Problem 129 · Practical builds

LLM Eval Runner with Caching and a Regression Gate

HardPractical multi-stage build

01Problem

Build an eval harness that runs a model over a test set, survives flaky APIs, avoids paying twice for the same call, and decides whether a candidate model may ship relative to a baseline.

Stages — finish each before reading the next

1Eval runner

run(cases) calls call(model, prompt) per case and grades output against expected (exact match by default), returning one Result per case in input order.

2Parallel with retries

Run cases on a thread pool. Retry transient errors (TimeoutError, ConnectionError) with exponential backoff via an injected sleep; record permanent errors immediately without retrying.

3Result caching

Cache successful outputs by SHA-256 of (model, prompt) so reruns skip the API; failures are never cached. Results report whether they came from cache.

4Regression gate

regression_gate(baseline, candidate, margin, z) pairs results by case id and computes a confidence interval on the per-case pass-rate difference. It passes only if the lower bound is at least -margin (non-inferiority).

02Key insight

Two design calls matter. Cache on a hash of everything that determines the output (model and prompt; add temperature and system prompt in practice) and never cache errors. And gate on a paired interval, not two raw pass rates: both models see the same cases, so per-case differences remove case difficulty from the variance and a small eval set cannot wave through a real regression.

03Solution

solution.pypython97 lines

import hashlib
import math
import threading
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Callable, Optional


@dataclass(frozen=True)
class Case:
    id: str
    prompt: str
    expected: str


@dataclass(frozen=True)
class Result:
    case_id: str
    output: Optional[str]
    passed: bool
    attempts: int
    cached: bool
    error: Optional[str] = None


def cache_key(prompt: str, model: str) -> str:
    return hashlib.sha256(f"{model}\x00{prompt}".encode("utf-8")).hexdigest()


def exact_match(output: str, expected: str) -> bool:
    return output.strip() == expected.strip()


class EvalRunner:
    def __init__(self, model: str, call: Callable[[str, str], str],
                 grader: Callable[[str, str], bool] = exact_match, max_workers: int = 4,
                 max_attempts: int = 3, retry_on: tuple = (TimeoutError, ConnectionError),
                 sleep: Callable[[float], None] = lambda s: None, backoff_s: float = 0.5) -> None:
        self.model, self.call, self.grader = model, call, grader
        self.max_workers, self.max_attempts, self.retry_on = max_workers, max_attempts, retry_on
        self.sleep, self.backoff_s = sleep, backoff_s
        self.cache: dict[str, str] = {}
        self.calls = 0
        self._lock = threading.Lock()

    def _one(self, case: Case) -> Result:
        key = cache_key(case.prompt, self.model)
        with self._lock:
            hit = self.cache.get(key)
        if hit is not None:
            return Result(case.id, hit, self.grader(hit, case.expected), 0, True)
        err: Optional[BaseException] = None
        for attempt in range(1, self.max_attempts + 1):
            with self._lock:
                self.calls += 1
            try:
                out = self.call(self.model, case.prompt)
            except self.retry_on as e:          # transient: back off and retry
                err = e
                if attempt < self.max_attempts:
                    self.sleep(self.backoff_s * 2 ** (attempt - 1))
                continue
            except Exception as e:              # permanent: record, never cache
                return Result(case.id, None, False, attempt, False, f"{type(e).__name__}: {e}")
            with self._lock:
                self.cache[key] = out          # only successful outputs are cached
            return Result(case.id, out, self.grader(out, case.expected), attempt, False)
        return Result(case.id, None, False, self.max_attempts, False, f"{type(err).__name__}: {err}")

    def run(self, cases: list[Case]) -> list[Result]:
        with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
            return list(pool.map(self._one, cases))  # map keeps input order


@dataclass(frozen=True)
class Gate:
    passed: bool
    delta: float
    lo: float
    hi: float
    n: int


def regression_gate(baseline: list[Result], candidate: list[Result],
                    margin: float = 0.02, z: float = 1.96) -> Gate:
    """Paired non-inferiority: pass only if the CI's lower bound on (cand - base) >= -margin."""
    b = {r.case_id: r.passed for r in baseline}
    c = {r.case_id: r.passed for r in candidate}
    shared = sorted(b.keys() & c.keys())
    if not shared:
        raise ValueError("no shared cases")
    d = [int(c[i]) - int(b[i]) for i in shared]
    n = len(d)
    mean = sum(d) / n
    var = sum((x - mean) ** 2 for x in d) / (n - 1) if n > 1 else 0.0
    half = z * math.sqrt(var / n)
    return Gate(mean - half >= -margin, mean, mean - half, mean + half, n)

Try it first.

TimeO(n) calls on a cold cache, O(n / workers) wall time; gate O(n)

SpaceO(n + cache size)

tests.py · 15 passing

tests.pypython41 lines

import threading as _t
cases = [Case(f"c{i}", f"q{i}", f"a{i}") for i in range(20)]
fails: dict[str, int] = {"q3": 2, "q7": 5}   # transient failures before success
seen: dict[str, int] = {}
lock = _t.Lock()
def model_call(model: str, prompt: str) -> str:
    with lock:
        seen[prompt] = seen.get(prompt, 0) + 1
        n = seen[prompt]
    if prompt == "q5":
        raise ValueError("content filter")
    if n <= fails.get(prompt, 0):
        raise TimeoutError("slow")
    return "a" + prompt[1:] if prompt != "q9" else "wrong"
sleeps: list[float] = []
r = EvalRunner("m1", model_call, max_workers=8, max_attempts=3, sleep=sleeps.append)
res = r.run(cases)
assert [x.case_id for x in res] == [c.id for c in cases]          # order preserved
by = {x.case_id: x for x in res}
assert by["c3"].passed and by["c3"].attempts == 3
assert not by["c7"].passed and by["c7"].error == "TimeoutError: slow" and by["c7"].attempts == 3
assert by["c5"].attempts == 1 and by["c5"].error.startswith("ValueError")
assert not by["c9"].passed and by["c9"].output == "wrong"
assert sum(x.passed for x in res) == 17
assert r.calls == 20 + 2 + 2 and sorted(sleeps) == [0.5, 0.5, 1.0, 1.0]
# cache: rerun only re-calls the cases that errored
res2 = r.run(cases)
assert r.calls == 24 + 1 + 3 and sum(x.cached for x in res2) == 18
assert by["c7"].passed is False and {x.case_id: x for x in res2}["c7"].passed  # q7 succeeds on call 6
assert cache_key("p", "m1") != cache_key("p", "m2") != cache_key("m1", "p")
# regression gate
def mk(passes: list[bool]) -> list[Result]:
    return [Result(f"c{i}", "", p, 1, False) for i, p in enumerate(passes)]
base = mk([True] * 100)
g = regression_gate(base, mk([True] * 100))
assert g.passed and g.delta == 0 and g.lo == 0
g = regression_gate(base, mk([False] + [True] * 99))
assert abs(g.delta + 0.01) < 1e-12 and abs(g.lo + 0.0296) < 1e-9 and not g.passed
assert regression_gate(base, mk([False] + [True] * 99), margin=0.05).passed
assert not regression_gate(base, mk([False] * 20 + [True] * 80), margin=0.05).passed
assert regression_gate(mk([False] * 10 + [True] * 90), mk([True] * 100)).passed

04Staff-level follow-up

How would you run this in CI for real? Persist the cache keyed by a full request fingerprint, set temperature 0 or sample k times per case, grade with an LLM judge whose own agreement is measured, and size the eval set so the CI half-width is below the margin you care about (about 1.96 * sqrt(var / n)).