摘要

本文探讨AI Agent在多步推理任务中面临的“黑盒”挑战，系统介绍可观测性（Observability）的核心概念、技术栈与实践方法，旨在为开发者提供一套破解Agent内部决策过程的技术框架。

1. 引言：AI Agent的“黑盒”困境

  • 现象描述：复杂Agent任务（如代码生成、数据分析、规划决策）的输出结果难以追溯和解释。
  • 核心痛点：
    • 用户只看到最终答案，无法理解推理路径。
    • 调试困难：当输出错误时，难以定位是哪个步骤出了问题。
    • 信任缺失：在关键业务场景中，不可解释的决策难以被采纳。
  • 可观测性的价值：从“黑盒”到“白盒”或“灰盒”，提升透明度、可调试性与可信度。

2. 可观测性（Observability）的核心三要素

借鉴软件工程的可观测性理念，将其适配到AI Agent领域。

  • 日志（Logs）：记录Agent执行过程中的离散事件。
    • 内容：工具调用、提示词（Prompt）输入输出、中间结果、异常信息。
    • 挑战：信息过载、结构化与语义化。
  • 指标（Metrics）：量化Agent的性能与状态。
    • 性能指标：单步耗时、Token消耗、成本。
    • 质量指标：工具调用成功率、推理路径的置信度、与目标的偏离度。
  • 追踪（Traces）：还原单次请求的完整端到端执行链路。
    • 核心：将一次用户查询触发的多步工具调用、LLM推理串联成一个有向无环图（DAG）。
    • 价值：可视化推理路径，分析瓶颈，复现问题。

3. 构建Agent可观测性系统的技术栈

  • 3.1 日志与事件采集
    • 框架原生支持（如LangChain Callbacks, LlamaIndex Callback Managers）。
    • 自定义装饰器或中间件拦截Agent每一步操作。
    • 标准化日志格式（如OpenTelemetry Logs）。
  • 3.2 分布式链路追踪
    • 概念引入：Span（单个LLM调用或工具调用）、Trace（完整会话）。
    • 集成方案：使用OpenTelemetry进行埋点，将Trace数据导出到Jaeger、Zipkin或云服务。
    • 可视化：展示Agent的“思维链”图谱。
  • 3.3 指标监控与告警
    • 定义关键业务与技术指标（SLO）。
    • 使用Prometheus采集，Grafana展示仪表盘。
    • 设置告警规则（如耗时激增、失败率上升）。
  • 3.4 会话重放与调试
    • 持久化存储完整的会话上下文（Prompt、工具参数、返回结果）。
    • 构建“时间旅行”调试器，可回放任意历史请求的执行过程。
    • 与版本控制系统集成，关联代码/提示词变更。

3.5 三要素对比总览

下表从 Agent 可观测性视角，横向对比日志、指标、追踪在核心内容、采集方式、存储工具和典型用途上的关键差异，帮助读者快速选择适合当前阶段的观测手段。

维度 日志（Logs） 指标（Metrics） 追踪（Traces）
核心内容 离散事件记录：工具调用、Prompt/Response、异常堆栈、中间步骤文本 聚合数值：单步耗时、Token 消耗、成功率、偏离度等时序或聚合数据 端到端调用链：一次请求触发的 LLM 调用与工具调用有向无环图（DAG）
采集方式 Callback 钩子、装饰器、中间件拦截；框架原生日志模块 自定义埋点（Counter/Gauge/Histogram）；框架/库自带 Metrics 暴露 OpenTelemetry SDK 自动/手动埋点；Span 嵌套构造 Trace
常用存储 Elasticsearch + Kibana、Loki、CloudWatch Logs、本地文件 Prometheus + Grafana、Datadog、VictoriaMetrics Jaeger、Zipkin、Tempo、Grafana、Arize、LangSmith 等专用平台
在 Agent 场景的典型用途 调试单步异常、审计提示词与工具输入输出、会话回放 监控 Agent 整体健康度、成本控制、SLO 告警、A/B 测试对比 可视化推理路径、定位瓶颈步骤、归因错误源头、优化工具链顺序

4. 实践：为LangChain Agent添加可观测性

  • 4.1 利用LangChain Callbacks进行基础日志

    下面的代码实现了一个自定义 CallbackHandler，继承自 BaseCallbackHandler，在 LLM 调用、工具调用、链式调用的开始和结束时自动记录关键信息，帮助开发者理解 Agent 的内部执行流程。

