backend/eval/run_eval.py

"""CUAD evaluation harness — the accuracy slide.

Scores the clause classifier against CUAD's expert labels:
for each contract and each category we evaluate, CUAD gives gold answer
spans (character offsets into the contract text). We predict spans = the
clauses our pipeline tags with that category. A prediction matches a gold
span if their character-range Jaccard overlap >= --threshold (default 0.3,
since our clause spans are deliberately wider than CUAD's tight answers
recall is what matters most here).

Reports per-category precision / recall / F1.

    .venv/bin/python -m eval.run_eval --limit 50                       (from backend/)
    .venv/bin/python -m eval.run_eval --limit 50 --classifier zeroshot
"""
from __future__ import annotations

import argparse
import json
import os
import pathlib
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1]))

from app.classification import classify_all  # noqa: E402
from app.schema import Clause, SourceSpan  # noqa: E402
from app.segmentation import CLAUSE_START  # noqa: E402

CUAD_JSON = pathlib.Path(__file__).resolve().parents[2] / "data" / "cuad" / "CUAD_v1.json"

# our category key -> substring of the CUAD question that identifies it
CATEGORY_TO_CUAD_QUESTION = {
    "auto_renewal": "Renewal Term",
    "liability_cap": "Cap On Liability",
    "governing_law": "Governing Law",
    "exclusivity": "Exclusivity",
    "termination": "Termination For Convenience",
    "insurance": "Insurance",
    "audit_rights": "Audit Rights",
    "assignment": "Anti-Assignment",
    "notice": "Notice Period To Terminate Renewal",
    "ip_ownership": "Ip Ownership Assignment",
    "warranty": "Warranty Duration",
    "term": "Expiration Date",
}


def segment_plain_text(text: str) -> list[Clause]:
    """Lightweight clause split for raw CUAD text (no PDF offsets needed)."""
    lines = text.split("\n")
    offsets, pos = [], 0
    for ln in lines:
        offsets.append(pos)
        pos += len(ln) + 1
    starts = [i for i, ln in enumerate(lines) if CLAUSE_START.match(ln)]
    clauses = []
    if not starts:
        starts = [0]
    for n, li in enumerate(starts):
        s = offsets[li]
        e = offsets[starts[n + 1]] if n + 1 < len(starts) else len(text)
        clauses.append(Clause(
            id=f"c{n}", text=text[s:e],
            span=SourceSpan(start_char=s, end_char=e, page=0)))
    return clauses


def jaccard(a: tuple[int, int], b: tuple[int, int]) -> float:
    inter = max(0, min(a[1], b[1]) - max(a[0], b[0]))
    union = (a[1] - a[0]) + (b[1] - b[0]) - inter
    return inter / union if union else 0.0


def overlap_recall(gold: tuple[int, int], pred: tuple[int, int]) -> float:
    """Fraction of the gold span covered by the prediction."""
    inter = max(0, min(gold[1], pred[1]) - max(gold[0], pred[0]))
    return inter / (gold[1] - gold[0]) if gold[1] > gold[0] else 0.0


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--limit", type=int, default=50, help="contracts to evaluate")
    ap.add_argument("--threshold", type=float, default=0.3,
                    help="match if gold-span coverage >= threshold")
    ap.add_argument("--classifier",
                    choices=["rules", "zeroshot", "legalbert", "fusion", "auto"], default="rules",
                    help="classification backend to score (default: rules). 'fusion' is "
                         "the shipped default (fine-tuned LegalBERT + zero-shot fill); on the "
                         "scored CUAD categories it is identical to 'legalbert'.")
    args = ap.parse_args()
    os.environ["CLASSIFIER"] = args.classifier

    if not CUAD_JSON.exists():
        sys.exit(f"CUAD not found at {CUAD_JSON}. Run scripts/download_cuad.py first.")

    data = json.loads(CUAD_JSON.read_text())["data"]
    data = data[: args.limit]

    stats = {k: {"tp": 0, "fp": 0, "fn": 0} for k in CATEGORY_TO_CUAD_QUESTION}

    for doc in data:
        for para in doc["paragraphs"]:
            text = para["context"]
            clauses = segment_plain_text(text)
            classify_all(clauses)

            # gold spans per category
            gold: dict[str, list[tuple[int, int]]] = {k: [] for k in stats}
            for qa in para["qas"]:
                for key, q_sub in CATEGORY_TO_CUAD_QUESTION.items():
                    if q_sub in qa["question"]:
                        for ans in qa["answers"]:
                            s = ans["answer_start"]
                            gold[key].append((s, s + len(ans["text"])))

            for key in stats:
                preds = [(c.span.start_char, c.span.end_char)
                         for c in clauses if key in c.categories]
                golds = gold[key]
                matched_gold, matched_pred = set(), set()
                for gi, g in enumerate(golds):
                    for pi, p in enumerate(preds):
                        if overlap_recall(g, p) >= args.threshold:
                            matched_gold.add(gi)
                            matched_pred.add(pi)
                stats[key]["tp"] += len(matched_gold)
                stats[key]["fn"] += len(golds) - len(matched_gold)
                stats[key]["fp"] += len(preds) - len(matched_pred)

    print(f"\nCUAD eval — {len(data)} contracts, classifier={args.classifier}, "
          f"gold-coverage threshold {args.threshold}\n")
    print(f"{'category':<18} {'P':>6} {'R':>6} {'F1':>6}   tp/fp/fn")
    print("-" * 56)
    macro = []
    for key, s in stats.items():
        p = s["tp"] / (s["tp"] + s["fp"]) if s["tp"] + s["fp"] else 0.0
        r = s["tp"] / (s["tp"] + s["fn"]) if s["tp"] + s["fn"] else 0.0
        f1 = 2 * p * r / (p + r) if p + r else 0.0
        macro.append(f1)
        print(f"{key:<18} {p:6.2f} {r:6.2f} {f1:6.2f}   "
              f"{s['tp']}/{s['fp']}/{s['fn']}")
    print("-" * 56)
    print(f"{'macro-F1':<18} {'':>6} {'':>6} {sum(macro) / len(macro):6.2f}")


if __name__ == "__main__":
    main()