当你的Agent在多步推理中悄悄"发疯"，你连它在哪里死的都不知道

在这里插入图片描述

一、引子：一个让所有开发者后背发凉的故事

2025年3月，Fedora基础设施团队经历了一场噩梦。

一个自动化CI脚本在清理临时构建目录时，因为环境变量泄露导致 rm -rf $BUILD_DIR 中的变量解析为空——变成了 rm -rf /。等运维发现时，构建服务器上的 /usr、/etc、/var 已经被"梳洗"了一遍，整个系统接近瘫痪。

事后复盘发现：这个脚本经过了6步处理流程，第2步的环境检查通过了，第3步的权限验证也通过了，第4步居然还打印了一行 “Cleaning up: /tmp/build-12345”。问题出在第5步——一个看似无害的路径拼接函数中，$BUILD_DIR 在被赋值前就被另一个函数副作用置空了。而这个中间状态，没有任何日志。

这个故事和AI Agent有什么关系？

关系大了。

你的Agent正在做的事情，本质上和这个脚本一模一样：

用户提问 → 意图识别 → 工具选择 → 参数构造 → 工具调用 → 结果解析 → 推理决策 → 下一步行动……

每一步都有可能出错。而且，Agent的多步推理比脚本复杂一万倍——因为每一步都是LLM生成的，非线性、非确定性、非可复现。

二、AI Agent为什么需要可观测性？

2.1 传统监控已经失效

先问一个问题：你用什么来监控你的Agent？

大概率还停留在这三板斧：

监控维度 传统服务的监控方式 Agent场景下失效的原因
响应状态 检查HTTP状态码200/500 工具调用返回200但推理路径完全跑偏，状态码无法反映"是否做了正确的事"
延迟 P99响应时间 > 5s告警 即使延迟正常，Agent可能在第一个步骤就误解了意图，后面的所有操作都是"无用功但耗时正常"
错误率 5xx错误占比超阈值 Agent最常见的"错误"不是抛异常，而是用错了工具、用了错的参数、或者编造了一个看似合理的答案

这种断层的根本原因在于：传统监控关注的是服务是否在运行，而Agent场景需要关注的是服务是否在正确地思考。

Agent出问题的方式和传统服务有本质区别：

  • 服务没挂，但Agent在兜圈子：绕了5轮工具调用后说"我不知道"，每轮API都是200
  • API返回200，但Agent"发疯"：成功调用了计算器API算出2+2=4，却在下一次推理中声称"计算结果为42"
  • 没有报错，但路径诡异：查天气的需求，Agent先去调了地图API定位，又调了百科API查城市信息，最后才碰天气API——多消耗了3倍token
  • 复现不可能：同样的prompt、同样的温度参数、同样的工具列表，两次跑出来的推理路径可能完全不同
  • 幻觉被包装成事实：没有调用任何工具，Agent就"自信"地回答了一个需要实时数据的问题，用户完全看不出来

2.2 可观测性的三个层次

Agent可观测性不是要不要的问题，而是做几个层次的问题。我认为需要覆盖三个递进层次：

L1 失败检测 → L2 问题定位 → L3 行为优化

L1：失败检测

最基本的需求——Agent出问题了你要能知道。

  • Agent执行超时（比如超过30秒还在转圈）
  • 工具调用连续失败（比如API密钥过期）
  • Token消耗异常（一次对话用了几十万token）

这里的挑战是：Agent所谓的"失败"不一定是抛异常。更多时候是"看似正常地做了一件错误的事"。

L2：问题定位

知道出事了还不够，你得知道哪一步出的问题。

  • 意图识别错了？还是工具选错了？
  • 工具调用成功但结果解析错了？
  • 参数格式错了？还是工具本身返回了异常数据？

这要求每一轮的输入、输出、中间状态都被结构化的记录下来，而不是写一堆print然后用人眼去翻。

L3：行为优化

最高层次——基于追踪数据持续改进Agent行为：

  • 发现频繁错误选择某个工具 → 优化该工具的description
  • 发现某类意图识别准确率低 → 改进system prompt中的分类示例
  • 发现某条推理路径消耗过多token → 增加路径约束

没有L1和L2，L3就是空中楼阁。大部分团队的问题不是"不想优化"，而是连出了什么问题都不知道。

2.3 从分布式追踪到Agent追踪

如果你做过微服务开发，一定接触过分布式追踪（Distributed Tracing）。它的核心思想是：用一个trace ID贯穿所有服务调用，把一次请求的完整调用链路串起来。

