Agent 可观测性：生产环境 AI Agent 的指标、追踪与实时告警

一、为什么 Agent 可观测性不同于传统服务监控

凌晨 3:14 的生产事故

凌晨 3:14，你的手机没有响。没有 PagerDuty，没有飞书告警，没有任何通知。第二天早上打开 Grafana，一切看起来都很正常——QPS 稳定，P99 延迟在预算内，错误率 0%。直到客服团队转来 7 条用户投诉，你才发现：昨晚的 Agent 在连续 3 个小时里，对每一笔退款请求都给出了「无法处理」的回复。

这不是传统意义上的「故障」。API 没有返回 500，数据库没有报连接超时，Pod 没有 OOM。Agent 的每一层基础设施都是健康的——只是 LLM 在一个特定 prompt 路径下做出了错误的工具选择，而你对此完全没有任何可见性。

这个场景揭示了一个根本性的问题：传统服务监控的三根支柱（Metrics · Tracing · Alerting）在 Agent 场景中需要被重新定义。

传统微服务监控 vs Agent 可观测性

在传统微服务架构中，一套 RED（Rate, Errors, Duration）指标基本覆盖了 90% 的监控需求：请求速率告诉你有多少流量，错误率告诉你服务是否健康，延迟告诉你用户体验如何。这套体系之所以有效，是因为微服务的执行路径是确定性的——同一个 HTTP 请求，经过同一个代码分支，产生同一个结果。

Agent 从根本上打破了这三个假设：

关键差异在第三行和第四行。传统服务中，延迟是网络和数据库的延迟——这些都是可预测、可优化的。Agent 的延迟中，LLM 推理时间可能占到 80%，而且同一个 prompt 两次调用的推理时间可以差 3 倍（取决于模型负载、token 数量、输出长度）。同理，传统服务的成本是固定的基础设施费用；Agent 的成本是 variable 的——每次请求消耗的 token 数量不同，直接影响到账单。

这就是为什么理解 AI Agent 的基本工作原理是可观测性的前提——如果你不知道 Agent 的推理循环中每一步在做什么，你就不可能知道该监控什么。

三根支柱：Metrics（发生了什么）· Tracing（怎么发生的）· Alerting（何时行动）

Agent 可观测性的三根支柱，对应三个不同的问题：

1. Metrics（指标）——回答「What」。当前生产环境中，Agent 的任务完成率是多少？P99 延迟是多少？每小时 token 消耗是否正常？工具调用成功率有没有下降趋势？Metrics 提供的是聚合后的态势感知。没有 metrics，你只能等用户投诉。

2. Tracing（追踪）——回答「How」。当指标显示 P99 延迟飙升时，你需要 Tracing 告诉你「是哪一步慢了」。一次 Agent 请求的内部链路可能是：LLM 推理（800ms）→ 工具调用 A（200ms）→ LLM 推理（1200ms）→ 工具调用 B（4500ms ← 这里慢了）→ LLM 推理（600ms）。没有 Tracing，你只知道总延迟 7300ms，但不知道 4500ms 花在了工具调用 B 上。正如下一节会详细展开的，Span 层级是 Agent Tracing 的核心。

3. Alerting（告警）——回答「When」。不是每个指标波动都值得叫醒 on-call。告警的挑战在 Agent 场景中尤为突出：传统服务的告警逻辑很简单（error_rate > 1%），但 Agent 的告警需要组合信号——仅 error_rate 高可能是一次 LLM 抖动，但 error_rate 高同时 token_cost 飙升同时 task_completion_rate 下降，说明真正出了问题。

这三根支柱的关系是：Metrics 是仪表盘，Tracing 是显微镜，Alerting 是哨兵。没有 Metrics，你不知道出事了；没有 Tracing，你不知道为什么出事；没有 Alerting，你出事了也没人知道。

为什么不能直接复用现有的监控体系

如果你已经有了 Prometheus + Grafana + ELK 的成熟监控体系，为什么不能直接套用到 Agent 上？三个结构性原因：

1. 缺少 LLM 维度的指标。你的现有 Prometheus 指标可能覆盖了 HTTP 请求数、数据库连接数、CPU/内存使用率——但没有任何一个指标告诉你 Agent 每小时消耗了多少 token、每次推理的 token 效率（完成任务消耗的 token 数）、或者 模型延迟 vs 工具延迟 的占比。这些指标对成本优化和性能调优至关重要，但在传统监控体系中不存在。

2. 缺少 Agent 特有的失败模式。幻觉（hallucination）不是异常。工具调用失败不是 HTTP 500。Agent 选择了一个不应该选的操作（如误删数据）——从 API 响应码来看是 200 成功，但从业务结果来看是灾难。传统监控的「错误」定义根本覆盖不到这些场景。

3. Trace 语义不通。传统分布式追踪的 Span 层级通常是：HTTP → Service → DB Query。Agent 的 Span 层级是：request → LLM reasoning → tool selection → tool execution → LLM reasoning（循环）。如果你把 Agent 的每一轮推理都当成一个平级的 Span 而不是嵌套的父子关系，Trace 的可视化将是一团乱麻——你无法区分「第 1 轮推理调了工具 A」和「第 3 轮推理调了工具 B」。

下面这张架构概览图展示了 Agent 可观测性的整体数据流——从应用代码到最终告警的完整链路：

┌─────────────────────────────────────────────────────────────────┐
│                        Agent 可观测性架构                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                   │
│  ┌──────────────┐    ┌──────────────────┐    ┌─────────────────┐ │
│  │  Agent Code   │───▶│ OpenTelemetry SDK │───▶│ OTLP Collector  │ │
│  │ (Python/TS)  │    │ (Span + Metrics)  │    │ (gRPC/HTTP)     │ │
│  └──────┬───────┘    └──────────────────┘    └───────┬─────────┘ │
│         │                                            │            │
│         │  ┌──────────────────┐          ┌───────────┴─────────┐ │
│         └─▶│ Prometheus       │          │                      │ │
│            │ Metrics Registry │          ▼                      ▼ │
│            └────────┬─────────┘   ┌───────────┐   ┌────────────┐ │
│                     │             │   Jaeger   │   │  Grafana    │ │
│                     ▼             │  / Tempo   │   │ (Dashboards)│ │
│            ┌───────────────┐      └───────────┘   └──────┬─────┘ │
│            │ /metrics      │                              │       │
│            │ (Prometheus)  │                     ┌────────▼─────┐ │
│            └───────────────┘                     │  Alertmanager │ │
│                                                   └──────┬───────┘ │
│                                                          │         │
│                                                  ┌───────▼───────┐ │
│                                                  │ Slack/PagerDuty│ │
│                                                  │ /Feishu        │ │
│                                                  └───────────────┘ │
└──────────────────────────────────────────────────────────────────┘

接下来的三个章节，我们将逐层构建这套体系。第 2 节定义「该监控什么」（信号分类），第 3 节实现「如何追踪」（OpenTelemetry），第 4 节构建「如何暴露和可视化」（Prometheus + Grafana）。后续章节（Part 2）将覆盖告警规则、Trace-Metric-Log 关联、以及渐进式落地路径。

二、Agent 可观测性信号分类：该监控哪些指标

定义「该监控什么」是可观测性体系中最重要的一步——如果你不知道要看什么，再好的工具也帮不了你。Agent 的信号不是简单的「加上几个 Prometheus counter 就完事」，而是需要从四个层次进行系统分类。每一层回答不同的运维问题，每一层对应不同的 Receiver 角色。

四层信号体系

我们将 Agent 的可观测性信号分为四个层次，从底层基础设施到顶层业务价值逐层递进：

四层之间的关系是递进依赖的：L1 健康不代表 L3 健康（Agent 可能在高 QPS 下持续给出错误答案），L3 健康不代表 L4 健康（Agent 能正确完成任务但用户觉得体验很差）。只有当四个层次的信号都亮绿灯时，你才能说「Agent 在生产环境中运行正常」。

