AI Agent可观测性与调试完全教程

1. AI Agent调试的挑战与痛点

AI Agent系统与传统软件有本质区别——它的行为是非确定性的。同一个输入，不同次执行可能走完全不同的路径。这让调试变得异常困难：

• 执行路径不可预测：Agent根据LLM推理动态决定调用哪些工具、以什么顺序调用，传统断点调试几乎无效
• 错误链路长：一个最终输出错误可能源于prompt理解偏差、工具返回异常、上下文截断等多个环节
• 黑盒推理：LLM的决策过程不透明，难以判断为什么Agent选择了某个工具而非另一个
• 成本隐性增长：一个陷入循环的Agent可能在几分钟内烧掉大量token预算
• 多轮对话状态爆炸：长对话中上下文不断膨胀，历史信息可能干扰当前决策

传统日志（print、logger.info）在这种场景下完全不够用。需要一套完整的可观测性体系来应对。

2. 可观测性三大支柱

日志（Logs）

结构化事件记录，是调试的基础。Agent系统需要记录的不仅是文本日志，还包括结构化的输入/输出对：

import json
import time
from dataclasses import dataclass, asdict
from typing import Any
from datetime import datetime

@dataclass
class AgentLog:
    timestamp: str
    trace_id: str
    span_id: str
    event_type: str       # "llm_call" | "tool_call" | "retrieval" | "decision"
    input_data: Any
    output_data: Any
    metadata: dict
    duration_ms: float
    token_usage: dict | None = None
    error: str | None = None

class StructuredLogger:
    def __init__(self, service_name: str):
        self.service_name = service_name

    def log(self, event: AgentLog):
        entry = asdict(event)
        entry["service"] = self.service_name
        # 输出JSON格式，方便ELK/Loki等系统采集
        print(json.dumps(entry, ensure_ascii=False, default=str))

    def log_llm_call(self, trace_id: str, span_id: str,
                     prompt: str, response: str,
                     duration_ms: float, tokens: dict):
        self.log(AgentLog(
            timestamp=datetime.utcnow().isoformat(),
            trace_id=trace_id,
            span_id=span_id,
            event_type="llm_call",
            input_data=prompt[:500],  # 截断避免日志过大
            output_data=response[:500],
            metadata={"model": "gpt-4o"},
            duration_ms=duration_ms,
            token_usage=tokens,
        ))

    def log_tool_call(self, trace_id: str, span_id: str,
                      tool_name: str, args: dict, result: Any,
                      duration_ms: float, error: str = None):
        self.log(AgentLog(
            timestamp=datetime.utcnow().isoformat(),
            trace_id=trace_id,
            span_id=span_id,
            event_type="tool_call",
            input_data={"tool": tool_name, "args": args},
            output_data=str(result)[:500],
            metadata={"tool_name": tool_name},
            duration_ms=duration_ms,
            error=error,
        ))

指标（Metrics）

数值型时间序列数据，用于监控趋势和设置告警：

from collections import defaultdict
import threading