Agent的可观测性本质是分布式追踪在LLM推理层的一次平移：

分布式追踪中的概念 Agent场景的对应物
一次HTTP请求 一次Agent调用（用户提问→最终回答）
Trace ID Agent执行的全局唯一标识
Span（服务A→服务B） Observation（LLM调用、工具调用、推理步骤）
父Span/子Span 嵌套的推理步骤（意图识别→工具调用→结果解析）
错误日志 LLM生成结果的质量评分
服务拓扑 Agent的决策路径拓扑

理解了这一点，就理解了Langfuse和Opik的底层设计逻辑——它们就是把分布式追踪那一套成熟的方法论，搬到了LLM和Agent这个全新的领域。

2.4 一个真实的生产教训

2024年底，某金融科技公司的故事在业内广为流传。

他们在生产环境部署了一个信贷审核Agent，设计流程如下：

用户提交材料 → Agent提取信息 → 调用征信API → 调用风控模型 → 综合决策 → 输出结果

看起来很完美，每个步骤职责清晰。

上线第二天，大量用户投诉"明明征信良好却被拒贷"。客服团队淹没在投诉中，业务方怒不可遏。

工程师排查了整整两天。第一天检查了征信API的调用日志——全部成功返回200。第二天手工插入了三十多个print日志，跑了几十遍复现，终于发现了真相：

  • 第2步：Agent正确提取了用户信息 ✅
  • 第3步：调用征信API成功，返回分数680（良好） ✅
  • 第4步：Agent自己脑补了一个"风险等级=高"的判断——在做这个判断时，它根本没有去调用风控模型，而是基于"征信报告中提到过一笔逾期记录"自己做了推理
  • 第5步：在这个错误的自作主张之上做综合决策，直接拒贷 ❌

风控模型根本就没被调用。

更可怕的是：没有任何监控告警发现了这个问题。因为：

  • 征信API调用成功了（返回200）
  • Agent最终返回了一个格式正确的答案（“尊敬的客户，您的贷款申请未通过审核”）
  • 没有抛异常、没有超时、所有指标看起来都正常

这就是Agent「多步推理黑盒问题」最典型、最昂贵的表现形式。

如果当时他们有可观测系统，追踪面板会清晰地显示：

credit_agent_trace (trace_id=xxx)
  1. info_extraction ✅ (312ms)
  2. credit_api_call ✅ (845ms, score=680)
  3. risk_model_call ❌ **没有被调用**
  4. decision_making ✅ (212ms, 使用了无来源的"高风险"判断)

问题在第3步就暴露无遗，根本不需要两天的排查。

三、多步推理黑盒问题的本质

3.1 经典"三步出错"模型

Agent的多步推理可以抽象为：

[Step 1] Input → LLM推理 → 决策/输出
[Step 2] 决策 → 工具调用 → 结果
[Step 3] 结果 → LLM推理 → 下一步
...
[Step N] 最终输出

每一步都有6种出错的模式：

模式 表现 原因
理解偏差 误解用户意图 prompt模糊
工具错选 选择了错误的API tool description不清晰
参数错配 正确的工具，参数格式错误 参数schema不合理
幻觉填充 没有调用任何工具就"编"出了答案 LLM过度自信
路径发散 正常路径走偏，开始兜圈子 缺少目标约束
级联失败 前一步的小错误被后续步骤放大 每步都在"离题万里"的基础上继续推理

3.2 为什么"加几个print"解决不了？

传统开发者的第一反应：加日志啊。

但在Agent场景下，这条路走不通：

  1. 每一步都是LLM生成的文本：不是固定代码路径，日志量失控
  2. 推理过程在模型内部：你只能看到输入和输出，中间的"思考过程"（如果是CoT可能能看到一部分，但结构化信息基本没有）
  3. 工具调用的上下文丢失：第3步用了第2步的结果，但第2步的完整结果已经被截断或合并了
  4. 关联性极差：日志之间没有trace ID关联，你根本不知道哪个日志属于哪个请求的哪一步

你需要的是分布式追踪的思维方式，应用到Agent场景。

四、Langfuse入门：开源LLM可观测标杆

Langfuse（https://github.com/langfuse/langfuse）是目前最流行的开源LLM可观测平台，GitHub 40k+ Stars。它把分布式追踪的思想引入LLM/Agent领域，提供了完整的Tracing、Evaluation、Prompt Management能力。

