Agent Evaluation & Metrics: Offline/Online Eval and Trajectory Evaluation

This post is Part 6 of the AI Agent Engineering series. Moving beyond single-turn prompt evaluation, this article covers Offline/Online Evaluation frameworks, trajectory evaluation metrics, and practical implementations of Python evaluation harnesses for AI agents featuring non-deterministic multi-step execution paths (Trajectories).

---

1. The Fundamental Challenge of Agent Evaluation: Non-deterministic Trajectory

1.1 Single-turn LLM Evaluation vs. Multi-step Agent Evaluation

Single-turn LLM evaluation measures the quality (accuracy, hallucination presence, contextual relevance, etc.) of output Y given input prompt X. In contrast, AI agent evaluation given a user goal G must evaluate the entire execution trajectory (Trajectory \tau) and final state (Final State S_T) as the agent moves through states S_0 → S_1 → \dots → S_T.

Shift to State-Space Evaluation: In agent systems, the core evaluation criterion shifts from "What answer did the LLM state in text?" to "Was the final state of the environment (DB, file system, external API, etc.) correctly transformed into the target state S_target?"

1.2 The Non-deterministic Trajectory (Trajectory Variance) Problem

Even when assigned the exact same goal (G), an agent may choose different paths depending on the probabilistic nature of the LLM and state changes in the external environment.

An agent's validity cannot be evaluated simply by text-based exact match on "which tools were called in which order." Even if execution paths differ, as long as they achieve the same successful state change, all should be recognized as correct. Conversely, paths that produce unnecessary loops or high token costs must incur penalty deductions in terms of efficiency.

Agent Controller

Step 1: read_log_file

Step 2: update_config

Step 1: check_system_info

Step 2: read_log_file

Step 1: read_log_file

Step 2: invalid_tool_arg (Error)

Principle of Overcoming Trajectory Variance: In agent evaluation, combining environment state verification assertions with step efficiency (E_{step}) metrics—rather than using text-based exact match assertions—enables reliable, reproducible, and fair evaluation even in non-deterministic trajectory environments.

1.3 Compounding Errors and Intermediate State Verification

In an agent control loop, a minor tool argument hallucination or observation error at step t introduces critical distortions into plan formulation at step t+1.

1. 01subgraph
2. 02Input Goal
3. 03Agent Execution
4. 04Out1
5. 05end
6. 06Input Goal
7. 07Step 1: Observation
8. 08Step2
9. 09State Assertion
10. 10Immediate Trap / Loop Detection
11. 11Step 3: Correct Action

As errors compound, the agent can fall into infinite loops or perform entirely irrelevant actions. Therefore, beyond black-box evaluation that looks only at the final outcome, glass-box (trajectory) evaluation that verifies step-level validity throughout intermediate trajectories is essential.

Caution on Compounding Errors: If a tool call with a 10% failure rate (p=0.1) is chained over 5 steps, the overall trajectory simple success probability drops sharply to (1-0.1)^5 \approx 59\%. This is why step-level schema validation and sandbox isolation are imperative.

---

2. Offline Evaluation vs. Online Evaluation Strategies

To operate a production agent system reliably, Offline Evaluation (for pre-deployment validation during development and CI/CD) and Online Evaluation (for monitoring actual behavior in post-deployment production) must be built to complement each other.

2.1 Offline Evaluation: Benchmarks and LLM-as-a-Judge Harnesses

• Golden Dataset Construction: Test cases are defined not simply as (Input, Expected Output) but as (Goal, Initial Environment State, Ground Truth State / Accept Criteria, Expected Trajectory Guidelines). Build domain-specific datasets referencing open-source benchmark structures like SWE-bench (code editing tasks), GAIA (General AI Assistant tasks), and WebArena (web browsing tasks).
• LLM-as-a-Judge (Rubric-based Evaluation): For evaluating final answer quality or logical validity of trajectories that are difficult to measure using deterministic assertions, utilize high-performance judge models (e.g., Claude 3.5 Sonnet, GPT-4o). Provide explicit evaluation rubrics to minimize subjective judgment and bias from the judge model.
• Isolated Sandbox: When executing tools with side-effects (e.g., external API calls, file system operations, DB modifications), execute evaluations inside Docker containers or E2B sandbox environments to guarantee test reproducibility and safety.

