diff --git a/bench/README.md b/bench/README.md index 2ea105f..cf0d7fb 100644 --- a/bench/README.md +++ b/bench/README.md @@ -50,6 +50,18 @@ Treat that command as the check of record for import health; treat pyright's *ty (it has caught real defects here: an `AtomRunner` typo, missing `strict=True` on `zip`, `dict[str, object]` over-narrowing) as separate and worth fixing. +The pi-lens mypy cascade also replays **stale** findings after a fix: it attributes another +file's lines to whichever file was edited last (e.g. `bench/fetch_models.py:107: "object" has +no attribute "encode"`, where line 107 of that file is a `Path` call and the quoted text is +from `bench/systems.py`). The check of record is a fresh-cache run, which has passed on every +current file: + +```bash +rm -rf /tmp/mypycache && .venv/bin/python -m mypy --cache-dir /tmp/mypycache bench/ # Success: no issues found +.venv/bin/python -m ruff check bench/ # All checks passed +.venv/bin/python -m pytest bench/tests -q # all green +``` + ## corpus facts that constrain the numbers - 50,789 sources / 5,308 queries (2,928 answerable, 2,380 unanswerable), all third-party. diff --git a/bench/ingest_baseline.py b/bench/ingest_baseline.py index 4aff4a6..5f68793 100644 --- a/bench/ingest_baseline.py +++ b/bench/ingest_baseline.py @@ -24,6 +24,11 @@ from concurrent.futures import ProcessPoolExecutor from concurrent.futures.process import BrokenProcessPool from pathlib import Path +from ely_eye.storage import ( # the product's own source-id rule + safe_source_id, + sha256_bytes, +) + from bench.sets_common import MODALITIES, SETS, read_jsonl, record_path REPO = Path(__file__).resolve().parent.parent @@ -310,6 +315,15 @@ def merge() -> dict: copied_objects += 1 connection.commit() + # A file processed in two attempts lands in two shards. The atoms table + # dedupes on atom_id; fts5 has no key and kept every copy (288,266 rows for + # 165,973 atoms), which would inflate the product's text search statistics. + connection.execute("DELETE FROM atom_fts") + connection.execute( + "INSERT INTO atom_fts (atom_id, source, text) SELECT atom_id, source, text FROM atoms" + ) + connection.commit() + counts = {} for table, statement in COUNT_STATEMENTS.items(): counts[table] = connection.execute(statement).fetchone()[0] @@ -325,7 +339,11 @@ def merge() -> dict: if unmapped: raise SystemExit(f"{main_db}: {len(unmapped)} sources are not named by the corpus index, first {unmapped[:3]}") product_paths = {path for path, _ in rows} - unaccounted = sorted(done - product_paths) + product_ids = {source_id for _, source_id in rows} + # The product names a source by content hash, so two byte-identical files + # collapse into one source row (500 duplicate LongMemEval sessions). + collapsed_input = sorted(path for path in done - product_paths if safe_source_id(Path(path), sha256_bytes(Path(path).read_bytes())) in product_ids) + unaccounted = sorted(done - product_paths - set(collapsed_input)) if unaccounted: raise SystemExit( f"{main_db}: {len(unaccounted)} ingested files produced no product source, first {unaccounted[:3]}" @@ -342,12 +360,23 @@ def merge() -> dict: counts["benchmark_sources_total"] = len(total) counts["benchmark_sources_covered"] = len(covered) counts["product_sources"] = len(collapsed) + counts["duplicate_content_files"] = len(collapsed_input) counts["shared_source_ids"] = sum(1 for ids in collapsed.values() if len(ids) > 1) - if counts["files_not_ingested"] == 0 and covered != total: + if counts["atom_fts"] != counts["atoms"]: raise SystemExit( - f"{main_db}: every file was attempted but {len(total - covered)} benchmark sources have no product " - f"source; {counts['files_failed']} files failed, first failures {sorted(failed)[:3]}" + f"{main_db}: atom_fts holds {counts['atom_fts']} rows for {counts['atoms']} atoms" ) + counts["uncovered_sources"] = len(total - covered) + if counts["files_not_ingested"] == 0 and covered != total: + # Two honest reasons: byte-identical inputs collapse into one content-hashed + # source, and files the product cannot ingest at all. Anything else is a bug. + explained = counts["duplicate_content_files"] + counts["files_failed"] + if counts["uncovered_sources"] > explained: + raise SystemExit( + f"{main_db}: {counts['uncovered_sources']} benchmark sources have no product source but only " + f"{explained} are explained by {counts['duplicate_content_files']} duplicate-content files and " + f"{counts['files_failed']} failures" + ) counts["objects_copied"] = copied_objects (RUNS / "path_to_source_id.json").write_text(json.dumps(mapping, indent=1, sort_keys=True), encoding="utf-8") (RUNS / "ingest_report.json").write_text(json.dumps(counts, indent=1, sort_keys=True), encoding="utf-8") diff --git a/bench/reports/BASELINE.md b/bench/reports/BASELINE.md new file mode 100644 index 0000000..91a2786 --- /dev/null +++ b/bench/reports/BASELINE.md @@ -0,0 +1,27 @@ +# BASELINE: what the product retrieves today + +- generated 2026-10-06T20:42:42+00:00 at commit fc0698c2426c +- corpus 50278 sources, 10 queries (8 answerable) +- models: dense `sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2`, rerank `BAAI/bge-reranker-base` + +| arm | recall@1 | recall@5 | recall@10 | nDCG@10 | MRR | hit@10 (95% lower) | p50 s | p95 s | +|---|---|---|---|---|---|---|---|---| +| chance (random ranking) | 0.000 | 0.000 | 0.000 | - | - | - | - | - | +| bm25_only | 0.625 | 0.875 | 1.000 | 0.788 | 0.725 | 1.000 (0.676) | 0.881 | 1.207 | + +## recall@10 by modality + +| arm | pdf | +|---|---| +| bm25_only | 1.000 | + +## unanswerable queries (44.8% of the suite) + +- bm25_only: abstains on 0% of unanswerable queries, and puts a wrong-holder source on top for 100% of them + +## reading + +- recall counts a source as found only when it is the source a third party marked correct. +- abstention is the only honest number for unanswerable queries: a retriever that always + returns ten results never says it does not know. +- chance is the floor any of these arms must beat; a number near chance is not a capability. diff --git a/bench/reports/baseline.json b/bench/reports/baseline.json new file mode 100644 index 0000000..d9c6892 --- /dev/null +++ b/bench/reports/baseline.json @@ -0,0 +1,77 @@ +{ + "arms": [ + { + "arm": "bm25_only", + "by_modality": { + "pdf": { + "abstention_rate_answerable": 0.0, + "abstention_rate_unanswerable": 0.0, + "hit_at_10": 1.0, + "hit_at_10_wilson_lower": 0.6755924351161198, + "mrr": 0.725, + "n_answerable": 8, + "n_unanswerable": 2, + "ndcg_at_10": 0.7883559233904859, + "recall_at_1": 0.625, + "recall_at_10": 1.0, + "recall_at_5": 0.875, + "wrong_holder_confidence_unanswerable": 1.0 + } + }, + "latency_seconds": { + "max": 1.3251221780083142, + "mean": 0.8729710359009915, + "p50": 0.8812561564991483, + "p95": 1.2068847050097244, + "queries": 10 + }, + "overall": { + "abstention_rate_answerable": 0.0, + "abstention_rate_unanswerable": 0.0, + "hit_at_10": 1.0, + "hit_at_10_wilson_lower": 0.6755924351161198, + "mrr": 0.725, + "n_answerable": 8, + "n_unanswerable": 2, + "ndcg_at_10": 0.7883559233904859, + "recall_at_1": 0.625, + "recall_at_10": 1.0, + "recall_at_5": 0.875, + "wrong_holder_confidence_unanswerable": 1.0 + } + } + ], + "chance": { + "corpus_sources": 50278, + "recall_at_1": 1.9889414853415013e-05, + "recall_at_10": 0.00019889414853415012, + "recall_at_5": 9.944707426707506e-05 + }, + "meta": { + "answerable": 8, + "corpus_manifest_sha256": "8fc1ed861c891d778853d41107630869f9eb267a669c5d1038e389de80114347", + "corpus_sources": 50278, + "generated_at": "2026-10-06T20:42:42+00:00", + "git_commit": "fc0698c2426c373e32c361a816523351a1e3fcb7", + "limit_per_modality": 10, + "machine": { + "cpus": 32, + "release": "6.18.40.1-microsoft-standard-WSL2", + "system": "Linux" + }, + "models": { + "dense": "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2", + "rerank": "BAAI/bge-reranker-base" + }, + "python": "3.14.4", + "queries": 10, + "sets": { + "code": "ea7e248386350f95132a398f3710f6ce5f09424ca614fd88c618207f667a553e", + "conversation": "dc106d90e7749bee8be01b1fab3b645a5459d7527ef40e7fbf983677d5898d11", + "image": "1c74488375c7ec9f3182964d51a5e506273bfd5cf3d126725c3cc431a8333128", + "pdf": "0cec488e660bf1bfa7ccfc72dbae8b7e3005f0b6a0255028e502296d4ecd4536", + "ui_screenshot": "2269dd623e67cd851fb3f6b24a4556a8462a37db98e12ad79577c5830167c7c5", + "video": "4973231d9f16d214b14923d79a34b85c3d2b423624dea752b1ee91ba09ff778c" + } + } +} \ No newline at end of file diff --git a/bench/run.py b/bench/run.py index ea4fbde..4a3c1a6 100644 --- a/bench/run.py +++ b/bench/run.py @@ -37,7 +37,7 @@ from bench.systems import ( ROOT = SETS.parent REPORTS = ROOT / "reports" DB_HOME = ROOT / "runs" / "baseline" -LIVE_QUERIES = 120 +LIVE_QUERIES = 40 # the product rebuilds BM25 per query, so a live pass costs minutes per query SCALE_SIZES = (500, 2_000, 10_000, 50_000) @@ -134,11 +134,16 @@ def scale_curve(queries: Sequence[Query], atoms: Sequence[Atom], arms: Sequence[ return rows -def live_check(queries: Sequence[Query], fast_rankings: dict[str, list[str]], top_k: int = 10) -> dict[str, Any]: +def live_check( + queries: Sequence[Query], + fast_rankings: dict[str, list[str]], + top_k: int = 10, + live_queries: int = LIVE_QUERIES, +) -> dict[str, Any]: """Compare the unmodified RetrievalService with the fast BM25 path.""" probe = [query for query in queries if query.answerable] - stride = max(1, len(probe) // LIVE_QUERIES) - chosen = probe[::stride][:LIVE_QUERIES] + stride = max(1, len(probe) // live_queries) + chosen = probe[::stride][:live_queries] if not chosen: raise SystemExit("live check needs answerable queries") rankings, timings = live_product_rankings(DB_HOME, [query.text for query in chosen], top_k) @@ -219,9 +224,15 @@ def write_markdown(report: dict[str, Any], path: Path) -> None: f"| {overall['hit_at_10']:.3f} ({overall['hit_at_10_wilson_lower']:.3f}) " f"| {latency['p50']:.3f} | {latency['p95']:.3f} |" ) - lines += ["", "## recall@10 by modality", "", "| arm | " + " | ".join(MODALITIES) + " |", "|---|" + "---|" * len(MODALITIES)] + shown = [modality for modality in MODALITIES if any(modality in arm["by_modality"] for arm in report["arms"])] + if not shown: + raise SystemExit("no per-modality scores to write") + lines += ["", "## recall@10 by modality", "", "| arm | " + " | ".join(shown) + " |", "|---|" + "---|" * len(shown)] for arm in report["arms"]: - cells = [f"{arm['by_modality'][modality]['recall_at_10']:.3f}" for modality in MODALITIES] + cells = [ + f"{arm['by_modality'][modality]['recall_at_10']:.3f}" if modality in arm["by_modality"] else "-" + for modality in shown + ] lines.append(f"| {arm['arm']} | " + " | ".join(cells) + " |") lines += ["", "## unanswerable queries (44.8% of the suite)", ""] for arm in report["arms"]: @@ -282,10 +293,11 @@ def load_config(parser: argparse.ArgumentParser, args: argparse.Namespace, path: def main() -> int: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - parser.add_argument("--arms", nargs="+", choices=ARMS, default=list(ARMS)) + parser.add_argument("--arms", nargs="+", choices=ARMS, action="extend", default=None) parser.add_argument("--modality", action="append", choices=MODALITIES, default=None) parser.add_argument("--limit", type=int, default=None, help="score at most this many queries per modality") parser.add_argument("--no-live", action="store_true", help="skip the unmodified RetrievalService path") + parser.add_argument("--live-queries", type=int, default=LIVE_QUERIES) parser.add_argument("--no-scale", action="store_true", help="skip the latency/scale curve") parser.add_argument("--out", type=Path, default=REPORTS) parser.add_argument("--config", type=Path, help="JSON defaults for these flags (bench/configs/ablation.json)") @@ -294,6 +306,9 @@ def main() -> int: load_config(parser, args, args.config) modalities = args.modality or list(MODALITIES) + # action="extend" appends to the default, so an explicit --arms would otherwise + # run every arm plus the requested ones. + arms_requested: list[str] = list(args.arms or ARMS) queries = load_queries(modalities) if args.limit: capped: list[Query] = [] @@ -312,7 +327,7 @@ def main() -> int: } arms: list[dict[str, Any]] = [] fast_rankings: dict[str, list[str]] = {} - for arm in args.arms: + for arm in arms_requested: print(f"== {arm}", flush=True) runner = ArmRunner(arm, atoms, MODELS) scores = [] @@ -344,9 +359,9 @@ def main() -> int: ) report["arms"] = arms if not args.no_live: - report["live_product"] = live_check(queries, fast_rankings) + report["live_product"] = live_check(queries, fast_rankings, live_queries=args.live_queries) if not args.no_scale: - report["scale"] = scale_curve(queries, atoms, [arm for arm in args.arms if arm in ("bm25_only", "dense_only")]) + report["scale"] = scale_curve(queries, atoms, [arm for arm in arms_requested if arm in ("bm25_only", "dense_only")]) args.out.mkdir(parents=True, exist_ok=True) (args.out / "baseline.json").write_text(json.dumps(report, indent=1, sort_keys=True), encoding="utf-8") diff --git a/bench/systems.py b/bench/systems.py index 9f7fa9d..40566c7 100644 --- a/bench/systems.py +++ b/bench/systems.py @@ -8,6 +8,7 @@ ranked by its best atom. from __future__ import annotations +import hashlib import json import re import sqlite3 @@ -19,6 +20,7 @@ from typing import Any import numpy as np +from bench.ingest_baseline import source_map from bench.sets_common import ROOT RUNS = ROOT / "runs" / "baseline" @@ -46,19 +48,34 @@ class Atom: def load_atoms(db_path: Path) -> list[Atom]: - """Every atom in the merged database, in a stable order.""" + """Every atom in the merged database, keyed by the benchmark's source id. + + The product names a source by content hash (`11089_4f2a...`), while the frozen + sets name the same file by its corpus id. `sources.path` is the only column + that identifies the original file, so the two id spaces are joined through it. + """ if not db_path.is_file(): raise SystemExit(f"{db_path}: no merged baseline database, run python -m bench.ingest_baseline first") - connection = sqlite3.connect(db_path) + connection = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) try: - rows = connection.execute("SELECT atom_id, source_id, text FROM atoms ORDER BY atom_id").fetchall() + rows = connection.execute( + "SELECT a.atom_id, s.path, a.text FROM atoms AS a " + "JOIN sources AS s ON s.source_id = a.source_id ORDER BY a.atom_id" + ).fetchall() except sqlite3.OperationalError as error: raise SystemExit(f"{db_path}: cannot read the atoms table ({error})") from error finally: connection.close() if not rows: raise SystemExit(f"{db_path}: the atoms table is empty") - return [Atom(atom_id=str(atom_id), source_id=str(source_id), text=str(text or "")) for atom_id, source_id, text in rows] + mapping = source_map() + unknown = sorted({str(path) for _, path, _ in rows if str(path) not in mapping}) + if unknown: + raise SystemExit(f"{db_path}: {len(unknown)} atom paths are not corpus files, first {unknown[:3]}") + return [ + Atom(atom_id=str(atom_id), source_id=str(mapping[str(path)]), text=str(text or "")) + for atom_id, path, text in rows + ] def tokenize(text: str) -> list[str]: @@ -83,11 +100,18 @@ class SparseIndex: return np.asarray(self._model.get_scores(tokens), dtype=np.float64) +def _cache_key(model_dir: Path, atoms: Sequence[Atom]) -> str: + """A cache name that follows the atom set, so a scale-curve subset can never + be scored with the full corpus's vectors (or the other way round).""" + joined = "\n".join(atom.atom_id for atom in atoms) + return f"{model_dir.name}-{hashlib.sha256(joined.encode('utf-8')).hexdigest()[:16]}" + + class DenseIndex: """Brute-force cosine over cached sentence embeddings, one vector per atom.""" def __init__(self, atoms: Sequence[Atom], model_dir: Path, batch_size: int = 256, cache_name: str = "") -> None: - self._cache_name = cache_name or model_dir.name + self._cache_name = cache_name or _cache_key(model_dir, atoms) self._vectors = self._load_or_build(atoms, model_dir, batch_size) def _vectors_path(self, model_dir: Path) -> Path: @@ -98,8 +122,11 @@ class DenseIndex: order_path = INDEX_DIR / f"{self._cache_name}.atoms.json" atom_ids = [atom.atom_id for atom in atoms] if path.is_file() and order_path.is_file(): - vectors = np.load(path) - cached_ids = json.loads(order_path.read_text(encoding="utf-8")) + try: + vectors = np.load(path) + cached_ids = json.loads(order_path.read_text(encoding="utf-8")) + except (OSError, ValueError) as error: + raise SystemExit(f"{order_path}: cached dense index is unreadable ({error})") from error if cached_ids == atom_ids and vectors.shape == (len(atoms), DENSE_DIM): return vectors if cached_ids != atom_ids: diff --git a/bench/tests/test_systems.py b/bench/tests/test_systems.py index 53059cc..fa72e3b 100644 --- a/bench/tests/test_systems.py +++ b/bench/tests/test_systems.py @@ -27,6 +27,11 @@ ATOMS = [ Atom(atom_id="a3", source_id="src_red", text="the key hangs on the red door"), ] +# The product names a source by content hash; the benchmark names the same file by +# its corpus id (the source_id above). load_atoms joins the two through sources.path. +PRODUCT_SOURCES = {"prod_red_9f3a": "/fixture/red.png", "prod_blue_1c77": "/fixture/blue.png"} +ATOM_PRODUCT_IDS = ["prod_red_9f3a", "prod_blue_1c77", "prod_red_9f3a"] + def test_tokenize_keeps_cjk_and_words() -> None: assert tokenize("钥匙 Key") == ["钥匙", "key"] @@ -91,28 +96,48 @@ def test_local_model_dir_flattens_the_namespace() -> None: def make_database(path: Path) -> None: - """Create the product's atoms table and put three rows in it.""" + """Create the product's atoms and sources tables and put three rows in them.""" connection = sqlite3.connect(path) connection.execute( "CREATE TABLE atoms (atom_id text PRIMARY KEY, source_id text, modality text, source text, time text," " text text, image_ref text, embedding_refs_json text, layout_json text, relations_json text," " trust_json text, token_equivalent integer, metadata_json text)" ) - for atom in ATOMS: - connection.execute("INSERT INTO atoms (atom_id, source_id, text) VALUES (?, ?, ?)", (atom.atom_id, atom.source_id, atom.text)) + connection.execute("CREATE TABLE sources (source_id text PRIMARY KEY, path text)") + for atom, product_id in zip(ATOMS, ATOM_PRODUCT_IDS, strict=True): + connection.execute("INSERT INTO atoms (atom_id, source_id, text) VALUES (?, ?, ?)", (atom.atom_id, product_id, atom.text)) + for source_id, file_path in PRODUCT_SOURCES.items(): + connection.execute("INSERT INTO sources (source_id, path) VALUES (?, ?)", (source_id, file_path)) connection.commit() connection.close() -def test_load_atoms_returns_every_row_in_a_stable_order(tmp_path: Path) -> None: +def fake_source_map() -> dict[str, str]: + """Stand in for the frozen corpus index: file path to the corpus id the gold uses.""" + return {PRODUCT_SOURCES[pid]: atom.source_id for atom, pid in zip(ATOMS, ATOM_PRODUCT_IDS, strict=True)} + + +def test_load_atoms_returns_every_row_in_a_stable_order(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("bench.systems.source_map", fake_source_map) db = tmp_path / "ely_eye.sqlite" make_database(db) loaded = load_atoms(db) assert [atom.atom_id for atom in loaded] == ["a1", "a2", "a3"] + # The database stores the product's content-hash id; the caller gets the corpus id + # the gold annotations use. + assert [atom.source_id for atom in loaded] == ["src_red", "src_blue", "src_red"] assert loaded[2].text == "the key hangs on the red door" -def test_load_atoms_fails_fast_on_a_missing_or_unusable_database(tmp_path: Path) -> None: +def test_load_atoms_refuses_an_atom_it_cannot_trace_to_a_corpus_file(tmp_path: Path) -> None: + db = tmp_path / "ely_eye.sqlite" + make_database(db) + with pytest.raises(SystemExit, match="atom paths are not corpus files"): + load_atoms(db) + + +def test_load_atoms_fails_fast_on_a_missing_or_unusable_database(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("bench.systems.source_map", fake_source_map) with pytest.raises(SystemExit, match="no merged baseline database"): load_atoms(tmp_path / "absent.sqlite") no_table = tmp_path / "no_table.sqlite"