4.1 安装部署

方式一：Docker Compose（推荐）

# 创建项目目录
mkdir langfuse-docker && cd langfuse-docker

# 下载官方docker-compose.yml
curl -o docker-compose.yml https://raw.githubusercontent.com/langfuse/langfuse/main/docker-compose.yml

# 生成安全密钥
openssl rand -hex 32  # 作为 ENCRYPTION_KEY
openssl rand -hex 16  # 作为 SALT

# 编辑 docker-compose.yml，修改以下环境变量：
# - ENCRYPTION_KEY: 上面生成的值
# - SALT: 上面生成的值
# - LANGFUSE_INIT_USER_NAME: admin
# - LANGFUSE_INIT_USER_PASSWORD: 你的密码

# 启动
docker compose up -d

启动后访问 http://localhost:3000，用设置的用户名密码登录。

方式二：pip 快速接入（不部署服务端，使用Langfuse云）

pip install langfuse openai

注册 Langfuse Cloud 获取 LANGFUSE_PUBLIC_KEY 和 LANGFUSE_SECRET_KEY。

4.2 基础配置

import os
from langfuse import Langfuse

# 方式一：环境变量
os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxxxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxxxx"
os.environ["LANGFUSE_HOST"] = "http://localhost:3000"  # 自托管地址

# 方式二：显式初始化
langfuse = Langfuse(
    secret_key="sk-lf-xxxxx",
    public_key="pk-lf-xxxxx",
    host="http://localhost:3000"
)

# 验证连接
assert langfuse.auth_check(), "Langfuse连接失败"

4.3 核心概念

Langfuse的追踪模型设计非常精妙，只有三个核心概念：

Trace（追踪）
 ├── name: "agent-execution"
 ├── input: 用户问题
 ├── output: 最终回答
 │
 ├── Observation: Generation（LLM调用）
 │   ├── name: "llm-call-1"
 │   ├── input: prompt
 │   ├── output: 模型回复
 │   ├── model: "gpt-4o"
 │   ├── usage: {prompt_tokens, completion_tokens}
 │   │
 │   └── Observation: Generation（子LLM调用）
 │       └── ...
 │
 ├── Observation: Span（工具调用/函数执行）
 │   ├── name: "search-tool"
 │   ├── input: {"query": "天气"}
 │   ├── output: {"temperature": 28}
 │   └── duration: 1.2s
 │
 └── Score（评分/评估指标）
     ├── name: "correctness"
     └── value: 0.95

4.4 基本使用：追踪LLM调用

方法一：装饰器模式（最简单）

from langfuse.decorators import observe, langfuse_context
from langfuse.openai import openai  # 注意：这是Langfuse包装后的OpenAI

@observe()
def generate_story(topic: str) -> str:
    langfuse_context.set_current_trace(
        name="story-generation",
        session_id="session-001",
        user_id="user-001"
    )
    
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是一个创意写手"},
            {"role": "user", "content": f"写一个关于{topic}的短故事"}
        ],
        temperature=0.7
    )
    
    langfuse_context.update_current_observation(
        input=topic,
        output=response.choices[0].message.content,
        usage=dict(response.usage)
    )
    
    return response.choices[0].message.content

story = generate_story("AI机器人")

方法二：低阶API（更灵活）

from langfuse import Langfuse
from openai import OpenAI

langfuse = Langfuse()
client = OpenAI()

# 创建trace
trace = langfuse.trace(
    name="chat-completion",
    input="什么是Agent可观测性？",
    session_id="session-123"
)

# 创建generation（追踪LLM调用）
generation = trace.generation(
    name="llm-call-1",
    model="gpt-4o",
    model_parameters={"temperature": 0.7, "max_tokens": 1000},
    input=[{"role": "user", "content": "什么是Agent可观测性？"}]
)

# 调用LLM
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "什么是Agent可观测性？"}]
)

# 记录结果
generation.end(
    output=response.choices[0].message.content,
    usage={
        "input": response.usage.prompt_tokens,
        "output": response.usage.completion_tokens,
        "unit": "TOKENS"
    }
)

trace.update(output=response.choices[0].message.content)

五、Opik入门：Comet的开源可观测方案

Opik（https://github.com/comet-ml/opik）是Comet团队推出的开源LLM评估与监控平台，定位是LLM应用的调试、评估和生产监控一体化解决方案。

5.1 安装部署

# 使用pip安装
pip install opik

# 或者使用poetry
poetry add opik

