Agent链路追踪：OpenTelemetry与Jaeger实战

引言

一次Agent对话可能涉及路由决策、向量检索、工具调用、LLM推理等十几个步骤，跨越多个微服务。当用户反馈"Agent回复变慢了"时，如果没有链路追踪，定位瓶颈就像大海捞针。OpenTelemetry + Jaeger的组合为Agent系统提供了端到端的请求追踪能力，让每一次对话的完整路径都清晰可见。

OpenTelemetry架构

┌──────────────────────────────────────────────────────────┐
│                  OpenTelemetry Architecture               │
│                                                          │
│  Agent Services                                          │
│  ┌─────────────────────────────────────────────┐         │
│  │  ┌──────┐  ┌──────┐  ┌──────┐  ┌──────┐  │         │
│  │  │Router│  │ Tool │  │ LLM  │  │Memory│  │         │
│  │  │Service│ │Service│ │Service│ │Service│ │         │
│  │  └──┬───┘  └──┬───┘  └──┬───┘  └──┬───┘  │         │
│  │     │         │         │         │       │         │
│  │  ┌──▼─────────▼─────────▼─────────▼───┐  │         │
│  │  │     OpenTelemetry SDK (Instrument)  │  │         │
│  │  └──────────────┬─────────────────────┘  │         │
│  └─────────────────┼───────────────────────┘         │
│                    │                                    │
│           ┌────────▼────────┐                          │
│           │  OTLP Exporter  │                          │
│           └────────┬────────┘                          │
│                    │                                    │
│           ┌────────▼────────┐                          │
│           │  OTel Collector │                          │
│           │  (接收+处理+导出) │                          │
│           └────┬───────┬────┘                          │
│                │       │                                │
│         ┌──────▼┐  ┌──▼──────┐                        │
│         │Jaeger  │  │Prometheus│                        │
│         │(Traces)│  │(Metrics) │                        │
│         └───────┘  └─────────┘                        │
└──────────────────────────────────────────────────────────┘

SDK初始化

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 Resource
from opentelemetry.instrumentation.grpc import GrpcInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor

def setup_tracing(service_name: str, otlp_endpoint: str = "otel-collector:4317"):
    """初始化OpenTelemetry追踪"""
    
    resource = Resource.create({
        "service.name": service_name,
        "service.version": "2.0.0",
        "deployment.environment": "production",
    })
    
    provider = TracerProvider(resource=resource)
    
    # OTLP导出器
    exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True)
    processor = BatchSpanProcessor(
        exporter,
        max_queue_size=2048,
        max_export_batch_size=512,
        export_timeout_millis=30000,
    )
    provider.add_span_processor(processor)
    
    trace.set_tracer_provider(provider)
    
    # 自动注入HTTP/gRPC调用
    GrpcInstrumentor().instrument()
    HTTPXClientInstrumentor().instrument()
    
    return trace.get_tracer(service_name)

Agent Span设计

