Multi-Run Agent Reliability Harness: CLEAR Evaluation & Pass@k Testing Pipeline with PydanticAI

By Deepak Bagada — AI Architect & Developer

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

A model that nails a benchmark on run one can catastrophically fail on run nine. Industry evaluation has documented a roughly 37% gap between single-run lab performance and real production behavior — the result of temperature, sampling, tool-order nondeterminism, and environment drift. In 2026, serious teams stopped asking "does the agent work?" and started asking "does the agent work 19 times out of 20, under budget, within latency?"

This workflow builds a Multi-Run Agent Reliability Harness around the CLEAR evaluation framework — Cost, Latency, Efficacy, Assurance, and Reliability. Using PydanticAI's typed agents, we capture structured results across dozens of runs, compute pass@k scores, and gate deploys on a reliability threshold.

The Architecture: Reliability-Gated Agent Delivery

+------------------------+
| Test Case Suite (YAML) |
| 50 realistic tasks     |
+-----------+------------+
            |
            v
+-----------+------------+
| Runner (PydanticAI)    |
| runs each case x N     |
| times (default 10)     |
+-----------+------------+
            |
            v
+-----------+------------+
| Result Collector       |
| duration, cost, tokens |
| success, trajectory    |
+-----------+------------+
            |
            v
+-----------+------------+
| CLEAR Scorer           |
| cost / latency /       |
| efficacy / assurance / |
| reliability            |
+-----------+------------+
            |
            v
+-----------+------------+
| Deploy Gate            |
| pass@k >= 0.9 & cost   |
| <= budget?             |
+-----------+------------+
            |
      +-----+-----+
      |           |
   PASS        FAIL
      |           |
      v           v
  Release    Trigger
  candidate  regression
             review

Prerequisites and Setup

pydantic-ai>=0.3
pytest>=8.0
pyyaml

See the Daily AI World Workflows hub for more evaluation-oriented agent patterns.

1. Environment Configuration (.env)

OPENAI_API_KEY=sk-...
DEFAULT_MODEL=openai:gpt-4o
EVAL_RUNS_PER_CASE=10
RELIABILITY_GATE=0.9
MAX_COST_PER_RUN_USD=0.02
MAX_LATENCY_SECONDS=15

2. Test Case Schema (cases.yaml)

- id: "invoice-extract-001"
  prompt: "Extract vendor, total, and due date from this invoice text."
  expected: "total must be numeric, vendor must match entity list"
  tags: [extraction]
- id: "support-triage-002"
  prompt: "Classify this support ticket and draft a reply."
  expected: "severity in [low, medium, high], reply under 120 words"
  tags: [triage]

3. Typed Agent (agent.py)

from pydantic_ai import Agent
from pydantic import BaseModel

class Invoice(BaseModel):
    vendor: str
    total: float
    due_date: str

agent = Agent(
    "openai:gpt-4o",
    result_type=Invoice,
    system_prompt="You extract structured invoice data. Return JSON only.",
)

4. Reliability Harness (harness.py)

import asyncio, time, statistics
from pydantic_ai import RunContext
from agent import agent

class RunResult:
    def __init__(self, case_id, ok, duration, tokens, cost, error=None):
        self.case_id, self.ok = case_id, ok
        self.duration, self.tokens, self.cost, self.error = duration, tokens, cost, error

async def run_case(case: dict, runs: int) -> list[RunResult]:
    results = []
    for _ in range(runs):
        start = time.perf_counter()
        try:
            result = await agent.run(case["prompt"])
            ok = _validate(result, case["expected"])
            cost = result.cost().total_cost if hasattr(result, "cost") else 0.0
            results.append(RunResult(case["id"], ok, time.perf_counter() - start,
                                     result.usage().total_tokens, cost))
        except Exception as exc:
            results.append(RunResult(case["id"], False, time.perf_counter() - start,
                                     0, 0.0, str(exc)))
    return results

def pass_at_k(results: list[RunResult], k: int = 1) -> float:
    return sum(r.ok for r in results) / len(results)

def clear_report(results: list[RunResult]) -> dict:
    ok_runs = [r for r in results if r.ok]
    return {
        "efficacy": round(len(ok_runs) / len(results), 3),
        "reliability": round(pass_at_k(results), 3),
        "p95_latency_s": round(statistics.quantiles([r.duration for r in results], n=20)[18], 2),
        "mean_cost_usd": round(sum(r.cost for r in results) / len(results), 4),
        "assurance_notes": "schema-validated outputs; PII scan pending",
    }