Opik服务部署：

# 克隆仓库
git clone https://github.com/comet-ml/opik.git
cd opik/deployment/installer

# 使用docker-compose启动
docker compose up -d

5.2 配置与初始化

import opik

# 方式一：本地自托管
opik.configure(use_local=True)

# 方式二：Comet云服务
# opik.configure(api_key="YOUR_API_KEY", workspace="YOUR_WORKSPACE")

# 方式三：环境变量
# export OPIK_URL_OVERRIDE=http://localhost:5173
# export OPIK_API_KEY=your-api-key

5.3 核心API使用

基础追踪：

import opik
from openai import OpenAI

opik.configure(use_local=True)

@opik.track()
def ask_llm(prompt: str) -> str:
    """追踪这个函数的输入输出和耗时"""
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

result = ask_llm("什么是Agent可观测性？")

追踪OpenAI调用（自动捕获token用量）：

import opik
from opik.integrations.openai import track_openai

opik.configure(use_local=True)

# 包装OpenAI客户端
openai_client = OpenAI()
track_openai(openai_client)  # 自动追踪所有后续调用

response = openai_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "解释多步推理"}]
)
# 这个调用会被自动追踪，包括prompt、completion、token用量

嵌套追踪：

import opik

@opik.track()
def step_1(query: str) -> str:
    """第一步：意图分析"""
    return "查询天气"

@opik.track()
def step_2(intent: str) -> dict:
    """第二步：调用工具"""
    return {"temperature": 28, "humidity": 60}

@opik.track()
def step_3(data: dict) -> str:
    """第三步：生成回答"""
    return f"当前温度{data['temperature']}°C"

@opik.track()
def agent_pipeline(query: str) -> str:
    intent = step_1(query)
    data = step_2(intent)
    answer = step_3(data)
    return answer

result = agent_pipeline("今天天气怎么样？")
# Opik自动展示调用树：
# agent_pipeline
#   ├── step_1 (33ms)
#   ├── step_2 (212ms)
#   └── step_3 (45ms)

5.4 Opik vs Langfuse：选型对比

维度 Langfuse Opik
部署复杂度 Docker Compose，含PostgreSQL+ClickHouse+Redis+MinIO 相对轻量
追踪粒度 Trace→Observation(Generation/Span/Event)三级 @track装饰器+OpenAI集成
评估能力 Score系统 + 集成RAGAS Dataset+Evaluate+LLM-as-a-judge
Prompt管理 内置Prompt管理 + Playground 基础支持
社区活跃度 40k+ Stars 相对较新，增长迅猛
定位 全栈LLM工程平台 LLM评估+监控专精
最佳场景 需要完整LLM DevOps流程的团队 以评估为中心的研发团队

建议： 如果团队刚起步，Langfuse的上手曲线更平缓，生态更成熟；如果团队对LLM评估有强诉求（自动评测、对比实验、回归测试），Opik的评估体系更完善。

六、实战：为Agent添加完整追踪

6.1 场景说明

我们构建一个"智能客服Agent"，功能：

  1. 用户提出问题
  2. Agent识别意图（查询订单/取消订单/退货/人工客服）
  3. 根据意图调用不同工具
  4. 综合结果生成回答

6.2 完整代码（基于Langfuse）

"""
智能客服Agent - 完整可观测性实现
依赖：pip install langfuse openai
"""

import os
import json
import time
from typing import Dict, Any, Optional

os.environ["LANGFUSE_PUBLIC_KEY"] = "pk-lf-xxxxx"
os.environ["LANGFUSE_SECRET_KEY"] = "sk-lf-xxxxx"
os.environ["LANGFUSE_HOST"] = "http://localhost:3000"
os.environ["OPENAI_API_KEY"] = "sk-your-openai-key"

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
from langfuse.openai import openai as langfuse_openai

# 初始化Langfuse
langfuse = Langfuse()
assert langfuse.auth_check(), "Langfuse连接失败"

# ============ 工具层 ============

# 模拟数据库查询
def query_order(order_id: str) -> Dict[str, Any]:
    """查询订单信息"""
    # 模拟数据库查询
    orders = {
        "ORD001": {"status": "已发货", "product": "AI鼠标", "price": 299, "date": "2025-01-15"},
        "ORD002": {"status": "待发货", "product": "机械键盘", "price": 599, "date": "2025-03-01"},
        "ORD003": {"status": "已取消", "product": "显示器", "price": 1999, "date": "2025-02-28"},
    }
    time.sleep(0.3)  # 模拟网络延迟
    return orders.get(order_id, {"status": "未找到", "error": f"订单{order_id}不存在"})

