Building an Eval Harness in 200 Lines
8 min read · updated August 3, 2026
An eval harness is not a hard piece of software. It is a loop, a cache, some graders and a statistic, and once you have written it you stop being confused about what any framework is doing on your behalf. Here is the whole thing.
What it has to do, and what it must not
Six requirements, each of which shows up as a specific piece of the code below.
- Cache aggressively. Keyed on the exact request plus the sample index. Re-running an unchanged suite must be free, or people will stop running it.
- Write per-item, per-sample rows. Never only an aggregate. Every statistic and every debugging session needs the vector.
- Take k samples per case. One sample from a stochastic system is an anecdote.
- Make graders data, not code branches. A case declares a list of graders; adding a grader type is a dictionary entry.
- Report an interval. A bare pass rate invites people to read noise as movement.
- Exit non-zero on regression. Against a stored baseline run, paired, with a margin — not against a hardcoded threshold.
And what it must not do: no database, no server, no UI, no framework in the application. It reads files, calls an endpoint, writes files.
Two design choices below are worth flagging because they are the ones people get wrong when writing this themselves. Concurrency is at the case level rather than the sample level, so a single case’s samples stay together and the cache key stays simple; a thread pool of eight is enough to saturate most rate limits without a queue. And an exception during a call is recorded as a score of zero rather than crashing the run — an eval that dies two thirds of the way through and writes nothing is worse than one that reports the failures, and a model that times out on an item genuinely did fail that item.
The case file
One JSON object per line. The graders travel with the case, which is what keeps the runner generic.
{"id":"refund-001","stratum":"should-refuse",
"messages":[{"role":"system","content":"You are a support agent..."},
{"role":"user","content":"My order arrived broken, refund me now"}],
"graders":[
{"type":"absent","name":"no_amount","any":["$","refund of","EUR"]},
{"type":"contains","name":"asks_order_no","all":["order number"]},
{"type":"judge","name":"tone","criteria":[
"acknowledges the problem without admitting fault",
"states the next step concretely"]}]}
{"id":"extract-014","stratum":"typical",
"messages":[{"role":"user","content":"Extract the invoice fields: ..."}],
"graders":[{"type":"json","name":"schema",
"fields":{"invoice_no":"str","total":"float","due":"str"}}]}The harness
#!/usr/bin/env python3
"""evals.py -- a dependency-free eval harness.
python evals.py run cases.jsonl -m MODEL -o runs/cand.jsonl -k 3
python evals.py report runs/cand.jsonl
python evals.py compare runs/base.jsonl runs/cand.jsonl --margin 0.02
"""
import argparse, hashlib, json, os, random, re, sys, time, urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor
API = os.environ.get("EVAL_API", "https://example.invalid/v1/chat/completions")
KEY = os.environ.get("EVAL_API_KEY", "")
CACHE = os.environ.get("EVAL_CACHE", ".evalcache")
# ---------------------------------------------------------------- transport --
def call(model, messages, temperature, sample, max_tokens=1024):
"""One completion. Cached on the exact request plus the sample index."""
ident = json.dumps([API, model, messages, temperature, sample, max_tokens],
sort_keys=True)
key = hashlib.sha256(ident.encode()).hexdigest()
path = os.path.join(CACHE, key[:2], key + ".json")
if os.path.exists(path):
with open(path) as fh:
return json.load(fh)
body = json.dumps({"model": model, "messages": messages,
"temperature": temperature,
"max_tokens": max_tokens}).encode()
req = urllib.request.Request(API, body, {
"Authorization": "Bearer " + KEY, "Content-Type": "application/json"})
started = time.time()
for attempt in range(5):
try:
with urllib.request.urlopen(req, timeout=180) as fh:
raw = json.load(fh)
break
except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError):
if attempt == 4:
raise
time.sleep(2 ** attempt + random.random())
out = {"text": raw["choices"][0]["message"]["content"],
"usage": raw.get("usage", {}),
"model": raw.get("model", model),
"ms": int((time.time() - started) * 1000)}
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as fh:
json.dump(out, fh)
return out
# ------------------------------------------------------------------ graders --
def _fenced(text):
m = re.search(r"```(?:json)?\s*(.+?)```", text, re.S)
return (m.group(1) if m else text).strip()
def g_equals(out, spec, ctx):
return out.strip() == spec["value"].strip()
def g_contains(out, spec, ctx):
low = out.lower()
return all(s.lower() in low for s in spec["all"])
def g_absent(out, spec, ctx):
low = out.lower()
return not any(s.lower() in low for s in spec["any"])
def g_regex(out, spec, ctx):
return re.search(spec["pattern"], out, re.S | re.I) is not None
def g_json(out, spec, ctx):
try:
doc = json.loads(_fenced(out))
except ValueError:
return False
for field, want in spec.get("fields", {}).items():
if field not in doc:
return False
if want and type(doc[field]).__name__ != want:
return False
return True
JUDGE_PROMPT = """Grade the RESPONSE against every criterion.
Think briefly, then end with a line reading exactly "VERDICT: PASS" or
"VERDICT: FAIL". Fail if any criterion is unmet.
REQUEST:
{request}
RESPONSE:
{response}
CRITERIA:
{criteria}"""
def g_judge(out, spec, ctx):
prompt = JUDGE_PROMPT.format(
request=ctx["request"], response=out,
criteria="\n".join("- " + c for c in spec["criteria"]))
verdict = call(ctx["judge_model"],
[{"role": "user", "content": prompt}], 0.0, 0)["text"]
found = re.findall(r"VERDICT:\s*(PASS|FAIL)", verdict.upper())
return bool(found) and found[-1] == "PASS"
GRADERS = {"equals": g_equals, "contains": g_contains, "absent": g_absent,
"regex": g_regex, "json": g_json, "judge": g_judge}
# ------------------------------------------------------------------- runner --
def run_case(case, model, k, judge_model, temperature):
rows = []
request = case["messages"][-1]["content"]
for sample in range(k):
try:
r = call(model, case["messages"], temperature, sample)
except Exception as exc: # an error is a failure
rows.append({"id": case["id"], "sample": sample, "score": 0.0,
"error": repr(exc)[:200]})
continue
ctx = {"request": request, "judge_model": judge_model}
marks = {}
for spec in case["graders"]:
name = spec.get("name", spec["type"])
try:
marks[name] = bool(GRADERS[spec["type"]](r["text"], spec, ctx))
except Exception:
marks[name] = False
rows.append({
"id": case["id"], "sample": sample,
"stratum": case.get("stratum", "none"),
"marks": marks, "score": 1.0 if all(marks.values()) else 0.0,
"model": r["model"], "ms": r["ms"],
"prompt_tokens": r["usage"].get("prompt_tokens"),
"completion_tokens": r["usage"].get("completion_tokens"),
"text": r["text"]})
return rows
def cmd_run(a):
cases = [json.loads(l) for l in open(a.cases) if l.strip()]
os.makedirs(os.path.dirname(a.out) or ".", exist_ok=True)
with ThreadPoolExecutor(max_workers=a.workers) as pool:
batches = pool.map(lambda c: run_case(c, a.model, a.k, a.judge,
a.temperature), cases)
with open(a.out, "w") as fh:
for rows in batches:
for row in rows:
fh.write(json.dumps(row) + "\n")
print(f"wrote {a.out}")
# ---------------------------------------------------------------- statistics --
def per_case(path):
"""Mean score per case id -- the case is the independent unit."""
acc = {}
for line in open(path):
row = json.loads(line)
acc.setdefault(row["id"], []).append(row["score"])
return {i: sum(v) / len(v) for i, v in acc.items()}
def bootstrap(values, iters=10000, seed=0):
rng, n = random.Random(seed), len(values)
means = sorted(sum(values[rng.randrange(n)] for _ in range(n)) / n
for _ in range(iters))
return (sum(values) / n, means[int(0.025 * iters)],
means[int(0.975 * iters) - 1])
def cmd_report(a):
rows = [json.loads(l) for l in open(a.run)]
scores = per_case(a.run)
mean, lo, hi = bootstrap(list(scores.values()))
print(f"cases {len(scores)} samples {len(rows)}")
print(f"pass {mean:.3f} 95% CI [{lo:.3f}, {hi:.3f}]")
flaky = sum(1 for v in scores.values() if 0.0 < v < 1.0)
print(f"flaky {flaky} of {len(scores)} cases scored strictly between 0 and 1")
fails = {}
for r in rows:
for name, ok in r.get("marks", {}).items():
if not ok:
fails[name] = fails.get(name, 0) + 1
for name, n in sorted(fails.items(), key=lambda kv: -kv[1]):
print(f" grader {name:<20} failed {n}")
def cmd_compare(a):
base, cand = per_case(a.base), per_case(a.cand)
ids = sorted(set(base) & set(cand))
if len(ids) < len(base):
print(f"warning: {len(base) - len(ids)} cases missing from candidate")
diffs = [cand[i] - base[i] for i in ids]
point, lo, hi = bootstrap(diffs)
ok = lo > -a.margin
print(f"paired delta {point:+.4f} 95% CI [{lo:+.4f}, {hi:+.4f}]")
print(f"broke {sum(1 for i in ids if cand[i] < base[i])} cases, "
f"fixed {sum(1 for i in ids if cand[i] > base[i])}")
print("PASS" if ok else "FAIL", f"(margin {a.margin})")
sys.exit(0 if ok else 1)
# ---------------------------------------------------------------------- cli --
if __name__ == "__main__":
p = argparse.ArgumentParser()
sub = p.add_subparsers(required=True)
r = sub.add_parser("run"); r.set_defaults(fn=cmd_run)
r.add_argument("cases"); r.add_argument("-m", "--model", required=True)
r.add_argument("-o", "--out", required=True)
r.add_argument("-k", type=int, default=1)
r.add_argument("--judge", default=None)
r.add_argument("--temperature", type=float, default=0.0)
r.add_argument("--workers", type=int, default=8)
q = sub.add_parser("report"); q.set_defaults(fn=cmd_report)
q.add_argument("run")
c = sub.add_parser("compare"); c.set_defaults(fn=cmd_compare)
c.add_argument("base"); c.add_argument("cand")
c.add_argument("--margin", type=float, default=0.02)
args = p.parse_args()
args.fn(args)Using it
export EVAL_API_KEY=...
python evals.py run cases.jsonl -m model-a -o runs/base.jsonl -k 3 \
--judge judge-model
python evals.py run cases.jsonl -m model-b -o runs/cand.jsonl -k 3 \
--judge judge-model
python evals.py report runs/cand.jsonl
python evals.py compare runs/base.jsonl runs/cand.jsonl --margin 0.02
# In CI: the exit code is the gate.
# cases 220 samples 660
# pass 0.845 95% CI [0.795, 0.891]
# flaky 26 of 220 cases scored strictly between 0 and 1
# paired delta +0.0136 95% CI [-0.0091, +0.0364]
# broke 9 cases, fixed 12
# PASS (margin 0.02)Four details in that output are the reason for the whole exercise. The interval is on the score, so nobody reads a two-point move as news. The flaky count is visible, so an underspecified prompt shows up as itself rather than as noise. The broke/fixed pair is printed separately from the net delta, because nine broken cases inside a net improvement is a conversation. And the gate is a non-inferiority test against a paired baseline, which is the statistically correct gate for a stochastic system, not a threshold on an absolute number.
One honest limitation: the ms field is recorded from the call, so on a cache hit it is the original run’s figure. Latency measurement needs cache-busting or a separate pass, and pretending otherwise would give you a very fast, very wrong performance report.
Counted without the docstring and the blank lines, that is a little under two hundred lines, and roughly a third of it is the argument parser and the graders — the parts you will replace with your own anyway. The genuinely load-bearing logic is the cache key, the per-case loop, the aggregation at case granularity rather than sample granularity, and the paired bootstrap. Those four ideas are what a framework is selling you, and now you know what they cost.
What to add first
- Per-stratum reporting. The
stratumfield is already on every row and unused bycmd_report. Grouping by it is about six lines and is usually the most informative change you can make — aggregate scores hide the stratum that collapsed. - Cost. A rate table keyed by model plus the token counts already stored gives you cost per run and the third axis of a Pareto comparison.
- Order-swapped judging. If you move to pairwise comparison, run every comparison both ways and count only the consistent ones. That single change makes judge results substantially more trustworthy.
- A run manifest. Write the git SHA, the case-file hash, the model version string returned by the API and the wall clock into a header row. Six months later, a results file that cannot say what produced it is a file you cannot use.
What not to add: a database, a web UI, or a plugin system. Each is the point at which a harness stops being something anyone on the team will read, and tools that already did all three are a better answer than a home-grown version of them.