class AgentMetrics:
    """Agent指标收集器"""

    def __init__(self):
        self._lock = threading.Lock()
        self._counters = defaultdict(int)
        self._histograms = defaultdict(list)

    def increment(self, name: str, value: int = 1, tags: dict = None):
        with self._lock:
            key = f"{name}:{tags}" if tags else name
            self._counters[key] += value

    def record_duration(self, name: str, duration_ms: float, tags: dict = None):
        with self._lock:
            key = f"{name}:{tags}" if tags else name
            self._histograms[key].append(duration_ms)

    def record_tokens(self, model: str, prompt_tokens: int, completion_tokens: int):
        with self._lock:
            self._counters[f"tokens.prompt.{model}"] += prompt_tokens
            self._counters[f"tokens.completion.{model}"] += completion_tokens
            self._counters[f"tokens.total.{model}"] += prompt_tokens + completion_tokens

    def get_summary(self) -> dict:
        """输出Prometheus兼容的指标摘要"""
        with self._lock:
            summary = {
                "counters": dict(self._counters),
                "latency_p50": {},
                "latency_p99": {},
            }
            for key, values in self._histograms.items():
                if values:
                    sorted_v = sorted(values)
                    summary["latency_p50"][key] = sorted_v[len(sorted_v) // 2]
                    summary["latency_p99"][key] = sorted_v[int(len(sorted_v) * 0.99)]
            return summary

# 全局指标实例
metrics = AgentMetrics()

# 使用示例
metrics.increment("agent.requests.total")
metrics.record_duration("agent.llm_call.duration_ms", 1234.5)
metrics.record_tokens("gpt-4o", prompt_tokens=500, completion_tokens=200)

追踪（Traces）

端到端的请求链路，是Agent可观测性中最关键的部分：

import uuid
from contextlib import contextmanager
from dataclasses import dataclass, field

@dataclass
class Span:
    trace_id: str
    span_id: str
    parent_id: str | None
    name: str
    start_time: float
    end_time: float = 0
    attributes: dict = field(default_factory=dict)
    events: list = field(default_factory=list)
    status: str = "ok"  # "ok" | "error"

    @property
    def duration_ms(self):
        return (self.end_time - self.start_time) * 1000

class Tracer:
    """分布式追踪器"""

    def __init__(self):
        self._spans: list[Span] = []
        self._current_span: Span | None = None

    @contextmanager
    def start_span(self, name: str, attributes: dict = None):
        span_id = str(uuid.uuid4())[:8]
        trace_id = self._current_span.trace_id if self._current_span else str(uuid.uuid4())[:12]
        parent_id = self._current_span.span_id if self._current_span else None

        span = Span(
            trace_id=trace_id,
            span_id=span_id,
            parent_id=parent_id,
            name=name,
            start_time=time.time(),
            attributes=attributes or {},
        )

        parent = self._current_span
        self._current_span = span

        try:
            yield span
        except Exception as e:
            span.status = "error"
            span.events.append({"event": "error", "message": str(e)})
            raise
        finally:
            span.end_time = time.time()
            self._spans.append(span)
            self._current_span = parent

    def get_trace(self, trace_id: str) -> list[Span]:
        return [s for s in self._spans if s.trace_id == trace_id]

3. LangSmith / LangFuse / Arize Phoenix 工具对比

LangSmith

LangChain官方的可观测性平台，与LangChain生态深度集成：

# 安装：pip install langsmith
import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"

# 使用LangChain时自动采集追踪
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent

llm = ChatOpenAI(model="gpt-4o")
# 所有LangChain调用自动上报到LangSmith
# 无需额外代码，零侵入

LangFuse

开源替代方案，支持自部署，API更灵活：

# 安装：pip install langfuse
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse(
    public_key="pk-xxx",
    secret_key="sk-xxx",
    host="https://cloud.langfuse.com",  # 或自部署地址
)

@observe()  # 自动追踪此函数
def my_agent_call(question: str) -> str:
    # 记录额外元数据
    langfuse_context.update_current_observation(
        metadata={"user_id": "user-123"},
        tags=["production", "customer-support"],
    )

    response = llm.invoke(question)
    return response.content

# 手动追踪LLM调用
@observe(as_type="generation")
def call_llm(prompt: str) -> str:
    start = time.time()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    # 上报token使用量
    langfuse_context.update_current_observation(
        usage={
            "input": response.usage.prompt_tokens,
            "output": response.usage.completion_tokens,
            "total": response.usage.total_tokens,
        },
        model="gpt-4o",
    )
    return response.choices[0].message.content

Arize Phoenix

专注LLM可观测性，提供强大的嵌入向量分析和幻觉检测：

# 安装：pip install arize-phoenix opentelemetry-sdk
import phoenix as px

# 启动本地Phoenix服务
px.launch_app()

from openinference.instrumentation.langchain import LangChainInstrumentor
from opentelemetry import trace as otel_trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

# 配置OpenTelemetry
provider = TracerProvider()
provider.add_span_processor(
    otel_trace.export.SimpleSpanProcessor(OTLPSpanExporter("http://localhost:6006/v1/traces"))
)
otel_trace.set_tracer_provider(provider)

# 自动注入LangChain
LangChainInstrumentor().instrument()

三者对比

选择建议：LangChain重度用户选LangSmith；需要自部署和成本控制选LangFuse；需要深度嵌入分析和幻觉检测选Phoenix。

4. Agent执行链路追踪

追踪一个完整的Agent执行链路，从用户输入到最终输出：

import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable
from opentelemetry import trace

tracer = trace.get_tracer("agent-tracer")

class TracedAgent:
    """带完整追踪的Agent"""

    def __init__(self, llm, tools: list, name: str = "agent"):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.name = name

    def run(self, user_input: str) -> str:
        with tracer.start_as_current_span("agent.run") as root_span:
            root_span.set_attribute("input", user_input[:200])
            messages = [{"role": "user", "content": user_input}]
            max_iterations = 10

            for iteration in range(max_iterations):
                with tracer.start_as_current_span(f"agent.iteration.{iteration}"):
                    # LLM决策
                    with tracer.start_as_current_span("llm.decide") as llm_span:
                        llm_span.set_attribute("iteration", iteration)
                        start = time.time()
                        response = self.llm.invoke(messages)
                        llm_span.set_attribute("duration_ms", (time.time() - start) * 1000)
                        llm_span.set_attribute("model", self.llm.model_name)

                    # 检查是否有工具调用
                    if not hasattr(response, "tool_calls") or not response.tool_calls:
                        root_span.set_attribute("iterations", iteration + 1)
                        root_span.set_attribute("final_answer", response.content[:500])
                        return response.content

                    messages.append(response)

                    # 执行工具调用
                    for tool_call in response.tool_calls:
                        with tracer.start_as_current_span(f"tool.{tool_call['name']}") as tool_span:
                            tool_span.set_attribute("tool_name", tool_call["name"])
                            tool_span.set_attribute("tool_args", str(tool_call["args"])[:300])

                            start = time.time()
                            try:
                                result = self.tools[tool_call["name"]].invoke(tool_call["args"])
                                tool_span.set_attribute("result_preview", str(result)[:300])
                            except Exception as e:
                                tool_span.set_status(trace.Status(trace.StatusCode.ERROR, str(e)))
                                result = f"工具执行错误: {e}"

                            tool_span.set_attribute("duration_ms", (time.time() - start) * 1000)
                            messages.append({
                                "role": "tool",
                                "tool_call_id": tool_call["id"],
                                "content": str(result),
                            })

            root_span.set_attribute("status", "max_iterations_reached")
            return "达到最大迭代次数，未能完成任务"

追踪数据可视化后呈现为树形结构：

agent.run [2.3s]
├── llm.decide [0.8s] → 决定调用 search_web
├── tool.search_web [1.2s] → 返回5条结果
├── llm.decide [0.6s] → 决定调用 read_document
├── tool.read_document [0.4s] → 返回文档内容
└── llm.decide [0.5s] → 生成最终回答

5. Token消耗与成本监控

Token是Agent系统的核心成本单元。失控的token消耗可能在几小时内耗尽预算。

from dataclasses import dataclass
from datetime import datetime, timedelta
from collections import defaultdict

@dataclass
class TokenBudget:
    """Token预算与成本追踪"""
    # 价格表（美元/1M tokens）
    PRICING = {
        "gpt-4o": {"input": 2.5, "output": 10.0},
        "gpt-4o-mini": {"input": 0.15, "output": 0.6},
        "claude-3-5-sonnet": {"input": 3.0, "output": 15.0},
        "deepseek-v3": {"input": 0.27, "output": 1.1},
    }

    def __init__(self, daily_budget_usd: float = 50.0):
        self.daily_budget = daily_budget_usd
        self.usage = defaultdict(lambda: {"input": 0, "output": 0, "cost": 0.0})
        self.daily_usage = defaultdict(float)

    def record(self, model: str, input_tokens: int, output_tokens: int,
               trace_id: str = None, user_id: str = None):
        """记录一次token使用"""
        pricing = self.PRICING.get(model, {"input": 1.0, "output": 3.0})
        cost = (input_tokens * pricing["input"] + output_tokens * pricing["output"]) / 1_000_000

        today = datetime.utcnow().strftime("%Y-%m-%d")
        self.daily_usage[today] += cost

        if trace_id:
            self.usage[trace_id]["input"] += input_tokens
            self.usage[trace_id]["output"] += output_tokens
            self.usage[trace_id]["cost"] += cost

        # 预算检查
        if self.daily_usage[today] > self.daily_budget * 0.8:
            self._alert_budget(f"当日成本已达预算的{self.daily_usage[today]/self.daily_budget*100:.0f}%")

        return cost

    def get_daily_report(self) -> dict:
        today = datetime.utcnow().strftime("%Y-%m-%d")
        return {
            "date": today,
            "total_cost_usd": round(self.daily_usage[today], 4),
            "budget_remaining_usd": round(self.daily_budget - self.daily_usage[today], 4),
            "budget_usage_pct": round(self.daily_usage[today] / self.daily_budget * 100, 1),
            "traces_count": len(self.usage),
        }

    def _alert_budget(self, message: str):
        """预算告警"""
        print(f"⚠️ [BUDGET ALERT] {message}")
        # 实际生产中接入Slack/钉钉/邮件告警

# 使用示例
budget = TokenBudget(daily_budget_usd=100.0)
cost = budget.record("gpt-4o", input_tokens=1500, output_tokens=800, trace_id="trace-001")
print(f"本次调用成本: ${cost:.4f}")

将token监控集成到OpenTelemetry中：

from opentelemetry import trace

def record_token_metrics(model: str, usage: dict, span: trace.Span = None):
    """将token使用量记录到OpenTelemetry span"""
    if span is None:
        span = trace.get_current_span()

    span.set_attribute("gen_ai.usage.input_tokens", usage.get("prompt_tokens", 0))
    span.set_attribute("gen_ai.usage.output_tokens", usage.get("completion_tokens", 0))
    span.set_attribute("gen_ai.usage.total_tokens", usage.get("total_tokens", 0))
    span.set_attribute("gen_ai.request.model", model)

6. 工具调用日志与错误诊断

Agent的工具调用是错误高发区。系统化的工具日志对诊断至关重要：

import traceback
from functools import wraps

class ToolCallLogger:
    """工具调用日志记录器"""

    def __init__(self):
        self.call_history: list[dict] = []

    def wrap_tool(self, tool_fn: Callable, tool_name: str) -> Callable:
        """包装工具函数，自动记录调用日志"""
        @wraps(tool_fn)
        def wrapper(*args, **kwargs):
            call_id = str(uuid.uuid4())[:8]
            log_entry = {
                "call_id": call_id,
                "tool_name": tool_name,
                "args": {"args": args, "kwargs": kwargs},
                "start_time": datetime.utcnow().isoformat(),
                "status": "pending",
            }

            start = time.time()
            try:
                result = tool_fn(*args, **kwargs)
                log_entry["status"] = "success"
                log_entry["result_preview"] = str(result)[:500]
                log_entry["result_type"] = type(result).__name__
                return result

            except TimeoutError:
                log_entry["status"] = "timeout"
                log_entry["error"] = "工具执行超时"
                raise

            except Exception as e:
                log_entry["status"] = "error"
                log_entry["error"] = str(e)
                log_entry["traceback"] = traceback.format_exc()
                raise

            finally:
                log_entry["duration_ms"] = (time.time() - start) * 1000
                log_entry["end_time"] = datetime.utcnow().isoformat()
                self.call_history.append(log_entry)

                # 日志输出
                status_emoji = "✅" if log_entry["status"] == "success" else "❌"
                print(
                    f"{status_emoji} Tool[{tool_name}] "
                    f"call_id={call_id} "
                    f"duration={log_entry['duration_ms']:.0f}ms "
                    f"status={log_entry['status']}"
                )

        return wrapper

    def get_error_summary(self) -> list[dict]:
        """获取错误摘要，用于诊断"""
        errors = [h for h in self.call_history if h["status"] != "success"]
        return [
            {
                "tool": e["tool_name"],
                "error": e.get("error"),
                "duration_ms": e.get("duration_ms"),
                "timestamp": e.get("start_time"),
            }
            for e in errors
        ]

# 使用示例
logger = ToolCallLogger()

def search_database(query: str) -> list:
    # 模拟数据库查询
    if not query:
        raise ValueError("查询不能为空")
    return [{"title": "结果1", "content": "相关内容"}]

# 包装工具
wrapped_search = logger.wrap_tool(search_database, "search_database")

# 调用时自动记录
try:
    result = wrapped_search("退款政策")
except Exception as e:
    pass  # 错误已被记录

# 查看错误摘要
print(logger.get_error_summary())

工具调用模式分析：

def analyze_tool_patterns(call_history: list[dict]) -> dict:
    """分析工具调用模式，发现潜在问题"""
    from collections import Counter

    tool_counts = Counter(h["tool_name"] for h in call_history)
    error_counts = Counter(h["tool_name"] for h in call_history if h["status"] != "success")
    avg_durations = {}

    for tool_name in tool_counts:
        durations = [h["duration_ms"] for h in call_history if h["tool_name"] == tool_name]
        avg_durations[tool_name] = sum(durations) / len(durations)

    return {
        "call_frequency": dict(tool_counts),
        "error_rate": {t: error_counts[t] / tool_counts[t] for t in tool_counts},
        "avg_latency_ms": avg_durations,
        "slowest_tool": max(avg_durations, key=avg_durations.get),
        "most_error_prone": max(error_counts, key=error_counts.get) if error_counts else None,
    }

7. Prompt版本管理与A/B测试

Prompt是Agent的核心资产，需要像代码一样进行版本管理：

import hashlib
import json
from datetime import datetime

class PromptRegistry:
    """Prompt版本注册中心"""

    def __init__(self, storage_path: str = "./prompts"):
        self.storage_path = storage_path
        self.prompts: dict[str, list[dict]] = {}

    def register(self, name: str, template: str, metadata: dict = None) -> str:
        """注册新版本的prompt"""
        version_hash = hashlib.sha256(template.encode()).hexdigest()[:8]

        entry = {
            "version": version_hash,
            "template": template,
            "created_at": datetime.utcnow().isoformat(),
            "metadata": metadata or {},
            "eval_score": None,
            "is_active": False,
        }

        if name not in self.prompts:
            self.prompts[name] = []

        # 检查是否已存在相同版本
        if any(v["version"] == version_hash for v in self.prompts[name]):
            return version_hash

        self.prompts[name].append(entry)
        return version_hash

    def activate(self, name: str, version: str):
        """激活指定版本"""
        for v in self.prompts.get(name, []):
            v["is_active"] = v["version"] == version

    def get_active(self, name: str) -> str | None:
        """获取当前活跃版本"""
        for v in self.prompts.get(name, []):
            if v["is_active"]:
                return v["template"]
        # 默认返回最新版本
        if self.prompts.get(name):
            return self.prompts[name][-1]["template"]
        return None

    def ab_test(self, name: str, variants: list[str], traffic_split: list[float] = None):
        """配置A/B测试"""
        if traffic_split is None:
            traffic_split = [1.0 / len(variants)] * len(variants)

        versions = []
        for template in variants:
            ver = self.register(name, template, {"ab_test": True})
            versions.append(ver)

        return {
            "name": name,
            "variants": versions,
            "traffic_split": traffic_split,
        }

# 使用示例
registry = PromptRegistry()

# 注册不同版本的系统prompt
v1 = registry.register("system_prompt", "你是一个技术助手，请简洁准确地回答问题。")
v2 = registry.register("system_prompt", "你是一位资深技术专家。回答时先分析问题本质，再给出方案，最后补充注意事项。")

# 配置A/B测试
ab_config = registry.ab_test("system_prompt", [
    "你是一个技术助手，请简洁准确地回答问题。",
    "你是一位资深技术专家。回答时先分析问题本质，再给出方案，最后补充注意事项。",
], traffic_split=[0.5, 0.5])

A/B测试结果收集：

class ABTestCollector:
    """A/B测试结果收集器"""

    def __init__(self):
        self.results: dict[str, list[dict]] = {}

    def record(self, test_name: str, variant: str,
               user_score: float = None, latency_ms: float = None,
               token_count: int = None):
        if test_name not in self.results:
            self.results[test_name] = []
        self.results[test_name].append({
            "variant": variant,
            "user_score": user_score,
            "latency_ms": latency_ms,
            "token_count": token_count,
            "timestamp": datetime.utcnow().isoformat(),
        })

    def analyze(self, test_name: str) -> dict:
        """分析A/B测试结果"""
        data = self.results.get(test_name, [])
        variants = {}
        for entry in data:
            v = entry["variant"]
            if v not in variants:
                variants[v] = {"scores": [], "latencies": [], "tokens": []}
            if entry["user_score"] is not None:
                variants[v]["scores"].append(entry["user_score"])
            if entry["latency_ms"] is not None:
                variants[v]["latencies"].append(entry["latency_ms"])
            if entry["token_count"] is not None:
                variants[v]["tokens"].append(entry["token_count"])

        summary = {}
        for v, metrics in variants.items():
            summary[v] = {
                "sample_size": len(metrics["scores"]),
                "avg_score": sum(metrics["scores"]) / len(metrics["scores"]) if metrics["scores"] else None,
                "avg_latency_ms": sum(metrics["latencies"]) / len(metrics["latencies"]) if metrics["latencies"] else None,
                "avg_tokens": sum(metrics["tokens"]) / len(metrics["tokens"]) if metrics["tokens"] else None,
            }
        return summary

8. Agent性能基准测试

建立标准化的测试集来量化Agent的能力变化：

from dataclasses import dataclass
from typing import Any

@dataclass
class BenchmarkCase:
    """单个基准测试用例"""
    id: str
    category: str          # "tool_use" | "reasoning" | "multi_step" | "error_recovery"
    input_query: str
    expected_tools: list[str]   # 期望调用的工具列表
    expected_answer: str | None # 期望答案（可选）
    max_iterations: int = 10
    max_latency_ms: float = 30000
    max_tokens: int = 5000

class AgentBenchmark:
    """Agent基准测试框架"""

    def __init__(self, agent, test_cases: list[BenchmarkCase]):
        self.agent = agent
        self.test_cases = test_cases

    def run(self) -> dict:
        results = []
        for case in self.test_cases:
            result = self._run_case(case)
            results.append(result)
        return self._aggregate(results)

    def _run_case(self, case: BenchmarkCase) -> dict:
        start = time.time()
        try:
            answer = self.agent.run(case.input_query)
            latency_ms = (time.time() - start) * 1000

            # 评估结果
            tool_match = self._check_tools(case.expected_tools)
            answer_quality = self._evaluate_answer(answer, case.expected_answer) if case.expected_answer else None
            latency_ok = latency_ms <= case.max_latency_ms

            return {
                "id": case.id,
                "category": case.category,
                "status": "pass",
                "latency_ms": latency_ms,
                "latency_ok": latency_ok,
                "tool_match": tool_match,
                "answer_quality": answer_quality,
            }
        except Exception as e:
            return {
                "id": case.id,
                "category": case.category,
                "status": "error",
                "error": str(e),
            }

    def _check_tools(self, expected: list[str]) -> bool:
        """检查实际调用的工具是否匹配期望"""
        # 实际实现需要从追踪数据中提取
        return True

    def _evaluate_answer(self, actual: str, expected: str) -> float:
        """简单的答案质量评估"""
        # 实际生产中用LLM-as-judge
        from difflib import SequenceMatcher
        return SequenceMatcher(None, actual, expected).ratio()

    def _aggregate(self, results: list[dict]) -> dict:
        passed = [r for r in results if r["status"] == "pass"]
        return {
            "total": len(results),
            "passed": len(passed),
            "pass_rate": len(passed) / len(results) if results else 0,
            "avg_latency_ms": sum(r.get("latency_ms", 0) for r in passed) / len(passed) if passed else 0,
            "by_category": self._group_by_category(results),
        }

    def _group_by_category(self, results: list[dict]) -> dict:
        groups = {}
        for r in results:
            cat = r["category"]
            if cat not in groups:
                groups[cat] = {"total": 0, "passed": 0}
            groups[cat]["total"] += 1
            if r["status"] == "pass":
                groups[cat]["passed"] += 1
        return {k: {**v, "pass_rate": v["passed"] / v["total"]} for k, v in groups.items()}

构建测试集：

test_cases = [
    BenchmarkCase(
        id="basic_search",
        category="tool_use",
        input_query="搜索Python异步编程的最佳实践",
        expected_tools=["web_search"],
        max_latency_ms=10000,
    ),
    BenchmarkCase(
        id="multi_step",
        category="multi_step",
        input_query="查找最近的AI论文，总结核心观点，并保存到笔记",
        expected_tools=["web_search", "read_paper", "save_note"],
        max_iterations=8,
        max_latency_ms=60000,
    ),
    BenchmarkCase(
        id="error_recovery",
        category="error_recovery",
        input_query="查询天气（工具可能失败）",
        expected_tools=["get_weather"],
        max_iterations=5,
    ),
]

benchmark = AgentBenchmark(agent, test_cases)
report = benchmark.run()
print(json.dumps(report, indent=2))

9. 生产环境Agent监控告警

Prometheus + Grafana集成

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 定义指标
agent_requests = Counter("agent_requests_total", "Total agent requests", ["status", "model"])
agent_latency = Histogram("agent_request_duration_seconds", "Agent request latency", ["model"])
agent_tokens = Counter("agent_tokens_total", "Total tokens used", ["model", "type"])
agent_cost = Gauge("agent_cost_dollars", "Estimated cost in dollars", ["model"])
active_agents = Gauge("agent_active_count", "Currently running agents")

# 在Agent中使用
class MonitoredAgent:
    def __init__(self, llm, model_name: str = "gpt-4o"):
        self.llm = llm
        self.model_name = model_name

    def run(self, query: str) -> str:
        active_agents.inc()
        start = time.time()

        try:
            response = self._execute(query)
            agent_requests.labels(status="success", model=self.model_name).inc()
            return response
        except Exception as e:
            agent_requests.labels(status="error", model=self.model_name).inc()
            raise
        finally:
            duration = time.time() - start
            agent_latency.labels(model=self.model_name).observe(duration)
            active_agents.dec()

# 启动Prometheus指标端口
start_http_server(8000)

告警规则配置

# prometheus/alert_rules.yml
groups:
  - name: agent_alerts
    rules:
      # 错误率告警
      - alert: AgentHighErrorRate
        expr: rate(agent_requests_total{status="error"}[5m]) / rate(agent_requests_total[5m]) > 0.1
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Agent错误率超过10%"

      # 延迟告警
      - alert: AgentHighLatency
        expr: histogram_quantile(0.99, rate(agent_request_duration_seconds_bucket[5m])) > 30
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Agent P99延迟超过30秒"

      # 成本告警
      - alert: AgentCostSpike
        expr: increase(agent_tokens_total[1h]) > 1000000
        labels:
          severity: warning
        annotations:
          summary: "过去1小时token消耗超过100万"

      # 活跃Agent数量
      - alert: AgentStuck
        expr: agent_active_count > 50
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "活跃Agent数量异常，可能存在卡死"

健康检查端点

from fastapi import FastAPI
from datetime import datetime

app = FastAPI()

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "timestamp": datetime.utcnow().isoformat(),
        "version": "1.2.0",
        "dependencies": {
            "llm_api": await check_llm_api(),
            "vector_db": await check_vector_db(),
            "tool_service": await check_tools(),
        },
    }

