Agent 全链路追踪：一次对话在模型、工具、存储间的完整路径

Agent 全链路追踪：一次对话在模型、工具、存储间的完整路径

一、为什么一次简单的对话却要翻几十个服务日志

排查 Agent 对话问题时，最让人抓狂的不是业务逻辑，而是追踪。一次对话经过 Prompt 拼接、模型调用、工具选择、工具执行、结果注入、二次推理……涉及五六个组件、十几个调用。出问题后，你得在模型服务的 Pod 日志、工具服务的 Sentinel 面板、向量数据库的 Query Log、消息队列的消费位点之间来回切，时间都花在"找日志"而非"读日志"。

核心矛盾在于：Agent 系统的调用链路远比传统微服务复杂。传统 API Gateway 进来的请求是一条线，Agent 的思维链本质上是一棵树——模型可能在推理中途分叉出多个工具调用，每个工具调用又可能产生子链。传统的 traceId 只能追踪线性调用，撑不起这种拓扑。

我们后来落地了一套方案：把 Agent 会话的全链路追踪，拆成 Session、Turn、Step、Span 四层模型，配合 OpenTelemetry 的 Baggage 机制注入对话上下文。排查问题时，输入session_id就能捞出整个对话的完整轨迹。

二、四层追踪模型与上下文传播机制

Agent 的调用特点决定了不能简单套用 RPC 的 Span 嵌套模型。Agent 的执行是"推理-行动-观测"的循环（ReAct 模式），每个循环是一个 Decision Point，模型在这个点上决定调用哪些工具。

下面是我们的四层追踪模型：

graph TD Session["Session: 一次完整对话"] Turn1["Turn 1: 第一轮对话"] Turn2["Turn 2: 第二轮对话"] Step1["Step: Prompt构建 + LLM调用"] Step2["Step: 工具调用 - 搜索"] Step3["Step: 工具调用 - 计算"] Step4["Step: 结果注入 + 二次推理"] Session --> Turn1 Session --> Turn2 Turn1 --> Step1 Turn1 --> Step2 Turn1 --> Step3 Turn1 --> Step4 SubStep2a["Span: API Request"] SubStep2b["Span: Response Parse"] SubStep3a["Span: Code Execute"] SubStep3b["Span: Sandbox Cleanup"] Step2 --> SubStep2a Step2 --> SubStep2b Step3 --> SubStep3a Step3 --> SubStep3b style Session fill:#4A90D9,color:#fff style Turn1 fill:#50B86C,color:#fff style Turn2 fill:#50B86C,color:#fff style Step1 fill:#F5A623,color:#fff style Step2 fill:#F5A623,color:#fff style Step3 fill:#F5A623,color:#fff style Step4 fill:#F5A623,color:#fff

• Session：一次完整对话，对应session_id。
• Turn：每轮用户输入+Agent回复构成一个 Turn，对应turn_id。
• Step：Agent 思维链中的一个决策点，可能是 LLM 推理、工具调用或结果整合。
• Span：对应 OpenTelemetry 的 Span，是单一的 RPC 调用或函数执行。

关键设计是在 Span 的 Attributes 里注入完整的对话上下文：session_id、turn_id、step_id、tool_name、model_name。这样 Collector 端可以通过session_id聚合出完整树，不需要应用层额外维护关联表。

上下文传播通过 OpenTelemetry 的 Context Propagation 实现。在 Python 侧：

from opentelemetry import trace, context from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator propagator = TraceContextTextMapPropagator()

将当前 trace context 注入到工具调用的 HTTP 请求头中，工具服务侧再提取还原。这样即使工具是外部服务，也能接上链路。

三、生产级实现：Trace Collector + 查询聚合

核心代码分两层：Agent 侧的埋点封装和 Collector 侧的聚合查询。

Agent 侧埋点

