1. 为什么 Agent 需要专门的可观测性

传统微服务可观测性关注请求-响应链路。Agent 系统则复杂得多：

  • 多步骤推理：一个用户请求可能触发 10-50 次 LLM 调用
  • 不确定输出：相同输入可能产生不同输出，难以复现问题
  • 工具调用链：Agent → tool1 → Agent → tool2 → … 形成深层调用链
  • 成本高昂：每次 LLM 调用都有实际成本，需要精准归因
  • 幻觉风险：输出质量难以量化监控

普通监控工具（Prometheus/Grafana）只覆盖指标，无法完整追踪 Agent 行为。需要专门的 Agent 可观测性方案。

2. 可观测性三大支柱

┌─────────────────────────────────────────────────┐
│               Agent Observability                │
├────────────┬──────────────┬────────────────────┤
│  Traces    │   Metrics    │      Logs          │
│  (追踪)    │   (指标)     │    (日志)          │
│            │              │                    │
│  ┌────────┐ │ ┌────────┐ │ ┌──────────────┐ │
│  │Span    │ │ │Counter │ │ │ Struct-     │ │
│  │Tree    │ │ │Gauge   │ │ │ ured Log   │ │
│  │Timeline│ │ │Histog. │ │ │ Event Log  │ │
│  └────────┘ │ └────────┘ │ └──────────────┘ │
│            │              │                    │
│  "发生了   │ "发生了几次" │ "发生了什么"     │
│   什么"    │ "耗时多久"   │ "为什么失败"     │
└────────────┴──────────────┴────────────────────┘
          统一关联：TraceID + SpanID + LogID

3. 分布式追踪设计

3.1 Agent Span 模型

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
import uuid

class SpanKind(str, Enum):
    LLM_CALL = "llm_call"
    TOOL_CALL = "tool_call"
    AGENT_STEP = "agent_step"
    RETRIEVAL = "retrieval"
    USER_INPUT = "user_input"
    SYSTEM = "system"

@dataclass
class SpanContext:
    trace_id: str
    span_id: str
    parent_span_id: Optional[str] = None

@dataclass
class SpanEvent:
    name: str
    timestamp: datetime
    attributes: dict = field(default_factory=dict)

@dataclass
class AgentSpan:
    """Agent 追踪 Span - 兼容 OpenTelemetry"""
    trace_id: str
    span_id: str
    parent_span_id: Optional[str]
    kind: SpanKind
    name: str
    agent_id: str
    session_id: str

    # 时间
    start_time: datetime = field(default_factory=datetime.now)
    end_time: Optional[datetime] = None
    duration_ms: Optional[float] = None

    # 输入/输出
    input: dict = field(default_factory=dict)
    output: dict = field(default_factory=dict)

    # 状态
    status: str = "ok"          # ok, error, timeout
    error_message: Optional[str] = None

    # 成本 & 性能
    token_usage: dict = field(default_factory=dict)  # {input: N, output: N}
    cost_usd: float = 0.0
    model: Optional[str] = None

    # 扩展属性
    attributes: dict = field(default_factory=dict)
    events: list[SpanEvent] = field(default_factory=list)
    child_spans: list[str] = field(default_factory=list)  # child span_ids

    def finish(self, status: str = "ok"):
        self.end_time = datetime.now()
        self.duration_ms = (self.end_time - self.start_time).total_seconds() * 1000
        self.status = status

    def add_event(self, name: str, attributes: dict = None):
        self.events.append(SpanEvent(
            name=name,
            timestamp=datetime.now(),
            attributes=attributes or {}
        ))

3.2 追踪器实现