@app.get("/health/detailed")
async def detailed_health():
    metrics_summary = agent_metrics.get_summary()
    return {
        "status": "healthy",
        "metrics": metrics_summary,
        "recent_errors": tool_logger.get_error_summary()[-10:],
        "budget": token_budget.get_daily_report(),
    }

10. 实战案例：构建Agent可观测平台

整合前述所有组件，构建一个完整的可观测平台：

from contextlib import contextmanager
from dataclasses import dataclass, field
import json

@dataclass
class ObservabilityConfig:
    """可观测平台配置"""
    service_name: str = "my-agent"
    langfuse_public_key: str = ""
    langfuse_secret_key: str = ""
    prometheus_port: int = 8000
    log_level: str = "INFO"
    enable_tracing: bool = True
    enable_metrics: bool = True
    token_budget_daily_usd: float = 50.0
    alert_webhook_url: str = ""

class AgentObservabilityPlatform:
    """Agent可观测平台 - 统一接入层"""

    def __init__(self, config: ObservabilityConfig):
        self.config = config
        self.logger = StructuredLogger(config.service_name)
        self.metrics = AgentMetrics()
        self.tracer = Tracer()
        self.token_budget = TokenBudget(config.token_budget_daily_usd)
        self.tool_logger = ToolCallLogger()
        self.prompt_registry = PromptRegistry()

        if config.enable_metrics:
            start_http_server(config.prometheus_port)

    @contextmanager
    def trace_agent_run(self, user_input: str, user_id: str = None):
        """追踪一次完整的Agent执行"""
        trace_id = str(uuid.uuid4())[:12]

        with self.tracer.start_span("agent.run") as root_span:
            root_span.attributes.update({
                "trace_id": trace_id,
                "user_id": user_id,
                "input": user_input[:200],
            })

            self.metrics.increment("agent.requests.total")
            start = time.time()

            try:
                yield {
                    "trace_id": trace_id,
                    "span": root_span,
                    "logger": self.logger,
                    "metrics": self.metrics,
                    "token_budget": self.token_budget,
                    "tool_logger": self.tool_logger,
                }
                root_span.attributes["status"] = "success"
                self.metrics.increment("agent.requests.success")

            except Exception as e:
                root_span.status = "error"
                root_span.attributes["error"] = str(e)
                self.metrics.increment("agent.requests.error")
                self._send_alert(f"Agent执行失败: {e}")
                raise

            finally:
                duration = (time.time() - start) * 1000
                self.metrics.record_duration("agent.run.duration_ms", duration)

    def get_dashboard_data(self) -> dict:
        """获取监控面板数据"""
        return {
            "metrics": self.metrics.get_summary(),
            "token_budget": self.token_budget.get_daily_report(),
            "recent_errors": self.tool_logger.get_error_summary()[-20:],
            "prompt_versions": {
                name: len(versions)
                for name, versions in self.prompt_registry.prompts.items()
            },
        }

    def _send_alert(self, message: str):
        """发送告警通知"""
        if self.config.alert_webhook_url:
            # 实际生产中发送到Slack/钉钉/PagerDuty
            print(f"🚨 ALERT: {message}")

