OpenTelemetry

引言 一次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的分析能够将排障效率提升数倍。 ...

Agent可观测性：你无法优化你看不见的东西 传统软件的可观测性已经相当成熟——我们有Prometheus做指标、ELK做日志、Jaeger做追踪。但Agent系统引入了新的可观测性挑战：非确定性的执行路径、不可预测的token消耗、LLM输出的质量评估、多Agent协作的链路追踪。2026年，Agent可观测性已经形成了一套独立的最佳实践。 Agent可观测性的三大支柱 ┌───────────────────────────────────────────────────┐ │ Agent 可观测性三大支柱 │ ├─────────────┬─────────────┬───────────────────────┤ │ │ │ │ │ 追踪 │ 日志 │ 指标 │ │ (Traces) │ (Logs) │ (Metrics) │ │ │ │ │ │ Agent执行 │ 结构化事件 │ 性能计数器 │ │ 链路追踪 │ 决策记录 │ 资源消耗 │ │ 跨Agent关联 │ 上下文快照 │ 质量评估 │ │ │ │ │ └─────────────┴─────────────┴───────────────────────┘ 1. 分布式追踪 Agent Trace模型 Agent系统的追踪比传统微服务更复杂，因为一个"请求"可能涉及多轮LLM调用、多次工具调用、跨多个Agent。 from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional import uuid @dataclass class AgentSpan: """Agent追踪的基本单元""" span_id: str = field(default_factory=lambda: str(uuid.uuid4())) parent_id: Optional[str] = None trace_id: str = "" # 基本信息 name: str = "" # span名称 span_type: str = "" # llm_call / tool_call / agent_step / sub_agent agent_name: str = "" # 哪个Agent start_time: datetime = field(default_factory=datetime.now) end_time: Optional[datetime] = None # Agent特有信息 input_data: Any = None output_data: Any = None model: str = "" # 使用的LLM模型 prompt_tokens: int = 0 completion_tokens: int = 0 cost: float = 0.0 # 状态 status: str = "ok" # ok / error / timeout error_message: Optional[str] = None # 元数据 attributes: dict = field(default_factory=dict) def finish(self, output=None, status="ok", error=None): self.end_time = datetime.now() self.output_data = output self.status = status self.error_message = error @property def duration_ms(self) -> float: if self.end_time: return (self.end_time - self.start_time).total_seconds() * 1000 return 0 追踪实现 class AgentTracer: """Agent追踪器""" def __init__(self, service_name="agent-service"): self.service_name = service_name self.spans: list[AgentSpan] = [] self.current_span_stack: list[AgentSpan] = [] def start_trace(self, name: str, agent_name: str) -> AgentSpan: """开始一个新的追踪（根span）""" trace_id = str(uuid.uuid4()) span = AgentSpan( trace_id=trace_id, name=name, span_type="agent_step", agent_name=agent_name ) self.spans.append(span) self.current_span_stack.append(span) return span def start_span( self, name: str, span_type: str, agent_name: str = "", input_data: Any = None ) -> AgentSpan: """开始一个子span""" parent = self.current_span_stack[-1] if self.current_span_stack else None span = AgentSpan( trace_id=parent.trace_id if parent else str(uuid.uuid4()), parent_id=parent.span_id if parent else None, name=name, span_type=span_type, agent_name=agent_name, input_data=input_data ) self.spans.append(span) self.current_span_stack.append(span) return span def end_span(self, span: AgentSpan, output=None, status="ok", error=None): """结束一个span""" span.finish(output, status, error) if self.current_span_stack and self.current_span_stack[-1].span_id == span.span_id: self.current_span_stack.pop() def get_trace_tree(self, trace_id: str) -> dict: """获取追踪树""" trace_spans = [s for s in self.spans if s.trace_id == trace_id] return self._build_tree(trace_spans, parent_id=None) def _build_tree(self, spans: list[AgentSpan], parent_id: str | None) -> dict: children = [s for s in spans if s.parent_id == parent_id] return [ { "span_id": s.span_id, "name": s.name, "type": s.span_type, "agent": s.agent_name, "duration_ms": s.duration_ms, "tokens": s.prompt_tokens + s.completion_tokens, "cost": s.cost, "status": s.status, "children": self._build_tree(spans, s.span_id) } for s in children ] 使用示例 tracer = AgentTracer() # 开始追踪 root = tracer.start_trace("用户咨询", "router_agent") # Agent执行LLM调用 llm_span = tracer.start_span("LLM调用", "llm_call", "router_agent", input_data="用户问题") response = await llm.complete("...") tracer.end_span(llm_span, output=response.text, status="ok") llm_span.prompt_tokens = response.usage.prompt_tokens llm_span.completion_tokens = response.usage.completion_tokens llm_span.cost = calculate_cost(response.usage, "gpt-4o") # Agent调用工具 tool_span = tracer.start_span("搜索知识库", "tool_call", "router_agent") results = await knowledge_base.search("query") tracer.end_span(tool_span, output=results) # 路由到子Agent sub_agent_span = tracer.start_span("专家Agent处理", "sub_agent", "expert_agent") # ... 子Agent内部会有自己的span ... tracer.end_span(sub_agent_span, output="最终回答") # 结束追踪 tracer.end_span(root, output="最终回答") # 查看追踪树 trace_tree = tracer.get_trace_tree(root.trace_id) 追踪可视化 追踪树可以渲染为瀑布图： ...