from typing import Any, Dict, List, Optional
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.messages import BaseMessage
from langchain_core.outputs import LLMResult
import logging
import json

# 配置日志记录器
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class ObservableCallbackHandler(BaseCallbackHandler):
    """
    自定义 CallbackHandler，用于记录 Agent 执行过程中每一个关键步骤的
    输入输出、耗时和异常信息，是实现 Agent 可观测性的基石。
    """

    # ---------- LLM 调用回调 ----------
    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        prompts: List[str],
        **kwargs: Any,
    ) -> None:
        """LLM 开始推理时触发，记录模型信息和输入的提示词。"""
        logger.info(
            "LLM 开始调用 | 模型: %s | 提示词长度: %s 字符",
            serialized.get("name", "unknown_llm"),
            len(prompts[0]) if prompts else 0,
        )

    def on_llm_end(self, response: LLMResult, **kwargs: Any) -> None:
        """LLM 推理结束时触发，记录返回的 Token 数量和响应摘要。"""
        token_usage = response.llm_output.get("token_usage", {}) if response.llm_output else {}
        # 从生成的 message 列表中提取第一个作为示例
        text_snippet = ""
        if response.generations and response.generations[0]:
            text_snippet = str(response.generations[0][0].message.content)[:150]
        logger.info(
            "LLM 调用结束 | Token 用量: %s | 响应摘要: %s...",
            token_usage,
            text_snippet,
        )

    def on_llm_error(self, error: Exception, **kwargs: Any) -> None:
        """LLM 调用发生异常时触发。"""
        logger.error("LLM 调用异常: %s", error)

    # ---------- 工具调用回调 ----------
    def on_tool_start(
        self,
        serialized: Dict[str, Any],
        input_str: str,
        **kwargs: Any,
    ) -> None:
        """工具开始执行时触发，记录工具名称和传入参数。"""
        logger.info(
            "工具开始执行 | 工具: %s | 输入: %s",
            serialized.get("name", "unknown_tool"),
            input_str[:200],
        )

    def on_tool_end(self, output: str, **kwargs: Any) -> None:
        """工具执行结束时触发，记录工具返回结果。"""
        logger.info(
            "工具执行结束 | 输出: %s",
            output[:200],
        )

    def on_tool_error(self, error: Exception, **kwargs: Any) -> None:
        """工具执行异常时触发。"""
        logger.error("工具执行异常: %s", error)

    # ---------- Chain / Agent 级回调 ----------
    def on_chain_start(
        self,
        serialized: Dict[str, Any],
        inputs: Dict[str, Any],
        **kwargs: Any,
    ) -> None:
        """Chain 或 Agent 开始运行时触发。"""
        logger.info(
            "Chain 开始 | 链: %s | 输入键: %s",
            serialized.get("name", "unknown_chain"),
            list(inputs.keys()),
        )

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
        """Chain 或 Agent 运行结束时触发，记录最终输出。"""
        logger.info(
            "Chain 结束 | 输出键: %s",
            list(outputs.keys()),
        )

    def on_chain_error(self, error: Exception, **kwargs: Any) -> None:
        """Chain / Agent 运行异常时触发。"""
        logger.error("Chain 执行异常: %s", error)

    # ---------- 文本回调 ----------
    def on_text(self, text: str, **kwargs: Any) -> None:
        """Agent 输出中间文本时触发（如 AgentAction 的日志）。"""
        logger.debug("Agent 中间输出: %s", text)


# ========== 使用示例 ==========
# 将自定义 Handler 传入 AgentExecutor
# from langchain.agents import create_openai_tools_agent
# from langchain.agents import AgentExecutor
#
# handler = ObservableCallbackHandler()
# agent_executor = AgentExecutor(
#     agent=agent,
#     tools=tools,
#     callbacks=[handler],  # 核心：在这里注入可观测性回调
# )
# result = agent_executor.invoke({"input": "帮我查一下今天的天气"})

  • 4.2 集成OpenTelemetry实现链路追踪

    下面的示例展示了如何使用 OpenTelemetry 为 Agent 的执行流程创建 Trace 和 Span，将 LLM 调用和工具调用映射为完整的调用链，方便在 Jaeger 等工具中可视化。

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    BatchSpanProcessor,
    ConsoleSpanExporter,
)
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict, List, Optional

