Write your first eval for an LLM app

[
 {"id": "rollback-command", "question": "How do I roll back production to release r41?", "source": "rollback.md", "must_include": ["docuctl rollback --env production --to r41"]},
 {"id": "migrations", "question": "Does a rollback undo database migrations?", "source": "rollback.md", "must_include": ["not rolled back"]},
 {"id": "revert-replaced", "question": "What replaced `docuctl release revert`?", "source": "rollback.md", "must_include": ["docuctl rollback"]},
 {"id": "token-expiry", "question": "How long is a CI deploy token valid?", "source": "tokens.md", "must_include": ["90 days"]},
 {"id": "config-file", "question": "Which file does docuctl read its configuration from?", "source": "config.md", "must_include": ["ship.toml"]},
 {"id": "prod-approval", "question": "What has to happen before a production deploy goes out?", "source": "deploy.md", "must_include": ["docuctl approve"]},
 {"id": "not-in-docs", "question": "How do I deploy to Kubernetes with docuctl?", "source": null, "must_include": ["not"]}
]

"""Run every eval case three times against the docs assistant and grade each answer. Run: python run_eval.py"""
import json
from pathlib import Path

import ollama
from pydantic import ValidationError

from assistant import NULL_RULE, DocsAnswer, DocsAnswerV2, ask

HERE = Path(__file__).parent
CASES = json.loads((HERE / "cases.json").read_text())


def grade(case: dict, answer: DocsAnswer) -> list[str]:
    """Reasons the answer fails this case; an empty list is a pass."""
    text = " ".join(filter(None, [answer.answer, answer.command])).replace("`", "").lower()
    problems = [f"cited {answer.source}, expected {case['source']}"] if answer.source != case["source"] else []
    problems += [f"missing {phrase!r}" for phrase in case["must_include"] if phrase.lower() not in text]
    return problems


def run_case(client: ollama.Client, case: dict, seed: int, answer_type=DocsAnswer, rules: str = "") -> dict:
    try:
        answer = ask(client, case["question"], answer_type=answer_type, rules=rules, temperature=0.7, seed=seed)
    except ValidationError as error:
        return {"case": case["id"], "seed": seed, "passed": False, "problems": [f"invalid output: {error.errors()[0]['type']}"]}
    problems = grade(case, answer)
    return {"case": case["id"], "seed": seed, "passed": not problems, "problems": problems,
            "answer": answer.model_dump()}


def report(label: str, results: list[dict]) -> dict:
    print(f"--- {label}")
    counts = {}
    for case in CASES:
        rows = [r for r in results if r["case"] == case["id"]]
        counts[case["id"]] = sum(r["passed"] for r in rows)
        problems = sorted({p for r in rows for p in r["problems"]})
        print(f"{case['id']:<17} {counts[case['id']]}/3  {'; '.join(problems)}")
    print(f"passed {sum(counts.values())} of {len(results)} trials\n")
    return counts


def main() -> None:
    client = ollama.Client()
    runs = {}
    versions = [("v1: source is required", DocsAnswer, ""),
                ("v2: source may be null", DocsAnswerV2, ""),
                ("v3: source may be null, prompt says when", DocsAnswerV2, NULL_RULE)]
    for label, answer_type, rules in versions:
        results = [run_case(client, case, seed, answer_type, rules) for case in CASES for seed in (1, 2, 3)]
        runs[label] = {"results": results, "passed": report(label, results)}
    output = HERE / "runs/local-docuctl"
    output.mkdir(parents=True, exist_ok=True)
    (output / "results.json").write_text(json.dumps(runs, indent=1) + "\n")
    baseline, *_, latest = (run["passed"] for run in runs.values())
    print("v3 regressions against v1:", [case for case in baseline if latest[case] < baseline[case]] or "none")


if __name__ == "__main__":
    main()

"""Unit tests for the grader, and checks on the recorded eval run. Run: python -m pytest -q"""
import json
from pathlib import Path

from assistant import DocsAnswer, DocsAnswerV2
from run_eval import CASES, grade

RUNS = json.loads(Path(__file__).with_name("results.json").read_text())
CASE = {c["id"]: c for c in CASES}


def test_a_correct_answer_passes_even_with_backticks_and_case_changes():
    answer = DocsAnswer(answer="Run `DOCUCTL ROLLBACK --env production --to r41`.", source="rollback.md")
    assert grade(CASE["rollback-command"], answer) == []


def test_the_command_field_counts_toward_the_required_phrases():
    answer = DocsAnswer(answer="Use the rollback command.", source="rollback.md", command="docuctl rollback --to <release>")
    assert grade(CASE["revert-replaced"], answer) == []


def test_a_wrong_source_and_a_missing_phrase_are_both_reported():
    answer = DocsAnswer(answer="It was removed in 2.0.", source="deploy.md")
    assert grade(CASE["revert-replaced"], answer) == ["cited deploy.md, expected rollback.md", "missing 'docuctl rollback'"]


def test_only_version_2_can_say_no_page_covers_the_question():
    assert DocsAnswerV2(answer="The docs do not cover it.", source=None).source is None
    assert "null" not in json.dumps(DocsAnswer.model_json_schema()["properties"]["source"])


def test_every_version_ran_every_case_three_times():
    for run in RUNS.values():
        assert sorted(r["case"] for r in run["results"]) == sorted(c["id"] for c in CASES for _ in range(3))


def test_neither_fix_made_the_not_in_docs_case_pass():
    assert [run["passed"]["not-in-docs"] for run in RUNS.values()] == [0, 0, 0]
    assert all(run["passed"]["revert-replaced"] == 0 for run in RUNS.values())


def test_fresh_eval_leaves_original_recording_unchanged(tmp_path, monkeypatch):
    import run_eval as app
    original = tmp_path / "results.json"
    original.write_text("original recording")
    monkeypatch.setattr(app, "HERE", tmp_path)
    monkeypatch.setattr(app.ollama, "Client", lambda: object())
    monkeypatch.setattr(app, "run_case", lambda _client, case, *_args: {"case": case["id"], "passed": False, "problems": []})
    app.main()
    assert original.read_text() == "original recording"
    current = json.loads((tmp_path / "runs/local-docuctl/results.json").read_text())
    assert len(current) == 3
    assert all(len(run["results"]) == 3 * len(app.CASES) for run in current.values())