引言 Agent 系统的"黑盒"问题是生产化的最大障碍之一。一个 Agent 调用了 3 个工具、经过 5 轮推理、消耗了 8000 tokens，但出了问题你却不知道在哪一步。2026年，OpenTelemetry 社区正式发布了 SemConv for GenAI 规范，为 LLM 可观测性提供了标准化方案。 一、为什么 Agent 可观测性不同于传统应用 传统微服务的可观测性关注：请求路径、延迟分布、错误率。Agent 系统增加了三个新维度： Token 维度：每次调用消耗多少 Token？成本如何分摊？ 推理维度：模型"想"了什么？为什么选择这个工具？为什么跳过某步？ 非确定性维度：相同输入可能产生不同输出，仅靠日志无法复现 二、OpenTelemetry GenAI 语义规范 2026年正式定稿的 GenAI SemConv 定义了以下核心 Attributes： # GenAI 基础属性 gen_ai.system: "openai" # 提供商 gen_ai.request.model: "gpt-5" # 模型名称 gen_ai.request.temperature: 0.7 # 采样温度 gen_ai.request.max_tokens: 4096 # 最大 Token # Token 使用 gen_ai.usage.input_tokens: 1523 # 输入 Token gen_ai.usage.output_tokens: 876 # 输出 Token gen_ai.usage.cost: 0.0234 # 本次调用成本（美元） # Agent 特有 gen_ai.agent.name: "research-agent" gen_ai.agent.tool.name: "web_search" gen_ai.agent.tool.result.quality: 0.85 gen_ai.agent.iteration: 3 # 第几轮迭代 # 工具调用 gen_ai.tool.name: "calculator" gen_ai.tool.input: '{"expr": "2+2"}' gen_ai.tool.output: '{"result": 4}' gen_ai.tool.duration_ms: 45 三、全链路追踪实现 架构概览 ┌─────────────────────────────────────────────────────┐ │ Agent Application │ │ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ │ │ │Step1│──►│Step2│──►│Step3│──►│Step4│ │ │ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ │ │ │ │ │ │ │ │ ┌──▼─────────▼─────────▼─────────▼──┐ │ │ │ OpenTelemetry SDK │ │ │ │ (Auto-instrumentation + Custom) │ │ │ └──────────────┬────────────────────┘ │ └─────────────────┼───────────────────────────────────┘ │ OTLP/gRPC ┌────────────▼────────────┐ │ OTel Collector │ │ (处理/采样/导出) │ └──┬─────┬─────┬──────────┘ │ │ │ ┌────▼┐ ┌─▼──┐ ┌▼─────┐ │Jaeger│ │Prom│ │Loki │ │(Trace)│ │(Met)│ │(Log)│ └─────┘ └────┘ └──────┘ Python 实现 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.instrumentation.openai import OpenAIInstrumentor # 1. 初始化 OTel provider = TracerProvider() processor = BatchSpanProcessor( OTLPSpanExporter(endpoint="http://otel-collector:4317") ) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # 2. 自动注入 OpenAI 调用的 Span OpenAIInstrumentor().instrument() # 3. Agent 自定义 Span tracer = trace.get_tracer("agent-system") class ObservabilityMiddleware: """Agent 可观测性中间件""" def __init__(self): self.tracer = trace.get_tracer("agent") async def on_agent_start(self, agent_name: str, input_data: dict): """Agent 启动时创建 Root Span""" self.root_span = self.tracer.start_span( f"agent.{agent_name}", attributes={ "gen_ai.agent.name": agent_name, "agent.input.size": len(str(input_data)), } ) async def on_llm_call(self, model: str, messages: list, **kwargs): """LLM 调用前记录""" ctx = trace.set_span_in_context(self.root_span) span = self.tracer.start_span( f"llm.{model}", context=ctx, attributes={ "gen_ai.request.model": model, "gen_ai.request.message_count": len(messages), "gen_ai.request.temperature": kwargs.get("temperature", 1.0), } ) return span async def on_llm_end(self, span, response): """LLM 调用后记录 Token 使用""" usage = response.usage span.set_attributes({ "gen_ai.usage.input_tokens": usage.prompt_tokens, "gen_ai.usage.output_tokens": usage.completion_tokens, "gen_ai.usage.total_tokens": usage.total_tokens, "gen_ai.usage.cost": calculate_cost( usage.prompt_tokens, usage.completion_tokens, response.model ), }) span.end() async def on_tool_call(self, tool_name: str, tool_input: dict): """工具调用追踪""" ctx = trace.set_span_in_context(self.root_span) span = self.tracer.start_span( f"tool.{tool_name}", context=ctx, attributes={ "gen_ai.tool.name": tool_name, "gen_ai.tool.input": json.dumps(tool_input)[:500], } ) return span async def on_agent_end(self, output: str): """Agent 结束""" self.root_span.set_attributes({ "agent.output.size": len(output), "agent.status": "success", }) self.root_span.end() Trace 可视化示例 在 Jaeger 中看到的典型 Agent Trace： ...