import time from typing import Optional, Dict, Any 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 from opentelemetry.sdk.resources import SERVICE_NAME, Resource # 初始化 TracerProvider，使用 OTLP 协议上报 # 为什么用 BatchSpanProcessor 而非 SimpleSpanProcessor： # Agent 的 Span 密度很高（一次对话可能有数十个 Span）， # 同步上报会造成 LLM 调用的额外等待，批量异步上报将延迟控制在 100ms 以内 resource = Resource(attributes={SERVICE_NAME: "agent-runtime"}) provider = TracerProvider(resource=resource) processor = BatchSpanProcessor( OTLPSpanExporter(endpoint="http://otel-collector:4317", insecure=True), # 每 512 个 Span 或每 5 秒刷新一次，避免内存堆积 max_export_batch_size=512, schedule_delay_millis=5000, ) provider.add_span_processor(processor) trace.set_tracer_provider(provider) tracer = trace.get_tracer(__name__) class AgentTraceContext: """封装 Agent 对话层级的 trace 上下文""" def __init__(self, session_id: str, user_id: str = ""): self.session_id = session_id self.user_id = user_id self.turn_counter = 0 def create_turn_span(self, user_input: str) -> trace.Span: """为每轮对话创建一个 Turn Span""" self.turn_counter += 1 turn_span = tracer.start_span( f"agent_turn_{self.turn_counter}", attributes={ "session.id": self.session_id, "turn.id": f"{self.session_id}:{self.turn_counter}", "user.id": self.user_id, "turn.input": user_input[:500], # 截断，避免 Span 过大 "turn.timestamp": int(time.time()), } ) return turn_span def create_step_span( self, parent_span: trace.Span, step_type: str, step_name: str, extra_attrs: Optional[Dict[str, Any]] = None ) -> trace.Span: """在 Turn 下创建 Step Span""" # 使用父 Span 的 Context 建立层级关系 ctx = trace.set_span_in_context(parent_span) step_span = tracer.start_span( f"{step_type}:{step_name}", context=ctx, attributes={ "session.id": self.session_id, "step.type": step_type, # "llm_call" | "tool_call" | "reasoning" "step.name": step_name, "step.timestamp": int(time.time()), **(extra_attrs or {}), } ) return step_span def propagate_context(self, headers: Dict[str, str], current_span: trace.Span): """ 将当前 trace context 注入到 HTTP headers 中。 调用外部工具时，必须调用此方法传递上下文。 """ ctx = trace.set_span_in_context(current_span) propagator = TraceContextTextMapPropagator() propagator.inject(headers, context=ctx) # 使用示例：Agent 执行循环 def agent_run(session_id: str, user_input: str): ctx = AgentTraceContext(session_id, user_id="user_123") with ctx.create_turn_span(user_input) as turn_span: try: # Step 1: Prompt 构建 with ctx.create_step_span(turn_span, "llm_call", "prompt_build") as step: prompt = build_prompt(user_input) step.set_attribute("prompt.length", len(prompt)) # Step 2: LLM 调用 with ctx.create_step_span(turn_span, "llm_call", "model_inference") as step: response = call_llm(prompt, trace_ctx=ctx, parent_span=step) step.set_attribute("model.name", "gpt-4") step.set_attribute("response.tokens", response.token_count) # Step 3: 工具调用（如果 LLM 返回工具调用） if response.tool_calls: for tool_call in response.tool_calls: with ctx.create_step_span( turn_span, "tool_call", tool_call.name, extra_attrs={"tool.args": str(tool_call.args)[:500]} ) as tool_span: result = execute_tool(tool_call, trace_ctx=ctx, parent_span=tool_span) tool_span.set_attribute("tool.success", result.success) tool_span.set_attribute("tool.duration_ms", result.duration_ms) # Step 4: 结果注入 + 二次推理 with ctx.create_step_span(turn_span, "llm_call", "final_response") as step: final = call_llm_with_tools_result(prompt, tool_results, trace_ctx=ctx, parent_span=step) turn_span.set_attribute("turn.success", True) return final except Exception as e: turn_span.set_attribute("turn.success", False) turn_span.record_exception(e) raise

