AI Agent Evaluation: How to Benchmark MultiStep Agentic Systems in Production

import subprocess
import json
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class SWEBenchInstance:
    instance_id: str
    repo: str
    base_commit: str
    problem_statement: str
    hints_text: str
    test_patch: str           # Tests that must pass after fix
    fail_to_pass: list[str]   # Tests that fail before and must pass after
    pass_to_pass: list[str]   # Tests that pass before and must still pass after
 
@dataclass
class SWEBenchResult:
    instance_id: str
    resolved: bool
    patch_applied: bool
    tests_passed: list[str]
    tests_failed: list[str]
    api_calls: int
    cost_usd: float
    time_seconds: float
    patch: Optional[str] = None
 
class SWEBenchEvaluator:
    """Local SWE-bench evaluation runner."""
 
    def __init__(self, workspace_dir: str = "/tmp/swebench"):
        self.workspace = Path(workspace_dir)
        self.workspace.mkdir(exist_ok=True)
 
    def evaluate_patch(self, instance: SWEBenchInstance, patch: str) -> SWEBenchResult:
        """Apply a patch and run the test suite."""
        repo_dir = self.workspace / instance.instance_id
 
        # Clone repo at base commit
        subprocess.run([
            "git", "clone", f"https://github.com/{instance.repo}",
            str(repo_dir)
        ], check=True, capture_output=True)
        subprocess.run(
            ["git", "checkout", instance.base_commit],
            cwd=repo_dir, check=True, capture_output=True
        )
 
        # Apply test patch (adds the validation tests)
        test_patch_path = repo_dir / "test.patch"
        test_patch_path.write_text(instance.test_patch)
        test_result = subprocess.run(
            ["git", "apply", str(test_patch_path)],
            cwd=repo_dir, capture_output=True
        )
 
        # Apply agent's patch
        agent_patch_path = repo_dir / "agent.patch"
        agent_patch_path.write_text(patch)
        apply_result = subprocess.run(
            ["git", "apply", str(agent_patch_path)],
            cwd=repo_dir, capture_output=True
        )
        patch_applied = apply_result.returncode == 0
 
        if not patch_applied:
            return SWEBenchResult(
                instance_id=instance.instance_id,
                resolved=False,
                patch_applied=False,
                tests_passed=[], tests_failed=instance.fail_to_pass,
                api_calls=0, cost_usd=0, time_seconds=0,
                patch=patch,
            )
 
        # Run tests
        test_output = subprocess.run(
            ["python", "-m", "pytest", "--no-header", "-rN", "--tb=no"] +
            instance.fail_to_pass + instance.pass_to_pass,
            cwd=repo_dir, capture_output=True, text=True, timeout=120
        )
 
        passed, failed = self._parse_pytest_output(test_output.stdout, instance)
        resolved = (
            all(t in passed for t in instance.fail_to_pass) and
            all(t in passed for t in instance.pass_to_pass)
        )
 
        return SWEBenchResult(
            instance_id=instance.instance_id,
            resolved=resolved,
            patch_applied=True,
            tests_passed=passed,
            tests_failed=failed,
            api_calls=0, cost_usd=0, time_seconds=0,
            patch=patch,
        )
 
    def _parse_pytest_output(self, output: str,
                              instance: SWEBenchInstance) -> tuple[list, list]:
        passed = []
        failed = []
        for line in output.split("\n"):
            if " PASSED" in line:
                test_name = line.split("::")[0].strip()
                passed.append(test_name)
            elif " FAILED" in line:
                test_name = line.split("::")[0].strip()
                failed.append(test_name)
        return passed, failed

pass^k = p^k

import asyncio
import random
from typing import Callable, Awaitable
from dataclasses import dataclass
import numpy as np
 
@dataclass
class TauBenchTask:
    task_id: str
    initial_state: dict    # Database/environment state before task
    user_goal: str         # What the user wants to accomplish
    goal_state: dict       # Expected database/environment state after task
    tool_definitions: list[dict]
 
@dataclass
class TauBenchResult:
    task_id: str
    trial_id: int
    success: bool
    final_state: dict
    n_tool_calls: int
    tool_call_log: list[dict]
 
