backend/evals/run.py

"""Eval runner.

Discovers <id>.gt.json files, runs the IDP pipeline on each paired document,
scores the prediction, and prints a per-type/per-difficulty report. Also writes
backend/evals/report.json and records rows in the metrics DB (mode='eval') so the
dashboard's Evals tab renders the same numbers.

Usage:
  python -m evals.run                    # full suite (configured router)
  python -m evals.run --type invoice     # filter by doc type
  python -m evals.run --policy offline   # force a routing policy
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

# allow `python -m evals.run` from backend/ and `python evals/run.py`
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from app.config import get_settings  # noqa: E402
from app.metrics import MetricsStore  # noqa: E402
from app.pipeline import process_document  # noqa: E402
from app.providers import build_registry  # noqa: E402
from app.router import ModelRouter  # noqa: E402
from evals import scorers  # noqa: E402

DOC_EXTS = (".pdf", ".png", ".jpg", ".jpeg", ".tif", ".tiff")


def discover(dataset_dir: Path, type_filter: str | None) -> list[tuple[Path, dict]]:
    out = []
    for gt_path in sorted(dataset_dir.glob("*.gt.json")):
        gt = json.loads(gt_path.read_text())
        if gt.get("_meta", {}).get("skip_eval"):
            continue  # showcase-only docs (e.g. the complex form) aren't scored here
        if type_filter and gt.get("doc_type") != type_filter:
            continue
        stem = gt_path.name[: -len(".gt.json")]
        doc = None
        for ext in DOC_EXTS:
            cand = dataset_dir / f"{stem}{ext}"
            if cand.exists():
                doc = cand
                break
        if doc is None:
            print(f"  ! no document file for {stem}, skipping")
            continue
        out.append((doc, gt))
    return out


def run_suite(type_filter: str | None = None, policy: str | None = None) -> dict:
    settings = get_settings()
    if policy:
        settings.routing_policy = policy
    registry = build_registry(settings)
    metrics = MetricsStore(settings.metrics_db_path)
    router = ModelRouter(registry, settings, metrics)

    cases = discover(settings.evals_dataset_dir, type_filter)
    if not cases:
        print("No eval cases found. Run: python scripts/generate_samples.py")
        return {}

    results = []
    for doc_path, gt in cases:
        meta = gt.get("_meta", {})
        clean_gt = {k: v for k, v in gt.items() if not k.startswith("_")}
        run = process_document(
            doc_path, router=router, settings=settings, metrics=metrics,
            doc_id=doc_path.stem, channel=meta.get("channel"),
            difficulty=meta.get("difficulty"), mode="eval",
            # let the classifier do its job; do NOT force the type (we score it)
        )
        pred = run["_state"]["extracted"] or {}
        score = scorers.score_document(pred, clean_gt)
        results.append({
            "doc_id": doc_path.stem,
            "predicted_type": run["_state"]["doc_type"],
            "difficulty": meta.get("difficulty", "n/a"),
            "channel": meta.get("channel", "n/a"),
            "confidence": run["_state"]["confidence"],
            "requires_review": run["_state"]["requires_review"],
            "cost_usd": run["total_cost_usd"],
            "score": score,
        })
    agg = scorers.aggregate(results)
    report = {"aggregate": agg, "documents": results,
              "routing_policy": settings.routing_policy,
              "active_tier": registry.capabilities()["active_tier"]}
    # Write to a writable location (committed copy locally, /tmp on serverless).
    for out_path in (settings.eval_report_committed, settings.eval_report_writable):
        try:
            out_path.write_text(json.dumps(report, indent=2))
            break
        except OSError:
            continue
    return report


def _print(report: dict) -> None:
    if not report:
        return
    agg = report["aggregate"]
    o = agg["overall"]
    print("\n" + "=" * 64)
    print(f" IDP EVAL REPORT  (tier={report['active_tier']}, policy={report['routing_policy']})")
    print("=" * 64)
    print(f" documents:            {o['documents']}")
    print(f" doc-type accuracy:    {_pct(o['doc_type_accuracy'])}")
    print(f" field exact-match:    {_pct(o['exact_match'])}")
    print(f" field F1:             {_pct(o['field_f1'])}")
    print(f" line-item F1:         {_pct(o['line_item_f1'])}")
    print(f" financial consistency:{_pct(o['financial_consistency_rate'])}")
    print("-" * 64)
    print(f" {'by type':<18}{'docs':>5}{'exact':>9}{'F1':>9}{'fin-ok':>9}")
    for t, g in agg["by_type"].items():
        print(f" {t:<18}{g['documents']:>5}{_pct(g['exact_match']):>9}"
              f"{_pct(g['field_f1']):>9}{_pct(g['financial_consistency_rate']):>9}")
    print("-" * 64)
    print(f" {'by difficulty':<18}{'docs':>5}{'exact':>9}{'F1':>9}{'fin-ok':>9}")
    for d, g in agg["by_difficulty"].items():
        print(f" {d:<18}{g['documents']:>5}{_pct(g['exact_match']):>9}"
              f"{_pct(g['field_f1']):>9}{_pct(g['financial_consistency_rate']):>9}")
    print("=" * 64)
    print(f" report → backend/evals/report.json\n")


def _pct(v) -> str:
    return "n/a" if v is None else f"{v*100:.1f}%"


def main() -> None:
    ap = argparse.ArgumentParser(description="Run the IDP eval suite.")
    ap.add_argument("--type", dest="type_filter", default=None)
    ap.add_argument("--policy", dest="policy", default=None,
                    choices=["auto", "cheap", "smart", "offline"])
    args = ap.parse_args()
    report = run_suite(args.type_filter, args.policy)
    _print(report)


if __name__ == "__main__":
    main()