LLM evaluation: build an eval suite and gate deploys in CI

#!/usr/bin/env python3
"""Offline LLM eval runner. Standard library only.
 
Usage: python3 run_evals.py [candidate|fixed]
Compares a scripted candidate against a scripted baseline, prints the
per-task report, and exits 1 if the deploy gate blocks.
"""
import json
import math
import sys
from collections import defaultdict
 
# Golden dataset. In your repo, keep this as a versioned golden.jsonl.
# These cases are illustrative; yours should come from sampled traffic.
GOLDEN = [
    {"id": "C1", "task": "classify", "input": "Charged twice in October", "expected": "billing"},
    {"id": "C2", "task": "classify", "input": "App crashes on login", "expected": "bug"},
    {"id": "C3", "task": "classify", "input": "How do I export invoices?", "expected": "how_to"},
    {"id": "C4", "task": "classify", "input": "Refund my annual plan", "expected": "billing"},
    {"id": "C5", "task": "classify", "input": "SSO setup fails with 403", "expected": "bug"},
    {"id": "E1", "task": "extract", "input": "Refund $40 on order A-17"},
    {"id": "E2", "task": "extract", "input": "Please return 25 EUR for B-09"},
    {"id": "E3", "task": "extract", "input": "Refund 18.50 GBP, order C-33"},
    {"id": "E4", "task": "extract", "input": "Refund $5 on order D-21"},
    {"id": "R1", "task": "retrieve", "input": "refund approval limits", "relevant": ["approvals", "refund-policy"]},
    {"id": "R2", "task": "retrieve", "input": "SSO error codes", "relevant": ["sso-errors"]},
    {"id": "R3", "task": "retrieve", "input": "invoice export formats", "relevant": ["export", "invoice-api"]},
    {"id": "R4", "task": "retrieve", "input": "dunning retry schedule", "relevant": ["dunning", "retries"]},
    {"id": "A1", "task": "answer", "input": "Can I get a refund after 30 days?", "rubric": ["within 30 days"]},
    {"id": "A2", "task": "answer", "input": "Why was my card charged twice?", "rubric": ["authorization"]},
    {"id": "A3", "task": "answer", "input": "Who approves a $600 refund?", "rubric": ["second approver"], "must_pass": True},
    {"id": "A4", "task": "answer", "input": "How long do dunning retries last?", "rubric": ["14 days"]},
]
 
REFUND_SCHEMA = {"order_id": str, "amount": (int, float), "currency": str}
CURRENCIES = {"USD", "EUR", "GBP"}
K = 3
 
# Scripted fake model outputs. Replace complete() with a real API call.
BASELINE = {
    "C1": "billing", "C2": "bug", "C3": "how_to", "C4": "how_to", "C5": "bug",
    "E1": '{"order_id": "A-17", "amount": 40, "currency": "USD"}',
    "E2": '{"order_id": "B-09", "amount": 25, "currency": "EUR"}',
    "E3": '{"order_id": "C-33", "amount": "18.50 GBP"}',
    "E4": '{"order_id": "D-21", "amount": 5, "currency": "USD"}',
    "R1": ["approvals", "refund-policy", "faq"],
    "R2": ["sso-errors", "login", "faq"],
    "R3": ["export", "faq", "billing-api"],
    "R4": ["dunning", "retries", "faq"],
    "A1": "You can request a refund within 30 days of purchase.",
    "A2": "Two authorization holds posted; one drops off in 3 days.",
    "A3": "Refunds over $500 need a second approver in finance.",
    "A4": "Dunning retries run for 14 days, then the plan pauses.",
}
CANDIDATE = BASELINE | {
    "C4": "billing",
    "E3": '{"order_id": "C-33", "amount": 18.5, "currency": "GBP"}',
    "R3": ["export", "invoice-api", "faq"],
    "R4": ["dunning", "faq", "billing-api"],
    "A3": "Refunds over $500 are approved by the finance team.",
}
FIXED = CANDIDATE | {
    "R4": ["dunning", "retries", "faq"],
    "A3": "Refunds over $500 need a second approver in finance.",
}
 
 
class ScriptedModel:
    """Stands in for a real client."""
 
    def __init__(self, name, outputs):
        self.name, self.outputs = name, outputs
 
    def complete(self, case):
        return self.outputs[case["id"]]
 
 
