Agent 可观测性：用 LangSmith 追踪 Agent 思考过程

LangSmith 是 LangChain 官方提供的 Agent 可观测性平台，可以可视化地记录和展示 Agent 每一步的思考过程、工具调用情况和 LLM 输入输出，帮助开发者调试和优化 Agent。

生产环境中的 Agent 出问题时，最常见的反应是：打开日志，看到一堆 HTTP 请求和 JSON 响应，然后不知道该从哪里开始排查。问题出在第几步？LLM 收到的 prompt 到底是什么？工具调用失败还是 LLM 选错了工具？日志里的信息有，但缺乏结构，无法直接回答这些问题。

这就是 Agent 可观测性要解决的问题：不是"记录发生了什么"，而是"让问题可以被追溯、被理解、被度量"。

1.1 生产 Agent 的调试困境

Agent可观测性层级图
Agent 可观测性三层架构——Trace（链路）、Run（运行）、Step（步骤）层层嵌套

传统服务的调试依赖三件套：日志、指标、报警。这套体系对确定性代码有效，因为相同输入产生相同输出，错误可以复现。

Agent 完全不同：

黑盒推理：LLM 的决策过程不透明。为什么这次调用了 send_email 而不是 save_draft？从日志里看不出来，只能猜。

多步骤错误传播：Agent 出错往往在第 5 步，但根因在第 2 步的工具返回了格式错误的数据。结构化日志如果没有"步骤间关联"，就无法追溯。

非确定性：同样的用户输入，两次执行可能走不同的路径、调用不同的工具、产生不同的输出。Bug 难以稳定复现。

延迟与成本难以归因：一次 Agent 执行耗时 15 秒，消耗 8000 token，但哪一步是瓶颈？token 花在了哪里？print 日志无法回答这些问题。

1.2 可观测性的三个层次

借鉴分布式系统的 OpenTelemetry（开放遥测标准，一套用于收集和统一分布式系统可观测数据的开源规范）标准，Agent 的可观测性分为三层：

层次	含义	包含信息	对应问题
Trace（链路）	一次完整的 Agent 任务执行	总耗时、总 token、最终结果	这次执行成功了吗？花了多少？
Span（步骤）	一个原子操作（LLM 调用、工具调用）	步骤耗时、输入/输出、状态	哪一步最慢？哪一步出错了？
Event（事件）	Span 内的关键时间点	参数值、中间状态变化	具体传了什么参数？返回了什么？

1.3 LangSmith 接入：3 行代码启用完整追踪

LangSmith 是 LangChain（一个构建 LLM 应用的开源 Python 框架）团队开发的 Agent 追踪平台，原生支持 LangChain/LangGraph，也支持自定义接入。

1.3.1 环境配置

bash

pip install langsmith openai langchain-openai

python

import os

# 三行配置，启用 LangSmith 追踪
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-agent-production"

# 之后所有 LangChain/LangGraph 调用自动追踪，无需修改业务代码

如果使用 LangChain 的 ChatOpenAI，启用追踪后，每次调用都会自动生成完整的 Trace，包含：输入 prompt、输出内容、token 消耗、延迟数据。

python

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate

# 启用追踪后，这段代码的每次执行都会在 LangSmith 上留下完整记录
llm = ChatOpenAI(model="gpt-4o", temperature=0)

@tool
def search_web(query: str) -> str:
    """Search the web for information about a topic."""
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Error: {e}"

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, [search_web, calculate], prompt)
executor = AgentExecutor(agent=agent, tools=[search_web, calculate], verbose=True)

# 每次调用自动记录到 LangSmith
result = executor.invoke({"input": "What is 15 * 23?"})

1.3.2 关联业务信息

生产环境中需要将 Agent 追踪与业务上下文关联，便于按用户、会话、请求 ID 查询。

python

from langsmith import traceable
from langchain_core.runnables.config import RunnableConfig

def run_agent_with_metadata(
    user_query: str,
    user_id: str,
    session_id: str,
) -> str:
    """为每次执行附加业务元数据"""
    config = RunnableConfig(
        metadata={
            "user_id": user_id,
            "session_id": session_id,
            "query_length": len(user_query),
        },
        tags=["production", "v2.1"],
        run_name=f"agent-{session_id[:8]}",  # 在 LangSmith 上的展示名称
    )
    result = executor.invoke({"input": user_query}, config=config)
    return result["output"]

1.4 自定义 Span：为关键业务逻辑打标记

LangChain 管辖之外的代码（自定义检索逻辑、数据库查询、第三方 API）需要手动打 Span。

python

from langsmith import traceable
import time

@traceable(name="tool_retrieval", run_type="retriever")
def retrieve_relevant_tools(query: str, top_k: int = 5) -> list[dict]:
    """
    用 @traceable 装饰器，自动创建一个 Span。
    run_type 告诉 LangSmith 这是什么类型的操作（retriever/llm/tool/chain）。
    """
    # 模拟工具检索逻辑
    start = time.time()
    results = []  # 实际场景替换为向量检索
    latency = time.time() - start

    # 返回值自动记录为 Span 的输出
    return results


