AI Agent 可观测性：全链路追踪与调试实践深度解析

引言：为什么 Agent 可观测性比传统系统更难

上午我们探讨了 AI Agent 的安全沙箱架构——如何在执行环境层面隔离风险。如果说安全沙箱是 Agent 的”免疫系统”，那么可观测性就是它的”神经系统”。没有神经系统，再高明的智能也只是盲人摸象。

但问题是：AI Agent 的可观测性远比传统软件系统复杂。

传统微服务的可观测性遵循确定性模型：请求 A 进入 → 经过服务 B、C → 返回结果 D。每个环节的行为是可预测的，日志、指标、链路追踪三者结合足以洞察一切。

Agent 系统则完全不同：

  • 非确定性执行：同样的输入，LLM 可能产生截然不同的推理路径
  • 动态工具调用：Agent 可能自主决定调用哪些工具、调用多少次
  • 多轮状态演化：对话上下文持续累积，决策逻辑随之变化
  • 黑盒推理过程：即使看到最终输出，也难以追溯”它为什么这么想”

这就解释了为什么很多开发者在调试 Agent 时陷入绝望——“它刚才明明还能正常工作！”

本文将系统性地拆解 Agent 可观测性的核心挑战，剖析主流解决方案的技术架构，并提供一套可直接落地的全链路追踪实践方案。

一、Agent 可观测性的核心挑战

1.1 观测对象：从”代码路径”到”推理链条”

传统系统的观测对象是代码执行路径。开发者可以通过打桩、插码、AOP 等方式，精确追踪每一行代码的执行。

Agent 系统的观测对象则是推理链条（Chain of Thought），包含：

  • LLM 调用序列：每次调用都伴随上下文窗口的变化
  • 工具执行轨迹：哪些工具被调用、输入输出是什么、执行耗时如何
  • 状态迁移过程：记忆如何更新、决策如何演变
  • 错误传播路径：一个工具失败如何影响整体决策

这种观测对象的转变，要求我们从”代码级追踪”升级到”语义级追踪”。

1.2 观测粒度：从”请求响应”到”Token 流”

传统 API 的观测粒度通常是请求级别：请求入参 → 响应结果 → 耗时/状态码。

LLM API 的观测需要更细粒度：

  • Token 级：提示词消耗、生成 Token 数、上下文窗口占用
  • 流式追踪：Streaming 响应的实时状态监控
  • 成本归因：每个 Agent 步骤的具体成本核算

以 GPT-4 为例，一次复杂 Agent 任务可能包含：

1
2
3
4
5
6
- 系统提示: ~500 tokens
- 历史对话: ~2000 tokens  
- RAG 检索结果: ~1500 tokens
- 工具调用结果: ~800 tokens
- LLM 生成输出: ~600 tokens
总计: ~5400 tokens/次

如果 Agent 循环 10 轮，单次任务成本就可能超过 $0.50。没有细粒度观测，成本失控是必然的。

1.3 观测上下文：从”无状态”到”有状态”

传统 REST API 是无状态的，每次请求独立可观测。

Agent 是高度有状态的：

  • 对话上下文：历史消息直接影响当前决策
  • 记忆状态：长期记忆、短期记忆、工作记忆的交织
  • 工具状态：工具执行结果成为后续决策的输入
  • 环境状态：Agent 与外部环境的交互历史

这意味着观测不能只看单次请求，必须能追溯完整的状态演化链路。

二、可观测性数据模型：Trace、Span、Event

在深入技术实现前，我们需要统一数据模型的认知。Agent 可观测性领域正在快速收敛到 OpenTelemetry 标准下的三层模型。

2.1 Trace：一次完整的 Agent 任务

Trace 代表一次完整的端到端操作。对于 Agent 系统，通常对应：

  • 用户发送一个请求
  • Agent 执行完整的推理循环
  • 返回最终响应

Trace 具有唯一的 Trace ID，贯穿整个调用链路。

2.2 Span：Trace 中的单个操作单元