以凌晨 3:14 那个事故为例：L1 指标一切正常（QPS 正常、延迟正常、error_count = 0），但 L3 层 tool_call_success_rate 虽然正常（工具调用都成功了），工具选择是错误的——LLM 没有选择「退款处理」工具，而是选择了「返回无法处理」的终止路径。这意味着你需要的不仅是「工具调用成功与否」，还需要「工具选择分布」——一个在传统监控体系中完全不存在的概念。

这也解释了为什么本系列的Agent 工具设计是可观测性的上游依赖——如果你在设计工具时没有定义清晰的工具分类（读操作 vs 写操作 vs 高风险操作），你就不可能在监控中区分「退款工具被调用了」和「退款工具被正确调用了」。

Python 信号定义

下面用 Python dataclass 和 Enum 将四层信号完整地定义出来。这个定义既是对信号分类的精确描述，也是你在实际代码中构建 metrics registry 的基础：

from __future__ import annotations

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


# ── L1: 标准信号（Standard Signals）───────────────────────────────────

@dataclass
class StandardSignals:
    """每个 HTTP 服务都需要的标准 RED 指标。Agent 服务也不例外。"""
    request_count: int = 0
    request_count_by_status: dict[str, int] = field(default_factory=dict)
    latency_p50_ms: float = 0.0
    latency_p95_ms: float = 0.0
    latency_p99_ms: float = 0.0
    error_count: int = 0


# ── L2: LLM 原生信号（LLM-Native Signals）─────────────────────────────

@dataclass
class LLMSignals:
    """LLM 特有的信号——传统微服务中不存在这些维度。"""
    tokens_consumed: int = 0
    tokens_per_step: list[float] = field(default_factory=list)
    model_latency_ms: float = 0.0
    input_tokens: int = 0
    output_tokens: int = 0
    tokens_per_dollar: float = 0.0


# ── L3: Agent 原生信号（Agent-Native Signals）──────────────────────────

class AgentStepOutcome(str, Enum):
    """Agent 推理单步的结果——比 L1 的 success/failure 更细粒度。"""
    COMPLETED = "completed"
    TOOL_CALL_FAILED = "tool_call_failed"
    HALLUCINATION_DETECTED = "hallucination"
    RECOVERY_ATTEMPTED = "recovery"
    MAX_STEPS_EXCEEDED = "max_steps"
    SAFETY_BLOCKED = "safety_blocked"

@dataclass
class AgentSignals:
    """Agent 特有的信号——衡量决策质量和可靠性。"""
    tool_call_success_rate: float = 0.0
    tool_call_count_per_run: list[int] = field(default_factory=list)
    reasoning_step_count: list[int] = field(default_factory=list)
    hallucination_detected: int = 0
    recovery_attempt_count: int = 0
    step_outcome_distribution: dict[str, int] = field(default_factory=dict)
    tool_selection_distribution: dict[str, int] = field(default_factory=dict)
    approval_trigger_count: int = 0
    approval_timeout_count: int = 0


# ── L4: 业务信号（Business Signals）───────────────────────────────────

@dataclass
class BusinessSignals:
    """业务层面的信号——回答「Agent 有没有产生价值」。"""
    task_completion_rate: float = 0.0
    user_feedback_score: float = 0.0
    user_feedback_count: int = 0
    cost_per_task: float = 0.0
    cost_per_1k_requests: float = 0.0
    abandonment_rate: float = 0.0


# ── 完整的 SignalBundle ────────────────────────────────────────────────

@dataclass
class AgentSignalBundle:
    """将所有四层信号聚合为一个结构体，作为 metrics registry 的数据源。"""
    standard: StandardSignals = field(default_factory=StandardSignals)
    llm: LLMSignals = field(default_factory=LLMSignals)
    agent: AgentSignals = field(default_factory=AgentSignals)
    business: BusinessSignals = field(default_factory=BusinessSignals)

    def snapshot(self) -> dict:
        """返回所有信号的扁平化字典，适合暴露给 /metrics endpoint。"""
        return {
            "agent_requests_total": self.standard.request_count,
            "agent_errors_total": self.standard.error_count,
            "agent_latency_p99_ms": self.standard.latency_p99_ms,
            "agent_tokens_consumed_total": self.llm.tokens_consumed,
            "agent_model_latency_ms": self.llm.model_latency_ms,
            "agent_tool_call_success_rate": self.agent.tool_call_success_rate,
            "agent_hallucination_total": self.agent.hallucination_detected,
            "agent_recovery_attempts_total": self.agent.recovery_attempt_count,
            "agent_task_completion_rate": self.business.task_completion_rate,
            "agent_user_feedback_score": self.business.user_feedback_score,
            "agent_cost_per_task_usd": self.business.cost_per_task,
        }

这个信号定义有几个值得注意的设计决策：

1. 用 dataclass 而非 dict。强类型定义意味着 IDE 自动补全、类型检查、以及后续 Prometheus client 集成时的字段校验。在实际项目中，你可以直接将这些 dataclass 作为内部状态存储，然后通过一个 snapshot() 方法导出到 Prometheus Counter/Gauge/Histogram。

2. AgentStepOutcome 是一个关键枚举。它在 L1 的 success/failure 之上增加了一层 Agent 特有的语义——工具调用失败和幻觉检测是两种完全不同的失败模式，需要不同的响应策略。工具调用失败可能是网络问题（retry 即可），幻觉可能是 prompt 设计问题（需要回滚 prompt 版本）。如果把它们都归为「error」，你就失去了精细化告警的能力。

3. tool_selection_distribution 是「静默失败」的探测器。回到凌晨 3:14 的事故——如果你监控了工具选择分布，你会立即发现「退款处理」工具的调用量降到了 0，而「返回无法处理」的终止路径调用量飙升。这个信号比任何 error counter 都更早地暴露了问题。这与Agent 评测框架中的在线评测指标直接对应——评测框架告诉你 Agent 的答案质量，信号体系告诉你 Agent 的行为模式。

4. cost_per_task 是连接技术和业务的桥梁。token 消耗是一个纯技术指标——工程师关心。但 product manager 关心的是「每个客服对话花了多少钱」。将 token 消耗 × 模型定价 ÷ 任务数 = cost_per_task，把技术指标翻译成业务语言，这才是让可观测性数据走出工程团队的钥匙。

在下一节，我们将看到这些信号如何被嵌入到 OpenTelemetry 的 Span 体系中，实现从「定义信号」到「采集信号」的跨越。

三、用 OpenTelemetry 为 Agent 添加分布式追踪

如果 Metrics 是仪表盘上的指针，Tracing 就是引擎盖下的高速摄像机——它能逐帧还原 Agent 在一次请求中的完整行为链路。对于 Agent 这种多步推理、多工具调用的非线性系统，Tracing 几乎是唯一能让你理解「Agent 在内部到底做了什么」的方式。

Span 层级设计

Agent 的 OpenTelemetry Span 层级遵循一个自然的嵌套结构——每次用户请求是一个 Root Span，Root Span 内部包含多次 LLM 调用 Span，每次 LLM 调用 Span 内部又包含多次工具调用 Span。层级关系如下：

root_span (agent.request)
├── llm_call_span (agent.llm.reasoning)     ← 第 1 轮推理
│   ├── tool_call_span (agent.tool.execute) ← 工具调用 A
│   └── tool_call_span (agent.tool.execute) ← 工具调用 B（可选：同一轮多工具）
├── llm_call_span (agent.llm.reasoning)     ← 第 2 轮推理
│   └── tool_call_span (agent.tool.execute) ← 工具调用 C
└── llm_call_span (agent.llm.reasoning)     ← 最终推理（产生最终回复）

这个层级设计有几个关键点：

1. 每一轮 LLM 推理是一个独立的 Span。一个典型的 Agent 请求可能经历 3~8 轮推理（ReAct / tool-use 循环），每一轮都应该是一个独立的 Span——这样你才能在 Tracing UI 中看到「第 2 轮推理慢了」而不是「整个请求慢了」。同时，reasoning_step_count（第 2 节定义的 L3 信号）可以直接从 Span 数量计算。

2. 工具调用 Span 是 LLM Span 的 Child。这是「因果关系」的正确表达——工具调用是 LLM 推理的结果，没有 LLM 的 tool_choice 就没有工具调用。将 tool_call_span 挂为 llm_call_span 的子 Span，意味着当你发现工具调用慢时，你可以立刻向上追溯：「这次工具调用是第几轮推理触发的？」