class ScriptedJudge:
    """Stands in for an LLM judge: passes if every rubric phrase appears."""
 
    def grade(self, case, output):
        missing = [p for p in case["rubric"] if p not in output.lower()]
        return not missing, f"missing {missing}" if missing else "ok"
 
 
def exact_match(case, output, _judge):
    ok = output.strip().lower() == case["expected"].lower()
    return float(ok), f"got {output!r}"
 
 
def schema_valid(case, output, _judge):
    try:
        data = json.loads(output)
    except json.JSONDecodeError as err:
        return 0.0, f"invalid JSON: {err}"
    for field, kind in REFUND_SCHEMA.items():
        if not isinstance(data.get(field), kind):
            return 0.0, f"bad field {field}: {data.get(field)!r}"
    if data["currency"] not in CURRENCIES:
        return 0.0, f"unknown currency {data['currency']}"
    return 1.0, "ok"
 
 
def recall_at_k(case, output, _judge):
    relevant = set(case["relevant"])
    score = len(relevant & set(output[:K])) / len(relevant)
    return score, f"recall@{K}={score:.2f}"
 
 
def judged(case, output, judge):
    passed, why = judge.grade(case, output)
    return float(passed), why
 
 
METRICS = {
    "classify": ("exact_match", exact_match, 0.90),
    "extract": ("schema_valid", schema_valid, 0.95),
    "retrieve": (f"recall@{K}", recall_at_k, 0.85),
    "answer": ("judged_pass", judged, 0.75),
}
 
 
def run(model, judge):
    results = {}
    for case in GOLDEN:
        _name, grader, _floor = METRICS[case["task"]]
        score, note = grader(case, model.complete(case), judge)
        results[case["id"]] = {"score": score, "passed": score == 1.0, "note": note}
    return results
 
 
def mcnemar_exact_p(new_fail, new_pass):
    """Two-sided exact McNemar: the discordant pairs are Binomial(n, 0.5)."""
    n = new_fail + new_pass
    if n == 0:
        return 1.0
    tail = sum(math.comb(n, i) for i in range(min(new_fail, new_pass) + 1)) / 2**n
    return min(1.0, 2 * tail)
 
 
def compare(base, cand):
    by_task = defaultdict(list)
    for case in GOLDEN:
        by_task[case["task"]].append(case)
    blockers, rows = [], []
    for task, cases in by_task.items():
        metric, _grader, floor = METRICS[task]
        ids = [c["id"] for c in cases]
        b_mean = sum(base[i]["score"] for i in ids) / len(ids)
        c_mean = sum(cand[i]["score"] for i in ids) / len(ids)
        new_fail = [i for i in ids if base[i]["passed"] and not cand[i]["passed"]]
        new_pass = [i for i in ids if not base[i]["passed"] and cand[i]["passed"]]
        rows.append((task, metric, b_mean, c_mean, new_fail, new_pass))
        if c_mean < floor:
            blockers.append(f"{task}: {metric} {c_mean:.3f} is below the floor {floor:.2f}")
        if len(new_fail) > len(new_pass):
            blockers.append(f"{task}: net regression, new failures {new_fail}")
    for case in GOLDEN:
        if case.get("must_pass") and not cand[case["id"]]["passed"]:
            blockers.append(f"must-pass case {case['id']} failed: {cand[case['id']]['note']}")
    return rows, blockers
 
 
