Creating a Repeatable Evaluation Pipeline

Ma Mahalakshmi V Updated 19 Sep 2026
7 min read ·Lesson 173 of 224

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.output_text 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 Memorizing Names 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 Reducing Unsupported Claims with Grounded Generation Testing Freshness-Sensitive AI Answers Production Considerations for Web-Grounded Applications Understanding File Search and Retrieval-Augmented Generation Creating and Organizing Vector Stores Uploading Documents for Retrieval Connecting Vector Stores to Responses API Requests Designing Document Metadata and Filtering Strategies Building a PDF Question-Answering Application Improving Retrieval Quality With Better Document Preparation Handling Missing Evidence and Retrieval Failures Combining File Search With Web Search Building a Production Knowledge-Base Assistant What the Code Interpreter Tool Is Designed For Running Python-Based Analysis Through the OpenAI SDK Uploading Datasets for Analysis Analyzing CSV and Spreadsheet Data Generating Charts and Data Summaries Handling Generated Files and Downloadable Artifacts Building a Data-Analysis Assistant Combining Code Execution with Structured Outputs Validating Generated Calculations and Results Security and Sandbox Considerations for Code Execution Understanding Multimodal Input with the OpenAI SDK Sending Images to a Model Image Analysis from URLs and Uploaded Files Extracting Text and Information from Screenshots Building an Image-Question-Answering Application Combining Image Input with Structured Output Analyzing Multiple Images in One Request Handling Image Quality and Input Limitations Designing Multimodal Prompts for Reliable Results Building a Practical Vision-Powered Python Application Understanding Speech-to-Text and Text-to-Speech Workflows Transcribing Audio with the OpenAI SDK Working with Uploaded Audio Files Handling Timestamps and Transcription Metadata Building a Meeting Transcription Workflow Generating Spoken Responses from Text Handling Long Audio and Processing Failures Combining Audio with Text and Tool Calling Building an End-to-End Python Voice Application What Embeddings Are and When to Use Them Generating Embeddings With the OpenAI API Preparing Text for Embedding Comparing Vectors With Cosine Similarity Building a Simple Semantic Search Engine in Python Storing Embeddings in a Database Metadata Filtering for Semantic Search Chunking Strategies for Better Retrieval Evaluating Semantic Search Quality Building a Document Similarity Application When Batch Processing Makes Sense Designing Large-Volume AI Processing Pipelines Using Asynchronous Python with the OpenAI SDK Running Concurrent Requests Safely Controlling Concurrency and Avoiding Rate Limits Tracking Batch Job Progress Handling Partial Failures in Bulk Workloads Retrying Failed Items Without Duplicating Successful Work Designing Resumable AI Processing Jobs Building a Production Batch-Processing Pipeline Batch Processing Makes Sense Large-Scale AI Processing Pipelines Async Python with OpenAI SDK Safe Concurrent Requests Concurrency & Rate Limits Batch Progress Tracking Partial Failure Handling Safe Retry Handling Resumable AI Jobs Production Batch Pipeline System–User Data Separation Reusable App Instructions Prompt Templates & Variables Extraction & Classification Prompts Summarization & Transformation Prompts Explicit Output Requirements Prompt Version Management Prompt Testing & Evaluation Reusable Python Prompt Library API Key Security Secure API Key Storage Secure Secret Management Prompt Injection Prevention Trusted vs. Untrusted Content Tool Argument Validation Sensitive Data Handling Secure Logging AI Action Authorization Production AI Security Checklist Why AI Applications Need Evaluation Beyond Unit Tests Unit Testing OpenAI SDK Integration Code Mocking API Responses in Python Tests Testing Structured Outputs Against Schemas Testing Tool-Calling Workflows Building a Small Evaluation Dataset Measuring Accuracy, Consistency, and Failure Rates Regression Testing Prompts and Model Changes Human Evaluation Versus Automated Evaluation Creating a Repeatable Evaluation Pipeline AI Request Monitoring Token Cost Management Usage Metrics Design Reducing Model Calls Prompt & Context Optimization Model Selection & Optimization AI Caching Strategies Interactive Latency Optimization Usage Dashboards & Budget Alerts Performance & Cost Checklist Every API Call Starts Fresh Fixing API Statelessness Server-Side Conversation Memory Limits of Response Chaining What We're Building Conversation Memory Challenges Preparing an OpenAI SDK Application for Deployment Environment-Specific Configuration for Development and Production Deploying a Python AI Service with Docker Container Health Checks and Startup Configuration Managing Secrets in Cloud Deployments Background Workers for Long-Running AI Tasks Queues and Asynchronous Job Architectures Scaling AI Workloads Horizontally Monitoring Production Incidents and Failures Production Deployment Checklist for OpenAI SDK Applications Reusable OpenAI Service Classes AI Client Dependency Injection Typed AI Responses Python Configuration Management AI Request Decorators Centralized AI Error Handling Clean SDK Abstractions Reusable OpenAI Utilities Internal AI Python Libraries SDK Integration Maintenance Production AI Chatbot Document Q&A System Web Research Assistant Customer Support Agent AI Data Analysis Assistant Image Analysis App Meeting Transcription & Summary Semantic Document Search Multi-Tool AI Agent Production OpenAI SDK App Why "It Looked Fine When I Tested It" Isn't Enough Timing Note Status Note Pre-Decision Status Note Current Availability Note
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.