3. Root Span 代表一次完整的用户请求。它的 duration 就是端到端延迟——这是 P50/P95/P99 延迟指标的来源。Root Span 的 status 同时指示了请求的整体结果（OK 或 ERROR）。

Span Attributes（属性）

OpenTelemetry Span 通过 key-value attributes 携带上下文信息。对于 Agent 场景，以下是关键的 Span attribute 设计：

这些 attribute 的设计原则是：足够精确定位问题，但不过度——不要把完整的 LLM prompt 和工具参数放进 Span attribute（这些属于审计日志的范畴，见Agent 审计日志设计）。Span attribute 用于「搜索和过滤」，审计日志用于「完整回放」。两者的配合是：在 Tracing UI 中通过 agent.tool.name=refund_processor 过滤出所有退款相关的请求 → 拿到 trace_id → 在审计日志中展开完整的决策链和参数。

完整代码实现

下面是一段可运行的 Python Agent 追踪代码，使用 opentelemetry-api 和 opentelemetry-sdk，通过 OTLP exporter 将 Span 发送到 Jaeger 或 Grafana Tempo：

"""
Agent 可观测性 — OpenTelemetry 分布式追踪完整示例
依赖: pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp

启动本地 Jaeger（开发环境）:
    docker run -d --name jaeger \
      -p 16686:16686 \
      -p 4318:4318 \
      jaegertracing/all-in-one:latest
然后访问 http://localhost:16686 查看 Trace。
"""

import time
import uuid
from contextlib import contextmanager
from typing import Any, Optional

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import SpanKind, Status, StatusCode


# ── 1. 初始化 TracerProvider ───────────────────────────────────────

resource = Resource.create({
    SERVICE_NAME: "agent-service-prod",
    "deployment.environment": "production",
})

provider = TracerProvider(resource=resource)

# OTLP HTTP exporter → local Jaeger (port 4318)
# 生产环境替换为你的 OTel Collector 地址
otlp_exporter = OTLPSpanExporter(
    endpoint="http://localhost:4318/v1/traces",
)

provider.add_span_processor(BatchSpanProcessor(otlp_exporter))
trace.set_tracer_provider(provider)

tracer = trace.get_tracer(__name__)


# ── 2. OpenTelemetry 上下文管理 ─────────────────────────────────────

@contextmanager
def traced_span(
    name: str,
    kind: SpanKind = SpanKind.INTERNAL,
    attributes: Optional[dict[str, Any]] = None,
    parent: Optional[trace.Span] = None,
):
    """创建一个 traced span 作为上下文管理器。

    用法:
        with traced_span("agent.llm.reasoning",
                         attributes={"agent.model.name": "gpt-4o"}) as span:
            span.set_attribute("agent.llm.tokens.input", 1520)
    """
    ctx = trace.set_span_in_context(parent) if parent else None
    span = tracer.start_span(name, kind=kind, attributes=attributes, context=ctx)
    try:
        yield span
    except Exception as e:
        span.set_status(Status(StatusCode.ERROR, str(e)))
        span.record_exception(e)
        raise
    finally:
        span.end()


# ── 3. 模拟 Agent 服务 ──────────────────────────────────────────────

def simulate_llm_call(model: str, step: int, input_tokens: int) -> tuple[int, float]:
    """模拟一次 LLM 推理调用，返回 (output_tokens, latency_ms)。"""
    output_tokens = int(input_tokens * 0.6)
    latency_ms = input_tokens * 0.3 + 200
    time.sleep(latency_ms / 1000 * 0.01)     # 模拟加速（真实代码去掉这行）
    return output_tokens, latency_ms


def simulate_tool_call(tool_name: str) -> tuple[float, bool]:
    """模拟一次工具调用，返回 (latency_ms, success)。"""
    latency_ms = 50 + hash(tool_name) % 200
    time.sleep(latency_ms / 1000 * 0.01)
    success = True
    return latency_ms, success


def agent_run(request_id: str, user_query: str) -> str:
    """一次完整的 Agent 请求——模拟 ReAct 循环。"""
    total_steps = 3
    tools_to_call = ["search_knowledge_base", "calculate", "format_response"]

    # ── Root Span：代表整个用户请求 ──
    with traced_span(
        "agent.request",
        kind=SpanKind.SERVER,
        attributes={
            "agent.request.id": request_id,
            "agent.agent.id": "prod-agent-03",
        },
    ) as root_span:

        final_reply: str = ""

        for step in range(1, total_steps + 1):
            model_name = "gpt-4o"
            input_tokens = 800 + step * 200

            # ── LLM Span：第 N 轮推理 ──
            with traced_span(
                "agent.llm.reasoning",
                kind=SpanKind.INTERNAL,
                attributes={
                    "agent.model.name": model_name,
                    "agent.model.temperature": 0.3,
                    "agent.reasoning.step": step,
                },
                parent=root_span,
            ) as llm_span:

                output_tokens, llm_latency = simulate_llm_call(
                    model_name, step, input_tokens
                )

                llm_span.set_attributes({
                    "agent.llm.tokens.input": input_tokens,
                    "agent.llm.tokens.output": output_tokens,
                    "agent.llm.duration_ms": int(llm_latency),
                })

                # ── Tool Call Span：工具执行 ──
                if step <= len(tools_to_call):
                    tool_name = tools_to_call[step - 1]
                    with traced_span(
                        "agent.tool.execute",
                        kind=SpanKind.CLIENT,
                        attributes={"agent.tool.name": tool_name},
                        parent=llm_span,
                    ) as tool_span:

                        tool_latency, tool_success = simulate_tool_call(tool_name)
                        tool_span.set_attributes({
                            "agent.tool.duration_ms": int(tool_latency),
                            "agent.tool.status": (
                                "success" if tool_success else "failure"
                            ),
                        })
                        if not tool_success:
                            tool_span.set_status(
                                Status(StatusCode.ERROR, f"{tool_name} failed")
                            )

                final_reply = (
                    "任务完成" if step == total_steps else "继续推理..."
                )

        root_span.set_attribute("agent.run.total_steps", total_steps)
        root_span.set_status(Status(StatusCode.OK))

    return final_reply


# ── 4. 演示入口 ─────────────────────────────────────────────────────

if __name__ == "__main__":
    for i in range(3):
        req_id = str(uuid.uuid4())
        print(f"[请求 {req_id[:8]}...] 开始...")
        reply = agent_run(req_id, f"用户查询 #{i+1}: 分析最近的销售数据")
        print(f"[请求 {req_id[:8]}...] 完成 → {reply}\n")

    provider.shutdown()
    print("✅ 所有 Span 已发送到 OTLP Collector (Jaeger: http://localhost:16686)")

代码解读

这段代码的核心是三个嵌套的 traced_span 上下文管理器——它们精确地建模了 Agent 的 Span 层级：

1. traced_span 上下文管理器。封装了 span 的创建、异常处理和关闭逻辑。在 __exit__ 时自动调用 span.end()，避免了手动管理 span 生命周期的痛苦。生产环境中，你可以将这个上下文管理器提取为一个独立的 decorator 或 middleware。

2. 属性设置时机。注意 LLM Span 的属性（token 数、延迟）是在 LLM 调用返回之后设置的——因为这些数据只有在 API 响应中才有。而在 Trace UI 中，你可以看到 span 的完整属性集，这使你能够按 agent.llm.tokens.input > 2000 过滤出「大 prompt」的请求。

3. BatchSpanProcessor。Span 不是实时发送的——它们在内存中批量堆积，每隔一定时间或达到一定数量后通过 OTLP exporter 发送。这减少了网络开销，但也意味着如果你的进程在 batch 刷新前崩溃，最近的 Span 会丢失。对于 Agent 场景，建议设置较小的 batch 间隔（如 1 秒）以减少数据丢失风险。