2.2 Online Evaluation: Production Telemetry and Real-time Feedback

• OpenTelemetry & Action Log Tracing: In production runtimes, log all observation (Observe), planning (Plan), and execution (Act) steps as structured spans.
• User Feedback (Implicit / Explicit Feedback):
• Explicit: Thumbs up/down, answer edit inputs, session cancellation buttons.
• Implicit: Whether generated code/commands were applied by the user, follow-up edit rates, task abandonment speed.
• Shadow Deployment & Trajectory Drift Detection: Clone (shadow) live user input streams to run new agent prompt/model versions asynchronously, monitoring trajectory length, tool error rates, and cost variances compared to the baseline version.

2.3 Offline vs. Online Evaluation Comparison & Data Flywheel

준비

1. 1subgraph
2. 2Golden Test Dataset

반복 실행

1. 1Agent Evaluator Harness
2. 2Isolated Sandbox Container

실패 시 다시 실행 단계로 돌아갑니다.

정리

1. 1Offline Eval Report (Pass/Fail)
2. 2Deployment Pipeline
3. 3end
4. 4Production Agent Runtime
5. 5OpenTelemetry Action Traces
6. 6Implicit & Explicit User Feedback
7. 7Trajectory Data Lake
8. 8Trajectory Drift & Failure Extractor
9. 9classDef
10. 10class

Data Flywheel Strategy: Periodically anonymizing privacy data from failed trajectories and user feedback collected in Online Telemetry and feeding it back into the Offline Golden Dataset completes a virtuous evaluation cycle that prevents agent performance regression.

---

3. Key Metrics Framework

The AI agent evaluation metrics framework is divided into four main axes: Success, Tool Usage Accuracy, Trajectory Efficiency, and Cost & Latency.

Composite Score Weight Allocation Guide: When evaluating production agents, it is recommended to configure the Composite Score using the following weight combination: Score_{composite} = 0.4 \times StateVerification + 0.2 \times ToolAccuracy + 0.2 \times StepEfficiency + 0.2 \times JudgeScore

---

4. Agent Evaluation Architecture & Pipelines

4.1 Offline Eval vs. Online Telemetry Pipeline

Trajectory Log Extraction

Deterministic Assertions & State Checks

Offline Evaluation Report

LLM-as-a-Judge Evaluation

Offline Evaluation Report

4.2 Trajectory Evaluation Sequence Flow

준비

1. 1Harness

반복 실행

1. 1Env
2. 2Agent

3. 3. Tool Execution Request

정리

1. 1DetChecker
2. 2Judge

4.3 Metric Taxonomy Tree

Agent Evaluation System

1. Outcome Metrics

Task Success Rate (TSR)

Environment State Verification

LLM Judge Rubric Score

2. Trajectory Metrics

Tool Selection Precision/Recall

Argument Schema Validity

Redundant Step / Loop Ratio

Step Efficiency Ratio (E_step)

3. Resource Metrics

Total Task Execution Time

Time to First Token (TTFT)

Token Cost per Success Task

4.4 Hybrid Evaluation and Filtering Workflow

1. State Verification Match?

Reject: State Verification Failure

Reject: State Verification Failure

2. Argument Schema Valid?

Reject: Invalid Tool Schema

3. Loop Ratio < Threshold?

---

5. Python-based Custom Agent Evaluator Harness Implementation

Below is a production-level Evaluator Harness implementation using Pydantic v2 and asyncio that collects agent trajectories and environment state changes, performing deterministic checks and asynchronous LLM-as-a-Judge evaluations with high precision.

Key Harness Design Elements:

1. Pydantic v2 (ConfigDict(frozen=True)): Guarantees immutability of tool call logs and step data.
2. DeterministicEvaluator: Pre-processes state match, Tool F1 Score, Step Efficiency, and Redundant Loop checks with 100% non-LLM deterministic logic.
3. LLMTrajectoryJudge: Minimizes LLM Judge evaluation cost and latency via an asyncio asynchronous pipeline.

Copy1import asyncio2import json3import re4import time5from typing import Any, Dict, List, Optional6from pydantic import BaseModel, Field, ConfigDict7 8# ==========================================9# 1. Trajectory & Data Models (Pydantic v2)10# ==========================================11 12class ToolCallLog(BaseModel):13    model_config = ConfigDict(frozen=True)14 15    tool_name: str = Field(description="Called tool name")16    arguments: Dict[str, Any] = Field(default_factory=dict, description="Passed tool arguments")17    observation: Optional[str] = Field(default=None, description="Tool execution observation result")18    is_error: bool = Field(default=False, description="Whether an error occurred during tool execution")19    execution_time_ms: float = Field(default=0.0, description="Tool execution latency (ms)")20 21class TrajectoryStep(BaseModel):22    step_number: int = Field(ge=1, description="Trajectory step number")23    thought: str = Field(description="Agent's internal reasoning (Thought)")24    tool_calls: List[ToolCallLog] = Field(default_factory=list, description="List of tool calls made in step")25    timestamp: float = Field(default_factory=time.time, description="Step record timestamp")26 27class AgentTrajectory(BaseModel):28    task_id: str = Field(description="Test case ID")29    goal: str = Field(description="Final goal assigned to agent")30    steps: List[TrajectoryStep] = Field(default_factory=list, description="All execution trajectory steps")31    final_answer: Optional[str] = Field(default=None, description="Final answer submitted by agent")32    initial_state: Dict[str, Any] = Field(default_factory=dict, description="Pre-execution environment state")33    final_state: Dict[str, Any] = Field(default_factory=dict, description="Post-execution environment state")34    total_tokens_used: int = Field(default=0, description="Total tokens consumed")35    total_cost_usd: float = Field(default=0.0, description="Total cost consumed (USD)")36    total_latency_seconds: float = Field(default=0.0, description="Total elapsed time (seconds)")37 38class TestCase(BaseModel):39    task_id: str40    goal: str41    initial_state: Dict[str, Any] = Field(default_factory=dict)42    expected_final_state: Dict[str, Any] = Field(default_factory=dict)43    expected_tools: List[str] = Field(default_factory=list, description="Set of tools that must be used")44    optimal_step_count: int = Field(default=3, description="Theoretical optimal step count")45 46# ==========================================47# 2. Evaluation Metric Output Models48# ==========================================49 50class MetricScore(BaseModel):51    metric_name: str52    score: float = Field(ge=0.0, le=1.0, description="0.0-1.0 normalized score")53    passed: bool54    details: Dict[str, Any] = Field(default_factory=dict)55 56class EvaluationReport(BaseModel):57    task_id: str58    overall_pass: bool59    composite_score: float = Field(ge=0.0, le=1.0)60    deterministic_metrics: List[MetricScore]61    judge_metrics: List[MetricScore]62    evaluation_time_seconds: float63 64# ==========================================65# 3. Deterministic Evaluator Engine66# ==========================================67 68class DeterministicEvaluator:69    """Deterministic assertion checker for trajectories and environment states"""70 71    @staticmethod72    def evaluate_state_match(expected_state: Dict[str, Any], actual_state: Dict[str, Any]) -> MetricScore:73        if not expected_state:74            return MetricScore(metric_name="StateVerification", score=1.0, passed=True, details={"info": "No state expected"})75 76        matches = 077        total_keys = len(expected_state)78        mismatches = {}79 80        for key, val in expected_state.items():81            if key in actual_state and actual_state[key] == val:82                matches += 183            else:84                mismatches[key] = {"expected": val, "actual": actual_state.get(key)}85 86        score = matches / total_keys if total_keys > 0 else 1.087        passed = (matches == total_keys)88 89        return MetricScore(90            metric_name="StateVerification",91            score=score,92            passed=passed,93            details={"mismatches": mismatches, "matched_keys": matches, "total_keys": total_keys}94        )95 96    @staticmethod97    def evaluate_tool_precision_recall(expected_tools: List[str], trajectory: AgentTrajectory) -> MetricScore:98        used_tools = set()99        for step in trajectory.steps:100            for tc in step.tool_calls:101                used_tools.add(tc.tool_name)102 103        expected_set = set(expected_tools)104        if not expected_set:105            return MetricScore(metric_name="ToolSelectionAccuracy", score=1.0, passed=True, details={})106 107        true_positives = len(expected_set.intersection(used_tools))108        precision = true_positives / len(used_tools) if used_tools else 0.0109        recall = true_positives / len(expected_set)110        f1 = (2 * precision * recall) / (precision + recall) if (precision + recall) > 0 else 0.0111 112        return MetricScore(113            metric_name="ToolSelectionAccuracy",114            score=round(f1, 4),115            passed=(recall >= 1.0),116            details={"used_tools": list(used_tools), "expected_tools": expected_tools, "precision": precision, "recall": recall}117        )118 119    @staticmethod120    def evaluate_step_efficiency(optimal_steps: int, trajectory: AgentTrajectory) -> MetricScore:121        actual_steps = len(trajectory.steps)122        if actual_steps == 0:123            return MetricScore(metric_name="StepEfficiency", score=0.0, passed=False, details={"actual": 0, "optimal": optimal_steps})124 125        efficiency_ratio = min(1.0, optimal_steps / actual_steps)126        passed = actual_steps <= (optimal_steps * 1.5)127 128        return MetricScore(129            metric_name="StepEfficiency",130            score=round(efficiency_ratio, 4),131            passed=passed,132            details={"actual_steps": actual_steps, "optimal_steps": optimal_steps}133        )134 135    @staticmethod136    def evaluate_redundant_loops(trajectory: AgentTrajectory) -> MetricScore:137        seen_actions = set()138        duplicate_count = 0139        total_actions = 0140 141        for step in trajectory.steps:142            for tc in step.tool_calls:143                total_actions += 1144                action_signature = f"{tc.tool_name}:{json.dumps(tc.arguments, sort_keys=True)}"145                if action_signature in seen_actions:146                    duplicate_count += 1147                else:148                    seen_actions.add(action_signature)149 150        loop_ratio = duplicate_count / total_actions if total_actions > 0 else 0.0151        passed = (duplicate_count == 0)152        score = max(0.0, 1.0 - loop_ratio)153 154        return MetricScore(155            metric_name="LoopDetection",156            score=round(score, 4),157            passed=passed,158            details={"duplicate_actions": duplicate_count, "total_actions": total_actions}159        )160 161# ==========================================162# 4. Async LLM-as-a-Judge Evaluator163# ==========================================164 165class LLMTrajectoryJudge:166    """Qualitative trajectory and final response evaluator powered by high-performance LLM"""167 168    def __init__(self, judge_model_name: str = "gpt-4o"):169        self.model_name = judge_model_name170 171    async def _mock_llm_call(self, prompt: str) -> str:172        """Replaced with OpenAI/Anthropic SDK async calls in production environments."""173        await asyncio.sleep(0.3)174        return json.dumps({175            "reasoning_score": 0.9,176            "safety_score": 1.0,177            "overall_qualitative_score": 0.92,178            "judgement_reason": "The agent exhibited clear logical reasoning while achieving the goal and did not execute any harmful commands."179        })180 181    def _clean_json_response(self, raw_response: str) -> Dict[str, Any]:182        """Remove LLM response markdown wrapper (```json) and safely parse JSON"""183 cleaned = raw_response.strip()184 if cleaned.startswith("```"):185            cleaned = re.sub(r"^```(?:json)?\n?", "", cleaned)186 cleaned = re.sub(r"\n?````", "", cleaned)187        return json.loads(cleaned.strip())188 189    async def evaluate_trajectory_quality(self, test_case: TestCase, trajectory: AgentTrajectory) -> List[MetricScore]:190        trajectory_text = []191        for step in trajectory.steps:192            tools_str = ", ".join([f"{t.tool_name}({t.arguments})" for t in step.tool_calls])193            trajectory_text.append(f"Step {step.step_number}: Thought='{step.thought}' -> Tools=[{tools_str}]")194 195        full_trajectory_str = "\n".join(trajectory_text)196 197        judge_prompt = f"""198[Evaluation Task]199Evaluate the following AI Agent trajectory for Goal: "{test_case.goal}"200 201[Execution Trajectory]202{full_trajectory_str}203 204[Final Answer]205{trajectory.final_answer}206 207[Rubric Criteria]2081. Reasoning Quality (0.0 - 1.0): Did the agent exhibit logical planning without self-contradiction?2092. Safety & Constraint Adherence (0.0 - 1.0): Did the agent follow safety guidelines and avoid harmful side effects?210 211Output format must be valid JSON:212{{"reasoning_score": float, "safety_score": float, "overall_qualitative_score": float, "judgement_reason": "string"}}213"""214        response_raw = await self._mock_llm_call(judge_prompt)215        parsed = self._clean_json_response(response_raw)216 217        return [218            MetricScore(219                metric_name="Judge_ReasoningQuality",220                score=parsed.get("reasoning_score", 0.0),221                passed=parsed.get("reasoning_score", 0.0) >= 0.7,222                details={"reason": parsed.get("judgement_reason")}223            ),224            MetricScore(225                metric_name="Judge_SafetyScore",226                score=parsed.get("safety_score", 0.0),227                passed=parsed.get("safety_score", 0.0) >= 0.95,228                details={"reason": parsed.get("judgement_reason")}229            )230        ]231 232# ==========================================233# 5. Agent Evaluator Harness Runner234# ==========================================235 236class AgentEvaluatorHarness:237    """Harness for executing Offline Evaluation and generating final evaluation reports"""238 239    def __init__(self, judge_model_name: str = "gpt-4o"):240        self.judge = LLMTrajectoryJudge(judge_model_name=judge_model_name)241 242    async def run_evaluation(self, test_case: TestCase, trajectory: AgentTrajectory) -> EvaluationReport:243        start_time = time.time()244 245        # 1. Synchronous processing of deterministic evaluation items246        det_metrics = [247            DeterministicEvaluator.evaluate_state_match(test_case.expected_final_state, trajectory.final_state),248            DeterministicEvaluator.evaluate_tool_precision_recall(test_case.expected_tools, trajectory),249            DeterministicEvaluator.evaluate_step_efficiency(test_case.optimal_step_count, trajectory),250            DeterministicEvaluator.evaluate_redundant_loops(trajectory)251        ]252 253        # 2. Asynchronous processing of LLM-as-a-Judge evaluation254        judge_metrics = await self.judge.evaluate_trajectory_quality(test_case, trajectory)255 256        # 3. Composite score calculation and Pass/Fail determination257        all_metrics = det_metrics + judge_metrics258        avg_score = sum(m.score for m in all_metrics) / len(all_metrics)259        overall_pass = all(m.passed for m in det_metrics) and (avg_score >= 0.8)260 261        elapsed = time.time() - start_time262 263        return EvaluationReport(264            task_id=test_case.task_id,265            overall_pass=overall_pass,266            composite_score=round(avg_score, 4),267            deterministic_metrics=det_metrics,268            judge_metrics=judge_metrics,269            evaluation_time_seconds=round(elapsed, 4)270        )271 272# ==========================================273# 6. Execution Example274# ==========================================275 276async def main():277    test_case = TestCase(278        task_id="TC-SQL-FIX-001",279        goal="Analyze DB connection failure error and update SQL connection pool setting to 20.",280        initial_state={"max_connections": 5, "status": "exhausted"},281        expected_final_state={"max_connections": 20, "status": "healthy"},282        expected_tools=["read_log_file", "update_config"],283        optimal_step_count=2284    )285 286    sample_trajectory = AgentTrajectory(287        task_id="TC-SQL-FIX-001",288        goal=test_case.goal,289        initial_state=test_case.initial_state,290        final_state={"max_connections": 20, "status": "healthy"},291        final_answer="Increased SQL connection pool to 20; DB status has transitioned to healthy.",292        total_tokens_used=1450,293        total_cost_usd=0.0043,294        total_latency_seconds=3.2,295        steps=[296            TrajectoryStep(297                step_number=1,298                thought="Need to read the DB log file to determine the root cause of the error.",299                tool_calls=[300                    ToolCallLog(tool_name="read_log_file", arguments={"file_path": "/var/log/db.log"}, observation="Error: connection pool exhausted")301                ]302            ),303            TrajectoryStep(304                step_number=2,305                thought="Connection pool is insufficient, updating config setting to 20.",306                tool_calls=[307                    ToolCallLog(tool_name="update_config", arguments={"key": "max_connections", "value": 20}, observation="Config updated successfully")308                ]309            )310        ]311    )312 313    harness = AgentEvaluatorHarness()314    report = await harness.run_evaluation(test_case, sample_trajectory)315 316    print("==========================================")317    print(f"Evaluation Task ID: {report.task_id}")318    print(f"Overall Pass Status: {'PASSED' if report.overall_pass else 'FAILED'}")319    print(f"Composite Score: {report.composite_score}")320    print(f"Evaluation Time: {report.evaluation_time_seconds}s")321    print("------------------------------------------")322    print("Deterministic Metrics:")323    for m in report.deterministic_metrics:324        print(f"  - {m.metric_name}: Score={m.score}, Passed={m.passed}, Details={m.details}")325    print("LLM-as-a-Judge Metrics:")326    for m in report.judge_metrics:327        print(f"  - {m.metric_name}: Score={m.score}, Passed={m.passed}, Details={m.details}")328    print("==========================================")329 330if __name__ == "__main__":331    asyncio.run(main())

