本文整理自 OpenTelemetry Semantic Conventions for Generative AI Systems、Langfuse Data Model、Arize: Best AI Observability Tools for Autonomous Agents in 2026、Morph AI Agent Tracing，并结合实际开发的一些经验进行整理。

Agent Tracing 设计

Agent 运行一次任务，背后可能包含几十次 LLM 调用、上百次工具执行、若干次检索与治理决策。如果只看最终输出，出问题后几乎无法定位：模型是在哪一轮开始跑偏的？哪个工具返回了错误但没有被正确处理？ token 和成本到底花在哪里？Tracing 解决的就是这个问题。Agent 的 trace 可以记录和揭示 Agent 链路中的层级结构（session / turn / tool / generation）、因果关系（哪次 tool call 对应哪次 LLM 调用）、业务语义（governance decision、approval、artifact）以及成本归因（token、latency、money）。

1. Agent Trace 基础

$a.$ OpenTelemetry GenAI Semantic Conventions

OpenTelemetry 的 GenAI Semantic Conventions 把生成式 AI 的遥测统一在 gen_ai.* 命名空间下。它目前还在快速演进，但核心方向已经稳定：把 LLM 调用、工具执行、Agent 编排、检索、记忆都映射到同一套 span taxonomy 上。

核心 gen_ai.operation.name 包括：

operation	含义
`chat`	一次聊天补全调用
`text_completion`	文本补全
`generate_content`	多模态生成
`embeddings`	向量嵌入
`execute_tool`	工具执行
`invoke_agent`	调用一个 agent
`create_agent`	创建 agent
`invoke_workflow` / `plan`	工作流/规划
`retrieval` / `search_memory` / `create_memory`	RAG 与记忆操作

一个典型的 ReAct turn 会被表达成一棵树：

invoke_agent (root)
├── chat (模型第一次推理)
├── execute_tool (read_file)
├── chat (模型根据工具结果再次推理)
├── execute_tool (write_file)
└── chat (最终回答)

规范还定义了事件：gen_ai.client.inference.operation.details 把完整请求参数以事件形式独立存储；gen_ai.evaluation.result 把评估结果挂到被评估的 span/trace 上。

这些约定的好处是：无论底层框架是 LangChain、LlamaIndex、CrewAI 还是自研框架，只要输出同样语义的 span，后端就能统一理解。

$b.$ Langfuse 的数据模型

Langfuse 是 Agent Trace 的经典开源组件之一。它的数据模型分为三层：

Session：把同一用户或同一业务任务的多个 trace 归为一组；
Trace：一次端到端请求，所有共享同一 trace_id 的 observation 构成一个 trace；
Observation：单个步骤，可嵌套。类型包括 event、span、generation、agent、tool、chain、retriever、evaluator、embedding、guardrail。

Langfuse 的 generation 专门表示一次 LLM API 调用，自动聚合 input/output token、成本、延迟、模型参数；span 表示有开始/结束时间的工作单元；event 表示时间点事件。

Langfuse 底层构建在 OpenTelemetry 之上，支持 OTLP 接收，也提供原生 SDK。这意味着既可以用 Langfuse 做 LLM 可观测，也可以把同一份 telemetry 同时发给 Datadog、Grafana、Braintrust 等传统 APM。

理解了这两层协议基础，我们就可以讨论如何自己实现一套不绑定特定后端的 Agent Tracing 架构了。

注：后面的 Agent Trace 是基于 Langfuse 的，其他的可以自己看文档噢，思路差不多的。

2. Agent Trace 组件统一架构

$a.$ 为什么要引入统一架构

在某个生产 Agent 的实现过程中，我使用 Langfuse 监测 Agent 的一些外挂插件的运行链路，每个模块独立在 Agent Loop 的 Hooks 中往 Langfuse 发送 Span 以及监听 Hooks，这造成了下面的问题：

同一次 tool/llm 调用被多个插件重复记录；
各平台使用不同的 trace_id，无法表示彼此间的关联关系；
tool span 与组件间的父子关系不稳定：原先组件应该是 tool 下面的子调用，但是由于各自为政导致它们变成同级的了；
观测后端故障可能拖慢甚至阻塞 Agent 主链路。

