Building an Agent Evaluation Harness with AutoGen: Complete Guide

Building an Agent Evaluation Harness with AutoGen: Complete Guide

As AI agents grow more capable and autonomous, evaluating their behavior becomes one of the most important challenges in production-grade LLM applications. Unlike single-turn prompts, agents make multi-step decisions, call tools, collaborate with other agents, and recover from errors. Traditional benchmarks fall short. This guide walks you through building a robust agent evaluation harness using Microsoft's AutoGen framework — from first principles to a working, extensible system you can drop into your own projects.

What Is an Agent Evaluation Harness?

An agent evaluation harness is a structured system that runs an agent (or multi-agent team) against a curated set of tasks, captures intermediate behavior, and scores outcomes against defined metrics. Think of it as a test suite — but instead of asserting function outputs, you're asserting that an agent reached a correct conclusion, used tools appropriately, stayed within budget, and produced safe, useful responses.

A good harness separates three concerns:

• Task definition — the input, expected behavior, and success criteria.
• Execution environment — the agent(s), tools, and runtime context.
• Evaluation — the metrics, judges, and reporting layer.

Why It Matters

Without an evaluation harness, agent development becomes guesswork. You tweak a prompt, run a few examples by hand, and ship based on vibes. This approach breaks down fast as agents become more complex. A harness gives you:

• Reproducibility — the same task always runs the same way.
• Regression detection — changes to prompts, tools, or models are measured, not assumed.
• Cost visibility — token usage, tool calls, and latency per task.
• Safety auditing — traces of every decision for post-hoc review.
• Comparison — apples-to-apples comparison between models, prompts, or agent topologies.

Prerequisites and Setup

Install AutoGen and a few supporting libraries. We'll use the v0.4+ async API, which is the current recommended approach.

pip install "autogen-agentchat" "autogen-ext[openai]" python-dotenv pydantic

Create a .env file with your API key:

OPENAI_API_KEY=sk-...

Designing the Harness Architecture

Our harness will consist of four components: a Task schema, a Trace recorder, a Scorer interface, and the Harness orchestrator that ties them together. Let's build each piece.

Defining the Task Schema

Every evaluation starts with a well-defined task. We use Pydantic to enforce structure and make tasks serializable to JSON for dataset management.

from pydantic import BaseModel, Field
from typing import Any, Optional

class Task(BaseModel):
    id: str
    prompt: str
    expected_answer: Optional[str] = None
    expected_tool_calls: list[str] = Field(default_factory=list)
    max_steps: int = 10
    max_tokens: int = 8000
    metadata: dict[str, Any] = Field(default_factory=dict)

class TaskResult(BaseModel):
    task_id: str
    answer: str
    tool_calls: list[str] = Field(default_factory=list)
    steps: int = 0
    tokens_used: int = 0
    elapsed_seconds: float = 0.0
    raw_messages: list[dict] = Field(default_factory=list)
    passed: bool = False
    scores: dict[str, float] = Field(default_factory=dict)

The expected_answer and expected_tool_calls fields are optional — some tasks have ground truth, others rely on LLM-based judging. The max_steps and max_tokens fields enforce budget constraints, which are themselves evaluation criteria.

Building the Trace Recorder

AutoGen agents emit messages through their conversation history. We need a way to capture these into a structured trace that the scorer can inspect. AutoGen v0.4+ supports a run_stream API that yields events as they happen.

import time
import json
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

async def run_task(agent: AssistantAgent, task: Task) -> TaskResult:
    start = time.time()
    termination = MaxMessageTermination(max_messages=task.max_steps)
    
    messages: list[dict] = []
    tool_calls: list[str] = []
    tokens_used = 0
    
    result = await agent.run(task=task.prompt, cancellation_token=None)
    
    for msg in result.messages:
        msg_dict = msg.to_json()
        messages.append(msg_dict)
        if msg_dict.get("type") == "ToolCallExecutionEvent":
            tool_calls.append(msg_dict.get("content", ""))
        # Extract token usage if available
        if hasattr(msg, "models_usage") and msg.models_usage:
            tokens_used += msg.models_usage.input_tokens + msg.models_usage.output_tokens
    
    answer = ""
    if messages:
        last = messages[-1]
        answer = last.get("content", "") if isinstance(last.get("content"), str) else str(last.get("content", ""))
    
    return TaskResult(
        task_id=task.id,
        answer=answer,
        tool_calls=tool_calls,
        steps=len(messages),
        tokens_used=tokens_used,
        elapsed_seconds=time.time() - start,
        raw_messages=messages,
    )

