AI Agent可观测性实战：从黑盒到透明的全链路追踪指南

weixin_56622231

266人浏览 · 2026-09-05 11:35:40

weixin_56622231 · 2026-09-05 11:35:40 发布

1. 引言：为什么 Agent 需要可观测性

随着大语言模型（LLM）从单轮问答走向多步推理的 Agent 应用，系统的行为复杂度急剧上升。一个 Agent 可能经历「理解任务 → 拆解子目标 → 调用工具 → 观察结果 → 修正计划 → 输出结论」等多个环节，任何一个环节出错都可能导致最终结果偏离预期。

传统监控体系只能回答「系统是否可用」，却无法回答「Agent 为什么做出这个决策」。这正是 AI Agent 可观测性要解决的核心问题：让多步推理过程变得透明、可追踪、可审计。

2. Agent 系统的黑盒困境

2.1 多步推理的复杂性

Agent 的推理链路通常包含多个环节，每个环节都可能引入不确定性：

模型自身的推理偏差
工具调用的参数错误
外部 API 返回的异常数据
上下文窗口截断导致的信息丢失

2.2 传统监控的局限

传统 APM（应用性能监控）关注的是请求延迟、错误率、吞吐量等指标，它们能告诉你「哪里慢了、哪里错了」，却无法解释「为什么模型选择了这个工具」「为什么 Agent 在第三步改变了计划」。

2.3 黑盒带来的实际风险

线上故障难以定位根因
模型行为漂移无法及时发现
合规审计缺乏推理过程证据
调试成本随链路长度指数增长

3. 可观测性的三大支柱

3.1 指标（Metrics）

用于回答「系统整体是否健康」：

工具调用成功率
单步推理耗时分布
重试次数统计
上下文使用率

3.2 日志（Logs）

用于回答「某个时刻发生了什么」：

每次 LLM 调用的输入输出
工具调用的请求与响应
状态变更事件
错误与异常堆栈

3.3 追踪（Traces）

用于回答「一次完整请求的完整链路」：

推理步骤的父子关系
每一步的输入输出快照
决策点的分支记录
跨服务调用的串联

4. 核心追踪模型设计

4.1 追踪单元：Span

一次 Agent 运行可以拆解为多个 Span，每个 Span 代表一个原子操作：

Agent Run（根 Span）
├── Plan Generation（规划）
├── Tool Call: search_web（工具调用）
│   └── HTTP Request（外部请求）
├── Observation Processing（结果观察）
├── Plan Revision（计划修正）
└── Final Answer（最终输出）

4.2 关键属性设计

每个 Span 应携带结构化属性，便于后续检索与分析：

agent_id：Agent 实例标识
step_index：当前推理步序号
model_name：使用的模型名称与版本
prompt_tokens / completion_tokens：Token 消耗
tool_name / tool_input / tool_output：工具调用详情
decision_reason：模型给出的决策理由

4.3 关联 ID 贯穿

通过 trace_id 将一次完整请求的所有 Span 串联，通过 parent_span_id 表达嵌套关系，形成一棵完整的调用树。

5. 推理过程的可视化

5.1 时间线视图

按时间顺序展示每个推理步骤，直观呈现「先做了什么、后做了什么、每步耗时多少」。

5.2 决策树视图

以树形结构展示 Agent 的决策分支，帮助定位「在哪一步出现了计划偏离」。

5.3 输入输出对比

并排展示每一步的输入与输出，快速发现「模型是否误解了工具返回的结果」。

6. 关键埋点方案

6.1 在 Agent 框架层埋点

如果使用 LangChain、LlamaIndex 等框架，可利用其内置的回调机制统一埋点：

from langchain.callbacks import BaseCallbackHandler

class TracingCallbackHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        # 记录 LLM 调用开始
        pass

    def on_llm_end(self, response, **kwargs):
        # 记录 LLM 输出与 token 消耗
        pass

    def on_tool_start(self, serialized, input_str, **kwargs):
        # 记录工具调用参数
        pass

    def on_tool_end(self, output, **kwargs):
        # 记录工具返回结果
        pass

6.2 在工具调用层埋点

对每个工具封装统一的追踪装饰器，自动记录入参、出参、耗时与异常。

6.3 在模型调用层埋点

记录每次 LLM 请求的完整 Prompt 与 Response，便于事后回放推理过程。

7. 数据采集与存储

7.1 采样策略

全量采样：适用于低流量场景或关键业务链路
头部采样：按请求维度决定是否采集整条链路
尾部采样：根据结果状态（如错误、超时）决定是否补采

7.2 存储选型

指标数据：Prometheus + Grafana
日志数据：ELK / Loki
追踪数据：Jaeger / Tempo / 自建 ClickHouse

7.3 成本控制

LLM 调用的 Prompt/Response 体积较大，建议对完整内容做压缩存储，索引只保留关键字段。

8. 基于追踪的调试实践

8.1 定位「工具误用」

当 Agent 反复调用同一个工具却得不到正确结果时，通过追踪中的 tool_input 与 tool_output 对比，快速判断是参数构造错误还是外部服务异常。

8.2 定位「计划漂移」

当 Agent 最终输出与用户意图偏离时，回放决策树，找到「哪一步开始偏离原始计划」。

8.3 定位「上下文丢失」

当 Agent 忘记早期信息时，检查每一步的上下文窗口使用率，判断是否因 Token 超限被截断。

