Agent可观测性:如何监控、调试、追踪你的智能体?(附完整方案) 🔍

导读:Agent上线了,但你不知道它在想什么、做了什么、为什么出错?可观测性(Observability) 是Agent从Demo走向生产的必备基础设施!今天给你一套完整的监控、调试、追踪方案!📊


一、Agent可观测性为什么难?🤔

1.1 传统应用 vs Agent应用

维度 传统应用 Agent应用
执行路径 确定性(代码写死的) 非确定性(LLM决策)
延迟 可预测 波动大(几秒到几分钟)
成本 主要是计算资源 Token消耗(按次计费)
错误类型 异常/超时 幻觉/死循环/工具调用失败
调试难度 看日志就行 需要看"思考过程"

💡 核心挑战:Agent的每一步决策都是LLM"想"出来的,你需要看到它的思考链,而不只是输入输出。

1.2 可观测性三大支柱

🔍 Agent可观测性

📋 日志 Logging

📊 指标 Metrics

🔗 追踪 Tracing

每步决策日志

工具调用记录

错误堆栈

响应延迟

Token消耗

成功率

完整执行链路

每步耗时

输入输出快照


二、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追踪界面

🔍 Trace

🤖 LLM Call
耗时: 2.3s
Token: 1500

🔧 Tool Call
search_web
耗时: 1.1s

🤖 LLM Call
耗时: 1.8s
Token: 800

🔧 Tool Call
read_file
耗时: 0.3s

🤖 LLM Call
耗时: 1.2s
Token: 600


三、自定义日志系统 📋

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 监控面板

📊 Agent监控面板

⚡ 延迟
P50: 3.2s
P99: 12.5s

💰 Token
今日: 2.3M
成本: ¥156

✅ 成功率
98.5%
↑0.3%

🔄 步数
平均: 4.2步
最大: 12步


五、调试技巧 🐛

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期,生产落地篇进行中…

作者:高炉炼铁智能化技术研究者,专注钢铁冶金与人工智能 交叉领域。

👍 如果觉得有帮助,请点赞、收藏、转发!
版权归作者所有,未经许可请勿抄袭,套用,商用(或其它具有利益性行为)
🔔 关注专栏,不错过后续精彩内容

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