7. Benchmarks and Agent Metrics

Overview

In the previous capsules you built a complete testing infrastructure: unit tests, integration tests, trajectory evaluation, LangSmith, and golden datasets with regression testing. All of that answers "does my agent work?" But there's a question you still can't answer precisely: how well does it work? Not "it seems to work fine." Rather: "task completion rate 87%, tool call accuracy 93%, average latency 3.8s, cost $0.07/task, reasoning quality 0.82." Numbers. Metrics. Benchmarks.

The difference between testing and benchmarking is the difference between "pass/fail" and "how much?" Tests tell you if something broke. Metrics tell you where you are and where you're heading. A benchmark is a standardized measurement you run periodically to quantify your agent's performance across multiple dimensions. With those numbers you can make decisions: is it worth switching from GPT-4o to GPT-4o-mini if task completion drops from 87% to 79% but the cost drops 70%? Without metrics, that decision is intuition. With metrics, it's engineering.

Connection to the module: This is the last technical capsule before the project. It takes everything built in capsules 02-06 and elevates it to aggregate production metrics. In capsule 05 you configured LangSmith to record every run. In 06 you created golden datasets for regression testing. Here you turn that data into a metrics dashboard that lets you talk about your agent with numbers, compare versions, and make data-driven decisions.

Core Metrics for Agents

An agent in production needs five fundamental metrics. Each one captures a different dimension of quality:

Core Metrics for Agents
═══════════════════════════════════════════════════════
1. Task Completion Rate    → Does it do what you ask?
2. Tool Call Accuracy      → Does it use the right tools?
3. Reasoning Quality       → Does it think well?
4. Latency                 → Is it fast?
5. Cost per Task           → Is it resource-efficient?
═══════════════════════════════════════════════════════
The first three measure QUALITY. The last two measure EFFICIENCY.

Defining each metric

from dataclasses import dataclass, field

@dataclass
class TaskResult:
    task_id: str
    input: str
    output: str
    completed: bool
    tools_used: list[str] = field(default_factory=list)
    steps: int = 0
    latency_ms: float = 0.0
    total_tokens: int = 0
    cost_usd: float = 0.0

Task Completion Rate — the percentage of tasks completed satisfactorily. What does "satisfactorily" mean? For a Research Agent, that it produced a report that answers the question and doesn't invent information. The definition is yours — but it needs to be consistent.

Tool Call Accuracy — the percentage of tool calls appropriate to the task. An agent that calls web_search for "what's 2+2?" completes the task, but with poor tool selection. It has two variants: precision (of the ones it used, how many were correct?) and recall (of the ones it should have used, how many did it use?). For agents, recall matters more.

Reasoning Quality — the quality of the reasoning. It requires LLM-as-Judge because there's no programmatic way to evaluate whether a plan was "sensible." It's the most expensive metric — each evaluation is an extra call to the LLM judge. Run it on a sample, not on the whole dataset.

Latency — total execution time. The P95 is more relevant than the mean: if your mean is 3.5s but the P95 is 12s, 5% of your users are waiting more than 12 seconds.

Cost per Task — how much each run costs. If your agent costs $0.08/task with 10,000 daily runs, that's $24,000/month. Cutting it to $0.05/task saves $9,000/month. Benchmarking becomes a business argument.

Implementing the Metrics

The calculation functions

def task_completion_rate(results: list[TaskResult]) -> dict:
    if not results:
        return {"rate": 0.0, "completed": 0, "total": 0}
    completed = sum(1 for r in results if r.completed)
    return {"rate": round(completed / len(results), 4),
            "completed": completed, "total": len(results)}

def tool_call_accuracy(results: list[TaskResult],
                        expected_tools: dict[str, list[str]]) -> dict:
    correct_calls, total_calls = 0, 0
    for r in results:
        expected = set(expected_tools.get(r.task_id, []))
        if not expected:
            continue
        total_calls += len(expected)
        correct_calls += len(expected & set(r.tools_used))
    accuracy = correct_calls / total_calls if total_calls > 0 else 1.0
    return {"accuracy": round(accuracy, 4), "correct": correct_calls, "total_calls": total_calls}

def latency_stats(results: list[TaskResult]) -> dict:
    if not results:
        return {}
    latencies = sorted(r.latency_ms for r in results)
    n = len(latencies)
    return {
        "mean_ms": round(sum(latencies) / n, 1),
        "p95_ms": round(latencies[int(n * 0.95)], 1),
        "p99_ms": round(latencies[int(n * 0.99)], 1),
    }