第三点有些绕，举个下面的例子，原本应该是下面这样的：

llm
└── tool:read_file
  └── governance:check  ← 正确的父子关系

但是如果每个插件或者模块都在自己内部实现 Trace 而不是让一个完全独立的组件来统一管理的话，由于模块内不清楚自己的父调用的信息，就会变成这样：

llm
├── tool:read_file        ← 后创建
└── governance:check      ← 先创建，但 parent 错了

这些问题的解决方式是建立一套后端无关的统一事件基础设施：业务代码只生产一次 canonical event，Langfuse、SQLite、SSE、Agent Governance 等都基于这个创建自己需要的视图，也即一次业务操作只生产一次 canonical event，所有平台只是投影。

$b.$ 具体架构

Agent Trace 组件的具体架构如下：

+-------------------------------------------------------+
|                     Agent Runtime                     |
+-------------------------------------------------------+
                           |
                           v
+-------------------------------------------------------+
|             Framework Adapter / Producer              |
|  - Listen to Agent Runtime lifecycle hooks            |
|  - Convert runtime events to canonical events         |
|  - Provide built-in local sinks (SQLite / SSE)        |
+-------------------------------------------------------+
                           |
                           v
+-------------------------------------------------------+
|                Canonical Event Library                |
|  - TraceContext / AgentEvent                          |
|  - EventEmitter / register_sink                       |
|  - Unified protocol decoupled from framework/backend  |
+-------------------------------------------------------+
                           |
         +-----------------+-----------------+
         |                 |                 |
         v                 v                 v
+-----------------+ +---------------+ +-----------------+
| Backend Adapter | | Backend Adapt | | Backend Adapter |
|  Langfuse Sink  | |  SQLite Sink  | |  SSE/JSONL Sink |
+-----------------+ +---------------+ +-----------------+

不同层次之间的关系如下：

[ Agent Runtime ]
       │
       ▼ (Raw Hooks)
[ Framework Adapter ] ──► Produce CanonicalEvent
       │
       ▼
[ Event Library ] ─────► Dispatch / Fan-out
       │
       ├─────────────────┬─────────────────┐
       ▼                 ▼                 ▼
[ Langfuse Sink ]  [ SQLite Sink ]   [ SSE / JSONL ]
       │                 │                 │
       ▼ (Project)       ▼ (Store)         ▼ (Stream)
[ Langfuse UI ]    [ SQLite DB ]     [ Client / Log ]

具体而言：

业务语义层 / Agent Runtime：只关心“现在发生了什么业务事件”，产出的是原始 hook 或运行时事件，不关心后面怎么存、怎么展示；
Framework Adapter / Producer：监听 Agent Runtime 的生命周期 hook，把原始运行时事件转换为 canonical event，同时也可以提供本地基础 sink（SQLite/SSE）；
Canonical Event Library：定义统一的事件 schema、trace context、错误分类、脱敏策略，提供 emit() API 和 register_sink()，负责把事件统一分发给所有注册的 sink；
后端适配层 / Backend Adapters：每个 sink 通过自己实现的消费接口独立消费 canonical event，映射到 Langfuse observation、OTel span、SQLite row、SSE stream 等。

这个划分的好处是：新增一个观测后端只需要新增一个 Backend Adapter；修改 Agent Runtime 的 hook 转换逻辑只需要改 Framework Adapter；业务语义演进只需要改 canonical event 定义，sink 会自动跟随；而 Canonical Event Library 里的核心协议可以稳定不变、被多个不同框架复用。这实现了不同模块的解耦。

$c.$ 核心抽象

$i.$ TraceContext：统一 Trace ID

TraceContext 用于记录一次事件属于哪次业务链路、哪次会话、哪一轮、哪一步、哪一次重试等：

@dataclass(frozen=True)
class TraceContext:
  trace_id: str        # 业务链路：UUID5(task_id or session_id)
  session_id: str = ""
  task_id: str = ""
  turn_id: str = ""    # 一次用户输入触发的一轮
  step_id: str = ""    # 当前操作：llm:xxx / tool:xxx / skill:xxx
  parent_step_id: str = ""
  attempt_id: str = "" # 物理重试尝试