4. 与审计日志的协作。注意到这段代码只记录了 Span attribute（轻量级的 key-value），而不是完整的 LLM prompt 和工具参数。这是因为 OpenTelemetry 的 Span attribute 有大小限制（通常不超过几 KB），且 OTel 后端和导出器对 attribute 和 event 大小也有限制。完整的数据（prompt、工具参数、工具返回值、审批记录）应该通过审计日志管道单独持久化——这正是Agent 审计日志设计中详细讨论的内容。Span 用于搜索和定位，审计日志用于完整回放——两者互补，而非替代。

5. OTLP Exporter 配置。上述代码使用的是 HTTP exporter（端口 4318），适合开发环境。生产环境中，你应该配置 OTLP gRPC exporter（端口 4317）并指向你的 OTel Collector 或直接指向 Grafana Tempo / Jaeger。关键参数包括 endpoint、headers（认证 token）、以及 TLS 配置。

有了 Trace，你就有了「显微镜」。但显微镜需要和「仪表盘」配合——下一节将构建 Agent 的指标管道，将第 2 节定义的四层信号暴露给 Prometheus 和 Grafana。

四、构建 Agent 指标管道：Prometheus + 自定义 Exporter

Tracing 给了你显微镜——你能看到每一次请求的内部链路。但显微镜无法告诉你「过去 5 分钟，Agent 的整体健康状况如何」。你需要 Metrics——聚合后的、时序化的、可查询的数值指标。本节构建 Agent 的指标管道：从应用代码中的 metrics registry，到 Prometheus 的 /metrics 端点，再到 Grafana 的可视化面板。

架构概览

Agent 指标管道的数据流是单向的、分层的：

Agent 代码（Python）
    │
    │  Counter.inc() / Gauge.set() / Histogram.observe()
    ▼
Prometheus Client Registry（内存中）
    │
    │  HTTP GET /metrics（Prometheus text format）
    ▼
Prometheus Server（scrape，每 15s 拉取一次）
    │
    │  时序数据库（TSDB）+ PromQL 查询
    ▼
Grafana（可视化 + 告警面板）
    │
    │  告警规则触发
    ▼
Alertmanager → Slack / PagerDuty / 飞书

在这个架构中，Agent 代码只做一件事：在正确的时机更新正确的指标。Prometheus server 负责定期拉取和存储，Grafana 负责可视化和告警触发。

指标类型：Counter / Gauge / Histogram

Prometheus 提供了四种核心指标类型，Agent 场景中使用其中三种：

一个常见的误区是用 Gauge 来计算成功率（手动做 success/total × 100）。正确做法是使用两个 Counter——agent_tool_calls_total 和 agent_tool_calls_failed_total——然后在 PromQL 中计算：

# PromQL: Agent 工具调用成功率（5 分钟窗口）
rate(agent_tool_calls_total[5m]) - rate(agent_tool_calls_failed_total[5m])
  /
rate(agent_tool_calls_total[5m])

这样做的优势是：Counter 不会在服务重启时丢失语义（从 0 开始，rate() 函数计算瞬时速率），而 Gauge 在重启后会跳变，导致告警误报。

Python 实现：Prometheus Client + Decorator 模式

下面是一段完整的 Agent 指标采集代码，使用 prometheus_client 库，通过 decorator 模式将指标采集逻辑与业务逻辑解耦：

"""
Agent 可观测性 — Prometheus 指标管道完整示例
依赖: pip install prometheus_client
启动后访问 http://localhost:9091/metrics 查看指标。
"""

import time
import functools
import threading
from typing import Callable, Any

from prometheus_client import (
    Counter, Gauge, Histogram, generate_latest,
    CollectorRegistry, REGISTRY,
)


# ── 1. 创建自定义 Registry（隔离 Agent 指标）─────────────────────────

agent_registry = CollectorRegistry()

# ── 2. 定义 Agent 指标 ───────────────────────────────────────────────

# L1 · 标准信号
agent_requests_total = Counter(
    "agent_requests_total", "Agent 请求总数",
    labelnames=["agent_id", "status"],
    registry=agent_registry,
)

agent_request_duration_seconds = Histogram(
    "agent_request_duration_seconds", "Agent 请求延迟（秒）",
    labelnames=["agent_id"],
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 60.0, 120.0],
    registry=agent_registry,
)

# L2 · LLM 原生信号
agent_tokens_consumed_total = Counter(
    "agent_tokens_consumed_total", "Agent 累计 token 消耗",
    labelnames=["agent_id", "model", "direction"],
    registry=agent_registry,
)

agent_model_latency_seconds = Histogram(
    "agent_model_latency_seconds", "LLM 模型推理延迟（秒）",
    labelnames=["agent_id", "model"],
    buckets=[0.1, 0.3, 0.5, 1.0, 2.0, 5.0, 10.0],
    registry=agent_registry,
)

# L3 · Agent 原生信号
agent_tool_calls_total = Counter(
    "agent_tool_calls_total", "工具调用总数",
    labelnames=["agent_id", "tool_name", "status"],
    registry=agent_registry,
)

agent_hallucination_total = Counter(
    "agent_hallucination_total", "检测到的幻觉总数",
    labelnames=["agent_id", "hallucination_type"],
    registry=agent_registry,
)

agent_reasoning_steps = Histogram(
    "agent_reasoning_steps", "每次请求的 LLM 推理步数",
    labelnames=["agent_id"],
    buckets=[1, 2, 3, 5, 8, 12, 20],
    registry=agent_registry,
)

agent_recovery_attempts_total = Counter(
    "agent_recovery_attempts_total", "错误恢复尝试总数",
    labelnames=["agent_id", "recovery_strategy"],
    registry=agent_registry,
)

# L4 · 业务信号
agent_task_completion_rate = Gauge(
    "agent_task_completion_rate", "任务完成率 (0.0 ~ 1.0)",
    labelnames=["agent_id", "task_type"],
    registry=agent_registry,
)

agent_cost_per_task_usd = Gauge(
    "agent_cost_per_task_usd", "单次任务平均成本（美元）",
    labelnames=["agent_id", "model"],
    registry=agent_registry,
)

# ── 3. Decorator: 自动采集指标 ───────────────────────────────────────

def observe_agent_request(agent_id: str = "prod-agent-03"):
    """Decorator: 自动为被装饰函数采集 Agent 请求指标。"""
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            start = time.time()
            status = "success"
            try:
                result = func(*args, **kwargs)
                return result
            except Exception:
                status = "failure"
                raise
            finally:
                duration = time.time() - start
                agent_requests_total.labels(
                    agent_id=agent_id, status=status
                ).inc()
                agent_request_duration_seconds.labels(
                    agent_id=agent_id
                ).observe(duration)
        return wrapper
    return decorator


def emit_tool_call_metric(
    agent_id: str, tool_name: str, status: str, duration_ms: float
):
    """在每次工具调用后调用此函数来更新指标。"""
    agent_tool_calls_total.labels(
        agent_id=agent_id, tool_name=tool_name, status=status
    ).inc()


def emit_llm_metric(
    agent_id: str, model: str, input_tokens: int,
    output_tokens: int, latency_seconds: float,
):
    """在每次 LLM 调用后调用此函数来更新指标。"""
    agent_tokens_consumed_total.labels(
        agent_id=agent_id, model=model, direction="input"
    ).inc(input_tokens)
    agent_tokens_consumed_total.labels(
        agent_id=agent_id, model=model, direction="output"
    ).inc(output_tokens)
    agent_model_latency_seconds.labels(
        agent_id=agent_id, model=model
    ).observe(latency_seconds)


# ── 4. 示例 Agent 请求处理 ───────────────────────────────────────────

@observe_agent_request(agent_id="prod-agent-03")
def handle_customer_query(query: str) -> str:
    """模拟 Agent 处理客户查询。"""
    agent_id = "prod-agent-03"
    model = "gpt-4o"

    emit_llm_metric(agent_id, model,
                    input_tokens=1200, output_tokens=350,
                    latency_seconds=0.8)
    emit_tool_call_metric(agent_id, "search_knowledge_base",
                          "success", duration_ms=180)

    emit_llm_metric(agent_id, model,
                    input_tokens=1800, output_tokens=400,
                    latency_seconds=1.1)
    emit_tool_call_metric(agent_id, "calculate",
                          "success", duration_ms=95)

    emit_llm_metric(agent_id, model,
                    input_tokens=1500, output_tokens=200,
                    latency_seconds=0.6)

    agent_task_completion_rate.labels(
        agent_id=agent_id, task_type="customer_query"
    ).set(0.92)

    return "已为您查询到最近的销售数据：Q2 营收同比增长 12.4%。"


