"""RAG eval harness for the /narrative explanatory layer. Reads `backend/tests/fixtures/narrative_golden_set.json`, runs each case through `rag_narrative.generate_narrative()` with `_call_openai_json` mocked to return the case's prepared LLM response, and scores aggregate metrics. Metrics: - kind_agreement_rate: result.kind == expected.kind - reason_agreement_rate: (result.reason or null) == (expected.reason or null) - citation_groundedness_rate: among happy_path cases, fraction returning `kind=narrative` AND all citations validate. - gate_correctness_rate: among low_context cases, fraction returning `kind=low_confidence` with the right reason. - hallucination_rejection_rate: among hallucinated_citation cases, fraction returning `kind=unavailable, reason=citation-hallucinated`. Writes a summary JSON to `factory/artifacts/RAG_EVAL_RESULT.json` so the result is committable and reviewable in PRs that touch rag_narrative.py. Run: python -m backend.scripts.run_rag_eval [--out PATH] Returns non-zero exit code if any baseline gate fails (kind_agreement < 1.0, gate_correctness < 1.0, hallucination_rejection < 1.0). Those gates are load-bearing — any regression in them is a real bug in the validation layer. """ from __future__ import annotations import argparse import json import sys from collections import Counter from pathlib import Path from typing import Any from unittest.mock import patch REPO_ROOT = Path(__file__).resolve().parents[3] GOLDEN_SET_PATH = REPO_ROOT / "backend" / "tests" / "fixtures" / "narrative_golden_set.json" DEFAULT_OUT_PATH = REPO_ROOT / "factory" / "artifacts" / "RAG_EVAL_RESULT.json" def _load_golden_set(path: Path) -> list[dict]: data = json.loads(path.read_text()) cases = data.get("cases", []) if not cases: raise ValueError(f"empty golden set at {path}") return cases def _build_context(spec: dict): """Materialize a NarrativeContext from a golden-set spec dict.""" from backend import rag_narrative return rag_narrative.NarrativeContext( queryFingerprint=spec["queryFingerprint"], trackId=spec["trackId"], title=spec["title"], artist=spec.get("artist"), queryWindow=tuple(spec["queryWindow"]), matchWindow=tuple(spec["matchWindow"]), rawCosine=float(spec["rawCosine"]), criteria=[ rag_narrative.CriterionContext(**c) for c in spec["criteria"] ], acrcloudCoverSongId=spec.get("acrcloudCoverSongId"), ) def _run_case(case: dict) -> dict: """Run one golden-set case end-to-end and capture the result vs expected. Returns a per-case report row. """ from backend import rag_narrative mocked = case["mocked_llm_response"] expected_kind = case["expected_kind"] expected_reason = case.get("expected_reason") must_not_call_llm = mocked == "MUST_NOT_BE_CALLED" # Sentinel "MUST_NOT_BE_CALLED" lives only in the JSON for readability; # the patched helper still needs a real return value if called. We use # None so a wrongful call surfaces as openai-error and the case will # fail the kind-agreement assertion, surfacing the bug. patch_return = None if must_not_call_llm else mocked context = _build_context(case["context"]) with patch( "backend.rag_narrative._call_openai_json", return_value=patch_return ) as call_mock: result = rag_narrative.generate_narrative( context, case["mode"], model_sha="eval-model-sha", catalog_sha="eval-catalog-sha", ) llm_was_called = call_mock.called actual_kind = getattr(result, "kind", None) or ( result.get("kind") if isinstance(result, dict) else None ) actual_reason = getattr(result, "reason", None) or ( result.get("reason") if isinstance(result, dict) else None ) return { "name": case["name"], "category": case["category"], "mode": case["mode"], "expected_kind": expected_kind, "actual_kind": actual_kind, "expected_reason": expected_reason, "actual_reason": actual_reason, "llm_was_called": llm_was_called, "must_not_call_llm": must_not_call_llm, "kind_match": actual_kind == expected_kind, "reason_match": (actual_reason or None) == (expected_reason or None), "gate_respected": (not must_not_call_llm) or (not llm_was_called), } def _aggregate(rows: list[dict]) -> dict: n = len(rows) kind_correct = sum(1 for r in rows if r["kind_match"]) reason_correct = sum(1 for r in rows if r["reason_match"]) by_cat = {} cat_total: Counter[str] = Counter() cat_kind_correct: Counter[str] = Counter() cat_reason_correct: Counter[str] = Counter() cat_gate_respected: Counter[str] = Counter() for r in rows: cat_total[r["category"]] += 1 if r["kind_match"]: cat_kind_correct[r["category"]] += 1 if r["reason_match"]: cat_reason_correct[r["category"]] += 1 if r["gate_respected"]: cat_gate_respected[r["category"]] += 1 for cat, total in cat_total.items(): by_cat[cat] = { "total": total, "kind_agreement_rate": round(cat_kind_correct[cat] / total, 4), "reason_agreement_rate": round(cat_reason_correct[cat] / total, 4), "gate_respected_rate": round(cat_gate_respected[cat] / total, 4), } return { "n_cases": n, "kind_agreement_rate": round(kind_correct / n, 4) if n else 0.0, "reason_agreement_rate": round(reason_correct / n, 4) if n else 0.0, "by_category": by_cat, "baseline_gates": { # Each gate is "must be 1.0 to pass" — any regression here is a # real bug in validation / gating, not noise. "happy_path_kind_agreement": by_cat.get("happy_path", {}).get("kind_agreement_rate", 0.0), "low_context_gate_correctness": by_cat.get("low_context", {}).get("reason_agreement_rate", 0.0), "hallucination_rejection": by_cat.get("hallucinated_citation", {}).get("reason_agreement_rate", 0.0), "malformed_rejection": by_cat.get("malformed_output", {}).get("reason_agreement_rate", 0.0), "openai_error_handling": by_cat.get("openai_error", {}).get("reason_agreement_rate", 0.0), }, } def run_eval(golden_set_path: Path = GOLDEN_SET_PATH) -> dict: """Public entry point — runs the eval and returns the aggregate dict. Pure function from disk path → results dict. The CLI wrapper writes the results to disk; the pytest gate calls this directly and asserts. """ cases = _load_golden_set(golden_set_path) rows = [_run_case(c) for c in cases] summary = _aggregate(rows) return {"summary": summary, "rows": rows} def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--out", type=Path, default=DEFAULT_OUT_PATH, help="Where to write the eval result JSON.", ) parser.add_argument( "--golden", type=Path, default=GOLDEN_SET_PATH, help="Override golden-set path (for experiments).", ) args = parser.parse_args() result = run_eval(args.golden) args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(json.dumps(result, indent=2) + "\n") summary = result["summary"] print(f"RAG eval — {summary['n_cases']} cases") print(f" kind agreement : {summary['kind_agreement_rate']:.2%}") print(f" reason agreement : {summary['reason_agreement_rate']:.2%}") print() print(" Baseline gates (each MUST be 1.0):") failed: list[str] = [] for gate, score in summary["baseline_gates"].items(): status = "OK " if score >= 1.0 else "FAIL" print(f" [{status}] {gate}: {score:.2%}") if score < 1.0: failed.append(gate) print() print(" By category:") for cat, stats in summary["by_category"].items(): print( f" {cat:<26} n={stats['total']:>2} kind={stats['kind_agreement_rate']:.2%} " f"reason={stats['reason_agreement_rate']:.2%} gate_respected={stats['gate_respected_rate']:.2%}" ) print() print(f"Result written to {args.out}") if failed: print(f"FAILED gates: {', '.join(failed)}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())