完整的使用流程：

# 初始化平台
config = ObservabilityConfig(
    service_name="customer-support-agent",
    token_budget_daily_usd=100.0,
    prometheus_port=8000,
)
platform = AgentObservabilityPlatform(config)

# 注册prompt版本
platform.prompt_registry.register(
    "system_prompt",
    "你是客服助手。基于知识库回答问题，不确定时说'我不确定'。",
)

# Agent执行（带完整观测）
def handle_request(user_input: str, user_id: str):
    with platform.trace_agent_run(user_input, user_id) as ctx:
        # 此处执行实际Agent逻辑
        # ctx中包含logger、metrics、token_budget等
        result = agent.run(user_input)

        # 记录token使用
        ctx["token_budget"].record(
            model="gpt-4o",
            input_tokens=1000,
            output_tokens=500,
            trace_id=ctx["trace_id"],
            user_id=user_id,
        )
        return result

# 查看仪表板
dashboard = platform.get_dashboard_data()
print(json.dumps(dashboard, indent=2))

11. 最佳实践与常见陷阱

最佳实践

1. 从第一天就接入可观测性：不要等到出了问题才想起加日志。在Agent开发初期就集成LangFuse或LangSmith，零成本获得全链路追踪
2. 结构化一切：所有日志、指标、追踪数据使用结构化格式（JSON），方便后续查询和分析
3. 为每次执行生成唯一trace_id：贯穿整个请求生命周期，从用户输入到最终输出，所有中间步骤共享同一个trace_id
4. 设置token预算硬上限：在代码层面设置每日/每用户的token消耗上限，超出后直接拒绝或降级服务
5. 工具调用设置超时：每个工具调用必须有超时控制，避免单个工具卡死导致整个Agent挂起
6. 保留prompt历史版本：每次修改prompt都记录版本，方便回滚和对比评估结果
7. 自动化回归测试：每次prompt或模型变更后，自动运行基准测试集，量化变化的影响
8. 采样率控制：生产环境中全量采集追踪数据成本过高，建议对正常请求采样10%，对错误请求采样100%