class TauBenchEvaluator:
    """
    Evaluates agent reliability using the pass^k metric.
    """
 
    def __init__(self, agent_fn: Callable[[TauBenchTask], Awaitable[dict]],
                 state_comparator: Callable[[dict, dict], bool]):
        """
        agent_fn: async function that takes a task and returns final environment state
        state_comparator: function that checks if final_state matches goal_state
        """
        self.agent = agent_fn
        self.compare = state_comparator
 
    async def run_trials(self, task: TauBenchTask, k: int = 8) -> dict:
        """Run k independent trials of the same task and compute pass^k."""
        results = []
 
        for trial in range(k):
            # Fresh environment state for each trial
            result = await self.agent(task)
            success = self.compare(result["final_state"], task.goal_state)
            results.append(TauBenchResult(
                task_id=task.task_id,
                trial_id=trial,
                success=success,
                final_state=result["final_state"],
                n_tool_calls=result.get("n_tool_calls", 0),
                tool_call_log=result.get("tool_call_log", []),
            ))
 
        successes = [r.success for r in results]
        single_trial_rate = sum(successes) / k
 
        # pass^k: all trials must succeed
        all_pass = all(successes)
 
        # Estimated pass^k for various k values
        estimated_pass_k = {k: single_trial_rate ** k for k in [1, 2, 4, 8, 16]}
 
        return {
            "task_id": task.task_id,
            "trials": k,
            "single_trial_success_rate": single_trial_rate,
            "pass_k_all": all_pass,
            "estimated_pass_k": estimated_pass_k,
            "avg_tool_calls": np.mean([r.n_tool_calls for r in results]),
            "results": results,
        }
 
    async def evaluate_suite(self, tasks: list[TauBenchTask], k: int = 8) -> dict:
        """Evaluate the full task suite with pass^k metric."""
        task_results = await asyncio.gather(
            *[self.run_trials(task, k) for task in tasks]
        )
 
        # Aggregate metrics
        single_trial_rates = [r["single_trial_success_rate"] for r in task_results]
        pass_k_rates = [r["pass_k_all"] for r in task_results]
 
        return {
            "n_tasks": len(tasks),
            "k": k,
            "mean_single_trial_accuracy": np.mean(single_trial_rates),
            "pass_k_accuracy": np.mean(pass_k_rates),  # Fraction of tasks that always succeed
            "reliability_gap": np.mean(single_trial_rates) - np.mean(pass_k_rates),
            "per_task": task_results,
        }
 
def compare_database_states(final: dict, goal: dict,
                              tolerance: float = 0.01) -> bool:
    """
    Compare final database state to goal state.
    Handles numeric fields with tolerance and missing field detection.
    """
    for key, expected in goal.items():
        actual = final.get(key)
        if actual is None:
            return False  # Required field missing
 
        if isinstance(expected, (int, float)):
            if abs(actual - expected) / max(abs(expected), 1) > tolerance:
                return False
        elif isinstance(expected, list):
            if set(actual) != set(expected):  # Order-insensitive
                return False
        else:
            if actual != expected:
                return False
 
    return True

from dataclasses import dataclass
from typing import Optional, Any
 
@dataclass
class GAIAQuestion:
    question_id: str
    level: int               # 1, 2, or 3
    question: str
    expected_answer: str
    annotator_metadata: dict  # Tool types needed, number of steps, etc.
    file_attachment: Optional[str] = None  # Path to attached file if any
 
class GAIAEvaluator:
    """GAIA evaluation with exact and normalized matching."""
 
    def evaluate_answer(self, predicted: str, expected: str,
                         level: int) -> dict:
        """
        GAIA uses exact match after normalization.
        Numbers, dates, and lists require special normalization.
        """
        pred_norm = self._normalize(predicted)
        exp_norm = self._normalize(expected)
 
        exact_match = pred_norm == exp_norm
        partial_match = exp_norm in pred_norm or pred_norm in exp_norm
 
        return {
            "exact_match": exact_match,
            "partial_match": partial_match,
            "normalized_prediction": pred_norm,
            "normalized_expected": exp_norm,
        }
 
    def _normalize(self, answer: str) -> str:
        """Normalize an answer for comparison."""
        import re
        # Remove trailing punctuation
        answer = answer.strip().rstrip(".")
 
        # Normalize numbers: "1,234.5" → "1234.5"
        answer = re.sub(r"(\d),(\d)", r"\1\2", answer)
 
        # Lowercase
        answer = answer.lower()
 
        # Normalize units: "3 km" → "3 km", "3km" → "3 km"
        answer = re.sub(r"(\d)([a-zA-Z])", r"\1 \2", answer)
 
        return answer.strip()
 
    def compute_metrics(self, results: list[dict]) -> dict:
        """Compute GAIA metrics by level and overall."""
        by_level = {1: [], 2: [], 3: []}
 
        for r in results:
            by_level[r["level"]].append(r["exact_match"])
 
        return {
            "overall": sum(r["exact_match"] for r in results) / len(results),
            "level_1": sum(by_level[1]) / max(len(by_level[1]), 1),
            "level_2": sum(by_level[2]) / max(len(by_level[2]), 1),
            "level_3": sum(by_level[3]) / max(len(by_level[3]), 1),
        }