class AgentTracer:
    """Agent 分布式追踪器"""
    def __init__(self, exporter: "TraceExporter"):
        self.exporter = exporter
        self._active_spans: dict[str, AgentSpan] = {}  # span_id -> span
        self._trace_tree: dict[str, list[str]] = {}    # trace_id -> [span_ids]

    def start_trace(self, session_id: str, agent_id: str,
                    name: str = "agent_session") -> tuple[str, str]:
        """开始一次 Agent 会话追踪"""
        trace_id = str(uuid.uuid4())
        root_span_id = self._new_span_id()
        span = AgentSpan(
            trace_id=trace_id,
            span_id=root_span_id,
            parent_span_id=None,
            kind=SpanKind.AGENT_STEP,
            name=name,
            agent_id=agent_id,
            session_id=session_id,
        )
        self._active_spans[root_span_id] = span
        self._trace_tree[trace_id] = [root_span_id]
        return trace_id, root_span_id

    def start_span(self, trace_id: str, parent_span_id: str, kind: SpanKind,
                   name: str, agent_id: str, session_id: str,
                   input_data: dict = None) -> str:
        span_id = self._new_span_id()
        span = AgentSpan(
            trace_id=trace_id,
            span_id=span_id,
            parent_span_id=parent_span_id,
            kind=kind,
            name=name,
            agent_id=agent_id,
            session_id=session_id,
            input=input_data or {},
        )
        self._active_spans[span_id] = span
        self._trace_tree.setdefault(trace_id, []).append(span_id)
        # 更新父 span 的 child 列表
        if parent_span_id in self._active_spans:
            self._active_spans[parent_span_id].child_spans.append(span_id)
        return span_id

    def finish_span(self, span_id: str, output_data: dict = None,
                     status: str = "ok", error: str = None):
        if span_id not in self._active_spans:
            return
        span = self._active_spans[span_id]
        span.finish(status)
        if output_data:
            span.output = output_data
        if error:
            span.error_message = error
        # 异步导出
        asyncio.create_task(self.exporter.export_span(span))

    def get_trace(self, trace_id: str) -> list[AgentSpan]:
        span_ids = self._trace_tree.get(trace_id, [])
        return [self._active_spans[sid] for sid in span_ids if sid in self._active_spans]

    @staticmethod
    def _new_span_id() -> str:
        return str(uuid.uuid4())[:16]  # 短 ID，便于展示

3.3 OpenTelemetry 集成

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

class OTelIntegration:
    """将 Agent Span 桥接到 OpenTelemetry"""
    def __init__(self, endpoint: str = "localhost:4317"):
        self.provider = TracerProvider()
        self.exporter = OTLPSpanExporter(endpoint=endpoint)
        self.processor = BatchSpanProcessor(self.exporter)
        self.provider.add_span_processor(self.processor)
        trace.set_tracer_provider(self.provider)
        self.tracer = trace.get_tracer("agent-system")

    def to_otel_span(self, agent_span: AgentSpan):
        """将 AgentSpan 转换为 OTel Span"""
        with self.tracer.start_as_current_span(
            agent_span.name,
            context=self._make_context(agent_span),
            kind=self._map_kind(agent_span.kind),
        ) as otel_span:
            # 设置属性
            otel_span.set_attribute("agent.id", agent_span.agent_id)
            otel_span.set_attribute("session.id", agent_span.session_id)
            otel_span.set_attribute("llm.model", agent_span.model or "")
            otel_span.set_attribute("cost.usd", agent_span.cost_usd)
            for k, v in agent_span.token_usage.items():
                otel_span.set_attribute(f"token.{k}", v)
            # 设置状态
            if agent_span.status == "error":
                otel_span.set_status(trace.Status(trace.StatusCode.ERROR, agent_span.error_message))
            # 添加事件
            for evt in agent_span.events:
                otel_span.add_event(evt.name, evt.attributes)

    def _map_kind(self, kind: SpanKind):
        mapping = {
            SpanKind.LLM_CALL: trace.SpanKind.CLIENT,
            SpanKind.TOOL_CALL: trace.SpanKind.INTERNAL,
            SpanKind.AGENT_STEP: trace.SpanKind.INTERNAL,
            SpanKind.RETRIEVAL: trace.SpanKind.CLIENT,
        }
        return mapping.get(kind, trace.SpanKind.INTERNAL)

4. 指标采集

4.1 关键指标定义

from prometheus_client import Counter, Histogram, Gauge, Summary
import time

