{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Trace-Based Agent Evaluation without an LLM Judge\n", "\n", "## Overview\n", "\n", "A final answer can look plausible even when an agent called the wrong tool, sent the wrong arguments, ignored its evidence, or exceeded a latency budget. This tutorial evaluates the *execution trace* instead of asking another language model for a subjective score. It produces reproducible per-case diagnostics and a suite-level quality gate suitable for local regression tests or CI.\n", "\n", "Everything runs offline with the Python standard library. The two agents are deterministic fixtures: one contains realistic routing mistakes, while the improved version demonstrates the same cases after the defects are corrected." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Detailed Explanation\n", "\n", "### What a trace tells us\n", "\n", "A useful trace records ordered tool calls, structured evidence, structured claims, a canonical rendered answer, latency, and any execution error. A test case declares the complete tool contract and evidence. The adapter renders the final answer from claims, removing independently generated prose as a second source of truth.\n", "\n", "### Agent Architecture\n", "\n", "![Trace-Based Agent Evaluation](../images/trace-based-agent-evaluation.svg)\n", "\n", "The evaluator is deliberately outside the agent. It consumes a trace through a small data contract, so the same scorer can compare a LangGraph workflow, a custom loop, or a hosted agent as long as each adapter emits the required fields." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Required Packages\n", "\n", "### No installation required\n", "\n", "The implementation uses `dataclasses`, `statistics`, and other Python standard-library modules." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from dataclasses import dataclass\n", "from math import ceil\n", "from time import perf_counter\n", "from typing import Any, Callable, Dict, List, Optional" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Implementation\n", "\n", "### Define evaluation cases and trace contracts\n", "\n", "Each case specifies only behavior that can be checked deterministically. `expected_calls` describes the complete allowed call sequence, while `expected_evidence` is compared with structured tool evidence and structured claims. The adapter derives a canonical answer from those claims so contradictory prose cannot receive credit." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "@dataclass(frozen=True)\n", "class EvalCase:\n", " \"\"\"Frozen tool-sequence, evidence, and latency expectations for one prompt.\"\"\"\n", " case_id: str\n", " prompt: str\n", " expected_calls: List[Dict[str, Any]]\n", " expected_evidence: Dict[str, Any]\n", " max_latency_ms: int\n", "\n", "\n", "@dataclass(frozen=True)\n", "class AgentTrace:\n", " \"\"\"Framework-neutral record of one agent execution.\"\"\"\n", " answer: str\n", " tool_calls: List[Dict[str, Any]]\n", " evidence: Dict[str, Any]\n", " claims: Dict[str, Any]\n", " latency_ms: int\n", " error: Optional[str] = None\n", "\n", "\n", "@dataclass(frozen=True)\n", "class TraceScore:\n", " \"\"\"Deterministic check results and weighted total for one case.\"\"\"\n", " case_id: str\n", " total: float\n", " checks: Dict[str, bool]\n", " failed_checks: List[str]" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Build a small regression dataset\n", "\n", "The suite covers weather and order lookup with different argument shapes. Expected values are literal fixtures, which keeps the evaluator independent from agent implementation." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "EVAL_CASES = [\n", " EvalCase(\n", " case_id=\"weather-paris\",\n", " prompt=\"What is the weather in Paris?\",\n", " expected_calls=[{\"name\": \"get_weather\", \"args\": {\"city\": \"Paris\"}}],\n", " expected_evidence={\"temperature\": \"18 C\"},\n", " max_latency_ms=500,\n", " ),\n", " EvalCase(\n", " case_id=\"weather-tokyo\",\n", " prompt=\"Do I need an umbrella in Tokyo?\",\n", " expected_calls=[{\"name\": \"get_weather\", \"args\": {\"city\": \"Tokyo\"}}],\n", " expected_evidence={\"condition\": \"rain\"},\n", " max_latency_ms=500,\n", " ),\n", " EvalCase(\n", " case_id=\"order-a100\",\n", " prompt=\"Has order A-100 shipped?\",\n", " expected_calls=[{\"name\": \"lookup_order\", \"args\": {\"order_id\": \"A-100\"}}],\n", " expected_evidence={\"status\": \"shipped\"},\n", " max_latency_ms=500,\n", " ),\n", " EvalCase(\n", " case_id=\"order-b200\",\n", " prompt=\"Where is order B-200?\",\n", " expected_calls=[{\"name\": \"lookup_order\", \"args\": {\"order_id\": \"B-200\"}}],\n", " expected_evidence={\"status\": \"processing\"},\n", " max_latency_ms=500,\n", " ),\n", "]" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Score one execution trace\n", "\n", "The four checks have explicit weights: correct tool sequence 25%, correct argument sequence 25%, grounded evidence 25%, and latency 25%. An execution error receives zero because a fast failure with no result should not earn partial production credit. Evidence must exactly match the structured tool result and claims, and the answer must be their deterministic rendering." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "WEIGHTS = {\"tool\": 0.25, \"arguments\": 0.25, \"evidence\": 0.25, \"latency\": 0.25}\n", "\n", "\n", "def render_claims(claims: Dict[str, Any]) -> str:\n", " \"\"\"Render structured claims into the only accepted answer representation.\"\"\"\n", " return \"; \".join(f\"{key}: {claims[key]}\" for key in sorted(claims))\n", "\n", "\n", "def score_trace(case: EvalCase, trace: AgentTrace) -> TraceScore:\n", " \"\"\"Score a complete trace against one explicit behavioral contract.\"\"\"\n", " if trace.error is not None:\n", " checks = {name: False for name in WEIGHTS}\n", " return TraceScore(case.case_id, 0.0, checks, [\"error\", *checks.keys()])\n", "\n", " actual_tools = [call.get(\"name\") for call in trace.tool_calls]\n", " expected_tools = [call[\"name\"] for call in case.expected_calls]\n", " actual_args = [call.get(\"args\") for call in trace.tool_calls]\n", " expected_args = [call[\"args\"] for call in case.expected_calls]\n", " checks = {\n", " \"tool\": actual_tools == expected_tools,\n", " \"arguments\": actual_args == expected_args,\n", " \"evidence\": (\n", " trace.evidence == case.expected_evidence\n", " and trace.claims == case.expected_evidence\n", " and trace.answer == render_claims(trace.claims)\n", " ),\n", " \"latency\": trace.latency_ms <= case.max_latency_ms,\n", " }\n", " total = round(sum(WEIGHTS[name] for name, passed in checks.items() if passed), 2)\n", " return TraceScore(\n", " case_id=case.case_id,\n", " total=total,\n", " checks=checks,\n", " failed_checks=[name for name, passed in checks.items() if not passed],\n", " )" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Create baseline and improved agent fixtures\n", "\n", "Both functions emit the same production-facing trace contract. The baseline routes one order question to the wrong tool and sends the wrong city for one weather question. The improved agent repairs both defects. Latencies are recorded constants so the tutorial is reproducible." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "TRACE_FIXTURES = {\n", " \"weather-paris\": (\"get_weather\", {\"city\": \"Paris\"}, {\"temperature\": \"18 C\"}, 120),\n", " \"weather-tokyo\": (\"get_weather\", {\"city\": \"Tokyo\"}, {\"condition\": \"rain\"}, 180),\n", " \"order-a100\": (\"lookup_order\", {\"order_id\": \"A-100\"}, {\"status\": \"shipped\"}, 90),\n", " \"order-b200\": (\"lookup_order\", {\"order_id\": \"B-200\"}, {\"status\": \"processing\"}, 110),\n", "}\n", "\n", "\n", "def trace_from_fixture(case: EvalCase, tool: str, args: Dict[str, Any]) -> AgentTrace:\n", " \"\"\"Emit a deterministic trace while allowing routing defects to be injected.\"\"\"\n", " _, _, evidence, latency = TRACE_FIXTURES[case.case_id]\n", " return AgentTrace(\n", " answer=render_claims(evidence),\n", " tool_calls=[{\"name\": tool, \"args\": args}],\n", " evidence=evidence,\n", " claims=evidence.copy(),\n", " latency_ms=latency,\n", " )\n", "\n", "\n", "def baseline_agent(case: EvalCase) -> AgentTrace:\n", " \"\"\"Return traces with two deliberate routing regressions.\"\"\"\n", " expected_tool, expected_args, _, _ = TRACE_FIXTURES[case.case_id]\n", " if case.case_id == \"weather-tokyo\":\n", " return trace_from_fixture(case, expected_tool, {\"city\": \"Kyoto\"})\n", " if case.case_id == \"order-b200\":\n", " return trace_from_fixture(case, \"search_web\", expected_args)\n", " return trace_from_fixture(case, expected_tool, expected_args)\n", "\n", "\n", "def improved_agent(case: EvalCase) -> AgentTrace:\n", " \"\"\"Return the corrected deterministic trace for every case.\"\"\"\n", " expected_tool, expected_args, _, _ = TRACE_FIXTURES[case.case_id]\n", " return trace_from_fixture(case, expected_tool, expected_args)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Aggregate the suite\n", "\n", "A case passes only at full credit. The report keeps per-case scores for diagnosis and summarizes pass rate, tool accuracy, and nearest-rank p95 latency. Empty suites are rejected explicitly, while an ordinary agent exception becomes a zero-credit trace so later cases still run. Exception latency uses a monotonic clock, rounded up with a 1 ms minimum so failed calls still contribute to latency metrics." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "Agent = Callable[[EvalCase], AgentTrace]\n", "\n", "\n", "def percentile(values: List[int], percentile_value: float) -> int:\n", " \"\"\"Return a nearest-rank percentile for a non-empty latency list.\"\"\"\n", " ordered = sorted(values)\n", " index = max(0, ceil(percentile_value * len(ordered)) - 1)\n", " return ordered[index]\n", "\n", "\n", "def evaluate_suite(agent: Agent, cases: List[EvalCase]) -> Dict[str, Any]:\n", " \"\"\"Run frozen cases and aggregate diagnostics and suite metrics.\"\"\"\n", " if not cases:\n", " raise ValueError(\"at least one evaluation case is required\")\n", "\n", " runs = []\n", " for case in cases:\n", " started_at = perf_counter()\n", " try:\n", " trace = agent(case)\n", " except Exception as exc:\n", " elapsed_ms = max(1, ceil((perf_counter() - started_at) * 1000))\n", " trace = AgentTrace(\n", " answer=\"\",\n", " tool_calls=[],\n", " evidence={},\n", " claims={},\n", " latency_ms=elapsed_ms,\n", " error=f\"{type(exc).__name__}: {exc}\",\n", " )\n", " runs.append({\"case\": case, \"trace\": trace, \"score\": score_trace(case, trace)})\n", "\n", " return {\n", " \"case_count\": len(runs),\n", " \"pass_rate\": sum(run[\"score\"].total == 1.0 for run in runs) / len(runs),\n", " \"tool_accuracy\": sum(run[\"score\"].checks[\"tool\"] for run in runs) / len(runs),\n", " \"p95_latency_ms\": percentile([run[\"trace\"].latency_ms for run in runs], 0.95),\n", " \"runs\": runs,\n", " }" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Turn metrics into a quality gate\n", "\n", "The gate returns structured failures instead of raising immediately, so CI can print all regressions in one run. Higher is better for pass rate and tool accuracy; lower is better for latency." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def check_quality_gate(\n", " report: Dict[str, Any], thresholds: Dict[str, float]\n", ") -> Dict[str, Any]:\n", " \"\"\"Return every metric regression against inclusive thresholds.\"\"\"\n", " failures = []\n", " if report[\"pass_rate\"] < thresholds[\"pass_rate\"]:\n", " failures.append(\n", " f\"pass_rate {report['pass_rate']:.2f} < {thresholds['pass_rate']:.2f}\"\n", " )\n", " if report[\"tool_accuracy\"] < thresholds[\"tool_accuracy\"]:\n", " failures.append(\n", " f\"tool_accuracy {report['tool_accuracy']:.2f} < {thresholds['tool_accuracy']:.2f}\"\n", " )\n", " if report[\"p95_latency_ms\"] > thresholds[\"p95_latency_ms\"]:\n", " failures.append(\n", " f\"p95_latency_ms {report['p95_latency_ms']} > {thresholds['p95_latency_ms']}\"\n", " )\n", " return {\"passed\": not failures, \"failures\": failures}" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Usage Example\n", "\n", "### Compare baseline and improved agents\n", "\n", "The baseline passes two of four cases. The case-level output shows whether the failure came from routing, arguments, evidence, latency, or execution itself." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "baseline_report = evaluate_suite(baseline_agent, EVAL_CASES)\n", "improved_report = evaluate_suite(improved_agent, EVAL_CASES)\n", "\n", "for label, report in [(\"baseline\", baseline_report), (\"improved\", improved_report)]:\n", " print(\n", " label,\n", " {\n", " \"pass_rate\": report[\"pass_rate\"],\n", " \"tool_accuracy\": report[\"tool_accuracy\"],\n", " \"p95_latency_ms\": report[\"p95_latency_ms\"],\n", " },\n", " )\n", " for run in report[\"runs\"]:\n", " print(\" \", run[\"score\"].case_id, run[\"score\"].failed_checks)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Enforce the regression threshold\n", "\n", "A repository test can assert on the final `passed` field or exit non-zero when it is false." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "THRESHOLDS = {\"pass_rate\": 1.0, \"tool_accuracy\": 1.0, \"p95_latency_ms\": 500}\n", "baseline_gate = check_quality_gate(baseline_report, THRESHOLDS)\n", "improved_gate = check_quality_gate(improved_report, THRESHOLDS)\n", "\n", "assert baseline_gate == {\n", " \"passed\": False,\n", " \"failures\": [\"pass_rate 0.50 < 1.00\", \"tool_accuracy 0.75 < 1.00\"],\n", "}\n", "assert improved_gate == {\"passed\": True, \"failures\": []}\n", "print(\"Improved agent passes the quality gate\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Comparison\n", "\n", "| Evaluation method | Reproducible | Diagnoses tool behavior | Handles open-ended prose | Cost |\n", "|---|---:|---:|---:|---:|\n", "| Exact answer matching | High | No | Poorly | Low |\n", "| LLM-as-judge | Depends on model and prompt | Sometimes | Well | Model calls |\n", "| Trace-based checks (this tutorial) | High | Yes | Through evidence contracts | Low |\n", "| Human review | With a rubric | Yes | Best | High |\n", "\n", "Trace checks are strongest for contracts a machine can observe: tool routing, parameters, structured claims, errors, ordering, latency, and cost. The canonical answer in this tutorial deliberately does not evaluate tone or nuanced synthesis. Mature evaluations combine deterministic gates for non-negotiable behavior with sampled human or calibrated judge review for subjective natural-language quality." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Additional Considerations\n", "\n", "- **Freeze the dataset.** Version cases and thresholds together so metric changes have an attributable cause. Keep a separate hidden set if prompts might overfit public cases.\n", "- **Redact traces.** Arguments and tool results can contain personal or secret data. Store the minimum fields needed for evaluation and apply retention limits.\n", "- **Measure autonomous success separately.** A run completed after human escalation is useful operationally but is not an autonomous pass; record both outcomes.\n", "- **Use distributions for real latency.** Four deterministic fixtures make the mechanics easy to see. Production gates need repeated runs, warm/cold separation, timeouts, and enough samples for percentiles.\n", "- **Do not reward fast errors.** This scorer zeros every check when the trace has an error. Report error rate independently as suites grow.\n", "- **Adapt at the boundary.** Convert framework-specific spans into `AgentTrace` rather than embedding LangGraph, LangSmith, OpenTelemetry, or vendor assumptions in the scorer.\n", "- **Audit the evaluator.** Hand-check literals and write mutation tests for wrong tools, arguments, missing evidence, slow traces, and errors. A broken gate can create more confidence than no gate." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## References\n", "\n", "- [LangSmith evaluation concepts](https://docs.smith.langchain.com/evaluation)\n", "- [OpenTelemetry trace specification](https://opentelemetry.io/docs/concepts/signals/traces/)\n", "- [NIST AI Risk Management Framework](https://www.nist.gov/itl/ai-risk-management-framework)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "📖 **Want the decisions behind agents like this one, explained without code?** [*AI Agents Made Simple*](https://europe-west1-rag-techniques-views-tracker.cloudfunctions.net/rag-techniques-tracker?notebook=all-agents-tutorials--trace-based-agent-evaluation&click=book-nb-agents&target=https%3A%2F%2Fwww.amazon.com%2Fdp%2FB0HNXRSRH2&retarget=0&text=book-nb-agents) by the author of this repository covers this and much more. Available on Amazon." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.9" } }, "nbformat": 4, "nbformat_minor": 5 }