def main():
    name = sys.argv[1] if len(sys.argv) > 1 else "candidate"
    outputs = {"candidate": CANDIDATE, "fixed": FIXED}[name]
    judge = ScriptedJudge()
    base = run(ScriptedModel("baseline", BASELINE), judge)
    cand = run(ScriptedModel(name, outputs), judge)
    rows, blockers = compare(base, cand)
 
    print(f"baseline vs {name}, {len(GOLDEN)} cases")
    print(f"{'task':<9}{'metric':<14}{'baseline':>9}{'cand':>7}  new_fail  new_pass")
    for task, metric, b_mean, c_mean, new_fail, new_pass in rows:
        print(f"{task:<9}{metric:<14}{b_mean:>9.3f}{c_mean:>7.3f}  "
              f"{','.join(new_fail) or '-':<9} {','.join(new_pass) or '-'}")
 
    b_pass = sum(r["passed"] for r in base.values())
    c_pass = sum(r["passed"] for r in cand.values())
    new_fail = sum(base[i]["passed"] and not cand[i]["passed"] for i in base)
    new_pass = sum(cand[i]["passed"] and not base[i]["passed"] for i in base)
    print(f"overall pass rate: {b_pass}/{len(base)} -> {c_pass}/{len(cand)}")
    print(f"exact McNemar p = {mcnemar_exact_p(new_fail, new_pass):.3f} "
          f"({new_fail} new failures, {new_pass} new passes)")
 
    print("gate: BLOCK" if blockers else "gate: PASS")
    for blocker in blockers:
        print(f"  - {blocker}")
    sys.exit(1 if blockers else 0)
 
 
if __name__ == "__main__":
    main()

$ python3 run_evals.py candidate
baseline vs candidate, 17 cases
task     metric         baseline   cand  new_fail  new_pass
classify exact_match       0.800  1.000  -         C4
extract  schema_valid      0.750  1.000  -         E3
retrieve recall@3          0.875  0.875  R4        R3
answer   judged_pass       1.000  0.750  A3        -
overall pass rate: 14/17 -> 15/17
exact McNemar p = 1.000 (2 new failures, 3 new passes)
gate: BLOCK
  - answer: net regression, new failures ['A3']
  - must-pass case A3 failed: missing ['second approver']
$ echo $?
1
$ python3 run_evals.py fixed
baseline vs fixed, 17 cases
task     metric         baseline   cand  new_fail  new_pass
classify exact_match       0.800  1.000  -         C4
extract  schema_valid      0.750  1.000  -         E3
retrieve recall@3          0.875  1.000  -         R3
answer   judged_pass       1.000  1.000  -         -
overall pass rate: 14/17 -> 17/17
exact McNemar p = 0.250 (0 new failures, 3 new passes)
gate: PASS
$ echo $?
0

#!/usr/bin/env python3
"""Sample-size math for LLM evals. Standard library only."""
import math
import random
from statistics import NormalDist
 
Z = NormalDist().inv_cdf
ALPHA, POWER = 0.05, 0.80
Z_A, Z_B = Z(1 - ALPHA / 2), Z(POWER)
 
 
def wilson(p, n, z=Z_A):
    centre = (p + z**2 / (2 * n)) / (1 + z**2 / n)
    half = z * math.sqrt(p * (1 - p) / n + z**2 / (4 * n**2)) / (1 + z**2 / n)
    return centre - half, centre + half
 
 
def unpaired_n(p1, p2):
    bar = (p1 + p2) / 2
    num = Z_A * math.sqrt(2 * bar * (1 - bar)) + Z_B * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2))
    return math.ceil(num**2 / (p1 - p2) ** 2)
 
 
def paired_n(delta, flip_up):
    """McNemar sample size. b = delta + flip_up flips down, flip_up flips up."""
    psi = delta + 2 * flip_up
    num = Z_A * math.sqrt(psi) + Z_B * math.sqrt(psi - delta**2)
    return math.ceil(num**2 / delta**2)
 
 
def exact_mcnemar_p(b, c):
    n = b + c
    if n == 0:
        return 1.0
    tail = sum(math.comb(n, i) for i in range(min(b, c) + 1)) / 2**n
    return min(1.0, 2 * tail)
 
 
def simulate_power(n, delta, flip_up, trials=4000, seed=7):
    rng = random.Random(seed)
    down, up = delta + flip_up, flip_up
    hits = 0
    for _ in range(trials):
        b = c = 0
        for _ in range(n):
            u = rng.random()
            if u < down:
                b += 1
            elif u < down + up:
                c += 1
        if b > c and exact_mcnemar_p(b, c) < ALPHA:
            hits += 1
    return hits / trials
 
 
print("Wilson 95% interval when the observed pass rate is 90%")
for n in (50, 100, 200, 400, 800, 1600):
    low, high = wilson(0.90, n)
    print(f"  n={n:>5}: {low:.3f} to {high:.3f} (width {high - low:.3f})")
 
