"""Score phase: compute all metrics and the threshold sweep from the cache. Purely offline -- it loads cached predictions (``eval.cache``), runs the pure aggregations (``eval.metrics``), and formats human-readable tables. Re-running with a different threshold grid or comparison never re-runs inference, which is the whole point of the two-phase split. The formatted output covers the four things the methodology asks for (data spec section 6): per-field precision/recall/F1, per-critical-field metrics, document routing stats, and the threshold sweep trade-off curve. It also prints the confidence distribution (so a flat sweep is explained by the backend exposing no per-field confidence) and, as analysis only, the lowest threshold that reaches a target auto-accept precision -- the operator still chooses the value. """ from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any from eval.cache import DEFAULT_CACHE_BASE, read_entries from eval.metrics import ( CRITICAL_FIELDS, THRESHOLDS, FieldMetric, SweepRow, compute_field_metrics, confidence_histogram, smallest_threshold_meeting, sweep_thresholds, ) from eval.normalize import is_present # Auto-accept precision target on critical fields (data spec section 6). TARGET_CRITICAL_PRECISION: float = 0.98 @dataclass(frozen=True) class ScoreReport: """Everything the score phase computed for one dataset slice.""" dataset: str n: int labeled_fields: tuple[str, ...] critical_labeled: tuple[str, ...] field_metrics: list[FieldMetric] sweep: list[SweepRow] confidence_hist: dict[float, int] n_error: int def _labeled_fields(entries: list[dict[str, Any]]) -> tuple[str, ...]: """Union of the ``labeled_fields`` recorded across cached entries.""" seen: list[str] = [] for entry in entries: for field in entry.get("labeled_fields", []): if field not in seen: seen.append(field) return tuple(seen) def _critical_labeled(labeled: tuple[str, ...], entries: list[dict[str, Any]]) -> tuple[str, ...]: """Critical fields the dataset actually labels *and* has gold present for. A critical field with no gold anywhere in the slice cannot be scored, so it is excluded from the critical-precision denominators. """ result: list[str] = [] for field in CRITICAL_FIELDS: if field not in labeled: continue if any(is_present(field, entry.get("gold", {}).get(field)) for entry in entries): result.append(field) return tuple(result) def build_report( dataset: str, *, cache_base: Path = DEFAULT_CACHE_BASE, thresholds: tuple[float, ...] = THRESHOLDS, ) -> ScoreReport: """Load the cache for a dataset and compute the full score report. Args: dataset: Dataset name whose cache to score. cache_base: Root cache directory. Defaults to ``eval/cache``. thresholds: The threshold grid to sweep. Returns: A :class:`ScoreReport`. Raises: FileNotFoundError: If no cached entries exist for the dataset. """ entries = read_entries(cache_base, dataset) if not entries: raise FileNotFoundError( f"No cached predictions for dataset {dataset!r} under {cache_base}. " "Run the predict phase first." ) labeled = _labeled_fields(entries) critical_labeled = _critical_labeled(labeled, entries) field_metrics = compute_field_metrics(entries, labeled) sweep = sweep_thresholds(entries, critical_labeled, thresholds) hist = confidence_histogram(entries) n_error = sum(1 for entry in entries if entry.get("error")) return ScoreReport( dataset=dataset, n=len(entries), labeled_fields=labeled, critical_labeled=critical_labeled, field_metrics=field_metrics, sweep=sweep, confidence_hist=hist, n_error=n_error, ) # --- Formatting ---------------------------------------------------------------- def _pct(value: float | None) -> str: """Format an optional ratio as a percentage, or 'n/a' when undefined.""" return " n/a" if value is None else f"{value * 100:5.1f}%" def _format_field_table(report: ScoreReport) -> list[str]: lines = [ "Per-field metrics (whole slice):", f" {'field':<16} {'P':>7} {'R':>7} {'F1':>7} {'pred':>4} {'gold':>4} {'ok':>4}", ] for metric in report.field_metrics: marker = " *" if metric.field in report.critical_labeled else " " lines.append( f"{marker}{metric.field:<16} {_pct(metric.precision)} {_pct(metric.recall)} " f"{_pct(metric.f1)} {metric.n_pred:>4} {metric.n_gold:>4} {metric.n_match:>4}" ) lines.append(" (* = critical field)") return lines def _format_sweep_table(report: ScoreReport, coarse_step: int = 5) -> list[str]: lines = [ "Threshold sweep (critical fields on the auto-accepted subset):", f" {'thr':>5} {'accept':>7} {'accept%':>8} {'crit P':>8} {'crit R':>8}", ] for index, row in enumerate(report.sweep): # Print a coarse grid plus the final threshold to keep it readable. is_grid = index % coarse_step == 0 or index == len(report.sweep) - 1 if not is_grid: continue lines.append( f" {row.threshold:>5.2f} {row.n_accepted:>7} {row.accept_rate * 100:>7.1f}% " f"{_pct(row.crit_precision)} {_pct(row.crit_recall)}" ) return lines def _format_confidence(report: ScoreReport) -> list[str]: lines = ["Confidence distribution (cached scores):"] for value, count in report.confidence_hist.items(): bar = "#" * count lines.append(f" {value:>5.2f} {count:>3} {bar}") return lines def _format_routing(report: ScoreReport) -> list[str]: target = smallest_threshold_meeting(report.sweep, TARGET_CRITICAL_PRECISION) lines = [ "Operating point analysis (you choose the threshold):", f" Target: auto-accept precision on critical fields " f"{report.critical_labeled or '(none labeled)'} >= " f"{TARGET_CRITICAL_PRECISION:.2f}", ] if not report.critical_labeled: lines.append( " This dataset labels none of total/tax/invoice_number, so critical " "auto-accept precision cannot be measured here." ) elif target is None: lines.append( " No threshold in the sweep reaches the target with any " "auto-accepted document (see the confidence distribution above)." ) else: lines.append( f" Lowest qualifying threshold: {target.threshold:.2f} " f"(accept {target.n_accepted}/{target.n_total} = " f"{target.accept_rate * 100:.1f}%, crit P {_pct(target.crit_precision)}, " f"crit R {_pct(target.crit_recall)})." ) return lines def format_report(report: ScoreReport) -> str: """Render a :class:`ScoreReport` as a plain-text report. Args: report: The computed score report. Returns: A multi-line string ready to print. """ header = [ "=" * 68, f"Evaluation: {report.dataset} (n={report.n}, errors={report.n_error})", f"Labeled fields: {', '.join(report.labeled_fields) or '(none)'}", "=" * 68, ] sections = [ header, _format_field_table(report), _format_confidence(report), _format_sweep_table(report), _format_routing(report), ] return "\n\n".join("\n".join(section) for section in sections)