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:#fffSession一次完整对话对应session_id。Turn每轮用户输入Agent回复构成一个 Turn对应turn_id。StepAgent 思维链中的一个决策点可能是 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(resourceresource) processor BatchSpanProcessor( OTLPSpanExporter(endpointhttp://otel-collector:4317, insecureTrue), # 每 512 个 Span 或每 5 秒刷新一次避免内存堆积 max_export_batch_size512, schedule_delay_millis5000, ) 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( fagent_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}, contextctx, 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, contextctx) # 使用示例Agent 执行循环 def agent_run(session_id: str, user_input: str): ctx AgentTraceContext(session_id, user_iduser_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_ctxctx, parent_spanstep) 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_ctxctx, parent_spantool_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_ctxctx, parent_spanstep) 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) raiseCollector 侧聚合查询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_size5, max_size20) 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_idrow[session_id], turn_idrow[turn_id], step_typerow[step_type], step_namerow[step_name], span_idrow[span_id], parent_span_idrow[parent_span_id], start_timerow[start_time], duration_msrow[duration_ms], attributesrow[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}四、这套方案不适合哪些场景缺点Span 膨胀问题一个复杂对话可能产生 50 个 SpanCollector 的存储成本上升很快。需要配置采样策略对成功且100ms 的 Span 按 10% 采样。Batching 带来的延迟BatchSpanProcessor 在进程异常退出时会丢失最后一批 Span。关键操作应设置OnEnd回调写本地日志兜底。跨语言传播的一致性Baggage 的传播在 Go/Python/Node.js 的 OTel SDK 表现不完全一致混用语言时需要 Mock 测试验证。禁用场景对延迟极度敏感、不允许额外 1-2ms 埋点开销的实时系统。对话内容包含强合规数据如金融交易指令不可将输入采样到 Span Attributes 中的场景。五、总结Agent 全链路追踪的核心是用对话语义层替代传统的RPC 调用层来组织链路。四层模型Session→Turn→Step→Span适配了 Agent 的树状执行特征配合 OpenTelemetry 的 Baggage 机制实现了跨服务的上下文透传。落地时Span 采样策略和异常退出的 Span 丢失是两个必须处理的工程问题。