Module 9. Evaluation, cost and reliability

Lab 9. Eval harness, dashboard and a routing experiment

Goal: take the Module 4 invoice extractor (the RAG assistant from Module 5 works too; see the note at the end), measure it on 50 or more labelled cases, add a cost and latency logger and a static HTML report, then test Haiku with a Sonnet fallback. Target: cut cost by 40% without losing accuracy.

Setup

cd claude-transformation            # the repo from Module 1
pip install anthropic matplotlib
mkdir module9 && cd module9

1. Cases and prices

Use sample_cases.py to draw a stratified sample from the invoices you used in Module 4 (as .txt files). It writes cases_todo.jsonl with empty answers. Label them, save as cases.jsonl, one JSON object per line. Include at least 10 Hebrew, 10 Russian, 5 mixed and 8 edge cases. One line looks like this:

{"id": "inv-017", "tag": "he", "text": "חשבונית מס 20417 ... סה\"כ לתשלום 1,170.00 ₪ ...", "expected": {"supplier": "מחסני הצפון בע\"מ", "invoice_number": "20417", "date": "2026-09-14", "currency": "ILS", "net": 1000.0, "vat": 170.0, "total": 1170.0}}

Create prices.json with the current per-million-token input and output prices for the three models, copied from the pricing page (the file is yours; the course does not state prices):

{
  "claude-haiku-5-5":  {"in": 0.0, "out": 0.0},
  "claude-sonnet-5-5": {"in": 0.0, "out": 0.0},
  "claude-opus-5-5":   {"in": 0.0, "out": 0.0}
}

sample_cases.py

"""sample_cases.py - draw a stratified sample of real invoice texts for labelling.

    python sample_cases.py ./client_invoices_txt 10   # 10 per stratum -> cases_todo.jsonl
Fill in "expected" by hand (or have Claude pre-fill it and then check every value yourself).
"""
import json
import random
import re
import sys
from collections import defaultdict
from pathlib import Path

FIELDS = ["supplier", "invoice_number", "date", "currency", "net", "vat", "total"]


def tag_of(text):
    he = bool(re.search(r"[֐-׿]", text))
    ru = bool(re.search(r"[Ѐ-ӿ]", text))
    lang = "mixed" if he and ru else "he" if he else "ru" if ru else "en"
    return lang + ("_long" if len(text) > 1500 else "")


def main():
    folder = Path(sys.argv[1])
    per_stratum = int(sys.argv[2]) if len(sys.argv) > 2 else 8
    random.seed(7)  # same sample every time
    strata = defaultdict(list)
    for p in sorted(folder.glob("*.txt")):
        text = p.read_text(encoding="utf-8", errors="replace")
        strata[tag_of(text)].append((p.stem, text))
    out = []
    for tag, items in sorted(strata.items()):
        for stem, text in random.sample(items, min(per_stratum, len(items))):
            out.append({"id": stem, "tag": tag, "text": text, "expected": {f: None for f in FIELDS}})
    with open("cases_todo.jsonl", "w", encoding="utf-8") as f:
        for c in out:
            f.write(json.dumps(c, ensure_ascii=False) + "\n")
    print({t: min(per_stratum, len(v)) for t, v in strata.items()}, "->", len(out), "cases")


if __name__ == "__main__":
    main()

2. eval.py (harness, graders, judge, logger, router)

"""eval.py - eval harness, cost/latency logger and router test for the invoice extractor.

    python eval.py --config sonnet --judge
    python eval.py --config haiku
    python eval.py --config routed
    python eval.py --config routed --min-exact 0.95   # CI gate: exit code 1 if below
Delete results_<config>.csv to start that config from scratch (otherwise finished cases are skipped).
"""
import argparse
import csv
import json
import re
import statistics
import sys
import time
import uuid
from pathlib import Path

import anthropic

# Check current model IDs at docs.claude.com before running
HAIKU, SONNET, OPUS = "claude-haiku-5-5", "claude-sonnet-5-5", "claude-opus-5-5"
JUDGE_MODEL = OPUS  # a different tier from the model under test, to limit self-preference
CONFIGS = {"haiku": (HAIKU, None), "sonnet": (SONNET, None), "routed": (HAIKU, SONNET)}
FIELDS = ["supplier", "invoice_number", "date", "currency", "net", "vat", "total"]
MONEY = {"net", "vat", "total"}