def cancel_order(order_id: str) -> Dict[str, Any]:
    """取消订单"""
    time.sleep(0.5)
    return {"success": True, "message": f"订单{order_id}已成功取消", "refund_amount": 299}

def return_order(order_id: str, reason: str) -> Dict[str, Any]:
    """申请退货"""
    time.sleep(0.4)
    return {"success": True, "message": f"订单{order_id}退货申请已提交", "reason": reason}

# ============ Agent核心 ============

@observe(name="intent-recognition")
def recognize_intent(user_input: str) -> str:
    """第一步：意图识别"""
    langfuse_context.update_current_observation(
        input=user_input
    )
    
    # 使用LLM进行意图分类
    response = langfuse_openai.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0.1,
        messages=[
            {"role": "system", "content": """
            你是一个客服意图识别系统。请将用户输入归类为以下之一：
            - query_order: 查询订单状态
            - cancel_order: 取消订单
            - return_product: 退货申请
            - human_service: 转人工客服
            - unknown: 无法识别
            
            只输出意图标签，不要输出其他内容。
            """},
            {"role": "user", "content": user_input}
        ]
    )
    
    intent = response.choices[0].message.content.strip()
    
    # 记录意图识别结果
    langfuse_context.update_current_observation(
        output=intent,
        metadata={"raw_response": intent}
    )
    
    return intent

@observe(name="parameter-extraction")
def extract_parameters(user_input: str, intent: str) -> Dict[str, Any]:
    """第二步：参数提取"""
    langfuse_context.update_current_observation(
        input={"user_input": user_input, "intent": intent}
    )
    
    response = langfuse_openai.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0.1,
        messages=[
            {"role": "system", "content": f"""
            从用户输入中提取所需参数。当前意图：{intent}
            
            如果意图是 query_order，需要参数：order_id
            如果意图是 cancel_order，需要参数：order_id
            如果意图是 return_product，需要参数：order_id, reason
            如果意图是 human_service，不需要参数
            如果意图是 unknown，不需要参数
            
            返回JSON格式。
            """},
            {"role": "user", "content": user_input}
        ]
    )
    
    try:
        params = json.loads(response.choices[0].message.content.strip().strip("`").replace("json\n", ""))
    except json.JSONDecodeError:
        params = {"error": "参数解析失败", "raw": response.choices[0].message.content}
    
    langfuse_context.update_current_observation(
        output=params
    )
    
    return params

@observe(name="tool-calling")
def call_tool(intent: str, params: Dict[str, Any]) -> Any:
    """第三步：工具调用（核心追踪点）"""
    langfuse_context.update_current_observation(
        name=f"tool-{intent}",
        input={"intent": intent, "params": params}
    )
    
    start_time = time.time()
    
    try:
        if intent == "query_order":
            result = query_order(params.get("order_id", ""))
        elif intent == "cancel_order":
            result = cancel_order(params.get("order_id", ""))
        elif intent == "return_product":
            result = return_order(params.get("order_id", ""), params.get("reason", ""))
        elif intent == "human_service":
            result = {"action": "transfer", "message": "正在转接人工客服..."}
        else:
            result = {"action": "clarify", "message": "请详细描述您的问题"}
    except Exception as e:
        result = {"error": str(e), "intent": intent}
    
    duration = time.time() - start_time
    
    langfuse_context.update_current_observation(
        output=result,
        metadata={
            "tool_duration_ms": round(duration * 1000, 2),
            "tool_name": intent
        }
    )
    
    return result

@observe(name="response-generation")
def generate_response(intent: str, tool_result: Any) -> str:
    """第四步：生成最终回答"""
    langfuse_context.update_current_observation(
        input={"intent": intent, "tool_result": tool_result}
    )
    
    response = langfuse_openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "你是一个友好的客服助手，用自然语言回复用户。"},
            {"role": "user", "content": f"""
            意图：{intent}
            工具调用结果：{json.dumps(tool_result, ensure_ascii=False)}
            
            请生成友好的回复。
            """}
        ]
    )
    
    answer = response.choices[0].message.content
    
    langfuse_context.update_current_observation(
        output=answer
    )
    
    return answer

# ============ Agent主流程 ============