class AgentMetrics:
    """Agent 系统关键指标"""
    def __init__(self, namespace: str = "agent"):
        # LLM 调用指标
        self.llm_requests = Counter(
            f"{namespace}_llm_requests_total",
            "Total LLM requests", ["model", "agent_id", "status"]
        )
        self.llm_latency = Histogram(
            f"{namespace}_llm_latency_seconds",
            "LLM request latency",
            ["model", "agent_id"],
            buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0]
        )
        self.llm_tokens = Counter(
            f"{namespace}_llm_tokens_total",
            "Total tokens consumed", ["model", "direction"]  # direction: input/output
        )
        self.llm_cost = Counter(
            f"{namespace}_llm_cost_usd_total",
            "Total LLM cost in USD", ["model", "agent_id"]
        )

        # Agent 行为指标
        self.agent_steps = Counter(
            f"{namespace}_agent_steps_total",
            "Total agent reasoning steps", ["agent_id", "step_type"]
        )
        self.agent_session_duration = Histogram(
            f"{namespace}_agent_session_duration_seconds",
            "Agent session duration", ["agent_id", "outcome"]  # outcome: success/failure/timeout
        )
        self.agent_tool_calls = Counter(
            f"{namespace}_agent_tool_calls_total",
            "Total tool calls", ["agent_id", "tool_name", "status"]
        )

        # 系统指标
        self.active_sessions = Gauge(
            f"{namespace}_active_sessions",
            "Current active sessions", ["agent_id"]
        )
        self.queue_depth = Gauge(
            f"{namespace}_queue_depth",
            "Current queue depth", ["queue_name"]
        )
        self.error_rate = Summary(
            f"{namespace}_error_rate",
            "Error rate over 5m window", ["agent_id", "error_type"]
        )

    def record_llm_call(self, model: str, agent_id: str, duration: float,
                        input_tokens: int, output_tokens: int, cost: float, status: str):
        self.llm_requests.labels(model=model, agent_id=agent_id, status=status).inc()
        self.llm_latency.labels(model=model, agent_id=agent_id).observe(duration)
        self.llm_tokens.labels(model=model, direction="input").inc(input_tokens)
        self.llm_tokens.labels(model=model, direction="output").inc(output_tokens)
        self.llm_cost.labels(model=model, agent_id=agent_id).inc(cost)

    def record_agent_step(self, agent_id: str, step_type: str):
        self.agent_steps.labels(agent_id=agent_id, step_type=step_type).inc()

    def record_tool_call(self, agent_id: str, tool_name: str, status: str):
        self.agent_tool_calls.labels(agent_id=agent_id, tool_name=tool_name, status=status).inc()

4.2 成本归因指标

from collections import defaultdict

class CostAttribution:
    """成本归因：按用户/会话/功能分解成本"""
    def __init__(self, redis_client):
        self.redis = redis_client

    async def record_cost(self, trace_id: str, user_id: str, feature: str,
                          model: str, cost: float, tokens: int):
        pipe = self.redis.pipeline()
        # 按用户聚合
        pipe.incrbyfloat(f"cost:user:{user_id}:total", cost)
        pipe.incrby(f"cost:user:{user_id}:tokens", tokens)
        # 按功能聚合
        pipe.incrbyfloat(f"cost:feature:{feature}:total", cost)
        # 按模型聚合
        pipe.incrbyfloat(f"cost:model:{model}:total", cost)
        # 按 trace 明细
        pipe.hset(f"cost:trace:{trace_id}", mapping={
            "user_id": user_id,
            "feature": feature,
            "model": model,
            "cost": str(cost),
            "tokens": str(tokens),
            "timestamp": str(time.time())
        })
        await pipe.execute()

    async def get_user_cost_report(self, user_id: str, days: int = 7) -> dict:
        return {
            "total_cost_usd": float(await self.redis.get(f"cost:user:{user_id}:total") or 0),
            "total_tokens": int(await self.redis.get(f"cost:user:{user_id}:tokens") or 0),
            "by_feature": await self._get_by_dimension(f"cost:feature", user_id, days),
            "by_model": await self._get_by_dimension(f"cost:model", user_id, days),
        }

5. 结构化日志

5.1 日志格式规范

import json
import logging
import sys
from datetime import datetime
from typing import Any

class StructuredLogger:
    """结构化日志：JSON 格式，便于 ELK/Loki 检索"""
    def __init__(self, service_name: str, level: str = "INFO"):
        self.service = service_name
        self.logger = logging.getLogger(service_name)
        self.logger.setLevel(getattr(logging, level))
        handler = logging.StreamHandler(sys.stdout)
        handler.setFormatter(self)
        self.logger.addHandler(handler)

    def format(self, record: logging.LogRecord) -> str:
        log_entry = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "service": self.service,
            "message": record.getMessage(),
            "trace_id": getattr(record, "trace_id", None),
            "span_id": getattr(record, "span_id", None),
            "session_id": getattr(record, "session_id", None),
            "agent_id": getattr(record, "agent_id", None),
            "user_id": getattr(record, "user_id", None),
        }
        # 添加额外字段
        if hasattr(record, "extra_fields"):
            log_entry.update(record.extra_fields)
        return json.dumps(log_entry, ensure_ascii=False)

    def bind(self, **kwargs) -> "BoundLogger":
        return BoundLogger(self, kwargs)

