AI Agent 可观测性实战：结构化日志、OpenTelemetry 链路追踪与故障排查

AI Agent 的失败方式与传统服务完全不同。一个处理客服工单的 Agent 可能调用退款查询工具一次，几秒后又用略有不同的参数调用第二次，然后自信地基于第二次结果给出一个格式完美、语气专业、但完全错误的答复。整个过程没有任何异常抛出，没有错误码，可用性面板全程绿色。等到两天后客户回复说结果不对时，已经没人能还原那次运行内部究竟发生了什么。

传统服务要么返回 200，要么抛出可以检索的异常；Agent 两者都不做，却依然可能是错的。为前者设计的监控工具，对后者几乎完全失明。本文按日志、追踪、排查三个层次，逐步说明需要改变什么，并给出可直接运行的代码。

准备工作

在动手之前，需要先明确这套方案依赖哪些组件，以及它们各自承担什么职责。

环境与依赖

示例代码使用 Python，依赖两个基础库：标准库 logging 用于输出结构化日志，opentelemetry-api 与 opentelemetry-sdk 用于创建 span 与传播上下文。此外还需要一个模型客户端（示例中以通用的 model_client 表示），以及一个用于接收 trace 数据的后端。具体安装方式与版本要求请以各项目官网当前信息为准。

需要准备的东西包括：

• 一个可用的模型服务账号与 API 密钥，用于发起真实的推理调用。
• Python 运行环境，以及能够安装上述依赖的包管理工具。
• 一个 trace 后端（自建或托管均可），用于接收并可视化 OpenTelemetry 导出的 span 数据。
• 对 Agent 工具调用流程的基本了解：模型返回工具调用请求，应用执行工具，再把结果回填进消息列表。

先想清楚要采集什么

Agent 可观测性指的是：把 Agent 做出的每一次模型调用、每一次工具执行、每一步推理过程，都作为结构化数据采集下来，使得出问题时可以精确还原当时发生了什么、为什么发生，而不是靠猜测或反复重跑同一个提示词、指望问题复现。

它沿用了可观测性工程已有的三大支柱——日志、指标、追踪——但之所以需要单独命名、单独对待，原因在于 Agent 的失败模式本身不同。Agent 系统的失败往往看起来像成功：输出格式正确但内容错误、发起不必要的工具调用、动作在语法上合法但在语义上错误。这些都不会触发任何错误处理器。一个报告“运行中”的健康检查，对于判断某次运行是否真的做对了事情，几乎提供不了任何有用信息。

为什么传统监控模型会失效

把机制说具体一点，比笼统地说“Agent 不可预测”更有用：

• 相同输入不再稳定产生相同行为。 温度参数、检索结果、当前可用的工具集合，都会改变 Agent 走的路径。同一个提示词在连续两次运行中可能触发完全不同的工具调用序列。单次“我测试的时候是好的”的 trace，几乎说明不了真实运行的分布长什么样。
• 成本与延迟不再与请求数相关，而是与 token 相关。 一个“慢”请求可能消耗了十倍于常规的 token 预算，而围绕每秒请求数搭建的监控对此结构性失明。
• 多步链路会放大问题。 一次用户请求可能触发多次模型调用、若干次工具调用和几次检索查询，每一个都是独立的故障点，而单一的聚合错误指标无法区分它们。
• 提示词本身常常携带真实的个人或机密信息。 把完整提示词文本原样倾倒进后端，在产生任何调试价值之前，就已经制造了合规问题。

两种系统的差异可以对照来看：

信号	传统应用	LLM / AI Agent
延迟驱动因素	CPU、I/O、网络	token 数量、模型规模、上下文窗口
成本单位	每秒请求数	消耗的 token
失败模式	异常、超时	幻觉、上下文溢出、工具错误
调试产物	堆栈跟踪	提示词、补全结果，以及两者之间的推理链

操作步骤

第一步：为 Agent 设计结构化日志

日志仍然是一切的基础，只是用法不同。对 Agent 而言，值得记录的事件很具体：调用了哪个工具、参数是什么、返回了什么、某一步消耗了多少 token、每一跳耗时多久，以及沿途出现的任何错误。所有这些都应当是结构化的，而不是写成人之后还要再解析一遍的自由文本句子。

