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..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 @@ -82,3 +83,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, Any]) -> str: + return str(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[str, Any]) -> dict[str, Any]: + 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