| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import math |
| import re |
| import time |
| from collections import Counter, defaultdict |
| from pathlib import Path |
| from typing import Callable |
|
|
| import numpy as np |
| from sklearn.feature_extraction.text import TfidfVectorizer |
| from sklearn.metrics.pairwise import cosine_similarity |
|
|
| ROOT = Path(__file__).resolve().parent |
| DATA_PATH = ROOT / "benchmark-data.json" |
| CROSS_DATA_PATH = DATA_PATH |
| DEFAULT_OUTPUT = ROOT / "my-results-monolingual.json" |
|
|
| SEMANTIC_MODELS = { |
| "multilingual-e5-small": { |
| "repo_id": "intfloat/multilingual-e5-small", |
| "query_prefix": "query: ", |
| "passage_prefix": "passage: ", |
| "languages": 94, |
| "license": "MIT", |
| "role": "efficient multilingual retrieval model", |
| }, |
| "multilingual-minilm": { |
| "repo_id": "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2", |
| "query_prefix": "", |
| "passage_prefix": "", |
| "languages": 50, |
| "license": "Apache-2.0", |
| "role": "compact multilingual sentence similarity baseline", |
| }, |
| "labse": { |
| "repo_id": "sentence-transformers/LaBSE", |
| "query_prefix": "", |
| "passage_prefix": "", |
| "languages": 110, |
| "license": "Apache-2.0", |
| "role": "high-coverage multilingual embedding reference", |
| }, |
| } |
|
|
|
|
| def load_rows(path: Path = DATA_PATH) -> list[dict]: |
| payload = json.loads(path.read_text(encoding="utf-8")) |
| split = "cross-language" if getattr(load_rows, "use_cross", False) else "monolingual" |
| return payload["splits"][split]["rows"] |
|
|
| def language_of(row: dict) -> str: |
| return row.get("language") or row.get("query_language") or "unknown" |
|
|
| def pair_of(row: dict) -> str | None: |
| q = row.get("query_language") |
| p = row.get("passage_language") |
| return f"{q}->{p}" if q and p else None |
|
|
|
|
| def candidate_passages(row: dict) -> list[str]: |
| return [row["positive_passage"], row["hard_negative"], row["easy_negative"]] |
|
|
|
|
| def _result_record(row: dict, scores: list[float], elapsed_ms: float) -> dict: |
| order = np.argsort(np.asarray(scores))[::-1].tolist() |
| rank = order.index(0) + 1 |
| return { |
| "id": row["id"], |
| "language": language_of(row), |
| "language_pair": pair_of(row), |
| "domain": row["domain"], |
| "positive_rank": rank, |
| "scores": [float(x) for x in scores], |
| "latency_ms": float(elapsed_ms), |
| } |
|
|
|
|
| def evaluate_scorer(rows: list[dict], scorer: Callable[[str, list[str]], list[float]]) -> list[dict]: |
| out = [] |
| for row in rows: |
| passages = candidate_passages(row) |
| t0 = time.perf_counter() |
| scores = scorer(row["query"], passages) |
| elapsed_ms = (time.perf_counter() - t0) * 1000 |
| out.append(_result_record(row, scores, elapsed_ms)) |
| return out |
|
|
|
|
| def char_tfidf_scorer(query: str, passages: list[str]) -> list[float]: |
| vec = TfidfVectorizer(analyzer="char_wb", ngram_range=(3, 5), sublinear_tf=True) |
| matrix = vec.fit_transform(passages + [query]) |
| return cosine_similarity(matrix[-1], matrix[:-1]).ravel().tolist() |
|
|
|
|
| def word_tfidf_scorer(query: str, passages: list[str]) -> list[float]: |
| vec = TfidfVectorizer(analyzer="word", ngram_range=(1, 2), sublinear_tf=True, lowercase=True) |
| matrix = vec.fit_transform(passages + [query]) |
| return cosine_similarity(matrix[-1], matrix[:-1]).ravel().tolist() |
|
|
|
|
| def hybrid_tfidf_scorer(query: str, passages: list[str]) -> list[float]: |
| char_scores = np.asarray(char_tfidf_scorer(query, passages)) |
| word_scores = np.asarray(word_tfidf_scorer(query, passages)) |
| return (0.6 * char_scores + 0.4 * word_scores).tolist() |
|
|
|
|
| def tokenize(text: str) -> list[str]: |
| return re.findall(r"\b\w+\b", text.lower(), flags=re.UNICODE) |
|
|
|
|
| def bm25_scorer(query: str, passages: list[str], k1: float = 1.5, b: float = 0.75) -> list[float]: |
| docs = [tokenize(p) for p in passages] |
| q = tokenize(query) |
| n = len(docs) |
| avgdl = sum(len(d) for d in docs) / max(n, 1) |
| df = Counter() |
| for d in docs: |
| for term in set(d): |
| df[term] += 1 |
| scores = [] |
| for doc in docs: |
| tf = Counter(doc) |
| dl = len(doc) |
| score = 0.0 |
| for term in q: |
| if term not in tf: |
| continue |
| idf = math.log(1 + (n - df[term] + 0.5) / (df[term] + 0.5)) |
| denom = tf[term] + k1 * (1 - b + b * dl / max(avgdl, 1e-9)) |
| score += idf * (tf[term] * (k1 + 1)) / denom |
| scores.append(score) |
| return scores |
|
|
|
|
| def summarize(name: str, kind: str, results: list[dict], *, metadata: dict | None = None, load_seconds: float | None = None) -> dict: |
| n = len(results) |
| by_language = {} |
| by_domain = {} |
| by_pair = {} |
| for field, target in [("language", by_language), ("domain", by_domain), ("language_pair", by_pair)]: |
| for value in sorted({r[field] for r in results}): |
| part = [r for r in results if r[field] == value] |
| target[value] = { |
| "n": len(part), |
| "accuracy_at_1": sum(r["positive_rank"] == 1 for r in part) / len(part), |
| "mrr": sum(1 / r["positive_rank"] for r in part) / len(part), |
| } |
| return { |
| "name": name, |
| "kind": kind, |
| "status": "executed", |
| "n": n, |
| "accuracy_at_1": sum(r["positive_rank"] == 1 for r in results) / n, |
| "mrr": sum(1 / r["positive_rank"] for r in results) / n, |
| "avg_latency_ms": sum(r["latency_ms"] for r in results) / n, |
| "load_seconds": load_seconds, |
| "by_language": by_language, |
| "by_domain": by_domain, |
| "by_language_pair": {k:v for k,v in by_pair.items() if k is not None}, |
| "metadata": metadata or {}, |
| } |
|
|
|
|
| def run_lexical(rows: list[dict]) -> list[dict]: |
| scorers = { |
| "char-tfidf": char_tfidf_scorer, |
| "word-tfidf": word_tfidf_scorer, |
| "hybrid-tfidf": hybrid_tfidf_scorer, |
| "bm25": bm25_scorer, |
| } |
| return [summarize(name, "lexical", evaluate_scorer(rows, scorer)) for name, scorer in scorers.items()] |
|
|
|
|
| def run_semantic(rows: list[dict], key: str) -> dict: |
| spec = SEMANTIC_MODELS[key] |
| try: |
| from sentence_transformers import SentenceTransformer |
| except ImportError as exc: |
| raise RuntimeError("sentence-transformers is required for semantic benchmarks") from exc |
|
|
| t0 = time.perf_counter() |
| model = SentenceTransformer(spec["repo_id"]) |
| load_seconds = time.perf_counter() - t0 |
| results = [] |
| for row in rows: |
| passages = candidate_passages(row) |
| q = spec["query_prefix"] + row["query"] |
| docs = [spec["passage_prefix"] + p for p in passages] |
| t1 = time.perf_counter() |
| q_emb = model.encode([q], normalize_embeddings=True) |
| p_emb = model.encode(docs, normalize_embeddings=True) |
| scores = (q_emb @ p_emb.T)[0].tolist() |
| elapsed_ms = (time.perf_counter() - t1) * 1000 |
| results.append(_result_record(row, scores, elapsed_ms)) |
| return summarize(key, "semantic", results, metadata=spec, load_seconds=load_seconds) |
|
|
|
|
| def configured_semantic_placeholders() -> list[dict]: |
| return [ |
| { |
| "name": key, |
| "kind": "semantic", |
| "status": "configured-not-executed", |
| "n": 0, |
| "accuracy_at_1": None, |
| "mrr": None, |
| "avg_latency_ms": None, |
| "load_seconds": None, |
| "by_language": {}, |
| "by_domain": {}, |
| "metadata": spec, |
| } |
| for key, spec in SEMANTIC_MODELS.items() |
| ] |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description="Benchmark multilingual retrieval methods on Research Retrieval Eval Mini.") |
| parser.add_argument("--models", nargs="*", default=["lexical"], help="lexical, all, or semantic keys") |
| parser.add_argument("--split", choices=["monolingual", "cross-language"], default="monolingual") |
| parser.add_argument("--output", type=Path, default=None) |
| args = parser.parse_args() |
|
|
| data_path = DATA_PATH if args.split == "monolingual" else CROSS_DATA_PATH |
| load_rows.use_cross = args.split == "cross-language" |
| rows = load_rows(data_path) |
| output = args.output or (DEFAULT_OUTPUT if args.split == "monolingual" else ROOT / "my-results-cross-language.json") |
| requested = set(args.models) |
| models = run_lexical(rows) if ("lexical" in requested or "all" in requested) else [] |
| semantic_keys = list(SEMANTIC_MODELS) if "all" in requested else [x for x in requested if x in SEMANTIC_MODELS] |
| if not semantic_keys and ("lexical" in requested): |
| models.extend(configured_semantic_placeholders()) |
| else: |
| for key in semantic_keys: |
| try: |
| models.append(run_semantic(rows, key)) |
| except Exception as exc: |
| spec = SEMANTIC_MODELS[key] |
| models.append({ |
| "name": key, |
| "kind": "semantic", |
| "status": "error", |
| "error": str(exc), |
| "n": 0, |
| "accuracy_at_1": None, |
| "mrr": None, |
| "avg_latency_ms": None, |
| "load_seconds": None, |
| "by_language": {}, |
| "by_domain": {}, |
| "metadata": spec, |
| }) |
|
|
| payload = { |
| "benchmark": "Research Retrieval Eval Mini", |
| "split": args.split, |
| "dataset_rows": len(rows), |
| "languages": sorted({language_of(r) for r in rows}), |
| "language_pairs": sorted({pair_of(r) for r in rows if pair_of(r)}), |
| "domains": sorted({r["domain"] for r in rows}), |
| "metrics": ["accuracy_at_1", "mrr", "avg_latency_ms"], |
| "models": models, |
| "interpretation_note": "This is a small diagnostic benchmark. Results should not be generalized to broader retrieval quality without larger external evaluation.", |
| } |
| output.parent.mkdir(parents=True, exist_ok=True) |
| output.write_text(json.dumps(payload, indent=2, ensure_ascii=False), encoding="utf-8") |
| print(json.dumps(payload, indent=2, ensure_ascii=False)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