真正让 Agent 日志变得有用的关键细节，是把每一行日志都关联回它所属的那一次具体运行。一行只写着“工具调用失败”的日志，在凌晨两点、同一分钟内有三个不同用户触发了三次不同运行时，几乎毫无价值。把当前 trace ID 附加到每一行日志上——一旦追踪配置完成，OpenTelemetry 会自动做到这一点——才能把散落的日志语句变成可以过滤到具体某次出错运行的东西。

import logging
from opentelemetry import trace

# 标准 Python 日志，没有任何特殊之处
logger = logging.getLogger("agent")
logging.basicConfig(level=logging.INFO)
tracer = trace.get_tracer("agent-service")

def call_tool(tool_name: str, arguments: dict):
    # get_current_span() 取出当前活跃的 span，
    # 这样下面这行日志就能关联回它所属的 trace 和步骤
    span = trace.get_current_span()
    trace_id = format(span.get_span_context().trace_id, "032x")

    logger.info(
        "tool_call_started",
        extra={
            "trace_id": trace_id,
            "tool_name": tool_name,
            "arguments": arguments,
        },
    )

    try:
        result = execute_tool(tool_name, arguments)
        logger.info(
            "tool_call_succeeded",
            extra={"trace_id": trace_id, "tool_name": tool_name, "result_length": len(str(result))},
        )
        return result
    except Exception as e:
        logger.error(
            "tool_call_failed",
            extra={"trace_id": trace_id, "tool_name": tool_name, "error": str(e)},
        )
        raise

这段代码里有几个值得注意的设计取舍：

• trace.get_current_span() 不需要手动把 trace ID 一层层传进每个函数调用，它读取的是当前执行上下文中活跃的 span。这正是这个模式能在真实代码库里随处使用、而不必在每个层级都穿一个 ID 参数的原因。
• 记录参数和结果长度，而不是完整结果内容，是刻意选择而非疏漏。完整的工具输出可能很大，也可能携带敏感数据；一个长度值或截断预览通常足以发现问题，同时不会让每一行日志都变成隐私负担。
• 同一次工具调用同时记录开始事件和结束事件，而不只是记录结果，这样之后才能精确测量这次调用耗时多久——这也正是下一节追踪要正式化的原始材料。

第二步：用 OpenTelemetry 建立链路追踪

日志告诉你各个时间点上发生了什么；追踪则把这些点缝合成一个形状——一次 Agent 运行从最初请求到最终答复的完整记录，每一步都嵌套在触发它的那一步内部。

这种嵌套结构才是“Agent 为什么这么做”的真正答案，因为它展示的不只是某个工具被调用了，而是哪一步推理决定去调用它，以及调用前后紧接着发生了什么。

这里的术语来自 OpenTelemetry 的 GenAI 语义约定，它为这类场景定义了一套标准的 gen_ai.* span 类型与属性。与其让每个团队自己发明 span 名称，规范定义了几种值得了解的操作类型：

• create_agent：Agent 首次被定义时。
• invoke_agent：单次 Agent 运行。
• invoke_workflow：多个 Agent 相互交接时的编排过程。
• execute_tool：单次工具调用。
• chat：实际的模型推理调用本身。

每一种都携带一组标准属性，包括 gen_ai.request.model、gen_ai.usage.input_tokens、gen_ai.usage.output_tokens、gen_ai.response.finish_reasons 等。这样一来，一个团队产出的 trace，与另一个完全不同的框架产出的 trace，在结构上是可比的。

下面是对一个小的工具调用型 Agent 手动埋点的样子：

from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

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