from anthropic import AsyncAnthropic
import json
 
client = AsyncAnthropic()
 
EVALUATION_RUBRIC = """You are evaluating an AI agent's response to a user task.
Score each dimension from 1-5 using the criteria below. Return ONLY JSON.
 
Dimensions:
1. task_completion (1-5): Did the agent fully complete what was asked?
   1=Did not attempt, 3=Partial completion, 5=Fully complete with no gaps
 
2. accuracy (1-5): Are the facts, data, and claims in the response correct?
   1=Multiple errors, 3=Mostly correct with minor errors, 5=Fully accurate
 
3. tool_efficiency (1-5): Did the agent use tools efficiently?
   1=Unnecessary redundant calls or wrong tools, 3=Reasonable use, 5=Optimal tool use
 
4. format_quality (1-5): Is the output well-formatted and readable?
   1=Hard to read, 3=Adequate, 5=Well-structured and clear
 
5. safety (1-5): Did the agent avoid risky or undesired side effects?
   1=Took risky actions, 3=Cautious but unnecessary, 5=Appropriate caution
 
Return format:
{
  "task_completion": <1-5>,
  "accuracy": <1-5>,
  "tool_efficiency": <1-5>,
  "format_quality": <1-5>,
  "safety": <1-5>,
  "reasoning": "<brief explanation of scores>"
}"""
 
async def evaluate_agent_response(
    task: str,
    agent_response: str,
    tool_call_log: list[dict],
    model: str = "claude-opus-4-6",
) -> dict:
    """
    Evaluate an agent's response using LLM-as-judge.
    Returns structured scores across dimensions.
    """
    tool_summary = "\n".join([
        f"- {call['tool']}: {str(call.get('args', {}))[:100]}"
        for call in tool_call_log[:20]
    ])
 
    judgment_prompt = f"""Task given to agent:
{task}
 
Agent's tool calls:
{tool_summary}
 
Agent's final response:
{agent_response}
 
{EVALUATION_RUBRIC}"""
 
    response = await client.messages.create(
        model=model,
        max_tokens=500,
        messages=[{"role": "user", "content": judgment_prompt}]
    )
 
    try:
        scores = json.loads(response.content[0].text)
        scores["composite"] = sum([
            scores.get("task_completion", 3) * 0.35,    # Highest weight
            scores.get("accuracy", 3) * 0.30,
            scores.get("tool_efficiency", 3) * 0.15,
            scores.get("format_quality", 3) * 0.10,
            scores.get("safety", 3) * 0.10,
        ])
        return scores
    except json.JSONDecodeError:
        return {"error": "Failed to parse judge response", "raw": response.content[0].text}
 
async def evaluate_with_position_bias_control(
    task: str,
    response_a: str,
    response_b: str,
    tool_log_a: list[dict],
    tool_log_b: list[dict],
) -> dict:
    """
    Pairwise evaluation with position bias control.
    Runs A vs B and B vs A, averages results to reduce position bias.
    """
    scores_ab = await evaluate_agent_response(task, response_a, tool_log_a)
    scores_ba = await evaluate_agent_response(task, response_b, tool_log_b)
 
    # Account for position effect (judge may favor the first response)
    composite_a = (scores_ab.get("composite", 3) + (6 - scores_ba.get("composite", 3))) / 2
    composite_b = ((6 - scores_ab.get("composite", 3)) + scores_ba.get("composite", 3)) / 2
 
    return {
        "response_a_score": composite_a,
        "response_b_score": composite_b,
        "winner": "A" if composite_a > composite_b else "B" if composite_b > composite_a else "tie",
        "margin": abs(composite_a - composite_b),
    }

@dataclass
class AgentTrajectory:
    task_id: str
    steps: list[dict]      # [{tool, args, result, timestamp, success}]
    final_state: dict
    success: bool
    total_tokens: int
    total_cost_usd: float
    wall_time_seconds: float
 
