| """Evidence diagnostics and report.json aggregation.""" |
| from __future__ import annotations |
|
|
| from collections import defaultdict |
|
|
| MEMORY_LOG_MAX_CHARS = 2_000_000 |
|
|
|
|
| def evidence_diagnostic(q, mem_visits: list[int]) -> dict: |
| """Whether the evidence was inside the memory block, separating a retrieval failure from a reasoning failure.""" |
| if mem_visits is None: |
| return {"evidence_total": len(q.evidence_visits), |
| "evidence_reachable": None, |
| "evidence_found": None, |
| "evidence_truncated": None} |
| reachable = [v for v in q.evidence_visits if v <= q.ckpt_visit] |
| found = set(reachable) & set(mem_visits) |
| return {"evidence_total": len(q.evidence_visits), |
| "evidence_reachable": len(reachable), |
| "evidence_found": len(found), |
| "evidence_truncated": bool(reachable) and not found} |
|
|
|
|
| def build_report(rows: list[dict], meta: dict, usage_totals: dict) -> dict: |
| """Aggregate graded rows into report.json.""" |
| def table(key): |
| agg = defaultdict(lambda: [0, 0]) |
| for r in rows: |
| agg[r[key]][1] += 1 |
| agg[r[key]][0] += int(r["pass"]) |
| return {k: {"pass": a[0], "n": a[1], "acc": round(a[0] / a[1], 4)} |
| for k, a in sorted(agg.items())} |
|
|
| total = len(rows) |
| passed = sum(int(r["pass"]) for r in rows) |
| infra = sum(1 for r in rows if r.get("verdict") in ("error", "?")) |
| scored = total - infra |
| scored_pass = sum(int(r["pass"]) for r in rows if r.get("verdict") not in ("error", "?")) |
| reachable = sum(r.get("evidence_reachable") or 0 for r in rows) |
| found = sum(r.get("evidence_found") or 0 for r in rows) |
| return { |
| **meta, |
| "total": total, "pass": passed, "accuracy": round(passed / total, 4) if total else None, |
| "scored": scored, "infra_failures": infra, |
| "accuracy_excl_infra": round(scored_pass / scored, 4) if scored else None, |
| "by_axis": table("axis"), |
| "by_answer_type": table("answer_type"), |
| "by_grounding": table("grounding"), |
| "evidence_truncated_count": sum(int(bool(r.get("evidence_truncated"))) |
| for r in rows if r.get("evidence_truncated") is not None), |
| "evidence_recall": round(found / reachable, 4) if reachable else None, |
| "context_truncated_count": sum(int(r["context_truncated"]) for r in rows), |
| "errors": sum(1 for r in rows if r["verdict"] == "error"), |
| "judge_failures": sum(1 for r in rows if r["verdict"] == "?"), |
| "llm_usage": usage_totals, |
| } |
|
|
|
|
| def print_report(rep: dict) -> None: |
| print(f"\n{rep['method']} | answer={rep['answer_model']} judge={rep['judge_model']} " |
| f"| patient={rep['patient']}") |
| print(f"{'axis':28} {'pass/n':>10} acc") |
| for ax, t in rep["by_axis"].items(): |
| print(f"{ax:28} {t['pass']}/{t['n']:>8} {t['acc']:.2f}") |
| print(f"{'TOTAL':28} {rep['pass']}/{rep['total']:>8} {rep['accuracy']:.2f}") |
| if rep.get("infra_failures"): |
| acc2 = rep.get("accuracy_excl_infra") |
| shown = "n/a" if acc2 is None else f"{acc2:.2f}" |
| print(f"{'TOTAL excl. infra':28} {rep['pass']}/{rep['scored']:>8} {shown}" |
| f" ({rep['infra_failures']} infra failure(s) dropped from both sides)") |
| rec = rep.get("evidence_recall") |
| jf = rep.get("judge_failures", 0) |
| print(f"evidence_truncated: {rep['evidence_truncated_count']}, " |
| f"evidence_recall: {'n/a' if rec is None else f'{rec:.2f}'}, " |
| f"context_truncated: {rep['context_truncated_count']}, errors: {rep['errors']}" |
| + (f", judge_failures: {jf}" if jf else "")) |
|
|