This function runs a single agent against a single task and returns a populated TaskResult. The raw message history is preserved so scorers can inspect the full reasoning chain.

Creating the Scorer Interface

Scorers evaluate a TaskResult against its Task and return a numeric score. We define a base interface and then implement several concrete scorers covering the most common evaluation dimensions.

from abc import ABC, abstractmethod

class Scorer(ABC):
    name: str
    
    @abstractmethod
    async def score(self, task: Task, result: TaskResult) -> float:
        ...

class ExactMatchScorer(Scorer):
    name = "exact_match"
    
    async def score(self, task: Task, result: TaskResult) -> float:
        if not task.expected_answer:
            return 0.0
        return 1.0 if task.expected_answer.strip().lower() in result.answer.strip().lower() else 0.0

class ToolCallScorer(Scorer):
    name = "tool_calls"
    
    async def score(self, task: Task, result: TaskResult) -> float:
        if not task.expected_tool_calls:
            return 1.0
        called = set(result.tool_calls)
        expected = set(task.expected_tool_calls)
        if not expected:
            return 1.0
        return len(expected & called) / len(expected)

class BudgetScorer(Scorer):
    name = "budget"
    
    async def score(self, task: Task, result: TaskResult) -> float:
        step_ratio = result.steps / task.max_steps if task.max_steps else 0
        token_ratio = result.tokens_used / task.max_tokens if task.max_tokens else 0
        if step_ratio > 1 or token_ratio > 1:
            return 0.0
        return 1.0 - (step_ratio + token_ratio) / 2

These three scorers cover exact answer matching, tool usage correctness, and budget adherence. But many tasks don't have a single correct answer — they require qualitative judgment. For those, we use an LLM-as-judge scorer.

Adding an LLM-as-Judge Scorer

LLM-based evaluation is essential for open-ended tasks. The judge receives the task, the agent's answer, and a rubric, then returns a score. We use a separate model client to avoid self-evaluation bias where possible.

class LLMJudgeScorer(Scorer):
    name = "llm_judge"
    
    def __init__(self, model_client: OpenAIChatCompletionClient, rubric: str):
        self.model_client = model_client
        self.rubric = rubric
    
    async def score(self, task: Task, result: TaskResult) -> float:
        judge_prompt = f"""You are an expert evaluator. Score the agent's answer on a scale of 0.0 to 1.0.

Task: {task.prompt}
Agent Answer: {result.answer}

Rubric:
{self.rubric}

Respond with ONLY a JSON object: {{"score": float, "reasoning": "brief explanation"}}"""
        
        response = await self.model_client.create([{"role": "user", "content": judge_prompt}])
        content = response.content
        try:
            import re
            match = re.search(r'\{[^}]+\}', content)
            if match:
                data = json.loads(match.group())
                return float(data.get("score", 0.0))
        except (json.JSONDecodeError, ValueError):
            pass
        return 0.0

The rubric is task-specific. For example, a coding task might use: "The answer must include working code that solves the problem, handles edge cases, and includes brief explanations." A research task might use: "The answer must be factual, cite sources, and directly address the question."

Assembling the Harness

Now we bring everything together. The Harness class accepts an agent factory (so each task gets a fresh agent), a list of tasks, and a list of scorers. It runs all tasks, collects results, and produces a summary report.

import asyncio
from dataclasses import dataclass, field
from typing import Callable, Awaitable

AgentFactory = Callable[[], AssistantAgent]

@dataclass
class HarnessReport:
    total_tasks: int = 0
    passed: int = 0
    results: list[TaskResult] = field(default_factory=list)
    avg_scores: dict[str, float] = field(default_factory=dict)
    total_tokens: int = 0
    total_seconds: float = 0.0