def cost_per_task(results: list[TaskResult]) -> dict:
    if not results:
        return {}
    costs = [r.cost_usd for r in results]
    return {"mean_cost_usd": round(sum(costs) / len(costs), 4),
            "total_cost_usd": round(sum(costs), 4)}

Reasoning quality with LLM-as-Judge

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

class ReasoningJudgment(BaseModel):
    plan_quality: int = Field(ge=1, le=5)
    tool_selection: int = Field(ge=1, le=5)
    coherence: int = Field(ge=1, le=5)
    completeness: int = Field(ge=1, le=5)
    justification: str

REASONING_JUDGE_PROMPT = """Evaluate the reasoning quality of an AI agent.

## Assigned task
{task}

## The agent's trajectory
{trajectory}

## Final answer
{output}

Evaluate each dimension on a 1-5 scale. Be rigorous — a 5 requires excellence."""

def reasoning_quality(task: str, trajectory: str, output: str) -> float:
    judge = ChatOpenAI(model="gpt-4o", temperature=0)
    structured = judge.with_structured_output(ReasoningJudgment)
    prompt = REASONING_JUDGE_PROMPT.format(task=task, trajectory=trajectory, output=output)
    judgment = structured.invoke(prompt)
    return round(
        (judgment.plan_quality + judgment.tool_selection
         + judgment.coherence + judgment.completeness) / 20.0, 4)

The benchmark runner

import time
from langchain_core.messages import HumanMessage, AIMessage

@dataclass
class BenchmarkConfig:
    name: str
    model: str
    dataset_path: str
    n_runs: int = 1

@dataclass
class BenchmarkReport:
    config: BenchmarkConfig
    task_completion: dict = field(default_factory=dict)
    tool_accuracy: dict = field(default_factory=dict)
    latency: dict = field(default_factory=dict)
    cost: dict = field(default_factory=dict)
    reasoning_quality_mean: float = 0.0
    results: list[TaskResult] = field(default_factory=list)
    timestamp: str = ""

    def summary(self) -> str:
        return "\n".join([
            f"\n{'='*60}",
            f"  BENCHMARK: {self.config.name} | Model: {self.config.model}",
            f"{'='*60}",
            f"  Task Completion:  {self.task_completion.get('rate', 0):.1%}",
            f"  Tool Accuracy:    {self.tool_accuracy.get('accuracy', 0):.1%}",
            f"  Reasoning:        {self.reasoning_quality_mean:.2f}",
            f"  Latency (mean):   {self.latency.get('mean_ms', 0):.0f}ms",
            f"  Latency (P95):    {self.latency.get('p95_ms', 0):.0f}ms",
            f"  Cost (mean):      ${self.cost.get('mean_cost_usd', 0):.4f}",
            f"{'='*60}",
        ])