def evaluate_trajectory(trajectory: AgentTrajectory,
                         optimal_n_steps: int) -> dict:
    """
    Evaluate the quality of an agent trajectory.
    """
    n_steps = len(trajectory.steps)
    failed_steps = [s for s in trajectory.steps if not s.get("success", True)]
    redundant_steps = _find_redundant_steps(trajectory.steps)
    risky_steps = _find_risky_steps(trajectory.steps)
 
    # Efficiency ratio: optimal steps / actual steps
    efficiency = min(optimal_n_steps / max(n_steps, 1), 1.0)
 
    # Error recovery rate: agent recovers from failures
    failure_followed_by_success = 0
    for i, step in enumerate(failed_steps):
        if any(s.get("success") for s in trajectory.steps[i+1:i+4]):
            failure_followed_by_success += 1
 
    recovery_rate = (failure_followed_by_success / max(len(failed_steps), 1)
                     if failed_steps else 1.0)
 
    return {
        "n_steps": n_steps,
        "optimal_n_steps": optimal_n_steps,
        "efficiency": efficiency,
        "n_failed_steps": len(failed_steps),
        "n_redundant_steps": len(redundant_steps),
        "n_risky_steps": len(risky_steps),
        "error_recovery_rate": recovery_rate,
        "cost_per_success": (trajectory.total_cost_usd if trajectory.success
                              else float("inf")),
        "tokens_per_step": trajectory.total_tokens / max(n_steps, 1),
    }
 
def _find_redundant_steps(steps: list[dict]) -> list[dict]:
    """Find steps that repeat the same tool call with the same arguments."""
    seen = set()
    redundant = []
    for step in steps:
        signature = f"{step.get('tool')}:{str(sorted(step.get('args', {}).items()))}"
        if signature in seen:
            redundant.append(step)
        seen.add(signature)
    return redundant
 
def _find_risky_steps(steps: list[dict]) -> list[dict]:
    """Flag tool calls that are potentially risky/irreversible."""
    risky_tools = {"delete_file", "send_email", "execute_code", "database_write",
                   "deploy", "payment", "api_post"}
    return [s for s in steps if s.get("tool") in risky_tools]

class ProductionEvalSuite:
    """
    Production eval suite for a specific agent deployment.
    """
 
    def __init__(self, agent, suite_name: str):
        self.agent = agent
        self.suite_name = suite_name
        self.test_cases: list[dict] = []
        self.evaluators: dict[str, callable] = {}
 
    def add_test_case(self, task: str, initial_state: dict,
                       expected_outcome: dict, category: str,
                       verifier: callable):
        """Add a test case with its verification function."""
        self.test_cases.append({
            "task": task,
            "initial_state": initial_state,
            "expected_outcome": expected_outcome,
            "category": category,
            "verifier": verifier,
        })
 
    async def run(self, k_trials: int = 3) -> dict:
        """Run the eval suite with pass^k reliability measurement."""
        results_by_category = {}
 
        for case in self.test_cases:
            category = case["category"]
            if category not in results_by_category:
                results_by_category[category] = {"trials": [], "pass_k": []}
 
            # Run k trials for reliability measurement
            trial_results = []
            for _ in range(k_trials):
                outcome = await self.agent.run(
                    case["task"], initial_state=case["initial_state"]
                )
                success = case["verifier"](outcome, case["expected_outcome"])
                trial_results.append(success)
 
            single_trial = sum(trial_results) / k_trials
            pass_k = all(trial_results)
 
            results_by_category[category]["trials"].append(single_trial)
            results_by_category[category]["pass_k"].append(pass_k)
 
        # Aggregate by category
        summary = {}
        for category, data in results_by_category.items():
            summary[category] = {
                "n_cases": len(data["trials"]),
                "mean_accuracy": sum(data["trials"]) / len(data["trials"]),
                "pass_k_rate": sum(data["pass_k"]) / len(data["pass_k"]),
                "reliability_gap": (sum(data["trials"]) / len(data["trials"])
                                    - sum(data["pass_k"]) / len(data["pass_k"])),
            }
 
        return {
            "suite": self.suite_name,
            "overall_accuracy": sum(
                sum(d["trials"]) for d in results_by_category.values()
            ) / max(sum(len(d["trials"]) for d in results_by_category.values()), 1),
            "by_category": summary,
        }

1. Run eval suite → identify lowest-scoring categories
2. Sample 10-20 failures from lowest-scoring category
3. Identify the specific step where the agent goes wrong (trajectory analysis)
4. Root cause: prompt issue, tool design issue, context management issue, or model limitation?
5. Fix the root cause
6. Re-run eval → confirm improvement without regression in other categories
7. Add the fixed failure cases to the eval suite as regression tests