9. 开源工具与生态

9.1 LangSmith

LangChain 官方可观测平台，开箱即用的追踪与评估能力。

9.2 Langfuse

开源 LLM 可观测平台，支持自托管，提供完整的追踪、评估与提示词管理。

9.3 OpenTelemetry GenAI 语义约定

社区正在推进的标准化方案，让 Agent 追踪数据可以接入通用可观测体系。

10. 总结与展望

AI Agent 可观测性不是可选项，而是生产级 Agent 应用的必备能力。它让开发者能够：

快速定位多步推理中的故障根因
持续评估模型行为是否符合预期
为合规审计提供完整的推理证据链

随着 Agent 应用走向复杂化，可观测性体系也将从「能用」走向「好用」，成为 Agent 工程质量的基础设施。
—# 十一、可观测性三大支柱深入传统软件的可观测性有三大支柱：日志（Logs）、指标（Metrics）、追踪（Traces）。Agent系统也不例外，但每根支柱都有Agent特有的形态。## 11.1 日志（Logs）：Agent的"黑匣子"日志是最基础的可观测性数据，记录了系统运行时发生的离散事件。### Agent日志的特殊性传统软件日志通常记录：请求到达、处理完成、错误发生。Agent日志需要记录更多：- 用户输入的原始内容- 系统提示词（System Prompt）- 模型的每一步思考（Thought）- 模型选择的工具（Action）- 工具的输入参数（Action Input）- 工具的返回结果（Observation）- 模型的最终回答（Final Answer）- 每一步的Token使用量- 每一步的耗时### 日志级别设计| 级别 | 记录内容 | 适用场景 ||------|---------|---------|| DEBUG | 完整的推理链、工具调用细节、Token使用 | 开发调试 || INFO | 关键步骤、工具调用摘要、最终结果 | 生产环境常规 || WARN | 异常但可恢复的情况（工具重试、模型降级） | 生产环境监控 || ERROR | 不可恢复的错误（工具调用失败、模型超时） | 生产环境告警 |### 结构化日志Agent日志必须是结构化的，便于后续分析和检索。pythonimport jsonimport timefrom datetime import datetimeclass AgentLogger: def __init__(self, agent_name="default"): self.agent_name = agent_name self.trace_id = None def start_trace(self, user_input): """开始一次Agent调用追踪""" self.trace_id = f"trace_{int(time.time()*1000)}" self._log("INFO", "trace_start", { "user_input": user_input, "timestamp": datetime.now().isoformat() }) return self.trace_id def log_thought(self, step, thought): """记录模型思考""" self._log("DEBUG", "thought", { "step": step, "thought": thought, "timestamp": datetime.now().isoformat() }) def log_tool_call(self, step, tool_name, tool_input, duration_ms, token_usage=None): """记录工具调用""" self._log("INFO", "tool_call", { "step": step, "tool_name": tool_name, "tool_input": tool_input, "duration_ms": duration_ms, "token_usage": token_usage, "timestamp": datetime.now().isoformat() }) def log_tool_result(self, step, tool_name, result, success=True): """记录工具返回结果""" self._log("DEBUG" if success else "WARN", "tool_result", { "step": step, "tool_name": tool_name, "success": success, "result_preview": str(result)[:500], # 只记录前500字符 "timestamp": datetime.now().isoformat() }) def log_final_answer(self, answer, total_duration_ms, total_token_usage): """记录最终回答""" self._log("INFO", "final_answer", { "answer": answer, "total_duration_ms": total_duration_ms, "total_token_usage": total_token_usage, "timestamp": datetime.now().isoformat() }) def log_error(self, error, step=None): """记录错误""" self._log("ERROR", "error", { "error": str(error), "step": step, "timestamp": datetime.now().isoformat() }) def _log(self, level, event_type, data): """输出结构化日志""" log_entry = { "trace_id": self.trace_id, "agent_name": self.agent_name, "level": level, "event_type": event_type, **data } print(json.dumps(log_entry, ensure_ascii=False))# 使用示例logger = AgentLogger(agent_name="my_agent")trace_id = logger.start_trace("帮我查一下今天的天气")logger.log_thought(step=1, thought="用户想查天气，我需要调用天气查询工具")logger.log_tool_call(step=1, tool_name="weather_api", tool_input={"city": "北京"}, duration_ms=230)logger.log_tool_result(step=1, tool_name="weather_api", result="北京今天晴，25度")logger.log_final_answer(answer="北京今天晴，25度，适合出门", total_duration_ms=1500, total_token_usage=350)### 日志存储与检索Agent日志量大，需要合适的存储方案：| 存储方案 | 优点 | 缺点 | 适用场景 ||---------|------|------|---------|| Elasticsearch | 全文检索强、生态成熟 | 运维复杂、成本高 | 中大型团队 || Loki | 轻量、与Grafana集成好 | 全文检索弱 | 中小团队 || ClickHouse | 列存、分析快 | 不适合全文检索 | 数据分析场景 || 云服务（CloudWatch等） | 免运维、集成好 | 成本高、锁定厂商 | 云上部署 |## 11.2 指标（Metrics）：Agent的"体检报告"指标是可聚合的数值数据，反映系统的整体健康状况。### Agent核心指标#### 1. 性能指标- 端到端延迟：从用户输入到最终回答的总耗时- 首Token延迟（TTFT）：从用户输入到模型输出第一个Token的耗时- Token生成速度：模型每秒生成的Token数- 工具调用延迟：每个工具的调用耗时- 重试次数：工具调用或模型调用的重试次数#### 2. 成本指标- 总Token使用量：输入Token + 输出Token- 平均每次调用Token量：总Token / 调用次数- Token成本：根据模型定价计算的成本- 工具调用成本：第三方API调用的费用#### 3. 质量指标- 任务成功率：成功完成任务的比例- 工具调用成功率：工具调用成功的比例- 用户满意度：用户点赞/点踩的比例- 人工介入率：需要人工介入的比例- 幻觉率：模型产生幻觉的比例#### 4. 系统指标- 并发请求数：同时处理的请求数- 队列长度：等待处理的请求数- 模型超时率：模型调用超时的比例- 限流触发率：触发限流的比例### 指标采集代码示例pythonimport timefrom collections import defaultdictclass AgentMetrics: def __init__(self): self.latencies = [] self.token_usages = [] self.tool_calls = defaultdict(list) self.success_count = 0 self.total_count = 0 self.error_count = 0 def record_request(self, duration_ms, token_usage, success=True): """记录一次请求""" self.latencies.append(duration_ms) self.token_usages.append(token_usage) self.total_count += 1 if success: self.success_count += 1 else: self.error_count += 1 def record_tool_call(self, tool_name, duration_ms, success=True): """记录工具调用""" self.tool_calls[tool_name].append({ "duration_ms": duration_ms, "success": success }) def get_summary(self): """获取指标摘要""" if not self.latencies: return {} sorted_latencies = sorted(self.latencies) return { "total_requests": self.total_count, "success_rate": self.success_count / self.total_count, "error_rate": self.error_count / self.total_count, "avg_latency_ms": sum(self.latencies) / len(self.latencies), "p50_latency_ms": sorted_latencies[len(sorted_latencies)//2], "p95_latency_ms": sorted_latencies[int(len(sorted_latencies)*0.95)], "p99_latency_ms": sorted_latencies[int(len(sorted_latencies)*0.99)], "avg_token_usage": sum(self.token_usages) / len(self.token_usages), "total_token_usage": sum(self.token_usages), "tool_stats": self._get_tool_stats() } def _get_tool_stats(self): """获取工具调用统计""" stats = {} for tool_name, calls in self.tool_calls.items(): durations = [c["duration_ms"] for c in calls] successes = [c["success"] for c in calls] stats[tool_name] = { "total_calls": len(calls), "success_rate": sum(successes) / len(calls), "avg_duration_ms": sum(durations) / len(durations), "max_duration_ms": max(durations) } return stats# 使用示例metrics = AgentMetrics()# 模拟一次请求start = time.time()# ... Agent处理逻辑 ...duration = (time.time() - start) * 1000metrics.record_request(duration_ms=duration, token_usage=350, success=True)metrics.record_tool_call("weather_api", duration_ms=230, success=True)# 查看指标summary = metrics.get_summary()print(f"成功率: {summary['success_rate']:.2%}")print(f"平均延迟: {summary['avg_latency_ms']:.0f}ms")print(f"P95延迟: {summary['p95_latency_ms']:.0f}ms")print(f"总Token: {summary['total_token_usage']}")### 指标可视化指标需要可视化才能发挥价值，常用的可视化工具：- Grafana：最流行的开源可视化平台- Prometheus：时序数据库 + 告警- Datadog：商业可观测性平台- 云厂商监控：CloudWatch、Cloud Monitor等### 告警规则设计| 指标 | 告警阈值 | 告警级别 | 说明 ||------|---------|---------|------|| 错误率 | >5% | P1 | 系统异常 || P95延迟 | >30s | P2 | 性能下降 || 模型超时率 | >10% | P2 | 模型服务异常 || Token使用量突增 | >日均200% | P3 | 可能被滥用 || 队列长度 | >100 | P2 | 系统过载 |## 11.3 追踪（Traces）：Agent的"行程单"追踪记录了一次请求经过的所有组件和步骤，是定位问题的核心工具。### Agent追踪的特殊性传统软件追踪通常记录：请求 → 服务A → 服务B → 数据库 → 返回。Agent追踪更复杂，需要记录：- 用户输入- 系统提示词- 模型推理（可能多轮）- 工具调用（可能多个）- 每一步的思考过程- 最终回答这是一个有向无环图（DAG），而不是简单的线性链路。### 追踪模型设计Trace（一次完整的Agent调用）├── Span 1: 用户输入接收├── Span 2: 系统提示词组装├── Span 3: 第一轮模型推理│ ├── Span 3.1: 模型API调用│ └── Span 3.2: 输出解析├── Span 4: 工具调用 - 天气查询│ ├── Span 4.1: 参数组装│ ├── Span 4.2: API请求│ └── Span 4.3: 响应解析├── Span 5: 第二轮模型推理│ ├── Span 5.1: 模型API调用│ └── Span 5.2: 输出解析└── Span 6: 最终回答生成### OpenTelemetry集成OpenTelemetry是可观测性的行业标准，Agent系统也应该基于它构建。pythonfrom opentelemetry import tracefrom opentelemetry.sdk.trace import TracerProviderfrom opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessorimport time# 初始化OpenTelemetrytrace.set_tracer_provider(TracerProvider())tracer = trace.get_tracer(__name__)# 添加控制台导出器（生产环境用Jaeger/Zipkin等）exporter = ConsoleSpanExporter()span_processor = SimpleSpanProcessor(exporter)trace.get_tracer_provider().add_span_processor(span_processor)class TracedAgent: def __init__(self): self.tracer = tracer def run(self, user_input): """运行Agent，带完整追踪""" with self.tracer.start_as_current_span("agent.run") as span: span.set_attribute("user_input", user_input) span.set_attribute("agent.version", "1.0.0") # 第一步：模型推理 with self.tracer.start_as_current_span("model.inference") as model_span: model_span.set_attribute("model.name", "gpt-4") model_span.set_attribute("step", 1) start = time.time() # ... 调用模型 ... thought = "用户想查天气，需要调用天气工具" action = "weather_api" action_input = {"city": "北京"} model_span.set_attribute("thought", thought) model_span.set_attribute("action", action) model_span.set_attribute("action_input", str(action_input)) model_span.set_attribute("duration_ms", (time.time()-start)*1000) # 第二步：工具调用 with self.tracer.start_as_current_span(f"tool.{action}") as tool_span: tool_span.set_attribute("tool.name", action) tool_span.set_attribute("tool.input", str(action_input)) start = time.time() # ... 调用工具 ... result = "北京今天晴，25度" tool_span.set_attribute("tool.result_preview", result[:200]) tool_span.set_attribute("duration_ms", (time.time()-start)*1000) tool_span.set_attribute("success", True) # 第三步：生成最终回答 with self.tracer.start_as_current_span("model.final_answer") as final_span: final_span.set_attribute("step", 2) # ... 生成最终回答 ... final_answer = "北京今天晴，25度，适合出门" final_span.set_attribute("final_answer", final_answer) span.set_attribute("final_answer", final_answer) return final_answer# 使用示例agent = TracedAgent()result = agent.run("帮我查一下今天的天气")print(result)### 追踪后端选择| 后端 | 优点 | 缺点 | 适用场景 ||------|------|------|---------|| Jaeger | 开源、轻量、UI友好 | 大规模性能一般 | 中小团队 || Zipkin | 开源、简单 | UI较简陋 | 简单场景 || Tempo | 与Grafana集成好 | 功能相对简单 | Grafana生态 || Datadog | 功能全、免运维 | 成本高 | 中大型团队 || LangSmith | Agent专属、功能强 | 锁定LangChain生态 | LangChain用户 |—# 十二、Agent专属可观测性深入除了传统的三大支柱，Agent系统还有一些专属的可观测性需求。## 12.1 推理链追踪（Thought Process Tracking）推理链追踪记录了模型的每一步思考过程，是理解Agent行为的核心。### 为什么需要推理链追踪- 问题定位：Agent出错时，能看到是哪一步想错了- 行为理解：理解Agent为什么做出某个决策- 质量评估：评估模型的推理质量- 合规审计：为监管提供完整的决策证据链### 推理链数据结构pythonfrom dataclasses import dataclass, fieldfrom typing import List, Optional, Dict, Anyfrom datetime import datetime@dataclassclass ThoughtStep: """单步思考""" step: int thought: str # 模型的思考内容 action: str # 选择的工具 action_input: Dict[str, Any] # 工具输入 observation: Optional[str] = None # 工具返回结果 timestamp: datetime = field(default_factory=datetime.now) duration_ms: float = 0.0 token_usage: Optional[Dict[str, int]] = None@dataclassclass ReasoningTrace: """完整推理链""" trace_id: str user_input: str system_prompt: str steps: List[ThoughtStep] = field(default_factory=list) final_answer: Optional[str] = None total_duration_ms: float = 0.0 total_token_usage: Optional[Dict[str, int]] = None success: bool = True error: Optional[str] = None def add_step(self, step: ThoughtStep): """添加一步思考""" self.steps.append(step) def to_dict(self): """转为字典""" return { "trace_id": self.trace_id, "user_input": self.user_input, "system_prompt": self.system_prompt, "steps": [ { "step": s.step, "thought": s.thought, "action": s.action, "action_input": s.action_input, "observation": s.observation, "timestamp": s.timestamp.isoformat(), "duration_ms": s.duration_ms, "token_usage": s.token_usage } for s in self.steps ], "final_answer": self.final_answer, "total_duration_ms": self.total_duration_ms, "total_token_usage": self.total_token_usage, "success": self.success, "error": self.error }### 推理链可视化推理链需要可视化才能直观理解，常见的可视化方式：1. 时间线视图：按时间顺序展示每一步2. 流程图视图：展示思考和工具调用的流程3. 对比视图：对比多次调用的推理链差异4. 搜索视图：按关键词搜索推理链## 12.2 工具调用追踪（Tool Call Tracking）工具调用是Agent与外部世界交互的方式，需要详细追踪。### 工具调用追踪内容- 工具名称- 工具输入参数- 工具调用时间- 工具返回结果- 工具调用耗时- 工具调用是否成功- 重试次数- 错误信息（如果失败）### 工具调用统计分析pythonfrom collections import defaultdictimport statisticsclass ToolCallAnalyzer: def __init__(self): self.calls = [] def record_call(self, tool_name, input_params, output, duration_ms, success=True, error=None): """记录一次工具调用""" self.calls.append({ "tool_name": tool_name, "input_params": input_params, "output_preview": str(output)[:500], "duration_ms": duration_ms, "success": success, "error": error }) def get_tool_stats(self): """获取工具统计""" tool_stats = defaultdict(lambda: { "total_calls": 0, "success_calls": 0, "failed_calls": 0, "durations": [], "errors": [] }) for call in self.calls: stats = tool_stats[call["tool_name"]] stats["total_calls"] += 1 stats["durations"].append(call["duration_ms"]) if call["success"]: stats["success_calls"] += 1 else: stats["failed_calls"] += 1 stats["errors"].append(call["error"]) # 计算统计指标 result = {} for tool_name, stats in tool_stats.items(): durations = stats["durations"] result[tool_name] = { "total_calls": stats["total_calls"], "success_rate": stats["success_calls"] / stats["total_calls"], "avg_duration_ms": statistics.mean(durations), "p50_duration_ms": statistics.median(durations), "p95_duration_ms": sorted(durations)[int(len(durations)*0.95)] if durations else 0, "max_duration_ms": max(durations) if durations else 0, "common_errors": self._get_common_errors(stats["errors"]) } return result def _get_common_errors(self, errors, top_n=5): """获取常见错误""" from collections import Counter error_counts = Counter(str(e) for e in errors if e) return error_counts.most_common(top_n) def get_slow_tools(self, threshold_ms=1000): """获取慢工具""" slow = [] for call in self.calls: if call["duration_ms"] > threshold_ms: slow.append({ "tool_name": call["tool_name"], "duration_ms": call["duration_ms"], "input_params": call["input_params"] }) return sorted(slow, key=lambda x: x["duration_ms"], reverse=True)# 使用示例analyzer = ToolCallAnalyzer()analyzer.record_call("weather_api", {"city": "北京"}, "晴，25度", 230, True)analyzer.record_call("weather_api", {"city": "上海"}, "多云，22度", 5000, False, "Timeout")analyzer.record_call("search_api", {"query": "AI"}, "结果...", 800, True)stats = analyzer.get_tool_stats()for tool, s in stats.items(): print(f"{tool}: 成功率={s['success_rate']:.0%}, 平均延迟={s['avg_duration_ms']:.0f}ms")## 12.3 Token使用追踪（Token Usage Tracking）Token是大模型的"货币"，需要精确追踪每一步的Token使用。### Token追踪内容- 输入Token数（Prompt Tokens）- 输出Token数（Completion Tokens）- 总Token数- 缓存命中Token数（如果使用了Prompt缓存）- 每一步的Token使用- 每次调用的Token使用- 每个用户的Token使用- 每个模型的Token使用### Token成本计算python# 模型定价（示例，实际以官方为准）MODEL_PRICING = { "gpt-4": { "input_per_1k": 0.03, # 美元/1K Token "output_per_1k": 0.06 }, "gpt-3.5-turbo": { "input_per_1k": 0.0015, "output_per_1k": 0.002 }, "claude-3-opus": { "input_per_1k": 0.015, "output_per_1k": 0.075 }}class TokenCostTracker: def __init__(self): self.usage = [] def record_usage(self, model, input_tokens, output_tokens, user_id=None, trace_id=None): """记录Token使用""" cost = self._calculate_cost(model, input_tokens, output_tokens) self.usage.append({ "model": model, "input_tokens": input_tokens, "output_tokens": output_tokens, "total_tokens": input_tokens + output_tokens, "cost_usd": cost, "user_id": user_id, "trace_id": trace_id }) return cost def _calculate_cost(self, model, input_tokens, output_tokens): """计算成本""" pricing = MODEL_PRICING.get(model) if not pricing: return 0.0 input_cost = (input_tokens / 1000) * pricing["input_per_1k"] output_cost = (output_tokens / 1000) * pricing["output_per_1k"] return input_cost + output_cost def get_daily_summary(self): """获取每日汇总""" from collections import defaultdict daily = defaultdict(lambda: { "total_tokens": 0, "total_cost": 0.0, "by_model": defaultdict(lambda: {"tokens": 0, "cost": 0.0}) }) for u in self.usage: day = "today" # 实际应按日期分组 daily[day]["total_tokens"] += u["total_tokens"] daily[day]["total_cost"] += u["cost_usd"] daily[day]["by_model"][u["model"]]["tokens"] += u["total_tokens"] daily[day]["by_model"][u["model"]]["cost"] += u["cost_usd"] return dict(daily) def get_top_users(self, top_n=10): """获取Token使用最多的用户""" from collections import defaultdict user_usage = defaultdict(lambda: {"tokens": 0, "cost": 0.0}) for u in self.usage: if u["user_id"]: user_usage[u["user_id"]]["tokens"] += u["total_tokens"] user_usage[u["user_id"]]["cost"] += u["cost_usd"] return sorted(user_usage.items(), key=lambda x: x[1]["tokens"], reverse=True)[:top_n]# 使用示例tracker = TokenCostTracker()cost = tracker.record_usage("gpt-4", input_tokens=1500, output_tokens=350, user_id="user_001")print(f"本次成本: ${cost:.4f}")summary = tracker.get_daily_summary()print(f"今日总Token: {summary['today']['total_tokens']}")print(f"今日总成本: ${summary['today']['total_cost']:.4f}")### Token优化建议通过Token使用分析，可以发现优化机会：1. 系统提示词优化：系统提示词是否过长，能否精简2. 上下文管理：是否保留了不必要的历史对话3. Prompt缓存：是否使用了Prompt缓存技术4. 模型选择：简单任务能否用更便宜的模型5. 输出长度控制：是否限制了输出长度—# 十三、主流可观测性工具深入## 13.1 LangSmithLangSmith是LangChain官方推出的可观测性平台，专为LLM应用设计。### 核心功能1. 追踪（Tracing）：完整记录LLM调用的每一步2. 评估（Evaluation）：自动化评估LLM输出质量3. 提示词管理（Prompt Management）：版本化管理提示词4. 数据集（Datasets）：管理测试数据集5. 监控（Monitoring）：实时监控LLM应用性能### 集成代码示例pythonfrom langchain_openai import ChatOpenAIfrom langchain.agents import AgentExecutor, create_openai_functions_agentfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_core.tools import toolimport os# 设置LangSmith环境变量os.environ["LANGCHAIN_TRACING_V2"] = "true"os.environ["LANGCHAIN_API_KEY"] = "your-api-key"os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"@tooldef get_weather(city: str) -> str: """查询指定城市的天气""" return f"{city}今天晴，25度"@tooldef search_web(query: str) -> str: """搜索网络信息""" return f"搜索结果：{query}相关信息..."# 创建Agentprompt = ChatPromptTemplate.from_messages([ ("system", "你是一个有用的助手"), ("user", "{input}"), MessagesPlaceholder(variable_name="agent_scratchpad"),])llm = ChatOpenAI(model="gpt-4", temperature=0)tools = [get_weather, search_web]agent = create_openai_functions_agent(llm, tools, prompt)agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)# 运行Agent（自动追踪到LangSmith）result = agent_executor.invoke({"input": "帮我查一下北京今天的天气，然后搜索一下北京有什么好玩的"})print(result["output"])### LangSmith评估功能pythonfrom langsmith import Clientfrom langsmith.evaluation import evaluateclient = Client()# 定义评估函数def evaluate_answer(output: str, reference: str) -> dict: """评估回答质量""" # 这里可以用LLM作为评判者 score = 0.8 # 示例分数 return {"score": score, "comment": "回答质量良好"}# 运行评估results = evaluate( lambda inputs: agent_executor.invoke(inputs), data="my-test-dataset", evaluators=[evaluate_answer], experiment_prefix="v1.0-evaluation")## 13.2 LangFuseLangFuse是开源的LLM可观测性平台，支持自托管。### 核心功能1. 追踪：完整的LLM调用追踪2. 评估：自动化评估3. 提示词管理：版本化提示词4. 成本分析：详细的成本分析5. 用户管理：多用户支持### 与LangSmith的对比| 特性 | LangSmith | LangFuse ||------|-----------|----------|| 开源 | 否 | 是 || 自托管 | 否 | 是 || 价格 | 按使用量付费 | 免费（自托管）/ 云服务付费 || 与LangChain集成 | 原生 | 良好 || UI | 优秀 | 良好 || 社区 | 大 | growing |## 13.3 Phoenix（Arize AI）Phoenix是Arize AI推出的开源LLM可观测性工具，专注于LLM应用的评估和调试。### 核心功能1. 嵌入可视化：将LLM输出可视化到嵌入空间2. 聚类分析：自动聚类相似的输出3. 异常检测：检测异常的LLM输出4. 评估：自动化评估5. 追踪：完整的调用追踪### 适用场景- 发现LLM输出的异常模式- 理解用户查询的分布- 评估RAG系统的检索质量- 调试模型幻觉问题## 13.4 OpenTelemetry GenAIOpenTelemetry正在推进GenAI的语义约定标准化，让Agent追踪数据可以接入通用的可观测性体系。### 核心概念- GenAI语义约定：定义LLM调用的标准属性- Instrumentation：自动埋点库- Collector：数据收集器- Exporter：数据导出器### 优势- 标准化：与通用可观测性体系兼容- 厂商中立：不锁定特定厂商- 生态丰富：大量后端支持—# 十四、实战：构建完整的Agent可观测性体系## 14.1 架构设计一个完整的Agent可观测性体系应该包含以下组件：┌─────────────────────────────────────────────────┐│ Agent应用 ││ ┌─────────┐ ┌─────────┐ ┌─────────┐ ││ │ 日志埋点 │ │ 指标埋点 │ │ 追踪埋点 │ ││ └────┬────┘ └────┬────┘ └────┬────┘ │└───────┼─────────────┼─────────────┼─────────────┘ │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 日志收集 │ │ 指标收集 │ │ 追踪收集 │ │ Filebeat │ │Prometheus│ │ OTel │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 日志存储 │ │ 指标存储 │ │ 追踪存储 │ │ ES │ │Prometheus│ │ Jaeger │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ └──────────────┼──────────────┘ ▼ ┌──────────────┐ │ 可视化平台 │ │ Grafana │ └──────────────┘## 14.2 日志系统搭建### Elasticsearch + Filebeat + Kibana（EFK）yaml# Filebeat配置示例filebeat.inputs: - type: log enabled: true paths: - /var/log/agent/*.log json.keys_under_root: true json.add_error_key: trueoutput.elasticsearch: hosts: ["elasticsearch:9200"] index: "agent-logs-%{+yyyy.MM.dd}"setup.template.name: "agent-logs"setup.template.pattern: "agent-logs-*"### 关键日志查询json// 查询错误率最高的Trace{ "size": 0, "aggs": { "by_trace": { "terms": { "field": "trace_id", "size": 10 }, "aggs": { "error_count": { "filter": { "term": { "level": "ERROR" } } } } } }}## 14.3 指标系统搭建### Prometheus + Grafanapython# Agent指标暴露（Prometheus格式）from prometheus_client import Counter, Histogram, Gauge, start_http_serverimport time# 定义指标request_total = Counter('agent_requests_total', 'Total agent requests', ['agent_name', 'status'])request_duration = Histogram('agent_request_duration_seconds', 'Request duration', ['agent_name'])token_usage = Counter('agent_token_usage_total', 'Total token usage', ['agent_name', 'model'])active_requests = Gauge('agent_active_requests', 'Active requests', ['agent_name'])# 启动HTTP服务start_http_server(8000)# 在Agent中使用def handle_request(agent_name, user_input): active_requests.labels(agent_name=agent_name).inc() start = time.time() try: # ... 处理请求 ... result = "..." request_total.labels(agent_name=agent_name, status="success").inc() return result except Exception as e: request_total.labels(agent_name=agent_name, status="error").inc() raise finally: duration = time.time() - start request_duration.labels(agent_name=agent_name).observe(duration) active_requests.labels(agent_name=agent_name).dec()### Grafana仪表盘设计关键面板：1. 请求量趋势：QPS折线图2. 延迟分布：P50/P95/P99延迟3. 错误率：错误率趋势图4. Token使用：Token使用量和成本5. 工具调用：各工具调用量和成功率6. 模型分布：各模型使用占比7. 用户活跃：活跃用户数## 14.4 追踪系统搭建### Jaeger + OpenTelemetryyaml# docker-compose.ymlversion: '3'services: jaeger: image: jaegertracing/all-in-one:latest ports: - "16686:16686" # UI - "4317:4317" # OTLP gRPC - "4318:4318" # OTLP HTTP environment: - COLLECTOR_OTLP_ENABLED=true``````python# Agent端配置from opentelemetry import tracefrom opentelemetry.sdk.trace import TracerProviderfrom opentelemetry.sdk.trace.export import BatchSpanProcessorfrom opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter# 配置OTLP导出器exporter = OTLPSpanExporter(endpoint="http://jaeger:4317")span_processor = BatchSpanProcessor(exporter)# 设置TracerProvidertrace.set_tracer_provider(TracerProvider())trace.get_tracer_provider().add_span_processor(span_processor)—# 十五、故障排查实战案例## 15.1 案例一：Agent回答质量突然下降### 现象用户反馈Agent回答质量下降，经常答非所问。### 排查步骤1. 查看指标：错误率正常，但用户满意度下降2. 查看追踪：随机抽取几个Trace，发现推理链变短了3. 查看日志：发现模型从gpt-4降级到了gpt-3.5-turbo4. 根因：gpt-4 API限流，自动降级到了更便宜的模型### 解决方案- 增加gpt-4的API配额- 优化降级策略，关键任务不降级- 增加模型切换的告警## 15.2 案例二：Agent响应变慢### 现象用户反馈Agent响应越来越慢。### 排查步骤1. 查看指标：P95延迟从5秒上升到30秒2. 查看追踪：发现工具调用耗时增加3. 查看日志：发现某个第三方API响应变慢4. 根因：第三方API服务降级### 解决方案- 增加工具调用超时- 增加重试机制- 准备备用工具- 增加第三方API的监控告警## 15.3 案例三：Agent成本突增### 现象某天Token使用量突增3倍。### 排查步骤1. 查看指标：Token使用量突增2. 查看追踪：发现某个用户的Trace特别长3. 查看日志：发现该用户在循环调用Agent4. 根因：用户脚本异常，无限循环调用### 解决方案- 增加用户级限流- 增加异常调用检测- 增加成本告警—# 十六、性能优化## 16.1 可观测性本身的性能开销可观测性会带来性能开销，需要控制在合理范围内。### 开销来源- 日志序列化和输出- 指标采集和聚合- 追踪数据采集和导出- 网络传输开销### 优化方法1. 采样：不是所有请求都记录完整追踪，按比例采样2. 异步：可观测性数据异步处理，不阻塞主流程3. 批量：批量导出数据，减少网络请求4. 压缩：压缩传输数据5. 分级：生产环境只记录关键信息，调试环境记录完整信息### 采样策略pythonimport randomclass SamplingStrategy: def __init__(self, sample_rate=0.1): self.sample_rate = sample_rate def should_trace(self, trace_id=None): """决定是否记录完整追踪""" # 基于Trace ID的确定性采样（同一个Trace要么全记要么全不记） if trace_id: hash_val = hash(trace_id) % 1000 return hash_val < self.sample_rate * 1000 # 随机采样 return random.random() < self.sample_rate def should_log_detail(self, level="INFO"): """决定是否记录详细日志""" if level == "ERROR": return True # 错误总是记录 return random.random() < self.sample_rate# 使用示例sampler = SamplingStrategy(sample_rate=0.1) # 10%采样if sampler.should_trace(trace_id): # 记录完整追踪 passelse: # 只记录关键指标 pass## 16.2 存储成本优化可观测性数据量大，存储成本高，需要优化。### 优化方法1. 分层存储：热数据存SSD，冷数据存对象存储2. 数据降采样：历史数据降采样，保留趋势3. 自动过期：设置数据保留策略，自动删除过期数据4. 压缩存储：压缩存储数据5. 只存必要数据：不是所有数据都需要长期保存### 保留策略建议| 数据类型 | 热存储（SSD） | 冷存储（对象存储） | 总保留期 ||---------|-------------|-----------------|---------|| 详细日志 | 7天 | 30天 | 90天 || 聚合指标 | 30天 | 1年 | 2年 || 详细追踪 | 3天 | 7天 | 30天 || 错误日志 | 30天 | 1年 | 永久 |—# 十七、行业应用案例## 17.1 金融行业：智能客服Agent### 挑战- 回答质量要求高，不能误导用户- 合规要求严格，需要完整审计- 用户量大，成本控制重要### 可观测性方案- 完整的推理链追踪，用于合规审计- 回答质量自动评估，发现问题及时告警- 详细的Token使用追踪，控制成本- 用户满意度监控，持续优化### 效果- 回答质量提升20%- 合规审计效率提升50%- 成本降低15%## 17.2 医疗行业：医疗咨询Agent### 挑战- 回答准确性要求极高，不能出错- 隐私要求高，患者数据不能泄露- 需要完整的决策证据链### 可观测性方案- 完整的推理链追踪，用于医疗审计- 回答准确性自动评估- 敏感信息检测，防止患者数据泄露- 异常回答检测，及时人工介入### 效果- 回答准确率达到95%- 隐私泄露事件为零- 医疗审计效率提升60%## 17.3 企业自动化：办公Agent### 挑战- 工具调用多，容易出错- 涉及企业内部系统，权限控制重要- 需要快速定位问题### 可观测性方案- 完整的工具调用追踪- 权限使用审计- 异常操作检测和告警- 快速问题定位### 效果- 问题定位时间从小时级降到分钟级- 异常操作减少80%- 运维效率提升40%—# 十八、未来趋势## 18.1 大模型时代的可观测性新挑战大模型的兴起带来了新的可观测性挑战：- 推理过程不透明：大模型的"黑盒"特性，难以理解内部推理- 多模态：文本、图像、音频、视频的统一可观测性- Agent自主性增强：Agent自主决策增多，需要更细粒度的追踪- 模型记忆：大模型的长期记忆，需要追踪记忆的读写## 18.2 可观测性技术发展方向- AI驱动的可观测性：用AI自动分析可观测性数据，发现异常- 预测性可观测性：预测可能发生的问题，提前预警- 自动化根因分析：自动定位问题根因，减少人工排查- 实时可观测性：实时分析和告警，秒级响应- 统一可观测性：日志、指标、追踪、事件的统一平台## 18.3 标准化进程可观测性正在走向标准化：- OpenTelemetry GenAI语义约定：LLM调用的标准属性- OpenInference：开源的LLM追踪标准- Model Context Protocol（MCP）：模型上下文协议- Agent Communication Protocol（ACP）：Agent通信协议—# 十九、总结与建议## 19.1 Agent可观测性体系建设路线图### 第一阶段：基础监控（1-2周）- 记录基本日志（请求、错误、耗时）- 采集基本指标（QPS、延迟、错误率）- 搭建基础监控仪表盘- 设置基本告警### 第二阶段：追踪与分析（2-4周）- 接入分布式追踪- 记录完整的推理链- 记录工具调用详情- 记录Token使用情况- 搭建追踪查询界面### 第三阶段：评估与优化（1-2月）- 建立回答质量评估体系- 建立自动化评估流程- 分析可观测性数据，发现优化点- 持续优化Agent性能和质量### 第四阶段：智能化（持续）- AI驱动的异常检测- 自动化根因分析- 预测性告警- 智能优化建议## 19.2 工具选型建议| 团队规模 | 推荐方案 | 理由 ||---------|---------|------|| 个人/小团队 | LangSmith云服务 | 免运维、开箱即用、免费额度够用 || 中小团队 | LangFuse自托管 + Prometheus + Grafana | 开源、成本低、可定制 || 中大型团队 | OpenTelemetry + Jaeger + Prometheus + Elasticsearch | 标准化、可扩展、生态成熟 || 企业级 | 商业可观测性平台（Datadog等）+ 定制化 | 功能全、支持好、SLA保障 |## 19.3 最佳实践1. 从第一天就开始：可观测性不是事后补救，要从项目开始就建设2. 结构化日志：所有日志都要是结构化的，便于分析3. 统一Trace ID：整个调用链使用同一个Trace ID，便于关联4. 采样策略：合理设置采样率，平衡可观测性和成本5. 告警降噪：避免告警疲劳，只告警真正重要的问题6. 定期复盘：定期分析可观测性数据，持续优化7. 隐私保护：可观测性数据中可能包含敏感信息，注意脱敏## 19.4 最后的话Agent可观测性不是可选项，而是生产级Agent应用的必备能力。它让开发者能够：- 快速定位多步推理中的故障根因- 持续评估模型行为是否符合预期- 为合规审计提供完整的推理证据链- 优化性能和成本- 提升用户满意度随着Agent应用走向复杂化，可观测性体系也将从"能用"走向"好用"，成为Agent工程质量的基础设施。如果你正在构建Agent应用，还没有建设可观测性体系，赶紧开始吧。毕竟，你无法优化你无法测量的东西。
在这里插入图片描述