class Harness:
    def __init__(
        self,
        agent_factory: AgentFactory,
        tasks: list[Task],
        scorers: list[Scorer],
    ):
        self.agent_factory = agent_factory
        self.tasks = tasks
        self.scorers = scorers
    
    async def evaluate(self) -> HarnessReport:
        report = HarnessReport(total_tasks=len(self.tasks))
        
        for task in self.tasks:
            agent = self.agent_factory()
            result = await run_task(agent, task)
            
            for scorer in self.scorers:
                score = await scorer.score(task, result)
                result.scores[scorer.name] = score
            
            result.passed = all(s >= 0.7 for s in result.scores.values()) if result.scores else False
            report.results.append(result)
            report.total_tokens += result.tokens_used
            report.total_seconds += result.elapsed_seconds
            if result.passed:
                report.passed += 1
            
            print(f"[{task.id}] passed={result.passed} scores={result.scores}")
        
        for scorer in self.scorers:
            scores = [r.scores.get(scorer.name, 0) for r in report.results]
            report.avg_scores[scorer.name] = sum(scores) / len(scores) if scores else 0.0
        
        return report

Putting It All Together: A Complete Example

Let's build a concrete evaluation run. We'll create a simple math-and-tool agent, define a small task dataset, and run the harness with all our scorers.

import os
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.tools import FunctionTool

# --- Define a simple tool ---
def calculate(expression: str) -> str:
    """Evaluate a math expression and return the result."""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Error: {e}"

calc_tool = FunctionTool(calculate, description="Evaluate a math expression", name="calculate")

# --- Model clients ---
api_key = os.getenv("OPENAI_API_KEY")
agent_model = OpenAIChatCompletionClient(model="gpt-4o-mini", api_key=api_key)
judge_model = OpenAIChatCompletionClient(model="gpt-4o", api_key=api_key)

# --- Agent factory ---
def make_agent() -> AssistantAgent:
    return AssistantAgent(
        name="math_agent",
        model_client=agent_model,
        tools=[calc_tool],
        system_message="You are a math assistant. Use the calculate tool for any arithmetic. Give concise final answers.",
    )

# --- Task dataset ---
tasks = [
    Task(
        id="math_001",
        prompt="What is 17 * 23 + 45?",
        expected_answer="436",
        expected_tool_calls=["calculate"],
        max_steps=6,
        max_tokens=3000,
    ),
    Task(
        id="math_002",
        prompt="Compute the square root of 144, then multiply by 5.",
        expected_answer="60",
        expected_tool_calls=["calculate"],
        max_steps=8,
        max_tokens=4000,
    ),
    Task(
        id="math_003",
        prompt="Explain why the sum of two odd numbers is always even, then verify with 7 + 11.",
        expected_answer="18",
        expected_tool_calls=["calculate"],
        max_steps=10,
        max_tokens=6000,
        metadata={"requires_explanation": True},
    ),
]

# --- Scorers ---
scorers = [
    ExactMatchScorer(),
    ToolCallScorer(),
    BudgetScorer(),
    LLMJudgeScorer(
        judge_model,
        rubric="Score 1.0 if the answer is correct and clearly explained. 0.5 if partially correct. 0.0 if wrong.",
    ),
]

# --- Run ---
async def main():
    harness = Harness(agent_factory=make_agent, tasks=tasks, scorers=scorers)
    report = await harness.evaluate()
    
    print("\n===== EVALUATION REPORT =====")
    print(f"Tasks: {report.total_tasks}")
    print(f"Passed: {report.passed}/{report.total_tasks}")
    print(f"Total tokens: {report.total_tokens}")
    print(f"Total time: {report.total_seconds:.2f}s")
    print("Average scores:")
    for name, avg in report.avg_scores.items():
        print(f"  {name}: {avg:.3f}")
    
    print("\nPer-task breakdown:")
    for r in report.results:
        print(f"  {r.task_id}: passed={r.passed} steps={r.steps} tokens={r.tokens_used} scores={r.scores}")

asyncio.run(main())

When you run this, you'll see per-task progress printed live, followed by a summary report showing pass rates, average scores per metric, and token/time costs. This output becomes your baseline — any change to the agent's prompt, model, or tools can now be measured against it.

Extending to Multi-Agent Teams

AutoGen's strength is multi-agent collaboration. The harness works with teams too — just swap the agent factory to return a RoundRobinGroupChat or any other team type. The run_task function already captures the full message stream, so scorers can inspect which agent said what.

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import MaxMessageTermination