class BoundLogger:
    """绑定上下文的 Logger"""
    def __init__(self, logger: StructuredLogger, bound: dict):
        self._logger = logger
        self._bound = bound

    def _log(self, level: str, msg: str, **kwargs):
        extra = {**self._bound, **kwargs}
        record = self._logger.logger.makeRecord(
            self._logger.logger.name, getattr(logging, level),
            "(unknown)", 0, msg, [], None
        )
        record.extra_fields = extra
        for k, v in extra.items():
            setattr(record, k, v)
        self._logger.logger.handle(record)

    def info(self, msg: str, **kwargs): self._log("INFO", msg, **kwargs)
    def warning(self, msg: str, **kwargs): self._log("WARNING", msg, **kwargs)
    def error(self, msg: str, **kwargs): self._log("ERROR", msg, **kwargs)
    def debug(self, msg: str, **kwargs): self._log("DEBUG", msg, **kwargs)

5.2 日志事件类型

class AgentLogEvents:
    """Agent 关键生命周期事件"""
    @staticmethod
    def session_started(logger: BoundLogger, session_id: str, user_id: str):
        logger.info("agent.session.started",
                   session_id=session_id, user_id=user_id,
                   event_type="session_lifecycle")

    @staticmethod
    def llm_call_started(logger: BoundLogger, model: str, prompt_length: int):
        logger.info("llm.call.started",
                   model=model, prompt_length=prompt_length,
                   event_type="llm_call")

    @staticmethod
    def llm_call_completed(logger: BoundLogger, model: str, latency_ms: float,
                           input_tokens: int, output_tokens: int, cost: float):
        logger.info("llm.call.completed",
                   model=model, latency_ms=latency_ms,
                   input_tokens=input_tokens, output_tokens=output_tokens,
                   cost_usd=cost, event_type="llm_call")

    @staticmethod
    def tool_call_started(logger: BoundLogger, tool_name: str, args: dict):
        logger.info("tool.call.started",
                   tool_name=tool_name, tool_args=args,
                   event_type="tool_call")

    @staticmethod
    def tool_call_completed(logger: BoundLogger, tool_name: str,
                            success: bool, result_summary: str):
        logger.info("tool.call.completed",
                   tool_name=tool_name, success=success,
                   result_summary=result_summary[:200],
                   event_type="tool_call")

    @staticmethod
    def hallucination_detected(logger: BoundLogger, claim: str, confidence: float):
        logger.warning("agent.hallucination.detected",
                      claim=claim[:100], confidence=confidence,
                      event_type="quality")

    @staticmethod
    def session_completed(logger: BoundLogger, session_id: str,
                          total_steps: int, total_cost: float, outcome: str):
        logger.info("agent.session.completed",
                   session_id=session_id, total_steps=total_steps,
                   total_cost_usd=total_cost, outcome=outcome,
                   event_type="session_lifecycle")

6. 完整可观测性集成

class ObservableAgent:
    """集成可观测性的 Agent 基类"""
    def __init__(self, agent_id: str, tracer: AgentTracer,
                 metrics: AgentMetrics, logger: StructuredLogger):
        self.agent_id = agent_id
        self.tracer = tracer
        self.metrics = metrics
        self.logger = logger.bind(agent_id=agent_id)
        self._current_trace_id: str | None = None
        self._current_span_id: str | None = None

    async def run(self, user_query: str, user_id: str, session_id: str) -> str:
        # 开始追踪
        trace_id, root_span_id = self.tracer.start_trace(session_id, self.agent_id)
        self._current_trace_id = trace_id
        self._current_span_id = root_span_id

        # 绑定 trace 上下文到日志
        self.logger = self.logger.bind(trace_id=trace_id, session_id=session_id, user_id=user_id)

        session_start = time.time()
        try:
            self.logger.info("agent.session.started", query=user_query[:100])
            self.metrics.active_sessions.labels(agent_id=self.agent_id).inc()

            # 执行 Agent 逻辑（带追踪）
            result = await self._execute_with_tracing(user_query)

            duration = time.time() - session_start
            self.metrics.agent_session_duration.labels(
                agent_id=self.agent_id, outcome="success"
            ).observe(duration)
            self.logger.info("agent.session.completed",
                           duration_seconds=duration, outcome="success")
            self.tracer.finish_span(root_span_id, {"result": result}, "ok")
            return result

        except Exception as e:
            duration = time.time() - session_start
            self.metrics.agent_session_duration.labels(
                agent_id=self.agent_id, outcome="failure"
            ).observe(duration)
            self.logger.error("agent.session.failed", error=str(e), exc_info=True)
            self.tracer.finish_span(root_span_id, status="error", error=str(e))
            raise
        finally:
            self.metrics.active_sessions.labels(agent_id=self.agent_id).dec()

    async def _execute_with_tracing(self, query: str) -> str:
        # LLM 调用
        llm_span_id = self.tracer.start_span(
            self._current_trace_id, self._current_span_id,
            SpanKind.LLM_CALL, "llm_chat", self.agent_id,
            self._get_session_id(), {"query": query}
        )
        llm_start = time.time()
        try:
            self.logger.info("llm.call.started")
            response = await self._call_llm(query)
            latency = time.time() - llm_start
            self.metrics.record_llm_call(
                model="gpt-4o", agent_id=self.agent_id, duration=latency,
                input_tokens=100, output_tokens=50, cost=0.003, status="ok"
            )
            self.tracer.finish_span(llm_span_id, {"response": response}, "ok")
            return response
        except Exception as e:
            self.tracer.finish_span(llm_span_id, status="error", error=str(e))
            raise

    def _get_session_id(self) -> str:
        return self.logger._bound.get("session_id", "")