@traceable(name="database_query", run_type="tool")
def query_user_database(
    sql: str,
    database: str = "prod",
) -> dict:
    """复杂的数据库查询，需要独立追踪"""
    # 函数参数自动记录为 Span 的输入
    result = {"rows": [], "count": 0}
    return result


# 嵌套追踪：父 Span 包含子 Span，形成树状结构
@traceable(name="complex_pipeline")
def run_data_pipeline(user_request: str) -> str:
    tools = retrieve_relevant_tools(user_request)  # 子 Span
    db_result = query_user_database("SELECT * FROM users")  # 子 Span
    # 后续 LLM 调用也自动记录
    return "pipeline complete"

1.5 关键指标：Token 消耗、工具调用成功率、平均步骤数

LangSmith 的 Dashboard 支持自定义指标聚合。以下是生产环境最重要的三类指标：

1.5.1 Token 消耗热图

python

from langsmith import Client
from datetime import datetime, timedelta

ls_client = Client()

def get_token_usage_stats(project_name: str, days: int = 7) -> dict:
    """统计最近 N 天的 token 消耗分布"""
    runs = ls_client.list_runs(
        project_name=project_name,
        start_time=datetime.now() - timedelta(days=days),
        run_type="llm",  # 只统计 LLM 调用
    )

    token_stats = {"prompt_tokens": [], "completion_tokens": []}
    for run in runs:
        if run.token_usage:
            token_stats["prompt_tokens"].append(run.token_usage.get("prompt_tokens", 0))
            token_stats["completion_tokens"].append(run.token_usage.get("completion_tokens", 0))

    total_prompt = sum(token_stats["prompt_tokens"])
    total_completion = sum(token_stats["completion_tokens"])
    n = len(token_stats["prompt_tokens"])

    return {
        "total_llm_calls": n,
        "avg_prompt_tokens": total_prompt / n if n else 0,
        "avg_completion_tokens": total_completion / n if n else 0,
        "total_cost_usd": (total_prompt * 0.0025 + total_completion * 0.01) / 1000,
    }

1.5.2 工具调用成功率

python

def get_tool_success_rate(project_name: str) -> dict[str, float]:
    """统计各工具的调用成功率"""
    runs = ls_client.list_runs(
        project_name=project_name,
        run_type="tool",
    )

    tool_stats: dict[str, dict] = {}
    for run in runs:
        name = run.name
        if name not in tool_stats:
            tool_stats[name] = {"total": 0, "errors": 0}
        tool_stats[name]["total"] += 1
        if run.error:
            tool_stats[name]["errors"] += 1

    return {
        name: 1 - (stats["errors"] / stats["total"])
        for name, stats in tool_stats.items()
        if stats["total"] > 0
    }

1.5.3 平均步骤数（Agent 效率指标）

python

def get_agent_step_stats(project_name: str) -> dict:
    """统计 Agent 执行的平均步骤数——步骤数过多说明 Agent 在兜圈子"""
    chain_runs = ls_client.list_runs(
        project_name=project_name,
        run_type="chain",
        is_root=True,  # 只看顶层 Trace
    )

    step_counts = []
    for run in chain_runs:
        # 通过子 Span 数量估算步骤数
        child_runs = list(ls_client.list_runs(parent_run_id=run.id, run_type="llm"))
        step_counts.append(len(child_runs))

    if not step_counts:
        return {}

    return {
        "avg_steps": sum(step_counts) / len(step_counts),
        "max_steps": max(step_counts),
        "p95_steps": sorted(step_counts)[int(len(step_counts) * 0.95)],
    }

1.6 Langfuse：开源自托管替代方案

LangSmith 是 SaaS 服务，数据上传到 LangChain 的服务器。对于数据合规要求严格的场景（金融、医疗、政府），Langfuse 是更好的选择——完全开源，支持 Docker 自托管。

bash

# Docker Compose 一键部署 Langfuse
git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up -d
# 访问 http://localhost:3000

python

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse(
    public_key="your-public-key",
    secret_key="your-secret-key",
    host="http://localhost:3000",  # 自托管地址
)

@observe()  # 自动创建 Span
def process_with_llm(prompt: str) -> str:
    """Langfuse 的 @observe 等价于 LangSmith 的 @traceable"""
    # 在 Span 内手动添加业务指标
    langfuse_context.update_current_observation(
        metadata={"prompt_length": len(prompt)},
        tags=["production"],
    )

    # 调用 LLM（需要使用 Langfuse 的 OpenAI wrapper 才能自动追踪 token）
    from langfuse.openai import openai as langfuse_openai
    response = langfuse_openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
    )
    return response.choices[0].message.content