# ── 5. 启动 Metrics HTTP Server ───────────────────────────────────────

def start_metrics_server(port: int = 9091) -> threading.Thread:
    """启动独立的 /metrics HTTP 端点（供 Prometheus scrape）。"""
    from http.server import HTTPServer, BaseHTTPRequestHandler

    class MetricsHandler(BaseHTTPRequestHandler):
        def do_GET(self):
            if self.path == "/metrics":
                self.send_response(200)
                self.send_header(
                    "Content-Type", "text/plain; charset=utf-8"
                )
                self.end_headers()
                self.wfile.write(generate_latest(agent_registry))
            elif self.path == "/health":
                self.send_response(200)
                self.send_header("Content-Type", "text/plain")
                self.end_headers()
                self.wfile.write(b"ok")
            else:
                self.send_response(404)
                self.end_headers()

    server = HTTPServer(("0.0.0.0", port), MetricsHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    print(
        f"📊 Agent metrics server on http://0.0.0.0:{port}/metrics"
    )
    return thread


if __name__ == "__main__":
    start_metrics_server(port=9091)

    print("模拟 Agent 请求...")
    for i in range(5):
        query = f"用户查询 #{i+1}: 帮我查一下最新数据"
        reply = handle_customer_query(query)
        print(f"  [{i+1}/5] 完成 → {reply}")

    print("\n✅ 指标已就绪。访问 http://localhost:9091/metrics 查看。")
    import signal
    signal.pause()

代码设计要点

1. 自定义 Registry 隔离 Agent 指标。如果你的服务同时有 Agent 组件和非 Agent 组件（如一个 FastAPI 服务中既有 REST API 也有 Agent 处理器），使用自定义 Registry 可以将 Agent 指标与应用指标分离，避免命名冲突和语义混淆。

2. Label 设计是门艺术。每个指标的 label 决定了你在 PromQL 中能如何切分和聚合。以上面的 agent_tool_calls_total 为例，选择了 [agent_id, tool_name, status] 三个 label——这使你能够按工具名称聚合（sum by(tool_name)）、按状态过滤（status="failure"）、或按 agent 实例对比（sum by(agent_id)）。但要避免高基数 label（如 user_id）——每个唯一的 label 组合都会在 Prometheus 中创建一个新的时间序列，过多的序列会导致内存膨胀。user_id 级别的指标更适合用 Tracing 和审计日志来追踪。

3. Histogram buckets 需要根据实际数据分布调整。Agent 的延迟分布通常比传统 API 宽得多——P50 可能 3 秒，P99 可能 60 秒。上面的 agent_request_duration_seconds 使用了 [0.5, 1.0, ..., 120.0] 的 buckets，覆盖了从快速响应到长任务的范围。如果你的 Agent 经常有 3 分钟以上的任务，需要增加 180.0 和 300.0 的 bucket。

4. Decorator 模式降低侵入性。observe_agent_request decorator 使指标采集对业务代码几乎透明——你只需要在函数上加一行 decorator，请求计数和延迟就被自动采集了。对于 LLM 调用和工具调用，使用显式的 emit_llm_metric() 和 emit_tool_call_metric() 函数，因为在 Agent 的推理循环中，这些调用的上下文（model 名称、tool 名称、token 数）因步而异，难以用 decorator 统一处理。错误恢复重试次数也是一个关键的观测信号——Agent 错误恢复中详细讨论了 retry 策略。

Prometheus 抓取配置

为了让 Prometheus 定期从 Agent 服务拉取指标，需要在 prometheus.yml 中添加一个 scrape job：

# prometheus.yml — Agent metrics scrape config

global:
  scrape_interval: 15s
  evaluation_interval: 15s

scrape_configs:
  - job_name: "agent-service"
    metrics_path: "/metrics"
    scrape_interval: 15s
    scrape_timeout: 10s
    static_configs:
      - targets:
          - "agent-prod-01:9091"
          - "agent-prod-02:9091"
          - "agent-prod-03:9091"
        labels:
          cluster: "production"
          service: "agent-service"

  - job_name: "agent-service-staging"
    metrics_path: "/metrics"
    scrape_interval: 60s
    static_configs:
      - targets:
          - "agent-staging:9091"
        labels:
          cluster: "staging"
          service: "agent-service"

关键配置说明：

• scrape_interval: 15s — 生产环境的推荐值。对于 Agent 这种延迟敏感的负载，15 秒足以捕捉到 P99 延迟的尖峰（尖峰通常持续 30~60 秒）。如果你的 Agent QPS 很高（> 100），可以考虑 10 秒间隔。
• scrape_timeout: 10s — 超时时间应小于 scrape_interval，避免一次慢抓取阻塞下一次。
• static_configs — 适合固定 IP 的部署。如果你的 Agent 在 Kubernetes 中运行，使用 kubernetes_sd_configs 进行服务发现，自动追踪 Pod 的创建和销毁。
• cluster 和 service 的 external labels — 在 Grafana 中按环境切换面板、在多集群部署中聚合指标时非常有用。

启动 Prometheus 后，你可以在 Prometheus UI（http://localhost:9090）中执行 PromQL 查询来验证指标是否正确采集：

# 验证：查看 Agent 请求速率（过去 5 分钟）
rate(agent_requests_total{cluster="production"}[5m])

# 验证：查看各工具调用失败率
rate(agent_tool_calls_total{status="failure"}[5m])
  /
rate(agent_tool_calls_total[5m])

# 验证：P99 延迟
histogram_quantile(0.99,
  rate(agent_request_duration_seconds_bucket[5m]))

至此，Agent 的指标管道已经建立完成：Agent 代码通过 prometheus_client 更新 Counter/Gauge/Histogram → Prometheus 定期 scrape /metrics 端点 → 指标存储在 TSDB 中，随时可以通过 PromQL 查询。下一阶段（第五、六节，将在 Part 2 中覆盖）将在此基础上构建告警规则和 Trace-Metric-Log 的关联体系。

值得强调的是，可观测性基础设施本身是需要被观测的。正如MCP 协议生产实践中所讨论的——MCP 调用和普通工具调用一样需要 Tracing 覆盖。在你的 Agent 体系中，MCP 工具的每一次调用也应该产生 agent.tool.execute Span 和对应的 Prometheus 指标，确保你的可观测性覆盖没有盲区。

五、Agent 实时告警：定义触发条件、阈值与分级响应

有了 Metrics（知道发生了什么）和 Tracing（知道怎么发生的），接下来需要一个「哨兵」——在问题演变成事故之前，通知正确的人采取行动。这就是 Alerting 的角色。但 Agent 的告警远比传统服务复杂：你不仅要回答「有没有异常」，还要回答「这个异常有多严重」和「它是真实问题还是 LLM 抖动」。

Agent 告警的五个关键信号

传统服务的告警通常只关注一个信号——error_rate > threshold → 告警。Agent 需要监控五个维度的信号，每个维度都有独特的失效模式：

这五个信号不能各自独立告警——否则你会收到大量噪音。正确的做法是组合告警：单个信号异常可能是噪声，两个以上信号同时异常才是真正的故障。例如：error_rate 高 + token_cost 飙升 + task_completion_rate 下降 = 真正的生产故障（可能是 LLM 陷入了工具选择循环）。仅 error_rate 高 = 可能是一次短暂的模型抖动，观察 5 分钟再决定。

告警分级：P1 / P2 / P3

不是每个告警都值得在凌晨 3 点叫醒 on-call。Agent 告警需要明确的分级体系：

分级的关键原则是：P1 只留给「用户正在受影响」的场景。如果你把每个 P99 抖动都设为 P1，你的 on-call 团队会迅速产生告警疲劳（alert fatigue），最终忽略所有告警——包括真正致命的那个。

Prometheus Alertmanager 告警规则

下面是一组可直接使用的 Prometheus 告警规则 YAML，覆盖了上述五个关键信号和三个告警级别：

# prometheus-alerting-rules.yml
# Agent 可观测性 — Prometheus 告警规则

groups:
  - name: agent-critical-alerts
    rules:
      # ── P1: Agent 错误率飙升（组合条件）─────────────────
      - alert: AgentHighErrorRate
        expr: |
          (
            rate(agent_tool_calls_total{status="failure"}[5m])
            /
            rate(agent_tool_calls_total[5m])
          ) > 0.05
          and
          rate(agent_task_completion_rate[5m]) < 0.5
        for: 3m
        labels:
          severity: critical
          team: agent-platform
        annotations:
          summary: "Agent 错误率飙升且任务完成率下降"
          description: >
            Agent {{ $labels.agent_id }} 的 5 分钟工具调用失败率
            为 {{ $value | humanizePercentage }}，
            同时任务完成率低于 50%。P1 级别——请立即检查。

      # ── P1: Agent P99 延迟飙升 ──────────────────────────
      - alert: AgentP99LatencySpike
        expr: |
          histogram_quantile(0.99,
            rate(agent_request_duration_seconds_bucket[5m])
          ) > 3 * (
            histogram_quantile(0.99,
              rate(agent_request_duration_seconds_bucket[1h] offset 1h)
            )
          )
        for: 5m
        labels:
          severity: critical
          team: agent-platform
        annotations:
          summary: "Agent P99 延迟飙升至基线的 3 倍以上"
          description: >
            Agent {{ $labels.agent_id }} 的 P99 延迟当前为
            {{ $value | humanizeDuration }}，超过 1 小时前基线的 3 倍。

  - name: agent-warning-alerts
    rules:
      # ── P2: Token 日消耗超出日均 ─────────────────────────
      - alert: AgentTokenCostSpike
        expr: |
          sum(increase(agent_tokens_consumed_total[1d]))
          >
          1.5 * avg(
            sum(increase(agent_tokens_consumed_total[1d] offset 1d))
          )
        for: 1h
        labels:
          severity: warning
          team: agent-platform
        annotations:
          summary: "Agent Token 消耗超过日均 1.5 倍"
          description: >
            过去 24 小时的 Token 总消耗已超过前一日均值的 1.5 倍。
            请检查是否有推理步数异常增加或 prompt 膨胀。

      # ── P2: 工具调用失败率异常 ──────────────────────────
      - alert: AgentToolFailureSpike
        expr: |
          rate(agent_tool_calls_total{status="failure"}[5m])
          /
          rate(agent_tool_calls_total[5m])
          > 0.03
        for: 5m
        labels:
          severity: warning
          team: agent-platform
        annotations:
          summary: "工具 {{ $labels.tool_name }} 调用失败率超过 3%"
          description: >
            工具 {{ $labels.tool_name }} 在 Agent {{ $labels.agent_id }}
            上的 5 分钟失败率为 {{ $value | humanizePercentage }}。

  - name: agent-info-alerts
    rules:
      # ── P3: 幻觉检测异常 ────────────────────────────────
      - alert: AgentHallucinationDetected
        expr: |
          increase(agent_hallucination_total[1h]) > 5
        for: 10m
        labels:
          severity: info
          team: agent-platform
        annotations:
          summary: "Agent 检测到幻觉 (1h 内 > 5 次)"
          description: >
            Agent {{ $labels.agent_id }} 在过去 1 小时内检测到
            {{ $value }} 次幻觉。请检查最近的 prompt 变更或模型版本。

Python Webhook 通知处理器

告警规则触发后，Alertmanager 通过 webhook 将告警推送到你的通知服务。下面是一个 Python Webhook 处理器，支持 Slack 和飞书双通道：

"""
Agent 告警 Webhook 处理器 — 接收 Alertmanager 告警并推送到 Slack / 飞书
依赖: pip install flask requests
"""

import json
import requests
from flask import Flask, request, jsonify

app = Flask(__name__)

# ── 配置 ──────────────────────────────────────────────────
SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/xxx"
FEISHU_WEBHOOK_URL = "https://open.feishu.cn/open-apis/bot/v2/hook/xxx"


def format_alert(alert: dict) -> dict:
    """将 Alertmanager 告警格式化为 Slack Block Kit 消息。"""
    severity = alert["labels"].get("severity", "info")
    severity_emoji = {
        "critical": "🔴", "warning": "🟡", "info": "🔵"
    }
    emoji = severity_emoji.get(severity, "⚪")

    return {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"{emoji} [{severity.upper()}] {alert['annotations'].get('summary', 'Unknown alert')}",
                },
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": (
                        f"*描述:* {alert['annotations'].get('description', 'N/A')}\n"
                        f"*实例:* {alert['labels'].get('instance', 'N/A')}\n"
                        f"*触发时间:* {alert.get('startsAt', 'N/A')}\n"
                        f"*状态:* {alert.get('status', 'N/A')}"
                    ),
                },
            },
            {"type": "divider"},
        ]
    }