# ========== 1. 初始化 OpenTelemetry Tracer ==========
# 创建资源标识，在追踪后端中可通过服务名筛选数据
resource = Resource(attributes={SERVICE_NAME: "langchain-agent-service"})

# 配置 TracerProvider 并设置全局
provider = TracerProvider(resource=resource)

# 将 Span 导出到控制台（生产环境可替换为 OTLP Exporter 对接 Jaeger/Tempo 等）
console_exporter = ConsoleSpanExporter()
span_processor = BatchSpanProcessor(console_exporter)
provider.add_span_processor(span_processor)
trace.set_tracer_provider(provider)

# 获取当前服务的 Tracer 实例
tracer = trace.get_tracer(__name__)


class OpenTelemetryCallbackHandler(BaseCallbackHandler):
    """
    集成了 OpenTelemetry 的 CallbackHandler：
    - 为每个 LLM 调用和工具调用创建子 Span
    - 在 Span 中记录关键属性（模型名、Token 数、工具参数等）
    - 出现异常时在 Span 上设置异常状态
    """

    def on_llm_start(
        self,
        serialized: Dict[str, Any],
        prompts: List[str],
        **kwargs: Any,
    ) -> None:
        """创建 LLM 调用的子 Span，记录模型和输入长度。"""
        # 用当前上下文创建子 Span
        span = tracer.start_span("llm_call")
        span.set_attribute("model", serialized.get("name", "unknown_llm"))
        span.set_attribute("prompt_length", len(prompts[0]) if prompts else 0)
        # 将 Span 存储在 kwargs 中，以便在 on_llm_end 中结束
        kwargs["__otel_span__"] = span

    def on_llm_end(self, response, **kwargs: Any) -> None:
        """结束 LLM Span，记录 Token 使用情况。"""
        span = kwargs.get("__otel_span__")
        if span:
            token_usage = (
                response.llm_output.get("token_usage", {})
                if response.llm_output
                else {}
            )
            span.set_attribute("token_usage", str(token_usage))
            span.end()

    def on_llm_error(self, error: Exception, **kwargs: Any) -> None:
        """LLM 报错时在 Span 上记录异常并结束。"""
        span = kwargs.get("__otel_span__")
        if span:
            span.record_exception(error)
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(error)))
            span.end()

    def on_tool_start(
        self,
        serialized: Dict[str, Any],
        input_str: str,
        **kwargs: Any,
    ) -> None:
        """创建工具调用的子 Span，记录工具名和输入参数。"""
        span = tracer.start_span("tool_call")
        span.set_attribute("tool_name", serialized.get("name", "unknown_tool"))
        span.set_attribute("tool_input", input_str[:200])
        kwargs["__otel_span__"] = span

    def on_tool_end(self, output: str, **kwargs: Any) -> None:
        """结束工具 Span，记录输出片段。"""
        span = kwargs.get("__otel_span__")
        if span:
            span.set_attribute("tool_output", output[:200])
            span.end()

    def on_tool_error(self, error: Exception, **kwargs: Any) -> None:
        """工具报错时在 Span 上记录异常并结束。"""
        span = kwargs.get("__otel_span__")
        if span:
            span.record_exception(error)
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(error)))
            span.end()

    def on_chain_start(
        self,
        serialized: Dict[str, Any],
        inputs: Dict[str, Any],
        **kwargs: Any,
    ) -> None:
        """为整个 Agent / Chain 创建根 Span。"""
        span = tracer.start_span("agent_execution")
        span.set_attribute(
            "chain_name", serialized.get("name", "unknown_chain")
        )
        kwargs["__otel_root_span__"] = span

    def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) -> None:
        """结束根 Span。"""
        span = kwargs.get("__otel_root_span__")
        if span:
            span.end()

    def on_chain_error(self, error: Exception, **kwargs: Any) -> None:
        """Agent 报错时在根 Span 上记录异常并结束。"""
        span = kwargs.get("__otel_root_span__")
        if span:
            span.record_exception(error)
            span.set_status(trace.Status(trace.StatusCode.ERROR, str(error)))
            span.end()


