File size: 3,662 Bytes
750b88d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """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 ""))
|