trace_id 用 UUID5 从 task_id/session_id 确定性生成，不同插件/进程无需共享状态即可得到相同 ID；
step_id 相同表示逻辑操作，attempt_id 递增表示物理重试，避免 error 与 success 被错误合并（否则只用 step_id 的话后面尝试的结果会覆盖前面的）；
child() 方法默认将 parent_step_id 设为当前 step_id，形成 span 树。

$ii.$ AgentEvent：统一事件信封

统一事件信封把“发生了什么”与“怎么展示”解耦。payload 里放业务数据，metadata 里放平台无关的上下文，error 用结构化描述符而不是原始 traceback。

@dataclass
class AgentEvent:
  event_type: str            # llm.start / tool.end / governance.error ...
  context: TraceContext
  status: str = "ok"
  producer: str = "agent_trace"
  payload: Dict[str, Any] = field(default_factory=dict)
  metadata: Dict[str, Any] = field(default_factory=dict)
  error: Optional[Dict[str, Any]] = None
  artifact_refs: list[Dict[str, Any]] = field(default_factory=list)
  duration_ms: Optional[int] = None
  schema_version: str = "agent.event.v1"

$iii.$ ErrorDescriptor：入口无关的错误分类

Agent Runtime 中的错误可能来自 LLM API、工具执行、治理决策、网络超时、参数校验等。不同入口的错误如果直接塞进 trace，后端看到的是五花八门的字符串。统一分类后，无论是 Langfuse 还是 AlertManager，看到的都是一致的 error_class 和 retryable 标记。

class ErrorClass(str, Enum):
  TEMPORARY_NETWORK = "TEMPORARY_NETWORK"
  TIMEOUT = "TIMEOUT"
  TOOL_EXECUTION_ERROR = "TOOL_EXECUTION_ERROR"
  TOOL_CONTRACT_ERROR = "TOOL_CONTRACT_ERROR"
  RESOURCE_NOT_FOUND = "RESOURCE_NOT_FOUND"

@dataclass(frozen=True)
class ErrorDescriptor:
  error_class: ErrorClass
  code: str
  message: str
  retryable: bool
  source: str = "runtime"
  details: Dict[str, Any] = field(default_factory=dict)

$iv.$ EventEmitter：把事件发送到各个 Sink

业务代码只调用 emit(event)，后端的 Langfuse、SQLite、SSE 等 sink 各自独立消费。每个 sink 有自己的有界队列和后台 worker，一个 sink 阻塞不会影响其他 sink，也不会拖垮 Agent 主链路。

class EventEmitter:
  def __init__(self):
    self._sinks: Dict[str, _SinkWorker] = {}

  def register_sink(self, name: str, sink: Sink, maxsize: int = 2048):
    self._sinks[name] = _SinkWorker(name, sink, maxsize)

  def emit(self, event: AgentEvent):
    safe = redact_payload(event.payload)
    event.payload = safe
    for worker in self._sinks.values():
      worker.enqueue(event)  # 非阻塞 put_nowait，队列满则丢弃并标记 degraded

$d.$ 与 OpenTelemetry / Langfuse 的映射

Canonical Event 层不需要直接依赖 OTel SDK，因为 OTel Span 语义无法直接表达 task/turn/attempt/governance decision/artifact_ref 等业务概念。合理的层次是：

Business canonical event → OTel/Langfuse adapter。

Canonical Event	Langfuse Observation	OTel Span Kind
`llm.start` / `llm.end`	`generation`	`chat`
`tool.start` / `tool.end`	`tool`	`execute_tool`
`skill.` / `resource.`	`span`	`invoke_agent` / custom
`governance.*`	`guardrail`	custom
`session/turn`	`chain` / root span	`invoke_agent`

Langfuse 对 usage key 的聚合规则是：所有含 input 的 key 计为 input total，含 output 的 key 计为 output total。因此在映射前要做统一归一化：

{
  "input": input_tokens,
  "output": output_tokens,
  "cache_read_input_tokens": cache_read_tokens,
  "cache_creation_input_tokens": cache_write_tokens,
  "reasoning_tokens": reasoning_tokens,
}

