10.6 可观测性
在上一节中，我们讨论了"反馈闭环"——如何通过用户反馈与自动化评估信号不断迭代 Agent。然而，反馈的前提是看得见。如果 Agent 的每一次推理都像黑箱一样不可观测，那么任何优化都只是盲人摸象。本节将聚焦"可观测性"这一贯穿整个评估与优化体系的基础能力：如何让 Agent 的运行过程透明、可度量、可回溯。
仪表盘类比：想象你在驾驶一架飞机。如果仪表盘上没有任何读数——引擎温度、油耗、高度、速度——你无从判断飞行状态是否正常，更无从在异常时做出正确判断。Agent 系统的"仪表盘"就是可观测性体系：日志是飞行记录仪，指标是仪表上的数字读数，追踪是整条航线的回放。没有仪表盘的飞行，本质上是在赌命。
10.6.1 为什么 Agent 需要可观测性
传统 Web 应用的可观测性通常聚焦于 HTTP 请求延迟、数据库查询耗时、CPU/内存占用等基础设施指标。Agent 系统在此基础上多出几个独特的可观测性需求：
- LLM 调用是非确定性的——同样的输入可能产生不同输出，需要完整记录每次调用的输入、输出、模型参数与 Token 消耗。
- 多步推理链路长——一个用户请求可能触发数十次 LLM 调用与工具调用的组合，没有端到端追踪几乎无法定位问题。
- 成本与质量高度耦合——更强的模型更贵，但效果不一定成比例提升；需要将质量信号与成本信号关联分析。
- 用户感知差异大——同一套 Agent 在不同用户、不同场景下表现可能天差地别，需要按维度切片分析。
可以说，Agent 系统的可观测性不只是"运维问题"，更是"产品问题"——它直接决定了你能否回答"我的 Agent 到底好不好用"这个核心问题。
10.6.2 可观测性三支柱
在云原生领域，可观测性（Observability）通常被拆解为三大支柱：
可观测性 (Observability)
├── 日志 (Logs) -> 记录"发生了什么"：离散事件，上下文丰富
├── 指标 (Metrics) -> 量化"表现如何"：聚合数值，适合告警
└── 追踪 (Traces) -> 追踪"完整链路"：请求在系统中的流转路径三者各有侧重又互为补充。在 Agent 场景下，这三者的含义需要被进一步细化：
| 维度 | 关注指标 | 典型工具 |
|---|---|---|
| LLM 调用 | 延迟、Token 用量、成功率、模型版本 | LangSmith |
| 工具调用 | 调用次数、失败率、平均耗时 | LangFuse |
| 检索质量 | 召回率、相关性分数、命中率 | Phoenix (Arize) |
| 对话质量 | 用户满意度、任务完成率、轮次分布 | 自定义 |
| 成本 | 按模型/用户/功能的成本分摊 | LangSmith / LangFuse |
下面我们逐一介绍主流的可观测性工具及其在 Agent 场景中的实战用法。
10.6.3 LangSmith 实战
LangSmith 是 LangChain 官方推出的 Agent 可观测性平台，与 LangChain 生态深度集成。它的核心能力是通过 @traceable 装饰器自动注入追踪上下文，无需手动埋点。
# 从 LangSmith SDK 导入客户端与追踪装饰器
from langsmith import Client, traceable
# 导入 LangChain 的 Tracer 回调，用于自动追踪链式调用
from langchain.callbacks.tracers.langchain import LangChainTracer
import os
# 开启 LangChain V2 追踪协议
os.environ["LANGCHAIN_TRACING_V2"] = "true"
# 设置 API Key（生产环境中应使用密钥管理服务而非硬编码）
os.environ["LANGCHAIN_API_KEY"] = "your-key"
# 指定项目名，所有追踪数据将归集到该项目下
os.environ["LANGCHAIN_PROJECT"] = "my-agent"
# 创建 LangSmith 客户端实例，用于查询和反馈
client = Client()
@traceable(run_type="chain", name="agent_execute")
def run_agent(query: str):
"""带追踪的 Agent 执行函数。
@traceable 会自动捕获函数的输入、输出、执行时间，
并注入一个 run_id 属性到函数对象上，可用于后续关联反馈。
"""
# 这里是 Agent 的核心逻辑（示例中简化处理）
result = {"answer": "处理结果", "tool_calls": 3}
# 为本次运行记录一条结构化反馈（例如由评估器自动评分）
client.create_feedback(
run_id=run_agent.run_id, # run_id 由 @traceable 自动注入
key="task_complexity", # 反馈的键名，如"任务复杂度"
score=0.8 # 分值，可表示 0~1 的归一化分数
)
return result
# 查询项目中的运行记录（可按执行顺序、错误状态等过滤）
runs = client.list_runs(
project_name="my-agent", # 限定项目
execution_order=1, # 只看顶层运行（不含子调用）
error=False # 排除出错的运行
)
# 遍历并打印每次运行的关键指标
for run in runs:
print(f"Run: {run.name}, Latency: {run.latency:.2f}s, Tokens: {run.total_tokens}")关键解读：
@traceable是零侵入埋点的核心。它会自动包装函数，在调用前后创建一个"运行"（Run）记录，包含输入、输出、耗时、Token 等信息。create_feedback将人类评分或自动评估分数与某次运行关联起来，为后续的优化提供标注。list_runs支持丰富的过滤条件，是事后分析的基础入口。
10.6.4 LangFuse 集成
LangFuse 是一款开源的 Agent 可观测性平台，支持自托管，适合对数据隐私要求较高的场景。它的 API 设计与 LangSmith 类似，但提供了更细粒度的步骤级追踪。
# 从 LangFuse SDK 导入主客户端
from langfuse import Langfuse
# 导入装饰器与上下文工具，用于在函数内部更新追踪信息
from langfuse.decorators import observe, langfuse_context
# 初始化 LangFuse 客户端，传入密钥与服务地址
langfuse = Langfuse(
secret_key="sk-lf-...", # 私钥，用于写入数据
public_key="pk-lf-...", # 公钥，用于只读访问
host="https://cloud.langfuse.com" # SaaS 或自托管地址
)
@observe() # @observe 会将函数标记为一个可追踪的"观察"节点
def agent_step(step_name: str, input_data: dict):
"""带 LangFuse 追踪的 Agent 单步执行。"""
# 更新当前追踪的元信息（追踪级别：名称、标签、版本等）
langfuse_context.update_current_trace(
name=f"agent_{step_name}", # 追踪名称，用于在面板中检索
metadata={"version": "1.2.0"}, # 记录 Agent 版本，便于版本对比
tags=["production", "agent"] # 打标签，支持按标签筛选
)
# 更新当前观察节点的输入与步骤元数据
langfuse_context.update_current_observation(
input=input_data, # 记录输入数据
metadata={"step": step_name} # 记录步骤名
)
# 执行实际的步骤逻辑（此处为占位）
result = process_step(input_data)
# 记录输出与资源使用量（Token 与成本）
langfuse_context.update_current_observation(
output=result, # 记录输出结果
usage={"tokens": 150, "cost": 0.002} # 记录 Token 消耗与估算成本
)
return result
# 事后为某条追踪打分（可用于人工评估或自动评估反馈）
langfuse.score(
trace_id="trace-xxx", # 目标追踪 ID
name="user_satisfaction", # 评分维度名
value=4.5, # 分值（1~5 制）
comment="回答准确" # 评语
)与 LangSmith 的对比：
| 对比维度 | LangSmith | LangFuse |
|---|---|---|
| 部署方式 | SaaS 为主 | SaaS + 开源自托管 |
| 生态绑定 | 深度绑定 LangChain | 框架无关 |
| 成本 | 按用量收费 | 开源免费（自托管） |
| 数据隐私 | 数据存储在 LangChain 云端 | 可完全私有化 |
对于数据不能出内网的企业场景，LangFuse 自托管是更稳妥的选择。
10.6.5 成本归因
Agent 的每一次 LLM 调用都涉及真金白银的消耗。如果不能精确地知道"钱花在了哪里"，就无从进行成本优化。成本归因的目标是将总开销按模型、用户、功能模块等维度拆分，让每一分钱都可追溯。
# 导入数据类装饰器，用于定义简洁的数据结构
from dataclasses import dataclass, field
# 导入 defaultdict，用于按维度聚合成本
from collections import defaultdict
@dataclass
class CostTracker:
"""追踪每个请求的成本，支持多维度归因。"""
model_costs: dict # 模型单价表：{model_name: {"input": x, "output": y}}
records: list = None # 存储所有成本记录的列表
def __post_init__(self):
"""dataclass 的初始化后钩子，用于初始化可变默认值。"""
self.records = [] # 避免使用可变默认参数的陷阱
def record(self, model: str, input_tokens: int, output_tokens: int,
user_id: str = None, feature: str = None):
"""记录一次 LLM 调用的成本。
参数:
model: 使用的模型名称（如 gpt-4o）
input_tokens: 输入 Token 数量
output_tokens: 输出 Token 数量
user_id: 发起请求的用户 ID（可选，用于按用户归因）
feature: 功能模块名（可选，用于按功能归因）
"""
# 计算输入成本：Token 数 / 1000 * 单价
input_cost = input_tokens / 1000 * self.model_costs[model]["input"]
# 计算输出成本：同理
output_cost = output_tokens / 1000 * self.model_costs[model]["output"]
# 总成本 = 输入 + 输出
total_cost = input_cost + output_cost
# 将本次调用的完整信息记录下来
self.records.append({
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": total_cost,
"user_id": user_id,
"feature": feature
})
return total_cost # 返回本次成本，方便即时展示
def summary(self):
"""生成成本汇总报告，按模型和用户维度聚合。"""
# 计算总成本
total = sum(r["cost"] for r in self.records)
# 按模型聚合成本
by_model = defaultdict(float)
# 按用户聚合成本
by_user = defaultdict(float)
for r in self.records:
by_model[r["model"]] += r["cost"] # 累加到对应模型
by_user[r["user_id"] or "unknown"] += r["cost"] # 累加到对应用户
return {
"total_cost": total, # 总成本
"total_requests": len(self.records), # 总请求数
"by_model": dict(by_model), # 按模型分摊的成本
"by_user": dict(by_user) # 按用户分摊的成本
}
# 使用示例
tracker = CostTracker({
"gpt-4o": {"input": 0.0025, "output": 0.01}, # GPT-4o 单价（美元/千Token）
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006} # GPT-4o-mini 单价
})
# 记录一次 GPT-4o 调用：500 输入 Token + 300 输出 Token
tracker.record("gpt-4o", 500, 300, user_id="user_1", feature="chat")
# 打印成本汇总
print(tracker.summary())成本优化的常见思路：
- 模型降级：对简单任务使用更便宜的模型（如将 GPT-4o 降级为 GPT-4o-mini），可节省 90% 以上成本。
- 缓存复用：对重复性高的请求引入缓存层，避免重复调用 LLM。
- Token 裁剪：通过 Prompt 压缩、历史消息摘要等方式减少输入 Token。
10.6.6 Phoenix (Arize) 可观测性
Phoenix 是 Arize 公司推出的开源可观测性工具，基于 OpenTelemetry 标准，特别擅长 Embedding 可视化与检索质量分析。
# 导入 Phoenix 主模块
import phoenix as px
# 导入 LangChain 的自动埋点器
from phoenix.trace.langchain import LangChainInstrumentor
# 启动 Phoenix 本地服务（默认端口 6006）
session = px.launch_app()
# 自动检测 LangChain 调用并注入追踪（零代码侵入）
LangChainInstrumentor().instrument()
# 如果需要手动记录自定义 Span，可以使用 OpenTelemetry API
from opentelemetry import trace
# 获取一个 Tracer 实例
tracer = trace.get_tracer(__name__)
# 创建一个名为 "rag_retrieval" 的 Span
with tracer.start_as_current_span("rag_retrieval") as span:
# 在 Span 上设置自定义属性，用于后续分析
span.set_attribute("num_docs", 5) # 检索到的文档数
span.set_attribute("retrieval_time_ms", 120) # 检索耗时（毫秒）
# ... 在此处执行实际的检索逻辑Phoenix 的独特价值在于它提供了 Embedding 可视化 能力——你可以在面板中直观地看到检索文档在向量空间中的分布，快速发现"语义漂移"或"召回偏差"等问题。
10.6.7 自定义监控面板
在生产环境中，除了使用第三方平台，有时也需要搭建轻量级的自定义监控。下面的代码实现了一个滑动窗口式的 Agent 运行时监控器：
import time
# deque 是双端队列，设置 maxlen 后自动丢弃旧数据，适合滑动窗口
from collections import deque
from dataclasses import dataclass, field
@dataclass
class AgentMonitor:
"""Agent 运行时监控器，基于滑动窗口统计。"""
window_size: int = 100 # 滑动窗口大小：保留最近 100 条记录
# 延迟记录队列（自动保留最近 window_size 条）
latencies: deque = field(default_factory=lambda: deque(maxlen=100))
# Token 使用量记录队列
token_usage: deque = field(default_factory=lambda: deque(maxlen=100))
error_counts: int = 0 # 错误计数
total_requests: int = 0 # 总请求数
def record(self, latency: float, tokens: int, error: bool = False):
"""记录一次请求的指标。"""
self.latencies.append(latency) # 记录延迟（秒）
self.token_usage.append(tokens) # 记录 Token 消耗
self.total_requests += 1 # 总请求数 +1
if error:
self.error_counts += 1 # 错误时累加错误计数
def get_stats(self):
"""计算并返回当前窗口内的统计指标。"""
if not self.latencies:
return {"status": "no_data"} # 无数据时返回占位
latencies = list(self.latencies) # 转为列表以便排序
tokens = list(self.token_usage)
return {
"total_requests": self.total_requests, # 累计请求数
"error_rate": self.error_counts / max(self.total_requests, 1), # 错误率
"latency": {
"avg": sum(latencies) / len(latencies), # 平均延迟
"p50": sorted(latencies)[len(latencies)//2], # 中位数
"p95": sorted(latencies)[int(len(latencies)*0.95)], # P95 延迟
"p99": sorted(latencies)[int(len(latencies)*0.99)] # P99 延迟
},
"tokens": {
"avg_per_request": sum(tokens) / len(tokens), # 平均 Token/请求
"total": sum(tokens) # 总 Token 消耗
}
}
# 使用示例：模拟 50 次请求
monitor = AgentMonitor()
for i in range(50):
# 模拟延迟（0~1 秒之间随机）、Token 用量递增、每 20 次出一次错
monitor.record(time.time() % 1.0, 200 + i * 10, error=(i % 20 == 0))
# 输出统计结果
import json
print(json.dumps(monitor.get_stats(), indent=2))指标解读：
- P50/P95/P99 延迟：分别表示 50%、95%、99% 的请求在多少时间内完成。P99 延迟是发现"长尾问题"的关键指标——如果 P99 远高于 P50，说明有少量请求异常缓慢。
- 错误率：错误请求数 / 总请求数。通常生产环境要求错误率低于 1%。
- 平均 Token/请求：反映 Agent 的 Token 效率，是成本控制的核心指标。
10.6.8 告警体系
光有监控还不够，还需要在指标异常时主动告警。以下是告警体系的设计要点：
| 告警类型 | 触发条件 | 建议阈值 | 响应动作 |
|---|---|---|---|
| 错误率告警 | 滑动窗口内错误率超阈值 | > 5% | 立即通知值班人员 |
| 延迟告警 | P95 延迟超阈值 | > 30s | 降级模型或限流 |
| 成本告警 | 日成本超预算 | > 预算的 120% | 触发成本优化策略 |
| Token 异常 | 单次请求 Token 超阈值 | > 10000 | 检查是否存在无限循环 |
| 可用性告警 | 连续健康检查失败 | > 3 次 | 自动切换到备用模型 |
告警设计的一条重要原则是：宁可少报，不可滥报。频繁的误报会导致"告警疲劳"，使真正重要的告警被淹没。
10.6.9 常见误区
在实际工程中，团队在搭建 Agent 可观测性时容易陷入以下误区：
误区一：只记录成功请求
很多团队只追踪正常完成的请求，过滤掉错误请求的追踪数据。这恰恰丢掉了最有价值的信息——错误请求的追踪记录是定位 Bug 的第一手资料。正确做法是全量记录，在分析时再按需过滤。
误区二：日志等同于可观测性
有些团队认为"我们已经有日志了，所以可观测性没问题"。但日志只是三支柱之一。没有指标就无法快速感知系统整体状态，没有追踪就无法理解多步推理的因果链路。三者缺一不可。
误区三：可观测性是上线后才需要的事
不少团队在开发阶段完全不接入可观测性工具，等到上线后遇到问题才匆忙补上。此时往往缺乏历史基线数据，无法判断"是变差了还是一直如此"。正确做法是从第一天起就接入追踪，让可观测性成为开发流程的一部分。
误区四：追踪粒度不当
追踪粒度过粗（只记录顶层请求）会丢失子步骤信息，无法定位具体是哪一步出了问题；追踪粒度过细（记录每行代码）则会产生海量数据，拖慢系统性能。合理的做法是在 Agent 的每次 LLM 调用、工具调用、检索操作处设置追踪节点，既不过粗也不过细。
误区五：忽视成本维度
很多团队只关注延迟和正确率，忽视了成本可观测性。在 Agent 场景下，成本是与质量、延迟并列的第三维度——一个延迟很低、质量很好但每次调用花费 10 美元的 Agent，在实际部署中同样不可接受。
10.6.10 工具选型决策
面对众多可观测性工具，如何选择？以下是一份决策参考：
是否使用 LangChain 生态？
├── 是 -> 优先考虑 LangSmith（集成度最高）
│ └── 数据能否上云？
│ ├── 是 -> 直接使用 LangSmith SaaS
│ └── 否 -> 使用 LangFuse 自托管
└── 否 -> 是否需要 Embedding 可视化？
├── 是 -> Phoenix (Arize)
└── 否 -> LangFuse（框架无关，功能全面）在实际项目中，也可以组合使用多种工具：例如用 LangSmith 做开发期的追踪分析，用 LangFuse 做生产期的持久化记录，用 Phoenix 做 Embedding 诊断。
10.6.11 本节小结
本节围绕 Agent 可观测性展开了系统性讨论，核心要点如下：
| 要点 | 说明 |
|---|---|
| 三支柱 | 日志（Logs）、指标（Metrics）、追踪（Traces）构成可观测性的基石 |
| Agent 特殊性 | 非确定性、多步链路、成本耦合、用户感知差异要求更细粒度的可观测 |
| LangSmith | LangChain 官方追踪平台，@traceable 装饰器实现零侵入埋点 |
| LangFuse | 开源替代方案，支持自托管，适合数据隐私要求高的场景 |
| 成本归因 | 按模型/用户/功能维度追踪 Token 消耗，是成本优化的前提 |
| Phoenix | 基于 OpenTelemetry 的开源方案，擅长 Embedding 可视化 |
| 自定义监控 | 滑动窗口统计 + 分位数延迟 + 错误率，轻量级方案 |
| 告警体系 | 错误率、延迟、成本、Token 异常等多维度告警，注意避免告警疲劳 |
| 常见误区 | 全量记录、三支柱缺一不可、开发期即接入、追踪粒度适中、重视成本 |
一句话总结：可观测性不是"锦上添花"，而是 Agent 系统的"生命体征监测仪"——没有它，你永远不知道系统是在健康运行还是在慢性崩溃。
至此，第 10 章"评估与优化"的内容已全部讲解完毕。我们从评估指标体系出发，依次讨论了自动评估、人工评估、A/B 测试、反馈闭环，最终落脚于可观测性——这条从"度量"到"看见"再到"改进"的完整链路，构成了 Agent 持续优化的闭环。
然而，无论 Agent 多么智能、可观测性多么完善，安全与风险始终是不可回避的底线问题。Agent 拥有了工具调用、代码执行、网络访问的能力，也意味着它拥有了造成真实世界损害的能力。在第 11 章中，我们将深入探讨 Agent 的安全与风险：从 Prompt 注入防御到工具权限管控，从越权防护到审计合规，确保 Agent 在发挥价值的同时不会成为安全隐患的放大器。