CASES_FILE = Path("cases.jsonl")
PRICES_FILE = Path("prices.json")  # {"model-id": {"in": usd_per_million, "out": usd_per_million}}
LOG_FILE = Path("calls.jsonl")
HUMAN_FILE = Path("human_labels.csv")  # optional: case_id,human_score (0/1/2) for judge calibration

client = anthropic.Anthropic(max_retries=3, timeout=45.0)

SYSTEM = """You extract fields from one supplier invoice (Hebrew, Russian or English).
Return only a JSON object with keys: supplier, invoice_number, date (YYYY-MM-DD),
currency (ISO code), net, vat, total (numbers, no symbols). Use null if a field is missing.
Do not guess."""

RUBRIC = """You grade an invoice extraction. You get the invoice text and the extracted JSON.
Score 2 if every value is supported by the text, 1 if one value is doubtful or badly
formatted, 0 if any value is invented or clearly wrong. Judge only against the text.
Reply with only JSON: {"reason": "<one sentence>", "score": 0|1|2}"""


def call(model, system, user, meta, max_tokens=400):
    """One API call plus one structured log line (prompt size, tokens, latency)."""
    t0 = time.perf_counter()
    msg = client.messages.create(
        model=model,
        max_tokens=max_tokens,
        system=system,
        messages=[{"role": "user", "content": user}],
    )
    sec = time.perf_counter() - t0
    text = "".join(b.text for b in msg.content if b.type == "text")
    rec = {
        "ts": round(time.time(), 1), **meta, "model": model,
        "in": msg.usage.input_tokens, "out": msg.usage.output_tokens,
        "sec": round(sec, 3), "stop": msg.stop_reason,
    }
    with LOG_FILE.open("a", encoding="utf-8") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    return text, rec


def parse_json(text):
    try:
        return json.loads(text[text.index("{"): text.rindex("}") + 1])
    except ValueError:  # includes JSONDecodeError
        return None


def acceptable(d):
    """Cheap router check: shape, date format, and net + vat = total."""
    if not isinstance(d, dict) or any(k not in d for k in FIELDS):
        return False
    try:
        if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", str(d["date"])):
            return False
        total = float(d["total"])
        if d["net"] is not None and d["vat"] is not None:
            return abs(float(d["net"]) + float(d["vat"]) - total) <= 0.02
        return True
    except (TypeError, ValueError):
        return False


def try_extract(model, text, meta):
    try:
        raw, rec = call(model, SYSTEM, text, {**meta, "role": "extract"})
        return parse_json(raw), rec
    except anthropic.APIError as e:  # retries already done by the SDK
        print(f"  {model} failed: {type(e).__name__}", file=sys.stderr)
        return None, None


def extract(text, primary, fallback, meta):
    """Primary model first; if the output fails the check (or the call fails), use the fallback."""
    recs = []
    data, rec = try_extract(primary, text, meta)
    if rec:
        recs.append(rec)
    used, escalated = primary, False
    if fallback and not acceptable(data):
        used, escalated = fallback, True
        data, rec = try_extract(fallback, text, meta)
        if rec:
            recs.append(rec)
    return data, used, escalated, recs


def norm_name(s):
    s = re.sub(r"[^\w\s]", "", str(s or "").casefold())
    s = re.sub(r"\b(ltd|inc|llc|gmbh|בעמ|ооо)\b", "", s)
    return " ".join(s.split())


def field_ok(name, exp, got):
    if exp is None or got is None:
        return exp is None and got is None
    if name in MONEY:
        try:
            return abs(float(exp) - float(got)) <= 0.01
        except (TypeError, ValueError):
            return False
    if name == "supplier":
        return norm_name(exp) == norm_name(got)
    if name == "invoice_number":
        return re.sub(r"\W", "", str(exp)).casefold() == re.sub(r"\W", "", str(got)).casefold()
    return str(exp).strip().upper() == str(got).strip().upper()


def grade(expected, got):
    """Code grader: per-field booleans and the share of fields correct."""
    got = got if isinstance(got, dict) else {}
    oks = {f: field_ok(f, expected.get(f), got.get(f)) for f in FIELDS}
    return oks, sum(oks.values()) / len(FIELDS)


