Agent 可观测性：线上出问题，怎么追踪和查日志

Agent 线上答错了，你连它当时调了什么、为什么这么决定都不知道——这不是夸张，是大多数第一次上线 Agent 的团队真实踩过的坑。

工单进来：用户说 Agent 给出了一个完全不对的结论。你打开服务器日志，看到的是一行"请求处理成功，耗时 3.2s"。里面发生了什么？它调了几个工具？第二个工具的返回值是啥？模型为什么没用那个返回值？一概不知。你只能凭猜测重现，凭运气复现。

这就是"没有可观测"的代价：出了事，你不是在排查，你是在赌。

---

为什么 Agent 比普通服务更需要可观测性

普通 API 服务，一次请求 → 一次响应，出错了看参数、看返回码、看堆栈，基本能定位。Agent 不一样：

一次用户请求，内部可能发生十几步。模型想、决定调哪个工具、工具执行、模型看结果再想、再调工具……每一步都可能出问题，而且问题之间会叠加——第二步工具返回了个空值，模型用空值"推理"出了一个错误结论，到第五步你看到的已经是被掩盖了好几层的症状。

还有一个 Agent 特有的问题：模型的决策不是代码，你看不到"为什么"。普通 bug，你能读 if/else 理解逻辑；Agent 调错工具，你只知道它调错了——但它当时拿到的上下文是什么、描述让它产生了什么判断，这些藏在那次会话的历史里，没记下来就永远找不回来。

加上 token 计费，一个失控的 Agent（比如工具一直报错，它一直重试）能在几分钟内烧掉你没预料到的费用，没有监控你甚至不知道在烧。

所以：Agent 上线，可观测不是"锦上添花"，是基础设施，跟异常捕获一个级别。

---

该记哪五类信息

先搞清楚要记什么，再谈怎么记。

① Trace：完整决策链路

这是最核心的一类。一次完整的 Agent trace 至少要包含：

• 输入：用户原始消息是什么
• 每一步模型决策：模型这一轮说了什么、stop_reason 是什么
• 工具调用：调了哪个工具、传的参数是什么（这是定位"调错工具"的关键）
• 工具返回：工具返回了什么（注意敏感信息脱敏，见后面的坑）
• 下一步决策：模型拿到工具结果后怎么判断的
• 最终结果：给用户的回答是什么

这条链路要用一个 trace_id 串起来。同一次会话的所有日志条目都带同一个 trace_id，出了事你用 trace_id 过滤，整条链路一目了然，不用在海量日志里手动对时间戳。

② Token 与成本

每次模型调用记下 input_tokens / output_tokens，折算成费用（以官方最新价格为准）。这两个数字能告诉你：某一类请求是不是异常烧钱、哪个工具调用回传了超长内容撑大了 token 消耗。

同时记累计 token：一次多轮 Agent 会话总共用了多少 token，方便发现"这个 Agent 越聊越贵"的问题。

③ 延迟

细分记每一步耗时：整体端到端延迟、模型推理耗时、每个工具执行耗时。这样能分清楚是模型慢还是工具慢——很多时候外部 API 工具慢，优化方向完全不同。

④ 错误与重试

工具调用失败了几次？模型有没有因为工具失败而一直重试？每次错误的类型是什么（网络超时、业务逻辑拒绝、参数校验失败）？

记这些的目的是：发现"重试风暴"——Agent 在某个工具上卡了，模型不断重试，token 和时间都在白烧。没有这个记录你根本看不出来。

⑤ 最终结果质量

这一类往往被忽略，但在持续改进阶段最有用。可以是：

• 用户有没有追问（间接信号：追问越多说明第一次没答好）
• 有没有触发人工审核或用户反馈
• 如果有 eval 流程，记下 eval 分数

这类信息配合 trace，能让你做"结果差 → 回查当时链路"的闭环分析。关于 eval 本身的方法，见 Agent eval：非确定性输出怎么测（6.1 节）。

---

怎么落地：结构化日志 + trace ID + 工具选型

结构化日志

日志输出 JSON，不要 print 字符串。字符串日志出了事靠正则匹配，JSON 日志直接用日志平台查询字段。每条日志至少包含：

