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.

Goal G: DB Connection Pool Configuration → Agent Controller branches into three trajectories.

The original shows three action cards for Path C but labels its total as 4 Steps. Both are preserved; no missing fourth action is invented.

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.

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

Remove identifying information from failed trajectories and user feedback before adding them to the offline evaluation dataset. Use those cases to test whether a known failure recurs.

---

3. Key Metrics Framework

A reproducible three-task evaluation fixture

This table is an illustrative fixture, not production measurements. Each outcome can be checked against the declared target state.

If A and B are the declared business tasks, their Task Success Rate is 2/2. Tool argument validity is 4/5 = 80% across all calls, while the unauthorized-action block rate is 1/1. Do not mix these denominators into a single '100% successful' claim. Report task outcomes, tool contracts and policy outcomes separately.

Pass@k asks whether at least one of up to k attempts succeeds on the same problem; it is not interchangeable with a single-run Task Success Rate. State reset rules, the retry policy, sample sizes and uncertainty must be reported.

The 0.4/0.2/0.2/0.2 composite score weights below are arbitrary illustrative weights, not a standard or measured optimum. A secret leak or unauthorized database mutation must fail its separate safety gate, regardless of the weighted quality score.

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

The two pipelines below have separate inputs. This model contains no deployment edge or feedback arrow between them.

4.2 Trajectory Evaluation Sequence Flow

4.3 Metric Taxonomy Tree

Agent Evaluation System connects to three metric categories. Each row lists the children of that category.

4.4 Hybrid Evaluation and Filtering Workflow

Trajectory Log Received enters check 1. Only Pass reaches the next stage.

After all three checks, the path is Trigger Async LLM-as-a-Judge → Evaluate Reasoning & Safety Rubrics → Generate Final Evaluation Report.

---

5. Python-based Custom Agent Evaluator Harness Implementation

This Evaluator Harness uses Pydantic v2 and asyncio to collect trajectories and state changes, then run deterministic checks and asynchronous LLM-as-a-Judge evaluations.

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

Agent evaluation checks both the final state and the path taken to reach it. Offline checks, LLM judges, and online telemetry examine different parts of task success, efficiency, and permitted behavior.

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.