def send_slack(message: dict):
    """发送格式化消息到 Slack。"""
    resp = requests.post(SLACK_WEBHOOK_URL, json=message, timeout=5)
    resp.raise_for_status()


def send_feishu(message: dict):
    """发送格式化消息到飞书（飞书的 webhook 消息格式）。"""
    text_blocks = []
    for block in message.get("blocks", []):
        if block["type"] == "header":
            text_blocks.append(block["text"]["text"])
        elif block["type"] == "section":
            text_blocks.append(block["text"]["text"])

    feishu_payload = {
        "msg_type": "interactive",
        "card": {
            "header": {
                "title": {"content": text_blocks[0] if text_blocks else "告警", "tag": "plain_text"},
                "template": "red" if "critical" in (text_blocks[0] or "").lower() else "yellow",
            },
            "elements": [
                {"tag": "div", "text": {"tag": "lark_md", "content": "\n".join(text_blocks[1:])}},
                {"tag": "hr"},
            ],
        },
    }
    resp = requests.post(FEISHU_WEBHOOK_URL, json=feishu_payload, timeout=5)
    resp.raise_for_status()


@app.route("/webhook/alertmanager", methods=["POST"])
def alertmanager_webhook():
    """接收 Alertmanager 的 webhook 请求。"""
    data = request.get_json(force=True)
    alerts = data.get("alerts", [])

    for alert in alerts:
        if alert.get("status") != "firing":
            continue

        severity = alert["labels"].get("severity", "info")
        formatted = format_alert(alert)

        # P1 → 双通道; P2 → Slack only; P3 → 不通知
        if severity == "critical":
            send_slack(formatted)
            send_feishu(formatted)
        elif severity == "warning":
            send_slack(formatted)

    return jsonify({"status": "ok"})


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=9093)
    print("🔔 Alertmanager webhook 处理器已启动 :9093")

代码设计要点：飞书的消息格式不同于 Slack——飞书使用「卡片消息」（card）而非 Slack 的 Block Kit。上述代码通过 format_alert() 生成统一的中间格式，再分别适配 Slack 和飞书的 API。P3 级别的告警不推送通知，仅在 Prometheus 中记录——这是避免告警疲劳的关键设计。

"静默失败"的告警难题

Agent 告警中最隐蔽的问题是静默失败（Silent Failure）：Agent 返回 HTTP 200，API 响应格式正确，工具调用全部成功——但给出的答案是完全错误的。凌晨 3:14 的事故就是静默失败的典型案例。

静默失败无法通过传统的 error_rate 或 latency 指标捕获。你需要评估信号（evaluation signals）——即在生产环境中持续运行的自动评测，来判断 Agent 的输出质量是否合格。具体做法：

• 在线评测埋点：在生产请求中采样 1%~5% 的流量，将 Agent 的输出发送到评测 pipeline（如另一个 LLM 作为评判器），检测事实性错误、工具选择错误、或回答不完整
• 工具选择分布监控：如果「退款处理」工具的调用量从每小时 200 次骤降到 0——即使所有工具调用都返回 success，这也是一个 P1 级别的信号（Agent 不再做退款了）
• 用户反馈信号：如果用户频繁点击「这个回答没用」或客服转接率飙升，这些是静默失败的最直接信号——但它们有滞后性（问题已经影响了真实用户）