@observe(name="customer-service-agent")
def customer_service_agent(user_input: str) -> str:
    """客服Agent主流程——带完整可观测性追踪"""
    
    # 设置trace元数据
    langfuse_context.set_current_trace(
        name="customer-service-agent",
        session_id=f"session-{int(time.time())}",
        metadata={
            "version": "v2.1.0",
            "environment": "production",
            "tags": ["customer-service", "production"]
        }
    )
    
    print(f"\n===== 用户输入: {user_input} =====")
    
    # Step 1: 意图识别
    intent = recognize_intent(user_input)
    print(f"[追踪] 意图识别 → {intent}")
    
    # Step 2: 参数提取
    params = extract_parameters(user_input, intent)
    print(f"[追踪] 参数提取 → {params}")
    
    # Step 3: 工具调用
    tool_result = call_tool(intent, params)
    print(f"[追踪] 工具调用结果 → {json.dumps(tool_result, ensure_ascii=False, indent=2)}")
    
    # 记录评分：工具调用是否成功
    if "error" in tool_result:
        langfuse_context.score(
            name="tool_success",
            value=0,
            comment=f"工具调用失败: {tool_result['error']}"
        )
    else:
        langfuse_context.score(
            name="tool_success",
            value=1,
            comment="工具调用成功"
        )
    
    # Step 4: 生成回答
    answer = generate_response(intent, tool_result)
    print(f"[追踪] 最终回答 → {answer}")
    
    # 更新trace的output
    langfuse_context.update_current_trace(output=answer)
    
    # 记录延迟评分
    langfuse_context.score(
        name="latency_category",
        value=1
    )
    
    return answer


# ============ 执行 ============

if __name__ == "__main__":
    test_cases = [
        "查询我的订单ORD001的状态",
        "我要取消订单ORD002",
        "帮我退货ORD003，键盘有按键不灵敏",
        "我想找人工客服",
    ]
    
    for test in test_cases:
        print(f"\n{'='*50}")
        result = customer_service_agent(test)
        print(f"[最终回复] {result}")
        print(f"{'='*50}")

6.3 输出解读

运行上述代码后，Langfuse控制台上会看到：

Trace级别视图：

customer-service-agent (session-session-123456)
├── intent-recognition
│   ├── input: "查询我的订单ORD001的状态"
│   └── output: "query_order"
├── parameter-extraction
│   ├── input: {"user_input": "...", "intent": "query_order"}
│   └── output: {"order_id": "ORD001"}
├── tool-query_order
│   ├── input: {"intent": "query_order", "params": {"order_id": "ORD001"}}
│   ├── output: {"status": "已发货", ...}
│   └── duration: 312ms
├── response-generation
│   └── output: "您的订单ORD001已发货..."
└── Scores:
    ├── tool_success: 1.0
    └── latency_category: 1.0

每个节点可展开查看完整输入输出、耗时、token用量。

七、追踪的五维设计

一个完整的Agent追踪系统需要覆盖五个维度：

7.1 输入输出追踪

最基本的要求：记录每一步的进出。

@observe()
def agent_step(input_data: dict) -> dict:
    # Langfuse自动捕获函数参数作为input
    # 返回值作为output
    result = {"status": "ok", "data": "..."}
    return result

7.2 中间步骤追踪

Agent的关键在于"中间过程"。用Span记录每一个子步骤：

# 在低阶API中创建子span
span = trace.span(
    name="tool-execution",
    input={"tool": "search", "query": "北京天气"},
    metadata={"retry_count": 0, "timeout": 5}
)

# ... 执行工具调用 ...

span.end(
    output={"temperature": 28, "humidity": 60},
    metadata={"http_status": 200, "cache_hit": True}
)

7.3 工具调用追踪

这是最重要的追踪点。工具调用是Agent与外界交互的边界，出错概率最高：

@observe(name="tool-call")
def tracked_tool_call(tool_name: str, **kwargs):
    """所有工具调用的统一入口"""
    start = time.time()
    
    try:
        result = actual_tool_function(**kwargs)
        success = True
    except Exception as e:
        result = {"error": str(e)}
        success = False
    
    duration = time.time() - start
    
    # 记录工具调用详情
    langfuse_context.update_current_observation(
        name=f"tool-{tool_name}",
        input=kwargs,
        output=result,
        metadata={
            "duration_ms": round(duration * 1000, 2),
            "success": success,
            "tool_name": tool_name
        }
    )
    
    # 工具调用评分
    langfuse_context.score(
        name=f"tool_{tool_name}_latency",
        value=round(duration, 2)
    )
    
    return result

