From a6b3017cf638336eca01c84a35c90d5ec744cd33 Mon Sep 17 00:00:00 2001 From: Tobias Perelstein <5562156+tobocop2@users.noreply.github.com> Date: Fri, 17 Jul 2026 00:59:35 -0400 Subject: [PATCH] Add a blind, noise-calibrated retrieval eval harness Retrieval tuning has been graded by an ad-hoc kit living outside the repo; every future change needs a versioned gate. evals/retrieval generates a question battery from any indexed library (topical questions from sampled passages, known-item questions from sampled documents, count questions with exact streaming-scan oracles), answers it against two lilbee servers over the same index with per-row checkpointing, judges answers blind under opaque ids with one arm duplicated to measure the judge's noise floor, and scores mechanically: per-dimension means with the noise floor as the error bar plus exact pass/fail for counts and document lookups. Streaming scans and reservoir sampling keep memory flat on very large libraries. Lives outside src/ so it never ships in the package; lint and format gates cover it. bb-yjbr --- Makefile | 6 +- evals/__init__.py | 1 + evals/retrieval/README.md | 107 ++++++++++++++ evals/retrieval/__init__.py | 1 + evals/retrieval/__main__.py | 6 + evals/retrieval/answers.py | 133 ++++++++++++++++++ evals/retrieval/blinding.py | 119 ++++++++++++++++ evals/retrieval/checkpoint.py | 36 +++++ evals/retrieval/cli.py | 200 ++++++++++++++++++++++++++ evals/retrieval/judging.py | 97 +++++++++++++ evals/retrieval/llm.py | 82 +++++++++++ evals/retrieval/questions.py | 211 ++++++++++++++++++++++++++++ evals/retrieval/report.py | 57 ++++++++ evals/retrieval/scoring.py | 130 +++++++++++++++++ evals/retrieval/store_scan.py | 137 ++++++++++++++++++ pyproject.toml | 8 ++ tests/evals/__init__.py | 0 tests/evals/test_eval_answers.py | 120 ++++++++++++++++ tests/evals/test_eval_blinding.py | 121 ++++++++++++++++ tests/evals/test_eval_checkpoint.py | 40 ++++++ tests/evals/test_eval_cli.py | 173 +++++++++++++++++++++++ tests/evals/test_eval_judging.py | 69 +++++++++ tests/evals/test_eval_questions.py | 128 +++++++++++++++++ tests/evals/test_eval_report.py | 49 +++++++ tests/evals/test_eval_scoring.py | 158 +++++++++++++++++++++ tests/evals/test_eval_store_scan.py | 85 +++++++++++ 26 files changed, 2271 insertions(+), 3 deletions(-) create mode 100644 evals/__init__.py create mode 100644 evals/retrieval/README.md create mode 100644 evals/retrieval/__init__.py create mode 100644 evals/retrieval/__main__.py create mode 100644 evals/retrieval/answers.py create mode 100644 evals/retrieval/blinding.py create mode 100644 evals/retrieval/checkpoint.py create mode 100644 evals/retrieval/cli.py create mode 100644 evals/retrieval/judging.py create mode 100644 evals/retrieval/llm.py create mode 100644 evals/retrieval/questions.py create mode 100644 evals/retrieval/report.py create mode 100644 evals/retrieval/scoring.py create mode 100644 evals/retrieval/store_scan.py create mode 100644 tests/evals/__init__.py create mode 100644 tests/evals/test_eval_answers.py create mode 100644 tests/evals/test_eval_blinding.py create mode 100644 tests/evals/test_eval_checkpoint.py create mode 100644 tests/evals/test_eval_cli.py create mode 100644 tests/evals/test_eval_judging.py create mode 100644 tests/evals/test_eval_questions.py create mode 100644 tests/evals/test_eval_report.py create mode 100644 tests/evals/test_eval_scoring.py create mode 100644 tests/evals/test_eval_store_scan.py diff --git a/Makefile b/Makefile index 432ace662..45c7e1120 100644 --- a/Makefile +++ b/Makefile @@ -1,14 +1,14 @@ .PHONY: lint format format-check typecheck test test-ci test-ci-serial test-ci-forked test-integration imports-check check clean install demo demo-prep demo-publish build publish release promote release-promote docs docs-api docs-site site site-serve site-tar dns-setup qa-pod-volume qa-pod-up qa-pod-logs qa-pod-down lint: - uv run ruff check src/ tests/ tools/qa/ scripts/qa/ + uv run ruff check src/ tests/ tools/qa/ scripts/qa/ evals/ uv run python scripts/check_style_rules.py format: - uv run ruff format src/ tests/ tools/qa/ scripts/qa/ + uv run ruff format src/ tests/ tools/qa/ scripts/qa/ evals/ format-check: - uv run ruff format --check src/ tests/ tools/qa/ scripts/qa/ + uv run ruff format --check src/ tests/ tools/qa/ scripts/qa/ evals/ typecheck: uv run mypy src/lilbee/ diff --git a/evals/__init__.py b/evals/__init__.py new file mode 100644 index 000000000..5af0170ea --- /dev/null +++ b/evals/__init__.py @@ -0,0 +1 @@ +"""Version-controlled evaluation harnesses; never shipped with the package.""" diff --git a/evals/retrieval/README.md b/evals/retrieval/README.md new file mode 100644 index 000000000..0e0159a25 --- /dev/null +++ b/evals/retrieval/README.md @@ -0,0 +1,107 @@ +# Retrieval eval harness + +Blind, noise-calibrated A/B evaluation of lilbee's end-to-end retrieval +quality. Two lilbee servers (arm A and arm B: old vs new version, or two +configs) answer the same question battery over the SAME index; a judge model +grades answers blind; exact-truth questions are checked mechanically against +the store. Lives outside `src/` on purpose: it never ships in the package. + +## How it works + +1. **questions** generates the battery from an existing indexed library: + - *topical*: the configured chat model writes one question from each + sampled stored passage, so the passage that must support the answer is + known ground truth. Passages are reservoir-sampled from a streaming scan + (one per source); nothing materializes the index, so it scales to very + large libraries. + - *known_item*: asks what a sampled document is about; ground truth is the + document's head chunks, captured in the same scan. + - *count*: asks how many chunks and documents mention a mid-frequency + term. Ground truth is an exact streaming scan of the LanceDB store; no + judge involved. +2. **answer** runs the battery against one server (`/api/ask`), one run per + arm. It waits for the server's health route, retries each question three + times, and checkpoints every row to JSONL, so a killed pod run resumes + without redoing completed questions. +3. **judge** shuffles every gradable answer under an opaque id and grades + them one at a time: the judge sees only question + ground truth + one + answer, never arm labels, and never knows a comparison is happening. Arm + B's answers are judged twice under different ids; the disagreement between + those two passes is the judge's noise floor. Grades are checkpointed too. +4. **score** unblinds mechanically: per-dimension means (faithfulness, + relevance, citation, each 0-2) with the noise floor as the error bar, plus + exact pass/fail for count and known-item questions. Writes machine-readable + `results.jsonl`. +5. **report** renders `results.jsonl` as markdown; any cross-arm delta at or + below the noise floor is labeled within noise. + +## Running on a pod + +Typical detached run against a large private corpus: one shared index, two +installed lilbee versions serving on different ports. + +```bash +# One venv per arm, both pointed at the same data root. +OLD_PORT=8081 NEW_PORT=8082 +"$OLD_VENV/bin/lilbee" --data-dir "$DATA_ROOT" serve --port "$OLD_PORT" & +"$NEW_VENV/bin/lilbee" --data-dir "$DATA_ROOT" serve --port "$NEW_PORT" & + +cd ~/lilbee # the repo checkout; run everything from the repo root + +# 1. Questions (uses the configured chat model; scans the index directly). +uv run python -m evals.retrieval questions \ + --data-root "$DATA_ROOT" --out /tmp/eval/questions.jsonl + +# 2. Answers, one run per arm. Interrupted runs resume from the checkpoint. +uv run python -m evals.retrieval answer \ + --questions /tmp/eval/questions.jsonl --base-url "http://127.0.0.1:$OLD_PORT" \ + --arm old --out /tmp/eval/answers-old.jsonl +uv run python -m evals.retrieval answer \ + --questions /tmp/eval/questions.jsonl --base-url "http://127.0.0.1:$NEW_PORT" \ + --arm new --out /tmp/eval/answers-new.jsonl + +# 3. Blind judging. --answers-b is the arm judged twice for the noise floor. +# Grades are checkpointed; re-running the same command resumes. +uv run python -m evals.retrieval judge \ + --questions /tmp/eval/questions.jsonl \ + --answers-a /tmp/eval/answers-old.jsonl --answers-b /tmp/eval/answers-new.jsonl \ + --work-dir /tmp/eval/judge + +# 4 + 5. Score and render. +uv run python -m evals.retrieval score \ + --questions /tmp/eval/questions.jsonl \ + --answers-a /tmp/eval/answers-old.jsonl --answers-b /tmp/eval/answers-new.jsonl \ + --work-dir /tmp/eval/judge --out /tmp/eval/results.jsonl +uv run python -m evals.retrieval report \ + --results /tmp/eval/results.jsonl --out /tmp/eval/report.md +``` + +Question generation and judging talk to the configured lilbee chat model by +default (set `LILBEE_CHAT_MODEL` before step 1, and switch it to a stronger +judge model before step 3 if both run locally). The judge can instead be any +OpenAI-compatible endpoint: + +```bash +export LILBEE_EVAL_JUDGE_BASE_URL="https://api.example.com/v1" +export LILBEE_EVAL_JUDGE_MODEL="judge-model-name" +export LILBEE_EVAL_JUDGE_API_KEY="..." # optional +``` + +## Determinism and resume + +- `--seed` (questions, judge) makes sampling, blinding ids, and shuffles + reproducible. Re-running `judge` with the same seed and inputs regenerates + the identical blind set, so the grades checkpoint stays valid. +- `answer` and `judge` both checkpoint per row. Kill and re-run freely; only + unfinished work repeats. +- Answers that hard-fail after all retries are recorded as failures and score + zero everywhere (prefailed: they never reach the judge). + +## Files in a judge work dir + +| file | contents | +| --- | --- | +| `blind_rows.jsonl` | what judges see: gid, question, ground truth, answer | +| `gid_map.json` | secret gid to (qid, arm, replicate) mapping | +| `prefailed.json` | gids scored zero without judging | +| `grades.jsonl` | checkpointed judge output, one row per gid | diff --git a/evals/retrieval/__init__.py b/evals/retrieval/__init__.py new file mode 100644 index 000000000..778cae64f --- /dev/null +++ b/evals/retrieval/__init__.py @@ -0,0 +1 @@ +"""Blind, noise-calibrated A/B evaluation of lilbee retrieval quality.""" diff --git a/evals/retrieval/__main__.py b/evals/retrieval/__main__.py new file mode 100644 index 000000000..0f9016f24 --- /dev/null +++ b/evals/retrieval/__main__.py @@ -0,0 +1,6 @@ +"""Runs the retrieval eval CLI: python -m evals.retrieval .""" + +from evals.retrieval.cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/evals/retrieval/answers.py b/evals/retrieval/answers.py new file mode 100644 index 000000000..d3db37578 --- /dev/null +++ b/evals/retrieval/answers.py @@ -0,0 +1,133 @@ +"""Answer collection from a running lilbee server, checkpointed for pod restarts.""" + +from __future__ import annotations + +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +import httpx + +from evals.retrieval.checkpoint import JsonlCheckpoint +from evals.retrieval.questions import Question + +HEALTH_ROUTE = "/api/health" +ASK_ROUTE = "/api/ask" +ASK_TIMEOUT_SECONDS = 600.0 +ANSWER_ATTEMPTS = 3 +ANSWER_RETRY_DELAY_SECONDS = 5.0 +WARMUP_ATTEMPTS = 180 +WARMUP_POLL_SECONDS = 5.0 + + +@dataclass +class AnswerRow: + """One arm's answer to one question, or its hard failure.""" + + qid: str + arm: str + answer: str + sources: list[str] + cited_sources: list[str] + seconds: float + error: str | None + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> AnswerRow: + return cls( + qid=data["qid"], + arm=data["arm"], + answer=data["answer"], + sources=list(data["sources"]), + cited_sources=list(data["cited_sources"]), + seconds=data["seconds"], + error=data["error"], + ) + + +def make_http_client() -> httpx.Client: + return httpx.Client(timeout=ASK_TIMEOUT_SECONDS) + + +def wait_for_server( + base_url: str, + client: httpx.Client, + *, + attempts: int = WARMUP_ATTEMPTS, + poll: float = WARMUP_POLL_SECONDS, +) -> None: + """Block until the server's health route answers 200.""" + for _ in range(attempts): + try: + if client.get(f"{base_url.rstrip('/')}{HEALTH_ROUTE}").status_code == httpx.codes.OK: + return + except httpx.HTTPError: + pass + time.sleep(poll) + raise RuntimeError(f"server at {base_url} never became healthy") + + +def _ask(client: httpx.Client, base_url: str, question: str, top_k: int) -> AnswerRow: + response = client.post( + f"{base_url.rstrip('/')}{ASK_ROUTE}", json={"question": question, "top_k": top_k} + ) + response.raise_for_status() + body = response.json() + return AnswerRow( + qid="", + arm="", + answer=body["answer"], + sources=[chunk["source"] for chunk in body["sources"]], + cited_sources=[chunk["source"] for chunk in body.get("cited_sources", [])], + seconds=0.0, + error=None, + ) + + +def answer_questions( + questions: list[Question], + base_url: str, + arm: str, + out_path: Path, + *, + top_k: int = 0, + attempts: int = ANSWER_ATTEMPTS, + retry_delay: float = ANSWER_RETRY_DELAY_SECONDS, + client: httpx.Client | None = None, +) -> list[AnswerRow]: + """Answer every question not already checkpointed; return this run's rows.""" + http = client or make_http_client() + checkpoint = JsonlCheckpoint(out_path, "qid") + wait_for_server(base_url, http) + rows: list[AnswerRow] = [] + for question in questions: + if question.qid in checkpoint: + continue + started = time.monotonic() + row: AnswerRow | None = None + error = "" + for attempt in range(attempts): + try: + row = _ask(http, base_url, question.question, top_k) + break + except (httpx.HTTPError, KeyError, ValueError) as exc: + error = f"{type(exc).__name__}: {exc}" + if attempt + 1 < attempts: + time.sleep(retry_delay) + if row is None: + # A question that fails all attempts is itself a result. + row = AnswerRow( + qid="", arm="", answer="", sources=[], cited_sources=[], seconds=0.0, error=error + ) + row.qid = question.qid + row.arm = arm + row.seconds = round(time.monotonic() - started, 2) + checkpoint.append(row.to_dict()) + rows.append(row) + status = "ERROR" if row.error else "ok" + print(f"[{arm}] {question.qid} {status}", flush=True) + return rows diff --git a/evals/retrieval/blinding.py b/evals/retrieval/blinding.py new file mode 100644 index 000000000..ed4023cc3 --- /dev/null +++ b/evals/retrieval/blinding.py @@ -0,0 +1,119 @@ +"""Blind row construction and mechanical unblinding. + +Every gradable (question, answer) pair becomes a row under an opaque gid. +The noise arm's answers appear twice (replicates 0 and 1) so the judge's +per-question disagreement with itself is measurable; all rows shuffle +together, so a judge cannot tell arms, replicates, or that a comparison is +happening at all. +""" + +from __future__ import annotations + +import random +from dataclasses import asdict, dataclass +from typing import Any, NamedTuple + +from evals.retrieval.answers import AnswerRow +from evals.retrieval.questions import Question, QuestionKind + +GROUND_CHARS = 2400 +ANSWER_CHARS = 2400 +GID_SPACE = 10**9 +NOISE_REPLICATES = 2 +JUDGED_KINDS = (QuestionKind.TOPICAL, QuestionKind.KNOWN_ITEM) + + +@dataclass(frozen=True) +class BlindRow: + """What a judge sees: no qid, no arm, no replicate.""" + + gid: str + question: str + source: str + ground: str + answer: str + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass(frozen=True) +class BlindAssignment: + """The secret side of a gid; never shown to a judge.""" + + qid: str + arm: str + replicate: int + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> BlindAssignment: + return cls(qid=data["qid"], arm=data["arm"], replicate=data["replicate"]) + + +class BlindSet(NamedTuple): + rows: list[BlindRow] + assignments: dict[str, BlindAssignment] + prefailed: list[str] + + +def _new_gid(rng: random.Random, taken: set[str]) -> str: + gid = f"g{rng.randrange(GID_SPACE):09d}" + while gid in taken: + gid = f"g{rng.randrange(GID_SPACE):09d}" + return gid + + +def build_blind_rows( + questions: list[Question], + answers_by_arm: dict[str, dict[str, AnswerRow]], + noise_arm: str, + rng: random.Random, +) -> BlindSet: + """Blind rows for every judged question, with the noise arm duplicated. + + Missing, errored, and empty answers are prefailed: they score zero without + wasting a judge call, and their gids never reach a judge. + """ + rows: list[BlindRow] = [] + assignments: dict[str, BlindAssignment] = {} + prefailed: list[str] = [] + for question in questions: + if question.kind not in JUDGED_KINDS: + continue + for arm, answers in answers_by_arm.items(): + replicates = NOISE_REPLICATES if arm == noise_arm else 1 + for replicate in range(replicates): + gid = _new_gid(rng, set(assignments)) + assignments[gid] = BlindAssignment(qid=question.qid, arm=arm, replicate=replicate) + answer = answers.get(question.qid) + if answer is None or answer.error or not answer.answer.strip(): + prefailed.append(gid) + continue + rows.append( + BlindRow( + gid=gid, + question=question.question, + source=question.source, + ground=question.ground_passage[:GROUND_CHARS], + answer=answer.answer[:ANSWER_CHARS], + ) + ) + rng.shuffle(rows) + return BlindSet(rows=rows, assignments=assignments, prefailed=prefailed) + + +def unblind( + assignments: dict[str, BlindAssignment], grades: dict[str, dict[str, int]] +) -> dict[str, dict[int, dict[str, dict[str, int]]]]: + """Regroup blind grades as arm -> replicate -> qid -> scores.""" + regrouped: dict[str, dict[int, dict[str, dict[str, int]]]] = {} + for gid, assignment in assignments.items(): + by_replicate = regrouped.setdefault(assignment.arm, {}) + by_qid = by_replicate.setdefault(assignment.replicate, {}) + scores = grades.get(gid) + if scores is not None: + by_qid[assignment.qid] = scores + return regrouped diff --git a/evals/retrieval/checkpoint.py b/evals/retrieval/checkpoint.py new file mode 100644 index 000000000..9769c766c --- /dev/null +++ b/evals/retrieval/checkpoint.py @@ -0,0 +1,36 @@ +"""Append-only JSONL checkpointing so interrupted runs resume without rework.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + + +def load_jsonl(path: Path) -> list[dict[str, Any]]: + """Parse one JSON object per line, skipping blanks; missing file is empty.""" + if not path.exists(): + return [] + return [json.loads(line) for line in path.read_text().splitlines() if line.strip()] + + +class JsonlCheckpoint: + """Appends keyed records to a JSONL file and remembers which keys are done.""" + + def __init__(self, path: Path, key_field: str) -> None: + self._path = path + self._key_field = key_field + self._done = {str(record[key_field]) for record in load_jsonl(path)} + path.parent.mkdir(parents=True, exist_ok=True) + + @property + def done(self) -> set[str]: + return set(self._done) + + def __contains__(self, key: str) -> bool: + return key in self._done + + def append(self, record: dict[str, Any]) -> None: + with self._path.open("a") as fh: + fh.write(json.dumps(record) + "\n") + self._done.add(str(record[self._key_field])) diff --git a/evals/retrieval/cli.py b/evals/retrieval/cli.py new file mode 100644 index 000000000..1a80a3f30 --- /dev/null +++ b/evals/retrieval/cli.py @@ -0,0 +1,200 @@ +"""Command-line entry point: questions, answer, judge, score, report.""" + +from __future__ import annotations + +import argparse +import json +import random +import sys +from pathlib import Path +from typing import Any + +from evals.retrieval.answers import AnswerRow, answer_questions, make_http_client +from evals.retrieval.blinding import BlindAssignment, build_blind_rows, unblind +from evals.retrieval.checkpoint import load_jsonl +from evals.retrieval.judging import DIMENSIONS, judge_rows +from evals.retrieval.llm import judge_chat_fn, lilbee_chat_fn, warm_chat +from evals.retrieval.questions import ( + COUNT_QUESTIONS, + DEFAULT_SEED, + KNOWN_ITEM_QUESTIONS, + TOPICAL_QUESTIONS, + Question, + build_questions, +) +from evals.retrieval.report import render_report +from evals.retrieval.scoring import build_results + +GID_MAP_FILE = "gid_map.json" +PREFAILED_FILE = "prefailed.json" +BLIND_ROWS_FILE = "blind_rows.jsonl" +GRADES_FILE = "grades.jsonl" + + +def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("".join(json.dumps(row) + "\n" for row in rows)) + + +def _load_questions(path: Path) -> list[Question]: + return [Question.from_dict(row) for row in load_jsonl(path)] + + +def _load_answer_arm(path: Path) -> tuple[str, dict[str, AnswerRow]]: + rows = [AnswerRow.from_dict(row) for row in load_jsonl(path)] + if not rows: + raise ValueError(f"no answer rows in {path}") + return rows[0].arm, {row.qid: row for row in rows} + + +def _load_arms(args: argparse.Namespace) -> dict[str, dict[str, AnswerRow]]: + arm_a, answers_a = _load_answer_arm(args.answers_a) + arm_b, answers_b = _load_answer_arm(args.answers_b) + if arm_a == arm_b: + raise ValueError(f"both answer files carry the arm label '{arm_a}'") + return {arm_a: answers_a, arm_b: answers_b} + + +def _cmd_questions(args: argparse.Namespace) -> int: + lancedb_dir = args.lancedb_dir or args.data_root / "data" / "lancedb" + chat = lilbee_chat_fn() + warm_chat(chat) + questions = build_questions( + lancedb_dir, + chat, + topical=args.topical, + known_item=args.known_item, + count=args.count, + seed=args.seed, + ) + _write_jsonl(args.out, [question.to_dict() for question in questions]) + print(f"wrote {len(questions)} questions -> {args.out}") + return 0 + + +def _cmd_answer(args: argparse.Namespace) -> int: + questions = _load_questions(args.questions) + rows = answer_questions( + questions, + args.base_url, + args.arm, + args.out, + top_k=args.top_k, + client=make_http_client(), + ) + failed = sum(1 for row in rows if row.error) + print(f"answered {len(rows)} new questions, {failed} errors -> {args.out}") + return 0 + + +def _cmd_judge(args: argparse.Namespace) -> int: + questions = _load_questions(args.questions) + answers_by_arm = _load_arms(args) + noise_arm = next(reversed(answers_by_arm)) + blind = build_blind_rows(questions, answers_by_arm, noise_arm, random.Random(args.seed)) + args.work_dir.mkdir(parents=True, exist_ok=True) + (args.work_dir / GID_MAP_FILE).write_text( + json.dumps({gid: a.to_dict() for gid, a in blind.assignments.items()}, indent=1) + ) + (args.work_dir / PREFAILED_FILE).write_text(json.dumps(blind.prefailed, indent=1)) + _write_jsonl(args.work_dir / BLIND_ROWS_FILE, [row.to_dict() for row in blind.rows]) + chat = judge_chat_fn() + warm_chat(chat) + grades = judge_rows(blind.rows, chat, args.work_dir / GRADES_FILE) + unreturned = len(blind.rows) - sum(1 for row in blind.rows if row.gid in grades) + print( + f"{len(blind.rows)} blind rows judged ({unreturned} unreturned), " + f"{len(blind.prefailed)} prefailed -> {args.work_dir}" + ) + return 0 + + +def _cmd_score(args: argparse.Namespace) -> int: + questions = _load_questions(args.questions) + answers_by_arm = _load_arms(args) + noise_arm = next(reversed(answers_by_arm)) + assignments = { + gid: BlindAssignment.from_dict(data) + for gid, data in json.loads((args.work_dir / GID_MAP_FILE).read_text()).items() + } + prefailed = json.loads((args.work_dir / PREFAILED_FILE).read_text()) + grades = { + record["gid"]: {k: v for k, v in record.items() if k != "gid"} + for record in load_jsonl(args.work_dir / GRADES_FILE) + } + for gid in prefailed: + grades[gid] = dict.fromkeys(DIMENSIONS, 0) + results = build_results(questions, answers_by_arm, unblind(assignments, grades), noise_arm) + _write_jsonl(args.out, results) + print(f"wrote {len(results)} result rows -> {args.out}") + return 0 + + +def _cmd_report(args: argparse.Namespace) -> int: + report = render_report(load_jsonl(args.results)) + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(report) + print(f"wrote {args.out}") + return 0 + + +def _add_arm_io_arguments(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--questions", type=Path, required=True) + parser.add_argument("--answers-a", type=Path, required=True) + parser.add_argument( + "--answers-b", type=Path, required=True, help="the arm judged twice for the noise floor" + ) + parser.add_argument("--work-dir", type=Path, required=True) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="evals.retrieval", description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + + questions = subparsers.add_parser("questions", help="generate the question battery") + questions.add_argument("--data-root", type=Path, required=True) + questions.add_argument( + "--lancedb-dir", + type=Path, + default=None, + help="override the default /data/lancedb", + ) + questions.add_argument("--out", type=Path, required=True) + questions.add_argument("--topical", type=int, default=TOPICAL_QUESTIONS) + questions.add_argument("--known-item", type=int, default=KNOWN_ITEM_QUESTIONS) + questions.add_argument("--count", type=int, default=COUNT_QUESTIONS) + questions.add_argument("--seed", type=int, default=DEFAULT_SEED) + questions.set_defaults(handler=_cmd_questions) + + answer = subparsers.add_parser("answer", help="answer the battery against one server") + answer.add_argument("--questions", type=Path, required=True) + answer.add_argument("--base-url", required=True) + answer.add_argument("--arm", required=True, help="label recorded on every row") + answer.add_argument("--out", type=Path, required=True) + answer.add_argument("--top-k", type=int, default=0) + answer.set_defaults(handler=_cmd_answer) + + judge = subparsers.add_parser("judge", help="blind-judge both arms' answers") + _add_arm_io_arguments(judge) + judge.add_argument("--seed", type=int, default=DEFAULT_SEED) + judge.set_defaults(handler=_cmd_judge) + + score = subparsers.add_parser("score", help="unblind grades and write results.jsonl") + _add_arm_io_arguments(score) + score.add_argument("--out", type=Path, required=True) + score.set_defaults(handler=_cmd_score) + + report = subparsers.add_parser("report", help="render results.jsonl as markdown") + report.add_argument("--results", type=Path, required=True) + report.add_argument("--out", type=Path, required=True) + report.set_defaults(handler=_cmd_report) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + try: + return int(args.handler(args)) + except (ValueError, FileNotFoundError, RuntimeError) as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 diff --git a/evals/retrieval/judging.py b/evals/retrieval/judging.py new file mode 100644 index 000000000..777ceb7db --- /dev/null +++ b/evals/retrieval/judging.py @@ -0,0 +1,97 @@ +"""Blind judging: one answer at a time against its ground truth.""" + +from __future__ import annotations + +import json +import re +import sys +import time +from enum import StrEnum +from pathlib import Path + +from evals.retrieval.blinding import BlindRow +from evals.retrieval.checkpoint import JsonlCheckpoint, load_jsonl +from evals.retrieval.llm import ChatFn + +SCORE_MIN = 0 +SCORE_MAX = 2 +JUDGE_ATTEMPTS = 3 +JUDGE_RETRY_DELAY_SECONDS = 5.0 + +_JSON_RE = re.compile(r"\{[^{}]*\}") + + +class Dimension(StrEnum): + FAITHFULNESS = "faithfulness" + RELEVANCE = "relevance" + CITATION = "citation" + + +DIMENSIONS = tuple(str(dimension) for dimension in Dimension) + +JUDGE_PROMPT = ( + "You are grading an answer produced by a document-search assistant.\n" + "Question: {question}\n\n" + "Ground-truth material the answer should be based on (from document " + "{source}):\n{ground}\n\n" + "Answer to grade:\n{answer}\n\n" + "Score STRICTLY as JSON with integer fields 0-2 each: " + '{{"faithfulness": _, "relevance": _, "citation": _}}. ' + "faithfulness: 2 = every claim supported by the ground material, " + "1 = minor unsupported detail, 0 = contradicts or invents. " + "relevance: 2 = directly answers the question, 1 = partial, 0 = misses. " + "citation: 2 = names or cites the correct document, 1 = cites nothing, " + "0 = cites a wrong document. Return ONLY the JSON." +) + + +def parse_grade(text: str) -> dict[str, int] | None: + """Scores clamped to the 0-2 scale, or None when the judge returned junk.""" + match = _JSON_RE.search(text) + if not match: + return None + try: + scores = json.loads(match.group(0)) + return { + dimension: max(SCORE_MIN, min(SCORE_MAX, int(scores[dimension]))) + for dimension in DIMENSIONS + } + except (KeyError, TypeError, ValueError): + return None + + +def judge_rows( + rows: list[BlindRow], + chat: ChatFn, + out_path: Path, + *, + attempts: int = JUDGE_ATTEMPTS, + retry_delay: float = JUDGE_RETRY_DELAY_SECONDS, +) -> dict[str, dict[str, int]]: + """Grade every row not already checkpointed; return all grades on disk.""" + checkpoint = JsonlCheckpoint(out_path, "gid") + for row in rows: + if row.gid in checkpoint: + continue + prompt = JUDGE_PROMPT.format( + question=row.question, source=row.source, ground=row.ground, answer=row.answer + ) + grade: dict[str, int] | None = None + for attempt in range(attempts): + try: + grade = parse_grade(chat(prompt)) + except Exception as exc: # a failed grade is skipped, not fatal + print(f"judge call failed for {row.gid}: {exc}", file=sys.stderr) + grade = None + if grade is not None: + break + if attempt + 1 < attempts: + time.sleep(retry_delay) + if grade is None: + print(f"no parseable grade for {row.gid}; leaving unreturned", file=sys.stderr) + continue + checkpoint.append({"gid": row.gid, **grade}) + return { + record["gid"]: {dimension: record[dimension] for dimension in DIMENSIONS} + for record in load_jsonl(out_path) + } diff --git a/evals/retrieval/llm.py b/evals/retrieval/llm.py new file mode 100644 index 000000000..de120d66b --- /dev/null +++ b/evals/retrieval/llm.py @@ -0,0 +1,82 @@ +"""Chat backends for question authoring and judging.""" + +from __future__ import annotations + +import os +import time +from collections.abc import Callable + +import httpx + +ChatFn = Callable[[str], str] + +JUDGE_BASE_URL_ENV = "LILBEE_EVAL_JUDGE_BASE_URL" +JUDGE_MODEL_ENV = "LILBEE_EVAL_JUDGE_MODEL" +JUDGE_API_KEY_ENV = "LILBEE_EVAL_JUDGE_API_KEY" + +CHAT_MAX_TOKENS = 256 +CHAT_TIMEOUT_SECONDS = 300.0 +WARM_ATTEMPTS = 30 +WARM_DELAY_SECONDS = 10.0 + + +def openai_chat_fn( + base_url: str, model: str, api_key: str | None = None, *, client: httpx.Client | None = None +) -> ChatFn: + """Prompt-to-text over an OpenAI-compatible /chat/completions endpoint.""" + http = client or httpx.Client(timeout=CHAT_TIMEOUT_SECONDS) + headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} + url = f"{base_url.rstrip('/')}/chat/completions" + + def chat(prompt: str) -> str: + payload = { + "model": model, + "messages": [{"role": "user", "content": prompt}], + "temperature": 0, + "max_tokens": CHAT_MAX_TOKENS, + } + response = http.post(url, json=payload, headers=headers) + response.raise_for_status() + return str(response.json()["choices"][0]["message"]["content"]) + + return chat + + +def lilbee_chat_fn() -> ChatFn: + """Prompt-to-text on the configured lilbee chat model, reasoning stripped.""" + from lilbee.app.services import get_services # heavy: builds the service container + from lilbee.retrieval.reasoning import strip_reasoning + + provider = get_services().provider + options = {"temperature": 0, "num_predict": CHAT_MAX_TOKENS, "think": False} + + def chat(prompt: str) -> str: + result = provider.chat([{"role": "user", "content": prompt}], options=options) + return strip_reasoning(result.text) + + return chat + + +def judge_chat_fn() -> ChatFn: + """The judge backend: env-configured OpenAI-compatible endpoint, else lilbee.""" + base_url = os.environ.get(JUDGE_BASE_URL_ENV) + if base_url: + model = os.environ.get(JUDGE_MODEL_ENV, "") + return openai_chat_fn(base_url, model, os.environ.get(JUDGE_API_KEY_ENV)) + return lilbee_chat_fn() + + +def warm_chat( + chat: ChatFn, attempts: int = WARM_ATTEMPTS, delay: float = WARM_DELAY_SECONDS +) -> None: + """Block until the chat backend answers; first calls race model load.""" + last: Exception | None = None + for _ in range(attempts): + try: + chat("ok") + except Exception as exc: # retried until the budget runs out + last = exc + time.sleep(delay) + else: + return + raise RuntimeError(f"chat backend never came up: {last}") diff --git a/evals/retrieval/questions.py b/evals/retrieval/questions.py new file mode 100644 index 000000000..c980b43e0 --- /dev/null +++ b/evals/retrieval/questions.py @@ -0,0 +1,211 @@ +"""Corpus-agnostic question generation from an existing lilbee index. + +Three kinds, each with ground truth attached at authoring time so judging +never needs the index again: + +- topical: written by the chat model FROM a sampled stored passage, so the + passage that must support the answer is known. +- known_item: asks what a sampled document is about; ground truth is the + document's head chunks. +- count: asks how many chunks and documents mention a term; ground truth is + an exact streaming scan of the store, no judge involved. +""" + +from __future__ import annotations + +import collections +import random +import re +import sys +import time +from dataclasses import asdict, dataclass +from enum import StrEnum +from pathlib import Path +from typing import Any + +from evals.retrieval.llm import ChatFn +from evals.retrieval.store_scan import ( + count_term_hits, + iter_chunks, + iter_source_names, + reservoir_sample, + scan_passages_and_heads, +) + +TOPICAL_QUESTIONS = 60 +KNOWN_ITEM_QUESTIONS = 20 +COUNT_QUESTIONS = 8 +MIN_PASSAGE_CHARS = 400 +PASSAGE_PROMPT_CHARS = 1800 +MIN_QUESTION_CHARS = 16 +TERM_DF_LOW = 0.05 +TERM_DF_HIGH = 0.4 +AUTHOR_ATTEMPTS = 3 +AUTHOR_RETRY_DELAY_SECONDS = 5.0 +DEFAULT_SEED = 20260714 + +_WORD_RE = re.compile(r"[a-z]{5,16}") + +QUESTION_PROMPT = ( + "Below is a passage from a document. Write ONE specific question that this " + "passage answers. The question must be answerable from the passage alone, " + "must not mention 'the passage' or 'the text', and must not quote more " + "than three consecutive words from it. Return ONLY the question.\n\n" + "Passage:\n{passage}" +) + + +class QuestionKind(StrEnum): + TOPICAL = "topical" + KNOWN_ITEM = "known_item" + COUNT = "count" + + +@dataclass +class CountOracle: + """Exact scan result a count answer is checked against.""" + + term: str + chunks: int + sources: int + + +@dataclass +class Question: + """One eval question with its ground truth attached.""" + + qid: str + kind: QuestionKind + question: str + source: str = "" + ground_passage: str = "" + oracle: CountOracle | None = None + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Question: + oracle = data.get("oracle") + return cls( + qid=data["qid"], + kind=QuestionKind(data["kind"]), + question=data["question"], + source=data.get("source", ""), + ground_passage=data.get("ground_passage", ""), + oracle=CountOracle(**oracle) if oracle else None, + ) + + +def parse_question(text: str) -> str | None: + """The first line as a question, or None when the model returned junk.""" + lines = text.strip().splitlines() + if not lines: + return None + candidate = lines[0].strip().strip('"') + if candidate.endswith("?") and len(candidate) >= MIN_QUESTION_CHARS: + return candidate + return None + + +def author_topical( + passages: list[tuple[str, str]], + chat: ChatFn, + *, + attempts: int = AUTHOR_ATTEMPTS, + retry_delay: float = AUTHOR_RETRY_DELAY_SECONDS, +) -> list[Question]: + """One question per sampled passage; a bad passage never kills the set.""" + questions: list[Question] = [] + for index, (source, passage) in enumerate(passages): + prompt = QUESTION_PROMPT.format(passage=passage[:PASSAGE_PROMPT_CHARS]) + response = None + for attempt in range(attempts): + try: + response = chat(prompt) + break + except Exception as exc: # a failed question is skipped, not fatal + print( + f"authoring attempt {attempt + 1} failed for {source}: {exc}", file=sys.stderr + ) + time.sleep(retry_delay) + if response is None: + continue + question = parse_question(response) + if question is None: + continue + questions.append( + Question( + qid=f"tq{index:03d}", + kind=QuestionKind.TOPICAL, + question=question, + source=source, + ground_passage=passage, + ) + ) + return questions + + +def sample_terms(passages: list[tuple[str, str]], count: int, rng: random.Random) -> list[str]: + """Mid-document-frequency terms drawn from the sampled passages. + + Document frequency is estimated over the sample, never the full index; + the exact oracle counts come from a full streaming scan afterwards. + """ + per_source_terms: dict[str, set[str]] = collections.defaultdict(set) + for source, passage in passages: + per_source_terms[source].update(_WORD_RE.findall(passage.lower())) + doc_freq: collections.Counter[str] = collections.Counter() + for terms in per_source_terms.values(): + doc_freq.update(terms) + total = len(per_source_terms) + if not total: + return [] + band = sorted(t for t, df in doc_freq.items() if TERM_DF_LOW <= df / total <= TERM_DF_HIGH) + rng.shuffle(band) + return band[:count] + + +def build_questions( + lancedb_dir: Path, + chat: ChatFn, + *, + topical: int = TOPICAL_QUESTIONS, + known_item: int = KNOWN_ITEM_QUESTIONS, + count: int = COUNT_QUESTIONS, + seed: int = DEFAULT_SEED, + min_passage_chars: int = MIN_PASSAGE_CHARS, +) -> list[Question]: + """The full question battery: sampled sources, two streaming chunk scans.""" + rng = random.Random(seed) + known_sources = sorted(reservoir_sample(iter_source_names(lancedb_dir), known_item, rng)) + scan = scan_passages_and_heads( + iter_chunks(lancedb_dir), + passage_count=topical, + min_passage_chars=min_passage_chars, + head_sources=set(known_sources), + rng=rng, + ) + questions = author_topical(scan.passages, chat) + for index, name in enumerate(known_sources): + questions.append( + Question( + qid=f"ki{index:03d}", + kind=QuestionKind.KNOWN_ITEM, + question=f"What is {name} about? Summarize it briefly.", + source=name, + ground_passage=scan.doc_heads.get(name, ""), + ) + ) + terms = sample_terms(scan.passages, count, rng) + hits = count_term_hits(iter_chunks(lancedb_dir), terms) if terms else {} + for index, term in enumerate(terms): + questions.append( + Question( + qid=f"ct{index:03d}", + kind=QuestionKind.COUNT, + question=f"How many documents mention {term}?", + oracle=CountOracle(term=term, chunks=hits[term].chunks, sources=hits[term].sources), + ) + ) + return questions diff --git a/evals/retrieval/report.py b/evals/retrieval/report.py new file mode 100644 index 000000000..603968c00 --- /dev/null +++ b/evals/retrieval/report.py @@ -0,0 +1,57 @@ +"""Markdown rendering of a scored eval run.""" + +from __future__ import annotations + +from typing import Any + +from evals.retrieval.judging import DIMENSIONS +from evals.retrieval.scoring import ResultRowType + + +def _delta_cell(first: float, second: float, noise: float) -> str: + delta = round(second - first, 3) + flag = "" if abs(delta) > noise else " (within noise)" + return f"{delta:+g}{flag}" + + +def render_report(rows: list[dict[str, Any]]) -> str: + """The human-readable report for a results.jsonl produced by score.""" + summary = next((row for row in rows if row["row_type"] == ResultRowType.SUMMARY), None) + if summary is None: + raise ValueError("results contain no summary row; run score first") + arms: dict[str, dict[str, Any]] = summary["arms"] + first, second = list(arms) + noise = summary["noise_floor"] + + lines = [ + "# Retrieval eval report", + "", + f"{summary['judged']} judge-graded questions per arm; the second arm was " + "judged twice under fresh opaque ids to measure the judge's noise floor. " + "Judges saw only question + ground truth + one answer; no arm labels.", + f"Judge noise floor (same answers re-graded blind): plus or minus {noise} " + "per dimension. Deltas at or below it are labeled within noise.", + "", + f"| dimension | {first} | {second} | delta |", + "| --- | --- | --- | --- |", + ] + for dimension in DIMENSIONS: + first_mean = arms[first]["means"][dimension] + second_mean = arms[second]["means"][dimension] + lines.append( + f"| {dimension} (0-2) | {first_mean} | {second_mean} " + f"| {_delta_cell(first_mean, second_mean, noise)} |" + ) + lines += [ + "", + f"| exact-truth check | {first} | {second} |", + "| --- | --- | --- |", + ] + for label, key in ( + ("count questions", "count_pass"), + ("known-item citation", "known_item_pass"), + ): + cells = [f"{arms[arm][key][0]}/{arms[arm][key][1]}" for arm in (first, second)] + lines.append(f"| {label} | {cells[0]} | {cells[1]} |") + lines.append(f"| hard failures | {arms[first]['errors']} | {arms[second]['errors']} |") + return "\n".join(lines) + "\n" diff --git a/evals/retrieval/scoring.py b/evals/retrieval/scoring.py new file mode 100644 index 000000000..1181be7bf --- /dev/null +++ b/evals/retrieval/scoring.py @@ -0,0 +1,130 @@ +"""Scoring: noise-floor math, per-dimension means, and exact-truth checks.""" + +from __future__ import annotations + +import re +import statistics +from enum import StrEnum +from typing import Any + +from evals.retrieval.answers import AnswerRow +from evals.retrieval.judging import DIMENSIONS +from evals.retrieval.questions import CountOracle, Question, QuestionKind + +_NUMBER_RE = re.compile(r"\d+") + +Grades = dict[str, dict[str, int]] +Unblinded = dict[str, dict[int, Grades]] + +ARM_REPLICATE = 0 +NOISE_REPLICATE = 1 + + +class ResultRowType(StrEnum): + QUESTION = "question" + SUMMARY = "summary" + + +def noise_floor(rep0: Grades, rep1: Grades) -> float: + """Mean absolute per-question, per-dimension disagreement between replicates.""" + shared = [qid for qid in rep0 if qid in rep1] + if not shared: + return 0.0 + per_question = [ + sum(abs(rep0[qid][d] - rep1[qid][d]) for d in DIMENSIONS) / len(DIMENSIONS) + for qid in shared + ] + return round(statistics.mean(per_question), 3) + + +def dimension_means(grades: Grades) -> dict[str, float]: + """Per-dimension mean scores over every graded question.""" + if not grades: + return dict.fromkeys(DIMENSIONS, 0.0) + return { + d: round(statistics.mean(scores[d] for scores in grades.values()), 3) for d in DIMENSIONS + } + + +def count_question_pass(oracle: CountOracle, answer: str) -> bool: + """Both oracle numbers must appear in the answer.""" + numbers = set(_NUMBER_RE.findall(answer)) + return {str(oracle.chunks), str(oracle.sources)} <= numbers + + +def known_item_pass(source: str, row: AnswerRow) -> bool: + """The expected document must be among the answer's cited sources.""" + return source in row.cited_sources + + +def _question_row( + question: Question, arm: str, answer: AnswerRow | None, grades: Grades +) -> dict[str, Any]: + row: dict[str, Any] = { + "row_type": ResultRowType.QUESTION, + "qid": question.qid, + "kind": question.kind, + "arm": arm, + "error": answer.error if answer is not None else "missing", + } + if question.kind in (QuestionKind.TOPICAL, QuestionKind.KNOWN_ITEM): + row["grades"] = grades.get(question.qid) + answered = answer is not None and not answer.error + if question.kind is QuestionKind.COUNT and question.oracle is not None: + row["exact_pass"] = answered and count_question_pass(question.oracle, answer.answer) + elif question.kind is QuestionKind.KNOWN_ITEM: + row["exact_pass"] = answered and known_item_pass(question.source, answer) + return row + + +def _arm_summary( + questions: list[Question], answers: dict[str, AnswerRow], grades: Grades +) -> dict[str, Any]: + count_hits = count_total = known_hits = known_total = errors = 0 + for question in questions: + answer = answers.get(question.qid) + if answer is None or answer.error: + errors += 1 + answer = None + if question.kind is QuestionKind.COUNT and question.oracle is not None: + count_total += 1 + count_hits += answer is not None and count_question_pass(question.oracle, answer.answer) + elif question.kind is QuestionKind.KNOWN_ITEM: + known_total += 1 + known_hits += answer is not None and known_item_pass(question.source, answer) + return { + "means": dimension_means(grades), + "count_pass": [count_hits, count_total], + "known_item_pass": [known_hits, known_total], + "errors": errors, + } + + +def build_results( + questions: list[Question], + answers_by_arm: dict[str, dict[str, AnswerRow]], + unblinded: Unblinded, + noise_arm: str, +) -> list[dict[str, Any]]: + """Per-question rows for every arm, then one summary row, results.jsonl shaped.""" + rows: list[dict[str, Any]] = [] + for question in questions: + for arm, answers in answers_by_arm.items(): + arm_grades = unblinded.get(arm, {}).get(ARM_REPLICATE, {}) + rows.append(_question_row(question, arm, answers.get(question.qid), arm_grades)) + noise_replicates = unblinded.get(noise_arm, {}) + summary: dict[str, Any] = { + "row_type": ResultRowType.SUMMARY, + "noise_floor": noise_floor( + noise_replicates.get(ARM_REPLICATE, {}), noise_replicates.get(NOISE_REPLICATE, {}) + ), + "judged": sum( + 1 for q in questions if q.kind in (QuestionKind.TOPICAL, QuestionKind.KNOWN_ITEM) + ), + "arms": { + arm: _arm_summary(questions, answers, unblinded.get(arm, {}).get(ARM_REPLICATE, {})) + for arm, answers in answers_by_arm.items() + }, + } + rows.append(summary) + return rows diff --git a/evals/retrieval/store_scan.py b/evals/retrieval/store_scan.py new file mode 100644 index 000000000..80d960742 --- /dev/null +++ b/evals/retrieval/store_scan.py @@ -0,0 +1,137 @@ +"""Streaming access to a lilbee LanceDB index; nothing materializes the table.""" + +from __future__ import annotations + +import random +from collections.abc import Iterable, Iterator, Sequence +from pathlib import Path +from typing import NamedTuple, TypeVar + +SCAN_BATCH_ROWS = 8192 +HEAD_CHUNK_COUNT = 3 +PASSAGE_OVERSAMPLE = 4 + +T = TypeVar("T") + + +class ChunkRow(NamedTuple): + source: str + chunk: str + chunk_index: int + + +class TermCounts(NamedTuple): + chunks: int + sources: int + + +class PassageScan(NamedTuple): + passages: list[tuple[str, str]] + doc_heads: dict[str, str] + + +def iter_chunks(lancedb_dir: Path) -> Iterator[ChunkRow]: + """Stream (source, chunk, chunk_index) rows batch by batch.""" + import lancedb # heavy: arrow + datafusion + + from lilbee.core.config.defaults import CHUNKS_TABLE + + table = lancedb.connect(str(lancedb_dir)).open_table(CHUNKS_TABLE) + columns = ["source", "chunk", "chunk_index"] + for batch in table.search().select(columns).to_batches(batch_size=SCAN_BATCH_ROWS): + yield from ( + ChunkRow(source, chunk, index) + for source, chunk, index in zip( + batch.column("source").to_pylist(), + batch.column("chunk").to_pylist(), + batch.column("chunk_index").to_pylist(), + strict=True, + ) + ) + + +def iter_source_names(lancedb_dir: Path) -> Iterator[str]: + """Stream indexed source filenames batch by batch.""" + import lancedb # heavy: arrow + datafusion + + from lilbee.core.config.defaults import SOURCES_TABLE + + table = lancedb.connect(str(lancedb_dir)).open_table(SOURCES_TABLE) + for batch in table.search().select(["filename"]).to_batches(batch_size=SCAN_BATCH_ROWS): + yield from batch.column("filename").to_pylist() + + +def reservoir_sample(items: Iterable[T], k: int, rng: random.Random) -> list[T]: + """Uniform sample of up to k items in one streaming pass (Algorithm R).""" + reservoir: list[T] = [] + for seen, item in enumerate(items): + if seen < k: + reservoir.append(item) + continue + slot = rng.randrange(seen + 1) + if slot < k: + reservoir[slot] = item + return reservoir + + +def count_term_hits(rows: Iterable[ChunkRow], terms: Sequence[str]) -> dict[str, TermCounts]: + """Exact chunk and distinct-source hit counts for every term in one pass.""" + needles = {term: term.lower() for term in terms} + chunk_hits = dict.fromkeys(terms, 0) + source_hits: dict[str, set[str]] = {term: set() for term in terms} + for row in rows: + if not row.chunk: + continue + text = row.chunk.lower() + for term, needle in needles.items(): + if needle in text: + chunk_hits[term] += 1 + source_hits[term].add(row.source) + return {term: TermCounts(chunk_hits[term], len(source_hits[term])) for term in terms} + + +def scan_passages_and_heads( + rows: Iterable[ChunkRow], + *, + passage_count: int, + min_passage_chars: int, + head_sources: set[str], + rng: random.Random, +) -> PassageScan: + """One streaming pass: sample candidate passages and collect document heads. + + Passages are reservoir-sampled with oversampling, then deduplicated to at + most one per source, so the pass never holds more than a few hundred rows. + """ + reservoir: list[ChunkRow] = [] + candidate_cap = passage_count * PASSAGE_OVERSAMPLE + heads: dict[str, dict[int, str]] = {source: {} for source in head_sources} + seen = 0 + for row in rows: + if row.source in heads and row.chunk_index < HEAD_CHUNK_COUNT: + heads[row.source][row.chunk_index] = row.chunk + if not row.chunk or len(row.chunk) < min_passage_chars: + continue + if seen < candidate_cap: + reservoir.append(row) + else: + slot = rng.randrange(seen + 1) + if slot < candidate_cap: + reservoir[slot] = row + seen += 1 + + passages: list[tuple[str, str]] = [] + picked_sources: set[str] = set() + for row in reservoir: + if row.source in picked_sources: + continue + picked_sources.add(row.source) + passages.append((row.source, row.chunk)) + if len(passages) == passage_count: + break + doc_heads = { + source: "\n".join(chunk for _, chunk in sorted(indexed.items())) + for source, indexed in heads.items() + if indexed + } + return PassageScan(passages=passages, doc_heads=doc_heads) diff --git a/pyproject.toml b/pyproject.toml index 8e323889a..f11aa5f13 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -177,6 +177,10 @@ select = [ # the ASCII look-alikes would break extraction. allowed-confusables = ["｜", "▁"] +[tool.ruff.lint.isort] +# evals/ lives at the repo root, outside the src roots above. +known-first-party = ["evals"] + [tool.ruff.lint.mccabe] max-complexity = 10 @@ -215,6 +219,10 @@ extend-immutable-calls = ["litestar.params.Body"] "PLR2004", "N802", "N806", "N818", "N817", ] +# Eval harness (and its tests): seeded random for reproducible sampling and +# blinding, nothing cryptographic (S311). +"evals/**/*.py" = ["S311"] +"tests/evals/**/*.py" = ["S311"] # Invokes the trusted `gh` CLI with controlled args to build the compat index. "tools/build_compat_index.py" = ["S603", "S607"] "conftest.py" = ["S101"] diff --git a/tests/evals/__init__.py b/tests/evals/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/evals/test_eval_answers.py b/tests/evals/test_eval_answers.py new file mode 100644 index 000000000..a9e9f6a16 --- /dev/null +++ b/tests/evals/test_eval_answers.py @@ -0,0 +1,120 @@ +"""Answer collection over HTTP: warm-up, retries, and checkpointed resume.""" + +import json + +import httpx +import pytest + +from evals.retrieval.answers import AnswerRow, answer_questions, wait_for_server +from evals.retrieval.checkpoint import load_jsonl +from evals.retrieval.questions import Question, QuestionKind + + +def _question(qid: str = "tq000") -> Question: + return Question(qid=qid, kind=QuestionKind.TOPICAL, question="Where?", source="a.txt") + + +def _client(handler) -> httpx.Client: + return httpx.Client(transport=httpx.MockTransport(handler), base_url="http://test") + + +def test_wait_for_server_returns_once_healthy(): + states = iter([503, 503, 200]) + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/api/health" + return httpx.Response(next(states)) + + wait_for_server("http://test", _client(handler), attempts=5, poll=0) + + +def test_wait_for_server_raises_after_budget(): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("refused") + + with pytest.raises(RuntimeError, match="never became healthy"): + wait_for_server("http://test", _client(handler), attempts=2, poll=0) + + +def _ask_response(answer: str = "By the pier.") -> httpx.Response: + return httpx.Response( + 200, + json={ + "answer": answer, + "sources": [{"source": "a.txt", "content_type": "text", "chunk": "c"}], + "cited_sources": [{"source": "a.txt", "content_type": "text", "chunk": "c"}], + }, + ) + + +def test_answer_questions_records_answer_sources_and_citations(tmp_path): + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200) + assert json.loads(request.content)["question"] == "Where?" + return _ask_response() + + out = tmp_path / "answers.jsonl" + rows = answer_questions( + [_question()], "http://test", "armA", out, retry_delay=0, client=_client(handler) + ) + assert len(rows) == 1 + row = AnswerRow.from_dict(load_jsonl(out)[0]) + assert row.arm == "armA" + assert row.answer == "By the pier." + assert row.sources == ["a.txt"] + assert row.cited_sources == ["a.txt"] + assert row.error is None + + +def test_answer_questions_retries_then_records_the_failure(tmp_path): + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200) + seen.append(request.url.path) + return httpx.Response(500) + + out = tmp_path / "answers.jsonl" + rows = answer_questions( + [_question()], + "http://test", + "armA", + out, + attempts=3, + retry_delay=0, + client=_client(handler), + ) + assert len(seen) == 3 + assert rows[0].error is not None + assert rows[0].answer == "" + + +def test_answer_questions_resumes_past_checkpointed_rows(tmp_path): + out = tmp_path / "answers.jsonl" + done = AnswerRow( + qid="tq000", + arm="armA", + answer="cached", + sources=[], + cited_sources=[], + seconds=0.1, + error=None, + ) + out.write_text(json.dumps(done.to_dict()) + "\n") + asked: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200) + asked.append(json.loads(request.content)["question"]) + return _ask_response("fresh") + + questions = [_question("tq000"), _question("tq001")] + rows = answer_questions( + questions, "http://test", "armA", out, retry_delay=0, client=_client(handler) + ) + assert [row.qid for row in rows] == ["tq001"] + assert asked == ["Where?"] + assert len(load_jsonl(out)) == 2 diff --git a/tests/evals/test_eval_blinding.py b/tests/evals/test_eval_blinding.py new file mode 100644 index 000000000..12e3a1152 --- /dev/null +++ b/tests/evals/test_eval_blinding.py @@ -0,0 +1,121 @@ +"""Blinding: opaque ids, a duplicated noise arm, and mechanical unblinding.""" + +import random + +from evals.retrieval.answers import AnswerRow +from evals.retrieval.blinding import build_blind_rows, unblind +from evals.retrieval.questions import Question, QuestionKind + + +def _question(qid: str, kind: QuestionKind = QuestionKind.TOPICAL) -> Question: + return Question( + qid=qid, + kind=kind, + question="Where did it happen?", + source="a.txt", + ground_passage="ground text", + ) + + +def _answer(qid: str, arm: str, answer: str = "an answer", error: str | None = None) -> AnswerRow: + return AnswerRow( + qid=qid, arm=arm, answer=answer, sources=[], cited_sources=[], seconds=0.1, error=error + ) + + +def _answers(questions, arm): + return {q.qid: _answer(q.qid, arm) for q in questions} + + +def test_noise_arm_is_judged_twice_and_other_arm_once(): + questions = [_question("tq000"), _question("tq001")] + blind = build_blind_rows( + questions, + {"A": _answers(questions, "A"), "B": _answers(questions, "B")}, + noise_arm="B", + rng=random.Random(5), + ) + assert len(blind.rows) == 6 + replicates = [(a.arm, a.replicate) for a in blind.assignments.values()] + assert replicates.count(("A", 0)) == 2 + assert replicates.count(("B", 0)) == 2 + assert replicates.count(("B", 1)) == 2 + + +def test_blind_rows_never_leak_arm_or_qid(): + questions = [_question("tq000")] + blind = build_blind_rows( + questions, + {"A": _answers(questions, "A"), "B": _answers(questions, "B")}, + noise_arm="B", + rng=random.Random(5), + ) + for row in blind.rows: + payload = f"{row.gid}{row.question}{row.ground}{row.answer}" + assert "tq000" not in payload + assert row.gid in blind.assignments + + +def test_count_questions_are_not_blind_judged(): + questions = [_question("ct000", QuestionKind.COUNT)] + blind = build_blind_rows( + questions, + {"A": _answers(questions, "A"), "B": _answers(questions, "B")}, + noise_arm="B", + rng=random.Random(5), + ) + assert blind.rows == [] + assert blind.assignments == {} + + +def test_failed_and_missing_answers_prefail_instead_of_judging(): + questions = [_question("tq000"), _question("tq001")] + answers_a = { + "tq000": _answer("tq000", "A", answer="", error="boom"), + } + blind = build_blind_rows( + questions, + {"A": answers_a, "B": _answers(questions, "B")}, + noise_arm="B", + rng=random.Random(5), + ) + assert len(blind.prefailed) == 2 + assert len(blind.rows) == 4 + for gid in blind.prefailed: + assert blind.assignments[gid].arm == "A" + + +def test_build_blind_rows_is_deterministic_for_a_seed(): + questions = [_question("tq000"), _question("tq001")] + arms = {"A": _answers(questions, "A"), "B": _answers(questions, "B")} + first = build_blind_rows(questions, arms, noise_arm="B", rng=random.Random(9)) + second = build_blind_rows(questions, arms, noise_arm="B", rng=random.Random(9)) + assert [row.gid for row in first.rows] == [row.gid for row in second.rows] + assert first.assignments == second.assignments + + +def test_unblind_regroups_by_arm_replicate_and_qid(): + questions = [_question("tq000")] + blind = build_blind_rows( + questions, + {"A": _answers(questions, "A"), "B": _answers(questions, "B")}, + noise_arm="B", + rng=random.Random(5), + ) + grades = {gid: {"faithfulness": 2, "relevance": 1, "citation": 0} for gid in blind.assignments} + regrouped = unblind(blind.assignments, grades) + assert regrouped["A"][0]["tq000"]["faithfulness"] == 2 + assert regrouped["B"][0]["tq000"]["relevance"] == 1 + assert regrouped["B"][1]["tq000"]["citation"] == 0 + + +def test_unblind_skips_unreturned_grades(): + questions = [_question("tq000")] + blind = build_blind_rows( + questions, + {"A": _answers(questions, "A"), "B": _answers(questions, "B")}, + noise_arm="B", + rng=random.Random(5), + ) + regrouped = unblind(blind.assignments, {}) + assert regrouped["A"][0] == {} diff --git a/tests/evals/test_eval_checkpoint.py b/tests/evals/test_eval_checkpoint.py new file mode 100644 index 000000000..54e437b5b --- /dev/null +++ b/tests/evals/test_eval_checkpoint.py @@ -0,0 +1,40 @@ +"""Checkpointed JSONL writing: interrupted runs resume without redoing work.""" + +import json + +from evals.retrieval.checkpoint import JsonlCheckpoint, load_jsonl + + +def test_load_jsonl_missing_file_returns_empty(tmp_path): + assert load_jsonl(tmp_path / "absent.jsonl") == [] + + +def test_load_jsonl_skips_blank_lines(tmp_path): + path = tmp_path / "rows.jsonl" + path.write_text('{"qid": "a"}\n\n{"qid": "b"}\n') + assert load_jsonl(path) == [{"qid": "a"}, {"qid": "b"}] + + +def test_append_records_and_marks_done(tmp_path): + path = tmp_path / "out.jsonl" + checkpoint = JsonlCheckpoint(path, "qid") + assert "q1" not in checkpoint + checkpoint.append({"qid": "q1", "answer": "x"}) + assert "q1" in checkpoint + assert load_jsonl(path) == [{"qid": "q1", "answer": "x"}] + + +def test_resume_skips_previously_written_keys(tmp_path): + path = tmp_path / "out.jsonl" + path.write_text(json.dumps({"qid": "q1"}) + "\n") + checkpoint = JsonlCheckpoint(path, "qid") + assert "q1" in checkpoint + assert "q2" not in checkpoint + checkpoint.append({"qid": "q2"}) + assert {row["qid"] for row in load_jsonl(path)} == {"q1", "q2"} + + +def test_done_returns_a_copy(tmp_path): + checkpoint = JsonlCheckpoint(tmp_path / "out.jsonl", "qid") + checkpoint.done.add("phantom") + assert "phantom" not in checkpoint diff --git a/tests/evals/test_eval_cli.py b/tests/evals/test_eval_cli.py new file mode 100644 index 000000000..8e1504821 --- /dev/null +++ b/tests/evals/test_eval_cli.py @@ -0,0 +1,173 @@ +"""End-to-end pipeline on synthetic fixtures via the CLI subcommands.""" + +import json + +from evals.retrieval import cli +from evals.retrieval.answers import AnswerRow +from evals.retrieval.checkpoint import load_jsonl +from evals.retrieval.questions import CountOracle, Question, QuestionKind +from evals.retrieval.scoring import ResultRowType + + +def _write_jsonl(path, rows): + path.write_text("".join(json.dumps(row) + "\n" for row in rows)) + + +def _fixture_files(tmp_path): + questions = [ + Question( + qid="tq000", + kind=QuestionKind.TOPICAL, + question="Where was the light?", + source="a.txt", + ground_passage="ground passage", + ), + Question( + qid="ct000", + kind=QuestionKind.COUNT, + question="How many documents mention 'lantern'?", + oracle=CountOracle(term="lantern", chunks=2, sources=1), + ), + ] + questions_path = tmp_path / "questions.jsonl" + _write_jsonl(questions_path, [q.to_dict() for q in questions]) + + def _answer(qid, arm, answer): + return AnswerRow( + qid=qid, + arm=arm, + answer=answer, + sources=["a.txt"], + cited_sources=["a.txt"], + seconds=0.2, + error=None, + ) + + answers_a = tmp_path / "answers-a.jsonl" + _write_jsonl( + answers_a, + [ + _answer("tq000", "old", "On the hill.").to_dict(), + _answer("ct000", "old", "2 chunks in 1 document.").to_dict(), + ], + ) + answers_b = tmp_path / "answers-b.jsonl" + _write_jsonl( + answers_b, + [ + _answer("tq000", "new", "On the headland.").to_dict(), + _answer("ct000", "new", "2 chunks in 1 document.").to_dict(), + ], + ) + return questions_path, answers_a, answers_b + + +def test_judge_score_report_pipeline(tmp_path, monkeypatch): + questions_path, answers_a, answers_b = _fixture_files(tmp_path) + work_dir = tmp_path / "work" + + fixed_grade = '{"faithfulness": 2, "relevance": 2, "citation": 1}' + monkeypatch.setattr(cli, "judge_chat_fn", lambda: lambda _prompt: fixed_grade) + monkeypatch.setattr(cli, "warm_chat", lambda chat: None) + + exit_code = cli.main( + [ + "judge", + "--questions", + str(questions_path), + "--answers-a", + str(answers_a), + "--answers-b", + str(answers_b), + "--work-dir", + str(work_dir), + "--seed", + "3", + ] + ) + assert exit_code == 0 + assert len(load_jsonl(work_dir / "grades.jsonl")) == 3 + gid_map = json.loads((work_dir / "gid_map.json").read_text()) + assert len(gid_map) == 3 + + results_path = tmp_path / "results.jsonl" + exit_code = cli.main( + [ + "score", + "--questions", + str(questions_path), + "--answers-a", + str(answers_a), + "--answers-b", + str(answers_b), + "--work-dir", + str(work_dir), + "--out", + str(results_path), + ] + ) + assert exit_code == 0 + rows = load_jsonl(results_path) + summary = rows[-1] + assert summary["row_type"] == ResultRowType.SUMMARY + assert set(summary["arms"]) == {"old", "new"} + assert summary["noise_floor"] == 0.0 + assert summary["arms"]["new"]["count_pass"] == [1, 1] + + report_path = tmp_path / "report.md" + exit_code = cli.main(["report", "--results", str(results_path), "--out", str(report_path)]) + assert exit_code == 0 + text = report_path.read_text() + assert "old" in text + assert "new" in text + assert "noise floor" in text.lower() + + +def test_judge_rejects_duplicate_arm_labels(tmp_path, monkeypatch): + questions_path, answers_a, _ = _fixture_files(tmp_path) + monkeypatch.setattr(cli, "judge_chat_fn", lambda: lambda _prompt: "{}") + monkeypatch.setattr(cli, "warm_chat", lambda chat: None) + exit_code = cli.main( + [ + "judge", + "--questions", + str(questions_path), + "--answers-a", + str(answers_a), + "--answers-b", + str(answers_a), + "--work-dir", + str(tmp_path / "w"), + ] + ) + assert exit_code == 1 + + +def test_answer_subcommand_writes_checkpointed_answers(tmp_path, monkeypatch): + questions_path, _, _ = _fixture_files(tmp_path) + out = tmp_path / "answers.jsonl" + + import httpx + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/health": + return httpx.Response(200) + return httpx.Response(200, json={"answer": "ok", "sources": [], "cited_sources": []}) + + client = httpx.Client(transport=httpx.MockTransport(handler), base_url="http://test") + monkeypatch.setattr(cli, "make_http_client", lambda: client) + exit_code = cli.main( + [ + "answer", + "--questions", + str(questions_path), + "--base-url", + "http://test", + "--arm", + "old", + "--out", + str(out), + ] + ) + assert exit_code == 0 + assert len(load_jsonl(out)) == 2 diff --git a/tests/evals/test_eval_judging.py b/tests/evals/test_eval_judging.py new file mode 100644 index 000000000..cb5735e84 --- /dev/null +++ b/tests/evals/test_eval_judging.py @@ -0,0 +1,69 @@ +"""Judge plumbing: grade parsing, clamping, and checkpointed judging.""" + +from evals.retrieval.blinding import BlindRow +from evals.retrieval.checkpoint import load_jsonl +from evals.retrieval.judging import judge_rows, parse_grade + + +def _row(gid: str) -> BlindRow: + return BlindRow(gid=gid, question="Where?", source="a.txt", ground="g", answer="a") + + +def test_parse_grade_extracts_json_from_noise(): + text = 'Sure. {"faithfulness": 2, "relevance": 1, "citation": 0} done.' + assert parse_grade(text) == {"faithfulness": 2, "relevance": 1, "citation": 0} + + +def test_parse_grade_clamps_out_of_range_scores(): + text = '{"faithfulness": 9, "relevance": -3, "citation": 1}' + assert parse_grade(text) == {"faithfulness": 2, "relevance": 0, "citation": 1} + + +def test_parse_grade_rejects_missing_dimensions_and_garbage(): + assert parse_grade('{"faithfulness": 2}') is None + assert parse_grade("no json here") is None + assert parse_grade('{"faithfulness": "high", "relevance": 1, "citation": 1}') is None + + +def test_judge_rows_grades_and_checkpoints(tmp_path): + out = tmp_path / "grades.jsonl" + grades = judge_rows( + [_row("g1"), _row("g2")], + lambda _prompt: '{"faithfulness": 2, "relevance": 2, "citation": 1}', + out, + retry_delay=0, + ) + assert set(grades) == {"g1", "g2"} + assert len(load_jsonl(out)) == 2 + + +def test_judge_rows_resumes_and_keeps_existing_grades(tmp_path): + out = tmp_path / "grades.jsonl" + judge_rows( + [_row("g1")], + lambda _prompt: '{"faithfulness": 1, "relevance": 1, "citation": 1}', + out, + retry_delay=0, + ) + judged: list[str] = [] + + def chat(prompt: str) -> str: + judged.append(prompt) + return '{"faithfulness": 2, "relevance": 2, "citation": 2}' + + grades = judge_rows([_row("g1"), _row("g2")], chat, out, retry_delay=0) + assert len(judged) == 1 + assert grades["g1"] == {"faithfulness": 1, "relevance": 1, "citation": 1} + assert grades["g2"] == {"faithfulness": 2, "relevance": 2, "citation": 2} + + +def test_judge_rows_retries_and_skips_unparseable_rows(tmp_path): + calls: list[str] = [] + + def bad_judge(prompt: str) -> str: + calls.append(prompt) + return "not json" + + grades = judge_rows([_row("g1")], bad_judge, tmp_path / "g.jsonl", attempts=2, retry_delay=0) + assert grades == {} + assert len(calls) == 2 diff --git a/tests/evals/test_eval_questions.py b/tests/evals/test_eval_questions.py new file mode 100644 index 000000000..262c8df09 --- /dev/null +++ b/tests/evals/test_eval_questions.py @@ -0,0 +1,128 @@ +"""Question generation: topical authoring, known-item, and count oracles.""" + +import random + +import pytest + +from evals.retrieval import questions as questions_mod +from evals.retrieval.questions import ( + CountOracle, + Question, + QuestionKind, + author_topical, + build_questions, + parse_question, + sample_terms, +) +from evals.retrieval.store_scan import ChunkRow + + +def test_parse_question_takes_first_line_and_strips_quotes(): + assert parse_question('"Where did the fleet anchor?"\nextra') == "Where did the fleet anchor?" + + +@pytest.mark.parametrize("text", ["", "not a question", "Why?", "Too short?"]) +def test_parse_question_rejects_non_questions(text): + assert parse_question(text) is None + + +def test_sample_terms_picks_mid_frequency_terms(): + passages = [(f"s{i}.txt", "common words everywhere") for i in range(10)] + passages += [("s0.txt", "unique zebra sighting"), ("s1.txt", "zebra again here")] + terms = sample_terms(passages, 5, random.Random(1)) + assert "common" not in terms + assert "zebra" in terms + + +def test_sample_terms_respects_count(): + passages = [(f"s{i}.txt", f"word{i}alpha word{i}beta shared") for i in range(20)] + assert len(sample_terms(passages, 3, random.Random(1))) <= 3 + + +def test_author_topical_builds_questions_with_ground_passages(): + picks = [("a.txt", "p" * 500), ("b.txt", "q" * 500)] + calls: list[str] = [] + + def chat(prompt: str) -> str: + calls.append(prompt) + return "What color was the harbor light?" + + authored = author_topical(picks, chat, retry_delay=0) + assert len(authored) == 2 + assert authored[0].kind is QuestionKind.TOPICAL + assert authored[0].source == "a.txt" + assert authored[0].ground_passage == "p" * 500 + assert "p" * 100 in calls[0] + + +def test_author_topical_retries_then_skips_hard_failures(): + attempts: list[int] = [] + + def flaky_chat(prompt: str) -> str: + attempts.append(1) + raise RuntimeError("model busy") + + authored = author_topical([("a.txt", "p" * 500)], flaky_chat, attempts=3, retry_delay=0) + assert authored == [] + assert len(attempts) == 3 + + +def test_author_topical_drops_malformed_questions(): + authored = author_topical([("a.txt", "p" * 500)], lambda _prompt: "no.", retry_delay=0) + assert authored == [] + + +def test_question_round_trips_through_dict(): + question = Question( + qid="ct000", + kind=QuestionKind.COUNT, + question="How many documents mention 'lantern'?", + oracle=CountOracle(term="lantern", chunks=4, sources=2), + ) + assert Question.from_dict(question.to_dict()) == question + + +def test_build_questions_assembles_all_three_kinds(monkeypatch): + rare_words = ["lantern", "beacon", "harbor", "vessel", "anchor", "compass"] + chunk_rows = [ChunkRow(f"doc{i}.txt", f"{rare_words[i]} " + "x" * 450, 0) for i in range(6)] + monkeypatch.setattr(questions_mod, "iter_chunks", lambda _dir: iter(chunk_rows)) + monkeypatch.setattr( + questions_mod, "iter_source_names", lambda _dir: iter(f"doc{i}.txt" for i in range(6)) + ) + + def chat(prompt: str) -> str: + return "Which lantern hung over the pier that night?" + + built = build_questions(lancedb_dir=None, chat=chat, topical=5, known_item=2, count=1, seed=11) + kinds = [q.kind for q in built] + assert kinds.count(QuestionKind.TOPICAL) == 5 + assert kinds.count(QuestionKind.KNOWN_ITEM) == 2 + assert kinds.count(QuestionKind.COUNT) == 1 + count_question = next(q for q in built if q.kind is QuestionKind.COUNT) + assert count_question.oracle is not None + assert count_question.oracle.chunks >= 1 + known = next(q for q in built if q.kind is QuestionKind.KNOWN_ITEM) + assert known.source in known.question + assert known.ground_passage + assert len({q.qid for q in built}) == len(built) + + +def test_build_questions_skips_the_count_scan_when_no_terms_qualify(monkeypatch): + scans: list[int] = [] + + def counted_iter_chunks(_dir): + scans.append(1) + return iter([ChunkRow("doc0.txt", "y" * 450, 0)]) + + monkeypatch.setattr(questions_mod, "iter_chunks", counted_iter_chunks) + monkeypatch.setattr(questions_mod, "iter_source_names", lambda _dir: iter(["doc0.txt"])) + built = build_questions( + lancedb_dir=None, + chat=lambda _p: "Where did the launch dock?", + topical=1, + known_item=1, + count=4, + seed=1, + ) + assert all(q.kind is not QuestionKind.COUNT for q in built) + assert len(scans) == 1 diff --git a/tests/evals/test_eval_report.py b/tests/evals/test_eval_report.py new file mode 100644 index 000000000..fb500af4a --- /dev/null +++ b/tests/evals/test_eval_report.py @@ -0,0 +1,49 @@ +"""Markdown report rendering from results rows.""" + +from evals.retrieval.report import render_report +from evals.retrieval.scoring import ResultRowType + + +def _summary(noise: float = 0.1) -> dict: + return { + "row_type": ResultRowType.SUMMARY, + "noise_floor": noise, + "judged": 4, + "arms": { + "A": { + "means": {"faithfulness": 1.5, "relevance": 1.8, "citation": 1.0}, + "count_pass": [1, 2], + "known_item_pass": [2, 2], + "errors": 0, + }, + "B": { + "means": {"faithfulness": 1.55, "relevance": 1.2, "citation": 1.0}, + "count_pass": [2, 2], + "known_item_pass": [1, 2], + "errors": 1, + }, + }, + } + + +def test_render_report_flags_deltas_within_the_noise_floor(): + report = render_report([_summary()]) + assert "| faithfulness (0-2) | 1.5 | 1.55 | +0.05 (within noise) |" in report + assert "| relevance (0-2) | 1.8 | 1.2 | -0.6 |" in report + + +def test_render_report_includes_exact_truth_and_failures(): + report = render_report([_summary()]) + assert "count questions | 1/2 | 2/2" in report + assert "known-item citation | 2/2 | 1/2" in report + assert "hard failures | 0 | 1" in report + assert "0.1" in report + + +def test_render_report_requires_a_summary_row(): + try: + render_report([{"row_type": ResultRowType.QUESTION}]) + except ValueError as exc: + assert "summary" in str(exc) + else: + raise AssertionError("expected ValueError") diff --git a/tests/evals/test_eval_scoring.py b/tests/evals/test_eval_scoring.py new file mode 100644 index 000000000..69cbc6b70 --- /dev/null +++ b/tests/evals/test_eval_scoring.py @@ -0,0 +1,158 @@ +"""Scoring: noise floor math, per-dimension means, and exact-truth checks.""" + +import pytest + +from evals.retrieval.answers import AnswerRow +from evals.retrieval.questions import CountOracle, Question, QuestionKind +from evals.retrieval.scoring import ( + ResultRowType, + build_results, + count_question_pass, + dimension_means, + known_item_pass, + noise_floor, +) + + +def _grades(f: int, r: int, c: int) -> dict[str, int]: + return {"faithfulness": f, "relevance": r, "citation": c} + + +def test_noise_floor_is_mean_absolute_disagreement_per_dimension(): + rep0 = {"q1": _grades(2, 2, 2), "q2": _grades(1, 1, 1)} + rep1 = {"q1": _grades(2, 1, 0), "q2": _grades(1, 1, 1)} + assert noise_floor(rep0, rep1) == pytest.approx(0.5) + + +def test_noise_floor_ignores_questions_missing_from_either_replicate(): + rep0 = {"q1": _grades(2, 2, 2), "only0": _grades(0, 0, 0)} + rep1 = {"q1": _grades(2, 2, 2), "only1": _grades(2, 2, 2)} + assert noise_floor(rep0, rep1) == 0.0 + + +def test_noise_floor_with_no_shared_questions_is_zero(): + assert noise_floor({}, {}) == 0.0 + + +def test_dimension_means(): + means = dimension_means({"q1": _grades(2, 2, 0), "q2": _grades(0, 1, 1)}) + assert means == {"faithfulness": 1.0, "relevance": 1.5, "citation": 0.5} + assert dimension_means({}) == {"faithfulness": 0.0, "relevance": 0.0, "citation": 0.0} + + +def test_count_question_pass_requires_both_oracle_numbers(): + oracle = CountOracle(term="lantern", chunks=14, sources=3) + assert count_question_pass(oracle, "It appears in 14 chunks across 3 documents.") + assert not count_question_pass(oracle, "It appears in 14 chunks.") + assert not count_question_pass(oracle, "It appears 15 times in 3 documents.") + + +def test_known_item_pass_requires_the_expected_citation(): + row = AnswerRow( + qid="ki000", + arm="A", + answer="It is a book.", + sources=["a.txt", "b.txt"], + cited_sources=["a.txt"], + seconds=0.1, + error=None, + ) + assert known_item_pass("a.txt", row) + assert not known_item_pass("b.txt", row) + + +def _pipeline_fixture(): + questions = [ + Question( + qid="tq000", + kind=QuestionKind.TOPICAL, + question="Where?", + source="a.txt", + ground_passage="ground", + ), + Question( + qid="ki000", + kind=QuestionKind.KNOWN_ITEM, + question="What is a.txt about?", + source="a.txt", + ground_passage="head", + ), + Question( + qid="ct000", + kind=QuestionKind.COUNT, + question="How many documents mention 'lantern'?", + oracle=CountOracle(term="lantern", chunks=2, sources=1), + ), + ] + + def _answer(qid, arm, answer, cited): + return AnswerRow( + qid=qid, + arm=arm, + answer=answer, + sources=cited, + cited_sources=cited, + seconds=0.2, + error=None, + ) + + answers = { + "A": { + "tq000": _answer("tq000", "A", "By the pier.", ["a.txt"]), + "ki000": _answer("ki000", "A", "A harbor log.", ["a.txt"]), + "ct000": _answer("ct000", "A", "2 chunks in 1 document.", []), + }, + "B": { + "tq000": _answer("tq000", "B", "Near the docks.", ["a.txt"]), + "ki000": _answer("ki000", "B", "A harbor log.", ["b.txt"]), + "ct000": _answer("ct000", "B", "About 7 mentions.", []), + }, + } + unblinded = { + "A": {0: {"tq000": _grades(2, 2, 2), "ki000": _grades(2, 2, 2)}}, + "B": { + 0: {"tq000": _grades(1, 1, 1), "ki000": _grades(1, 2, 1)}, + 1: {"tq000": _grades(1, 1, 0), "ki000": _grades(1, 2, 1)}, + }, + } + return questions, answers, unblinded + + +def test_build_results_emits_question_rows_and_a_summary(): + questions, answers, unblinded = _pipeline_fixture() + rows = build_results(questions, answers, unblinded, noise_arm="B") + summary = rows[-1] + assert summary["row_type"] == ResultRowType.SUMMARY + assert summary["noise_floor"] == pytest.approx(1 / 6, abs=1e-3) + arm_a = summary["arms"]["A"] + assert arm_a["means"]["faithfulness"] == 2.0 + assert arm_a["count_pass"] == [1, 1] + assert arm_a["known_item_pass"] == [1, 1] + assert arm_a["errors"] == 0 + arm_b = summary["arms"]["B"] + assert arm_b["count_pass"] == [0, 1] + assert arm_b["known_item_pass"] == [0, 1] + + question_rows = [r for r in rows if r["row_type"] == ResultRowType.QUESTION] + assert len(question_rows) == 6 + topical_a = next(r for r in question_rows if r["qid"] == "tq000" and r["arm"] == "A") + assert topical_a["grades"] == _grades(2, 2, 2) + count_b = next(r for r in question_rows if r["qid"] == "ct000" and r["arm"] == "B") + assert count_b["exact_pass"] is False + + +def test_build_results_counts_hard_failures(): + questions, answers, unblinded = _pipeline_fixture() + answers["B"]["tq000"] = AnswerRow( + qid="tq000", + arm="B", + answer="", + sources=[], + cited_sources=[], + seconds=0.0, + error="ConnectError: refused", + ) + del answers["B"]["ct000"] + rows = build_results(questions, answers, unblinded, noise_arm="B") + summary = rows[-1] + assert summary["arms"]["B"]["errors"] == 2 diff --git a/tests/evals/test_eval_store_scan.py b/tests/evals/test_eval_store_scan.py new file mode 100644 index 000000000..f6e5ab466 --- /dev/null +++ b/tests/evals/test_eval_store_scan.py @@ -0,0 +1,85 @@ +"""Streaming scan helpers: sampling and counting without materializing an index.""" + +import random + +from evals.retrieval.store_scan import ( + ChunkRow, + count_term_hits, + reservoir_sample, + scan_passages_and_heads, +) + + +def _rows(spec: dict[str, list[str]]) -> list[ChunkRow]: + return [ + ChunkRow(source, chunk, index) + for source, chunks in spec.items() + for index, chunk in enumerate(chunks) + ] + + +def test_reservoir_sample_returns_all_when_fewer_than_k(): + assert sorted(reservoir_sample(iter("abc"), 10, random.Random(1))) == ["a", "b", "c"] + + +def test_reservoir_sample_is_deterministic_for_a_seed(): + items = list(range(1000)) + first = reservoir_sample(iter(items), 5, random.Random(7)) + second = reservoir_sample(iter(items), 5, random.Random(7)) + assert first == second + assert len(first) == 5 + assert all(item in items for item in first) + + +def test_reservoir_sample_draws_from_the_whole_stream(): + samples = reservoir_sample(iter(range(10_000)), 50, random.Random(3)) + assert any(item > 5000 for item in samples) + + +def test_count_term_hits_counts_chunks_and_distinct_sources(): + rows = _rows( + { + "a.txt": ["the whale swam", "deep water"], + "b.txt": ["a whale breached", "another Whale sighting"], + "c.txt": ["nothing relevant"], + } + ) + counts = count_term_hits(rows, ["whale", "water", "absent"]) + assert counts["whale"].chunks == 3 + assert counts["whale"].sources == 2 + assert counts["water"] == (1, 1) + assert counts["absent"] == (0, 0) + + +def test_scan_collects_one_passage_per_source_and_respects_min_chars(): + long_a = "a" * 500 + long_b = "b" * 500 + rows = _rows({"a.txt": [long_a, long_a], "b.txt": ["short", long_b]}) + scan = scan_passages_and_heads( + rows, passage_count=5, min_passage_chars=400, head_sources=set(), rng=random.Random(1) + ) + assert sorted(source for source, _ in scan.passages) == ["a.txt", "b.txt"] + assert all(len(passage) >= 400 for _, passage in scan.passages) + + +def test_scan_caps_passages_at_requested_count(): + rows = _rows({f"s{i}.txt": ["x" * 450] for i in range(30)}) + scan = scan_passages_and_heads( + rows, passage_count=4, min_passage_chars=400, head_sources=set(), rng=random.Random(2) + ) + assert len(scan.passages) == 4 + assert len({source for source, _ in scan.passages}) == 4 + + +def test_scan_builds_doc_heads_in_chunk_order(): + rows = [ + ChunkRow("k.txt", "third", 2), + ChunkRow("k.txt", "first", 0), + ChunkRow("k.txt", "second", 1), + ChunkRow("k.txt", "far past the head", 9), + ChunkRow("other.txt", "ignored", 0), + ] + scan = scan_passages_and_heads( + rows, passage_count=0, min_passage_chars=400, head_sources={"k.txt"}, rng=random.Random(1) + ) + assert scan.doc_heads == {"k.txt": "first\nsecond\nthird"}