07 - Evaluation framework automation: scoring pipeline, batch processing, async evaluation

Automating evaluation: parallel scoring, cache and a gate on the PR

You have an evaluation set of 500 questions, three retrieval systems to compare and two versions of the prompt. Multiply: 500 × 3 × 2 makes 3000 judgments, and if for each you ask for three metrics the judge calls become 9000. How long they take depends on a single quantity, the latency of your judge, which depends on model, hardware and prompt length and which I cannot give you. What you can do is reason about how the time scales, and that is what this episode builds: a model for computing time, an asynchronous evaluator with a concurrency limit and a cache, and a gate that blocks the change that worsens the metrics.

The listings run with the standard library plus numpy only, without any model: the judge is fake, with simulated latency, because what is being studied is the orchestration and not the verdict. For what to measure with that judge see the episode on faithfulness; for how to build the evaluation set, the one on the golden dataset; for the overall picture of an evaluation pipeline, this article from the AI Engineering series.

How much time parallelism saves: a model, not a measurement

Before writing the orchestration it is worth knowing what to expect. The listing generates 3000 judge latencies with a lognormal distribution (median 8 seconds, long tail: a few very slow calls) and computes the completion time with k workers, assigning each call to the first free slot. The generator seed is fixed, so the numbers are reproducible; but they are the numbers of a model and not of a machine. Replace the median and the dispersion with those of your judge and redo the calculation.

import heapq
import numpy as np

# Model, not measurement: judge latencies with a median of 8 s and a long tail (lognormal).
rng = np.random.default_rng(7)
latencies = rng.lognormal(mean=np.log(8.0), sigma=0.4, size=3000)   # seconds, 500 x 3 x 2


def total_time(durations, workers):
    """Completion time with `workers` parallel slots (assign to the first free one)."""
    free = [0.0] * workers
    heapq.heapify(free)
    for d in durations:
        start = heapq.heappop(free)
        heapq.heappush(free, start + d)
    return max(free)


print(f"sum of latencies: {latencies.sum() / 3600:.2f} hours")
for k in (1, 3, 5, 10):
    t = total_time(latencies, k)
    print(f"{k:2d} workers: {t / 60:7.1f} min   (ideal {latencies.sum() / k / 60:7.1f})")

# The cache removes calls, it does not make them faster: duplicate tasks are dropped
for share in (0.0, 0.2, 0.4):
    remaining = latencies[int(len(latencies) * share):]
    print(f"cache {share:.0%}: 5 workers -> {total_time(remaining, 5) / 60:6.1f} min")

# Output:
# sum of latencies: 7.13 hours
#  1 workers:   427.7 min   (ideal   427.7)
#  3 workers:   142.7 min   (ideal   142.6)
#  5 workers:    85.7 min   (ideal    85.5)
# 10 workers:    43.0 min   (ideal    42.8)
# cache 0%: 5 workers ->   85.7 min
# cache 20%: 5 workers ->   69.5 min
# cache 40%: 5 workers ->   52.3 min

The result shows two facts. First, in this model the time scales almost as 1/k: with five workers the wait drops from about 428 to about 86 minutes, because the calls are independent and the ideal value (sum divided by k) is almost entirely reached. Second, in reality you don't choose that "k" freely: it is set by the judge's server, which beyond a certain concurrency serializes internally or rejects requests, and the gain stops. The right concurrency is found by measuring throughput as k grows on your hardware, not by copying someone else's number. The cache lines show that removing 40% of the calls takes the five-worker case from 86 to 52 minutes: the cache doesn't make the call fast, it avoids it.

The asynchronous evaluator: semaphore, cache and the problem of simultaneous duplicates

An asyncio.Semaphore is an internal counter: each acquire() decrements it, each release() increments it, and it never goes below zero. Whoever finds zero waits. Used with async with it limits to N the simultaneous calls to the judge, whatever the number of tasks launched with gather. This protects you from a common mistake: launching 3000 requests at once and getting cascading timeouts.

