From 9ee622820e6923802a8502b4d6a8abe57bad4e9f Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Mon, 14 Sep 2026 17:07:49 +0000 Subject: [PATCH 1/2] Evaluator: survive a failed question, and keep what was already paid for A run is bought question by question -- a rephrase call, a retrieval, a generation, then a judge pass over all of it. The report was written once, at the very end, after every model and every repeat. So a rate limit on question 18 of 20 raised, unwound the loop, and discarded the seventeen answers already purchased. Nothing was on disk. That is the reason the evaluation is unpleasant to run, and the reason it has not been run on the four retrieval changes that are owed one. **A failed question no longer ends the run.** It is named on stderr, listed under `failed_questions` in the report, and the rest are scored without it. **The references are re-aligned to the questions that survived.** This is the part that had to be got right: dropping a question from the middle and leaving the reference list alone would score every later answer against the wrong reference. That does not raise -- it produces numbers, and they look like a result. `_kept` filters the question list by the same indices, and a test pins the pairing. **Answers are written as they are produced.** `--transcript-log` appends one JSON object per answer and flushes, so whatever was bought survives the process that bought it. **`--concurrency` answers several questions at once.** Questions are independent, so this does not change what is measured. Results are placed by index rather than appended as they arrive -- appending would order them by completion time, which is the reference-misalignment bug by another route. Default is 1, so nothing changes unless asked. Six tests, with fakes rather than API calls. Four of them fail against an append-based implementation, which is the mutation worth guarding: order by arrival is plausible, passes casual inspection, and silently scores the wrong pairs. Verified: ruff and ruff format clean (the repo's pinned 0.7.4 -- a newer ruff reports RUF100 here for a rule this repo does not enable), mypy clean, full suite 237 passed. Co-Authored-By: Claude Opus 5 --- src/evaluation/README.md | 29 ++++ src/evaluation/evaluator.py | 249 ++++++++++++++++++++++++++--- tests/evaluation/test_evaluator.py | 124 ++++++++++++++ 3 files changed, 379 insertions(+), 23 deletions(-) diff --git a/src/evaluation/README.md b/src/evaluation/README.md index c1c7fcf..144494f 100644 --- a/src/evaluation/README.md +++ b/src/evaluation/README.md @@ -64,6 +64,35 @@ poetry run ./bin/evaluate --model gpt-4o-mini --repeat 3 --out report.json and every answer with its per-question scores — so a low score can be looked at rather than guessed about. +### Surviving a run that goes wrong + +A run is bought, question by question, and `--out` is written only at the very +end — after every model and every repeat. A rate limit on question 18 of 20 used +to discard the seventeen already paid for. + +```bash +# keep every answer as it is produced +poetry run ./bin/evaluate --model gpt-4o-mini --transcript-log run.jsonl + +# answer four questions at once +poetry run ./bin/evaluate --model gpt-4o-mini --concurrency 4 +``` + +`--transcript-log` appends one JSON object per answer, flushed immediately, so +whatever was bought survives the process that bought it. + +A question that fails no longer ends the run. It is named on stderr, listed +under `failed_questions` in the report, and the remaining questions are scored +without it — with the references re-aligned to the questions that survived. That +alignment is the part worth knowing about: dropping a question from the middle +and *not* re-aligning would score every later answer against the wrong +reference, which produces numbers rather than an error. + +`--concurrency` raises the rate of calls to the provider, so it makes rate +limits more likely — which is survivable now, and was not before. It does not +change what is measured: questions are independent, and results are placed by +index rather than appended as they arrive. + ## What it measures, exactly The chain `create_reactome_rag` builds, asked the **rephrased** question — which diff --git a/src/evaluation/evaluator.py b/src/evaluation/evaluator.py index 876c779..9252d61 100644 --- a/src/evaluation/evaluator.py +++ b/src/evaluation/evaluator.py @@ -30,9 +30,12 @@ import os import statistics import sys +import threading import time +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor from pathlib import Path -from typing import Any +from typing import Any, NamedTuple, cast import nltk from dotenv import load_dotenv @@ -97,10 +100,17 @@ def resolve_references( return [references.get(question) for question in questions] -def answer_questions( - chain: Any, rephrase: Any, questions: list[str] -) -> tuple[list[str], list[str], list[list[str]], float]: - """Rephrase each question as production does, then ask the chain. +class QuestionFailed(NamedTuple): + """A question that could not be answered, kept rather than thrown away.""" + + # Not `index`: on a NamedTuple that shadows tuple.index. + position: int + question: str + error: str + + +def answer_one(chain: Any, rephrase: Any, question: str) -> tuple[str, str, list[str]]: + """Rephrase as production does, then ask the chain. The rephrase is not optional and not cosmetic. `generate_answer` passes `rephrased_input` to the RAG chain, never the raw question, so retrieval in @@ -112,20 +122,139 @@ def answer_questions( the one it claims to measure: the same defect as building a private retriever, one step further up. """ - rephrased: list[str] = [] - answers: list[str] = [] - contexts: list[list[str]] = [] + standalone = rephrase.invoke({"user_input": question, "chat_history": []}) + response = chain.invoke({"input": standalone, "chat_history": []}) + return ( + standalone, + response["answer"], + [doc.page_content for doc in response["context"]], + ) + + +def answer_questions( + chain: Any, + rephrase: Any, + questions: list[str], + *, + concurrency: int = 1, + on_answered: Callable[[int, str, str, list[str]], None] | None = None, +) -> tuple[list[str], list[str], list[list[str]], float, list[QuestionFailed]]: + """Answer every question, keeping what succeeded when something fails. + + Two properties this needs that the serial version did not have. + + **A failure must not cost the whole run.** Every answer here is paid for -- + a rephrase call, a retrieval, a generation -- and the previous version held + all of it in memory until the last question returned, so a rate limit on + question 18 of 20 discarded the seventeen already bought. Failures are now + recorded and reported; the run continues. + + **Order must not depend on timing.** Results are placed by index, not + appended as they arrive, so a concurrent run scores the same questions + against the same references as a serial one. Getting this wrong would not + crash; it would silently score answers against the wrong references, which + is the kind of wrong that reads as a result. + """ + total = len(questions) + rephrased: list[str | None] = [None] * total + answers: list[str | None] = [None] * total + contexts: list[list[str] | None] = [None] * total + failures: list[QuestionFailed] = [] + done = 0 + lock = threading.Lock() started = time.monotonic() - for i, question in enumerate(questions, start=1): - print(f" [{i}/{len(questions)}] {question[:70]}", file=sys.stderr) - standalone = rephrase.invoke({"user_input": question, "chat_history": []}) - if standalone.strip() != question.strip(): - print(f" rephrased: {standalone.strip()[:70]}", file=sys.stderr) - rephrased.append(standalone) - response = chain.invoke({"input": standalone, "chat_history": []}) - answers.append(response["answer"]) - contexts.append([doc.page_content for doc in response["context"]]) - return rephrased, answers, contexts, time.monotonic() - started + + def run(index: int) -> None: + nonlocal done + question = questions[index] + try: + standalone, answer, context = answer_one(chain, rephrase, question) + # Deliberately broad: one bad question must not end the run. + except Exception as exc: + with lock: + done += 1 + failures.append(QuestionFailed(index, question, repr(exc))) + print( + f" [{done}/{total}] FAILED {question[:60]}: {exc!r}", + file=sys.stderr, + ) + return + + rephrased[index] = standalone + answers[index] = answer + contexts[index] = context + with lock: + done += 1 + print(f" [{done}/{total}] {question[:70]}", file=sys.stderr) + if standalone.strip() != question.strip(): + print( + f" rephrased: {standalone.strip()[:70]}", file=sys.stderr + ) + if on_answered is not None: + on_answered(index, standalone, answer, context) + + if concurrency > 1: + with ThreadPoolExecutor(max_workers=concurrency) as pool: + list(pool.map(run, range(total))) + else: + for index in range(total): + run(index) + + elapsed = time.monotonic() - started + + # Drop the failures, keeping the three lists aligned with each other. + kept = [i for i in range(total) if answers[i] is not None] + return ( + [cast(str, rephrased[i]) for i in kept], + [cast(str, answers[i]) for i in kept], + [cast(list[str], contexts[i]) for i in kept], + elapsed, + failures, + ) + + +def _kept(index: int, failures: list[QuestionFailed]) -> bool: + """Whether the question at this index produced an answer.""" + return all(failure.position != index for failure in failures) + + +def make_transcript_writer( + path: Path | None, model: str, run: int, questions: list[str] +) -> Callable[[int, str, str, list[str]], None] | None: + """Append each answer to a JSONL file as it is produced. + + The report is written once, at the end, after every model and every repeat. + That is fine when nothing goes wrong and expensive when something does: a + run that dies on the last question used to leave nothing at all, having paid + for every answer before it. + + One line per answer, flushed immediately, so whatever was bought survives + the process that bought it. + """ + if path is None: + return None + + path.parent.mkdir(parents=True, exist_ok=True) + lock = threading.Lock() + + def write(index: int, rephrased: str, answer: str, context: list[str]) -> None: + record = { + "model": model, + "run": run, + "index": index, + "question": questions[index], + "rephrased": rephrased, + "answer": answer, + "documents_retrieved": len(context), + } + line = json.dumps(record, sort_keys=True) + "\n" + # Serialised and flushed: concurrent workers share this file, and a + # partial line is worse than a missing one. + with lock, path.open("a", encoding="utf-8") as handle: + handle.write(line) + handle.flush() + + return write def build_chain(model: str, embeddings_dir: Path) -> tuple[Any, Any]: @@ -279,8 +408,27 @@ def main() -> None: "--embeddings-dir", type=Path, default=EmbeddingEnvironment.get_dir("reactome") ) parser.add_argument("--out", type=Path, help="Write the full report as JSON.") + parser.add_argument( + "--concurrency", + type=int, + default=1, + help="Answer this many questions at once (default 1). Questions are " + "independent, so this does not change what is measured -- but it does " + "raise the rate of calls to the provider, and a rate limit now costs " + "one question rather than the run.", + ) + parser.add_argument( + "--transcript-log", + type=Path, + help="Append every answer here as it is produced, one JSON object per " + "line. A run that dies partway keeps everything already paid for.", + ) args = parser.parse_args() + if args.concurrency < 1: + raise SystemExit("--concurrency must be at least 1.") + + failed_questions: list[dict[str, Any]] = [] models: list[str] = args.models or ["gpt-4o-mini"] if args.judge_model in models: raise SystemExit( @@ -321,19 +469,62 @@ def main() -> None: chain, rephrase = build_chain(model, embeddings_dir) for run in range(1, args.repeat + 1): print(f" {model} run {run}/{args.repeat}", file=sys.stderr) - rephrased, answers, contexts, elapsed = answer_questions( - chain, rephrase, questions + rephrased, answers, contexts, elapsed, failures = answer_questions( + chain, + rephrase, + questions, + concurrency=args.concurrency, + on_answered=make_transcript_writer( + args.transcript_log, model, run, questions + ), ) - seconds.append(elapsed / len(questions)) + if failures: + # Named, not counted. "3 failed" tells you nothing about + # whether the run is still worth reading. + print( + f" {len(failures)} of {len(questions)} questions failed:", + file=sys.stderr, + ) + for failure in failures: + print( + f" [{failure.position + 1}] {failure.question[:60]} " + f"-> {failure.error}", + file=sys.stderr, + ) + failed_questions.extend( + { + "model": model, + "run": run, + "question": failure.question, + "error": failure.error, + } + for failure in failures + ) + if not answers: + print( + f" every question failed for {model}; skipping scoring", + file=sys.stderr, + ) + continue + # Per answered question, so a run that lost some is still + # comparable on rate rather than on total. + seconds.append(elapsed / len(answers)) # Scored against the rephrased question, because that is what was # retrieved on and answered. Judging the answer against the original # wording would penalise a faithful answer for a rewrite the product # performs deliberately. + # The references have to follow the questions that survived. A + # failure removes a question from the middle of the list, so + # passing the full reference list would score every later answer + # against the wrong reference -- silently, and plausibly. + answered = [ + questions[i] for i in range(len(questions)) if _kept(i, failures) + ] aggregate, per_question = score( rephrased, answers, contexts, - resolve_references(questions, references), + resolve_references(answered, references), args.judge_model, ) runs.append(aggregate) @@ -351,7 +542,7 @@ def main() -> None: "scores": s, } for q, r, a, c, s in zip( - questions, + answered, rephrased, answers, contexts, @@ -366,7 +557,19 @@ def main() -> None: "transcripts": transcripts, } + # In the report, not only on stderr: a scored run with three questions + # missing is a different measurement from a complete one, and whoever reads + # the JSON later will not have the terminal output. + if failed_questions: + report["failed_questions"] = failed_questions + print_report(report) + if failed_questions: + print( + f"\n {len(failed_questions)} question-run(s) failed; scores above " + "cover only what succeeded.", + file=sys.stderr, + ) if args.out: args.out.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n") print(f"\nWrote {args.out}", file=sys.stderr) diff --git a/tests/evaluation/test_evaluator.py b/tests/evaluation/test_evaluator.py index 4b9133b..6c658d0 100644 --- a/tests/evaluation/test_evaluator.py +++ b/tests/evaluation/test_evaluator.py @@ -82,3 +82,127 @@ def test_a_missing_reference_is_none_and_not_empty_text() -> None: assert resolve_references(["q"], {"q": ""}) == [ "" ], "an explicitly empty reference is preserved, not turned into None" + + +class _FakeRephrase: + """Stands in for the rephrase chain; returns the question unchanged.""" + + def invoke(self, payload: dict) -> str: + return payload["user_input"] + + +class _FakeChain: + """A chain that answers, unless the question is one it is told to fail on.""" + + def __init__(self, fail_on: set[str] | None = None) -> None: + self.fail_on = fail_on or set() + + def invoke(self, payload: dict) -> dict: + question = payload["input"] + if question in self.fail_on: + raise RuntimeError(f"boom: {question}") + + class _Doc: + def __init__(self, text: str) -> None: + self.page_content = text + + return {"answer": f"answer to {question}", "context": [_Doc(f"ctx {question}")]} + + +def test_answer_questions_keeps_what_succeeded() -> None: + """One bad question must not discard the answers already paid for.""" + from evaluation.evaluator import answer_questions + + questions = ["q1", "q2", "q3", "q4"] + rephrased, answers, contexts, _elapsed, failures = answer_questions( + _FakeChain(fail_on={"q3"}), _FakeRephrase(), questions + ) + + assert [f.question for f in failures] == ["q3"] + assert answers == ["answer to q1", "answer to q2", "answer to q4"] + assert rephrased == ["q1", "q2", "q4"] + assert len(contexts) == 3 + + +def test_answer_questions_preserves_order_under_concurrency() -> None: + """Results are placed by index, never appended as they arrive. + + Appending would order answers by completion time, so a slow question would + push every later answer onto the wrong reference. That does not crash -- it + produces a plausible score for the wrong pairing. + """ + from evaluation.evaluator import answer_questions + + questions = [f"q{i}" for i in range(12)] + rephrased, answers, _contexts, _elapsed, failures = answer_questions( + _FakeChain(), _FakeRephrase(), questions, concurrency=6 + ) + + assert not failures + assert rephrased == questions + assert answers == [f"answer to {q}" for q in questions] + + +def test_answer_questions_order_holds_when_a_middle_question_fails() -> None: + """The alignment that matters: survivors keep their original order.""" + from evaluation.evaluator import answer_questions + + questions = [f"q{i}" for i in range(10)] + _rephrased, answers, _contexts, _elapsed, failures = answer_questions( + _FakeChain(fail_on={"q2", "q7"}), _FakeRephrase(), questions, concurrency=4 + ) + + assert sorted(f.question for f in failures) == ["q2", "q7"] + expected = [f"answer to q{i}" for i in range(10) if i not in (2, 7)] + assert answers == expected + + +def test_surviving_questions_line_up_with_their_references() -> None: + """A dropped question must not shift every later reference by one. + + This is the failure that reads as a result: scores come out, they are just + computed against the wrong pairing. + """ + from evaluation.evaluator import _kept, answer_questions + + questions = ["q0", "q1", "q2", "q3"] + references = {q: f"ref {q}" for q in questions} + + _rephrased, answers, _contexts, _elapsed, failures = answer_questions( + _FakeChain(fail_on={"q1"}), _FakeRephrase(), questions + ) + answered = [q for i, q in enumerate(questions) if _kept(i, failures)] + resolved = resolve_references(answered, references) + + assert answered == ["q0", "q2", "q3"] + assert resolved == ["ref q0", "ref q2", "ref q3"] + assert len(resolved) == len(answers) + + +def test_transcript_log_survives_the_run_that_wrote_it(tmp_path: Path) -> None: + """Every answer is on disk as it is produced, not held until the end.""" + import json + + from evaluation.evaluator import answer_questions, make_transcript_writer + + log = tmp_path / "nested" / "transcript.jsonl" + questions = ["q0", "q1", "q2"] + answer_questions( + _FakeChain(fail_on={"q1"}), + _FakeRephrase(), + questions, + concurrency=3, + on_answered=make_transcript_writer(log, "test-model", 1, questions), + ) + + records = [json.loads(line) for line in log.read_text().splitlines()] + assert {r["question"] for r in records} == {"q0", "q2"} + assert all(r["model"] == "test-model" for r in records) + # Written per answer, so a crash keeps what was already bought. + assert len(records) == 2 + + +def test_no_transcript_writer_when_no_path_given() -> None: + from evaluation.evaluator import make_transcript_writer + + assert make_transcript_writer(None, "m", 1, []) is None From d3d82fd335e0646993e1539b2866cf106e999144 Mon Sep 17 00:00:00 2001 From: Adam Wright Date: Mon, 14 Sep 2026 17:32:57 +0000 Subject: [PATCH 2/2] Type the test fakes so mypy sees them CI runs `mypy` over the whole tree; I had only run it against src/evaluation/evaluator.py, so the fakes in the new tests went unchecked and the lint job went red on a file I had just added. `_FakeRephrase.invoke` returned `payload["user_input"]` -- `Any` out of a bare `dict` -- from a function annotated `-> str`. Both fakes now take `dict[str, Any]` and say what they return. Run `poetry run ruff check . && poetry run ruff format --check . && poetry run mypy` before pushing; a subset is not the gate. Co-Authored-By: Claude Opus 5 --- tests/evaluation/test_evaluator.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/evaluation/test_evaluator.py b/tests/evaluation/test_evaluator.py index 6c658d0..0408837 100644 --- a/tests/evaluation/test_evaluator.py +++ b/tests/evaluation/test_evaluator.py @@ -6,6 +6,7 @@ """ from pathlib import Path +from typing import Any import pytest @@ -87,8 +88,8 @@ def test_a_missing_reference_is_none_and_not_empty_text() -> None: class _FakeRephrase: """Stands in for the rephrase chain; returns the question unchanged.""" - def invoke(self, payload: dict) -> str: - return payload["user_input"] + def invoke(self, payload: dict[str, Any]) -> str: + return str(payload["user_input"]) class _FakeChain: @@ -97,7 +98,7 @@ class _FakeChain: def __init__(self, fail_on: set[str] | None = None) -> None: self.fail_on = fail_on or set() - def invoke(self, payload: dict) -> dict: + def invoke(self, payload: dict[str, Any]) -> dict[str, Any]: question = payload["input"] if question in self.fail_on: raise RuntimeError(f"boom: {question}")