给 LLM 调用加上监控：Token 用量、延迟、错误率一目了然
上一篇文章搭建了第一个 LangGraph Agent，它能跑起来了。但运维的本能告诉我：能跑 ≠ 能运维。一个系统如果没有监控，就是在裸奔。
这篇文章给 Agent 加上可观测性——从零实现一个 LLM 调用监控模块，不依赖任何第三方平台。
为什么 LLM 可观测性比传统服务更难
传统服务的监控我们很熟悉：QPS、延迟、错误率、CPU、内存。但 LLM 调用引入了几个新维度：
| 传统服务 | LLM 调用 | 为什么不同 |
|---|---|---|
| 延迟只取决于代码 | 延迟取决于模型负载和 Token 数 | 同一个接口，长回复可能慢 10 倍 |
| 成本固定（服务器） | 成本按 Token 计费 | 每次调用都在花钱，必须精确计量 |
| 错误是明确的（500/超时） | 错误可能是语义性的 | API 返回 200 但内容可能是幻觉 |
| 无第三方依赖黑盒 | 模型是黑盒 | 你无法控制模型侧的行为 |
结论：LLM 可观测性需要同时监控系统指标（延迟、错误）和业务指标（Token、成本、质量）。
设计思路：用 LangChain 回调拦截每次 LLM 调用
LangChain 提供了 BaseCallbackHandler，可以在 LLM 调用的各个生命周期节点插入钩子：
on_llm_start → LLM 推理中... → on_llm_end （成功）
→ on_llm_error（失败）
我们在 on_llm_start 记录开始时间，在 on_llm_end 计算延迟、提取 Token 用量、估算成本。核心就三个类：
LLMCallRecord — 单次调用的指标快照
LLMObservability — 指标收集器，提供聚合查询和报告
ObservabilityCallback — LangChain 回调，连接 LLM 和收集器
实现：指标收集器
from dataclasses import dataclass, field
import time
import threading
@dataclass
class LLMCallRecord:
"""单次 LLM 调用的完整指标"""
timestamp: float = field(default_factory=time.time)
model: str = ""
duration_ms: float = 0.0
prompt_tokens: int = 0
completion_tokens: int = 0
success: bool = True
error: str | None = None
@property
def total_tokens(self) -> int:
return self.prompt_tokens + self.completion_tokens
class LLMObservability:
"""LLM 调用可观测性收集器"""
def __init__(self):
self.calls: list[LLMCallRecord] = []
self._lock = threading.Lock()
self._start_time = time.time()
def record(self, record: LLMCallRecord):
with self._lock:
self.calls.append(record)
# ─── 运维关心的聚合指标 ───
@property
def call_count(self) -> int:
return len(self.calls)
@property
def total_tokens(self) -> int:
return sum(c.total_tokens for c in self.calls)
@property
def avg_latency_ms(self) -> float:
if not self.calls:
return 0.0
return round(sum(c.duration_ms for c in self.calls) / len(self.calls), 1)
@property
def p95_latency_ms(self) -> float:
"""P95 延迟：95% 的请求在此时间内完成"""
if not self.calls:
return 0.0
sorted_latencies = sorted(c.duration_ms for c in self.calls)
idx = int(len(sorted_latencies) * 0.95)
return sorted_latencies[min(idx, len(sorted_latencies) - 1)]
@property
def error_rate(self) -> float:
if not self.calls:
return 0.0
return sum(1 for c in self.calls if not c.success) / len(self.calls)
设计要点：
threading.Lock保证线程安全——生产环境中 Agent 可能并发处理多个请求- P95 延迟比平均值更有意义——平均值会被少数慢请求拉高，P95 反映的是”大多数用户感受到的延迟”
- 所有指标都是惰性计算——只在需要时聚合，避免每次记录都重算
实现：回调处理器
from langchain_core.callbacks import BaseCallbackHandler
import time
class ObservabilityCallback(BaseCallbackHandler):
"""LangChain 回调：拦截 LLM 调用，记录指标"""
def __init__(self, obs: LLMObservability, model_name: str):
self.obs = obs
self.model_name = model_name
self._start_times: dict[str, float] = {}
def on_llm_start(self, serialized, prompts, *, run_id, **kwargs):
self._start_times[run_id] = time.time()
def on_llm_end(self, response, *, run_id, **kwargs):
start = self._start_times.pop(run_id, time.time())
duration_ms = (time.time() - start) * 1000
# 从 LLM 响应中提取 token 用量
token_usage = {}
if hasattr(response, "llm_output") and response.llm_output:
token_usage = response.llm_output.get("token_usage", {})
record = LLMCallRecord(
model=self.model_name,
duration_ms=round(duration_ms, 1),
prompt_tokens=token_usage.get("prompt_tokens", 0),
completion_tokens=token_usage.get("completion_tokens", 0),
)
self.obs.record(record)
def on_llm_error(self, error, *, run_id, **kwargs):
start = self._start_times.pop(run_id, time.time())
self.obs.record(LLMCallRecord(
model=self.model_name,
duration_ms=round((time.time() - start) * 1000, 1),
success=False,
error=str(error),
))
关键细节：
- 用
run_id作为 key 追踪每次调用的起始时间——LangChain 的每次 LLM 调用都有唯一run_id on_llm_end和on_llm_error是互斥的——一次调用只会触发其中一个token_usage的位置因模型而异——OpenAI 放在llm_output，Anthropic 放在response_metadata，需要兼容处理
成本估算
有了 Token 用量，成本估算就简单了。维护一个定价表：
MODEL_PRICING = {
"gpt-4o-mini": {"input": 0.15, "output": 0.60}, # $/1M tokens
"gpt-4o": {"input": 2.50, "output": 10.00},
"deepseek-chat": {"input": 0.14, "output": 0.28},
"claude-3.5-sonnet": {"input": 3.00, "output": 15.00},
}
def estimate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
pricing = MODEL_PRICING.get(model, {})
input_cost = (prompt_tokens / 1_000_000) * pricing.get("input", 0)
output_cost = (completion_tokens / 1_000_000) * pricing.get("output", 0)
return input_cost + output_cost
注意：定价会变，建议从配置文件或 API 动态获取，不要硬编码。
接入 Agent
接入方式只有一行变化——在 agent.stream() 的 config 中传入回调：
obs = LLMObservability()
def run_with_obs(question: str):
callback = ObservabilityCallback(obs, MODEL_NAME)
config = {
"configurable": {"thread_id": "demo-1"},
"callbacks": [callback], # ← 就这一行
}
for event in agent.stream(
{"messages": [HumanMessage(content=question)]},
config=config,
stream_mode="values",
):
# ... 处理事件 ...
对比第一篇的 run() 函数，唯一区别就是 config 中多了 callbacks。可观测性应该是外挂的，不应该侵入业务逻辑——这正是回调模式的优势。
实际效果
运行 observability_demo.py，三次对话后的输出：
👤 用户: 现在几点了？
🔧 [工具] get_current_time 返回: 2025-08-28 15:30:00
🤖 助手: 现在是 2025年8月28日 15:30:00。
👤 用户: 检查磁盘使用情况，然后看看 nginx 日志有没有异常
🔧 [工具] check_disk_usage 返回: {"total": "100G", "used": "72G"...}
🔧 [工具] read_recent_logs 返回: [ERROR] connect() failed to upstream...
🤖 助手: 磁盘使用率 72%，nginx 日志中发现一个错误...
👤 用户: 你好，介绍一下你自己
🤖 助手: 你好！我是 AI 运维助手，可以帮你...
=======================================================
📊 LLM 调用可观测性报告
=======================================================
运行时长: 12s
总调用次数: 5
成功: 5
失败: 0 (0.0%)
── Token 用量 ──
输入 Token: 3,847
输出 Token: 412
总计 Token: 4,259
── 延迟 ──
平均延迟: 1340ms
P95 延迟: 2180ms
── 成本 ──
估算总成本: $0.0008
=======================================================
三次用户对话触发了 5 次 LLM 调用（有些问题需要”LLM 推理 → 工具调用 → LLM 再次推理”），总成本不到 0.1 美分。
对接监控系统
to_json() 方法导出结构化数据，可以接入你的现有监控栈：
import json
metrics = obs.to_json()
# 方式 1：暴露为 HTTP 端点，Prometheus 定期抓取
@app.get("/metrics/llm")
def llm_metrics():
return metrics
# 方式 2：推送到 Pushgateway
# 方式 3：写入时序数据库（InfluxDB / VictoriaMetrics）
# 方式 4：直接打日志，ELK 采集
建议的告警规则：
| 指标 | 告警条件 | 原因 |
|---|---|---|
error_rate | > 1% | API 故障或配额耗尽 |
p95_latency_ms | > 5000ms | 模型响应变慢，可能影响用户体验 |
total_cost_usd | 日环比 > 200% | 防止异常调用导致账单爆炸 |
avg_tokens_per_call | 周环比 > 150% | 提示词可能膨胀，需要优化 |
下一步
这篇文章给 Agent 加上了”仪表盘”。但仪表盘只能告诉你出问题了，不能帮你挡问题。下一篇文章会深入 AI Gateway——在 Agent 和模型之间加一层网关，统一做认证、路由、限流和灾备。这是运维视角的 AI 基础设施核心组件。