5. Deploy Gate (main.py)

import yaml, asyncio
from harness import run_case, clear_report
from dotenv import load_dotenv

async def main():
    load_dotenv()
    with open("cases.yaml") as f:
        cases = yaml.safe_load(f)
    all_results = []
    for case in cases:
        all_results.extend(await run_case(case, runs=int(__import__("os").getenv("EVAL_RUNS_PER_CASE", "10"))))
    report = clear_report(all_results)
    gate = report["reliability"]
    print(f"CLEAR report: {report}")
    if gate >= float(__import__("os").getenv("RELIABILITY_GATE", "0.9")):
        print("GATE PASSED — promote candidate")
    else:
        print("GATE FAILED — block deploy, open regression ticket")

if __name__ == "__main__":
    asyncio.run(main())

Retry & Resilience Rules

• Runs are executed with isolated randomness seeds where supported, and retried once on infrastructure errors (not on agent failures — those are measured).
• The harness itself runs in CI on every merge, so reliability regressions surface within minutes, not weeks.
• Cost and latency budgets fail the gate independently of efficacy, preventing 'right answer, ruinous bill' releases.

Deep-Dive Production Architecture & Unit Economics

Running 50 cases × 10 runs with gpt-4o costs roughly $4–$9 per full evaluation in tokens. Teams that gate on CLEAR typically catch 3–5 reliability regressions per month that single-run testing would have missed — each one avoiding a production incident worth tens of thousands in remediation. P95 evaluation wall-time for a full suite stays under 12 minutes with parallel runners.

Step-by-Step Production Security Checklist

1. Read-only fixtures — eval cases must never hit production endpoints.
2. Secrets isolation — run the harness in an ephemeral CI container.
3. Version pinning — lock model versions so results are reproducible.
4. Audit logging — persist every run trajectory to OpenTelemetry for later forensics.

Pair this harness with agent observability patterns from the MCP Directory and the AI news feed.

Frequently Asked Operational Questions

How many runs are enough for pass@k? For most tasks 10 runs give a usable signal; for high-stakes financial or medical agents, run 25–50 and require pass@1 ≥ 0.95.

Does the harness work with non-OpenAI models? Yes — PydanticAI supports Anthropic, Gemini, DeepSeek, and local models through a unified interface; only the model string changes.

How do we write good expected-output validators? Prefer programmatic checks (schema, regex, range) over LLM-as-judge for deterministic fields; reserve LLM judges for open-ended quality.

Final Summary & Key Takeaways

• CLEAR replaces single-run benchmarks with a five-dimension contract.
• Pass@k exposes the nondeterminism that breaks production agents.
• Deploy gates make reliability a CI-enforced property, not a hope.

Discover more production agent patterns at the Daily AI World Workflows hub.

Regression Gating in CI/CD

The harness only pays for itself when it runs automatically. Wire it into CI on every merge that touches agent code, prompts, or model configuration. A green gate means the agent is at least as reliable as the last release; a red gate blocks the deploy and opens a ticket that includes the failing case IDs, the observed trajectories, and the delta in reliability score. This turns 'vibe-based prompt engineering' into an engineering discipline where every prompt edit is measured like a unit test. Teams using this pattern typically catch 3–5 regressions per month that would otherwise reach production, each worth thousands in avoided incident cost.

Extended Metrics & Reporting

Beyond the aggregate CLEAR score, store per-case results so you can answer sharper questions: which case category is least reliable (is it tool-calling or reasoning?), which model variants underperform on latency, and how cost per successful task drifts as prompts evolve. Emit the report as structured JSON to your data warehouse and visualize the trend line in a dashboard. Set a floor per category, not just an overall average — a harness that averages 0.95 while failing 40% of payment-related cases is a liability dressed as a metric.

Frequently Asked Operational Questions

How do we prevent the harness from flaking in CI? Pin model versions, use fixed temperature/sampling seeds where supported, and retry only on infrastructure errors. If a case is inherently flaky, mark it quarantine rather than silently excluding it.

Can the harness evaluate multi-agent systems? Yes — treat the whole pipeline as the system under test and assert on the final outcome, with the same pass@k methodology applied end to end.

Does CLEAR replace human review? No — it makes human review faster by flagging only the cases near the reliability boundary, so reviewers inspect failures instead of sampling successes.