Set up an eval harness¶

import json
import os
import re
import time
from dataclasses import dataclass, asdict
from pathlib import Path
from anthropic import Anthropic

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
MODEL = "claude-3-5-sonnet-20241022"
JUDGE_MODEL = "claude-3-5-sonnet-20241022"


@dataclass
class Result:
    id: str
    category: str
    prompt: str
    answer: str
    passed: bool
    failure_reason: str
    latency_ms: int


def call_model(prompt: str) -> tuple[str, int]:
    t0 = time.time()
    resp = client.messages.create(
        model=MODEL,
        max_tokens=400,
        messages=[{"role": "user", "content": prompt}],
    )
    latency_ms = int((time.time() - t0) * 1000)
    return resp.content[0].text, latency_ms


def evaluator_contains(answer: str, case: dict) -> tuple[bool, str]:
    needles = case.get("expected_contains", [])
    insensitive = case.get("case_insensitive", True)
    a = answer.lower() if insensitive else answer
    for n in needles:
        h = n.lower() if insensitive else n
        if h not in a:
            return False, f"missing substring: {n!r}"
    return True, ""


def evaluator_regex(answer: str, case: dict) -> tuple[bool, str]:
    pattern = case.get("expected_match")
    if not pattern:
        return True, ""
    flags = re.I if case.get("case_insensitive") else 0
    return (
        (True, "")
        if re.search(pattern, answer.strip(), flags)
        else (False, f"regex did not match: {pattern!r}")
    )


def evaluator_json(answer: str, case: dict) -> tuple[bool, str]:
    if not case.get("json_required"):
        return True, ""
    try:
        # Try to find JSON in the answer
        match = re.search(r"\{.*\}", answer, re.DOTALL)
        obj = json.loads(match.group(0)) if match else json.loads(answer)
    except (json.JSONDecodeError, AttributeError) as e:
        return False, f"not valid JSON: {e}"
    must = case.get("json_must_contain", {})
    for k, v in must.items():
        if obj.get(k) != v:
            return False, f"json[{k!r}] = {obj.get(k)!r}, expected {v!r}"
    return True, ""


def evaluator_judge(answer: str, case: dict) -> tuple[bool, str]:
    rubric = case.get("judge")
    if not rubric:
        return True, ""
    judge_prompt = f"""Evaluate the following answer against the rubric.

Rubric: {rubric}

Answer to evaluate:
<answer>{answer}</answer>

Reply with JSON: {{"pass": true|false, "reason": "..."}}"""
    resp = client.messages.create(
        model=JUDGE_MODEL,
        max_tokens=200,
        messages=[{"role": "user", "content": judge_prompt}],
    )
    try:
        match = re.search(r"\{.*\}", resp.content[0].text, re.DOTALL)
        verdict = json.loads(match.group(0))
        return verdict.get("pass", False), verdict.get("reason", "")
    except (json.JSONDecodeError, AttributeError):
        return False, "judge produced unparseable output"


EVALUATORS = [evaluator_contains, evaluator_regex, evaluator_json, evaluator_judge]


def run_one(case: dict) -> Result:
    answer, latency = call_model(case["prompt"])
    failures = []
    for fn in EVALUATORS:
        ok, why = fn(answer, case)
        if not ok:
            failures.append(why)
    return Result(
        id=case["id"],
        category=case["category"],
        prompt=case["prompt"],
        answer=answer,
        passed=not failures,
        failure_reason="; ".join(failures),
        latency_ms=latency,
    )


def main():
    cases = [json.loads(l) for l in Path("datasets/golden.jsonl").read_text().splitlines() if l.strip()]
    results = [run_one(c) for c in cases]

    # Print summary
    by_cat = {}
    for r in results:
        by_cat.setdefault(r.category, []).append(r)

    print(f"\n{'CATEGORY':<15} {'PASS':<8} {'TOTAL':<8} {'P95 ms':<8}")
    for cat, rs in sorted(by_cat.items()):
        passed = sum(1 for r in rs if r.passed)
        latencies = sorted(r.latency_ms for r in rs)
        p95 = latencies[int(len(latencies) * 0.95)] if latencies else 0
        print(f"{cat:<15} {passed:<8} {len(rs):<8} {p95:<8}")

    overall = sum(1 for r in results if r.passed)
    print(f"\nOverall: {overall}/{len(results)} passed ({100*overall/len(results):.1f}%)")

    # Save full report
    Path("results").mkdir(exist_ok=True)
    out = Path(f"results/run-{int(time.time())}.json")
    out.write_text(json.dumps([asdict(r) for r in results], indent=2))
    print(f"Detail saved to {out}")

    return 0 if overall == len(results) else 1


if __name__ == "__main__":
    import sys
    sys.exit(main())

import json
import sys
from pathlib import Path

def load(path):
    return {r["id"]: r for r in json.loads(Path(path).read_text())}

if len(sys.argv) != 3:
    print("Usage: diff.py baseline.json candidate.json")
    sys.exit(1)

base, cand = load(sys.argv[1]), load(sys.argv[2])

regressions = [i for i in base if base[i]["passed"] and not cand.get(i, base[i])["passed"]]
fixes = [i for i in cand if cand[i]["passed"] and not base.get(i, cand[i])["passed"]]

print(f"Regressions: {len(regressions)}")
for i in regressions:
    print(f"  {i}: {cand[i]['failure_reason']}")
print(f"\nFixes: {len(fixes)}")
for i in fixes:
    print(f"  {i}")

sys.exit(1 if regressions else 0)