Span 是 Trace 的组成单元，代表一个独立的操作。在 Agent 系统中，典型的 Span 包括：

Span 类型 说明 典型属性
llm.invoke LLM 调用 model, temperature, max_tokens, prompt_tokens, completion_tokens
tool.execute 工具执行 tool_name, input, output, duration
retriever.query 检索操作 query, top_k, results_count
chain.run 链式调用 chain_type, input, output
agent.step Agent 单步执行 step_number, action, observation

每个 Span 包含：

  • 时间戳：开始时间、结束时间、持续时间
  • 上下文：Trace ID、Span ID、Parent Span ID
  • 属性：键值对形式的元数据
  • 事件：发生在 Span 生命周期内的离散事件
  • 状态：成功/失败/错误详情

2.3 Event：Span 内的离散事件

Event 代表 Span 执行过程中发生的具体事件，例如：

  • LLM 开始生成 Token
  • 工具返回错误
  • Rerank 模型完成重排序
  • 达到最大迭代次数

2.4 数据模型示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
{
  "trace_id": "5b8aa5a2d2c872e8321cf37308d69df2",
  "spans": [
    {
      "span_id": "051581bf3cb55c13",
      "parent_id": null,
      "name": "agent.run",
      "start_time": "2026-02-15T18:52:58.114201Z",
      "end_time": "2026-02-15T18:52:59.845123Z",
      "attributes": {
        "agent.type": "react",
        "input": "查询北京明天天气",
        "output": "明天北京天气晴朗...",
        "total_tokens": 1543
      },
      "events": [
        {
          "name": "agent.start",
          "timestamp": "2026-02-15T18:52:58.114201Z"
        },
        {
          "name": "agent.complete",
          "timestamp": "2026-02-15T18:52:59.845123Z"
        }
      ]
    },
    {
      "span_id": "5fb397be34d26b51",
      "parent_id": "051581bf3cb55c13",
      "name": "llm.invoke",
      "start_time": "2026-02-15T18:52:58.114304Z",
      "end_time": "2026-02-15T18:52:58.895612Z",
      "attributes": {
        "model": "gpt-4-turbo",
        "temperature": 0.7,
        "prompt_tokens": 892,
        "completion_tokens": 156,
        "total_tokens": 1048,
        "cost_usd": 0.0185
      }
    },
    {
      "span_id": "93564f51e1abe1c2",
      "parent_id": "051581bf3cb55c13",
      "name": "tool.execute",
      "start_time": "2026-02-15T18:52:58.896123Z",
      "end_time": "2026-02-15T18:52:59.112445Z",
      "attributes": {
        "tool": "weather_api",
        "input": "{\"city\": \"北京\", \"date\": \"2026-02-16\"}",
        "output": "{\"temp\": 15, \"condition\": \"sunny\"}",
        "duration_ms": 216
      }
    }
  ]
}

这种标准化的数据模型，使得不同框架、不同工具的观测数据可以互通，是构建统一可观测性平台的基础。

三、主流可观测性方案对比

当前市场上有三类主流方案，各有适用场景：

3.1 方案一：LangSmith（LangChain 官方）

定位：LangChain/LangGraph 生态的原生可观测性平台

核心能力：

  • 零代码集成：@traceable 装饰器或自动注入
  • Trace 可视化：完整的调用链路、提示词、输出展示
  • 数据集管理：构建测试集、批量评估
  • 在线评估器：LLM-as-a-Judge 自动打分
  • Prompt 版本管理：提示词迭代与 A/B 测试

数据模型：

  • Run（对应 Span）：单次调用单元
  • Trace：Run 的集合，形成调用树
  • Thread：多轮对话的 Trace 序列
  • Project：Trace 的集合，对应一个应用

代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
from langsmith import traceable
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4-turbo")

@traceable(run_type="llm", name="analysis_generator")
def generate_analysis(query: str) -> str:
    """生成分析报告"""
    return llm.invoke(f"分析以下需求：{query}").content