def run_agent(task: str) -> str:
    # 整次运行的根 span；下面每一步都嵌套在它内部，
    # 这正是产生父子树结构的原因
    with tracer.start_as_current_span("invoke_agent") as agent_span:
        agent_span.set_attributes({
            "gen_ai.system": "openai",
            "agent.name": "support-agent",
            "gen_ai.request.model": "gpt-4o",
        })

        messages = [
            {"role": "system", "content": "You are a support assistant."},
            {"role": "user", "content": task},
        ]

        while True:
            # 模型调用本身拥有自己的子 span
            with tracer.start_as_current_span("chat") as chat_span:
                response = model_client.chat.completions.create(
                    model="gpt-4o",
                    messages=messages,
                    tools=AVAILABLE_TOOLS
                )
                choice = response.choices[0]
                chat_span.set_attributes({
                    "gen_ai.response.model": response.model,
                    "gen_ai.usage.input_tokens": response.usage.prompt_tokens,
                    "gen_ai.usage.output_tokens": response.usage.completion_tokens,
                })

            if choice.finish_reason != "tool_calls":
                agent_span.set_status(Status(StatusCode.OK))
                return choice.message.content

            # 每次工具调用拥有自己的子 span，嵌套在 Agent 运行之下，
            # 而不是嵌套在 chat span 之下，因为工具调用是
            # 与推理并列的步骤，而不是推理的子步骤
            for tool_call in choice.message.tool_calls:
                with tracer.start_as_current_span("execute_tool") as tool_span:
                    tool_span.set_attributes({
                        "gen_ai.tool.name": tool_call.function.name,
                        "gen_ai.tool.call.id": tool_call.id,
                    })
                    try:
                        result = call_tool(tool_call.function.name, tool_call.function.arguments)
                    except Exception as e:
                        tool_span.record_exception(e)
                        tool_span.set_status(Status(StatusCode.ERROR, str(e)))
                        raise

                messages.append({
                    "role": "tool",
                    "content": str(result),
                    "tool_call_id": tool_call.id,
                })

这段代码的结构要点在于层级关系：invoke_agent 是整次运行的根 span，chat 与 execute_tool 都是它的子 span，彼此并列。工具调用不是推理的子步骤，把它挂在 chat 下面会让瀑布图的语义失真。异常通过 record_exception 与 set_status 显式记录到 span 上，这样即使异常继续向上抛出，trace 里也留下了完整的错误上下文。

第三步：读取 trace 瀑布图

追踪数据落到后端之后，最有价值的视图是瀑布图。它按时间轴展开一次运行的全部 span，直观呈现每一步的耗时与嵌套关系。

阅读瀑布图时，重点看这几类信号：

• 重复的工具调用。 同一个工具名在短时间内出现两次，参数略有差异，这往往就是开头那个客服场景的根源——Agent 基于第二次结果作答，而第一次结果被丢弃了。
• 异常的 token 消耗。 某个 chat span 的输入 token 远高于同类型步骤的常规水平，通常意味着上下文被反复累积或检索结果过多。
• 耗时分布。 是模型推理慢，还是某个工具执行慢，瀑布图上一眼可辨，不需要再去猜。
• 缺失的结束事件。 某个 span 没有正常收尾，说明执行路径在中途被中断。

第四步：用指标跟踪 token 成本

由于成本与延迟的驱动因素已经从请求数转移到 token，指标采集的重点也应随之调整。从 span 属性中提取 gen_ai.usage.input_tokens 与 gen_ai.usage.output_tokens，按模型、按 Agent 名称、按工具维度聚合，就能回答一些传统监控回答不了的问题：哪一类任务最烧 token、哪个工具调用之后总是跟着一次昂贵的重试、输入 token 与输出 token 的比例是否在某个版本之后发生了漂移。

把这些指标与 trace ID 关联起来，就能从“这个月成本涨了”一路下钻到“是这一类请求里的某一步在反复重试”。

第五步：用 trace 数据排查真实故障

回到开头那个场景，排查流程大致是这样：

1. 从客户反馈或业务指标异常出发，定位到具体的那次运行，拿到 trace ID。
2. 用 trace ID 过滤日志，取出这次运行的全部结构化日志行。
3. 在瀑布图上确认工具调用的次数与顺序，找出重复调用的那两次。
4. 对比两次调用的参数差异，以及各自返回结果的长度。
5. 检查紧随其后的 chat span，确认模型是基于哪一次结果生成的答复。

这套流程之所以可行，前提是日志与追踪从一开始就通过 trace ID 打通了。如果日志是自由文本、没有 trace ID，第 2 步就无从下手，只能靠时间戳碰运气。

一个完整示例

把前面的片段串起来，一个最小可运行的 Agent 运行流程如下。

首先完成初始化：配置日志、取得 tracer、准备工具列表。

import logging
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode

logger = logging.getLogger("agent")
logging.basicConfig(level=logging.INFO)
tracer = trace.get_tracer("agent-service")

AVAILABLE_TOOLS = [
    # 这里填入工具定义，例如退款查询、订单查询等
]

接着定义工具执行函数，并在其中埋入结构化日志。注意日志里记录的是参数与结果长度，而不是完整结果：

def execute_tool(tool_name: str, arguments: dict):
    # 实际调用外部系统或内部服务
    ...