class AgentTracer:
    """Agent专用追踪器"""
    
    def __init__(self, tracer):
        self.tracer = tracer
    
    async def trace_request(
        self,
        session_id: str,
        user_input: str,
        handler: callable
    ):
        """追踪完整请求链路"""
        with self.tracer.start_as_current_span(
            "agent.request",
            attributes={
                "session.id": session_id,
                "input.length": len(user_input),
                "input.language": self._detect_language(user_input),
            }
        ) as request_span:
            try:
                result = await handler()
                request_span.set_attribute(
                    "response.length", len(result.get("response", ""))
                )
                request_span.set_attribute(
                    "response.quality_score", 
                    result.get("quality_score", 0)
                )
                request_span.set_status(trace.StatusCode.OK)
                return result
            except Exception as e:
                request_span.record_exception(e)
                request_span.set_status(
                    trace.Status(trace.StatusCode.ERROR, str(e))
                )
                raise
    
    async def trace_routing(
        self,
        user_input: str,
        routing_fn: callable
    ):
        """追踪路由决策"""
        with self.tracer.start_as_current_span(
            "agent.route",
            attributes={"input.preview": user_input[:100]}
        ) as span:
            decision = await routing_fn()
            
            span.set_attributes({
                "route.model": decision.get("model", ""),
                "route.tools": ",".join(decision.get("tools", [])),
                "route.confidence": decision.get("confidence", 0),
                "route.reason": decision.get("reason", ""),
            })
            return decision
    
    async def trace_tool_call(
        self,
        tool_name: str,
        params: dict,
        executor: callable
    ):
        """追踪工具调用"""
        with self.tracer.start_as_current_span(
            f"tool.{tool_name}",
            attributes={
                "tool.name": tool_name,
                "tool.params_hash": hashlib.md5(
                    json.dumps(params, sort_keys=True).encode()
                ).hexdigest()[:8],
            }
        ) as span:
            start = time.monotonic()
            try:
                result = await executor()
                latency_ms = (time.monotonic() - start) * 1000
                
                span.set_attributes({
                    "tool.latency_ms": latency_ms,
                    "tool.success": True,
                    "tool.result_size": len(str(result)),
                })
                return result
            except Exception as e:
                span.set_attributes({
                    "tool.success": False,
                    "tool.error": str(e)[:200],
                })
                span.record_exception(e)
                raise
    
    async def trace_llm_call(
        self,
        model: str,
        prompt: str,
        generator: callable
    ):
        """追踪LLM推理"""
        with self.tracer.start_as_current_span(
            f"llm.{model}",
            attributes={
                "llm.model": model,
                "llm.prompt_length": len(prompt),
            }
        ) as span:
            start = time.monotonic()
            result = await generator()
            latency_ms = (time.monotonic() - start) * 1000
            
            span.set_attributes({
                "llm.latency_ms": latency_ms,
                "llm.prompt_tokens": result.get("usage", {}).get("prompt_tokens", 0),
                "llm.completion_tokens": result.get("usage", {}).get("completion_tokens", 0),
                "llm.total_tokens": result.get("usage", {}).get("total_tokens", 0),
                "llm.finish_reason": result.get("finish_reason", ""),
            })
            return result

Span层级示例

agent.request (session=abc123) [2000ms]
├── agent.route [15ms]
│   ├── embedding.generate [8ms]
│   └── similarity.match [5ms]
├── memory.retrieve [50ms]
│   ├── vector.search [30ms]
│   └── context.assemble [15ms]
├── tool.search [800ms]
│   ├── http.get [600ms]
│   └── result.parse [50ms]
├── tool.calculator [20ms]
├── llm.gpt-4o [1100ms]
│   ├── prompt.build [5ms]
│   ├── api.call [1050ms]
│   └── response.parse [30ms]
└── response.format [15ms]

上下文传播

from opentelemetry.propagate import inject, extract
from opentelemetry.trace import get_current_span

class TraceContextPropagator:
    """跨服务Trace上下文传播"""
    
    @staticmethod
    def inject_to_headers(headers: dict = None) -> dict:
        """注入trace context到HTTP头"""
        headers = headers or {}
        inject(headers)
        return headers
    
    @staticmethod
    def extract_from_headers(headers: dict):
        """从HTTP头提取trace context"""
        return extract(headers)
    
    @staticmethod
    def get_current_trace_id() -> str:
        """获取当前trace ID"""
        span = get_current_span()
        if span and span.is_recording():
            return format(span.get_span_context().trace_id, "032x")
        return ""
    
    @staticmethod
    def get_current_span_id() -> str:
        """获取当前span ID"""
        span = get_current_span()
        if span and span.is_recording():
            return format(span.get_span_context().span_id, "016x")
        return ""


