Agent 可观测性 2026：OpenTelemetry for LLM 实践

引言

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：

agent.research-agent                    [2.3s]
├── llm.gpt-5                          [800ms]  in:1523 out:456 tokens
│   └── gen_ai.tool_call: web_search   [120ms]
├── tool.web_search                    [340ms]
│   └── http.get                       [280ms]
├── llm.gpt-5 (summarize)              [600ms]  in:2340 out:380 tokens
├── tool.write_file                    [45ms]
└── llm.gpt-5 (finalize)               [415ms]  in:890 out:234 tokens
    Total: in=4753 out=1070 cost=$0.0312

四、关键指标设计

Golden Signals for Agent

from opentelemetry import metrics

meter = metrics.get_meter("agent.metrics")

# 1. Agent 执行延迟
agent_duration = meter.create_histogram(
    "agent.duration",
    unit="ms",
    description="Agent execution duration"
)

# 2. Token 消耗
token_usage = meter.create_histogram(
    "agent.token.usage",
    unit="tokens",
    description="Token usage per agent run"
)

# 3. 工具调用成功率
tool_success = meter.create_counter(
    "agent.tool.success",
    description="Successful tool calls"
)
tool_failure = meter.create_counter(
    "agent.tool.failure",
    description="Failed tool calls"
)

# 4. Agent 自终止率
agent_self_terminate = meter.create_counter(
    "agent.terminate.self",
    description="Agent self-terminated (max iterations, etc.)"
)

# 5. 成本追踪
cost_counter = meter.create_counter(
    "agent.cost.total",
    unit="USD",
    description="Total LLM cost"
)

Prometheus 告警规则

# Agent 执行超时告警
- alert: AgentHighLatency
  expr: histogram_quantile(0.95, agent_duration_bucket) > 30000
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "Agent P95 latency > 30s"

# Token 消耗异常
- alert: TokenUsageSpike
  expr: rate(agent_token_usage_sum[5m]) > 100000
  for: 2m
  labels:
    severity: critical
  annotations:
    summary: "Token consumption > 100k/min"

# 工具失败率
- alert: ToolFailureRate
  expr: |
    rate(agent_tool_failure_total[5m]) / 
    (rate(agent_tool_success_total[5m]) + rate(agent_tool_failure_total[5m])) > 0.1
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "Tool failure rate > 10%"

五、结构化日志最佳实践

import structlog

logger = structlog.get_logger()

# 每条日志都携带 Trace 上下文
def log_agent_step(
    agent_name: str,
    step: str,
    trace_id: str,
    span_id: str,
    **kwargs
):
    logger.info(
        "agent.step",
        agent=agent_name,
        step=step,
        trace_id=trace_id,
        span_id=span_id,
        **kwargs  # 额外字段
    )

# 使用示例
log_agent_step(
    agent_name="research-agent",
    step="tool_selection",
    trace_id=current_trace_id(),
    span_id=current_span_id(),
    available_tools=["search", "calculator", "write"],
    selected_tool="search",
    selection_confidence=0.92,
    reasoning="User asked about latest news, search tool is most relevant"
)

六、成本归因模型

将 Token 成本精确归因到业务维度：

class CostAttribution:
    """Agent 成本归因模型"""
    
    def __init__(self):
        self.costs = {}  # {dimension: total_cost}
    
    def record(
        self,
        user_id: str,
        agent_name: str,
        task_type: str,
        tokens_in: int,
        tokens_out: int,
        model: str
    ):
        cost = self._calculate(tokens_in, tokens_out, model)
        
        # 多维度归因
        dimensions = [
            f"user:{user_id}",
            f"agent:{agent_name}",
            f"task:{task_type}",
            f"model:{model}",
        ]
        for dim in dimensions:
            self.costs[dim] = self.costs.get(dim, 0) + cost
    
    def report(self) -> dict:
        """生成成本报告"""
        return {
            "by_user": self._group_by("user"),
            "by_agent": self._group_by("agent"),
            "by_task": self._group_by("task"),
            "total": sum(self.costs.values()),
        }

七、采样策略

Agent 高频调用会导致 Trace 数据爆炸。推荐分层采样：

class AgentSampler:
    def __init__(self):
        self.error_sampler = AlwaysOnSampler()      # 错误必采
        self.slow_sampler = AlwaysOnSampler()       # 慢请求必采
        self.normal_sampler = TraceIdRatio(0.05)   # 正常请求采5%
    
    def should_sample(self, span) -> bool:
        if span.attributes.get("error"):
            return self.error_sampler.should_sample()
        if span.duration_ms > 10000:
            return self.slow_sampler.should_sample()
        return self.normal_sampler.should_sample()

结语

可观测性不是奢侈品，而是 Agent 生产化的基础设施。OpenTelemetry 为 LLM 可观测性提供了统一标准，让你不再被绑定在特定供应商的监控工具上。投资可观测性的回报是：更快的故障定位、更精准的成本优化、更高的用户信任。在 Agent 时代，看见即掌控。

加入讨论

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

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

引言#

一、为什么 Agent 可观测性不同于传统应用#

二、OpenTelemetry GenAI 语义规范#

三、全链路追踪实现#

架构概览#

Python 实现#

Trace 可视化示例#

四、关键指标设计#

Golden Signals for Agent#

Prometheus 告警规则#

五、结构化日志最佳实践#

六、成本归因模型#

七、采样策略#

结语#

加入讨论#

📌 相关推荐

引言