AI Agent 从入门到封神:24 讲打造你的超级智能体~系列文章21:Agent可观测性:如何监控、调试、追踪你的智能体?
·
Agent可观测性:如何监控、调试、追踪你的智能体?(附完整方案) 🔍
导读:Agent上线了,但你不知道它在想什么、做了什么、为什么出错?可观测性(Observability) 是Agent从Demo走向生产的必备基础设施!今天给你一套完整的监控、调试、追踪方案!📊
一、Agent可观测性为什么难?🤔
1.1 传统应用 vs Agent应用
| 维度 | 传统应用 | Agent应用 |
|---|---|---|
| 执行路径 | 确定性(代码写死的) | 非确定性(LLM决策) |
| 延迟 | 可预测 | 波动大(几秒到几分钟) |
| 成本 | 主要是计算资源 | Token消耗(按次计费) |
| 错误类型 | 异常/超时 | 幻觉/死循环/工具调用失败 |
| 调试难度 | 看日志就行 | 需要看"思考过程" |
💡 核心挑战:Agent的每一步决策都是LLM"想"出来的,你需要看到它的思考链,而不只是输入输出。
1.2 可观测性三大支柱
二、LangSmith:Agent的"X光机" 🏥
2.1 什么是LangSmith?
LangSmith = LangChain官方的可观测性平台,专门用于追踪、调试、评估Agent。
| 功能 | 说明 |
|---|---|
| 🔗 Trace追踪 | 可视化Agent的每一步执行 |
| 📊 指标监控 | 延迟、Token、成功率等 |
| 🧪 评估测试 | 自动化评估Agent质量 |
| 🐛 调试工具 | 回放任意一次执行 |
| 📈 数据分析 | 趋势分析、异常检测 |
2.2 接入LangSmith
import os
# 配置LangSmith
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-project"
# 之后的所有LangChain调用都会自动追踪
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
llm = ChatOpenAI(model="gpt-4o")
agent = create_agent(model=llm, tools=[...])
# 运行Agent - 自动记录到LangSmith
result = agent.invoke({"messages": [...]})
2.3 LangSmith追踪界面
三、自定义日志系统 📋
3.1 Agent日志框架
import logging
import json
import time
from functools import wraps
# 配置日志
logger = logging.getLogger("agent")
handler = logging.FileHandler("/tmp/agent.log")
handler.setFormatter(logging.Formatter(
'{"time":"%(asctime)s","level":"%(levelname)s","msg":%(message)s}'
))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
class AgentLogger:
"""Agent专用日志记录器"""
@staticmethod
def log_llm_call(model, prompt_tokens, completion_tokens, latency_ms):
logger.info(json.dumps({
"event": "llm_call",
"model": model,
"prompt_tokens": prompt_tokens,
"completion_tokens": completion_tokens,
"total_tokens": prompt_tokens + completion_tokens,
"latency_ms": latency_ms
}))
@staticmethod
def log_tool_call(tool_name, input_data, output_data, latency_ms, success=True):
logger.info(json.dumps({
"event": "tool_call",
"tool": tool_name,
"input": str(input_data)[:200],
"output": str(output_data)[:200],
"latency_ms": latency_ms,
"success": success
}))
@staticmethod
def log_agent_step(step_num, thought, action, observation):
logger.info(json.dumps({
"event": "agent_step",
"step": step_num,
"thought": thought[:200],
"action": action,
"observation": str(observation)[:200]
}))
@staticmethod
def log_error(error_type, error_msg, context=None):
logger.error(json.dumps({
"event": "error",
"type": error_type,
"message": error_msg,
"context": str(context)[:500] if context else None
}))
3.2 中间件集成
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage
@wrap_tool_call
def logging_middleware(request, handler):
"""记录每次工具调用的中间件"""
tool_name = request.tool_call["name"]
tool_input = request.tool_call["args"]
start = time.time()
try:
result = handler(request)
latency = int((time.time() - start) * 1000)
AgentLogger.log_tool_call(tool_name, tool_input, result.content, latency)
return result
except Exception as e:
latency = int((time.time() - start) * 1000)
AgentLogger.log_tool_call(tool_name, tool_input, str(e), latency, success=False)
return ToolMessage(content=f"Error: {e}", tool_call_id=request.tool_call["id"])
四、关键监控指标 📊
4.1 指标体系
| 类别 | 指标 | 说明 | 告警阈值 |
|---|---|---|---|
| ⚡ 性能 | 响应延迟P50 | 50%请求的延迟 | > 5s |
| ⚡ 性能 | 响应延迟P99 | 99%请求的延迟 | > 30s |
| 💰 成本 | 单次Token消耗 | 每次请求的Token | > 10000 |
| 💰 成本 | 日均成本 | 每日API费用 | > 预算 |
| ✅ 质量 | 任务成功率 | 成功完成的比例 | < 90% |
| ✅ 质量 | 工具调用成功率 | 工具调用成功比例 | < 95% |
| 🔄 循环 | 平均步数 | Agent完成任务的步数 | > 10步 |
| 🔄 循环 | 死循环检测 | 连续重复动作 | > 3次重复 |
4.2 监控面板
五、调试技巧 🐛
5.1 常见问题排查表
| 问题 | 可能原因 | 排查方法 |
|---|---|---|
| 🔄 死循环 | Agent反复调用同一工具 | 查看Trace,检查重复动作 |
| 💀 幻觉 | LLM编造信息 | 对比工具返回和Agent输出 |
| ⏰ 超时 | 工具调用太慢/步数太多 | 检查各步骤耗时 |
| 💸 成本过高 | Token消耗过多 | 分析Prompt长度和步数 |
| ❌ 工具失败 | 参数错误或权限不足 | 查看工具调用日志 |
| 🤷 答非所问 | System Prompt不够清晰 | 检查Prompt和输入 |
5.2 回放调试
# 保存每次执行的完整记录
def save_execution_trace(thread_id, state):
"""保存执行轨迹,用于回放调试"""
trace = {
"thread_id": thread_id,
"timestamp": time.time(),
"messages": [
{
"role": msg.type if hasattr(msg, 'type') else msg["role"],
"content": msg.content if hasattr(msg, 'content') else msg["content"]
}
for msg in state["messages"]
],
"total_steps": len([m for m in state["messages"] if hasattr(m, 'type') and m.type == "tool"]),
}
# 保存到文件/数据库
with open(f"/tmp/traces/{thread_id}.json", "w") as f:
json.dump(trace, f, ensure_ascii=False, indent=2)
六、可观测性工具对比 📊
| 工具 | 类型 | 优点 | 缺点 | 推荐场景 |
|---|---|---|---|---|
| LangSmith | 商业平台 | 功能全面,开箱即用 | 收费 | LangChain项目 |
| Phoenix | 开源 | 免费,自托管 | 需要自己部署 | 预算有限 |
| OpenTelemetry | 标准协议 | 通用,生态好 | 需要额外适配 | 多框架混合 |
| 自建日志 | 自定义 | 完全可控 | 开发成本高 | 特殊需求 |
七、本期小结 📝
| 知识点 | 核心内容 |
|---|---|
| 可观测性三支柱 | 日志、指标、追踪 |
| LangSmith | Agent专属的可观测性平台 |
| 关键指标 | 延迟、Token、成功率、步数 |
| 调试技巧 | 回放Trace、死循环检测 |
| 工具选择 | LangSmith/Phoenix/OTel/自建 |
🔥 没有可观测性的Agent = 黑箱! 你永远不知道它在想什么、为什么出错。把可观测性作为Agent上线的第一优先级,而不是事后补救!
📢 下期预告:《Agent安全攻防:Prompt注入、工具滥用、数据泄露,三大风险与防御策略》—— Agent的安全问题,比你想的更严重!🛡️
📌 三连走起!可观测性,让Agent不再是黑箱! 💪
📚 专栏第21/24期,生产落地篇进行中…
作者:高炉炼铁智能化技术研究者,专注钢铁冶金与人工智能 交叉领域。
👍 如果觉得有帮助,请点赞、收藏、转发!
版权归作者所有,未经许可请勿抄袭,套用,商用(或其它具有利益性行为)。
🔔 关注专栏,不错过后续精彩内容
更多推荐



所有评论(0)