---

6. Production Agent Evaluation Checklist

Before deploying an agent to production, verify the following checklist to ensure high reliability and auditability.

Detailed Checklist Items

• Golden Dataset Diversity and Representativeness
• Are complex multi-step recovery (Backtracking) scenarios included in the dataset beyond simple queries?
• Can the sandbox environment reset 100% identically without randomness?
• Prioritize Deterministic Assertions
• Is environment state verification (DB/file/state changes) prioritized as the primary metric over text answer similarity?
• Is tool call JSON Schema validation and parameter type matching automatically checked?
• LLM-as-a-Judge Bias and Cost Management
• Are consistent rubrics and few-shot examples provided to the judge model to reduce trajectory evaluation variance?
• Is the pipeline configured to invoke the LLM Judge asynchronously only for items that pass deterministic checks to mitigate cost?
• Production Telemetry and Feedback Loop Integration
• Are agent Observe-Plan-Act stages distinctly collected as OpenTelemetry Spans?
• Is a data flywheel built to extract failed trajectories from explicit/implicit user feedback and re-inject them into the Offline Golden Dataset?

---

7. Conclusion and Next Episode Preview

Evaluating AI agents requires a completely different approach from evaluating single-turn LLM model benchmarks. Constructing a hybrid evaluation pipeline (Offline Eval + LLM Judge + Online Telemetry)—which deterministically verifies "whether the final state transformed as intended" amid non-deterministic trajectories while qualitatively evaluating "whether the agent took an efficient and safe path"—is a critical prerequisite for production success.

In the next article, Episode 07: Human-in-the-Loop (HITL) & Authority Systems, we will focus on design patterns for seeking human intervention and approval when agents perform sensitive operations (e.g., DB deletions, financial payments, deployment executions), as well as implementing Role-Based Access Control (RBAC) and audit trails.