Introduction to Agent Evaluation Harnesses
Building AI agents is easy; proving they work reliably is hard. As agent architectures grow more complex—incorporating tool calls, multi-step reasoning, and dynamic routing—traditional evaluation methods fall short. An agent evaluation harness is a structured framework that systematically tests your agents across diverse scenarios, measuring correctness, efficiency, and robustness.
LangGraph, built on top of LangChain, provides a powerful graph-based abstraction for building stateful, multi-actor agents. By combining LangGraph's flexible agent runtime with a purpose-built evaluation harness, you can continuously validate agent behavior, catch regressions, and ship with confidence.
What Is an Agent Evaluation Harness?
An agent evaluation harness is a testing infrastructure specifically designed for agentic systems. Unlike unit tests that check isolated functions, an evaluation harness evaluates the end-to-end behavior of an agent across a curated set of test cases. It typically includes:
- Test cases — scenarios with inputs, expected outcomes, and evaluation criteria
- Agent runner — executes the agent against each test case in a controlled environment
- Evaluators — functions or LLM-based judges that score agent outputs and trajectories
- Reporting — aggregates results into metrics, traces, and visualizations
- Mocked tools — deterministic stand-ins for external services to ensure reproducibility
The key distinction from traditional testing is that agents produce trajectories—sequences of thoughts, tool calls, and observations—not just final answers. A good harness evaluates both the destination and the journey.
Why Evaluation Harnesses Matter
Without systematic evaluation, agent development becomes a guessing game. Here's why a dedicated harness is essential:
Catching Regressions
Every prompt tweak, tool modification, or model upgrade can silently break agent behavior. A harness lets you run the same suite of tests before and after changes, surfacing regressions before they reach production.
Measuring Trajectory Quality
Two agents might arrive at the same correct answer, but one took three efficient steps while the other took fifteen redundant ones. Evaluating trajectories helps you optimize for cost, latency, and reliability—not just correctness.
Enabling Safe Iteration
When you have a comprehensive test suite, you can refactor aggressively. The harness becomes a safety net that catches behavioral drift, letting you experiment with new architectures without fear.
Building Stakeholder Trust
Quantitative evaluation results—success rates, average steps, tool usage patterns—give stakeholders concrete evidence that your agent performs as advertised across real-world scenarios.
Setting Up Your Environment
Let's start by installing the necessary dependencies and setting up the project structure.
# Install required packages
pip install langgraph langchain langchain-openai langsmith
pip install pydantic pytest rich
# Project structure
# agent_eval/
# ├── __init__.py
# ├── agent.py # The LangGraph agent definition
# ├── tools.py # Tool definitions
# ├── harness/
# │ ├── __init__.py
# │ ├── runner.py # Test case runner
# │ ├── evaluators.py # Evaluation functions
# │ ├── cases.py # Test case definitions
# │ └── mocks.py # Mocked tools and environments
# └── run_evals.py # CLI entry pointSet your API keys as environment variables:
import os
os.environ["OPENAI_API_KEY"] = "sk-your-key-here"
os.environ["LANGSMITH_API_KEY"] = "ls-your-key-here"
os.environ["LANGSMITH_TRACING"] = "true"Building the Agent Under Test
First, let's create a simple but representative LangGraph agent that can search the web, perform calculations, and look up information in a knowledge base. This will be the system we evaluate.
# agent_eval/agent.py
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from typing import TypedDict, Annotated, List, Union
import operator
class AgentState(TypedDict):
messages: Annotated[List, operator.add]
tool_calls: List[dict]
final_answer: str
steps_taken: int
def create_agent(tools: list, model_name: str = "gpt-4o"):
"""Create a LangGraph agent with the given tools."""
llm = ChatOpenAI(model=model_name, temperature=0)
llm_with_tools = llm.bind_tools(tools)
tool_map = {tool.name: tool for tool in tools}
def agent_node(state: AgentState):
system_prompt = SystemMessage(content=(
"You are a helpful assistant. Use tools when needed "
"to answer questions accurately. Be concise."
))
response = llm_with_tools.invoke([system_prompt] + state["messages"])
return {
"messages": [response],
"steps_taken": state.get("steps_taken", 0) + 1
}
def tool_node(state: AgentState):
last_message = state["messages"][-1]
results = []
tool_calls_log = []
for tool_call in last_message.tool_calls:
tool = tool_map[tool_call["name"]]
result = tool.invoke(tool_call["args"])
results.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": str(result)
})
tool_calls_log.append({
"name": tool_call["name"],
"args": tool_call["args"],
"result": str(result)[:200]
})
return {
"messages": results,
"tool_calls": tool_calls_log,
"steps_taken": state.get("steps_taken", 0) + 1
}
def should_continue(state: AgentState):
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
if state.get("steps_taken", 0) >= 10:
return END
return "tools"
return END
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")
return workflow.compile()Now let's define the tools our agent can use:
# agent_eval/tools.py
from langchain_core.tools import tool
@tool
def search_web(query: str) -> str:
"""Search the web for information."""
# In production, this would call a real search API
return f"Search results for: {query}"
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression."""
try:
allowed = set("0123456789+-*/.() ")
if not all(c in allowed for c in expression):
return "Error: Invalid characters in expression"
result = eval(expression) # Safe due to character filtering
return str(result)
except Exception as e:
return f"Error: {str(e)}"
@tool
def lookup_knowledge_base(topic: str) -> str:
"""Look up information in the company knowledge base."""
kb = {
"return_policy": "Items can be returned within 30 days with receipt.",
"shipping": "Free shipping on orders over $50. Standard delivery 3-5 days.",
"warranty": "All products come with a 1-year manufacturer warranty.",
}
return kb.get(topic.lower().replace(" ", "_"), "No information found.")
ALL_TOOLS = [search_web, calculate, lookup_knowledge_base]Designing Test Cases
Test cases are the foundation of your evaluation harness. Each case should specify the input, expected behavior, and metadata for evaluation. Let's create a structured test case format:
# agent_eval/harness/cases.py
from pydantic import BaseModel, Field
from typing import Optional, List, Dict, Any
class TestCase(BaseModel):
id: str
description: str
input: str
expected_answer: Optional[str] = None
expected_tools: Optional[List[str]] = None
forbidden_tools: Optional[List[str]] = None
max_steps: int = 10
metadata: Dict[str, Any] = Field(default_factory=dict)
# Define your test suite
TEST_CASES = [
TestCase(
id="math_basic_001",
description="Simple arithmetic calculation",
input="What is 15 * 23 + 47?",
expected_answer="392",
expected_tools=["calculate"],
forbidden_tools=["search_web"],
max_steps=4,
),
TestCase(
id="kb_lookup_001",
description="Knowledge base retrieval",
input="What is the return policy?",
expected_answer="30 days",
expected_tools=["lookup_knowledge_base"],
forbidden_tools=["search_web"],
max_steps=4,
),
TestCase(
id="multi_step_001",
description="Multi-step reasoning requiring calculation and lookup",
input="If I order 3 items at $20 each, do I get free shipping? Also check the warranty.",
expected_tools=["calculate", "lookup_knowledge_base"],
max_steps=8,
),
TestCase(
id="no_tools_001",
description="Question that doesn't require tools",
input="What is the capital of France?",
expected_tools=[],
max_steps=2,
),
TestCase(
id="edge_case_001",
description="Handling invalid calculation input",
input="Calculate 5 / 0",
expected_tools=["calculate"],
max_steps=4,
metadata={"expect_error_handling": True},
),
]Creating Mocked Tools for Reproducibility
For evaluations to be reproducible, external dependencies must be deterministic. Mocked tools return controlled responses, ensuring the same test always produces the same trajectory:
# agent_eval/harness/mocks.py
from langchain_core.tools import tool
from typing import Dict
# Predefined mock responses for web search
MOCK_SEARCH_RESULTS: Dict[str, str] = {
"weather today": "Today: Sunny, 72°F, humidity 45%",
"latest news": "Top story: Tech company releases new AI framework",
"python tutorial": "Python is a high-level programming language...",
}
def create_mock_search_web():
@tool
def search_web(query: str) -> str:
"""Search the web for information."""
query_lower = query.lower()
for key, value in MOCK_SEARCH_RESULTS.items():
if key in query_lower:
return value
return f"No results found for: {query}"
return search_web
def create_mock_tools():
"""Create a full set of mocked tools for deterministic testing."""
from agent_eval.tools import calculate, lookup_knowledge_base
return [create_mock_search_web(), calculate, lookup_knowledge_base]Building the Evaluation Runner
The runner orchestrates test execution. It invokes the agent for each test case, captures the full trajectory, and passes results to evaluators:
# agent_eval/harness/runner.py
from agent_eval.agent import create_agent
from agent_eval.harness.cases import TestCase
from agent_eval.harness.mocks import create_mock_tools
from typing import List, Dict, Any
import time
import traceback
class TestResult(BaseModel):
test_id: str
description: str
passed: bool
input: str
output: str
trajectory: List[Dict[str, Any]]
tool_calls: List[Dict[str, Any]]
steps_taken: int
execution_time: float
errors: List[str]
scores: Dict[str, float]
class EvalRunner:
def __init__(self, use_mocks: bool = True, model_name: str = "gpt-4o"):
self.use_mocks = use_mocks
self.model_name = model_name
self.results: List[TestResult] = []
def run_single(self, test_case: TestCase) -> TestResult:
"""Run a single test case against the agent."""
tools = create_mock_tools() if self.use_mocks else get_real_tools()
agent = create_agent(tools, model_name=self.model_name)
errors = []
trajectory = []
tool_calls = []
output = ""
steps_taken = 0
start_time = time.time()
try:
initial_state = {
"messages": [{"role": "user", "content": test_case.input}],
"tool_calls": [],
"final_answer": "",
"steps_taken": 0,
}
result = agent.invoke(initial_state)
# Extract final answer
last_message = result["messages"][-1]
output = last_message.content if hasattr(last_message, "content") else str(last_message)
steps_taken = result.get("steps_taken", 0)
tool_calls = result.get("tool_calls", [])
# Build trajectory for analysis
for msg in result["messages"]:
trajectory.append({
"type": type(msg).__name__,
"content": str(msg.content)[:500] if hasattr(msg, "content") else str(msg)[:500],
"tool_calls": getattr(msg, "tool_calls", None),
})
except Exception as e:
errors.append(f"{type(e).__name__}: {str(e)}")
errors.append(traceback.format_exc())
execution_time = time.time() - start_time
return TestResult(
test_id=test_case.id,
description=test_case.description,
passed=len(errors) == 0,
input=test_case.input,
output=output,
trajectory=trajectory,
tool_calls=tool_calls,
steps_taken=steps_taken,
execution_time=execution_time,
errors=errors,
scores={},
)
def run_suite(self, test_cases: List[TestCase]) -> List[TestResult]:
"""Run all test cases and return results."""
self.results = []
for i, tc in enumerate(test_cases, 1):
print(f"[{i}/{len(test_cases)}] Running: {tc.id} - {tc.description}")
result = self.run_single(tc)
self.results.append(result)
status = "PASS" if result.passed else "FAIL"
print(f" -> {status} ({result.execution_time:.2f}s, {result.steps_taken} steps)")
return self.resultsImplementing Evaluators
Evaluators are the heart of the harness. They score different dimensions of agent performance. Let's implement several types: rule-based, trajectory-based, and LLM-as-judge evaluators.
# agent_eval/harness/evaluators.py
from agent_eval.harness.cases import TestCase
from agent_eval.harness.runner import TestResult
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from typing import Dict
import re
class Evaluator:
"""Base class for evaluators."""
name: str = "base"
def evaluate(self, test_case: TestCase, result: TestResult) -> float:
"""Return a score between 0.0 and 1.0."""
raise NotImplementedError
class ExactMatchEvaluator(Evaluator):
"""Check if the expected answer appears in the output."""
name = "exact_match"
def evaluate(self, test_case: TestCase, result: TestResult) -> float:
if not test_case.expected_answer:
return 1.0 # No expected answer, skip
if test_case.expected_answer.lower() in result.output.lower():
return 1.0
return 0.0
class ToolUsageEvaluator(Evaluator):
"""Verify the agent used the expected tools and avoided forbidden ones."""
name = "tool_usage"
def evaluate(self, test_case: TestCase, result: TestResult) -> float:
used_tools = {tc["name"] for tc in result.tool_calls}
score = 1.0
# Check expected tools were used
if test_case.expected_tools is not None:
expected = set(test_case.expected_tools)
if expected and not expected.issubset(used_tools):
missing = expected - used_tools
score -= 0.5 * (len(missing) / len(expected)) if expected else 0
# Check forbidden tools were not used
if test_case.forbidden_tools:
forbidden = set(test_case.forbidden_tools)
used_forbidden = forbidden & used_tools
if used_forbidden:
score -= 0.5 * (len(used_forbidden) / len(forbidden))
return max(0.0, min(1.0, score))
class StepEfficiencyEvaluator(Evaluator):
"""Penalize agents that take too many steps."""
name = "step_efficiency"
def evaluate(self, test_case: TestCase, result: TestResult) -> float:
if result.steps_taken <= 0:
return 0.0
if result.steps_taken <= test_case.max_steps:
return 1.0
# Linear penalty for exceeding max steps
excess = result.steps_taken - test_case.max_steps
return max(0.0, 1.0 - (excess / test_case.max_steps))
class ErrorHandlingEvaluator(Evaluator):
"""Check that the agent handles errors gracefully."""
name = "error_handling"
def evaluate(self, test_case: TestCase, result: TestResult) -> float:
if not test_case.metadata.get("expect_error_handling"):
return 1.0
# Check if output acknowledges the error without crashing
error_indicators = ["error", "cannot", "unable", "undefined", "invalid"]
has_error_ack = any(ind in result.output.lower() for ind in error_indicators)
no_crash = len(result.errors) == 0
return 1.0 if (has_error_ack and no_crash) else 0.0
class LLMJudgeEvaluator(Evaluator):
"""Use an LLM to judge the quality of the agent's response."""
name = "llm_judge"
def __init__(self, model_name: str = "gpt-4o"):
self.llm = ChatOpenAI(model=model_name, temperature=0)
def evaluate(self, test_case: TestCase, result: TestResult) -> float:
if not test_case.expected_answer:
return 1.0
judge_prompt = f"""You are evaluating an AI agent's response. Score it from 0.0 to 1.0.
Question: {test_case.input}
Expected answer should contain: {test_case.expected_answer}
Agent's answer: {result.output}
Scoring criteria:
- 1.0: Correct and complete answer
- 0.5: Partially correct or missing key information
- 0.0: Completely wrong or unhelpful
Respond with ONLY a number between 0.0 and 1.0."""
try:
response = self.llm.invoke([
SystemMessage(content="You are a strict but fair evaluator."),
HumanMessage(content=judge_prompt),
])
score = float(response.content.strip())
return max(0.0, min(1.0, score))
except (ValueError, Exception):
return 0.5 # Default to neutral if judge fails
class TrajectoryCoherenceEvaluator(Evaluator):
"""Use an LLM to evaluate whether the agent's trajectory makes sense."""
name = "trajectory_coherence"
def __init__(self, model_name: str = "gpt-4o"):
self.llm = ChatOpenAI(model=model_name, temperature=0)
def evaluate(self, test_case: TestCase, result: TestResult) -> float:
if not result.tool_calls:
return 1.0 # No tools used, trajectory is trivially coherent
trajectory_str = "\n".join([
f" Step {i+1}: {tc['name']}({tc['args']}) -> {tc['result']}"
for i, tc in enumerate(result.tool_calls)
])
prompt = f"""Evaluate if this agent's tool usage trajectory is coherent and efficient.
Question: {test_case.input}
Trajectory:
{trajectory_str}
Score from 0.0 to 1.0 where:
- 1.0: Every tool call was necessary and logical
- 0.5: Some unnecessary or redundant calls
- 0.0: Trajectory is incoherent or wasteful
Respond with ONLY a number between 0.0 and 1.0."""
try:
response = self.llm.invoke([HumanMessage(content=prompt)])
score = float(response.content.strip())
return max(0.0, min(1.0, score))
except (ValueError, Exception):
return 0.5Assembling the Full Evaluation Pipeline
Now let's tie everything together into a complete evaluation pipeline that runs all test cases, applies all evaluators, and generates a report:
# agent_eval/run_evals.py
from agent_eval.harness.cases import TEST_CASES
from agent_eval.harness.runner import EvalRunner
from agent_eval.harness.evaluators import (
ExactMatchEvaluator,
ToolUsageEvaluator,
StepEfficiencyEvaluator,
ErrorHandlingEvaluator,
LLMJudgeEvaluator,
TrajectoryCoherenceEvaluator,
)
from rich.console import Console
from rich.table import Table
import json
def run_evaluation_suite(use_mocks: bool = True, verbose: bool = False):
console = Console()
# Initialize runner and evaluators
runner = EvalRunner(use_mocks=use_mocks)
evaluators = [
ExactMatchEvaluator(),
ToolUsageEvaluator(),
StepEfficiencyEvaluator(),
ErrorHandlingEvaluator(),
LLMJudgeEvaluator(),
TrajectoryCoherenceEvaluator(),
]
# Run all test cases
console.print("\n[bold blue]Running Agent Evaluation Suite[/bold blue]\n")
results = runner.run_suite(TEST_CASES)
# Apply evaluators
console.print("\n[bold blue]Evaluating Results[/bold blue]\n")
for result in results:
test_case = next(tc for tc in TEST_CASES if tc.id == result.test_id)
for evaluator in evaluators:
score = evaluator.evaluate(test_case, result)
result.scores[evaluator.name] = score
# Generate report
generate_report(results, console, verbose)
# Save detailed results
with open("eval_results.json", "w") as f:
json.dump([r.dict() for r in results], f, indent=2, default=str)
console.print("\n[green]Results saved to eval_results.json[/green]")
return results
def generate_report(results, console, verbose):
# Summary table
table = Table(title="Evaluation Summary")
table.add_column("Test ID", style="cyan")
table.add_column("Status", style="bold")
table.add_column("Steps", justify="right")
table.add_column("Time(s)", justify="right")
table.add_column("Match", justify="right")
table.add_column("Tools", justify="right")
table.add_column("Efficiency", justify="right")
table.add_column("Judge", justify="right")
table.add_column("Coherence", justify="right")
table.add_column("Avg", justify="right", style="bold")
all_scores = []
for r in results:
scores = r.scores
avg = sum(scores.values()) / len(scores) if scores else 0.0
all_scores.append(avg)
status = "[green]PASS[/green]" if r.passed else "[red]FAIL[/red]"
table.add_row(
r.test_id,
status,
str(r.steps_taken),
f"{r.execution_time:.2f}",
f"{scores.get('exact_match', 0):.2f}",
f"{scores.get('tool_usage', 0):.2f}",
f"{scores.get('step_efficiency', 0):.2f}",
f"{scores.get('llm_judge', 0):.2f}",
f"{scores.get('trajectory_coherence', 0):.2f}",
f"{avg:.2f}",
)
console.print(table)
# Overall metrics
overall_avg = sum(all_scores) / len(all_scores) if all_scores else 0
pass_rate = sum(1 for r in results if r.passed) / len(results)
console.print(f"\n[bold]Overall Average Score:[/bold] {overall_avg:.2f}")
console.print(f"[bold]Pass Rate:[/bold] {pass_rate:.1%}")
console.print(f"[bold]Total Test Cases:[/bold] {len(results)}")
if verbose:
console.print("\n[bold blue]Detailed Trajectories[/bold blue]\n")
for r in results:
console.print(f"[cyan]{r.test_id}[/cyan]: {r.input}")
console.print(f" Output: {r.output[:200]}")
for tc in r.tool_calls:
console.print(f" Tool: {tc['name']}({tc['args']})")
if r.errors:
for err in r.errors:
console.print(f" [red]Error: {err}[/red]")
console.print()
if __name__ == "__main__":
run_evaluation_suite(use_mocks=True, verbose=True)Integrating with LangSmith for Tracing
LangSmith provides powerful tracing and observability. Integrating it with your harness gives you deep insights into agent execution:
# agent_eval/harness/langsmith_integration.py
from langsmith import Client
from agent_eval.harness.runner import TestResult
from typing import List
import os
class LangSmithReporter:
def __init__(self, project_name: str = "agent-eval"):
self.client = Client()
self.project_name = project_name
def create_dataset(self, test_cases):
"""Create a LangSmith dataset from test cases."""
dataset_name = f"agent-eval-{int(time.time())}"
dataset = self.client.create_dataset(
dataset_name=dataset_name,
description="Agent evaluation test cases"
)
for tc in test_cases:
self.client.create_example(
inputs={"input": tc.input, "test_id": tc.id},
outputs={"expected_answer": tc.expected_answer or ""},
dataset_id=dataset.id,
)
return dataset
def log_results(self, results: List[TestResult]):
"""Log evaluation results to LangSmith for visualization."""
for result in results:
self.client.create_run(
name=f"eval-{result.test_id}",
run_type="chain",
inputs={"input": result.input},
outputs={"output": result.output},
tags=["evaluation", result.test_id],
extra={
"scores": result.scores,
"steps_taken": result.steps_taken,
"execution_time": result.execution_time,
"tool_calls": result.tool_calls,
}
)
print(f"Logged {len(results)} results to LangSmith project: {self.project_name}")Best Practices for Agent Evaluation
1. Start Simple, Then Expand
Begin with a small set of 10-20 test cases covering your most common scenarios. As you discover edge cases in production, add them to the suite. A small suite that runs frequently is more valuable than a large suite that never runs.
2. Mock External Dependencies
Always use mocked tools in your evaluation harness. Real API calls introduce latency, cost, and non-determinism that make results hard to reproduce. Save real API testing for integration tests in a separate environment.
3. Evaluate Trajectories, Not Just Answers
A correct answer achieved through a convoluted path is still a problem. Include trajectory evaluators that check for unnecessary tool calls, redundant reasoning, and circular logic. The TrajectoryCoherenceEvaluator above is a good starting point.
4. Use Multiple Evaluator Types
Don't rely on a single metric. Combine rule-based evaluators (fast, deterministic) with LLM-based judges (flexible, nuanced). Rule-based evaluators catch obvious failures; LLM judges catch subtle quality issues.
5. Version Your Test Cases
Treat your test cases as code. Store them in version control, review changes, and tag them with metadata about when they were added and why. This creates an audit trail of what your agent is expected to handle.
6. Run Evaluations in CI/CD
Integrate the harness into your CI pipeline so every pull request triggers an evaluation run. Fail builds when scores drop below a threshold. Here's a simple GitHub Actions integration:
# .github/workflows/agent-eval.yml
name: Agent Evaluation
on: [pull_request]
jobs:
evaluate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.11"
- run: pip install -r requirements.txt
- name: Run evaluation suite
env:
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: python -m agent_eval.run_evals
- name: Check threshold
run: |
AVG_SCORE=$(python -c "import json; results=json.load(open('eval_results.json')); print(sum(sum(r['scores'].values())/len(r['scores']) for r in results)/len(results))")
echo "Average score: $AVG_SCORE"
if (( $(echo "$AVG_SCORE < 0.8" | bc -l) )); then
echo "Evaluation failed: average score below 0.8 threshold"
exit 1
fi7. Track Metrics Over Time
Store evaluation results across runs to identify trends. A gradual decline in scores might indicate model drift or accumulated technical debt. Use LangSmith or a simple database to track metrics historically:
# agent_eval/harness/metrics_tracker.py
import sqlite3
import json
from datetime import datetime
from agent_eval.harness.runner import TestResult
from typing import List
class MetricsTracker:
def __init__(self, db_path: str = "eval_metrics.db"):
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
self.conn.execute("""
CREATE TABLE IF NOT EXISTS eval_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT,
test_id TEXT,
passed INTEGER,
avg_score REAL,
steps_taken INTEGER,
execution_time REAL,
scores_json TEXT
)
""")
self.conn.commit()
def record(self, results: List[TestResult]):
timestamp = datetime.now().isoformat()
for r in results:
avg = sum(r.scores.values()) / len(r.scores) if r.scores else 0.0
self.conn.execute(
"INSERT INTO eval_runs (timestamp, test_id, passed, avg_score, steps_taken, execution_time, scores_json) VALUES (?, ?, ?, ?, ?, ?, ?)",
(timestamp, r.test_id, int(r.passed), avg, r.steps_taken, r.execution_time, json.dumps(r.scores))
)
self.conn.commit()
def get_trend(self, test_id: str, limit: int = 10):
cursor = self.conn.execute(
"SELECT timestamp, avg_score FROM eval_runs WHERE test_id = ? ORDER BY timestamp DESC LIMIT ?",
(test_id, limit)
)
return cursor.fetchall()
def close(self):
self.conn.close()8. Test Failure Modes Deliberately
Don't just test happy paths. Include test cases for invalid inputs, missing data, rate limits, and contradictory instructions. An agent that handles edge cases gracefully is far more valuable than one that only works in ideal conditions.
Advanced: A/B Testing Agent Variants
Once your harness is solid, you can use it to compare different agent configurations—different models, prompts, or tool sets:
# agent_eval/harness/ab_testing.py
from agent_eval.harness.runner import EvalRunner
from agent_eval.harness.cases import TEST_CASES
from typing import Dict, List
def compare_variants(variants: Dict[str, dict]):
"""Compare multiple agent configurations.
Args:
variants: Dict mapping variant name to config dict with keys like
'model_name', 'system_prompt', etc.
"""
all_results = {}
for name, config in variants.items():
print(f"\n=== Running variant: {name} ===")
runner = EvalRunner(
use_mocks=True,
model_name=config.get("model_name", "gpt-4o")
)
results = runner.run_suite(TEST_CASES)
all_results[name] = results
# Compare averages
print("\n=== Comparison Summary ===")
for name, results in all_results.items():
pass_rate = sum(1 for r in results if r.passed) / len(results)
avg_steps = sum(r.steps_taken for r in results) / len(results)
avg_time = sum(r.execution_time for r in results) / len(results)
print(f"{name}:")
print(f" Pass rate: {pass_rate:.1%}")
print(f" Avg steps: {avg_steps:.1f}")
print(f" Avg time: {avg_time:.2f}s")
return all_results
# Example usage
if __name__ == "__main__":
variants = {
"gpt-4o": {"model_name": "gpt-4o"},
"gpt-4o-mini": {"model_name": "gpt-4o-mini"},
}
compare_variants(variants)Conclusion
Building an agent evaluation harness with LangGraph transforms agent development from an ad-hoc process into a disciplined engineering practice. By systematically defining test cases, mocking external dependencies, implementing multi-dimensional evaluators, and integrating with CI/CD pipelines, you create a feedback loop that drives continuous improvement. The harness catches regressions early, quantifies trajectory quality, and gives you the confidence to iterate aggressively on agent design. Start with a small suite of representative test cases, expand as you learn from production, and treat your evaluation harness as a first-class artifact in your agent development lifecycle. The investment pays dividends every time you ship a change with evidence that your agent still performs as expected.