Agent线上事故排查手册从日志追踪到根因定位

一、没有可观测性，排查就是猜

传统服务的故障有明确报错栈，Agent的故障却常常是"答非所问""卡住不动""突然烧钱"。原因在于：Agent的行为由模型+上下文+工具共同决定，任何一个环节出问题，表象都可能是"回答变差了"。所以排查的第一前提，是上线前就把可观测性埋好。

二、结构化日志与trace_id贯穿

排查的第一步是能"连点成线"：一次Agent会话涉及用户输入、模型调用、工具调用、检索结果等多个环节，必须用同一个trace_id串起来：

import uuid
import logging
import json

class TraceContext:
    """贯穿一次Agent请求的追踪上下文"""
    def __init__(self):
        self.trace_id = uuid.uuid4().hex[:16]
        self.events = []

    def log(self, stage: str, **fields):
        entry = {"trace_id": self.trace_id, "stage": stage, **fields}
        self.events.append(entry)
        # 结构化输出，便于ELK/Loki检索
        logging.getLogger("agent").info(json.dumps(entry, ensure_ascii=False))
        return entry

def agent_with_tracing(user_input, tools):
    ctx = TraceContext()
    ctx.log("input", text=user_input, length=len(user_input))
    # 记录每次模型调用的输入输出token
    resp, usage = call_llm_with_usage(build_messages(user_input))
    ctx.log("llm", model=resp.model, in_tokens=usage.prompt_tokens,
            out_tokens=usage.completion_tokens, latency_ms=resp.latency_ms)
    for tool_call in resp.tool_calls:
        try:
            result = tools[tool_call.name](**tool_call.arguments)
            ctx.log("tool", name=tool_call.name, args=tool_call.arguments,
                    status="ok", result_preview=str(result)[:200])
        except Exception as e:
            ctx.log("tool", name=tool_call.name, status="error", error=str(e))
    ctx.log("output", text=resp.content[:500])
    return resp

核心原则：凡是模型看到的、工具返回的、Agent输出的，全部落日志。尤其是工具返回值，必须截断记录——上下文污染类故障全靠它定位。

三、六大高频故障模式速查表

故障	典型症状	根因	定位方法
Token超限截断	回答到一半戛然而止	上下文超窗口被截断	查llm日志的in_tokens与模型上限对比
工具死循环	同一工具反复调用N次	缺循环次数上限	查tool日志调用序列
上下文污染	回答内容混入乱码/文档原文	工具返回未清洗直接入上下文	查tool日志result_preview
格式漂移	结构化输出突然解析失败	Prompt改动/模型版本变化	查输出解析失败率趋势
密钥权限错误	工具调用401/403	环境变量缺失或过期	查tool日志status=error
缓存污染	新数据查不到/旧答案反复出现	缓存键设计缺陷	查缓存命中日志

四、实战定位流程：从告警到根因

接到告警后，按五步走，不要一上来就改代码：

# 第1步：确认影响面 —— 这个故障是单例还是大面积？
# 查询最近1小时按错误类型聚合
# SELECT stage, status, COUNT(*) FROM agent_logs
# WHERE ts > now() - 1h GROUP BY stage, status ORDER BY 3 DESC

# 第2步：抽取一条完整trace复现链路
# SELECT * FROM agent_logs WHERE trace_id = 'xxx' ORDER BY ts
# -> 人工按 input -> llm -> tool -> output 顺序读一遍

# 第3步：对照速查表定位故障类别
# 第4步：用同一输入在测试环境重放，确认可复现性
def replay(trace_id, log_store, agent):
    """把线上trace的输入重放到当前版本，验证是否已修复"""
    events = log_store.query(trace_id=trace_id)
    original_input = events[0]["text"]
    new_output = agent.run(original_input)
    return compare_with(events[-1]["text"], new_output)

# 第5步：修复 -> 重放验证 -> 灰度发布

重放是Agent排障的杀手锏：由于模型输出有随机性，复现不了时不要纠结，把trace里的完整上下文（含历史消息）喂给模型，人工分析它在哪个环节跑偏。

五、止损三板斧：熔断、降级、回滚

定位之前先止血。Agent故障的止损手段按顺序上：

class CircuitBreaker:
    """熔断器：连续失败超阈值自动断开，防止故障扩大"""
    def __init__(self, threshold=5, cooldown=60):
        self.failures = 0
        self.threshold = threshold
        self.cooldown = cooldown
        self.open_until = 0

    def call(self, fn, fallback=None):
        if time.time() < self.open_until:
            return fallback() if fallback else None  # 熔断态走兜底
        try:
            result = fn()
            self.failures = 0
            return result
        except Exception:
            self.failures += 1
            if self.failures >= self.threshold:
                self.open_until = time.time() + self.cooldown
                log_alert("circuit_open", threshold=self.threshold)
            raise

手段	动作	适用场景
熔断	停止调用故障依赖	工具API大面积报错
降级	切备用模型/简化流程	主模型超时、成本失控
回滚	恢复上一版Prompt/配置	新改动引入的回归

六、复盘与预防

每次事故后填写复盘清单，把教训沉淀成自动化防护：

[ ] 是否补充了对应故障的监控告警？

[ ] 该故障场景是否已加入黄金评测数据集？

[ ] 是否需要增加循环上限/超时/重试策略？

[ ] 日志是否缺少定位该问题所需的关键字段？

[ ] 是否更新了故障速查表？

**一句话总结**：Agent排障拼的不是临场反应，而是平时的日志完整度。把"模型看到了什么、工具返回了什么、Agent输出了什么"记全，80%的疑难杂症都能在五分钟内定位。建议每月做一次故障演练，用线上真实trace训练团队的排查手感。