7.4 耗时分析

每个Agent开发者最困惑的问题：“Agent到底慢在哪里？”

# Langfuse自动记录每个observation的耗时
# 在UI上可以按耗时排序，快速发现瓶颈

7.5 评分与评估

追踪不仅仅是记录，还要打分：

# 在整个trace级别打分
langfuse_context.score(
    name="user_satisfaction",
    value=0.85,
    comment="回答完整，但稍慢"
)

# 在具体observation级别打分
langfuse_context.score(
    name="tool_accuracy",
    value=0.9,
    data_type="NUMERIC",
    observation_id=observation_id  # 关联到具体的observation
)

八、可视化与告警

8.1 通过Langfuse Dashboard监控

Langfuse提供了几个开箱即用的视图：

Traces视图：

  • 按时间线展示所有请求
  • 支持按session、user、tag过滤
  • 可展开查看每一步详情

Analytics视图：

  • 平均延迟趋势
  • Token消耗统计
  • 工具调用频率/成功率
  • 评分分布

Scores视图：

  • 自定义评分指标的聚合
  • 低分请求快速定位

8.2 集成告警

Langfuse支持通过Webhook集成到PagerDuty、Slack等：

# 在Langfuse控制台 Settings → Webhooks 中配置
# 当满足条件时触发告警：
# 条件示例：
# - trace级别的score低于阈值
# - 单次请求token消耗超过限额
# - 工具调用失败率突升

也可以通过API轮询实现自定义告警：

import requests
import time

LANGFUSE_HOST = "http://localhost:3000"
BASIC_AUTH = ("pk-lf-key", "sk-lf-key")

def check_agent_health():
    """监控Agent健康状况"""
    
    # 获取最近5分钟的高延迟trace
    response = requests.get(
        f"{LANGFUSE_HOST}/api/public/traces",
        params={
            "limit": 50,
            "fromTimestamp": int((time.time() - 300) * 1000)
        },
        auth=BASIC_AUTH
    )
    
    traces = response.json().get("data", [])
    
    high_latency = [t for t in traces if t.get("latency", 0) > 10]
    failed_tools = [t for t in traces if any(
        s.get("name", "").startswith("tool_") and s.get("value", 1) < 0.5
        for s in t.get("scores", [])
    )]
    
    alerts = []
    if high_latency:
        alerts.append(f"⚠️ 发现{len(high_latency)}个高延迟Trace（>10s）")
    if failed_tools:
        alerts.append(f"🔴 发现{len(failed_tools)}个工具调用失败")
    
    if alerts:
        print("\n".join(alerts))
        # 可以发送到Slack/飞书等
    else:
        print("✅ Agent健康状况正常")
    
    return len(alerts) == 0

# 定时调用
check_agent_health()

8.3 Opik的可视化

Opik提供类似的能力：

import opik

# 在UI中可查看：
# 1. Trace Tree：完整的调用链视图
# 2. Experiment Dashboard：不同版本的对比
# 3. Dataset Manager：评估数据集管理
# 4. Feedback Scores：用户反馈聚合

九、最佳实践与踩坑经验

9.1 一定要做的几件事

1. Trace ID 贯穿全链路

# 不要把trace_id隔离在Agent内部
# 应该从HTTP请求头或消息队列中透传
import uuid

# 在API入口处生成或接收trace_id
trace_id = request.headers.get("X-Trace-Id", str(uuid.uuid4()))

# 传递给Agent
agent_response = agent.process(user_input, trace_id=trace_id)

2. 工具调用的input/output要结构化

不要只记录"调用了搜索API"，要记录：

# ✅ 好的做法
tool_log = {
    "tool_name": "weather_api",
    "parameters": {"city": "北京", "date": "2025-03-01"},
    "response_status": 200,
    "response": {"temp": 28, "humidity": 60},
    "latency_ms": 312,
    "retry_count": 0
}

# ❌ 不好的做法
tool_log = "已调用天气API"

3. 在关键节点打Score

# 在每个Agent步骤输出一个评估分数
langfuse_context.score(name=f"step_{step_name}_valid", value=1.0)
# 后续可以通过这些分数快速定位异常步骤

4. 控制数据量

Agent的追踪数据量是非常大的（一个10步的Agent可能产生30+个observation）。做好采样：

import random

# 生产环境采样率控制
SAMPLE_RATE = 0.1  # 10%的请求进行全量追踪

