Agent可观测性：你无法优化你看不见的东西
传统软件的可观测性已经相当成熟——我们有Prometheus做指标、ELK做日志、Jaeger做追踪。但Agent系统引入了新的可观测性挑战：非确定性的执行路径、不可预测的token消耗、LLM输出的质量评估、多Agent协作的链路追踪。2026年，Agent可观测性已经形成了一套独立的最佳实践。
Agent可观测性的三大支柱
┌───────────────────────────────────────────────────┐
│ Agent 可观测性三大支柱 │
├─────────────┬─────────────┬───────────────────────┤
│ │ │ │
│ 追踪 │ 日志 │ 指标 │
│ (Traces) │ (Logs) │ (Metrics) │
│ │ │ │
│ Agent执行 │ 结构化事件 │ 性能计数器 │
│ 链路追踪 │ 决策记录 │ 资源消耗 │
│ 跨Agent关联 │ 上下文快照 │ 质量评估 │
│ │ │ │
└─────────────┴─────────────┴───────────────────────┘
1. 分布式追踪
Agent Trace模型
Agent系统的追踪比传统微服务更复杂，因为一个"请求"可能涉及多轮LLM调用、多次工具调用、跨多个Agent。
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional
import uuid
@dataclass
class AgentSpan:
"""Agent追踪的基本单元"""
span_id: str = field(default_factory=lambda: str(uuid.uuid4()))
parent_id: Optional[str] = None
trace_id: str = ""
# 基本信息
name: str = "" # span名称
span_type: str = "" # llm_call / tool_call / agent_step / sub_agent
agent_name: str = "" # 哪个Agent
start_time: datetime = field(default_factory=datetime.now)
end_time: Optional[datetime] = None
# Agent特有信息
input_data: Any = None
output_data: Any = None
model: str = "" # 使用的LLM模型
prompt_tokens: int = 0
completion_tokens: int = 0
cost: float = 0.0
# 状态
status: str = "ok" # ok / error / timeout
error_message: Optional[str] = None
# 元数据
attributes: dict = field(default_factory=dict)
def finish(self, output=None, status="ok", error=None):
self.end_time = datetime.now()
self.output_data = output
self.status = status
self.error_message = error
@property
def duration_ms(self) -> float:
if self.end_time:
return (self.end_time - self.start_time).total_seconds() * 1000
return 0
追踪实现
class AgentTracer:
"""Agent追踪器"""
def __init__(self, service_name="agent-service"):
self.service_name = service_name
self.spans: list[AgentSpan] = []
self.current_span_stack: list[AgentSpan] = []
def start_trace(self, name: str, agent_name: str) -> AgentSpan:
"""开始一个新的追踪（根span）"""
trace_id = str(uuid.uuid4())
span = AgentSpan(
trace_id=trace_id,
name=name,
span_type="agent_step",
agent_name=agent_name
)
self.spans.append(span)
self.current_span_stack.append(span)
return span
def start_span(
self,
name: str,
span_type: str,
agent_name: str = "",
input_data: Any = None
) -> AgentSpan:
"""开始一个子span"""
parent = self.current_span_stack[-1] if self.current_span_stack else None
span = AgentSpan(
trace_id=parent.trace_id if parent else str(uuid.uuid4()),
parent_id=parent.span_id if parent else None,
name=name,
span_type=span_type,
agent_name=agent_name,
input_data=input_data
)
self.spans.append(span)
self.current_span_stack.append(span)
return span
def end_span(self, span: AgentSpan, output=None, status="ok", error=None):
"""结束一个span"""
span.finish(output, status, error)
if self.current_span_stack and self.current_span_stack[-1].span_id == span.span_id:
self.current_span_stack.pop()
def get_trace_tree(self, trace_id: str) -> dict:
"""获取追踪树"""
trace_spans = [s for s in self.spans if s.trace_id == trace_id]
return self._build_tree(trace_spans, parent_id=None)
def _build_tree(self, spans: list[AgentSpan], parent_id: str | None) -> dict:
children = [s for s in spans if s.parent_id == parent_id]
return [
{
"span_id": s.span_id,
"name": s.name,
"type": s.span_type,
"agent": s.agent_name,
"duration_ms": s.duration_ms,
"tokens": s.prompt_tokens + s.completion_tokens,
"cost": s.cost,
"status": s.status,
"children": self._build_tree(spans, s.span_id)
}
for s in children
]
使用示例
tracer = AgentTracer()
# 开始追踪
root = tracer.start_trace("用户咨询", "router_agent")
# Agent执行LLM调用
llm_span = tracer.start_span("LLM调用", "llm_call", "router_agent", input_data="用户问题")
response = await llm.complete("...")
tracer.end_span(llm_span, output=response.text,
status="ok")
llm_span.prompt_tokens = response.usage.prompt_tokens
llm_span.completion_tokens = response.usage.completion_tokens
llm_span.cost = calculate_cost(response.usage, "gpt-4o")
# Agent调用工具
tool_span = tracer.start_span("搜索知识库", "tool_call", "router_agent")
results = await knowledge_base.search("query")
tracer.end_span(tool_span, output=results)
# 路由到子Agent
sub_agent_span = tracer.start_span("专家Agent处理", "sub_agent", "expert_agent")
# ... 子Agent内部会有自己的span ...
tracer.end_span(sub_agent_span, output="最终回答")
# 结束追踪
tracer.end_span(root, output="最终回答")
# 查看追踪树
trace_tree = tracer.get_trace_tree(root.trace_id)
追踪可视化
追踪树可以渲染为瀑布图：
[router_agent] 用户咨询 2850ms 1520 tok ¥0.08
├── [router_agent] LLM调用 (意图识别) 320ms 180 tok ¥0.01
├── [router_agent] 搜索知识库 85ms 0 tok ¥0.00
├── [expert_agent] 专家Agent处理 2100ms 1240 tok ¥0.06
│ ├── [expert_agent] LLM调用 (分析) 450ms 320 tok ¥0.02
│ ├── [expert_agent] 工具调用 (数据查询) 180ms 0 tok ¥0.00
│ ├── [expert_agent] LLM调用 (生成回答) 680ms 540 tok ¥0.03
│ └── [expert_agent] LLM调用 (审查) 350ms 380 tok ¥0.02
└── [router_agent] LLM调用 (格式化输出) 280ms 200 tok ¥0.01
2. 结构化日志
日志格式
import json
import logging
class AgentLogger:
"""Agent专用结构化日志"""
def __init__(self, agent_name: str):
self.agent_name = agent_name
self.logger = logging.getLogger(agent_name)
def log(self, level: str, event: str, **kwargs):
"""结构化日志输出"""
log_entry = {
"timestamp": datetime.now().isoformat(),
"level": level,
"agent": self.agent_name,
"event": event,
**kwargs
}
self.logger.log(
getattr(logging, level.upper()),
json.dumps(log_entry, ensure_ascii=False, default=str)
)
def log_llm_call(self, model: str, prompt: str, response: str,
tokens: dict, latency: float):
self.log("INFO", "llm_call",
model=model,
prompt_length=len(prompt),
response_length=len(response),
prompt_tokens=tokens.get("prompt", 0),
completion_tokens=tokens.get("completion", 0),
latency_ms=latency
)
def log_tool_call(self, tool_name: str, args: dict, result: Any,
latency: float, success: bool):
self.log("INFO", "tool_call",
tool=tool_name,
args=args,
result_preview=str(result)[:200],
latency_ms=latency,
success=success
)
def log_decision(self, decision: str, reasoning: str,
alternatives: list, confidence: float):
self.log("INFO", "decision",
decision=decision,
reasoning=reasoning,
alternatives=alternatives,
confidence=confidence
)
def log_error(self, error_type: str, message: str, context: dict):
self.log("ERROR", "error",
error_type=error_type,
message=message,
context=context
)
3. 指标监控
核心指标定义
from dataclasses import dataclass
from collections import defaultdict
import time
@dataclass
class AgentMetrics:
"""Agent核心指标"""
# 性能指标
request_count: int = 0
success_count: int = 0
error_count: int = 0
timeout_count: int = 0
# 延迟指标
latencies: list[float] = None # ms
# Token指标
total_prompt_tokens: int = 0
total_completion_tokens: int = 0
# 成本指标
total_cost: float = 0.0
# 质量指标
user_feedback_scores: list[float] = None
# Agent特有指标
avg_iterations: float = 0.0 # 平均迭代次数
tool_call_count: int = 0 # 工具调用总次数
tool_success_rate: float = 0.0 # 工具成功率
escalation_count: int = 0 # 升级人工次数
def summary(self) -> dict:
return {
"total_requests": self.request_count,
"success_rate": self.success_count / max(1, self.request_count),
"error_rate": self.error_count / max(1, self.request_count),
"p50_latency_ms": self._percentile(self.latencies, 50),
"p95_latency_ms": self._percentile(self.latencies, 95),
"p99_latency_ms": self._percentile(self.latencies, 99),
"avg_tokens_per_request": (
self.total_prompt_tokens + self.total_completion_tokens
) / max(1, self.request_count),
"avg_cost_per_request": self.total_cost / max(1, self.request_count),
"avg_user_score": sum(self.user_feedback_scores) / max(1, len(self.user_feedback_scores)),
"tool_success_rate": self.tool_success_rate,
"escalation_rate": self.escalation_count / max(1, self.request_count)
}
@staticmethod
def _percentile(data: list[float], p: int) -> float:
if not data:
return 0
sorted_data = sorted(data)
idx = int(len(sorted_data) * p / 100)
return sorted_data[min(idx, len(sorted_data) - 1)]
关键指标仪表盘
┌─────────────────────────────────────────────────────┐
│ Agent 监控仪表盘 │
├─────────────────┬─────────────────┬─────────────────┤
│ 请求总数 │ 成功率 │ P95延迟 │
│ 12,847 │ 94.2% │ 2.1s │
│ ↑ 8.3% vs昨 │ ↑ 1.2% vs昨 │ ↓ 15% vs昨 │
├─────────────────┼─────────────────┼─────────────────┤
│ Token/请求 │ 成本/请求 │ 用户评分 │
│ 18,200 │ ¥0.42 │ 4.3/5.0 │
│ ↓ 5% vs昨 │ ↓ 7% vs昨 │ ↑ 0.2 vs昨 │
├─────────────────┼─────────────────┼─────────────────┤
│ 工具成功率 │ 升级人工率 │ 错误率 │
│ 91.5% │ 2.7% │ 3.1% │
│ → 持平 │ ↓ 0.3% vs昨 │ → 持平 │
└─────────────────┴─────────────────┴─────────────────┘
异常检测规则
class AnomalyDetector:
"""Agent异常检测"""
RULES = [
# 延迟异常
{"metric": "latency_p95", "threshold": 5000, "window": "5m",
"action": "alert", "message": "P95延迟超过5秒"},
# 错误率飙升
{"metric": "error_rate", "threshold": 0.10, "window": "5m",
"action": "alert", "message": "错误率超过10%"},
# Token消耗异常
{"metric": "avg_tokens", "threshold": 50000, "window": "1h",
"action": "alert", "message": "平均token消耗异常偏高"},
# 成本异常
{"metric": "cost_per_hour", "threshold": 100, "window": "1h",
"action": "alert", "message": "每小时成本超过¥100"},
# 成功率下降
{"metric": "success_rate", "threshold": 0.85, "window": "15m",
"action": "alert", "message": "成功率低于85%"},
# 工具调用失败率
{"metric": "tool_error_rate", "threshold": 0.15, "window": "10m",
"action": "warn", "message": "工具调用错误率超过15%"},
]
def check(self, metrics: dict) -> list[dict]:
alerts = []
for rule in self.RULES:
value = metrics.get(rule["metric"], 0)
threshold = rule["threshold"]
if rule["metric"] in ["success_rate", "tool_success_rate"]:
# 这些指标是越低越异常
if value < threshold:
alerts.append(rule)
else:
# 这些指标是越高越异常
if value > threshold:
alerts.append(rule)
return alerts
4. OpenTelemetry集成
2026年，OpenTelemetry已经成为Agent可观测性的事实标准：
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
# 配置OpenTelemetry
provider = TracerProvider()
provider.add_span_processor(
BatchSpanProcessor(
OTLPSpanExporter(endpoint="http://otel-collector:4317")
)
)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agent-service")
# 在Agent代码中使用
class TracedAgent:
def __init__(self, name: str):
self.name = name
self.tracer = trace.get_tracer(f"agent.{name}")
async def run(self, task: str) -> str:
with self.tracer.start_as_current_span(f"agent.{self.name}.run") as span:
span.set_attribute("agent.name", self.name)
span.set_attribute("agent.task", task)
# LLM调用
with self.tracer.start_as_current_span("llm.call"):
response = await self.llm.complete(task)
span.set_attribute("llm.tokens", response.usage.total_tokens)
span.set_attribute("llm.model", self.model)
# 工具调用
with self.tracer.start_as_current_span("tool.search"):
results = await self.tool.search(task)
span.set_attribute("tool.results_count", len(results))
return response
5. 质量评估
除了技术指标，Agent系统的输出质量也需要监控：
class QualityMonitor:
"""Agent输出质量监控"""
async def evaluate_response(self, query: str, response: str) -> dict:
"""使用LLM评估响应质量"""
eval_prompt = f"""评估以下AI响应的质量，给出1-5分：
用户问题: {query}
AI回答: {response}
评分维度:
1. 准确性: 回答是否正确？
2. 完整性: 是否完整回答了问题？
3. 相关性: 回答是否切题？
4. 清晰度: 表达是否清晰？
5. 安全性: 是否包含不当内容？
"""
result = await self.eval_llm.complete(eval_prompt)
return self._parse_scores(result.text)
async def detect_hallucination(self, response: str, sources: list[str]) -> float:
"""检测幻觉概率"""
# 检查response中的事实声明是否有sources支撑
prompt = f"""判断以下回答中的事实是否有来源支撑：
回答: {response}
来源: {sources}
返回幻觉概率（0-1），1表示完全虚构。
"""
result = await self.eval_llm.complete(prompt)
return float(result.text.strip())
结论
Agent可观测性不是锦上添花，而是生产部署的必要条件。一个不可观测的Agent系统就像一个黑箱——出了问题不知道为什么，性能下降不知道哪里瓶颈，成本飙升不知道哪个环节。
核心建议：
- 追踪先行：先实现完整追踪，再考虑日志和指标
- 结构化日志：所有日志都是JSON格式，可机器解析
- 关键指标看板：成功率、延迟、token消耗、成本是四大核心指标
- 异常告警：设置合理的告警阈值，避免告警疲劳
- 质量监控：技术指标之外，也要监控输出质量
可观测性的投入回报率极高——在100次故障中，有可观测性系统的平均故障恢复时间（MTTR）是没有的1/5。
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。