Creating a Repeatable Evaluation Pipeline

Ma Mahalakshmi V Updated 19 Sep 2026
7 min read

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.

  1. 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.
  2. 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/except around 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.

0 Comments

Reviewed before they appear

No comments yet.

OpenAI SDK
Introduction to the OpenAI SDK Setting Up Python Creating an API Key Your First Call — client.responses.create() and response.out Understanding Billing, Credits, and What a Request Costs Why Responses Replaced Chat Completions Anatomy of a Request: model, input, and instructions Anatomy of a Response: The Typed output Array, Not Just Text Roles: User, Assistant, and Developer/System Choosing a Model, and Reading the Models Page Instead of Mem Instructions vs. Input Writing Prompts That Get Consistent Results Few-Shot Examples Reasoning Models and the reasoning Parameter Debugging a Prompt That Misbehaves Why Streaming Matters for User Experience stream=True and Iterating Over Events Handling the Event Types You Actually Care About Background Mode for Long-Running Jobs Project — Add Live Streaming to Your Chatbot The Problem With Parsing Free Text JSON Schema and Strict Mode Pydantic Models With the SDK's Parse Helpers Handling Refusals and Validation Failures Project — A Resume-to-JSON Extractor Working With input_image input_file, PDFs, and the Files API Image Generation Speech-to-Text and Text-to-Speech Project: A PDF Question-Answering Script What Function Calling Is Defining a Tool Schema The Full Loop Multiple Tools Errors, Timeouts, and Untrusted Arguments Project: A Weather Assistant Web Search File Search and Vector Stores Code Interpreter Remote MCP Servers and Connectors Project: A Research Assistant What an Embedding Is, Without the Maths Generating and Storing Embeddings Similarity Search From Scratch Hosted Vector Stores vs. Rolling Your Own A Small RAG App Over a Folder of Notes Agents vs. a Single API Call — When You Need One pip install openai Giving Agents Tools Handoffs and Multi-Agent Triage Guardrails and Approvals Tracing and Observing What Your Agent Did A Multi-Agent Support Desk Error Codes and What Each One Means Retries, Timeouts, and Backoff Rate Limits and Spend Limits Prompt Caching and Cost Optimisation The Batch API for Bulk Work Async Clients and Concurrency Moderation and Safety Best Practices Designing the App Backend With FastAPI Streaming to a Simple Frontend Deploying and a Cost/Safety Checklist Why Web Search Is Useful for Current Information Using the Web Search Tool with the Responses API Configuring Search Behavior for Application Use Cases Understanding Citations and Source Attribution Where to Go Next Building a Research Assistant with Web Search Combining Web Search with Structured Outputs Handling Conflicting or Low-Quality Web Sources
Ask about this post
AI Ask about this post

Ask questions about Creating a Repeatable Evaluation Pipeline and get answers drawn from it.

Signed-in readers only.