PRICING = {
    "gpt-4o": {"input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000},
    "gpt-4o-mini": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
    "gpt-4.1-mini": {"input": 0.40 / 1_000_000, "output": 1.60 / 1_000_000},
}

def run_benchmark(agent, tasks: list[dict], config: BenchmarkConfig,
                   expected_tools: dict[str, list[str]] | None = None) -> BenchmarkReport:
    from datetime import datetime
    results = []
    for task in tasks:
        start = time.time()
        try:
            response = agent.invoke({"messages": [HumanMessage(content=task["input"])]})
            elapsed_ms = (time.time() - start) * 1000
            messages = response["messages"]
            output = messages[-1].content if messages else ""
            tools = [tc["name"] for m in messages
                     if isinstance(m, AIMessage) and m.tool_calls for tc in m.tool_calls]
            steps = sum(1 for m in messages if isinstance(m, AIMessage) and m.tool_calls)
            tokens = sum(getattr(m, "usage_metadata", {}).get("total_tokens", 0)
                         for m in messages if hasattr(m, "usage_metadata") and m.usage_metadata)
            price = PRICING.get(config.model, PRICING["gpt-4o-mini"])
            cost = tokens * (price["input"] + price["output"]) / 2
            completed = evaluate_completion(output, task.get("expected", {}))
            results.append(TaskResult(task["id"], task["input"], output, completed,
                                      tools, steps, elapsed_ms, tokens, cost))
        except Exception as e:
            elapsed_ms = (time.time() - start) * 1000
            results.append(TaskResult(task["id"], task["input"], f"ERROR: {e}",
                                      False, latency_ms=elapsed_ms))

    return BenchmarkReport(
        config=config, task_completion=task_completion_rate(results),
        tool_accuracy=tool_call_accuracy(results, expected_tools or {}),
        latency=latency_stats(results), cost=cost_per_task(results),
        results=results, timestamp=datetime.now().isoformat())

def evaluate_completion(output: str, expected: dict) -> bool:
    if not expected:
        return len(output) > 20
    checks_passed, checks_total = 0, 0
    if "answer_contains" in expected:
        for kw in expected["answer_contains"]:
            checks_total += 1
            if kw.lower() in output.lower():
                checks_passed += 1
    if "min_length" in expected:
        checks_total += 1
        if len(output) >= expected["min_length"]:
            checks_passed += 1
    return (checks_passed / checks_total) >= 0.7 if checks_total > 0 else len(output) > 20

Standardized Benchmarks

Creating a reusable suite

An ad hoc benchmark is useless. You need a standardized suite — the same dataset, the same evaluators, the same process. What changes is the agent:

import json
from dataclasses import asdict

@dataclass
class BenchmarkSuite:
    name: str
    version: str
    tasks: list[dict]
    expected_tools: dict[str, list[str]]
    thresholds: dict[str, float]

    def save(self, path: str):
        with open(path, "w") as f:
            json.dump(asdict(self), f, indent=2, ensure_ascii=False)

    @classmethod
    def load(cls, path: str) -> "BenchmarkSuite":
        with open(path) as f:
            return cls(**json.load(f))

    def evaluate_report(self, report: BenchmarkReport) -> dict:
        metric_map = {
            "task_completion_rate": report.task_completion.get("rate", 0),
            "tool_call_accuracy": report.tool_accuracy.get("accuracy", 0),
            "reasoning_quality": report.reasoning_quality_mean,
            "mean_latency_ms": report.latency.get("mean_ms", 0),
            "mean_cost_usd": report.cost.get("mean_cost_usd", 0),
        }
        results = {}
        for metric, threshold in self.thresholds.items():
            actual = metric_map.get(metric, 0)
            passed = actual <= threshold if metric in ("mean_latency_ms", "mean_cost_usd") else actual >= threshold
            results[metric] = {"actual": actual, "threshold": threshold, "passed": passed}
        return results

Trend tracking

A single benchmark doesn't say much. The power comes from running the same benchmark over time:

@dataclass
class BenchmarkHistory:
    suite_name: str
    entries: list[dict] = field(default_factory=list)

    def add_from_report(self, report: BenchmarkReport):
        self.entries.append({
            "timestamp": report.timestamp, "model": report.config.model,
            "task_completion_rate": report.task_completion.get("rate", 0),
            "tool_call_accuracy": report.tool_accuracy.get("accuracy", 0),
            "reasoning_quality": report.reasoning_quality_mean,
            "mean_latency_ms": report.latency.get("mean_ms", 0),
            "mean_cost_usd": report.cost.get("mean_cost_usd", 0),
        })

    def trend(self, metric: str, last_n: int = 5) -> list[float]:
        return [e[metric] for e in self.entries[-last_n:]]

    def detect_regression(self, metric: str, threshold: float = 0.05) -> bool:
        values = self.trend(metric, last_n=3)
        if len(values) < 2:
            return False
        latest, baseline = values[-1], sum(values[:-1]) / len(values[:-1])
        if metric in ("mean_latency_ms", "mean_cost_usd"):
            return latest > baseline * (1 + threshold)
        return latest < baseline * (1 - threshold)

    def save(self, path: str):
        with open(path, "w") as f:
            json.dump({"suite_name": self.suite_name, "entries": self.entries}, f, indent=2)

    @classmethod
    def load(cls, path: str) -> "BenchmarkHistory":
        with open(path) as f:
            data = json.load(f)
        h = cls(data["suite_name"])
        h.entries = data["entries"]
        return h

Run the benchmark weekly or after each release. The history shows you whether your agent is improving, holding steady, or degrading.

The Metrics Dashboard

The dashboard turns raw data into actionable information. You don't need a sophisticated frontend — a programmatic dashboard that prints metrics with deltas and alerts is enough:

class MetricsDashboard:
    def __init__(self, history: BenchmarkHistory, thresholds: dict[str, float]):
        self.history = history
        self.thresholds = thresholds

    def render(self):
        if not self.history.entries:
            return
        latest = self.history.entries[-1]
        previous = self.history.entries[-2] if len(self.history.entries) > 1 else None

        print(f"\n{'='*60}")
        print(f"  AGENT METRICS DASHBOARD | {latest.get('timestamp', '?')}")
        print(f"{'='*60}")

        for label, key, fmt in [("Task Completion", "task_completion_rate", ".1%"),
                                 ("Tool Accuracy", "tool_call_accuracy", ".1%"),
                                 ("Reasoning", "reasoning_quality", ".2f"),
                                 ("Latency", "mean_latency_ms", ".0f"),
                                 ("Cost/task", "mean_cost_usd", ".4f")]:
            val = latest.get(key, 0)
            delta = ""
            if previous and (diff := val - previous.get(key, 0)):
                delta = f" ({'+' if diff > 0 else ''}{format(diff, fmt)})"
            alert = ""
            if (t := self.thresholds.get(key)):
                violated = val > t if key in ("mean_latency_ms", "mean_cost_usd") else val < t
                alert = " [!]" if violated else ""
            print(f"  {label:<18} {format(val, fmt)}{delta}{alert}")

        regressions = [k for k in self.thresholds if self.history.detect_regression(k)]
        if regressions:
            print(f"\n  ALERTS: {', '.join(regressions)}")
        print(f"{'='*60}")

============================================================
  AGENT METRICS DASHBOARD | 2025-11-15T14:28
============================================================
  Task Completion    87.0% (+2.0%)
  Tool Accuracy      91.7% (+4.2%)
  Reasoning          0.78 (+0.03)
  Latency            2847 (-320)
  Cost/task          0.0012 (-0.0003)
============================================================

The deltas show direction. The [!] alerts show when something crossed a threshold. For production, wire the alerts into a cron job — don't wait to look at the dashboard manually.

Metrics for Multi-Agent

In a multi-agent system (Supervisor + Researcher + Analyst + Writer), the aggregate metrics tell you that there's a problem. The per-agent metrics tell you where:

Multi-agent system — Task completion: 72% [!]
  Supervisor routing accuracy:    95%  ← OK
  Researcher search relevance:    68%  ← HERE
  Analyst findings quality:       85%  ← OK
  Writer report completeness:     90%  ← OK

Per-agent and coordination metrics

@dataclass
class AgentMetrics:
    agent_name: str
    invocations: int = 0
    successes: int = 0
    avg_latency_ms: float = 0.0
    avg_tool_calls: float = 0.0

    @property
    def success_rate(self) -> float:
        return self.successes / self.invocations if self.invocations > 0 else 0.0

@dataclass
class CoordinationMetrics:
    total_handoffs: int = 0
    successful_handoffs: int = 0
    supervisor_reassignments: int = 0
    redundant_work: int = 0

    @property
    def handoff_success_rate(self) -> float:
        return self.successful_handoffs / self.total_handoffs if self.total_handoffs > 0 else 0.0

    @property
    def supervisor_efficiency(self) -> float:
        if self.total_handoffs == 0:
            return 1.0
        return 1.0 - (self.supervisor_reassignments / self.total_handoffs)

Three coordination metrics matter especially:

1. Handoff success rate: Do the transfers between agents preserve context? A failed handoff means the receiving agent didn't understand the task.
2. Supervisor efficiency: Does the supervisor reassign a lot? Frequent reassignments = poor routing.
3. Redundant work: Are multiple agents doing the same work? If the Researcher and the Analyst both search "quantum computing" because the result wasn't shared, there's redundancy that costs tokens.

Comparing Models with Benchmarks

The most frequent question in production: can I use a cheaper model without losing quality?

def compare_models(model_names: list[str], suite: BenchmarkSuite, tools: list) -> dict:
    from langgraph.prebuilt import create_react_agent
    all_reports = {}
    for model_name in model_names:
        print(f"\nEvaluating {model_name}...")
        model = ChatOpenAI(model=model_name, temperature=0)
        agent = create_react_agent(model, tools=tools)
        config = BenchmarkConfig(name=f"compare-{model_name}", model=model_name, dataset_path="")
        report = run_benchmark(agent, suite.tasks, config, suite.expected_tools)
        all_reports[model_name] = report
    return all_reports

def print_comparison_table(reports: dict[str, BenchmarkReport]):
    models = list(reports.keys())
    metrics = [
        ("Task Completion", lambda r: r.task_completion.get("rate", 0), ".1%"),
        ("Tool Accuracy", lambda r: r.tool_accuracy.get("accuracy", 0), ".1%"),
        ("Latency (ms)", lambda r: r.latency.get("mean_ms", 0), ".0f"),
        ("Cost ($/task)", lambda r: r.cost.get("mean_cost_usd", 0), ".4f"),
    ]
    header = f"{'Metric':<18}" + "".join(f"{m:<16}" for m in models)
    print(f"\n{header}\n{'='*(18 + 16*len(models))}")
    for label, extractor, fmt in metrics:
        values = [extractor(reports[m]) for m in models]
        row = f"{label:<18}" + "".join(f"{format(v, fmt):<16}" for v in values)
        print(row)

Metric            gpt-4o          gpt-4o-mini     gpt-4.1-mini
================================================================
Task Completion   92.0%           79.0%           87.0%
Tool Accuracy     95.0%           83.3%           91.7%
Latency (ms)      3200            1800            2100
Cost ($/task)     0.0089          0.0008          0.0014

With this table you make concrete decisions. GPT-4o is the best in quality but the most expensive. GPT-4o-mini drops to 79% — below the threshold. GPT-4.1-mini has the best balance: 87% completion at 84% less cost than GPT-4o. Without benchmarks, "GPT-4o is better" is an opinion. With benchmarks, it's a fact a product manager can use.

To go deeper, break it down by task category. If GPT-4o-mini gets 95% on happy paths but 40% on multi-step, you know it fails at coordinating multiple tools — and you can use complexity-based routing (M8's Router pattern) to assign models by task.

Connection to the Project

In the module's project (capsule 08), benchmarks and metrics are your final product:

┌──────────────────────────────────────────────────────────┐
│      BENCHMARKS IN THE PROJECT — RESEARCH AGENT v6        │
│                                                          │
│  1. A versioned BenchmarkSuite (JSON) with 20+ tasks     │
│  2. Metrics calculated:                                  │
│     ├─ Task completion rate (target: ≥ 80%)              │
│     ├─ Tool call accuracy (target: ≥ 85%)                │
│     ├─ Reasoning quality (target: ≥ 0.70)               │
│     ├─ Mean latency (target: ≤ 5000ms)                  │
│     └─ Cost per task (target: ≤ $0.10)                   │
│  3. Model comparison (gpt-4o vs gpt-4o-mini)             │
│  4. Dashboard + trend history                            │
└──────────────────────────────────────────────────────────┘

The result: you can say "my Research Agent has a task completion rate of 87%, tool call accuracy of 92%, latency of 3.8s, cost $0.07/task." That's data. That's engineering.

Troubleshooting

Problem 1: The metrics vary a lot between runs of the same benchmark

Cause: LLM non-determinism — even with temperature=0, variations of 5-10% are normal.

Solution: Run the benchmark N times (3 minimum) and report the average and standard deviation. If the deviation is > 5%, your expectations are too strict. Use answer_contains instead of exact_match.

Problem 2: Reasoning quality gives inconsistent scores across runs

Cause: The LLM judge has its own variability.

Solution: Use temperature=0 for the judge. Structure the output with with_structured_output. Implement multi-judge (3 evaluations, averaged) for release benchmarks. Reserve it for important comparisons — the cost triples.

Problem 3: The calculated costs don't match the OpenAI bill

Cause: Outdated pricing or uncounted tokens (tool calls, the system prompt, retries).

Solution: Use LangChain's usage_metadata, which includes input_tokens and output_tokens. Check monthly against your bill and adjust the multiplier.

Problem 4: The benchmark takes too long to run regularly

Cause: 50+ tasks with a real LLM, 3-5 seconds each. 4+ minutes total. Nobody runs it.

Solution: A quick benchmark (10 tasks, < 1 minute) for CI. A full benchmark (50+ tasks, with reasoning quality) weekly or per release.

Exercises

Exercise 1: Compute the five core metrics (Easy)

You have this list of TaskResults. Compute the five core metrics and determine which pass the thresholds (completion ≥ 80%, accuracy ≥ 85%, latency ≤ 5000ms, cost ≤ $0.10).

sample_results = [
    TaskResult("T-01", "Calculate 15*20", "300", True, ["calculator"], 1, 1200, 80, 0.001),
    TaskResult("T-02", "Look up AI news", "News...", True, ["web_search"], 1, 2500, 200, 0.003),
    TaskResult("T-03", "Mexico's GDP and 5%", "The GDP...", True, ["web_search", "calculator"], 2, 4800, 350, 0.005),
    TaskResult("T-04", "Capital of Japan", "Tokyo", True, [], 0, 800, 50, 0.0005),
    TaskResult("T-05", "Send an email", "I can't...", False, [], 0, 600, 40, 0.0004),
    TaskResult("T-06", "BTC price", "BTC...", True, ["web_search", "calculator"], 3, 5200, 400, 0.006),
    TaskResult("T-07", "", "Error", False, [], 0, 300, 20, 0.0002),
    TaskResult("T-08", "NYC weather", "Weather...", True, ["web_search"], 1, 2100, 150, 0.002),
    TaskResult("T-09", "1024*768", "786432", True, ["calculator"], 1, 900, 60, 0.0006),
    TaskResult("T-10", "Python vs Java", "Python...", True, ["web_search"], 2, 3800, 280, 0.004),
]
sample_expected = {"T-01": ["calculator"], "T-02": ["web_search"],
    "T-03": ["web_search", "calculator"], "T-05": [], "T-06": ["web_search", "calculator"],
    "T-08": ["web_search"], "T-09": ["calculator"], "T-10": ["web_search"]}

completion = task_completion_rate(sample_results)
print(f"Task Completion: {completion['rate']:.1%}")  # 80.0%

accuracy = tool_call_accuracy(sample_results, sample_expected)
print(f"Tool Accuracy: {accuracy['accuracy']:.1%}")  # 100% (the tools used match the expected)

lat = latency_stats(sample_results)
print(f"Latency mean: {lat['mean_ms']:.0f}ms, P95: {lat['p95_ms']:.0f}ms")

cost = cost_per_task(sample_results)
print(f"Cost mean: ${cost['mean_cost_usd']:.4f}")

print(f"\nCompletion >= 80%: {'PASS' if completion['rate'] >= 0.80 else 'FAIL'}")
print(f"Accuracy >= 85%:  {'PASS' if accuracy['accuracy'] >= 0.85 else 'FAIL'}")
print(f"Latency <= 5000ms: {'PASS' if lat['mean_ms'] <= 5000 else 'FAIL'}")
print(f"Cost <= $0.10:     {'PASS' if cost['mean_cost_usd'] <= 0.10 else 'FAIL'}")

Exercise 2: Build a BenchmarkSuite with thresholds (Easy)

Create a BenchmarkSuite with 15 tasks across 3 categories (simple, multi-tool, edge-case), with expected_tools and thresholds. Save it as JSON.

tasks = [
    {"id": "S-01", "input": "What's 25*4?", "expected": {"answer_contains": ["100"]}},
    {"id": "S-02", "input": "Capital of France", "expected": {"answer_contains": ["Paris"]}},
    {"id": "S-03", "input": "Look up Python news", "expected": {"min_length": 50}},
    {"id": "S-04", "input": "Square root of 144", "expected": {"answer_contains": ["12"]}},
    {"id": "S-05", "input": "Who created Linux?", "expected": {"answer_contains": ["Linus"]}},
    {"id": "M-01", "input": "Population of Brazil, calculate 10%", "expected": {"answer_contains": ["Brazil"]}},
    {"id": "M-02", "input": "GDP of Japan and Mexico, the difference", "expected": {"min_length": 100}},
    {"id": "M-03", "input": "Gold price, how much with $5000", "expected": {"answer_contains": ["gold"]}},
    {"id": "M-04", "input": "Tesla vs Toyota sales, the ratio", "expected": {"min_length": 100}},
    {"id": "M-05", "input": "Earth-Moon distance in miles", "expected": {"answer_contains": ["Moon"]}},
    {"id": "E-01", "input": "", "expected": {"min_length": 5}},
    {"id": "E-02", "input": "Calculate 0/0", "expected": {"min_length": 5}},
    {"id": "E-03", "input": "Send an email to test@test.com", "expected": {"answer_contains": ["I can't"]}},
    {"id": "E-04", "input": "Predict BTC tomorrow", "expected": {"min_length": 20}},
    {"id": "E-05", "input": "Search for asdfjkl12345", "expected": {"min_length": 10}},
]
expected = {"S-01": ["calculator"], "S-02": [], "S-03": ["web_search"], "S-04": ["calculator"],
    "S-05": ["web_search"], "M-01": ["web_search", "calculator"], "M-02": ["web_search", "calculator"],
    "M-03": ["web_search", "calculator"], "M-04": ["web_search", "calculator"],
    "M-05": ["web_search", "calculator"], "E-01": [], "E-02": ["calculator"], "E-03": [],
    "E-04": [], "E-05": ["web_search"]}
suite = BenchmarkSuite("research-agent-full-v1", "1.0.0", tasks, expected,
    {"task_completion_rate": 0.80, "tool_call_accuracy": 0.85,
     "reasoning_quality": 0.70, "mean_latency_ms": 5000, "mean_cost_usd": 0.10})
suite.save("benchmarks/research-agent-full-v1.json")

Exercise 3: Trend tracking with regression detection (Medium)

Simulate 5 benchmark runs with the last one showing degradation. Verify that detect_regression catches it.

from datetime import datetime, timedelta

history = BenchmarkHistory(suite_name="trend-test")
runs = [
    {"task_completion_rate": 0.85, "tool_call_accuracy": 0.90, "reasoning_quality": 0.78,
     "mean_latency_ms": 2800, "mean_cost_usd": 0.0015},
    {"task_completion_rate": 0.87, "tool_call_accuracy": 0.92, "reasoning_quality": 0.80,
     "mean_latency_ms": 2700, "mean_cost_usd": 0.0014},
    {"task_completion_rate": 0.88, "tool_call_accuracy": 0.93, "reasoning_quality": 0.82,
     "mean_latency_ms": 2600, "mean_cost_usd": 0.0013},
    {"task_completion_rate": 0.86, "tool_call_accuracy": 0.91, "reasoning_quality": 0.79,
     "mean_latency_ms": 2750, "mean_cost_usd": 0.0014},
    {"task_completion_rate": 0.72, "tool_call_accuracy": 0.91, "reasoning_quality": 0.79,
     "mean_latency_ms": 4200, "mean_cost_usd": 0.0020},
]
base = datetime.now() - timedelta(days=5)
for i, run in enumerate(runs):
    history.entries.append({"timestamp": (base + timedelta(days=i)).isoformat(),
                            "model": "gpt-4o-mini", **run})

for metric in ["task_completion_rate", "tool_call_accuracy", "mean_latency_ms", "mean_cost_usd"]:
    reg = history.detect_regression(metric)
    trend = [round(v, 3) for v in history.trend(metric, 3)]
    print(f"{metric:<25} trend={trend}  {'REGRESSION' if reg else 'OK'}")

# task_completion_rate: 0.88 → 0.72 = REGRESSION (-18%)
# mean_latency_ms: 2600 → 4200 = REGRESSION (+62%)
# tool_call_accuracy: stable = OK

Exercise 4: Model comparison with a recommendation (Hard)

Implement recommend_model that takes reports from multiple models and a profile ("quality_first", "cost_first", "balanced"), and returns the recommended model with a justification based on weighted scores.

@dataclass
class ModelRecommendation:
    profile: str
    recommended_model: str
    justification: str
    scores: dict[str, float]

def recommend_model(reports: dict[str, BenchmarkReport], profile: str = "balanced") -> ModelRecommendation:
    weights = {
        "quality_first":  {"completion": 0.35, "accuracy": 0.25, "latency": 0.15, "cost": 0.25},
        "cost_first":     {"completion": 0.15, "accuracy": 0.15, "latency": 0.20, "cost": 0.50},
        "balanced":       {"completion": 0.25, "accuracy": 0.25, "latency": 0.20, "cost": 0.30},
    }
    w = weights.get(profile, weights["balanced"])
    model_scores = {}
    for name, r in reports.items():
        comp = r.task_completion.get("rate", 0)
        acc = r.tool_accuracy.get("accuracy", 0)
        lat_norm = max(0, 1.0 - (r.latency.get("mean_ms", 5000) / 10000))
        cost_norm = max(0, 1.0 - (r.cost.get("mean_cost_usd", 0.01) / 0.05))
        model_scores[name] = round(
            w["completion"] * comp + w["accuracy"] * acc
            + w["latency"] * lat_norm + w["cost"] * cost_norm, 4)

    best = max(model_scores, key=model_scores.get)
    br = reports[best]
    return ModelRecommendation(profile, best,
        f"{best} (score {model_scores[best]:.3f}): "
        f"completion {br.task_completion.get('rate',0):.0%}, "
        f"cost ${br.cost.get('mean_cost_usd',0):.4f}/task",
        model_scores)

# quality_first → gpt-4o   |  cost_first → gpt-4o-mini  |  balanced → gpt-4.1-mini

Exercise 5: A benchmarking pipeline for CI (Hard)

Build a BenchmarkPipeline that: (a) loads the suite from JSON, (b) runs N runs for confidence, (c) compares against the history, (d) generates a JSON artifact with pass/fail, (e) updates the history only if it passes.

import sys

class BenchmarkPipeline:
    def __init__(self, suite_path: str, history_path: str):
        self.suite = BenchmarkSuite.load(suite_path)
        self.history_path = history_path
        try:
            self.history = BenchmarkHistory.load(history_path)
        except FileNotFoundError:
            self.history = BenchmarkHistory(suite_name=self.suite.name)

    def run(self, agent, model_name: str, n_runs: int = 3,
            output_path: str = "benchmark-artifact.json") -> dict:
        config = BenchmarkConfig(name=self.suite.name, model=model_name, dataset_path="")
        reports = [run_benchmark(agent, self.suite.tasks, config, self.suite.expected_tools)
                   for _ in range(n_runs)]

        avg = {k: sum(getattr_metric(r, k) for r in reports) / n_runs
               for k in ["task_completion_rate", "tool_call_accuracy", "mean_latency_ms", "mean_cost_usd"]}

        thresh = {}
        for m, t in self.suite.thresholds.items():
            actual = avg.get(m, 0)
            passed = actual <= t if m in ("mean_latency_ms", "mean_cost_usd") else actual >= t
            thresh[m] = {"actual": round(actual, 6), "threshold": t, "passed": passed}

        overall = all(r["passed"] for r in thresh.values())
        if overall:
            self.history.entries.append({"timestamp": datetime.now().isoformat(),
                "model": model_name, **{k: round(v, 6) for k, v in avg.items()}})
            self.history.save(self.history_path)

        artifact = {"timestamp": datetime.now().isoformat(), "suite": self.suite.name,
                     "model": model_name, "metrics": avg, "thresholds": thresh,
                     "overall_pass": overall, "history_updated": overall}
        with open(output_path, "w") as f:
            json.dump(artifact, f, indent=2)
        print(f"Benchmark {'PASS' if overall else 'FAIL'}")
        return artifact

def getattr_metric(r, key):
    mapping = {"task_completion_rate": r.task_completion.get("rate", 0),
               "tool_call_accuracy": r.tool_accuracy.get("accuracy", 0),
               "mean_latency_ms": r.latency.get("mean_ms", 0),
               "mean_cost_usd": r.cost.get("mean_cost_usd", 0)}
    return mapping.get(key, 0)

# pipeline = BenchmarkPipeline("benchmarks/suite.json", "benchmarks/history.json")
# artifact = pipeline.run(agent, "gpt-4o-mini", n_runs=3)
# sys.exit(0 if artifact["overall_pass"] else 1)

Summary

In this capsule you defined your agent's quantitative language — the numbers that replace "it seems to work":

• Five core metrics: Task completion rate, tool call accuracy, reasoning quality, latency, cost per task. The first three measure quality. The last two measure efficiency. Together they define the complete performance profile.
• Standardized benchmarks: A versioned BenchmarkSuite with tasks, expected tools, and thresholds. You run the same benchmark every time — what changes is the agent.
• Trend tracking: BenchmarkHistory records every run. detect_regression compares the latest run against the previous average. The deltas tell you direction, the alerts tell you when to act.
• Multi-agent metrics: Per-agent metrics reveal where the problem is. Coordination metrics (handoff success, supervisor efficiency) reveal integration problems between agents.
• Data-driven model comparison: The same benchmark, different models. ModelRecommendation automates the decision by profile (quality_first, cost_first, balanced). Without benchmarks, "GPT-4o is better" is an opinion. With benchmarks, it's data.

Next capsule: Project: Testing and Evaluation — you integrate the whole module (unit tests, trajectory evaluation, LangSmith, golden datasets, regression testing, and benchmarks) into Research Agent v6.

Additional Resources

1. LangSmith — Evaluation Metrics — Official documentation on evaluation metrics in LangSmith for agents
2. Evaluating LLM Applications — LangChain — LangChain's guide to evaluation chains and metrics for LLM applications
3. OpenAI — Pricing — Per-model pricing reference for cost-per-task calculations
4. Building Effective Agents — Anthropic — Anthropic's perspective on measuring agent quality
5. How to Evaluate AI Agents — Hamel Husain — Benchmarking and metrics for agents in production