静默失败是连接「可观测性」和「评测」的桥梁——我们将在 FAQ 中详细讨论两者之间的关系。先让我们进入最关键的能力：Trace、Metric、Log 的关联。

六、追踪、指标与日志的三位一体关联

如果 Metrics 是仪表盘，Tracing 是显微镜，Logging 就是黑匣子——它记录了 Agent 内部的每一步决策和每一个参数。三者的真正威力不在于各自独立使用，而在于关联查询：从 Metrics 发现异常 → 跳转到 Tracing 定位具体的请求 → 展开 Log 查看完整的上下文。没有这种关联能力，你会在三个独立的工具之间反复切换，拼凑碎片化信息——就像拿着三张不同的地图却找不到同一个地点。

完整的排查工作流

假设凌晨 3:14，你被 P1 告警叫醒。Prometheus 显示工具调用失败率飙升至 8%，但 Grafana 面板上没有任何 HTTP 5xx 错误。排查路径如下：

告警触发（Prometheus）
  │  Alertmanager → Slack: "Agent tool_call_failure_rate > 8%"
  │
  ▼
在 Grafana 中 drill down
  │  点击 metric → "Exemplar: trace_id=4f8a2c..."
  │
  ▼
打开 Jaeger/Tempo，输入 trace_id=4f8a2c...
  │  展开 Agent Span 树：
  │    root (7.2s)
  │    ├── llm_call #1 (1.1s)
  │    │   └── tool_call: search_db (4.3s ❌ timeout)
  │    ├── llm_call #2 (0.9s)
  │    │   └── tool_call: search_db (4.0s ❌ timeout)
  │    └── llm_call #3 (0.5s) → "无法处理请求"
  │
  ▼
从 Span 中提取 trace_id → 跳转到审计日志系统
  │  grep trace_id=4f8a2c... → 完整 LLM prompt + 工具参数 + 响应体
  │  发现：search_db 工具的超时阈值设为 4s，而数据库查询实际需要 6~8s
  │
  ▼
结论：数据库优化导致查询变慢，Agent 工具调用连续超时后放弃。

这个工作流之所以高效，是因为三个系统中的数据通过 trace_id 和 span_id 紧密关联。如果你没有做这种关联，你的排查过程会是：在 Grafana 中看到异常 → 翻日志找对应时间段的请求 → 手动匹配哪个请求的延迟数据对应哪条日志 → 在脑海中重建链路。对于一个 QPS 100 的 Agent 服务，这几乎是不可能的。

Prometheus Exemplars：指标到追踪的桥梁

Prometheus 的 Exemplar 机制允许你在指标数据点上附加一个 trace_id，使得在 Grafana 中可以直接从 metric 图表跳转到对应的 Trace。以 Histogram 为例：

# 在 Prometheus Python client 中记录 Histogram 时附加 exemplar
agent_request_duration_seconds.labels(agent_id="prod-03").observe(
    duration,
    # exemplar: 关联到具体的 trace
    exemplar={"trace_id": "4f8a2c1d3e9b6a07"}
)

在 Grafana 中配置好数据源关联（Prometheus ↔ Jaeger/Tempo）后，当你鼠标悬停在 P99 延迟的尖峰上时，Grafana 会显示一个「Exemplar」标记，点击即可跳转到该 Trace 的详细面板。这是从指标到追踪的最短路径——不需要手动翻日志、不需要猜测时间窗口、不需要 copy-paste trace_id。

结构化日志与 Trace Context 注入

仅仅在日志中「顺便写上 trace_id」是不够的——你需要确保每一行日志都自动携带 trace_id 和 span_id，这样你才能在日志系统中精确过滤出某一次请求的完整决策链。下面是一段 Python 日志配置代码，使用 OpenTelemetry 自动注入 trace context：

"""
Agent 结构化日志 — 自动注入 trace_id 和 span_id
依赖: pip install opentelemetry-api opentelemetry-sdk python-json-logger
"""

import logging
import sys
from pythonjsonlogger import jsonlogger

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter


# ── 1. 初始化 OTel TracerProvider ────────────────────────────
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer(__name__)


# ── 2. 自定义 JSON Formatter：注入 trace context ─────────────
class TraceContextFormatter(jsonlogger.JsonFormatter):
    """在每个日志记录中自动注入当前的 trace_id 和 span_id。"""

    def add_fields(self, log_record, record, message_dict):
        super().add_fields(log_record, record, message_dict)

        current_span = trace.get_current_span()
        span_context = current_span.get_span_context()

        if span_context.is_valid:
            log_record["trace_id"] = format(span_context.trace_id, "032x")
            log_record["span_id"] = format(span_context.span_id, "016x")
        else:
            log_record["trace_id"] = None
            log_record["span_id"] = None


# ── 3. 配置 root logger ──────────────────────────────────────
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(TraceContextFormatter(
    "%(asctime)s %(levelname)s %(name)s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S",
))

logging.basicConfig(level=logging.INFO, handlers=[handler])
logger = logging.getLogger("agent-service")


# ── 4. 使用示例 ──────────────────────────────────────────────
def process_agent_request(request_id: str):
    """在 Span 上下文内，所有日志自动携带 trace context。"""
    with tracer.start_as_current_span("agent.request") as span:
        span.set_attribute("agent.request.id", request_id)

        logger.info("开始处理请求", extra={"request_id": request_id})
        # → 输出: {..., "trace_id": "4f8a2c...", "span_id": "1d3e9b6a...",
        #           "message": "开始处理请求", "request_id": "..."}

        # LLM 调用
        with tracer.start_as_current_span("agent.llm.reasoning") as llm_span:
            logger.info("LLM 推理中", extra={
                "model": "gpt-4o", "step": 1
            })
            # → 输出: {..., "trace_id": "4f8a2c...", "span_id": "7c1a4f2d...",
            #           "message": "LLM 推理中", "model": "gpt-4o", "step": 1}

            # 工具调用
            with tracer.start_as_current_span("agent.tool.execute") as tool_span:
                logger.info("工具调用", extra={
                    "tool_name": "search_knowledge_base", "duration_ms": 180
                })
                # → 输出: {..., "trace_id": "4f8a2c...", "span_id": "f93b2e8a...",
                #           "message": "工具调用", "tool_name": "..."}

        logger.info("请求完成", extra={
            "request_id": request_id, "status": "success"
        })

设计要点：

• TraceContextFormatter 自动注入——不需要在每个 logger.info() 调用中手动传递 trace_id。只要代码在 tracer.start_as_current_span() 的上下文内，日志就自动携带正确的 trace context。
• JSON 格式输出——结构化日志是生产环境的必需品。JSON 使你可以用 jq 或 Loki/ELK 高效查询，而不是用 grep 和正则表达式挣扎。python-json-logger 库提供了一个即插即用的 JSON formatter。
• extra 参数的约定——使用 extra={...} 传递业务上下文（如 request_id、model、tool_name），这些字段会被追加到 JSON 日志中，方便在日志系统中按字段过滤。

「单一面板」反模式

一个常见的错误是试图把所有可观测性信息塞进一个 Grafana Dashboard——左侧放 QPS 和延迟曲线，中间放 Trace 列表，右侧放日志搜索结果，底部放 Alert 状态。这种「单一面板」看起来功能齐全，实际上完全不可用：

• 信息密度过高——on-call 工程师在凌晨被叫醒时，需要的是「问题在哪里」而不是「所有数据都在这里」
• 上下文不共享——点击 Grafana 中的一个 metric 数据点时，Trace 面板不会自动跳转到对应时间窗口的 Trace 列表
• 滚动同步失效——当你在日志面板中滚到 03:14:27 时，指标面板仍然显示 03:10~03:20 的整个窗口

正确的做法是关联而非堆砌：

1. 一个概览 Dashboard（40 秒扫一眼就知道系统是否健康）——只有 6~8 个核心面板
2. 一个单体请求视图（输入 trace_id → 展开完整的 Span 树 + 关联的日志行）——在 Jaeger 或 Grafana Tempo 中实现，不需要自己拼
3. 一个工具详情 Dashboard（按工具名称筛选，展示每个工具的成功率、延迟分布和调用量趋势）
4. 面板间链接：在概览 Dashboard 的 P99 延迟面板上配置 data link，点击数据点自动跳转到 Jaeger 中对应时间范围的 Trace 列表