{
  "trace_id": "agt-20260625-abc123",
  "session_id": "usr-session-xyz",
  "step": 3,
  "event": "tool_call",
  "tool_name": "search_db",
  "tool_input": {"query": "用户ID 12345 的订单"},
  "tool_output_preview": "返回 3 条记录",
  "tokens_used": {"input": 1240, "output": 88},
  "latency_ms": 342,
  "timestamp": "2026-06-25T10:23:01Z"
}

trace_id 用"前缀+日期+随机串"格式，方便按日期和业务线过滤。

trace ID 串起一次会话

会话开始时生成一个 trace_id，这次 Agent 运行的所有日志都带这个 ID。如果 Agent 是多轮对话，还需要一个 session_id 把多次运行关联起来。

两个 ID 分工：trace_id 标识"这一次 Agent 跑"，session_id 标识"这个用户的这段对话"。排查单次异常用 trace_id，分析用户体验用 session_id。

可观测工具选型

现成工具（以官方文档为准，功能/价格随时变化）：

• LangSmith（LangChain 出品）：对 LangChain 生态集成最顺滑，提供可视化 trace 树、eval 功能、数据集管理。如果你用 LangChain/LangGraph，接入成本最低。详情见 LangSmith 官方文档。
• Langfuse：开源可自托管，对框架无关，SDK 轻量，社区活跃。适合不想把数据传给第三方的团队。详情见 Langfuse 官方文档。
• OpenTelemetry 思路：如果你的团队已经有 OTel 基础设施（Jaeger/Grafana Tempo），可以把 Agent 的 span 接入进去，复用现有监控体系。Agent 每一步作为一个 span，trace 视图天然就有了。详情见 OpenTelemetry 官方文档。

自建轻量日志：如果暂时不接第三方，最低成本是：结构化 JSON 日志 → 写文件或发到 ELK/Loki → Grafana 查询。这套接入成本低，但可视化不如专门的 LLM 可观测工具直观。

工具本质上都是帮你收集 → 存储 → 查询 → 可视化这四步，选哪个看你的团队栈和数据主权要求，别被名字迷惑。

---

给 Agent 加结构化 trace 日志：代码示例

下面这段代码是可复制粘贴的轻量 trace 装饰器，不依赖任何第三方可观测 SDK，直接用标准库实现结构化日志。在这基础上，你可以替换成任何你选的可观测工具。

import anthropic
import json
import logging
import time
import uuid
from datetime import datetime, timezone
from functools import wraps
from typing import Any, Callable

# ────────────────────────────────────────────────
# 1. 结构化日志配置：输出 JSON，方便日志平台解析
# ────────────────────────────────────────────────
class JsonFormatter(logging.Formatter):
    def format(self, record):
        log_obj = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
        }
        if isinstance(record.msg, dict):
            log_obj.update(record.msg)
        else:
            log_obj["message"] = record.getMessage()
        return json.dumps(log_obj, ensure_ascii=False)

logger = logging.getLogger("agent_trace")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)


# ────────────────────────────────────────────────
# 2. trace 上下文：用 trace_id 串起一次完整会话
# ────────────────────────────────────────────────
class TraceContext:
    def __init__(self, session_id: str = None):
        self.trace_id = f"agt-{datetime.now(timezone.utc).strftime('%Y%m%d')}-{uuid.uuid4().hex[:8]}"
        self.session_id = session_id or f"sess-{uuid.uuid4().hex[:8]}"
        self.step = 0
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        self.start_time = time.time()

    def next_step(self) -> int:
        self.step += 1
        return self.step

    def log(self, event: str, **kwargs):
        """统一日志入口，自动带上 trace_id / session_id / step"""
        logger.info({
            "trace_id": self.trace_id,
            "session_id": self.session_id,
            "step": self.step,
            "event": event,
            **kwargs
        })


# ────────────────────────────────────────────────
# 3. 工具执行追踪：记录调用、返回、耗时、错误
# ────────────────────────────────────────────────
def trace_tool(ctx: TraceContext, tool_name: str, tool_input: dict, func: Callable) -> Any:
    t0 = time.time()
    ctx.log("tool_call_start", tool_name=tool_name, tool_input=tool_input)
    try:
        result = func(**tool_input)
        latency_ms = int((time.time() - t0) * 1000)
        # 只记前 200 字符，避免超长返回撑大日志
        preview = str(result)[:200] + ("…" if len(str(result)) > 200 else "")
        ctx.log("tool_call_end",
                tool_name=tool_name,
                status="ok",
                result_preview=preview,
                latency_ms=latency_ms)
        return result
    except Exception as e:
        latency_ms = int((time.time() - t0) * 1000)
        ctx.log("tool_call_error",
                tool_name=tool_name,
                error=str(e),
                latency_ms=latency_ms)
        return f"工具执行失败: {e}"


