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 丢失是两个必须处理的工程问题。