src/eval/run_eval.py

"""Stage 10: run the golden eval set against the live pipeline.

For each query in eval/golden_set.jsonl:
  - Recall@K: did any expected_book / expected_page show up in top-K retrieval?
  - Coverage (COMPARISON): fraction of expected_books that appear in the answer.
  - must_mention: do we observe each required phrase in the cited passages?
  - hallucination: count of `ungrounded_claims` in the response.

Writes results to eval/results_{ISO_TIMESTAMP}.json so each run is its own row
of history. Always commit results — the user uses them as a changelog.

Run:
    python -m src.eval.run_eval
    python -m src.eval.run_eval --no-llm    # retrieval-only (skip Stage 8 calls)
    python -m src.eval.run_eval --top-k 20
"""

from __future__ import annotations

import argparse
import json
from datetime import datetime, timezone
from pathlib import Path

from src.config import load_config


def _load_golden(path: Path) -> list[dict]:
    rows: list[dict] = []
    with path.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                rows.append(json.loads(line))
    return rows


def _check_recall(retrieved_books: set[str], retrieved_pages: set[tuple[str, int]], expected: dict) -> dict:
    exp_books = set(expected.get("expected_books") or [])
    exp_pages = {(b, p) for b in exp_books for p in (expected.get("expected_pages") or [])}
    book_hit = bool(exp_books & retrieved_books) if exp_books else None
    page_hit = bool(exp_pages & retrieved_pages) if exp_pages else None
    return {"book_recall": book_hit, "page_recall": page_hit}


def _check_must_mention(retrieved_text: str, must: list[str] | None) -> dict:
    if not must:
        return {"must_mention_present": None, "missing": []}
    missing = [m for m in must if m not in retrieved_text]
    return {"must_mention_present": len(missing) == 0, "missing": missing}


def run_eval(no_llm: bool = False, top_k: int = 10) -> dict:
    cfg = load_config()
    golden_path = Path(cfg.section("eval").get("golden_set", "eval/golden_set.jsonl"))
    if not golden_path.is_absolute():
        golden_path = Path.cwd() / golden_path
    rows = _load_golden(golden_path)

    # Lazy imports so --no-llm works even if the router LLM key isn't set.
    from src.stage7_retrieval.retrieve import retrieve

    per_query: list[dict] = []
    for q in rows:
        hits = retrieve(q["query"], top_k=top_k)
        retrieved_books = {h.book_id for h in hits}
        retrieved_pages = {(h.book_id, h.page_number) for h in hits}
        retrieved_text = "\n".join(h.text for h in hits)

        record: dict = {
            "query_id": q["query_id"],
            "query": q["query"],
            "expected_books": q.get("expected_books") or [],
            "retrieval": {
                "k": top_k,
                "retrieved_books": sorted(retrieved_books),
                **_check_recall(retrieved_books, retrieved_pages, q),
                **_check_must_mention(retrieved_text, q.get("must_mention")),
            },
        }

        if not no_llm:
            from src.stage8_router.router import answer

            try:
                resp = answer(q["query"], force_mode=q.get("mode"))
                consulted = set(resp.get("books_consulted") or [])
                exp_books = set(q.get("expected_books") or [])
                coverage = (
                    (len(consulted & exp_books) / len(exp_books)) if exp_books else None
                )
                record["generation"] = {
                    "mode": resp.get("mode"),
                    "coverage": coverage,
                    "ungrounded_claims_count": len(resp.get("ungrounded_claims") or []),
                    "confidence": resp.get("confidence"),
                }
            except Exception as e:  # noqa: BLE001
                record["generation"] = {"error": f"{type(e).__name__}: {e}"}

        per_query.append(record)

    # Aggregate
    n = len(per_query)
    book_recall = sum(1 for r in per_query if r["retrieval"]["book_recall"]) / n if n else 0
    page_recall_eligible = [r for r in per_query if r["retrieval"]["page_recall"] is not None]
    page_recall = (
        sum(1 for r in page_recall_eligible if r["retrieval"]["page_recall"])
        / len(page_recall_eligible)
    ) if page_recall_eligible else None
    summary = {
        "n_queries": n,
        "book_recall_at_k": book_recall,
        "page_recall_at_k": page_recall,
        "k": top_k,
    }

    out = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "summary": summary,
        "per_query": per_query,
    }
    out_path = cfg.paths.eval_dir / f"results_{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}.json"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"\nResults: {out_path}")
    print(json.dumps(summary, indent=2))
    return out


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Run the eval harness")
    ap.add_argument("--no-llm", action="store_true", help="retrieval only; skip Stage 8 LLM calls")
    ap.add_argument("--top-k", type=int, default=10)
    args = ap.parse_args(argv)
    run_eval(no_llm=args.no_llm, top_k=args.top_k)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())