The cache is more insidious. The natural implementation checks a dictionary, calls the judge if the key is missing, and saves the result when the call ends. But with gather all the tasks start before the first one finishes, so none ever finds a key already present: the duplicates wait in parallel and all pay for the call. The remedy, called single-flight, is to store in the cache the Task just created and not the result: whoever arrives later awaits the same Task. In the listing the difference reads in two lines, cache[k] = asyncio.create_task(...) against cache[k] = await .... Keeping the Task in the dictionary also serves a second purpose: the Python documentation warns that the event loop keeps only weak references to tasks, and a task that is not referenced can disappear mid-execution.

import asyncio
import hashlib
import json


class Judge:
    """Fake judge: no model, just simulated latency and a score derived from the hash."""

    def __init__(self, concurrency):
        self.sem = asyncio.Semaphore(concurrency)
        self.in_flight = 0
        self.peak = 0
        self.calls = 0

    async def evaluate(self, item, metric):
        async with self.sem:
            self.in_flight += 1
            self.peak = max(self.peak, self.in_flight)
            self.calls += 1
            await asyncio.sleep(0.01)                    # the "model" takes time
            self.in_flight -= 1
        h = hashlib.sha256(f"{item}|{metric}".encode()).digest()
        return h[0] / 255


def key(item, metric):
    return hashlib.sha256(json.dumps([item, metric], sort_keys=True).encode()).hexdigest()


async def run(items, metrics, concurrency, single_flight):
    judge = Judge(concurrency)
    cache = {}          # key -> result (naive) or Task (single-flight)

    async def one(i, m):
        k = key(i, m)
        if single_flight:
            if k not in cache:                           # the Task goes into the cache IMMEDIATELY
                cache[k] = asyncio.create_task(judge.evaluate(i, m))
            return await cache[k]
        if k in cache:
            return cache[k]
        cache[k] = await judge.evaluate(i, m)            # it enters the cache only when the call ends
        return cache[k]

    results = await asyncio.gather(*(one(i, m) for i in items for m in metrics))
    return judge, results


# 100 answers, but only 40 are distinct: it happens when the same test runs on several variants
items = [f"answer-{i % 40}" for i in range(100)]
metrics = ["faithfulness", "context_precision"]
for sf in (False, True):
    j, r = asyncio.run(run(items, metrics, concurrency=5, single_flight=sf))
    name = "single-flight" if sf else "naive cache  "
    print(f"{name}: requests={len(r)} judge calls={j.calls} peak concurrency={j.peak}")

# Output:
# naive cache  : requests=200 judge calls=200 peak concurrency=5
# single-flight: requests=200 judge calls=80 peak concurrency=5

The outcome is clear-cut. With the naive cache the 200 requests (100 answers × 2 metrics, of which only 80 are distinct) produce 200 judge calls, that is, zero saving; with single-flight the calls are 80. In both cases the observed peak concurrency is 5, the value of the semaphore, confirming that the limit holds. The judge here is fake: the score derives from the hash of the input, so the same answer always receives the same score, which is the property a cache requires. With a real LLM at positive temperature two identical calls can give different results, and the cache fixes the first one: it is a conscious compromise, not a defect.

What you put in the key decides when the cache is correct:

• the text of the question, the context and the answer, because they change the verdict;
• the name of the metric and the version of its judging prompt;
• the judge model and its version, because if you change it without knowing, the scores are no longer comparable;
• not the time or a run identifier, which would make every key unique.

The practical rule is that the key contains everything that, if it changed, would change the judge's answer, and nothing else. A key that is too poor returns old scores for new inputs, the hardest error to notice; one that is too rich wipes out the savings. Sorting the JSON keys, as sort_keys=True does, prevents the same content from producing two different hashes. For transient failures add a limited number of retries and a per-call timeout with asyncio.wait_for, because a blocked call occupies a semaphore slot forever.

The gate: blocking the change that worsens the metrics

Automating the measurement serves to decide. The gate is a script that compares the results using two criteria: an absolute minimum for each metric and a maximum regression against the baseline, that is, the last accepted version. The second criterion avoids a defect of the absolute minimum alone: a metric can fall from 0.95 to 0.81 without ever violating the 0.80 threshold, and the change that caused it would pass. The script returns an exit code, 0 or 1, which is all that continuous integration needs to know.

THRESHOLDS = {"faithfulness": 0.80, "context_precision": 0.75}   # absolute minimums
MAX_DROP = 0.03                                                   # regression allowed against the baseline