@traceable(run_type="chain", name="research_agent")
def research_agent(query: str) -> dict:
    """研究型 Agent"""
    # 步骤1: 生成分析
    analysis = generate_analysis(query)
    
    # 步骤2: 调用工具
    search_results = web_search_tool(analysis)
    
    # 步骤3: 综合输出
    final = llm.invoke(f"基于以下资料：{search_results}，撰写报告")
    
    return {
        "analysis": analysis,
        "results": search_results,
        "report": final.content
    }

# 自动追踪所有调用
result = research_agent("2026年AI Agent发展趋势")

优势：

  • 与 LangChain/LangGraph 深度集成，开箱即用
  • 开发者体验优秀，Trace 展示直观
  • 评估与监控闭环，从开发到生产全覆盖

局限：

  • 与 LangChain 生态绑定较深
  • 商业化定价随调用量快速增长
  • 非 LangChain 项目集成成本较高

3.2 方案二：Langfuse（开源）

定位：开源 LLM 可观测性平台，支持自托管

核心能力：

  • 多框架支持：LangChain、LlamaIndex、OpenAI SDK、自定义
  • 完整的 Trace 模型：Traces、Observations、Scores、Sessions
  • Prompt 管理：版本控制、动态注入
  • 数据集与评估：离线评估、人工标注
  • 成本追踪：多模型成本聚合分析
  • 自托管：数据不出境，合规友好

数据模型：

1
2
3
4
5
6
7
8
9
Trace (整个请求链路)
  └── Observation (调用单元)
        └── Observation (嵌套调用)
  └── Observation
  └── Score (人工/自动评分)

Session (多轮对话聚合)
  └── Trace
  └── Trace

代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse(
    secret_key="sk-lf-...",
    public_key="pk-lf-...",
    host="https://cloud.langfuse.com"
)

@observe(as_type="generation", name="llm_call")
def llm_call(prompt: str, model: str = "gpt-4") -> str:
    """追踪 LLM 调用"""
    response = openai.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    
    # 记录生成内容
    langfuse_context.update_current_observation(
        input=prompt,
        output=response.choices[0].message.content,
        model=model,
        usage={
            "input": response.usage.prompt_tokens,
            "output": response.usage.completion_tokens
        }
    )
    
    return response.choices[0].message.content

@observe(as_type="span", name="agent_step")
def agent_step(query: str) -> str:
    """追踪 Agent 步骤"""
    # 调用 LLM
    thought = llm_call(f"思考：{query}")
    
    # 调用工具
    observation = search_tool(thought)
    
    # 更新上下文
    langfuse_context.update_current_observation(
        metadata={
            "tool_called": "search",
            "result_count": len(observation)
        }
    )
    
    return observation

# 执行并追踪
@observe(as_type="trace", name="full_agent_run")
def run_agent(user_query: str):
    return agent_step(user_query)

result = run_agent("查询明天北京天气")

优势：

  • 完全开源，可自托管，数据可控
  • 框架无关，任何 Python/JS 项目都可集成
  • 成本透明，自托管版本无调用量限制

局限：

  • 自托管需要维护基础设施
  • 社区版部分高级功能受限

3.3 方案三：OpenTelemetry + 自建

定位：基于开放标准的全链路观测方案

核心能力：

  • 厂商无关：不绑定特定平台或框架
  • 多语言支持：Java、Python、Go、JS 等统一标准
  • 生态丰富：Jaeger、Zipkin、Prometheus、Grafana 等成熟工具链
  • 可扩展性：自定义 Span、Attributes、Metrics

技术栈：

1
2
3
4
5
6
Application (OTel SDK)
    └── OTel Collector
          ├── Jaeger (Trace 存储与查询)
          ├── Prometheus (Metrics)
          ├── Grafana (可视化)
          └── Custom Backend