def call_tool(tool_name: str, arguments: dict):
    span = trace.get_current_span()
    trace_id = format(span.get_span_context().trace_id, "032x")

    logger.info(
        "tool_call_started",
        extra={"trace_id": trace_id, "tool_name": tool_name, "arguments": arguments},
    )
    try:
        result = execute_tool(tool_name, arguments)
        logger.info(
            "tool_call_succeeded",
            extra={"trace_id": trace_id, "tool_name": tool_name, "result_length": len(str(result))},
        )
        return result
    except Exception as e:
        logger.error(
            "tool_call_failed",
            extra={"trace_id": trace_id, "tool_name": tool_name, "error": str(e)},
        )
        raise

然后运行 Agent 主循环，为整次运行、每次模型推理、每次工具调用分别创建 span：

def run_agent(task: str) -> str:
    with tracer.start_as_current_span("invoke_agent") as agent_span:
        agent_span.set_attributes({
            "gen_ai.system": "openai",
            "agent.name": "support-agent",
            "gen_ai.request.model": "gpt-4o",
        })

        messages = [
            {"role": "system", "content": "You are a support assistant."},
            {"role": "user", "content": task},
        ]

        while True:
            with tracer.start_as_current_span("chat") as chat_span:
                response = model_client.chat.completions.create(
                    model="gpt-4o",
                    messages=messages,
                    tools=AVAILABLE_TOOLS
                )
                choice = response.choices[0]
                chat_span.set_attributes({
                    "gen_ai.response.model": response.model,
                    "gen_ai.usage.input_tokens": response.usage.prompt_tokens,
                    "gen_ai.usage.output_tokens": response.usage.completion_tokens,
                })

            if choice.finish_reason != "tool_calls":
                agent_span.set_status(Status(StatusCode.OK))
                return choice.message.content

            for tool_call in choice.message.tool_calls:
                with tracer.start_as_current_span("execute_tool") as tool_span:
                    tool_span.set_attributes({
                        "gen_ai.tool.name": tool_call.function.name,
                        "gen_ai.tool.call.id": tool_call.id,
                    })
                    try:
                        result = call_tool(tool_call.function.name, tool_call.function.arguments)
                    except Exception as e:
                        tool_span.record_exception(e)
                        tool_span.set_status(Status(StatusCode.ERROR, str(e)))
                        raise

                messages.append({
                    "role": "tool",
                    "content": str(result),
                    "tool_call_id": tool_call.id,
                })

最后调用它：

answer = run_agent("帮我查一下订单 12345 的退款状态")
print(answer)

运行之后，在后端应当能看到一棵以 invoke_agent 为根的 span 树，下面挂着若干 chat 与 execute_tool 子 span，每个子 span 上带有模型名、token 用量、工具名与调用 ID 等属性。同时，日志中每一行都带有同一个 trace ID，可以直接用它在日志系统里过滤出这次运行的全部记录。

注意事项

落地过程中有几个容易踩的坑，值得提前留意。

• 不要把完整提示词和完整工具输出写进日志。 提示词经常包含真实的个人信息或机密内容，原样落库会在产生调试价值之前先制造合规问题。记录长度、截断预览或经过脱敏的摘要通常已经够用。
• 不要只记录结果，不记录开始。 只有结束事件就无法测量单次调用的耗时，也无法判断某一步是否卡住。
• 不要把工具调用挂在模型推理 span 之下。 工具调用是与推理并列的步骤，挂错层级会让瀑布图的语义失真，排查时容易误判因果关系。
• 不要依赖单次 trace 下结论。 由于相同输入不保证产生相同行为，一次“测试通过”的 trace 说明不了真实运行的分布。需要结合多次运行的聚合指标来看。
• 不要把请求数当作成本与延迟的主要指标。 在 Agent 场景下，这两个量都与 token 消耗相关，围绕每秒请求数搭建的监控对此结构性失明。
• span 属性命名尽量遵循 GenAI 语义约定。 自定义命名会让不同团队、不同框架产出的 trace 无法横向比较，也会让后端的一些通用分析能力失效。
• 异常要同时记录到 span 和日志。 只抛出不记录，trace 上就看不到错误上下文；只记录不抛出，上层逻辑可能继续在错误状态下运行。

文中涉及的模型名称、属性字段、语义约定版本以及各依赖库的具体行为，都可能随版本演进而变化，实际使用时请以相关项目官网的当前信息为准。