Collector 侧聚合查询

import asyncpg from typing import List, Dict, Any from dataclasses import dataclass @dataclass class TraceSpan: session_id: str turn_id: str step_type: str step_name: str span_id: str parent_span_id: str start_time: int duration_ms: int attributes: Dict[str, Any] class TraceQueryService: """基于 ClickHouse 的全链路查询服务""" def __init__(self, clickhouse_dsn: str): # ClickHouse 在宽表查询和时序聚合上秒杀 MySQL， # Agent 的 Span 数据写入量大（每天百万级），必须用列存 self.pool = None self.clickhouse_dsn = clickhouse_dsn async def init(self): self.pool = await asyncpg.create_pool(self.clickhouse_dsn, min_size=5, max_size=20) async def query_session_trace(self, session_id: str) -> List[TraceSpan]: """根据 session_id 查询完整对话链路""" query = """ SELECT session_id, turn_id, step_type, step_name, span_id, parent_span_id, toUnixTimestamp(start_time) as start_time, duration_ms, attributes FROM agent_traces WHERE session_id = $1 AND date >= today() - INTERVAL 7 DAY -- 限制时间范围，避免全表扫描 ORDER BY start_time ASC """ async with self.pool.acquire() as conn: rows = await conn.fetch(query, session_id) traces = [] for row in rows: traces.append(TraceSpan( session_id=row['session_id'], turn_id=row['turn_id'], step_type=row['step_type'], step_name=row['step_name'], span_id=row['span_id'], parent_span_id=row['parent_span_id'], start_time=row['start_time'], duration_ms=row['duration_ms'], attributes=row['attributes'], )) return traces async def analyze_session_performance(self, session_id: str) -> Dict[str, Any]: """分析单次会话的性能瓶颈""" query = """ SELECT step_type, step_name, count() as call_count, avg(duration_ms) as avg_duration, quantile(0.95)(duration_ms) as p95_duration, max(duration_ms) as max_duration FROM agent_traces WHERE session_id = $1 GROUP BY step_type, step_name ORDER BY avg_duration DESC """ async with self.pool.acquire() as conn: rows = await conn.fetch(query, session_id) bottlenecks = [] for row in rows: bottlenecks.append({ "step_type": row['step_type'], "step_name": row['step_name'], "call_count": row['call_count'], "avg_duration_ms": row['avg_duration'], "p95_duration_ms": row['p95_duration'], "max_duration_ms": row['max_duration'], }) return {"session_id": session_id, "bottlenecks": bottlenecks}

四、这套方案不适合哪些场景

缺点：

1. Span 膨胀问题：一个复杂对话可能产生 50+ 个 Span，Collector 的存储成本上升很快。需要配置采样策略（对成功且<100ms 的 Span 按 10% 采样）。
2. Batching 带来的延迟：BatchSpanProcessor 在进程异常退出时会丢失最后一批 Span。关键操作应设置OnEnd回调写本地日志兜底。
3. 跨语言传播的一致性：Baggage 的传播在 Go/Python/Node.js 的 OTel SDK 表现不完全一致，混用语言时需要 Mock 测试验证。

禁用场景：

• 对延迟极度敏感、不允许额外 1-2ms 埋点开销的实时系统。
• 对话内容包含强合规数据（如金融交易指令），不可将输入采样到 Span Attributes 中的场景。

五、总结

Agent 全链路追踪的核心，是用"对话语义层"替代传统的"RPC 调用层"来组织链路。四层模型（Session→Turn→Step→Span）适配了 Agent 的树状执行特征，配合 OpenTelemetry 的 Baggage 机制实现了跨服务的上下文透传。落地时，Span 采样策略和异常退出的 Span 丢失是两个必须处理的工程问题。