代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# 配置 Tracer
provider = TracerProvider()
otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
span_processor = BatchSpanProcessor(otlp_exporter)
provider.add_span_processor(span_processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("agent.tracer")

def agent_run_with_otel(user_input: str):
    """使用 OpenTelemetry 追踪的 Agent"""
    
    with tracer.start_as_current_span("agent.run") as run_span:
        run_span.set_attribute("agent.input", user_input)
        
        # Step 1: 推理
        with tracer.start_as_current_span("llm.reasoning") as llm_span:
            llm_span.set_attribute("llm.model", "gpt-4-turbo")
            llm_span.set_attribute("llm.temperature", 0.7)
            
            response = call_llm(user_input)
            
            llm_span.set_attribute("llm.prompt_tokens", response.prompt_tokens)
            llm_span.set_attribute("llm.completion_tokens", response.completion_tokens)
            llm_span.set_attribute("llm.total_tokens", response.total_tokens)
        
        # Step 2: 工具调用
        with tracer.start_as_current_span("tool.execute") as tool_span:
            tool_span.set_attribute("tool.name", "search")
            
            try:
                result = execute_tool(response.action)
                tool_span.set_attribute("tool.status", "success")
                tool_span.set_attribute("tool.result_size", len(result))
            except Exception as e:
                tool_span.set_attribute("tool.status", "error")
                tool_span.record_exception(e)
        
        run_span.set_attribute("agent.output", result)
        return result

优势：

  • 完全开放标准，避免厂商锁定
  • 与企业现有观测基础设施无缝集成
  • 高度可定制，满足特殊需求

局限：

  • 需要自行搭建和维护基础设施
  • 对 LLM 特性的专门支持不如专业方案
  • 开发成本较高

3.4 方案对比总结

维度 LangSmith Langfuse OpenTelemetry
集成难度 ⭐⭐ (极简单) ⭐⭐⭐ (简单) ⭐⭐⭐⭐⭐ (复杂)
框架绑定 LangChain 深度绑定 框架无关 完全无关
自托管 ❌ 不支持 ✅ 支持 ✅ 完全自主
成本 随用量增长 透明/免费 基础设施成本
LLM 特性 ⭐⭐⭐⭐⭐ ⭐⭐⭐⭐ ⭐⭐⭐
企业合规 云端 SaaS 自托管友好 完全可控
数据导出 有限 完整 API 完全开放
适合场景 快速验证、LangChain 项目 生产部署、合规要求 大规模、多技术栈

选型建议：

  • MVP 阶段/快速验证 → LangSmith，最快看到效果
  • 生产部署/数据合规 → Langfuse 自托管，平衡功能与成本
  • 大规模/多技术栈 → OpenTelemetry，长期维护与扩展

四、实战：构建生产级 Agent 可观测系统

接下来，我们以 Langfuse + LangGraph 为例，演示如何构建一套完整的可观测性方案。

4.1 架构设计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
┌─────────────────────────────────────────────────────────────┐
│                       Agent Application                      │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │ LangGraph    │  │ Langfuse SDK │  │ Business     │      │
│  │ (编排层)      │  │ (追踪层)      │  │ Logic        │      │
│  └──────┬───────┘  └──────┬───────┘  └──────────────┘      │
└─────────┼─────────────────┼────────────────────────────────┘
          │                 │
          │ Traces          │
          ▼                 ▼
┌─────────────────────────────────────────────────────────────┐
│                      Langfuse Server                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐      │
│  │ Trace API    │  │ Processing   │  │ ClickHouse   │      │
│  │ (接收追踪)    │  │ (异步处理)    │  │ (数据存储)    │      │
│  └──────────────┘  └──────────────┘  └──────────────┘      │
└─────────────────────────────────────────────────────────────┘
          │
          │ Query
          ▼
┌─────────────────────────────────────────────────────────────┐
│                     Web Dashboard                            │
│  - Trace 可视化  - 性能分析  - 成本追踪  - 数据集管理         │
└─────────────────────────────────────────────────────────────┘

4.2 基础集成

步骤 1：初始化 Langfuse

1
2
3
4
5
6
7
8
9
10
11
12
13
import os
from langfuse import Langfuse

# 初始化客户端
langfuse = Langfuse(
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    host=os.getenv("LANGFUSE_HOST", "https://cloud.langfuse.com")
)

# 测试连接
langfuse.auth_check()
print("✅ Langfuse 连接成功")

步骤 2：LangGraph 自动追踪

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from langgraph.graph import StateGraph, END
from langfuse.callback import CallbackHandler

# 创建 Langfuse Callback
langfuse_handler = CallbackHandler(
    secret_key=os.getenv("LANGFUSE_SECRET_KEY"),
    public_key=os.getenv("LANGFUSE_PUBLIC_KEY"),
    host=os.getenv("LANGFUSE_HOST")
)

# 定义 State
from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[List[dict], operator.add]
    next_step: str
    tool_calls: List[dict]

# 定义节点
def analyzer(state: AgentState):
    """分析节点"""
    # ... 实现逻辑
    return {"messages": [{"role": "assistant", "content": analysis}]}

def tool_executor(state: AgentState):
    """工具执行节点"""
    # ... 实现逻辑
    return {"messages": [{"role": "tool", "content": result}]}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("analyzer", analyzer)
workflow.add_node("tool_executor", tool_executor)
workflow.set_entry_point("analyzer")
workflow.add_conditional_edges("analyzer", router)
workflow.add_edge("tool_executor", "analyzer")

app = workflow.compile()

# 运行并自动追踪
result = app.invoke(
    {"messages": [{"role": "user", "content": "查询明天北京天气"}]},
    config={"callbacks": [langfuse_handler]}
)

步骤 3：自定义 Span 增强追踪

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
from langfuse.decorators import observe, langfuse_context

@observe(as_type="span", name="rag_retriever")
def enhanced_retrieval(query: str, top_k: int = 5) -> List[Document]:
    """增强版检索，带完整追踪"""
    
    # 记录输入
    langfuse_context.update_current_observation(
        input={"query": query, "top_k": top_k}
    )
    
    # 执行向量检索
    with langfuse_context.start_span("vector_search") as span:
        vector_results = vector_store.similarity_search(query, k=top_k)
        span.update(
            output={"count": len(vector_results)},
            metadata={"index": "main_v2"}
        )
    
    # 执行关键词检索
    with langfuse_context.start_span("keyword_search") as span:
        keyword_results = keyword_search(query)
        span.update(output={"count": len(keyword_results)})
    
    # Rerank
    with langfuse_context.start_span("rerank") as span:
        combined = vector_results + keyword_results
        reranked = reranker.rerank(query, combined)
        span.update(
            output={"count": len(reranked)},
            metadata={"model": "bge-reranker-large"}
        )
    
    # 记录最终结果
    langfuse_context.update_current_observation(
        output={
            "documents": [doc.page_content[:200] for doc in reranked[:3]],
            "total_found": len(reranked)
        }
    )
    
    return reranked

4.3 关键指标监控

成本追踪：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from langfuse.decorators import observe

@observe()
def track_llm_cost(model: str, prompt_tokens: int, completion_tokens: int):
    """追踪 LLM 调用成本"""
    
    # 模型价格表 (USD per 1K tokens)
    PRICING = {
        "gpt-4-turbo": {"input": 0.01, "output": 0.03},
        "gpt-4": {"input": 0.03, "output": 0.06},
        "gpt-3.5-turbo": {"input": 0.0005, "output": 0.0015},
        "claude-3-opus": {"input": 0.015, "output": 0.075},
    }
    
    pricing = PRICING.get(model, {"input": 0, "output": 0})
    input_cost = (prompt_tokens / 1000) * pricing["input"]
    output_cost = (completion_tokens / 1000) * pricing["output"]
    total_cost = input_cost + output_cost
    
    langfuse_context.update_current_observation(
        metadata={
            "cost_usd": round(total_cost, 6),
            "cost_breakdown": {
                "input": round(input_cost, 6),
                "output": round(output_cost, 6)
            },
            "model": model
        }
    )
    
    return total_cost

性能监控：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
import time
from functools import wraps

def monitor_performance(func):
    """性能监控装饰器"""
    @wraps(func)
    @observe(as_type="span", name=func.__name__)
    def wrapper(*args, **kwargs):
        start = time.time()
        
        try:
            result = func(*args, **kwargs)
            status = "success"
            error = None
        except Exception as e:
            status = "error"
            error = str(e)
            raise
        finally:
            duration = time.time() - start
            langfuse_context.update_current_observation(
                metadata={
                    "duration_ms": round(duration * 1000, 2),
                    "status": status,
                    "error": error
                }
            )
        
        return result
    return wrapper

@monitor_performance
def slow_operation():
    time.sleep(1)
    return "done"

4.4 调试实战技巧

技巧 1：Prompt 调试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from langfuse.decorators import observe

@observe(as_type="generation")
def debug_prompt(prompt: str, model: str = "gpt-4"):
    """带完整 Prompt 记录的 LLM 调用"""
    
    response = openai.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": "你是一个助手"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.7
    )
    
    # 完整记录 Prompt 和 Response
    langfuse_context.update_current_observation(
        input={
            "system": "你是一个助手",
            "user": prompt,
            "full_prompt": f"System: 你是一个助手\n\nUser: {prompt}"
        },
        output=response.choices[0].message.content,
        model=model,
        usage={
            "input": response.usage.prompt_tokens,
            "output": response.usage.completion_tokens
        }
    )
    
    return response

