Agent 出了 Bug 你怎么知道？全链路可观测性与错误自愈实战

当你把一个 AI Agent 部署到生产环境，很快就会发现一个尴尬的事实：传统监控完全失效。

HTTP 返回 200，延迟正常，错误率为零——但 Agent 的回答是幻觉。用户投诉了，你翻遍日志，只看到一堆 JSON，根本还原不出 Agent 在第几步拐错了弯。这不是个别现象，而是 2026 年 Agent 工程的普遍困境：你有一个能跑的 Agent，但你不知道它为什么跑对了，也不知道它为什么跑错了。

本文从实战角度出发，给出 AI Agent 可观测性与错误自愈的完整方案——不是理论框架，是你今天就能动手改的东西。

为什么传统 APM 对 Agent 无效

传统应用监控基于「请求→响应」模型：固定代码路径，确定输出，堆栈即可定位。AI Agent 完全不同：

维度	传统应用	AI Agent
执行路径	固定代码分支	每次推理可能走不同策略
输出确定性	相同输入→相同输出	相同输入→可能不同输出
错误定位	看堆栈和错误码	需要分析推理链路和上下文
性能瓶颈	通常在 IO 或计算	可能在模型推理、RAG 召回或工具等待
质量评估	功能测试通过即可	需要持续评估输出准确性和可靠性

一个典型 Agent 执行链路涉及五个以上环节：提示词组装 → 模型推理 → RAG 检索 → 工具调用 → 结果整合。任何一个环节异常都可能导致最终输出质量下降，而你只看到了最终结果。

可观测性三支柱：Trace、Metric、Evaluation

1. Agent Trace：执行链路全记录

Trace 是可观测性的核心。它记录一个完整 Agent 任务从输入到输出的每一步，每一步都应记录：

• 输入内容（含动态注入的上下文和知识片段）
• 执行耗时与 Token 消耗
• 模型决策依据（推理链 / Chain-of-Thought）
• 工具调用参数与返回结果
• 执行状态（成功、重试、降级、失败）

关键实践：为每个 Agent 执行分配唯一的 Trace ID，支持跨服务追踪。Multi-Agent 协作时，子 Agent 的 Trace 应作为父 Trace 的子 Span 挂载，形成完整的调用树。