def judge(case, data, meta):
    """LLM-as-judge with a rubric. Returns 0, 1, 2 or None."""
    prompt = (f"INVOICE TEXT:\n{case['text']}\n\n"
              f"EXTRACTED JSON:\n{json.dumps(data, ensure_ascii=False)}")
    try:
        raw, _ = call(JUDGE_MODEL, RUBRIC, prompt, {**meta, "role": "judge"}, max_tokens=200)
    except anthropic.APIError:
        return None
    score = (parse_json(raw) or {}).get("score")
    return score if score in (0, 1, 2) else None


def cost_usd(recs, prices):
    return sum(r["in"] * prices[r["model"]]["in"] + r["out"] * prices[r["model"]]["out"]
               for r in recs) / 1_000_000


def pct(values, p):
    s = sorted(values)
    return s[min(len(s) - 1, int(p / 100 * len(s)))]


COLS = (["run_id", "config", "case_id", "tag", "model_used", "escalated", "exact", "score", "judge"]
        + [f"f_{f}" for f in FIELDS] + ["in_tokens", "out_tokens", "latency_s", "cost_usd"])


def summarize(path, min_exact=None):
    with path.open(encoding="utf-8", newline="") as f:
        rows = list(csv.DictReader(f))
    n = len(rows)
    exact = sum(r["exact"] == "1" for r in rows) / n
    lat = [float(r["latency_s"]) for r in rows]
    cost_1k = sum(float(r["cost_usd"]) for r in rows) / n * 1000
    print(f"\n{path.name}: {n} cases")
    print(f"  exact match      {exact:.1%}")
    print(f"  mean field score {statistics.mean(float(r['score']) for r in rows):.3f}")
    print(f"  escalated        {sum(r['escalated'] == '1' for r in rows) / n:.1%}")
    print(f"  latency p50/p95  {pct(lat, 50):.2f}s / {pct(lat, 95):.2f}s")
    print(f"  cost per 1000    {cost_1k:.4f} USD")
    judged = [r for r in rows if r["judge"] != ""]
    if judged:
        print(f"  judge mean       {statistics.mean(float(r['judge']) for r in judged):.2f} of 2")
        if HUMAN_FILE.exists():
            with HUMAN_FILE.open(encoding="utf-8", newline="") as f:
                human = {r["case_id"]: int(r["human_score"]) for r in csv.DictReader(f)}
            pairs = [(int(r["judge"]), human[r["case_id"]]) for r in judged if r["case_id"] in human]
            if pairs:
                agree = sum(a == b for a, b in pairs) / len(pairs)
                print(f"  judge vs human   {agree:.1%} agreement on {len(pairs)} labelled cases")
    if min_exact is not None and exact < min_exact:
        print(f"FAIL: exact match {exact:.1%} is below the gate {min_exact:.1%}")
        sys.exit(1)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", choices=list(CONFIGS), required=True)
    ap.add_argument("--judge", action="store_true", help="also run the LLM judge")
    ap.add_argument("--limit", type=int, help="only the first N cases (smoke test)")
    ap.add_argument("--min-exact", type=float, help="CI gate on exact-match rate")
    args = ap.parse_args()

    cases = [json.loads(line) for line in CASES_FILE.read_text(encoding="utf-8").splitlines() if line.strip()]
    if args.limit:
        cases = cases[: args.limit]
    prices = json.loads(PRICES_FILE.read_text(encoding="utf-8"))
    primary, fallback = CONFIGS[args.config]
    out_path = Path(f"results_{args.config}.csv")

    done = set()
    if out_path.exists():  # idempotent resume: skip cases already written
        with out_path.open(encoding="utf-8", newline="") as f:
            done = {r["case_id"] for r in csv.DictReader(f)}
    new_file = not out_path.exists()
    run_id = uuid.uuid4().hex[:8]

    with out_path.open("a", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=COLS)
        if new_file:
            w.writeheader()
        for c in cases:
            if c["id"] in done:
                continue
            meta = {"run_id": run_id, "config": args.config, "case_id": c["id"]}
            data, used, escalated, recs = extract(c["text"], primary, fallback, meta)
            oks, score = grade(c["expected"], data)
            j = judge(c, data, meta) if args.judge and data else None
            row = {
                "run_id": run_id, "config": args.config, "case_id": c["id"], "tag": c.get("tag", ""),
                "model_used": used, "escalated": int(escalated), "exact": int(all(oks.values())),
                "score": round(score, 4), "judge": "" if j is None else j,
                "in_tokens": sum(r["in"] for r in recs), "out_tokens": sum(r["out"] for r in recs),
                "latency_s": round(sum(r["sec"] for r in recs), 3),
                "cost_usd": round(cost_usd(recs, prices), 8),
            }
            row.update({f"f_{k}": int(v) for k, v in oks.items()})
            w.writerow(row)
            f.flush()
            print(f"{c['id']:>8} {c.get('tag', ''):<8} {used:<18} exact={row['exact']}")
    summarize(out_path, args.min_exact)


