Evaluation Harness

{
  "document_id": "doc_demo_a3b4c5d6e7f8",
  "doc_class": "master_service_agreement",
  "fields": {
    "parties": ["Acme Corp", "Customer Inc"],
    "effective_date": "2024-03-15",
    "term_length_months": 12,
    "renewal_terms": {
      "auto_renew": true,
      "term_length_months": 12,
      "notice_period_days": 60
    },
    "governing_law": "Delaware"
  }
}

from __future__ import annotations
from datetime import date
from typing import Any


def normalize_str(v: Any) -> str:
    return str(v).strip().lower() if v else ""


def match_string(extracted: Any, gold: Any) -> bool:
    return normalize_str(extracted) == normalize_str(gold)


def match_date(extracted: Any, gold: Any, tolerance_days: int = 1) -> bool:
    try:
        if not extracted or not gold: return False
        e = date.fromisoformat(extracted)
        g = date.fromisoformat(gold)
        return abs((e - g).days) <= tolerance_days
    except (ValueError, TypeError):
        return False


def match_int(extracted: Any, gold: Any, tolerance: int = 0) -> bool:
    try:
        return abs(int(extracted) - int(gold)) <= tolerance
    except (ValueError, TypeError):
        return False


def match_renewal_terms(extracted: dict | None, gold: dict | None) -> bool:
    if extracted is None and gold is None: return True
    if extracted is None or gold is None: return False
    return (extracted.get("auto_renew") == gold.get("auto_renew")
            and match_int(extracted.get("term_length_months"), gold.get("term_length_months"))
            and match_int(extracted.get("notice_period_days"), gold.get("notice_period_days"), tolerance=5))


def match_parties(extracted: list, gold: list) -> bool:
    if not extracted or not gold: return False
    e = sorted(normalize_str(x) for x in extracted)
    g = sorted(normalize_str(x) for x in gold)
    return e == g


def score_extraction(extracted: dict, gold: dict) -> dict:
    """Return per-field correctness + overall accuracy."""
    results = {
        "parties": match_parties(extracted.get("parties", []), gold.get("parties", [])),
        "effective_date": match_date(
            extracted.get("effective_date", {}).get("value"),
            gold.get("effective_date"),
        ),
        "term_length_months": match_int(
            extracted.get("term_length_months", {}).get("value"),
            gold.get("term_length_months"),
        ),
        "renewal_terms": match_renewal_terms(
            extracted.get("renewal_terms", {}).get("value"),
            gold.get("renewal_terms"),
        ),
        "governing_law": match_string(
            extracted.get("governing_law", {}).get("value"),
            gold.get("governing_law"),
        ),
    }
    total = len(results)
    correct = sum(results.values())
    return {
        "per_field": results,
        "correct": correct,
        "total": total,
        "accuracy": correct / total,
    }

from __future__ import annotations
import json
from pathlib import Path
from collections import defaultdict

from ci.config import settings
from ci.extraction.ocr import extract_text
from ci.extraction.llm import extract_msa
from ci.eval.scoring import score_extraction


def run_eval(gold_dir: Path) -> dict:
    """Run extraction over the gold set; return aggregate report."""
    pdfs = sorted(gold_dir.glob("*.pdf"))
    if not pdfs:
        raise RuntimeError(f"No PDFs in {gold_dir}")

    per_file = []
    field_correct: dict[str, int] = defaultdict(int)
    field_total: dict[str, int] = defaultdict(int)

    for pdf in pdfs:
        gold_path = pdf.with_suffix(".gold.json")
        if not gold_path.exists():
            print(f"Skip {pdf.name}: no gold file")
            continue
        gold = json.loads(gold_path.read_text())

        file_bytes = pdf.read_bytes()
        text, _ = extract_text(file_bytes)
        extraction = extract_msa(text)
        scored = score_extraction(extraction.model_dump(mode="json"), gold["fields"])

        per_file.append({"file": pdf.name, **scored})
        for field, was_correct in scored["per_field"].items():
            field_total[field] += 1
            if was_correct: field_correct[field] += 1

    overall_correct = sum(f["correct"] for f in per_file)
    overall_total = sum(f["total"] for f in per_file)
    return {
        "model": settings.extraction_model,
        "n_documents": len(per_file),
        "overall_accuracy": overall_correct / overall_total if overall_total else 0.0,
        "per_field_accuracy": {
            f: field_correct[f] / field_total[f] for f in field_total
        },
        "per_file": per_file,
    }

def render_report(eval_result: dict) -> str:
    lines = []
    lines.append(f"# Extraction Eval Report")
    lines.append(f"")
    lines.append(f"- **Model**: {eval_result['model']}")
    lines.append(f"- **Documents**: {eval_result['n_documents']}")
    lines.append(f"- **Overall accuracy**: {eval_result['overall_accuracy']:.1%}")
    lines.append(f"")
    lines.append(f"## Per-field accuracy")
    for f, acc in sorted(eval_result['per_field_accuracy'].items()):
        marker = "✅" if acc >= 0.9 else "⚠️" if acc >= 0.7 else "❌"
        lines.append(f"- {marker} **{f}**: {acc:.1%}")
    lines.append(f"")
    lines.append(f"## Per-document results")
    lines.append(f"| File | Correct/Total | Accuracy |")
    lines.append(f"|------|--------------|----------|")
    for f in eval_result['per_file']:
        lines.append(f"| {f['file']} | {f['correct']}/{f['total']} | {f['accuracy']:.0%} |")
    return "\n".join(lines)

from pathlib import Path
from ci.eval.runner import run_eval
from ci.eval.report import render_report


@app.command("eval")
def eval_(gold_dir: Path = Path(settings.gold_dir)) -> None:
    """Run eval against the gold set."""
    console.print(f"Running eval against {gold_dir}...")
    result = run_eval(gold_dir)
    report = render_report(result)
    console.print(report)
    Path("eval_report.md").write_text(report)
    console.print(f"\nReport saved to eval_report.md")

ci eval

# Extraction Eval Report

- **Model**: claude-sonnet-5-5
- **Documents**: 7
- **Overall accuracy**: 88.6%

## Per-field accuracy
- ✅ **effective_date**: 100.0%
- ⚠️ **governing_law**: 85.7%
- ✅ **parties**: 100.0%
- ⚠️ **renewal_terms**: 71.4%
- ⚠️ **term_length_months**: 85.7%

## Per-document results
| File | Correct/Total | Accuracy |
|------|--------------|----------|
| acme_msa_2024.pdf | 5/5 | 100% |
| ...