常见陷阱

1. 日志过多导致成本爆炸：追踪数据会占用大量存储。对LLM的输入输出做截断（如限制500字符），对高频事件做采样
2. 只监控不告警：收集了大量指标但没有配置告警规则，数据躺在那里无人看。关键指标必须配告警
3. 忽略延迟的长尾分布：平均延迟看起来正常，但P99可能很差。关注P50/P95/P99而非平均值
4. 追踪数据泄露敏感信息：日志中包含用户个人信息、API密钥等。必须在日志层做脱敏处理
5. 过度依赖单一工具：LangSmith好用但它是SaaS服务。如果服务宕机或网络不通，你的可观测性就归零。考虑多工具冗余
6. 忽视Agent循环检测：Agent可能陷入"调用工具→结果不满意→再调用"的死循环。必须设置最大迭代次数，并监控迭代次数分布
7. prompt和代码混在一起管理：prompt应该独立于代码进行版本管理和部署。将prompt存储在数据库或配置中心，而非硬编码在代码中

调试清单

当Agent行为异常时，按以下顺序排查：

1. 查看trace日志，确认执行路径是否符合预期
2. 检查每一步的LLM输入/输出，确认推理是否正确
3. 检查工具调用的参数和返回值，确认工具是否正常工作
4. 检查token使用量，确认是否因上下文截断导致信息丢失
5. 检查prompt版本，确认是否有近期变更
6. 运行基准测试集，确认问题是个例还是系统性退化
7. 对比正常和异常case的trace，找出差异点

这套可观测性体系不是一次性搭建就完事的——它需要随着Agent能力的增长持续迭代。从最简单的结构化日志开始，逐步增加指标、追踪、告警，最终形成完整的可观测平台。