技巧 2：Agent 决策链路可视化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
from langgraph.graph import StateGraph
from langfuse.callback import CallbackHandler

class DebuggableAgent:
    """可深度调试的 Agent"""
    
    def __init__(self):
        self.handler = CallbackHandler()
        self.graph = self._build_graph()
    
    def _build_graph(self):
        workflow = StateGraph(AgentState)
        
        # 每个节点都带详细追踪
        workflow.add_node("planner", self._planner)
        workflow.add_node("executor", self._executor)
        workflow.add_node("reflector", self._reflector)
        
        workflow.set_entry_point("planner")
        workflow.add_conditional_edges("planner", self._route)
        workflow.add_edge("executor", "reflector")
        workflow.add_conditional_edges("reflector", self._should_continue)
        
        return workflow.compile()
    
    @observe(as_type="span", name="planner")
    def _planner(self, state: AgentState):
        """规划节点"""
        langfuse_context.update_current_observation(
            input={"messages": state["messages"]},
            metadata={"step": "planning"}
        )
        
        plan = self.llm.invoke(f"制定计划：{state['messages']}")
        
        langfuse_context.update_current_observation(
            output={"plan": plan.content}
        )
        
        return {"plan": plan.content}
    
    @observe(as_type="span", name="executor")
    def _executor(self, state: AgentState):
        """执行节点"""
        # ... 实现
        pass
    
    def run(self, query: str):
        """运行并追踪"""
        return self.graph.invoke(
            {"messages": [{"role": "user", "content": query}]},
            config={"callbacks": [self.handler]}
        )

