Building a Production Eval Harness

An eval harness is the engineering infrastructure that makes evaluation a first-class, automated part of your development workflow rather than a periodic manual activity. It loads your golden dataset, runs each example through the current system, scores the outputs, aggregates results, and reports a clear pass or fail. When this runs on every pull request, quality regressions become as catchable as type errors.

Core components of an eval harness

• Dataset loader: reads examples from a versioned source (JSONL file, database, or dataset registry)
• Runner: calls the system under test for each example and captures the output with retries and error handling
• Scorer: compares outputs to expected results using exact match, embedding similarity, or an LLM judge
• Aggregator: computes pass rates, mean scores, and per-category breakdowns from individual scores
• Reporter: emits results in a format your CI system can consume (exit code, JUnit XML, GitHub check annotation)

A minimal but production-worthy harness

import json, asyncio
from pathlib import Path

async def run_eval(
    dataset_path: str,
    system_under_test,
    scorer,
    concurrency: int = 5,
) -> dict:
    examples = [json.loads(l) for l in Path(dataset_path).read_text().splitlines()]
    semaphore = asyncio.Semaphore(concurrency)

    async def run_one(example: dict) -> dict:
        async with semaphore:
            try:
                output = await system_under_test(example["input"])
                score = scorer(output=output, example=example)
            except Exception as e:
                output, score = None, 0.0
                print(f"Error on {example['id']}: {e}")
            return {"id": example["id"], "score": score, "output": output}

    results = await asyncio.gather(*[run_one(ex) for ex in examples])
    pass_rate = sum(r["score"] >= 0.7 for r in results) / len(results)
    return {
        "pass_rate": pass_rate,
        "mean_score": sum(r["score"] for r in results) / len(results),
        "n": len(results),
        "results": results,
    }

if __name__ == "__main__":
    results = asyncio.run(run_eval("data/golden.jsonl", my_pipeline, my_scorer))
    print(json.dumps(results, indent=2))
    raise SystemExit(0 if results["pass_rate"] >= 0.85 else 1)

Scoring strategies by task type

The scorer is the component that most needs to be tailored to your task. For classification and extraction tasks, exact match or schema validation works. For factual Q&A, string inclusion of key facts combined with an LLM check for faithfulness works. For open-ended generation, an LLM judge with a rubric is necessary. Build the scorer as a replaceable component so you can swap strategies without changing the rest of the harness.

Parallelism and cost management

Running a 200-example eval sequentially at 1-2 seconds per example takes several minutes. Adding concurrency reduces wall-clock time dramatically but requires careful throttling to avoid rate limit errors. Implement a semaphore or rate limiter that keeps request rate below your API quota, add exponential backoff on rate limit errors, and track the total token cost of each eval run in your budget.