有了 Trace、Metric、Log 的完整关联能力，你就具备了生产级 Agent 运维的核心工具链。但工具链再好，也需要正确的落地方案——下一节我们来设计一条从零起步的渐进式路线。

七、渐进式落地路线：从零到完整可观测性

对于大多数团队来说，一次性实施四层信号 + Tracing + 告警 + 日志关联是不现实的——既没有那么多人力，也没有那么迫切的业务需求。正确的做法是分阶段渐进式落地，每个阶段交付一个明确的能力增量，并且每个阶段都有一个「good enough」检查点——到达这个点就可以暂停可观测性建设，把精力转回业务开发。

五阶段落地路线

每个阶段的具体行动

Phase 0 · 结构化日志（必须最先做）

这是整个可观测性体系的基石。没有结构化日志，你连「一个请求到底做了什么」都无法追溯。关键动作：① 引入 python-json-logger 将日志格式化为 JSON；② 实现 TraceContextFormatter 自动注入 trace_id 和 span_id；③ 将日志发送到集中存储（Loki 通过 promtail、或 ELK）；④ 配置日志保留策略（生产环境建议 30 天）。完成后，你应该能在 Loki 中执行 {trace_id="4f8a2c..."} 查询并立即看到完整的请求链路日志。

Phase 1 · 基础 Tracing（价值最大的增量）

在你有了结构化日志之后，下一步是为 Agent 的推理循环添加 Span。不需要立刻上 OTel Collector 和分布式 Tracing 集群——先用 ConsoleSpanExporter 本地调试，确认 Span 层级正确（root → llm_call → tool_call），再部署 Jaeger all-in-one 或 Grafana Tempo。关键动作：① 实现 traced_span 上下文管理器；② 定义 Span 命名规范（agent.llm.reasoning、agent.tool.execute）；③ 设置关键 attribute（model、tokens、tool_name、step）；④ 部署 Jaeger/Tempo 并验证 Trace 可视化。

Phase 2 · 核心指标（让非工程师也能看懂系统状态）

Tracing 是工程师的工具，Metrics 是所有人的工具——产品经理、业务负责人、SRE 都需要通过 Grafana 面板了解 Agent 的状态。关键动作：① 定义 6 个核心 Prometheus 指标（Counter + Histogram）；② 通过 prometheus_client 暴露 /metrics 端点；③ 配置 Prometheus scrape job；④ 构建 2 个 Grafana Dashboard——概览面板（QPS、P99、token 消耗、错误率）和工具详情面板（每个工具的成功率和延迟分布）；⑤ 为关键指标添加 Exemplar。

Phase 3 · 告警规则（从被动查看到主动通知）

在你有 2~4 周的生产数据后，你已经知道了 Agent 的「正常基线」——正常的 P99 延迟是多少、正常的 token 日消耗是多少、正常的工具调用成功率是多少。有这些基线数据后，告警阈值才有意义。关键动作：① 基于历史基线设置告警阈值（不是拍脑袋填数字）；② 编写 Prometheus 告警规则 YAML；③ 实现 webhook 通知处理器；④ 第一周只开 P2 级别的告警，验证无误报后再开 P1；⑤ 每周 review 告警规则，调整阈值、删除噪音规则。

Phase 4 · 完整关联（持续优化，没有终点）

前四个阶段已经覆盖了 80% 的需求。Phase 4 是持续优化阶段——包括 Exemplar 关联的完善、业务指标的深化（如按用户类型切分的 task_completion_rate）、在线评测信号的接入（如 hallucination 检测结果自动进入 Prometheus）、以及成本面板的精细化（按模型、按 Agent 实例、按任务类型拆分 token 成本）。

「Good Enough」检查点：何时停止添加可观测性

可观测性建设有一个递减回报的拐点——超过某个点后，每增加一个指标带来的价值小于维护它的成本。判断你是否到达「good enough」的三个信号：

1. 过去 3 次生产事故中，你都能在 10 分钟内定位到根因。如果还需要手动 grep 日志或猜测问题来源，说明你的 Tracing 或日志关联还不完善。
2. 告警误报率 < 10%。如果 on-call 收到的大多数告警都是误报，说明你的阈值设置或告警规则需要优化——这不是「再加一个指标」能解决的问题。
3. 团队中非 on-call 成员也能在 5 分钟内判断系统是否健康。如果你的 Grafana 面板只有你自己看得懂，说明可视化设计还不够直观。

当这三个条件都满足时，停止增加新的可观测性组件，把精力转向业务能力建设。可观测性的目标是让你信任你的系统——而不是让你无休止地监控它。

八、开源方案 vs 商业平台：如何选型

Agent 可观测性领域正在快速分化——一边是完全开源的 OTel + Prometheus + Grafana 技术栈，另一边是 LangSmith、Datadog LLM 等商业平台。两者不是互斥的：大多数成熟团队的最佳实践是以开源栈为基础，在需要时引入商业平台作为补充。本节提供一套系统化的选型框架。

决策矩阵：你属于哪种画像？

方案一：纯开源自建（推荐起步方案）

优势：零许可费用、完全可控、与现有 Prometheus 体系无缝集成、数据不离开 VPC。劣势：需要专人维护（估计 0.2~0.5 个 FTE）、没有 LLM 特有的可视化（如 prompt 版本对比、token 成本拆解）。

方案二：开源 Agent 平台（中间路线）

LangFuse 和 Arize Phoenix 填补了纯 OTel 栈的空白——它们提供了 LLM 和 Agent 特有的可视化能力（如 prompt 版本和性能关联、token 成本按模型拆分、LLM 调用链回放），同时保持开源和自托管。如果你的团队已经有 OTel + Prometheus 基础，优先考虑 LangFuse——它通过 OTel integration 可以接收你的现有 Span，不需要额外插桩。

方案三：商业平台

商业平台的核心价值不是「替代开源栈」，而是提供LLM 和 Agent 特有的高级功能——如 LangSmith 的 prompt 版本对比和回归测试、Datadog 的端到端延迟拆解（HTTP → Agent → LLM → Tool → LLM → Response）。如果你只需要基础的指标和 Tracing，开源栈完全足够。

推荐选型路径

基于团队画像和需求，以下是三条典型的选型路径：

我们的核心建议：从 OTel + Prometheus 开始。理由很简单——OTel 是行业标准，不锁定任何厂商；Prometheus 是你大概率已经有的基础设施；这套组合的成本为零，且足以覆盖 Agent 生产环境 80% 的可观测性需求。当你的 Agent 日请求量突破 10 万、或者你需要 prompt 版本对比等高级功能时，再评估是否引入 LangSmith 或 Datadog LLM。这样你花掉的每一分钱都在解决真实存在的问题，而不是为「万一有用」的功能预付费。

常见问题 (FAQ)

下一步阅读 / Next Steps

Agent 可观测性不是一个孤立的话题——它与 Agent 生产化的所有方面都紧密相连。以下是你接下来应该探索的方向：

• Agent 审计日志设计 — 如何记录 Agent 每次决策的完整轨迹（prompt、工具参数、LLM 输出、审批记录），满足合规要求和事后回溯需求
• Agent 评测框架 — 构建在线和离线评测体系，量化 Agent 的输出质量，连接可观测性信号与质量指标
• Agent 错误恢复 — 设计多级错误恢复策略（重试、降级、安全终止），并监控恢复成功率
• Agent 工具权限控制 — 在工具执行层实现最小权限、审批流程和风险评分，将安全事件纳入可观测性
• Agent 安全评测 — 检测 prompt 注入、数据泄露、权限绕过等安全风险，将安全信号集成到告警体系
• MCP 协议生产实践 — 在生产环境中部署和管理 MCP 工具，确保 MCP 调用与 Agent 可观测性无缝集成

每一篇文章都建立在可观测性的基础之上——因为你无法管理你看不到的东西。有了本文构建的指标、追踪和告警体系，你就有了 Agent 生产化的「眼睛」和「耳朵」。接下来，用它来让你的 Agent 更可靠、更安全、更值得信赖。