LangSmith 与 Langfuse 的对比：

维度	LangSmith	Langfuse
部署方式	SaaS（托管）	开源自托管 / SaaS
LangChain 集成	原生，零配置	需要 wrapper
数据主权	数据在 LangChain 服务器	数据完全自控
评估功能	内置 human evaluation	内置 + 自定义评估
免费额度	有限	自托管无限制
生产就绪性	成熟	快速成长中

1.7 OpenTelemetry 标准化：脱离平台依赖

长期来看，锁定在单一可观测性平台是风险。OpenTelemetry（OTel）是 CNCF（云原生计算基金会，管理 Kubernetes 等云原生开源项目的国际组织）的开放标准，通过统一的 SDK 发出 Trace，后端可以是 Jaeger（一个开源分布式追踪系统）、Datadog（一个商业监控和分析平台）、LangSmith 或任意支持 OTel 的系统。

python

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# 配置 OTel TracerProvider
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="http://localhost:4317")  # OTLP 接收端
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)

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

def run_agent_otel(task: str) -> str:
    """使用 OTel 标准 API 追踪 Agent 执行"""
    with tracer.start_as_current_span("agent.run") as span:
        span.set_attribute("agent.task", task)
        span.set_attribute("agent.model", "gpt-4o")

        for step in range(10):
            with tracer.start_as_current_span("agent.step") as step_span:
                step_span.set_attribute("agent.step_number", step)

                # LLM 调用
                with tracer.start_as_current_span("llm.call") as llm_span:
                    llm_span.set_attribute("llm.model", "gpt-4o")
                    # 实际 LLM 调用
                    llm_span.set_attribute("llm.prompt_tokens", 500)
                    llm_span.set_attribute("llm.completion_tokens", 100)

                # 判断是否结束
                done = True  # 简化示例
                if done:
                    span.set_attribute("agent.steps_taken", step + 1)
                    return "task completed"

    return "max steps reached"

1.8 线上问题定位实战：一个典型 Bug 的完整排查过程

问题现象：生产环境中，有约 15% 的 Agent 任务返回"I couldn't complete the task"，但没有明显报错。

第一步：在 LangSmith 上过滤失败 Trace

python

# 找到失败的 Trace（通过 error 字段或输出关键词过滤）
failed_runs = ls_client.list_runs(
    project_name="production",
    filter='and(eq(status, "error"), gt(start_time, "2024-01-15"))',
)

# 或者用输出内容过滤
all_runs = ls_client.list_runs(project_name="production", run_type="chain", is_root=True)
suspicious = [r for r in all_runs if r.outputs and "couldn't complete" in str(r.outputs)]
print(f"Suspicious runs: {len(suspicious)}")

第二步：查看具体失败 Trace 的工具调用链

通过 LangSmith UI 展开某个失败的 Trace，发现规律：失败的任务全部在调用 get_user_profile 工具后的 LLM 推理步骤失败。

第三步：对比失败与成功的工具返回值

python

# 找到失败 Trace 中 get_user_profile 工具的输出
tool_runs = ls_client.list_runs(
    project_name="production",
    run_type="tool",
    filter='eq(name, "get_user_profile")',
)

error_outputs = []
for run in tool_runs:
    # 检查父 Trace 是否失败
    parent = ls_client.read_run(run.parent_run_id)
    if parent.error or "couldn't complete" in str(parent.outputs):
        error_outputs.append(run.outputs)

print("Failed tool outputs sample:")
print(error_outputs[:3])

第四步：发现根因

分析输出发现，失败时 get_user_profile 返回的是 {"error": "user_not_found", "code": 404} ——这是正常的业务逻辑（用户不存在），但工具的描述没有说明这种情况，LLM 遇到这个返回值时不知道该如何处理，最终放弃任务。

修复方案：更新工具描述，明确说明"当用户不存在时返回 error 字段，此时应提示用户检查 ID"。工具描述更新后，失败率从 15% 降至 1.2%。

关键洞察：没有 LangSmith 的追踪，这个问题会被归类为"LLM 随机错误"，无法定位到工具描述这个真正的根因。

1.9 小结

Agent 可观测性的核心价值不是"监控系统运行状态"，而是"让非确定性的 AI 行为变得可理解、可追溯、可改进"。

建立可观测性基础设施的优先级建议：

先接入 LangSmith（3 行代码，立刻获得基础追踪能力）
为关键业务逻辑打自定义 Span（工具检索、外部 API 调用、数据库查询）
建立定期指标回顾机制（每周查看工具成功率和平均步骤数趋势）
数据合规场景切换 Langfuse（自托管，数据不出域）
长期向 OTel 标准迁移（避免平台锁定）

下一篇将讨论 Human-in-the-Loop 设计：当 Agent 面临不可逆操作时，如何优雅地暂停并请求人工确认。