一个简化的 Langfuse Sink 实现如下。只需要根据 event_type 决定 observation 类型，并把 trace_id、parent_step_id、input/output 挂上即可。如果 governance 事件比 tool start 先到达，sink 内部需要缓存 pending children，等 parent observation 创建后再回放，保证父子关系正确，否则就会出现前面说的父子关系不稳定问题：

class LangfuseSink:
  def __call__(self, event: AgentEvent):
    ctx = event.context
    otype = "generation" if event.event_type.startswith("llm.") else \
            "tool" if event.event_type.startswith("tool.") else \
            "guardrail" if event.event_type.startswith("governance.") else "span"

    if event.event_type.endswith(".start"):
      if otype == "generation":
        self.langfuse.generation(
          id=ctx.step_id,
          trace_id=ctx.trace_id,
          parent_observation_id=ctx.parent_step_id or None,
          name=event.payload.get("model", "llm"),
          input=event.payload.get("input_summary"),
          model=event.payload.get("model"),
        )
      else:
        self.langfuse.span(
          id=ctx.step_id,
          trace_id=ctx.trace_id,
          parent_observation_id=ctx.parent_step_id or None,
          name=event.payload.get("name", otype),
          input=event.payload.get("input_summary"),
        )

3. 业务 Agent Tracing 的关键设计

$a.$ Trace ID

$i.$ 根据业务 ID 生成

如果业务相关逻辑不是直接耦合在 Agent Loop 中、它们不在同一线程时，可以通过下面的方式在对每一个业务 event 生成唯一 Trace ID：

import uuid

NAMESPACE_TRACE = uuid.UUID("6ba7b810-9dad-11d1-80b4-00c04fd430c8")

def make_trace_id(task_id: str = "", session_id: str = "") -> str:
  seed = task_id or session_id
  return str(uuid.uuid5(NAMESPACE_TRACE, seed))

# 同一任务下，所有模块用同一个 task_id 计算，得到相同 trace_id；session_id 亦然
trace_id = make_trace_id(task_id="task-xxx")

只要所有模块都从同一个 task_id 或 session_id 计算，就能得到相同的 trace_id。

$ii.$ 隐式上下文传播：降低业务代码侵入

如果业务相关逻辑耦合在 Agent Loop 中、在同一个线程时，可以直接让 Agent Loop 生成一个 trace_id 运行：

await runWithRunContext({ runId, traceId, observer, stepExecutor }, () =>
  runAgentTurn(),
);

$iii.$ 与框架 Hook 的配合

很多业务 Agent 会基于一个自研或开源框架构建。框架通常提供 on_session_start、on_tool_start 等 hook，业务 Agent 不需要从头实现传播，只需要在 hook 回调里把框架的上下文转换为业务 TraceContext，再 emit canonical event 即可。这里就简单带过了。

$b.$ 生命周期覆盖

业务 Agent 的生命周期不是简单的 session/turn/llm/tool，而是包含业务语义的业务步骤，例如下面的例子：

session.start
├── turn.start
│   ├── llm.request / llm.response
│   ├── tool.start
│   │   └── tool.end
│   ├── skill.start / skill.end
│   └── resource.start / resource.end
├── approval.request / approval.response
└── session.end

业务层的 tracing 需要回答“模型是在哪一步做出错误决策的”，所以它要把业务 step 类型（如 intent_route、retrieval、governance）显式表达出来，而不是只靠 Trace 框架的 llm / tool（比如 Langfuse），这样才能在排查问题时能还原完整决策链。因此我们需要记录更详细的 trace，比如：

llm.request 要记录 model、message_count、tool_count；
tool.start 要记录 tool_name、args、是谁发起的（parent_step_id）；
governance.end 要记录 decision（allow/audit/approval/deny），并且 parent 指向对应的 tool；
subagent.start 要传播 parent trace context，避免子 agent 产生无法关联的匿名 trace。

在具体实现中，只需要在对应阶段 emit trace 即可，例如：

context = TraceContext(
  trace_id=make_trace_id(task_id="task-xxx"),
  session_id="session-yyy",
  turn_id="turn-1",
  step_id="tool:read_file:abc",
  parent_step_id="llm:turn-1",
)