def make_team() -> RoundRobinGroupChat:
    planner = AssistantAgent(
        name="planner",
        model_client=agent_model,
        system_message="Break down the problem into steps. Pass to the executor.",
    )
    executor = AssistantAgent(
        name="executor",
        model_client=agent_model,
        tools=[calc_tool],
        system_message="Execute the plan using the calculate tool. Report results.",
    )
    return RoundRobinGroupChat(
        participants=[planner, executor],
        termination_condition=MaxMessageTermination(max_messages=12),
    )

For teams, you'll want to adapt run_task to call team.run() instead of agent.run(). The rest of the harness — scorers, reporting, task schema — stays identical. This separation of concerns is what makes the design extensible.

Best Practices

• Version your datasets. Store tasks as JSON files in a versioned directory. Tag every evaluation run with the dataset version, model version, and agent configuration so results are comparable over time.
• Use a separate judge model. Self-evaluation (same model as agent and judge) inflates scores. Use a stronger model for judging, or at minimum a different model family.
• Cap concurrency carefully. Running tasks in parallel speeds things up, but rate limits and cost spikes are real. Start with sequential execution, then add asyncio.Semaphore for controlled parallelism.
• Log full traces. Save raw_messages to disk for every run. When a task fails, you need the full conversation to debug — a pass/fail boolean isn't enough.
• Define pass thresholds deliberately. The 0.7 threshold in our example is arbitrary. Tune it per use case: safety-critical tasks might require 1.0 on certain scorers, while creative tasks might accept 0.5.
• Test your scorers. Scorers are code too. Create golden examples where you know the expected score and verify your scorers produce them. A buggy judge silently corrupts your entire evaluation.
• Watch for prompt leakage. If your judge prompt includes the agent's full trace, the judge might reward verbose answers. Keep judge inputs focused on the final answer and rubric unless you specifically need to evaluate the process.
• Track cost over time. Log total_tokens and elapsed_seconds across runs. A model upgrade that improves accuracy but triples cost might not be worth it.

Persisting Results

Evaluation results are only useful if you can compare them across runs. Here's a simple persistence layer that writes results to JSON and appends a summary to a CSV log.

import csv
from datetime import datetime
from pathlib import Path

def save_report(report: HarnessReport, output_dir: str = "eval_results"):
    out = Path(output_dir)
    out.mkdir(exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    
    # Full results as JSON
    results_file = out / f"results_{timestamp}.json"
    with open(results_file, "w") as f:
        json.dump({
            "timestamp": timestamp,
            "total_tasks": report.total_tasks,
            "passed": report.passed,
            "total_tokens": report.total_tokens,
            "total_seconds": report.total_seconds,
            "avg_scores": report.avg_scores,
            "results": [r.model_dump() for r in report.results],
        }, f, indent=2, default=str)
    
    # Append summary to CSV
    csv_file = out / "summary.csv"
    write_header = not csv_file.exists()
    with open(csv_file, "a", newline="") as f:
        writer = csv.writer(f)
        if write_header:
            writer.writerow(["timestamp", "total_tasks", "passed", "total_tokens", "total_seconds"] + list(report.avg_scores.keys()))
        writer.writerow([timestamp, report.total_tasks, report.passed, report.total_tokens, f"{report.total_seconds:.2f}"] + [f"{v:.3f}" for v in report.avg_scores.values()])
    
    print(f"Results saved to {results_file}")
    print(f"Summary appended to {csv_file}")

Call save_report(report) at the end of your main() function. Over time, the CSV becomes a dashboard of how your agent's performance evolves with each change.

Conclusion

Building an agent evaluation harness with AutoGen gives you a repeatable, measurable foundation for agent development. By separating task definition, execution, and scoring into distinct components, you create a system that scales from a handful of smoke-test tasks to large benchmark datasets. The scorers we built — exact match, tool call verification, budget adherence, and LLM-as-judge — cover the most common evaluation needs, and the plugin architecture means you can add domain-specific scorers without touching the core. Start small: define five to ten representative tasks, run the harness, and establish a baseline. From there, every prompt tweak, model swap, or tool addition becomes a measured experiment rather than a leap of faith. The agents that ship to production are the ones you can trust — and trust comes from evidence.