# 在gRPC metadata中传播
class GrpcTraceInterceptor:
    """gRPC trace拦截器"""
    
    async def intercept(self, method, request, context):
        # 从metadata提取trace context
        metadata = dict(context.invocation_metadata())
        trace_context = TraceContextPropagator.extract_from_headers(metadata)
        
        tracer = trace.get_tracer(__name__)
        
        with tracer.start_as_current_span(
            f"grpc.{method}",
            context=trace_context,
        ) as span:
            # 注入trace context到响应metadata
            response_metadata = TraceContextPropagator.inject_to_headers()
            context.set_trailing_metadata(
                [(k, v) for k, v in response_metadata.items()]
            )
            
            return await method(request, context)

Jaeger查询与分析

class JaegerAnalyzer:
    """Jaeger数据分析器"""
    
    def __init__(self, jaeger_url: str):
        self.url = jaeger_url
    
    async def find_slow_traces(
        self,
        service: str,
        min_duration_ms: int = 2000,
        limit: int = 20
    ) -> list:
        """查找慢trace"""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.url}/api/traces",
                params={
                    "service": service,
                    "limit": limit,
                    "minDuration": f"{min_duration_ms}ms",
                    "lookback": "1h",
                }
            )
            return response.json()["data"]
    
    async def analyze_bottleneck(
        self,
        trace_id: str
    ) -> dict:
        """分析trace瓶颈"""
        trace = await self._get_trace(trace_id)
        
        spans = self._flatten_spans(trace)
        
        # 找到最耗时的span
        slowest = max(spans, key=lambda s: s["duration"])
        
        # 分析Span层级
        tree = self._build_span_tree(spans)
        
        # 找到关键路径
        critical_path = self._find_critical_path(tree)
        
        return {
            "trace_id": trace_id,
            "total_duration_ms": tree["duration"],
            "slowest_span": {
                "name": slowest["operationName"],
                "duration_ms": slowest["duration"],
                "service": slowest["process"]["serviceName"],
            },
            "critical_path": [
                {
                    "span": s["operationName"],
                    "duration_ms": s["duration"],
                    "service": s["process"]["serviceName"],
                }
                for s in critical_path
            ],
            "span_count": len(spans),
        }

性能影响控制

class TracingPerformanceGuard:
    """追踪性能守护——控制追踪开销"""
    
    def __init__(self):
        self.sampling_rates = {
            "fast_path": 0.05,    # <500ms的请求5%采样
            "normal": 0.2,        # 500ms-2s的请求20%采样
            "slow": 1.0,          # >2s的请求100%采样
            "error": 1.0,         # 错误请求100%采样
        }
    
    def should_trace(self, estimated_duration_ms: int = 0) -> bool:
        """决定是否追踪"""
        if estimated_duration_ms > 2000:
            rate = self.sampling_rates["slow"]
        elif estimated_duration_ms > 500:
            rate = self.sampling_rates["normal"]
        else:
            rate = self.sampling_rates["fast_path"]
        
        return random.random() < rate
    
    @contextmanager
    def conditional_span(self, tracer, name: str, should_trace: bool):
        """条件性创建span"""
        if should_trace:
            with tracer.start_as_current_span(name) as span:
                yield span
        else:
            yield None

总结

OpenTelemetry为Agent系统提供了统一的、标准化的链路追踪能力。精心设计的Span层级让每一次Agent对话的完整路径都清晰可见——从路由决策到工具调用，从记忆检索到LLM推理。Jaeger的可视化让性能瓶颈一目了然，基于trace的分析能够将排障效率提升数倍。

核心原则：链路追踪的投入产出比极高——它不仅帮你定位问题，更帮你理解系统。在生产环境中，即使只有1%的采样率，也能捕获足够的信息用于性能分析和故障排查。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。

引言#

OpenTelemetry架构#

SDK初始化#

Agent Span设计#

Span层级示例#

上下文传播#

Jaeger查询与分析#

性能影响控制#

总结#

核心原则：链路追踪的投入产出比极高——它不仅帮你定位问题，更帮你理解系统。在生产环境中，即使只有1%的采样率，也能捕获足够的信息用于性能分析和故障排查。#

加入讨论#

📌 相关推荐

引言