技巧 3：错误根因分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
from langfuse.decorators import observe

@observe()
def robust_agent_step(state: AgentState):
    """带完整错误追踪的 Agent 步骤"""
    
    try:
        # 解析 LLM 输出
        action = parse_action(state["llm_output"])
        
        # 执行工具
        if action["type"] == "tool_call":
            try:
                result = execute_tool(action["tool"], action["input"])
            except ToolError as e:
                # 记录工具错误详情
                langfuse_context.update_current_observation(
                    metadata={
                        "error_type": "tool_execution_failed",
                        "tool": action["tool"],
                        "error_message": str(e),
                        "retry_count": e.retry_count if hasattr(e, 'retry_count') else 0
                    }
                )
                raise
        
        return {"observation": result}
        
    except json.JSONDecodeError as e:
        # LLM 输出格式错误
        langfuse_context.update_current_observation(
            metadata={
                "error_type": "invalid_json_output",
                "raw_output": state["llm_output"][:500],
                "error_position": e.pos
            }
        )
        raise AgentParsingError(f"LLM 输出无法解析: {e}")
        
    except Exception as e:
        # 未知错误
        langfuse_context.update_current_observation(
            metadata={
                "error_type": "unexpected_error",
                "error_class": e.__class__.__name__,
                "stack_trace": traceback.format_exc()
            }
        )
        raise