# ========== 2. 使用示例 ==========
# from langchain.agents import create_openai_tools_agent
# from langchain.agents import AgentExecutor
#
# otel_handler = OpenTelemetryCallbackHandler()
# agent_executor = AgentExecutor(
#     agent=agent,
#     tools=tools,
#     callbacks=[otel_handler],
# )
# result = agent_executor.invoke({"input": "帮我查一下今天的天气"})
# # 此时控制台会输出完整的 Trace 信息，可在 Jaeger 等后端查看调用链路
  • 4.3 构建一个简单的可观测性控制台
    • 前端展示：请求列表、Trace详情视图、关键指标图表。
    • 后端服务：聚合与查询追踪数据。

5. 高级主题：超越基础可观测性

  • 5.1 意图与计划识别
    • 在Trace中标注Agent的“意图”（Goal）和生成的“计划”（Plan）。
  • 5.2 因果分析与归因
    • 当最终答案错误时，自动分析是哪个推理步骤或工具调用导致了偏差。
  • 5.3 基于可观测性的Agent优化
    • 利用Trace数据发现低效或冗余的步骤，优化提示词或工具链。
    • A/B测试不同Agent架构或模型的效果。

6. 挑战与未来展望

  • 挑战：
    • 性能开销：可观测性引入的延迟。
    • 信息敏感度：日志可能包含敏感数据。
    • 标准化缺失：不同框架、不同厂商的Agent数据格式不统一。
  • 展望：
    • 可观测性即代码（Observability as Code）。
    • AI驱动的根因分析（AI for AI Ops）。
    • 与评估（Evaluation）框架的深度集成。

7. 总结

  • 可观测性是释放AI Agent在生产环境潜力的关键。
  • 从日志、指标、追踪三个支柱入手，由简入繁。
  • 目标不仅是“看见”，更是“理解”和“优化”Agent的行为。

8. 术语表

术语 英文全称 简要解释
可观测性 Observability 通过外部输出（日志、指标、追踪）推断系统内部状态的能力，在 Agent 领域用于还原黑盒决策过程
日志 Logs 系统运行时产生的离散事件记录，例如工具调用的输入输出、异常堆栈
指标 Metrics 反映系统性能与质量的可聚合数值，如 Token 消耗、工具调用成功率
追踪 Traces 记录一次请求端到端执行路径的调用链，由多个 Span 组成
Span Span 追踪中的最小执行单元，代表一次 LLM 调用或工具调用
有向无环图 DAG (Directed Acyclic Graph) Agent 多步推理路径的图形化表示，节点为操作，边为依赖关系
SLO Service Level Objective 服务等级目标，定义系统在特定时间窗口内应达到的指标阈值
OpenTelemetry OpenTelemetry 开源的可观测性框架，提供统一的 API、SDK 和采集器，用于生成和导出 Trace、Metrics、Logs
Callback Callback 框架提供的钩子机制，允许在 Agent/LLM/工具执行的不同阶段插入自定义逻辑
OTLP OpenTelemetry Protocol OpenTelemetry 定义的遥测数据传输协议，用于将数据发送到可观测性后端
Jaeger Jaeger Uber 开源的分布式链路追踪系统，兼容 OpenTelemetry 的 Trace 数据
Prometheus Prometheus 开源的系统监控与告警套件，用于采集和查询时序指标
Grafana Grafana 开源的可视化分析平台，常用于展示 Prometheus 指标仪表盘和 Trace 视图
LLM Large Language Model 大语言模型，如 GPT-4、Claude 等，是 AI Agent 的推理核心

9. 参考资料

官方文档

开源工具仓库

推荐阅读

# 人工智能
Logo
智能体开发者社区

中国智能体开发者社区，聚焦智能体与大模型开发，提供前沿资讯、实用工具链、开源项目及行业案例。通过技术沙龙、开发者大赛等活动，促进经验交流与协作，助力开发者快速构建创新智能应用。

更多推荐

cover

Day 006 — Multi-Agent + MCP/A2A + 安全 + 可观测性

avatar 智能体开发者社区
cover

规避 90% 搭建故障，OpenClaw 2.7.9 Windows、macOS 双系统安装操作流程

avatar 智能体开发者社区
cover

【安心陪诊 Agent】作品介绍文档怎么写：800 字限制下如何讲清创新点

avatar 智能体开发者社区