if __name__ == "__main__":
    main()

3. report.py (static dashboard with matplotlib)

"""report.py - build report.html (static, charts embedded as PNG) from results_*.csv."""
import base64
import csv
import html
import io
import statistics
from datetime import date
from pathlib import Path

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt

CONFIGS = ["sonnet", "haiku", "routed"]
BASELINE = "sonnet"
COLORS = {"sonnet": "#0f7a72", "haiku": "#8a9a9f", "routed": "#c8642c"}


def load(config):
    path = Path(f"results_{config}.csv")
    if not path.exists():
        return None
    with path.open(encoding="utf-8", newline="") as f:
        rows = list(csv.DictReader(f))
    n = len(rows)
    lat = sorted(float(r["latency_s"]) for r in rows)
    tags = sorted({r["tag"] for r in rows})
    by_tag = {}
    for t in tags:
        sub = [r for r in rows if r["tag"] == t]
        by_tag[t] = sum(r["exact"] == "1" for r in sub) / len(sub)
    return {
        "n": n,
        "exact": sum(r["exact"] == "1" for r in rows) / n,
        "score": statistics.mean(float(r["score"]) for r in rows),
        "cost_1k": sum(float(r["cost_usd"]) for r in rows) / n * 1000,
        "p50": lat[n // 2],
        "p95": lat[min(n - 1, int(n * 0.95))],
        "esc": sum(r["escalated"] == "1" for r in rows) / n,
        "by_tag": by_tag,
    }


def png(fig):
    buf = io.BytesIO()
    fig.savefig(buf, format="png", dpi=110, bbox_inches="tight")
    plt.close(fig)
    data = base64.b64encode(buf.getvalue()).decode()
    return f'<img alt="chart" src="data:image/png;base64,{data}">'


def bar(names, values, title, ylabel, fmt, zero_one=False):
    fig, ax = plt.subplots(figsize=(4.2, 3))
    ax.bar(names, values, color=[COLORS[n] for n in names])
    ax.set_title(title, fontsize=10)
    ax.set_ylabel(ylabel)
    if zero_one:
        ax.set_ylim(0, 1)
    for i, v in enumerate(values):
        ax.text(i, v, fmt.format(v), ha="center", va="bottom", fontsize=8)
    ax.spines[["top", "right"]].set_visible(False)
    return png(fig)


def main():
    data = {c: load(c) for c in CONFIGS}
    data = {c: d for c, d in data.items() if d}
    names = list(data)
    charts = [
        bar(names, [data[c]["exact"] for c in names], "Exact match", "share of cases", "{:.1%}", True),
        bar(names, [data[c]["cost_1k"] for c in names], "Cost per 1000 documents", "USD", "{:.3f}"),
    ]

    fig, ax = plt.subplots(figsize=(4.2, 3))
    xs = range(len(names))
    ax.bar([x - 0.2 for x in xs], [data[c]["p50"] for c in names], 0.4, label="p50", color="#0f7a72")
    ax.bar([x + 0.2 for x in xs], [data[c]["p95"] for c in names], 0.4, label="p95", color="#c8642c")
    ax.set_xticks(list(xs))
    ax.set_xticklabels(names)
    ax.set_title("Latency", fontsize=10)
    ax.set_ylabel("seconds")
    ax.legend(frameon=False)
    ax.spines[["top", "right"]].set_visible(False)
    charts.append(png(fig))

    fig, ax = plt.subplots(figsize=(4.2, 3))
    for c in names:
        ax.scatter(data[c]["cost_1k"], data[c]["exact"], s=80, color=COLORS[c])
        ax.annotate(c, (data[c]["cost_1k"], data[c]["exact"]), textcoords="offset points", xytext=(6, 6), fontsize=8)
    ax.set_xlabel("cost per 1000 documents (USD)")
    ax.set_ylabel("exact match")
    ax.set_title("Accuracy vs cost", fontsize=10)
    ax.spines[["top", "right"]].set_visible(False)
    charts.append(png(fig))

    rows = []
    for c in names:
        d = data[c]
        rows.append(f"<tr><td>{c}</td><td>{d['n']}</td><td>{d['exact']:.1%}</td><td>{d['cost_1k']:.3f}</td>"
                    f"<td>{d['p50']:.2f}</td><td>{d['p95']:.2f}</td><td>{d['esc']:.0%}</td></tr>")
    tags = sorted({t for d in data.values() for t in d["by_tag"]})
    tag_rows = "".join(
        f"<tr><td>{html.escape(t)}</td>" + "".join(f"<td>{data[c]['by_tag'].get(t, float('nan')):.0%}</td>" for c in names) + "</tr>"
        for t in tags)

    verdict = "Need results for sonnet and routed to judge the goal."
    if BASELINE in data and "routed" in data:
        saving = 1 - data["routed"]["cost_1k"] / data[BASELINE]["cost_1k"]
        drop = data[BASELINE]["exact"] - data["routed"]["exact"]
        ok = saving >= 0.40 and drop <= 0.01
        verdict = (f"Routed vs {BASELINE}: cost {saving:+.0%} lower, exact match {drop * 100:+.1f} points lower. "
                   f"Goal (40% cheaper, no more than 1 point lost): {'MET' if ok else 'NOT MET'}.")

    page = f"""<!doctype html><html lang="en"><head><meta charset="utf-8"><title>Eval and cost report</title>
<style>body{{font-family:system-ui,sans-serif;max-width:960px;margin:24px auto;padding:0 16px;color:#14232b}}
table{{border-collapse:collapse;width:100%;margin:12px 0}}td,th{{border-bottom:1px solid #d5dfe1;padding:6px 10px;text-align:left}}
.grid{{display:flex;flex-wrap:wrap;gap:12px}}.grid img{{max-width:100%;height:auto}}.v{{padding:10px 14px;background:#dcefed;border-left:3px solid #0f7a72}}</style></head>
<body><h1>Eval and cost report</h1><p>Generated {date.today().isoformat()}</p>
<p class="v">{html.escape(verdict)}</p>
<table><tr><th>config</th><th>cases</th><th>exact</th><th>USD / 1000 docs</th><th>p50 s</th><th>p95 s</th><th>escalated</th></tr>{''.join(rows)}</table>
<div class="grid">{''.join(charts)}</div>
<h2>Exact match by case type</h2><table><tr><th>tag</th>{''.join(f'<th>{c}</th>' for c in names)}</tr>{tag_rows}</table>
</body></html>"""
    Path("report.html").write_text(page, encoding="utf-8")
    print("wrote report.html")


if __name__ == "__main__":
    main()

Steps

1. Smoke test: python eval.py --config sonnet --limit 5. Check that calls.jsonl and results_sonnet.csv appear and the numbers look sane.
2. Baseline: python eval.py --config sonnet --judge. Open the CSV, read every failed row, and decide whether the model or your label is wrong. Fix labels before you trust any score.
3. Calibrate the judge: pick 30 rows, score them yourself into human_labels.csv (case_id,human_score), and re-run the summary. Below about 80% agreement, rewrite the rubric in RUBRIC.
4. Cheap model: python eval.py --config haiku. Compare accuracy by case type. Where does Haiku fail?
5. Routing: python eval.py --config routed. Read the escalation rate. Tune acceptable() (stricter check, more escalation, more accuracy) until accuracy is within one point of Sonnet.
6. Dashboard: python report.py, open report.html. The top line states whether the 40% goal was met.
7. Add the CI gate: --min-exact 0.95 (use your own number) and copy the workflow sketch from 9.4 into your repo.
8. Write five lines: the accuracy and cost of each configuration, which one you would ship, and what you would do next (prompt caching on the system prompt, batching for the nightly backlog).

Using the Module 5 RAG assistant instead: replace SYSTEM, FIELDS and grade(). Cases become question, expected facts (strings that must appear) and the source document ID. Code graders check that the answer contains the facts and cites the right source; the LLM judge rubric checks "every claim is supported by the retrieved text". The rest of the harness stays the same.

Expected result: Haiku handles most clean documents; the escalation rate shows how many need Sonnet. If the routed run costs at least 40% less than Sonnet alone with accuracy within one point, you have met the goal. If not, say so in your notes: a failed trade-off is a valid, reportable result.