Introduction to Agent Evaluation Harnesses
When you build AI agents that make decisions, call tools, and produce structured outputs, you need a way to measure whether they actually work. An agent evaluation harness is a systematic framework that runs your agent against a curated set of test cases, captures its outputs, and scores them against expected behaviors. Think of it as unit testing, but for non-deterministic systems that involve language models.
Pydantic AI is a Python framework that brings type safety and structured outputs to agent development. Because Pydantic AI agents already produce validated, typed outputs through Pydantic models, it is an excellent foundation for building a rigorous evaluation harness. You can leverage the same schemas you use in production to define what a "correct" answer looks like.
Why Evaluation Harnesses Matter
LLM-based agents are probabilistic. The same prompt can produce different outputs on different runs, and small changes to a system prompt or tool definition can silently degrade performance. Without an evaluation harness, you are flying blind.
- Regression detection: Catch performance drops when you change prompts, models, or tools.
- Model comparison: Compare GPT-4o against Claude or a local model on the same task set.
- Prompt iteration: Quantify whether your prompt "improvements" actually improve anything.
- CI/CD integration: Block deployments when evaluation scores fall below a threshold.
- Documentation: An eval suite documents what your agent is expected to handle.
Prerequisites and Setup
Install Pydantic AI and a few supporting libraries. We will use pytest as the test runner and rich for readable console output of results.
pip install pydantic-ai pytest rich pydantic
Create a project structure that separates your agent, your evaluation cases, and the harness itself:
agent_eval_project/
├── agent/
│ ├── __init__.py
│ └── support_agent.py
├── evals/
│ ├── __init__.py
│ ├── cases.py
│ ├── scorers.py
│ └── harness.py
└── run_evals.py
Building the Agent Under Test
Let us build a simple customer support agent that classifies a user message and decides on an action. The agent returns a structured response, which makes evaluation far easier than parsing free text.
# agent/support_agent.py
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from typing import Literal
class SupportResponse(BaseModel):
category: Literal["billing", "technical", "general", "complaint"]
urgency: Literal["low", "medium", "high"] = Field(
description="How quickly this needs attention"
)
suggested_action: str = Field(
description="A concrete next step for the support team"
)
confidence: float = Field(
ge=0.0, le=1.0,
description="Confidence score for this classification"
)
support_agent = Agent(
model="openai:gpt-4o-mini",
output_type=SupportResponse,
system_prompt=(
"You are a customer support triage agent. Analyze the incoming "
"message and classify it. Always suggest a concrete action. "
"Be conservative with urgency: only mark 'high' for account "
"lockouts, data loss, or angry customers threatening to leave."
),
)
async def run_support_agent(message: str) -> SupportResponse:
result = await support_agent.run(message)
return result.output
Because the output is a Pydantic model, invalid responses are rejected automatically. This means your evaluation harness can rely on the schema being correct and focus on scoring the content of the response.
Defining Evaluation Cases
An evaluation case is a single test scenario. It contains the input, the expected properties of the output, and optional metadata. Define a typed structure for your cases so the harness can validate them.
# evals/cases.py
from pydantic import BaseModel
from typing import Optional
class EvalCase(BaseModel):
id: str
input: str
expected_category: Optional[str] = None
expected_urgency: Optional[str] = None
min_confidence: float = 0.5
action_must_contain: Optional[list[str]] = None
tags: list[str] = []
eval_cases: list[EvalCase] = [
EvalCase(
id="billing_001",
input="I was charged twice for my subscription this month.",
expected_category="billing",
expected_urgency="medium",
min_confidence=0.6,
action_must_contain=["refund", "charge"],
tags=["billing", "duplicate-charge"],
),
EvalCase(
id="tech_001",
input="I can't log in, it says my account is locked.",
expected_category="technical",
expected_urgency="high",
min_confidence=0.7,
action_must_contain=["reset", "unlock", "password"],
tags=["technical", "login"],
),
EvalCase(
id="complaint_001",
input=(
"I have been a customer for 5 years and this is the worst "
"service I have ever experienced. I am cancelling."
),
expected_category="complaint",
expected_urgency="high",
min_confidence=0.6,
action_must_contain=["escalat", "retention", "loyalty"],
tags=["complaint", "churn-risk"],
),
EvalCase(
id="general_001",
input="What are your business hours?",
expected_category="general",
expected_urgency="low",
min_confidence=0.5,
action_must_contain=None,
tags=["general", "faq"],
),
]
Notice that not every field is required for every case. A flexible case definition lets you write partial expectations — you might only care about category for some cases, while others need precise urgency and action checks.
Writing Scorers
Scorers are functions that take an evaluation case and the agent's actual output, then return a score and a reason. Keeping scorers modular lets you mix and match them per case and reuse them across agents.
# evals/scorers.py
from agent.support_agent import SupportResponse
from evals.cases import EvalCase
class ScoreResult(BaseModel):
scorer_name: str
passed: bool
score: float # 0.0 to 1.0
reason: str
def score_category(case: EvalCase, output: SupportResponse) -> ScoreResult:
if case.expected_category is None:
return ScoreResult(
scorer_name="category",
passed=True,
score=1.0,
reason="No category expectation set",
)
passed = output.category == case.expected_category
return ScoreResult(
scorer_name="category",
passed=passed,
score=1.0 if passed else 0.0,
reason=f"Expected {case.expected_category}, got {output.category}",
)
def score_urgency(case: EvalCase, output: SupportResponse) -> ScoreResult:
if case.expected_urgency is None:
return ScoreResult(
scorer_name="urgency",
passed=True,
score=1.0,
reason="No urgency expectation set",
)
passed = output.urgency == case.expected_urgency
return ScoreResult(
scorer_name="urgency",
passed=passed,
score=1.0 if passed else 0.0,
reason=f"Expected {case.expected_urgency}, got {output.urgency}",
)
def score_confidence(case: EvalCase, output: SupportResponse) -> ScoreResult:
passed = output.confidence >= case.min_confidence
return ScoreResult(
scorer_name="confidence",
passed=passed,
score=output.confidence,
reason=(
f"Confidence {output.confidence:.2f} "
f"{'>=' if passed else '<'} threshold {case.min_confidence}"
),
)
def score_action_keywords(case: EvalCase, output: SupportResponse) -> ScoreResult:
if case.action_must_contain is None:
return ScoreResult(
scorer_name="action_keywords",
passed=True,
score=1.0,
reason="No keyword expectations set",
)
action_lower = output.suggested_action.lower()
matched = [kw for kw in case.action_must_contain if kw.lower() in action_lower]
passed = len(matched) > 0
return ScoreResult(
scorer_name="action_keywords",
passed=passed,
score=len(matched) / len(case.action_must_contain),
reason=(
f"Matched keywords: {matched} "
f"out of {case.action_must_contain}"
),
)
ALL_SCORERS = [
score_category,
score_urgency,
score_confidence,
score_action_keywords,
]
Each scorer returns a normalized score between 0 and 1, plus a human-readable reason. The reason is critical — when a case fails, you want to know exactly why without re-running the agent manually.
Building the Harness
The harness ties everything together. It runs each case through the agent, applies all scorers, aggregates results, and produces a report. Design it to be async-friendly since Pydantic AI agents are asynchronous.
# evals/harness.py
import asyncio
from dataclasses import dataclass, field
from typing import Callable
from rich.console import Console
from rich.table import Table
from agent.support_agent import run_support_agent, SupportResponse
from evals.cases import EvalCase
from evals.scorers import ScoreResult, ALL_SCORERS
console = Console()
@dataclass
class CaseResult:
case: EvalCase
output: SupportResponse | None
scores: list[ScoreResult] = field(default_factory=list)
error: str | None = None
@property
def passed(self) -> bool:
if self.error:
return False
return all(s.passed for s in self.scores)
@property
def average_score(self) -> float:
if not self.scores:
return 0.0
return sum(s.score for s in self.scores) / len(self.scores)
@dataclass
class HarnessReport:
results: list[CaseResult]
total: int
passed: int
failed: int
pass_rate: float
mean_score: float
async def run_single_case(
case: EvalCase,
scorers: list[Callable],
) -> CaseResult:
try:
output = await run_support_agent(case.input)
scores = [scorer(case, output) for scorer in scorers]
return CaseResult(case=case, output=output, scores=scores)
except Exception as e:
return CaseResult(case=case, output=None, error=str(e))
async def run_eval_suite(
cases: list[EvalCase],
scorers: list[Callable] | None = None,
concurrency: int = 5,
) -> HarnessReport:
if scorers is None:
scorers = ALL_SCORERS
semaphore = asyncio.Semaphore(concurrency)
async def bounded_run(case: EvalCase) -> CaseResult:
async with semaphore:
return await run_single_case(case, scorers)
results = await asyncio.gather(*[bounded_run(c) for c in cases])
total = len(results)
passed = sum(1 for r in results if r.passed)
failed = total - passed
pass_rate = passed / total if total else 0.0
mean_score = (
sum(r.average_score for r in results) / total if total else 0.0
)
return HarnessReport(
results=results,
total=total,
passed=passed,
failed=failed,
pass_rate=pass_rate,
mean_score=mean_score,
)
def print_report(report: HarnessReport) -> None:
console.print()
console.print(
f"[bold]Evaluation Results[/bold] — "
f"{report.passed}/{report.total} passed "
f"({report.pass_rate:.0%}), "
f"mean score: {report.mean_score:.2f}",
style="bold cyan",
)
table = Table(show_header=True, header_style="bold magenta")
table.add_column("Case ID", style="dim")
table.add_column("Status")
table.add_column("Avg Score")
table.add_column("Category")
table.add_column("Urgency")
table.add_column("Failures")
for r in report.results:
status = "[green]PASS[/green]" if r.passed else "[red]FAIL[/red]"
avg = f"{r.average_score:.2f}"
category = r.output.category if r.output else "ERROR"
urgency = r.output.urgency if r.output else "ERROR"
if r.error:
failures = r.error[:60]
else:
failed_scorers = [s.scorer_name for s in r.scores if not s.passed]
failures = ", ".join(failed_scorers) if failed_scorers else "—"
table.add_row(r.case.id, status, avg, category, urgency, failures)
console.print(table)
console.print()
for r in report.results:
if not r.passed:
console.print(f"[red]✗ {r.case.id}[/red]")
if r.error:
console.print(f" Error: {r.error}")
for s in r.scores:
if not s.passed:
console.print(f" [{s.scorer_name}] {s.reason}")
console.print()
The harness uses a semaphore to limit concurrency, preventing rate limit errors when you have many cases. It also catches exceptions per case so one failure does not crash the entire run.
Running the Evaluation
Create an entry point script that loads the cases, runs the harness, and prints the report.
# run_evals.py
import asyncio
from evals.cases import eval_cases
from evals.harness import run_eval_suite, print_report
async def main():
report = await run_eval_suite(eval_cases)
print_report(report)
# Exit with non-zero code if pass rate is below threshold
# Useful for CI/CD pipelines
threshold = 0.75
if report.pass_rate < threshold:
print(f"Pass rate {report.pass_rate:.0%} below threshold {threshold:.0%}")
exit(1)
if __name__ == "__main__":
asyncio.run(main())
Run it with:
python run_evals.py
You will see a rich-formatted table showing each case, its pass/fail status, average score, and which scorers failed. The script exits with code 1 if the pass rate falls below your threshold, making it suitable for CI integration.
Adding LLM-as-a-Judge Scoring
Keyword matching and exact field comparison are fast and deterministic, but they cannot judge whether a suggested action is genuinely helpful. An LLM-as-a-judge scorer uses another model to evaluate the quality of the output. This is where Pydantic AI shines again — you can define a structured rubric and get a typed score back.
# evals/llm_judge.py
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from agent.support_agent import SupportResponse
from evals.cases import EvalCase
from evals.scorers import ScoreResult
class JudgeVerdict(BaseModel):
score: float = Field(
ge=0.0, le=1.0,
description="Quality score for the suggested action"
)
reasoning: str
is_safe: bool = Field(
description="Whether the action is safe to execute without human review"
)
judge_agent = Agent(
model="openai:gpt-4o",
output_type=JudgeVerdict,
system_prompt=(
"You are an expert evaluator for customer support responses. "
"Given a customer message and the agent's suggested action, "
"score the action on helpfulness, safety, and appropriateness. "
"Return a score from 0.0 (terrible) to 1.0 (excellent)."
),
)
async def score_with_llm_judge(
case: EvalCase,
output: SupportResponse,
) -> ScoreResult:
prompt = (
f"Customer message: {case.input}\n\n"
f"Agent category: {output.category}\n"
f"Agent urgency: {output.urgency}\n"
f"Agent suggested action: {output.suggested_action}\n"
f"Agent confidence: {output.confidence}\n\n"
f"Evaluate the suggested action."
)
result = await judge_agent.run(prompt)
verdict = result.output
passed = verdict.score >= 0.7 and verdict.is_safe
return ScoreResult(
scorer_name="llm_judge",
passed=passed,
score=verdict.score,
reason=f"{verdict.reasoning} (safe={verdict.is_safe})",
)
Integrate the LLM judge into the harness by adding it to the scorer list. Because it is async, you will need a small modification to support async scorers. Update run_single_case to handle both sync and async scorers:
# Updated run_single_case in evals/harness.py
import inspect
async def run_single_case(case, scorers):
try:
output = await run_support_agent(case.input)
scores = []
for scorer in scorers:
if inspect.iscoroutinefunction(scorer):
result = await scorer(case, output)
else:
result = scorer(case, output)
scores.append(result)
return CaseResult(case=case, output=output, scores=scores)
except Exception as e:
return CaseResult(case=case, output=None, error=str(e))
Then pass the judge scorer alongside the deterministic ones:
from evals.llm_judge import score_with_llm_judge
from evals.scorers import ALL_SCORERS
scorers = ALL_SCORERS + [score_with_llm_judge]
report = await run_eval_suite(eval_cases, scorers=scorers)
Best Practices
- Start with deterministic scorers. Exact match and keyword checks are cheap, fast, and reproducible. Add LLM judges only for dimensions that truly require semantic understanding.
- Version your eval cases. Store cases in version control and tag them. When you change the case set, you change what you are measuring, so track those changes deliberately.
- Run evaluations multiple times. Because agents are non-deterministic, a single run is noisy. Run each case 3–5 times and report the average or the pass rate across runs.
- Separate dev and regression sets. Use a small, fast set during development and a larger, comprehensive set for regression checks before deployment.
- Log everything. Save the full agent output, token usage, latency, and scorer reasons for every run. This creates a historical record you can mine when performance drifts.
- Use a cheaper model for the judge than for the agent. Or better, use a stronger model for judging than for the agent under test, so the judge can catch errors the weaker agent makes.
- Watch for judge bias. LLM judges can be biased toward verbose or polite responses. Calibrate your judge against human-labeled examples periodically.
- Set realistic thresholds. A 100% pass rate is often unrealistic for LLM agents. Choose thresholds based on your current baseline and ratchet them up over time.
Extending the Harness
The harness described here is a starting point. Real-world evaluation systems often add additional capabilities:
- Golden dataset management: Store cases in JSON or YAML files, or in a database, so non-developers can contribute test cases.
- Tool call evaluation: If your agent calls tools, score whether it called the right tool with the right arguments, not just the final output.
- Trajectory scoring: Evaluate the full reasoning chain, not just the final answer, to catch agents that arrive at correct answers through flawed logic.
- A/B comparison: Run two agent configurations side by side and compare scores case by case to decide which version ships.
- Cost and latency tracking: Capture token counts and response times per case so you can optimize for cost, not just accuracy.
Here is a quick example of capturing token usage from Pydantic AI's result object:
result = await support_agent.run(message)
usage = result.usage()
print(f"Input tokens: {usage.input_tokens}")
print(f"Output tokens: {usage.output_tokens}")
print(f"Total tokens: {usage.total_tokens}")
You can store these values in CaseResult and include them in the report to track cost trends over time.
Conclusion
Building an agent evaluation harness with Pydantic AI gives you a structured, type-safe way to measure agent quality over time. By combining deterministic scorers for fast feedback with LLM-as-a-judge scorers for semantic quality, you get a comprehensive picture of how your agent performs. The key is to treat your evaluation suite as a first-class artifact: version it, maintain it, and run it regularly. As your agent evolves, your eval suite becomes the safety net that catches regressions before your users do, and the scoreboard that tells you whether each change actually moved the needle.