7. 可观测性数据流水线

Agent 运行时
    │
    ├── Traces ──→ OTel Collector ──→ Jaeger/Tempo ──→ Grafana
    │
    ├── Metrics ──→ Prometheus ──────────────────────→ Grafana
    │
    └── Logs ────→ Loki/ELK ───────────────────────→ Grafana
                      │
                      └──→ 告警规则 → AlertManager → Slack/PagerDuty

7.1 OpenTelemetry Collector 配置

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 5s
    send_batch_size: 100
  memory_limiter:
    check_interval: 5s
    limit_mib: 1000
  attributes:
    actions:
      - key: agent.id
        action: insert
        value: "unknown"
  resource:
    attributes:
      - key: deployment.environment
        value: "production"
        action: insert

exporters:
  otlp/jaeger:
    endpoint: jaeger:4317
    tls:
      insecure: true
  prometheus:
    endpoint: 0.0.0.0:8889
  loki:
    endpoint: http://loki:3100/loki/api/v1/push

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch, memory_limiter, attributes]
      exporters: [otlp/jaeger]
    metrics:
      receivers: [otlp]
      processors: [batch, memory_limiter]
      exporters: [prometheus]
    logs:
      receivers: [otlp]
      processors: [batch, memory_limiter]
      exporters: [loki]

8. 告警规则

# prometheus-alerts.yaml
groups:
  - name: agent_alerts
    rules:
      - alert: AgentHighErrorRate
        expr: |
          sum(rate(agent_agent_steps_total{status="error"}[5m])) by (agent_id)
          / sum(rate(agent_agent_steps_total[5m])) by (agent_id) > 0.05
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Agent {{ $labels.agent_id }} 错误率超过 5%"

      - alert: AgentHighLatency
        expr: |
          histogram_quantile(0.95,
            sum(rate(agent_llm_latency_seconds_bucket[5m])) by (le, agent_id)
          ) > 5
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Agent {{ $labels.agent_id }} P95 延迟超过 5 秒"

      - alert: AgentHighCost
        expr: |
          rate(agent_llm_cost_usd_total[1h]) > 10
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Agent 每小时成本超过 $10"

      - alert: AgentHallucinationDetected
        expr: |
          increase(agent_quality_hallucination_total[10m]) > 5
        labels:
          severity: warning
        annotations:
          summary: "检测到 {{ $value }} 次幻觉"

9. 总结

Agent 可观测性的核心在于全链路关联：

维度关键指标工具选型
追踪每次 LLM 调用、工具调用、推理步骤OpenTelemetry + Jaeger/Tempo
指标延迟、吞吐、错误率、成本Prometheus + Grafana
日志结构化 JSON，关联 trace_idLoki/ELK + Grafana
告警错误率、延迟、成本异常AlertManager + Slack

设计原则：

  1. 每个 Span 必须携带 trace_id + span_id + agent_id + session_id
  2. 每个 Log 行必须关联 trace_id
  3. 成本必须归因到用户/会话/功能
  4. 告警规则优先覆盖：错误率 > 延迟 > 成本

推荐技术栈：OpenTelemetry（统一采集）→ OTel Collector（处理与转发）→ Jaeger + Prometheus + Loki（存储）→ Grafana（统一展示）→ AlertManager（告警）。

加入讨论

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

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