def gate(results, baseline):
    """Return (exit_code, reasons). 0 = ok, 1 = block the merge."""
    reasons = []
    for metric, minimum in THRESHOLDS.items():
        now, before = results[metric], baseline[metric]
        if now < minimum:
            reasons.append(f"{metric} {now:.3f} below the minimum {minimum:.2f}")
        if before - now > MAX_DROP:
            reasons.append(f"{metric} dropped by {before - now:.3f} against the baseline")
    return (1 if reasons else 0), reasons


baseline = {"faithfulness": 0.86, "context_precision": 0.80}
for name, run in [("pr-a", {"faithfulness": 0.85, "context_precision": 0.79}),
                  ("pr-b", {"faithfulness": 0.79, "context_precision": 0.80})]:
    code, reasons = gate(run, baseline)
    print(name, "exit", code, reasons)

# Output:
# pr-a exit 0 []
# pr-b exit 1 ['faithfulness 0.790 below the minimum 0.80', 'faithfulness dropped by 0.070 against the baseline']

The first change passes. The second violates both criteria: faithfulness is 0.79, below the minimum of 0.80, and has dropped by 0.07 against the baseline. The thresholds in the listing (0.80, 0.75, allowed drop 0.03) are examples: calibrate the tolerance on the variability of your metric at constant code, repeating the measurement several times, otherwise a gate that is too tight blocks on noise and ends up disabled by whoever it annoys.

The outline of the flow on GitHub Actions is short: it starts only when the paths that affect quality change, runs the evaluation on a small, fast subset, then launches the gate. For it to really block the merge, the check must be set as required in the branch protection rules: the GitHub documentation explains that required checks must have a successful, skipped or neutral status before the protected branch can be modified. The subset for PRs and the full nightly set answer two different questions: "did I break something?" and "how is the system doing overall?".

# Schema to adapt, not executed: the folders and scripts are those of your repository.
name: rag-eval-gate
on:
  pull_request:
    paths: ["retrieval/**", "prompts/**", "eval/**"]   # it starts only if something relevant changes
jobs:
  eval:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r eval/requirements.txt
      - run: python eval/run_eval.py --dataset eval/golden_ci.json --out eval/results.json
      - run: python eval/gate.py eval/results.json eval/baseline.json   # exit 1 = failed job

Tracking runs, and when you don't need it

A score without context says nothing three months later: you need to know with which dataset, which prompt, which judge model it was obtained. Tracking tools such as MLflow offer functions for this; its documentation describes start_run to open a run and log_param and log_metric to record parameters and metrics. For a small team a versioned JSON file next to the code, with the parameters and scores of each run, does the same job without a service to maintain. Switch to a dedicated tool when you compare many runs and need an interface to do so, not before.

Not everything should be automated right away. With a set of fifty questions and a fast judge, the manual cycle takes minutes and the infrastructure would cost more than the problem. The reasonable order is: first the reproducible evaluation script, then the cache, then parallelism, then the gate, finally tracking. Each step is justified by a real pain you have already felt.

Frequently asked questions

How many concurrent workers should I use?

As many as the judge's server can handle without serializing or rejecting requests. You find it by measuring throughput as concurrency grows on your own hardware and stopping where it stops increasing. The value depends on model, hardware and prompt length.

Can the cache distort the results?

Yes, if the key omits something that changes the verdict, such as the version of the prompt or of the judge model. A correct cache contains in the key everything that influences the answer and nothing else, and is emptied when you change the judge.

Why do I need the gate if I already look at the numbers?

Because looking at the numbers by hand requires someone to remember to do it at every change. The gate makes the regression a merge condition, not a habit, and it is traceable in the PR history.

Is an LLM judge deterministic?

Not necessarily. Lowering the temperature reduces variability but does not always eliminate it. That is why gates use tolerances calibrated on the measured variability and not exact comparisons.

What you can do now

You can estimate how the time of an evaluation changes as concurrency varies, limit the calls with a semaphore, prevent the cache from failing with simultaneous duplicates, write a gate with an absolute minimum and a maximum regression and attach it to a PR. You also know which of these pieces to wait on until they are needed. In the next episode, evaluation-driven optimization, you use this infrastructure to compare variants with a statistical test.

The same series also includes production monitoring.