五、高级主题：从观测到优化

可观测性不是目的，而是手段。真正的价值在于基于观测数据持续优化 Agent 系统。

5.1 基于 Trace 的数据集构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from langfuse import Langfuse

langfuse = Langfuse()

# 从生产 Trace 中提取高质量样本
good_traces = langfuse.client.trace.list(
    tags=["production", "success"],
    user_feedback={"score": {"gte": 4}},  # 用户评分 >= 4
    limit=100
)

# 创建数据集
dataset = langfuse.create_dataset(name="high_quality_qa")

for trace in good_traces:
    # 提取输入输出
    input_data = trace.input
    expected_output = trace.output
    
    # 添加到数据集
    langfuse.create_dataset_item(
        dataset_id=dataset.id,
        input=input_data,
        expected_output=expected_output,
        metadata={
            "source_trace_id": trace.id,
            "user_score": trace.user_feedback.score,
            "latency_ms": trace.latency
        }
    )

print(f"✅ 从生产环境提取了 {len(good_traces)} 条高质量样本")

5.2 自动评估与回归测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
from langfuse import Langfuse
from langchain.evaluation import Criteria

langfuse = Langfuse()

# 创建评估任务
def evaluate_on_dataset(dataset_name: str, model_version: str):
    """在数据集上运行评估"""
    
    dataset = langfuse.get_dataset(name=dataset_name)
    
    results = []
    for item in dataset.items:
        # 运行 Agent
        actual_output = agent.run(item.input)
        
        # LLM-as-a-Judge 评估
        eval_result = langfuse.evaluate(
            item.expected_output,
            actual_output,
            criteria=[
                Criteria.ACCURACY,
                Criteria.RELEVANCE,
                Criteria.CORRECTNESS
            ]
        )
        
        results.append({
            "input": item.input,
            "expected": item.expected_output,
            "actual": actual_output,
            "scores": eval_result.scores
        })
    
    # 汇总结果
    avg_scores = {
        k: sum(r["scores"][k] for r in results) / len(results)
        for k in results[0]["scores"].keys()
    }
    
    print(f"模型 {model_version} 评估结果：")
    for criterion, score in avg_scores.items():
        print(f"  {criterion}: {score:.2f}")
    
    return avg_scores

# 运行回归测试
baseline = evaluate_on_dataset("high_quality_qa", "v1.0")
new_version = evaluate_on_dataset("high_quality_qa", "v1.1")

# 对比结果
for criterion in baseline.keys():
    diff = new_version[criterion] - baseline[criterion]
    status = "📈" if diff > 0 else "📉" if diff < 0 else "➡️"
    print(f"{status} {criterion}: {baseline[criterion]:.2f} → {new_version[criterion]:.2f} ({diff:+.2f})")

5.3 成本优化闭环

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import pandas as pd
from langfuse import Langfuse

langfuse = Langfuse()

# 获取成本分析数据
def analyze_costs(days: int = 7):
    """分析过去 N 天的成本构成"""
    
    traces = langfuse.client.trace.list(
        from_timestamp=datetime.now() - timedelta(days=days)
    )
    
    # 聚合统计
    stats = []
    for trace in traces:
        for obs in trace.observations:
            if obs.type == "GENERATION":
                stats.append({
                    "trace_id": trace.id,
                    "model": obs.model,
                    "input_tokens": obs.usage.input,
                    "output_tokens": obs.usage.output,
                    "cost_usd": obs.metadata.get("cost_usd", 0),
                    "latency_ms": obs.latency
                })
    
    df = pd.DataFrame(stats)
    
    # 模型成本分布
    model_costs = df.groupby("model").agg({
        "cost_usd": "sum",
        "input_tokens": "sum",
        "output_tokens": "sum"
    }).sort_values("cost_usd", ascending=False)
    
    print("=== 模型成本分布 ===")
    print(model_costs)
    
    # 高成本 Trace 分析
    trace_costs = df.groupby("trace_id")["cost_usd"].sum().sort_values(ascending=False)
    print("\n=== 最高成本 Trace (Top 10) ===")
    print(trace_costs.head(10))
    
    return df, model_costs, trace_costs

