Creating a Repeatable Evaluation Pipeline
What "Repeatable" Actually Requires
Every technique in this unit — dependency-injected, mockable client code (Lessons 1-3), schema and tool-routing tests (Lessons 4-5), a versioned dataset (Lesson 6), the accuracy/consistency/failure-rate metrics (Lesson 7), regression comparison against a baseline (Lesson 8), and a place for human review (Lesson 9) — has so far been demonstrated as an independent piece. A repeatable pipeline is what ties these pieces into a single process that runs the same way every time, produces the same kind of report every time, and can be triggered automatically whenever code, prompts, or models change — the same discipline that makes ordinary software CI trustworthy, applied to a system that includes a model.
Repeatability here has a precise meaning: given the same inputs (same code, same dataset, same model version), the pipeline produces the same decision (pass or fail) through the same steps, in the same order, without a person needing to remember to run any particular check manually.
The Two Stages, Run in Sequence
A repeatable pipeline for an AI application runs two distinct kinds of checks, in a specific order, because the second stage is expensive and pointless to run if the first stage is broken.
- Unit and integration tests (Lessons 1-5) — fast, free, deterministic. If these fail, there is a code bug, and there is no reason to spend money running an evaluation against broken code.
- Evaluation run (Lessons 6-8) — slower, costs real money, measures model output quality. This only runs once the first stage passes.
import subprocess
import sys
def run_unit_tests() -> bool:
"""Runs the fast, mock-based test suite. Returns True if all tests pass."""
result = subprocess.run(
[sys.executable, "-m", "pytest", "-m", "not integration", "-q"],
capture_output=True,
text=True,
)
print(result.stdout)
if result.returncode != 0:
print(result.stderr, file=sys.stderr)
return result.returncode == 0
def test_run_unit_tests_reports_failure_on_nonzero_exit_code(monkeypatch):
class FakeCompletedProcess:
returncode = 1
stdout = "1 failed, 3 passed"
stderr = "AssertionError in test_something"
monkeypatch.setattr(subprocess, "run", lambda *a, **k: FakeCompletedProcess())
passed = run_unit_tests()
assert passed is False
print("PASS: run_unit_tests correctly reports failure from a nonzero exit code")
run_unit_tests wraps a pytest invocation as a Python function specifically so the pipeline can make a decision (continue or stop) based on its result, rather than only printing output for a human to read. The test uses monkeypatch (Lesson 3) to simulate a failing test run without actually needing a failing test to exist on disk — the same fake-the-dependency technique used throughout this unit, now applied to testing the pipeline's own control flow.
Running the Evaluation Stage and Producing a Verdict
The evaluation stage reuses the dataset, metrics, and comparison functions built earlier in this unit. Structuring it as a function that returns a clear pass/fail verdict — not just a printed report — is what allows the pipeline to act on the result automatically.
def run_evaluation_stage(dataset: list[dict], predict_fn, baseline_metrics: dict) -> dict:
"""predict_fn takes an input string and returns a predicted category or None on failure."""
results = []
raw_results = []
for example in dataset:
try:
prediction = predict_fn(example["input"])
except Exception:
prediction = None
raw_results.append({"predicted_category": prediction})
results.append({
"expected_category": example["expected_category"],
"predicted_category": prediction,
})
candidate_metrics = {
"accuracy": compute_accuracy(results),
"failure_rate": compute_failure_rate(raw_results),
}
comparison = compare_against_baseline(baseline_metrics, candidate_metrics)
return {"metrics": candidate_metrics, "comparison": comparison}
def test_run_evaluation_stage_passes_when_predictions_match_expectations():
dataset = [
{"input": "I was charged twice", "expected_category": "billing"},
{"input": "App crashes on upload", "expected_category": "technical"},
]
def perfect_predict_fn(text: str) -> str:
return "billing" if "charged" in text else "technical"
baseline_metrics = {"accuracy": 1.0, "failure_rate": 0.0}
outcome = run_evaluation_stage(dataset, perfect_predict_fn, baseline_metrics)
assert outcome["metrics"]["accuracy"] == 1.0
assert outcome["comparison"]["passed"] is True
print("PASS: run_evaluation_stage passes when the candidate matches the baseline")
def test_run_evaluation_stage_records_exceptions_as_failures():
dataset = [{"input": "anything", "expected_category": "billing"}]
def broken_predict_fn(text: str) -> str:
raise RuntimeError("simulated API timeout")
baseline_metrics = {"accuracy": 0.9, "failure_rate": 0.0}
outcome = run_evaluation_stage(dataset, broken_predict_fn, baseline_metrics)
assert outcome["metrics"]["failure_rate"] == 1.0
assert outcome["comparison"]["passed"] is False
print("PASS: run_evaluation_stage treats an exception as a recorded failure, not a crash")
The try/except inside the loop is deliberate and important: a single failing prediction (a timeout, a malformed response) must not crash the entire evaluation run and lose every other result — it should be recorded as exactly the kind of failure compute_failure_rate (Lesson 7) is designed to count. predict_fn is itself injected as a parameter, following the same dependency-injection principle from Lesson 2 — in a real pipeline it would wrap a call to the actual OpenAI client, while these tests pass in simple, controllable Python functions to verify the pipeline's control flow without spending any real API budget on testing the pipeline itself.
Assembling the Full Pipeline
def run_full_pipeline(dataset: list[dict], predict_fn, baseline_metrics: dict) -> int:
"""Returns a process exit code: 0 for success, 1 for failure."""
print("Stage 1: running unit tests...")
if not run_unit_tests():
print("Unit tests failed. Aborting before running the evaluation stage.")
return 1
print("Stage 2: running evaluation against baseline...")
outcome = run_evaluation_stage(dataset, predict_fn, baseline_metrics)
if not outcome["comparison"]["passed"]:
print(f"Evaluation regression detected: {outcome['comparison']}")
return 1
print(f"Pipeline passed. Metrics: {outcome['metrics']}")
return 0
This function is the pipeline's single entry point, and its structure encodes the ordering rule stated earlier: it returns immediately after Stage 1 if unit tests fail, never spending money on Stage 2 for code that is already known to be broken. The int return value (0 or 1) matters specifically because it is the same convention a shell or a CI system uses to decide whether a job succeeded — this is what lets the function back a real command-line entry point.
Wiring It Into Continuous Integration
A repeatable pipeline is only truly repeatable once it runs without a person remembering to trigger it. A CI configuration (illustrated here for GitHub Actions) runs run_full_pipeline automatically on relevant changes.
name: AI Pipeline
on:
pull_request:
paths:
- "app/**"
- "prompts/**"
- "eval/**"
jobs:
test-and-evaluate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
- run: pip install -r requirements.txt
- name: Run unit tests and evaluation pipeline
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: python run_pipeline.py
The paths filter ensures the pipeline only runs on pull requests that actually touch application code, prompts, or evaluation configuration — avoiding unnecessary API spend on unrelated changes such as documentation edits. OPENAI_API_KEY is injected from encrypted repository secrets, never committed to source control, so the evaluation stage's real API calls can run securely inside CI. If run_pipeline.py (a thin script calling run_full_pipeline and calling sys.exit with its return value) exits nonzero, the CI job fails, blocking the pull request from merging until the regression is addressed — turning the entire discipline built across this unit into an automatic gate rather than a manual, easily-skipped step.
Note: Exact CI syntax and secret-handling conventions vary by provider (GitHub Actions, GitLab CI, and others); verify against your platform's current documentation, since these configuration formats evolve independently of the OpenAI SDK itself.
Where Human Evaluation Fits Into an Automated Pipeline
Not everything belongs inside the automatic gate. Following Lesson 9's distinction, the CI pipeline above enforces automated checks — unit tests and metric-based regression comparison — on every relevant change, while human evaluation is scheduled separately (for example, a periodic job that samples recent production outputs and assigns them to reviewers) and feeds its findings back into the dataset itself: a case a human reviewer flags as poorly handled becomes a new dataset example with a documented expected answer, permanently strengthening the automated suite for every future run. This is how a repeatable pipeline improves over time rather than staying frozen at the quality of the day it was first written.
Common Mistakes
- Running the evaluation stage even when unit tests fail. This wastes API budget evaluating a system with a known code bug, and can produce a confusing regression report caused by the bug rather than by any real change in prompt or model quality.
- Letting a single failed prediction crash the entire evaluation run. Without the
try/exceptaround each prediction, one timeout or malformed response aborts the whole batch, losing every other result and making the failure rate itself impossible to measure. - Building a pipeline that only prints results instead of returning a real pass/fail signal. A pipeline that a human must read and interpret manually will eventually be skipped or ignored; a pipeline that returns a proper exit code can be trusted to gate a merge automatically.
Best Practices
- Always run fast, free unit tests before the slower, costlier evaluation stage, and stop immediately on unit test failure to avoid wasted spend and confusing results.
- Make every stage return a structured, actionable result (a boolean, an exit code, a comparison dictionary) rather than only human-readable text, so the pipeline can be automated end to end.
- Feed human evaluation findings back into the automated dataset over time, so the repeatable pipeline's coverage grows from real production experience rather than remaining fixed at its initial scope.