# ────────────────────────────────────────────────
# 4. 带 trace 的 Agent 主循环
# ────────────────────────────────────────────────
def run_agent_with_trace(user_message: str, tools: list, tool_functions: dict,
                         session_id: str = None, max_rounds: int = 10) -> str:
    client = anthropic.Anthropic()
    ctx = TraceContext(session_id=session_id)
    messages = [{"role": "user", "content": user_message}]

    ctx.log("agent_start", user_message=user_message)

    for round_num in range(max_rounds):
        ctx.next_step()
        t0 = time.time()

        response = client.messages.create(
            model="claude-opus-4-5",   # 以官方文档为准（截稿 2026-06）
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )

        latency_ms = int((time.time() - t0) * 1000)
        input_tokens = response.usage.input_tokens
        output_tokens = response.usage.output_tokens
        ctx.total_input_tokens += input_tokens
        ctx.total_output_tokens += output_tokens

        ctx.log("model_response",
                stop_reason=response.stop_reason,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                latency_ms=latency_ms)

        if response.stop_reason == "end_turn":
            final_text = "".join(
                block.text for block in response.content if hasattr(block, "text")
            )
            total_ms = int((time.time() - ctx.start_time) * 1000)
            ctx.log("agent_end",
                    status="success",
                    total_rounds=round_num + 1,
                    total_input_tokens=ctx.total_input_tokens,
                    total_output_tokens=ctx.total_output_tokens,
                    total_latency_ms=total_ms,
                    result_preview=final_text[:200])
            return final_text

        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []

            for block in response.content:
                if block.type == "tool_use":
                    result = trace_tool(ctx, block.name, block.input,
                                        tool_functions[block.name])
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result),
                    })

            messages.append({"role": "user", "content": tool_results})
        else:
            ctx.log("unexpected_stop_reason", stop_reason=response.stop_reason)
            break

    ctx.log("agent_end", status="max_rounds_exceeded", rounds=max_rounds)
    return "Agent 未在最大轮次内完成任务"

关键设计说明：

• TraceContext 在一次 Agent 运行开始时创建，trace_id 全局唯一，所有日志自动带上。
• trace_tool 包裹每次工具执行，记录入参、返回预览（截断防撑大）、耗时、异常。
• max_rounds 防止无限循环烧 token，超了就退出并记录。
• 所有日志用 JsonFormatter 输出，日志平台可直接按字段查询。

---

怎么用 trace 定位"它为什么调错工具"

这是最常见的线上排查场景。有了 trace 日志，排查步骤是这样的：

第一步：找到那次 trace

用出问题的时间范围或用户 session_id 过滤日志，找到对应的 trace_id。

第二步：按 trace_id 拉出完整链路

trace_id = "agt-20260625-abc123" 的所有事件按 step 排序：
step 1 → model_response: stop_reason=tool_use
step 1 → tool_call_start: tool_name="search_orders", tool_input={"user_id": null}
step 1 → tool_call_end: status="ok", result_preview="返回空列表"
step 2 → model_response: stop_reason=tool_use
step 2 → tool_call_start: tool_name="search_kb", tool_input={"query": "用户订单查询"}
...

第三步：找到第一个"异常信号"

上面的例子，tool_input={"user_id": null} 就是问题所在——模型调了正确的工具，但传入了 null 参数。这说明在这一步之前，用户 ID 没有被正确提取。往前翻 step 0 的 user_message 确认：用户原始消息里有没有提供 user_id？模型是解析失败了，还是工具的 input_schema 定义得不够严格？

结论：如果没有 tool_input 的完整记录，你只能看到"返回了空列表"，永远找不到根因是参数传错了。

---

故障排查表