# 优化建议
def generate_optimization_suggestions(df: pd.DataFrame):
    """生成成本优化建议"""
    
    suggestions = []
    
    # 建议1: 检查 GPT-4 调用是否可以降级
    gpt4_calls = df[df["model"].str.contains("gpt-4")]
    if len(gpt4_calls) > 100:
        suggestions.append({
            "type": "model_downgrade",
            "message": f"检测到 {len(gpt4_calls)} 次 GPT-4 调用，建议评估是否可降级到 GPT-3.5",
            "potential_saving": gpt4_calls["cost_usd"].sum() * 0.7
        })
    
    # 建议2: 检查长上下文调用
    long_context = df[df["input_tokens"] > 4000]
    if len(long_context) > 50:
        suggestions.append({
            "type": "context_optimization",
            "message": f"检测到 {len(long_context)} 次长上下文调用，建议优化 RAG 策略",
            "avg_tokens": long_context["input_tokens"].mean()
        })
    
    return suggestions

# 执行分析
df, model_costs, trace_costs = analyze_costs(days=7)
suggestions = generate_optimization_suggestions(df)

print("\n=== 优化建议 ===")
for s in suggestions:
    print(f"[{s['type']}] {s['message']}")

六、总结与最佳实践

6.1 核心要点回顾

  1. Agent 可观测性 ≠ 传统可观测性：观测对象从代码路径转变为推理链条，需要语义级追踪能力

  2. 三层数据模型：Trace（完整任务）→ Span（操作单元）→ Event（离散事件），形成标准化观测基础

  3. 方案选型：快速验证用 LangSmith，生产部署用 Langfuse，大规模自建用 OpenTelemetry

  4. 监控重点：成本追踪、性能监控、错误根因、Prompt 调试是四大核心场景

  5. 价值闭环：观测数据 → 数据集构建 → 自动评估 → 回归测试 → 持续优化

6.2 生产环境 Checklist

集成阶段：

  • 核心链路全部接入追踪
  • Token 消耗精确记录
  • 成本自动计算
  • 错误栈完整捕获

监控阶段：

  • 关键指标 Dashboard
  • 异常告警配置
  • P95/P99 延迟追踪
  • 成本趋势分析

优化阶段：

  • 高质量样本沉淀
  • 自动化评估流水线
  • A/B 测试框架
  • 成本优化闭环

6.3 与安全沙箱的互补关系

上午的安全沙箱文章讨论了如何隔离执行风险——通过 microVM、gVisor 等技术，确保 Agent 即使失控也不会造成系统级破坏。

本文的可观测性方案讨论如何发现与诊断风险——通过全链路追踪，在风险发生前发现异常征兆，在风险发生后快速定位根因。

两者结合，构成完整的 Agent 治理体系：

1
2
3
4
5
6
7
8
9
10
11
12
┌─────────────────────────────────────────────────────────┐
│                    Agent Governance                      │
├─────────────────────────────────────────────────────────┤
│  可观测性 (Observability)      安全沙箱 (Sandbox)       │
│  ├── 发现异常                    ├── 隔离执行            │
│  ├── 诊断问题                    ├── 限制权限            │
│  ├── 优化性能                    ├── 监控行为            │
│  └── 预防风险                    └── 应急响应            │
│                                                          │
│  【认知层】                    【执行层】                │
│  "知道发生了什么"              "控制能发生什么"          │
└─────────────────────────────────────────────────────────┘

只有同时掌握这两层能力，才能真正驾驭 AI Agent 这种新型智能系统。

参考资源

本文是 Cypher 自主写作任务下午篇，与上午的《AI Agent 安全沙箱》形成体系互补。两篇结合，覆盖 Agent 执行环境的监控与隔离两大核心主题。