print("\nCases to detect a drop from a 90% pass rate (alpha 0.05, power 0.80)")
print("  drop  unpaired per arm  paired (1.5% flip up)")
for drop in (0.10, 0.05, 0.03, 0.02):
    print(f"  {drop:.2f}  {unpaired_n(0.90, 0.90 - drop):>16}  {paired_n(drop, 0.015):>21}")
 
n = paired_n(0.05, 0.015)
print(f"\nSimulated power of exact McNemar, drop 0.05, n={n}: {simulate_power(n, 0.05, 0.015):.3f}")
for try_n in (260, 280, 300):
    print(f"Simulated power, n={try_n}: {simulate_power(try_n, 0.05, 0.015):.3f}")
 
print("\nOne-sided 95% upper bound on the failure rate after zero failures in n cases")
for n in (30, 100, 300, 1000):
    print(f"  n={n:>5}: {1 - 0.05 ** (1 / n):.4f}")

Wilson 95% interval when the observed pass rate is 90%
  n=   50: 0.786 to 0.957 (width 0.170)
  n=  100: 0.826 to 0.945 (width 0.119)
  n=  200: 0.851 to 0.934 (width 0.084)
  n=  400: 0.867 to 0.926 (width 0.059)
  n=  800: 0.877 to 0.919 (width 0.042)
  n= 1600: 0.884 to 0.914 (width 0.029)
 
Cases to detect a drop from a 90% pass rate (alpha 0.05, power 0.80)
  drop  unpaired per arm  paired (1.5% flip up)
  0.10               199                    100
  0.05               686                    249
  0.03              1774                    521
  0.02              3841                    979
 
Simulated power of exact McNemar, drop 0.05, n=249: 0.769
Simulated power, n=260: 0.779
Simulated power, n=280: 0.817
Simulated power, n=300: 0.855
 
One-sided 95% upper bound on the failure rate after zero failures in n cases
  n=   30: 0.0950
  n=  100: 0.0295
  n=  300: 0.0099
  n= 1000: 0.0030

name: evals
on:
  pull_request:
 
jobs:
  eval-gate:
    runs-on: ubuntu-latest
    timeout-minutes: 20
    steps:
      - uses: actions/checkout@v7
      - uses: actions/setup-python@v7
        with:
          python-version: "3.15"
      - name: Run the eval gate
        run: python3 evals/run_evals.py candidate

#!/usr/bin/env python3
"""Grade scripted agent runs by outcome and by trajectory rules."""
GOAL = {"R-1042": "refunded"}
EXPECTED_SEQUENCE = ["get_order", "verify_identity", "issue_refund"]
 
RUNS = {
    "T1": (["get_order", "verify_identity", "issue_refund"], {"R-1042": "refunded"}, "Refund issued."),
    "T2": (["verify_identity", "get_order", "issue_refund"], {"R-1042": "refunded"}, "Refund issued."),
    "T3": (["get_order", "issue_refund", "verify_identity"], {"R-1042": "refunded"}, "Refund issued."),
    "T4": (["get_order", "verify_identity"], {"R-1042": "open"}, "Refund issued."),
}
 
 
def outcome_passed(final_state):
    return final_state == GOAL
 
 
def exact_sequence_passed(calls):
    return calls == EXPECTED_SEQUENCE
 
 
def policy_passed(calls):
    """Identity check must happen before any refund."""
    if "issue_refund" not in calls:
        return True
    return "verify_identity" in calls[: calls.index("issue_refund")]
 
 
def text_claims_success(reply):
    return "refund issued" in reply.lower()
 
 
print(f"{'run':<4}{'reply says done':<17}{'outcome':<9}{'exact seq':<11}{'policy':<8}verdict")
for run, (calls, state, reply) in RUNS.items():
    outcome, policy = outcome_passed(state), policy_passed(calls)
    verdict = "pass" if outcome and policy else "fail"
    print(f"{run:<4}{str(text_claims_success(reply)):<17}{str(outcome):<9}"
          f"{str(exact_sequence_passed(calls)):<11}{str(policy):<8}{verdict}")

run reply says done  outcome  exact seq  policy  verdict
T1  True             True     True       True    pass
T2  True             True     False      True    pass
T3  True             True     False      False   fail
T4  True             False    False      True    fail