症状	可能根因	排查方向
Agent 一直循环不停，token 暴涨	工具持续失败，模型一直重试；或 max_rounds 未设置	查 trace 里 tool_call_error 事件是否反复出现；确认代码有 max_rounds 限制
工具调用参数是 null 或空	模型未从对话中提取到必要参数；工具 description 未说明"必须传什么"	查 trace 里 tool_call_start 的 tool_input；检查工具 description 和 input_schema 里 required 字段
结果明显错误但日志显示"success"	最终结果质量未记录；eval 缺失	在 agent_end 事件加上结果校验逻辑；参考 Agent eval 方法 增加质量度量
单次请求延迟极高，但不知道慢在哪	某个工具执行极慢；或模型推理轮次多	看 trace 里各步骤 latency_ms；对比 tool_call_end 和 model_response 的耗时，定位瓶颈
日志量大但查不到想要的信息	日志字段设计不合理，关键字段没记；或没有 trace_id 关联	回溯这五类信息，逐一检查日志里有没有覆盖；加 trace_id / session_id 索引

---

三类常见坑

坑一：日志太多，根本没法看

新上线的团队常见做法：把所有内容都 print 出来，包括完整的工具返回、完整的模型 response。结果日志平台一天几十 GB，搜索慢、存储贵，出了事反而在日志海里找不到重点。

解法：设计分级日志。工具返回只记 preview（前 200 字符），完整内容只在 debug 级别记录；正常运行只开 INFO 级别；怀疑有问题的请求可以临时调成 DEBUG 重跑。

坑二：没记工具入参

很多人记了工具名，但没记 tool_input。结果你能看到"调了 search_orders"，但不知道它传的 user_id 是什么——这样的日志排查起来跟没有差不多。

解法：tool_input 必记，这是定位"调错工具"和"参数传错"的核心字段。

坑三：敏感信息进了日志

工具入参可能包含用户的手机号、身份证、密码等；工具返回可能包含用户隐私数据。日志里如果有这些，既有合规风险，也有信息安全风险。

解法：在 trace_tool 里加脱敏层，对已知敏感字段（phone、id_card、password、token 等）打码；或在工具 schema 里标注敏感字段，框架层统一处理。不要把脱敏留给每个工具函数去做，容易漏。

---

常见问题

Q：我现在用的是低代码平台（Coze/Dify），有没有可观测？

低代码平台通常提供运行日志，但粒度和可控性不如自建。Dify 有 trace 功能，Coze 的日志粒度相对有限。如果你的 Agent 需要深入排查，建议看 Agent 部署形态：六种后端与 Cron 方式（6.3 节），评估是否需要迁到自己控制的环境。

Q：Langfuse 和 LangSmith 我该选哪个？

两个都是成熟工具，功能重叠较多。核心差别：LangSmith 对 LangChain 生态集成最顺滑，适合已经用 LangChain/LangGraph 的团队；Langfuse 开源可自托管，对框架无关，数据在自己手里。如果你不用 LangChain，优先考虑 Langfuse；如果数据不能出境，Langfuse 自托管是更好选择。具体功能和价格以官方文档为准，别根据这篇文章做最终决策。

Q：trace_id 我应该怎么传给用户，方便他们反馈时提供？

可以在响应 header 里带 X-Trace-Id，或者在 API 返回体的 metadata 字段里放 trace_id。如果是有界面的产品，可以在"反馈"入口里自动带上当前 trace_id，免得用户还要记住一个 ID。

Q：我只有几十个 Agent 请求/天，有必要搞这些吗？

量少不代表出问题的概率低——有时候恰恰是小流量下的偶发 bug 最难复现。而且早期建好可观测的习惯，等流量起来你不会手忙脚乱。最低成本做法：就用本节的轻量 trace 代码，输出到文件，不用接任何第三方工具，一天日志几百 KB，不影响性能。

---

小结

• 没有可观测的 Agent，出了事就是抓瞎——模型决策不是代码，不记下来就永远找不回来。
• 五类信息都要记：trace 链路、token/成本、延迟、错误重试、结果质量。
• trace_id 是可观测的骨架：一次 Agent 运行的所有日志带同一个 ID，排查时一条命令过滤出完整链路。
• 工具选型看团队栈和数据主权：LangSmith（LangChain 生态）/ Langfuse（框架无关/可自托管）/ OTel（已有监控基础设施）。
• 三个坑要提前规避：日志太多没法看、没记工具入参、敏感信息进日志。

下一步去看 Agent 部署形态：六种后端与 Cron 方式（6.3 节），了解生产部署的选型；或者看 可观测性：日志/监控/错误追踪与回溯，这是编程 L7 里更底层的监控体系，两套知识配合起来用。

👉 看看 AI 数字员工落地指南，或了解 数字员工搭建实战课。需要为企业落地方案，欢迎找我们聊 企业服务。