emit(AgentEvent(
  event_type="tool.start",
  context=context,
  payload={
    "tool_name": "read_file",
    "args": {"path": "config.yaml"},
  },
))

$c.$ 对悬挂操作的处理

如果在 session 结束时若仍有未闭合操作的，可以自动 emit llm.error / tool.error，避免 Langfuse 留下永久 open observation。这是个简单的小细节。

$d.$ 观测系统可以失败，业务不能失败

这是业务 Agent Tracing 最重要的工程原则之一。比如前面的 EventQueue，每个 sink 维护独立有界队列和后台 worker。队列满时丢弃事件，并标记 degraded 而不是阻塞主线程：

def enqueue(self, event: Any) -> bool:
  if self.degraded and time.monotonic() < self.next_attempt_at:
    self.dropped += 1
    return False
  try:
    self.queue.put_nowait(event)
    return True
  except Full:
    self.dropped += 1
    self._mark_degraded("queue_full")
    return False

4. Agent 框架的 Tracing 设计

前面的讨论主要围绕业务 Agent，而业务 Agent 一般会基于某个框架上开发。框架层的 Tracing 设计与业务 Agent 不同：它不感知业务语义，但要提供通用的生命周期事件、可插拔的观测接口，并且不能与执行控制混在一起。

$a.$ 通用生命周期事件

框架关心的是所有 Agent 都共有的生命周期：

session.start
├── turn.start
│   ├── llm.request
│   ├── llm.response
│   ├── llm.usage
│   ├── tool.start
│   │   └── tool.end
│   ├── plan.step.start
│   │   └── plan.step.end
│   └── subagent.start
│       └── subagent.stop
├── turn.end
└── session.end

框架不需要知道“这次 LLM 调用是在做意图路由还是最终总结”，它只记录“发生了一次 llm.request，model 是什么，usage 多少”。业务 Agent 在这些通用事件之上再添加自己的业务 step。

$b.$ Hook 机制

通过 Hooks 机制暴露生命周期扩展点。业务 Agent 只订阅 hook，不直接操作 trace。

class ObservabilityHooks(Hooks):
  async def on_session_start(self, ctx):
    self._start_span(ctx, "nonoka.run", {
      "nonoka.session_id": ctx.session.session_id,
      "gen_ai.request.model": ctx.agent.model,
    })

4.3 ExecutionTrace 与 Checkpoint 集成

框架通常把 trace 字段放进 SessionState，checkpoint 会序列化它，恢复时重建。

以 nonoka-agent 举例，nonoka agent 的 ObservabilityPipeline 中，本地 store 是同步写、失败了会抛出异常，第三方 exporter 则是 best-effort 的设计：

from typing import Protocol, runtime_checkable

@runtime_checkable
class Exporter(Protocol):
  async def emit(
    self,
    session_id: str,
    event_type: str,
    payload: dict,
  ) -> None:
    ...

class ObservabilityPipeline:
  def __init__(
    self,
    store: EventStore,
    exporters: list[Exporter] | None = None,
  ):
    self.store = store
    self.exporters = exporters or []

  async def append(
    self,
    session_id: str,
    event_type: str,
    payload: dict,
  ):
    # 1. 先脱敏，再写入任何存储
    safe = redact_payload(payload)

    # 2. 本地 store 同步写，失败上抛
    await self.store.append(session_id, event_type, safe)

    # 3. 第三方 exporter best-effort，失败吞掉
    for exporter in self.exporters:
      try:
        await exporter.emit(session_id, event_type, safe)
      except Exception:
        continue

Agent Tracing 设计

1. Agent Trace 基础

a.a.a. OpenTelemetry GenAI Semantic Conventions

b.b.b. Langfuse 的数据模型

2. Agent Trace 组件统一架构

a.a.a. 为什么要引入统一架构

b.b.b. 具体架构

c.c.c. 核心抽象

i.i.i. TraceContext：统一 Trace ID

ii.ii.ii. AgentEvent：统一事件信封

iii.iii.iii. ErrorDescriptor：入口无关的错误分类

iv.iv.iv. EventEmitter：把事件发送到各个 Sink

d.d.d. 与 OpenTelemetry / Langfuse 的映射