以下是基于 OpenTelemetry 的最小埋点方案：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# 初始化 Tracer
resource = Resource.create({"service.name": "my-agent-service"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agent-runtime")


def run_agent_with_trace(user_input: str):
    with tracer.start_as_current_span("agent.run") as root_span:
        root_span.set_attribute("input", user_input)
        root_span.set_attribute("agent.version", "2.3.1")

        # Step 1: 模型推理
        with tracer.start_as_current_span("agent.llm_call") as llm_span:
            response = llm_client.chat(model="gpt-4o", messages=[...])
            llm_span.set_attribute("model", "gpt-4o")
            llm_span.set_attribute("input_tokens", response.usage.prompt_tokens)
            llm_span.set_attribute("output_tokens", response.usage.completion_tokens)
            llm_span.set_attribute("latency_ms", response.latency_ms)

        # Step 2: 工具调用
        with tracer.start_as_current_span("agent.tool_call") as tool_span:
            tool_span.set_attribute("tool.name", "search_database")
            tool_span.set_attribute("tool.args", json.dumps(tool_args))
            result = execute_tool(tool_name, tool_args)
            tool_span.set_attribute("tool.status", "success" if result.ok else "error")

        root_span.set_attribute("output", final_output)
        return final_output

如果你用的是 LangChain/LangGraph，只需设置两个环境变量即可零侵入开启追踪：

export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="your-key"

LangSmith 会自动捕获 Chain、Agent、Tool 的调用链路，无需手写任何 span 代码。2026 年的 LangSmith 已原生支持 OpenTelemetry 导出，你可以同时把 Trace 数据发到 LangSmith 做 Agent 级调试，发到 Jaeger/Grafana 做系统级监控。

2. Agent Metric：运营级指标体系

Agent 的指标不能只看「响应时间」和「错误率」，需要建立业务 + 技术双层指标体系：

指标类别	核心指标	健康阈值建议
任务质量	任务完成率、幻觉率、首次正确率	完成率 > 90%，幻觉率 < 3%
执行效率	P50/P95/P99 延迟、Agent 循环次数、工具调用次数	P95 < 15s，平均循环 < 4 次
资源消耗	日均 Token 用量、API 调用成本、GPU 利用率	日均成本波动 < 20%
知识质量	RAG 命中率、Top-K 相关性得分、无结果率	命中率 > 85%，无结果 < 5%
安全合规	敏感数据触达次数、越权操作次数	越权 = 0

3. Agent Evaluation：持续评测闭环

与传统软件不同，AI Agent 的「Bug」往往不是代码错误，而是输出质量退化。评测不能只在上线前做一次，必须持续进行：

• 在线评测：对生产流量实时采样评估，监测质量趋势
• 回归评测：每次模型升级或 Prompt 变更后，用标准化测试集验证
• 对比评测：A/B 测试不同模型、Prompt 策略或 RAG 配置
• 人工评审：定期抽取 Agent 输出让业务专家评分

分层错误防御：从重试到自愈

Agent 的工具调用失败是常态，不是异常。关键在于如何让 Agent 从失败中恢复。以下是生产级的分层防御策略：

Layer 1：带抖动的指数退避重试

最基础的韧性层，但 90% 的团队都做错了——没有抖动（Jitter），所有客户端在同一时刻重试，造成惊群效应。

import asyncio
import random
from functools import wraps

def retry_with_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter: bool = True,
):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except (RateLimitError, TimeoutError) as e:
                    last_error = e
                    if attempt == max_retries:
                        break
                    delay = min(max_delay, base_delay * (2 ** attempt))
                    if jitter:
                        delay += random.uniform(0, 0.25 * delay)
                    await asyncio.sleep(delay)
            raise last_error
        return wrapper
    return decorator

@retry_with_backoff(max_retries=3, base_delay=1.0)
async def call_llm(prompt: str):
    return await llm_client.chat(prompt=prompt)

关键：抖动可以将重试风暴减少 60-80%。

Layer 2：错误分类——不是所有错误都该重试

from enum import Enum

class ErrorType(Enum):
    TRANSIENT = "transient"      # 可重试：限流、超时、网络抖动
    PERMANENT = "permanent"      # 不可重试：参数错误、权限不足
    RATE_LIMIT = "rate_limit"    # 需等待：429
    CONTENT_POLICY = "content"   # 内容安全：换模型或降级

def classify_error(error: Exception) -> ErrorType:
    if isinstance(error, RateLimitError):
        return ErrorType.RATE_LIMIT
    elif isinstance(error, (AuthenticationError, ValidationError)):
        return ErrorType.PERMANENT
    elif isinstance(error, ContentPolicyError):
        return ErrorType.CONTENT_POLICY
    else:
        return ErrorType.TRANSIENT

对 PERMANENT 错误立即失败，不要浪费重试次数；对 RATE_LIMIT 错误尊重 Retry-After 头；对 CONTENT_POLICY 错误换模型降级——分类先于行动。

Layer 3：熔断器——当下游真的挂了

当某个工具 API 连续失败，停止调用它，避免雪崩：

import time
from dataclasses import dataclass, field

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    cooldown_seconds: float = 60.0
    _failure_count: int = field(default=0, init=False)
    _state: str = field(default="closed", init=False)  # closed / open / half_open
    _last_failure_time: float = field(default=0.0, init=False)

    def can_execute(self) -> bool:
        if self._state == "closed":
            return True
        if self._state == "open":
            if time.time() - self._last_failure_time > self.cooldown_seconds:
                self._state = "half_open"
                return True  # 试探性放行
            return False
        return True  # half_open 放行一次

    def record_success(self):
        self._failure_count = 0
        self._state = "closed"

    def record_failure(self):
        self._failure_count += 1
        self._last_failure_time = time.time()
        if self._failure_count >= self.failure_threshold:
            self._state = "open"

# 每个工具一个熔断器
tool_breakers = {}

async def call_tool_with_breaker(tool_name: str, **kwargs):
    breaker = tool_breakers.setdefault(tool_name, CircuitBreaker())
    if not breaker.can_execute():
        raise CircuitOpenError(f"Tool {tool_name} circuit is open, fail fast")
    try:
        result = await execute_tool(tool_name, **kwargs)
        breaker.record_success()
        return result
    except Exception as e:
        breaker.record_failure()
        raise

Layer 4：模型降级与回退链

当主模型不可用，自动切换到备选模型：

FALLBACK_CHAIN = [
    {"provider": "anthropic", "model": "claude-sonnet-4-20250514"},
    {"provider": "openai", "model": "gpt-4o"},
    {"provider": "anthropic", "model": "claude-haiku-4-20250414"},
    {"provider": "local", "model": "llama-3-70b"},
]

async def call_with_fallback(prompt: str, context: list):
    for config in FALLBACK_CHAIN:
        try:
            return await llm_client.chat(
                model=config["model"], messages=context, timeout=15
            )
        except (RateLimitError, TimeoutError, ServiceUnavailableError):
            continue
    raise AllModelsUnavailableError("所有模型均不可用")

关键：回退时必须传递完整对话历史，确保降级模型继承计算上下文。

五大典型生产故障与排查

故障模式	症状	排查方法	预防措施
幻觉爆发	回答编造虚假数据	检查 RAG 命中率和知识源版本	幻觉检测器 + 无知识时拒绝回答
循环死锁	Agent 反复执行不收敛	分析 Trace 中循环步骤和决策变化	最大循环次数 + 循环检测断路器
工具雪崩	下游 API 超时导致整体瘫痪	监控工具调用延迟和错误率分布	超时 + 熔断 + 降级
上下文溢出	超长对话耗尽 Token 窗口	追踪每步 Token 消耗和上下文裁剪逻辑	智能上下文管理 + 分段处理
权限越界	Agent 访问不该接触的数据	审计日志中检查调用链和权限校验	最小权限 + 运行时权限校验

实战教训：某金融客户 Agent 上线第三天，因 RAG 索引未更新导致引用已作废利率政策。这个问题通过日志根本无法发现，只有评测系统检测到「知识源版本不一致」异常时才定位到根因。可观测性不能只有日志。

最小可行方案：30 分钟上手

如果你从零开始，按以下优先级实施：

1. 第一步（今天）：接入 Trace。如果用 LangChain，设两个环境变量即可。如果自研框架，用上面的 OpenTelemetry 埋点模板，确保每次执行有 Trace ID。
2. 第二步（本周）：加上重试 + 抖动 + 错误分类。这是投入产出比最高的改动。
3. 第三步（下周）：给外部工具加熔断器。先给最不稳定的工具加。
4. 第四步（持续）：建立评测数据集，实施在线采样评测。

结论

可观测性不是 Agent 上线后的「附加功能」，而是进入生产的前提条件。没有 Trace 的 Agent，就像没有仪表盘的飞机——或许能飞，但没有人敢坐。

2026 年的 Agent 工程已经过了「能跑就行」的阶段。OpenTelemetry + LangSmith 的组合让你可以同时拥有 Agent 级调试深度和系统级监控广度；分层错误防御让 Agent 从「一碰就碎」变成「越挫越勇」。先让 Agent 可观测，再让它可靠——这个顺序不能反。