可观测性与调试 - Agent 运行的透明度保障 / Observability and Debugging for Transparent Agent Operations
📅 创建时间：2026-05-08 🏷️ 标签：#Observability #OpenTelemetry #调试 #追踪 #日志 📚 前置知识：[[12-提示词注入防护]]
📋 本章目标
- 理解 Agent 可观测性的三大支柱（Logs / Metrics / Traces）
- 掌握 Agent 执行链路追踪的方法
- 理解 Token 使用量和成本的监控
- 掌握上下文窗口健康度的监控
- 能够为 Agent 构建完整的可观测性体系
第1部分：为什么 Agent 需要可观测性？
1.1 Agent 的调试困难
┌─────────────────────────────────────────────────────────────┐
│ Agent 调试 vs 普通代码调试 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 普通代码： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 确定性执行 │ │
│ │ 相同输入 → 相同输出 │ │
│ │ 断点调试、变量检查、易于复现 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Agent： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 非确定性行为 │ │
│ │ 相同输入 → 可能不同输出（随机性、上下文变化） │ │
│ │ LLM 内部是黑盒 │ │
│ │ 执行路径可能每次都不同 │ │
│ │ 失败可能是偶发的（概率性） │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Agent 需要"事后诸葛亮"式的调试： │
│ 不是边跑边看，而是跑完了能回放和分析 │
│ │
└─────────────────────────────────────────────────────────────┘1.2 可观测性三大支柱
┌─────────────────────────────────────────────────────────────┐
│ 可观测性三大支柱 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Logs（日志） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ "发生了什么" │ │
│ │ 时间戳 + 事件 + 上下文 │ │
│ │ Agent 的每一步操作、每次决策、每个错误 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Metrics（指标） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ "表现如何" │ │
│ │ Token 消耗、响应延迟、成功率、成本 │ │
│ │ 可以聚合、告警、可视化 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Traces（链路追踪） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ "怎么发生的" │ │
│ │ 一次请求的完整执行链路 │ │
│ │ 工具调用 → LLM 推理 → 工具调用 → ... │ │
│ │ 从头到尾的因果链 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘第2部分：链路追踪（Traces）
2.1 Agent 执行链路结构
┌─────────────────────────────────────────────────────────────┐
│ Agent 链路追踪结构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Session: sess_abc123 │
│ User Message: "帮我优化这个函数的性能" │
│ │ │
│ ├── Span: LLM Call #1 │
│ │ ├── Input Tokens: 1,234 │
│ │ ├── Output Tokens: 567 │
│ │ └── Model: claude-opus-4 │
│ │ │ │
│ │ ├── Span: Tool: read_file │
│ │ │ └── Result: [文件内容...] │
│ │ │ │
│ │ ├── Span: Tool: grep │
│ │ │ └── Result: [搜索结果...] │
│ │ │ │
│ │ ├── Span: Tool: bash (执行代码) │
│ │ │ ├── Command: python optimize.py │
│ │ │ └── Result: 性能提升 40% │
│ │ │ │
│ │ └── Span: Tool: edit (写入修改) │
│ │ └── Changes: 3 files modified │
│ │ │
│ ├── Span: LLM Call #2 (结果汇总) │
│ │ └── Output: "优化完成..." │
│ │ │
│ └── Span: Response Delivered │
│ └── Duration: 12.3s │
│ │
└─────────────────────────────────────────────────────────────┘2.2 OpenTelemetry 集成
python
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
from opentelemetry.sdk.resources import Resource
from opentelemetry.semconv.resource import ResourceAttributes
# 初始化追踪
provider = TracerProvider(
resource=Resource.create({
ResourceAttributes.SERVICE_NAME: "claude-code-agent",
ResourceAttributes.SERVICE_VERSION: "2.1.88",
ResourceAttributes.DEPLOYMENT_ENVIRONMENT: "production",
})
)
# 导出到 Jaeger / Grafana / Datadog
processor = BatchSpanProcessor(
OTLPSpanExporter(endpoint="http://otel-collector:4317")
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)
class ObservableAgent:
"""可观测的 Agent"""
def __init__(self, agent):
self.agent = agent
self.tracer = trace.get_tracer(__name__)
def run(self, user_input: str, session_id: str):
with self.tracer.start_as_current_span(
"agent.run",
attributes={
"session.id": session_id,
"user.input.length": len(user_input),
"user.input.preview": user_input[:100],
}
) as root_span:
# 记录 LLM 调用
with self.tracer.start_as_current_span("llm.call.1") as llm_span:
response = self.agent.call_llm(user_input)
llm_span.set_attribute("llm.model", response.model)
llm_span.set_attribute("llm.input_tokens", response.usage.input_tokens)
llm_span.set_attribute("llm.output_tokens", response.usage.output_tokens)
# 记录工具调用
for tool_call in response.tool_calls:
with self.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))
result = self.agent.execute_tool(tool_call)
tool_span.set_attribute("tool.result.length", len(str(result)))
tool_span.set_attribute("tool.duration_ms", result.duration_ms)
return response.final_output第3部分：关键指标监控
3.1 Agent 核心指标
┌─────────────────────────────────────────────────────────────┐
│ Agent 核心监控指标 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 性能指标： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ • 首次响应时间（TTFT）：用户发消息到收到第一条回复 │ │
│ │ • 端到端延迟：用户发消息到 Agent 完成 │ │
│ │ • 工具调用延迟：每个工具的执行时间 │ │
│ │ • LLM 推理延迟：API 调用的响应时间 │ │
│ │ • 每步延迟：Agent 的每轮循环耗时 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 成本指标： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ • 输入 Token 消耗（$ / M tokens） │ │
│ │ • 输出 Token 消耗（$ / M tokens） │ │
│ │ • 单次会话总成本 │ │
│ │ • 按用户/按任务的成本分布 │ │
│ │ • Token 利用率（实际使用 / 上下文窗口） │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 质量指标： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ • 任务成功率（完成任务 / 总任务） │ │
│ │ • 工具调用成功率（成功 / 总调用） │ │
│ │ • 错误率（各类型错误分布） │ │
│ │ • 用户满意度 / 反馈 │ │
│ │ • 上下文重置频率（Agent 失败后重试） │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘3.2 指标采集与告警
python
from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry
# 创建指标注册表
registry = CollectorRegistry()
# 计数器
llm_calls_total = Counter(
"agent_llm_calls_total",
"LLM 调用总次数",
["model", "status"],
registry=registry
)
tool_calls_total = Counter(
"agent_tool_calls_total",
"工具调用总次数",
["tool_name", "status"],
registry=registry
)
# 直方图
request_duration = Histogram(
"agent_request_duration_seconds",
"请求持续时间",
["endpoint", "method"],
buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0],
registry=registry
)
token_usage = Histogram(
"agent_token_usage",
"Token 消耗量",
["type"], # input / output
buckets=[100, 500, 1000, 5000, 10000, 50000],
registry=registry
)
# 仪表盘
active_sessions = Gauge(
"agent_active_sessions",
"活跃会话数",
registry=registry
)
context_health = Gauge(
"agent_context_health_percent",
"上下文健康度（剩余空间百分比）",
["session_id"],
registry=registry
)
# 使用示例
def track_llm_call(model: str, status: str, input_tokens: int, output_tokens: int):
llm_calls_total.labels(model=model, status=status).inc()
token_usage.labels(type="input").observe(input_tokens)
token_usage.labels(type="output").observe(output_tokens)
def track_tool_call(tool_name: str, status: str, duration_ms: float):
tool_calls_total.labels(tool_name=tool_name, status=status).inc()
request_duration.labels(endpoint=f"tool.{tool_name}", method="EXECUTE").observe(duration_ms / 1000)
# 告警规则（Prometheus AlertManager）
ALERT_RULES = """
groups:
- name: agent_alerts
rules:
- alert: HighTokenUsage
expr: rate(agent_token_usage_total[5m]) > 100000
for: 5m
annotations:
summary: "Token 使用率异常高"
- alert: HighErrorRate
expr: rate(agent_llm_calls_total{status="error"}[5m]) / rate(agent_llm_calls_total[5m]) > 0.05
for: 5m
annotations:
summary: "错误率超过 5%"
- alert: ToolCallFailure
expr: rate(agent_tool_calls_total{status="error"}[5m]) > 10
for: 2m
annotations:
summary: "工具调用失败率上升"
"""第4部分：会话追踪与回放
4.1 会话持久化
python
from datetime import datetime
from pathlib import Path
import json
class SessionRecorder:
"""会话录制器——记录 Agent 执行的完整轨迹"""
def __init__(self, session_dir: str = "~/.agent/sessions"):
self.session_dir = Path(session_dir).expanduser()
self.session_dir.mkdir(parents=True, exist_ok=True)
def start_session(self, session_id: str, metadata: dict) -> Path:
"""开始新会话"""
session_file = self.session_dir / f"{session_id}.jsonl"
metadata_record = {
"type": "session_start",
"session_id": session_id,
"timestamp": datetime.now().isoformat(),
**metadata
}
self._append(session_file, metadata_record)
return session_file
def record_turn(self, session_file: Path, turn: dict):
"""记录一个对话轮次"""
record = {
"type": "turn",
"timestamp": datetime.now().isoformat(),
**turn
}
self._append(session_file, record)
def record_span(self, session_file: Path, span: dict):
"""记录一个执行 span"""
record = {
"type": "span",
"timestamp": datetime.now().isoformat(),
**span
}
self._append(session_file, record)
def record_metric(self, session_file: Path, metric: dict):
"""记录一个指标采样"""
record = {
"type": "metric",
"timestamp": datetime.now().isoformat(),
**metric
}
self._append(session_file, record)
def _append(self, file: Path, record: dict):
"""追加写入 JSONL"""
with file.open("a") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
def replay_session(self, session_id: str) -> list[dict]:
"""回放会话"""
session_file = self.session_dir / f"{session_id}.jsonl"
if not session_file.exists():
return []
records = []
with session_file.open() as f:
for line in f:
records.append(json.loads(line))
return records4.2 调试工具
python
class AgentDebugger:
"""Agent 调试器"""
def __init__(self, session_recorder: SessionRecorder):
self.recorder = session_recorder
def analyze_session(self, session_id: str) -> dict:
"""分析会话，输出调试报告"""
records = self.recorder.replay_session(session_id)
# 统计
llm_calls = [r for r in records if r["type"] == "turn"]
tool_calls = [r for r in records if r["type"] == "span" and "tool" in r.get("name", "")]
# 计算总消耗
total_input_tokens = sum(r.get("input_tokens", 0) for r in llm_calls)
total_output_tokens = sum(r.get("output_tokens", 0) for r in llm_calls)
# 分析耗时
durations = [r.get("duration_ms", 0) for r in tool_calls]
slow_tools = [(r["name"], r["duration_ms"]) for r in tool_calls if r.get("duration_ms", 0) > 5000]
# 分析错误
errors = [r for r in records if r.get("status") == "error"]
# 分析上下文使用
context_usage = [r.get("context_tokens", 0) for r in llm_calls if "context_tokens" in r]
return {
"session_id": session_id,
"total_turns": len(llm_calls),
"total_tool_calls": len(tool_calls),
"total_input_tokens": total_input_tokens,
"total_output_tokens": total_output_tokens,
"estimated_cost": self._estimate_cost(total_input_tokens, total_output_tokens),
"slow_tools": slow_tools,
"error_count": len(errors),
"errors": errors[-5:], # 最近 5 个错误
"max_context_usage": max(context_usage) if context_usage else 0,
"avg_turn_duration_ms": sum(durations) / len(durations) if durations else 0,
}
def _estimate_cost(self, input_tokens: int, output_tokens: int) -> float:
"""估算成本（Claude 3.5 Sonnet 费率）"""
input_cost = input_tokens / 1_000_000 * 3.0 # $3 / M input
output_cost = output_tokens / 1_000_000 * 15.0 # $15 / M output
return input_cost + output_cost第5部分：上下文窗口健康监控
5.1 上下文健康指标
┌─────────────────────────────────────────────────────────────┐
│ 上下文健康监控 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Claude 3.5 Sonnet 上下文窗口：200,000 tokens │
│ │
│ 健康区间（0-60%）：绿色 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 0% ───────────|──────────|──────────|────────── 100% │
│ │ ↑ ↑ ↑ │
│ │ 60% 80% 90% │
│ │ ↓ ↓ ↓ │
│ │ [绿色] [黄色] [红色] │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 上下文使用 > 80%：触发压缩提醒 │
│ 上下文使用 > 90%：强制执行压缩 │
│ 上下文使用 = 100%：截断，后续消息无法处理 │
│ │
└─────────────────────────────────────────────────────────────┘5.2 上下文监控实现
python
from dataclasses import dataclass
from datetime import datetime
@dataclass
class ContextHealthStatus:
session_id: str
current_tokens: int
max_tokens: int
usage_percent: float
health_level: str # "healthy" / "warning" / "critical"
recommended_action: str
class ContextHealthMonitor:
"""上下文健康监控"""
def __init__(self, max_tokens: int = 200_000):
self.max_tokens = max_tokens
self.warning_threshold = 0.8 # 80% 警告
self.critical_threshold = 0.9 # 90% 危险
def check(self, session_id: str, current_tokens: int) -> ContextHealthStatus:
"""检查上下文健康状态"""
usage_percent = current_tokens / self.max_tokens
if usage_percent < self.warning_threshold:
health_level = "healthy"
action = "无需操作"
elif usage_percent < self.critical_threshold:
health_level = "warning"
action = "建议执行上下文压缩"
else:
health_level = "critical"
action = "必须执行压缩，否则即将截断"
# 记录指标
self.record_metric(session_id, current_tokens, usage_percent)
return ContextHealthStatus(
session_id=session_id,
current_tokens=current_tokens,
max_tokens=self.max_tokens,
usage_percent=usage_percent,
health_level=health_level,
recommended_action=action
)
def record_metric(self, session_id: str, tokens: int, usage: float):
"""记录到 Prometheus"""
context_health.labels(session_id=session_id).set(usage * 100)
if usage >= self.warning_threshold:
# 发送告警
self._send_alert(session_id, usage, tokens)
def _send_alert(self, session_id: str, usage: float, tokens: int):
"""发送上下文告警"""
# 接入告警系统
alert_message = (
f"[上下文告警] Session {session_id}: "
f"使用 {tokens:,} tokens ({usage:.1%})"
)
print(f"ALERT: {alert_message}")第6部分：Claude Code 的可观测性实践
┌─────────────────────────────────────────────────────────────┐
│ Claude Code 可观测性亮点 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 全链路 OpenTelemetry 集成： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 每一个 LLM 调用、工具执行、文件操作 │ │
│ │ 都有对应的 trace span │ │
│ │ 可以导入 Grafana / Jaeger / Datadog 查看 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 详细的工具执行日志： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Tool: bash │ │
│ │ Command: python optimize.py │ │
│ │ Duration: 1,234ms │ │
│ │ Exit Code: 0 │ │
│ │ Output Length: 456 bytes │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Token 使用透明： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ /usage 命令显示实时 Token 消耗 │ │
│ │ 每次 LLM 调用记录输入/输出 Token 数 │ │
│ │ 会话结束汇总总消耗 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 自动压缩的可见性： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ autocompact 日志清晰显示： │ │
│ │ • 压缩前 Token 数 │ │
│ │ • 压缩后 Token 数 │ │
│ │ • 保留的关键信息 │ │
│ │ • 丢弃的信息 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘核心总结
总结1：可观测性三大支柱
Logs（日志） → 发生了什么
Metrics（指标）→ 表现如何
Traces（追踪）→ 怎么发生的总结2：Agent 关键监控指标
性能：TTFT、端到端延迟、每步耗时
成本：输入/输出 Token、总成本
质量：成功率、错误率、用户满意度
健康：上下文使用率总结3：会话录制价值
JSONL 格式：每行一条记录，易于追加和解析
完整回放：从头到尾还原执行过程
统计分析：定位慢工具、错误模式、成本热点章节测试
测试1：可观测性支柱
可观测性的三大支柱是什么？各自回答什么问题？
测试2：链路追踪
在链路追踪中，"span"是什么概念？
测试3：上下文健康
上下文使用率超过多少时应触发压缩？
测试4：Token 监控
以下哪个是 Agent 需要特别监控的独特指标？ A. HTTP 响应码 B. 数据库连接数 C. Token 消耗量和上下文使用率 D. CPU 利用率
测试5：会话录制
Claude Code 使用什么格式录制会话？
参考答案
测试1答案
答案：
- Logs（日志）：发生了什么事件（时间戳 + 事件 + 上下文）
- Metrics（指标）：表现如何（可聚合的数值：延迟、成本、成功率）
- Traces（链路追踪）：怎么发生的（一次请求的完整执行链路）
测试2答案
答案：Span（跨度）是链路追踪中的基本单元，代表一次操作或一个调用。Span 可以嵌套（父 span 包含子 span），记录该操作的开始时间、结束时间、属性（如工具名、输入输出大小等）。
测试3答案
答案：超过 80% 时应触发压缩警告，超过 90% 时必须压缩，接近 100% 时会截断后续消息。
测试4答案
答案：C（Token 消耗量和上下文使用率）
解析：A、B、D 是普通服务也有的指标。Token 消耗和上下文使用率是 Agent 特有的关键指标，直接影响成本和 Agent 可用性。
测试5答案
答案：JSONL 格式（每行一条 JSON 记录）
解析：Claude Code 的会话录制使用 JSONL 格式，易于追加写入、传输，也便于后续解析和分析回放。
相关笔记
- [[05-agent-workflow]] - 工作流中的调试需求
- [[04-memory-management]] - 压缩机制影响可观测性
- [[08-real-world-applications]] - 生产环境监控的实际应用
下一步学习
- [ ] 阅读 23 - 模型路由
学习状态：🟡 开始学习