Add a blind, noise-calibrated retrieval eval harness

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/

evals/__init__.py

@@ -0,0 +1 @@
+"""Version-controlled evaluation harnesses; never shipped with the package."""

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 |

evals/retrieval/__init__.py

@@ -0,0 +1 @@
+"""Blind, noise-calibrated A/B evaluation of lilbee retrieval quality."""

evals/retrieval/__main__.py

@@ -0,0 +1,6 @@
+"""Runs the retrieval eval CLI: python -m evals.retrieval <subcommand>."""
+
+from evals.retrieval.cli import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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

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