Introduction to Agent Evaluation Harnesses
Building AI agents is the easy part. Knowing whether they actually work — consistently, reliably, and at scale — is the hard part. When you ship a CrewAI crew into production, you need a way to measure how well it performs across a range of inputs, edge cases, and failure modes. That's where an agent evaluation harness comes in.
An evaluation harness is a structured framework that runs your agents against a predefined set of test cases, captures their outputs, scores them against expected outcomes or quality rubrics, and produces actionable metrics. Think of it as a unit test suite, but designed for the non-deterministic, probabilistic behavior of LLM-powered agents.
Why Evaluation Matters for CrewAI Agents
CrewAI agents differ from simple LLM calls. They reason, delegate, use tools, and collaborate with other agents. Each of those steps introduces variability. Without evaluation, you're flying blind when you:
- Swap one LLM provider for another (e.g., GPT-4o to Claude 3.5)
- Change a system prompt or role definition
- Add or remove a tool from an agent's toolkit
- Modify the crew's process flow or delegation rules
- Promote a crew from prototype to production
A well-built harness gives you regression protection. Before you ship a change, you run the suite. If accuracy drops on any test case, you know immediately.
Core Components of an Evaluation Harness
Before writing code, let's define the architecture. A robust harness has five components:
- Test cases — input scenarios paired with expected outputs or evaluation criteria
- Runner — orchestrates crew execution for each test case, handling retries and timeouts
- Evaluators — functions or LLM-as-judge prompts that score outputs
- Metrics aggregator — collects scores and computes summary statistics
- Reporter — formats results for human review (console, JSON, HTML)
Setting Up Your Project
Start by installing CrewAI and creating a project structure. We'll assume Python 3.10+.
pip install crewai crewai-tools python-dotenv pydanticCreate the following directory layout:
eval_harness/
├── crews/
│ └── research_crew.py
├── harness/
│ ├── __init__.py
│ ├── runner.py
│ ├── evaluators.py
│ ├── metrics.py
│ └── reporter.py
├── test_cases/
│ └── research_cases.json
└── run_evals.pyBuilding a Sample Crew to Evaluate
First, let's define a crew we can evaluate. This is a simple research crew with two agents: a researcher and a writer.
# crews/research_crew.py
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool
search_tool = SerperDevTool()
researcher = Agent(
role="Senior Research Analyst",
goal="Find accurate, up-to-date information on the given topic",
backstory="You are an expert analyst with 15 years of experience "
"synthesizing complex information into clear insights.",
tools=[search_tool],
verbose=False,
allow_delegation=False,
)
writer = Agent(
role="Technical Writer",
goal="Produce a clear, well-structured report from research findings",
backstory="You are a Pulitzer-nominated writer who excels at making "
"technical topics accessible without losing accuracy.",
verbose=False,
allow_delegation=False,
)
def build_research_crew(topic: str) -> Crew:
research_task = Task(
description=f"Research the topic: '{topic}'. "
f"Find at least 3 key facts and 2 recent developments.",
expected_output="A bullet list of key findings with sources.",
agent=researcher,
)
writing_task = Task(
description="Write a 300-word summary report based on the research findings.",
expected_output="A polished markdown report with a title, summary, and key points.",
agent=writer,
)
return Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential,
verbose=False,
)Defining Test Cases
Test cases are the backbone of your harness. Each case should have an input, an expected output (or expected properties), and metadata. Store them as JSON for portability.
// test_cases/research_cases.json
[
{
"id": "rc_001",
"category": "factual",
"input": "the impact of quantum computing on cryptography",
"expected_keywords": ["Shor", "RSA", "post-quantum", "encryption"],
"min_word_count": 200,
"max_word_count": 500
},
{
"id": "rc_002",
"category": "recent_events",
"input": "latest developments in fusion energy",
"expected_keywords": ["ITER", "tokamak", "net energy", "plasma"],
"min_word_count": 200,
"max_word_count": 500
},
{
"id": "rc_003",
"category": "edge_case",
"input": "",
"expected_keywords": [],
"should_fail": true,
"min_word_count": 0,
"max_word_count": 0
}
]Notice the third test case — an empty input. Good harnesses test failure modes, not just happy paths. Your crew should handle this gracefully, and your harness should verify that it does.
Building the Runner
The runner executes the crew for each test case and captures the result. It needs to handle timeouts, exceptions, and output extraction.
# harness/runner.py
import time
import traceback
from dataclasses import dataclass, field
from typing import Any, Optional
from crewai import Crew
@dataclass
class RunResult:
test_id: str
success: bool
output: Optional[str]
error: Optional[str]
duration_seconds: float
metadata: dict = field(default_factory=dict)
def run_single_case(crew: Crew, test_case: dict, timeout: int = 120) -> RunResult:
start = time.time()
test_id = test_case["id"]
try:
result = crew.kickoff()
elapsed = time.time() - start
output = str(result) if result else ""
return RunResult(
test_id=test_id,
success=True,
output=output,
error=None,
duration_seconds=round(elapsed, 2),
metadata={"raw_result": result.raw if hasattr(result, "raw") else None},
)
except Exception as e:
elapsed = time.time() - start
return RunResult(
test_id=test_id,
success=False,
output=None,
error=f"{type(e).__name__}: {str(e)}\n{traceback.format_exc()}",
duration_seconds=round(elapsed, 2),
)
def run_suite(crew_factory, test_cases: list, timeout: int = 120) -> list:
results = []
for case in test_cases:
print(f" Running test {case['id']}...")
crew = crew_factory(case["input"])
result = run_single_case(crew, case, timeout)
results.append(result)
return resultsWriting Evaluators
Evaluators inspect the output and produce a score. You'll typically use a mix of deterministic checks (keyword presence, word count) and LLM-as-judge evaluations for subjective quality.
# harness/evaluators.py
import re
from dataclasses import dataclass
from typing import Callable
@dataclass
class EvalScore:
name: str
passed: bool
score: float # 0.0 to 1.0
detail: str
def keyword_evaluator(output: str, expected_keywords: list) -> EvalScore:
if not expected_keywords:
return EvalScore("keyword_check", True, 1.0, "No keywords required.")
output_lower = output.lower()
found = [kw for kw in expected_keywords if kw.lower() in output_lower]
score = len(found) / len(expected_keywords)
missing = set(expected_keywords) - set(found)
detail = f"Found {len(found)}/{len(expected_keywords)}."
if missing:
detail += f" Missing: {missing}"
return EvalScore("keyword_check", score >= 0.5, score, detail)
def word_count_evaluator(output: str, min_words: int, max_words: int) -> EvalScore:
word_count = len(re.findall(r"\b\w+\b", output))
passed = min_words <= word_count <= max_words
detail = f"Word count: {word_count} (expected {min_words}-{max_words})"
return EvalScore("word_count", passed, 1.0 if passed else 0.0, detail)
def failure_evaluator(output: str, should_fail: bool, error: str) -> EvalScore:
if should_fail:
passed = error is not None and len(error) > 0
return EvalScore("expected_failure", passed, 1.0 if passed else 0.0,
f"Expected failure, got: {'failure' if passed else 'success'}")
return EvalScore("expected_failure", True, 1.0, "No failure expected.")
def evaluate_result(run_result, test_case: dict) -> list:
scores = []
if run_result.success and run_result.output:
scores.append(keyword_evaluator(
run_result.output, test_case.get("expected_keywords", [])
))
scores.append(word_count_evaluator(
run_result.output,
test_case.get("min_word_count", 0),
test_case.get("max_word_count", 10000),
))
else:
scores.append(EvalScore("keyword_check", False, 0.0, "No output produced."))
scores.append(failure_evaluator(
run_result.output or "",
test_case.get("should_fail", False),
run_result.error or "",
))
return scoresAdding an LLM-as-Judge Evaluator
For subjective quality dimensions — clarity, accuracy, tone — use an LLM to judge the output. This is the most powerful evaluator type, but it adds cost and latency.
# Add to harness/evaluators.py
from crewai import Agent, Task, Crew, Process
from litellm import completion
import json
import os
def llm_judge_evaluator(output: str, input_topic: str, rubric: str) -> EvalScore:
prompt = f"""You are an expert evaluator. Score the following AI-generated report.
Topic: {input_topic}
Report:
---
{output}
---
Evaluation rubric:
{rubric}
Respond ONLY with valid JSON in this format:
{{"score": 0.0-1.0, "reasoning": "one sentence explanation", "passed": true/false}}
"""
try:
response = completion(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
api_key=os.getenv("OPENAI_API_KEY"),
)
content = response.choices[0].message.content.strip()
# Strip markdown code fences if present
if content.startswith(""):
content = content.split("")[1]
if content.startswith("json"):
content = content[4:]
result = json.loads(content)
return EvalScore(
"llm_judge",
result.get("passed", result.get("score", 0) >= 0.7),
float(result.get("score", 0.0)),
result.get("reasoning", "No reasoning provided."),
)
except Exception as e:
return EvalScore("llm_judge", False, 0.0, f"Judge error: {e}")Aggregating Metrics
Once you have scores from all evaluators across all test cases, aggregate them into summary metrics. This is what you'll review after each run.
# harness/metrics.py
from collections import defaultdict
from dataclasses import dataclass, field
@dataclass
class SuiteMetrics:
total_cases: int = 0
passed_cases: int = 0
failed_cases: int = 0
pass_rate: float = 0.0
avg_duration: float = 0.0
evaluator_scores: dict = field(default_factory=dict)
case_results: list = field(default_factory=list)
def aggregate(results: list, all_scores: list) -> SuiteMetrics:
metrics = SuiteMetrics()
metrics.total_cases = len(results)
durations = []
eval_totals = defaultdict(lambda: {"sum": 0.0, "count": 0})
for run_result, scores in zip(results, all_scores):
case_passed = all(s.passed for s in scores)
if case_passed:
metrics.passed_cases += 1
else:
metrics.failed_cases += 1
durations.append(run_result.duration_seconds)
case_detail = {
"test_id": run_result.test_id,
"success": run_result.success,
"passed": case_passed,
"duration": run_result.duration_seconds,
"scores": [
{"name": s.name, "passed": s.passed, "score": s.score, "detail": s.detail}
for s in scores
],
}
if run_result.error:
case_detail["error"] = run_result.error[:500]
metrics.case_results.append(case_detail)
for s in scores:
eval_totals[s.name]["sum"] += s.score
eval_totals[s.name]["count"] += 1
metrics.pass_rate = metrics.passed_cases / metrics.total_cases if metrics.total_cases else 0
metrics.avg_duration = round(sum(durations) / len(durations), 2) if durations else 0
metrics.evaluator_scores = {
name: round(v["sum"] / v["count"], 3) for name, v in eval_totals.items()
}
return metricsReporting Results
The reporter formats metrics for human consumption. A console reporter is essential for local development; a JSON reporter is essential for CI/CD pipelines.
# harness/reporter.py
import json
def report_console(metrics):
print("\n" + "=" * 60)
print(" EVALUATION SUITE RESULTS")
print("=" * 60)
print(f" Total cases: {metrics.total_cases}")
print(f" Passed: {metrics.passed_cases}")
print(f" Failed: {metrics.failed_cases}")
print(f" Pass rate: {metrics.pass_rate:.1%}")
print(f" Avg duration: {metrics.avg_duration}s")
print("-" * 60)
print(" Evaluator Scores:")
for name, score in metrics.evaluator_scores.items():
status = "PASS" if score >= 0.7 else "WARN" if score >= 0.5 else "FAIL"
print(f" [{status}] {name:20s} {score:.3f}")
print("-" * 60)
print(" Case Details:")
for case in metrics.case_results:
status = "PASS" if case["passed"] else "FAIL"
print(f" [{status}] {case['test_id']} ({case['duration']}s)")
for s in case["scores"]:
flag = "+" if s["passed"] else "-"
print(f" {flag} {s['name']}: {s['detail']}")
print("=" * 60 + "\n")
def report_json(metrics, filepath: str):
data = {
"total_cases": metrics.total_cases,
"passed_cases": metrics.passed_cases,
"failed_cases": metrics.failed_cases,
"pass_rate": metrics.pass_rate,
"avg_duration": metrics.avg_duration,
"evaluator_scores": metrics.evaluator_scores,
"case_results": metrics.case_results,
}
with open(filepath, "w") as f:
json.dump(data, f, indent=2)
print(f"JSON report saved to {filepath}")Wiring It All Together
Now create the entry point that loads test cases, runs the suite, evaluates, aggregates, and reports.
# run_evals.py
import json
import os
from dotenv import load_dotenv
from crews.research_crew import build_research_crew
from harness.runner import run_suite
from harness.evaluators import evaluate_result
from harness.metrics import aggregate
from harness.reporter import report_console, report_json
load_dotenv()
def main():
with open("test_cases/research_cases.json") as f:
test_cases = json.load(f)
print(f"Loaded {len(test_cases)} test cases. Starting evaluation...\n")
results = run_suite(build_research_crew, test_cases, timeout=180)
all_scores = []
for run_result, test_case in zip(results, test_cases):
scores = evaluate_result(run_result, test_case)
all_scores.append(scores)
metrics = aggregate(results, all_scores)
report_console(metrics)
report_json(metrics, "eval_results.json")
# Exit code for CI integration
exit(0 if metrics.pass_rate >= 0.8 else 1)
if __name__ == "__main__":
main()Run it with:
python run_evals.pyBest Practices
1. Version Your Test Cases
Keep test cases in version control alongside your crew definitions. When you change a prompt, you can immediately see which test cases regress. Tag test case files with the crew version they correspond to.
2. Use Deterministic Seeds Where Possible
LLM outputs are non-deterministic, but you can reduce variance by setting temperature to 0 during evaluation runs and pinning model versions. This won't eliminate all variance, but it makes results more reproducible.
3. Separate Fast and Slow Test Suites
Not every test needs to run on every commit. Create a fast suite (5-10 cases, deterministic evaluators only) for local development and a comprehensive suite (50+ cases, including LLM-judge) for pre-release validation.
4. Track Metrics Over Time
Save JSON results from each run with a timestamp. Over time, you'll build a picture of how your crew's performance evolves. A simple trend line of pass rate per commit is invaluable for catching gradual regressions.
5. Include Adversarial and Edge Cases
Happy-path tests give you a false sense of security. Include cases with empty inputs, extremely long inputs, ambiguous queries, and prompts designed to trigger tool misuse. Your harness should verify the crew fails gracefully, not just that it succeeds on easy inputs.
6. Cache Tool Calls During Evaluation
If your agents use web search or API tools, cache the responses during evaluation runs. This reduces cost, speeds up the suite, and makes results more reproducible. CrewAI tools can be wrapped with a caching layer using functools.lru_cache or a disk-based cache like diskcache.
7. Set Realistic Pass Thresholds
Don't set your CI gate at 100% pass rate. LLM variance means some cases will occasionally fail even with a good crew. A threshold of 80-90% with mandatory passing of critical cases is more practical. Mark critical cases explicitly in your test case JSON and enforce them separately.
Conclusion
Building an evaluation harness for your CrewAI agents transforms agent development from guesswork into a measurable, iterative process. By defining test cases, running your crew against them, scoring outputs with both deterministic and LLM-based evaluators, and tracking metrics over time, you gain the confidence to refactor prompts, swap models, and add tools without fear of silent regressions. Start small — a handful of test cases and two or three evaluators is enough to deliver immediate value — then expand your suite as your crews grow in complexity. The investment pays off the first time your harness catches a regression before your users do.