def should_trace(user_input: str) -> bool:
    """判断是否需要对本次请求进行全量追踪"""
    # 规则1：包含敏感操作
    if any(keyword in user_input for keyword in ["退款", "投诉", "取消"]):
        return True
    # 规则2：随机采样
    return random.random() < SAMPLE_RATE

9.2 踩坑实录

踩坑1：忘记flush导致数据丢失

Langfuse默认是批量异步上报的，如果程序立即退出，数据可能丢失：

# 在程序退出前flush
langfuse.flush()

踩坑2：嵌套装饰器导致trace混乱

@observe()
def outer():
    inner()

@observe()
def inner():
    pass

# 这样嵌套使用时，inner会被正确归到outer下面
# 但如果你在inner中手动创建了一个新trace，就会断链

踩坑3：OpenAI客户端被包装多次

# ❌ 不要这样做
from langfuse.openai import openai as lf_openai
from openai import OpenAI

lf_openai.chat.completions.create(...)  # 这个会被追踪
OpenAI().chat.completions.create(...)   # 这个不会！

# ✅ 全部使用Langfuse包装的客户端
from langfuse.openai import openai as lf_openai
client = lf_openai

踩坑4：Prompt里包含大量日志信息

# ❌ 不要在prompt中传入整个trace
prompt = f"用户说：{user_input}。以下是之前的日志：{json.dumps(all_logs)}"

# ✅ 只传入摘要或结构化信息
prompt = f"用户说：{user_input}。历史步骤：{summarize_steps(agent_memory)}"

踩坑5：Opik本地服务默认不持久化数据

# 自托管Opik时，务必配置持久化存储
# 在 docker-compose.yml 中确保 volumes 配置正确
volumes:
  - opik_data:/app/data

9.3 生产环境配置参考

采样策略：
├── 正常请求：10% 全量追踪
├── 敏感请求（退款/投诉）：100% 全量追踪
└── 高价值用户：100% 全量追踪

数据保留：
├── 全量数据：7天
├── 聚合指标：30天
└── 评估数据集：永久

告警规则：
├── 工具调用失败率 > 5% → P0告警
├── 平均延迟 > 15s → P1告警
├── Token浪费率 > 30% → P2告警
└── 用户评分 < 0.6 → 自动拉入分析队列

十、总结

写到最后，回到开头那个Fedora的故事。

那个rm -rf /的灾难之所以发生，核心原因是中间步骤的隐式状态变更没有被追踪——第2步的环境变量在第5步被副作用清空了，没有任何人能发现。

AI Agent也是一样。每个LLM调用都是一个"隐式状态变更"，每一步的推理结果都不可预测。如果没有可观测性，你根本无法回答这三个问题：

  1. Agent在做什么？（实时追踪当前执行到哪一步）
  2. Agent为什么这么做？（追溯每一步的推理依据）
  3. Agent做得怎么样？（量化评估每次执行的质量）

Langfuse和Opik就是解决这三个问题的专业工具。它们把分布式追踪的思想移植到了Agent领域，让每一个LLM调用、每一次工具执行、每一步推理过程都可以被记录、回溯和评估。

最后送你一条铁律：

不加追踪的Agent，上线就是在盲飞。

无论你选择Langfuse还是Opik，无论你用装饰器还是低阶API——从写第一行Agent代码开始，就把可观测性焊死进去。等你真的碰到Agent"发疯"的时候，你会感谢自己的这个决定。

参考资源

  • Langfuse官方文档：https://langfuse.com/docs
  • Langfuse GitHub：https://github.com/langfuse/langfuse
  • Langfuse Python SDK：https://pypi.org/project/langfuse/
  • Opik官方文档：https://comet.com/docs/opik
  • Opik GitHub：https://github.com/comet-ml/opik
  • Opik Python SDK：https://pypi.org/project/opik/
  • OpenAI API参考：https://platform.openai.com/docs/api-reference

本文配套代码已上传：https://github.com/your-repo/agent-observability-demo

# 人工智能
Logo
AI Agent技术社区

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

更多推荐

cover

2026 年第 30 周 Agent＋Java 岗位观察：Harness 开始进入招聘要求

avatar AI Agent技术社区
cover

别急着让AI Agent拥有权限：没有监控的智能体，比Bug代码更危险

avatar AI Agent技术社区
cover

小白程序员入门大模型AI Agent的6+1核心概念，轻松构建实用AI应用！

avatar AI Agent技术社区