LLM Agent故障实时检测与自动修复实战指南

最近在做大模型 Agent 相关项目时，最让人头疼的往往不是模型能力不够，而是 Agent 在真实运行中频繁出现各种意外失败：工具调用参数解析不了、外部 API 超时、上下文窗口被撑爆、重试多次依旧卡死……更麻烦的是，这些失败通常要等用户反馈，或者定时任务跑完才发现，整个过程非常被动。有一回线上任务链断了快两个小时，界面上却一直显示“执行中”，最后排查日志才发现某个中间步骤的 JSON 输出字段少了一个引号。

为了让这类问题能被提前感知并自动恢复，我整理了一套 LLM Agent 失败的实时检测与修复方案。本文会把失败类型建模、Agent 执行状态机、监控埋点、修复策略到完整代码示例串起来，尽量做到可以直接在工程中落地。无论你是刚接触 Agent 开发的初学者，还是已经将 Agent 接入生产环境的开发者，都能从中找到可复用思路。

1. LLM Agent 失败的本质与类型

1.1 为什么失败不能只靠模型层判断

先区分两个概念：大语言模型（LLM）本身只负责文本生成，而 Agent 是一个完整的决策与执行系统。

我们通常说的 Agent 包含：任务理解、规划拆解、工具调用、结果观察、最终回复等多个环节。任何一个环节出问题，都会导致整体任务失败。

如果把 Agent 当成一个黑盒去监控，只能看到“最终返回错误”或“长时间没响应”，无法定位问题发生在哪一步。因此实时检测的第一步，是先把 Agent 的执行链路显式拆开，记录每一步的状态变化。实际项目中我推荐把检测重点放在“输出格式是否合法、工具调用是否成功、上下文是否超限、重试是否失控”四个维度上。

1.2 最常见的四类 LLM Agent 失败

第一类是格式化错误。模型返回的结果不是预期结构，例如应该输出 JSON 却返回了 Markdown 文本，或者 JSON 被截断导致json.loads直接报错。这类现象在长输出场景下尤其常见。

第二类是工具执行异常。Agent 调用外部 API、数据库、文件系统时，可能出现超时、限流、参数不合法、依赖服务临时不可用等情况。严格说这类错误不一定是 Agent 自身逻辑造成的，但 Agent 必须能处理，不能把工具异常原样抛给用户。

第三类是上下文窗口溢出。多轮对话和工具结果累积后，prompt 长度可能超过模型 token 上限。一旦溢出，轻则程序报错，重则模型开始“遗忘”早期指令，行为变得不稳定。

第四类是重试与循环失控。Agent 在遇到失败后如果无脑重试，可能进入死循环，浪费 tokens 并且延长故障时间。这类问题需要靠修复策略中的最大重试次数和熔断机制控制。

1.3 实时检测 + 修复的价值

如果没有实时检测，通常只能做“事后追溯”：用户反馈、翻日志、复现问题。这个周期可能是小时级甚至天级。如果加上实时检测和自动修复，系统可以在秒级发现异常，并根据失败类型选择重试、降级或者转人工。

从工程角度看，实时检测解决“看得见”的问题，自动修复解决“恢复快”的问题。两者配合，才能降低 Agent 任务的 MTTR（平均恢复时间）。这也是本文重点演示的内容。

2. 环境准备与项目结构

2.1 运行环境

本文示例基于 Python 编写，建议使用 Python 3.10 及以上版本。操作系统不限，Windows、macOS、Linux 都可以。

这里有一个取舍需要说明：为了让你能在本地直接跑通，本文的例子没有真正接入大模型 API，而是用一个模拟 Agent 执行器来复现失败场景。这样可以避免申请 API Key、配置网络、产生 tokens 费用等额外成本。你理解了检测与修复流程后，再替换成真实的大模型调用即可。

2.2 依赖库

示例代码全部使用 Python 标准库，不需要额外安装第三方包。核心依赖如下：

• enum：定义 Agent 状态和失败类型。
• dataclasses：定义执行记录结构。
• logging：输出结构化日志。
• time：统计耗时和实现退避等待。
• abc：定义修复策略接口（如果你的工程扩展性要求高）。

如果你需要在真实项目中接入大模型，可以自行安装openai、langchain等 SDK，版本需要根据项目实际情况调整。本文重点演示的是检测框架，而不是某个 SDK 的具体用法。

2.3 项目结构

agent_failure_monitor/ ├── agent_failure_types.py # 失败类型、状态枚举、运行记录 ├── agent_errors.py # 自定义异常 ├── failure_detector.py # 实时检测器 ├── repair_engine.py # 修复引擎 ├── simulated_agent.py # 模拟 Agent 执行器 ├── main.py # 示例入口 └── requirements.txt # 可选，本示例无需依赖

下面按这个结构逐文件实现。

3. 核心原理：Agent 执行的监测链路

3.1 Agent 执行循环

一个典型的 Agent 执行循环可以拆成以下几步：

1. 接收任务。
2. 对任务进行规划，拆成多个步骤。
3. 针对当前步骤生成工具调用指令。
4. 执行工具并获取结果。
5. 把工具结果放回上下文，继续下一步。
6. 全部步骤完成后生成最终回复。

每一步都可能出现不同类型的失败。例如第 3 步可能产生 JSON 解析错误，第 4 步可能发生工具超时，第 5 步可能触发上下文溢出。所以检测器不能只在 Agent 末尾兜底，而要在“每轮迭代”的边界上埋点。

3.2 关键监控指标

在设计检测器时，我建议至少关注以下几个指标：

• 开始时间与结束时间。
• 当前状态是规划中、工具调用中还是已完成。
• 每一次执行是第几次尝试。
• 失败类型是什么。
• 重试是否已经超过阈值。

这些指标看起来简单，却是故障定位的基础。很多生产事故难排查，就是因为日志里缺少“状态 + 尝试次数 + 失败类型”这三个维度。

3.3 用状态机表达 Agent 生命周期

给 Agent 定义一个显式状态机，能避免“状态混乱”的问题。例如：

• STARTED：任务启动。
• PLANNING：正在规划。
• TOOL_CALLING：正在调用工具。
• OBSERVING：正在观察工具结果。
• FINISHED：正常结束。
• FAILED：执行失败。

检测器在每个状态间切换时记录日志，一旦进入FAILED状态立即触发修复策略。这样做会话等级的可观测性会好很多。

4. 实战：构建 LLM Agent 故障实时检测与修复系统

4.1 定义失败类型、状态与运行记录

先创建agent_failure_types.py，用于统一管理枚举和数据模型。

文件路径：agent_failure_monitor/agent_failure_types.py

from enum import Enum from dataclasses import dataclass, field from typing import Optional import time class AgentState(str, Enum): """Agent 执行状态。""" STARTED = "started" PLANNING = "planning" TOOL_CALLING = "tool_calling" OBSERVING = "observing" FINISHED = "finished" FAILED = "failed" class AgentFailureType(str, Enum): """Agent 失败类型，用于后续修复策略路由。""" TOOL_CALL_PARSE_ERROR = "tool_call_parse_error" TOOL_EXECUTION_TIMEOUT = "tool_execution_timeout" CONTEXT_WINDOW_OVERFLOW = "context_window_overflow" RETRY_EXCEEDED = "retry_exceeded" OUTPUT_VALIDATION_FAILED = "output_validation_failed" UNKNOWN_ERROR = "unknown_error" @dataclass class AgentRunRecord: """ 一次 Agent 运行的完整记录，方便在检测与修复过程中传递状态。 """ agent_id: str task: str state: AgentState = AgentState.STARTED attempts: int = 0 start_time: float = field(default_factory=time.time) end_time: Optional[float] = None error: Optional[str] = None failure_type: Optional[str] = None def finish( self, state: AgentState, error: Optional[str] = None, failure_type: Optional[str] = None, ) -> None: self.end_time = time.time() self.state = state self.error = error self.failure_type = failure_type def duration(self) -> float: end = self.end_time or time.time() return round(end - self.start_time, 3)

这里把失败类型设计成枚举，而不是散落的字符串，是为了后续修复策略可以做更可靠的分发。如果直接用字符串，很容易在代码里写错单词且不会报错。

4.2 定义自定义异常

接下来是agent_errors.py，把不同类型的 Agent 失败映射成具体异常类。

文件路径：agent_failure_monitor/agent_errors.py

class ToolCallParseError(Exception): """模型输出无法解析成合法的工具调用参数时抛出。""" pass class ToolExecutionTimeoutError(Exception): """工具调用超过指定时间上限时抛出。""" pass class ContextWindowOverflowError(Exception): """上下文窗口接近或超过模型限制时抛出。""" pass class OutputValidationError(Exception): """Agent 最终输出未通过规则校验时抛出。""" pass class RetryExceededError(Exception): """修复重试次数超过上限时抛出。""" pass

把这些异常类单独放在一个模块里，是为了避免在检测器中直接依赖某个具体 Agent 的实现。任何 Agent 模块都可以抛出这些异常，检测器统一捕获并分类。

4.3 实现实时检测器

实时检测器是整个系统的核心。它提供一个watch方法，接收一个可执行的 Agent 函数，并自动完成如下工作：

1. 创建运行记录。
2. 执行 Agent 函数。
3. 捕获所有异常。
4. 将异常映射为规范化失败类型。
5. 输出结构化日志。
6. 返回执行结果与运行记录。

文件路径：agent_failure_monitor/failure_detector.py

import logging from typing import Any, Callable, Dict, Optional, Tuple from agent_errors import ( ToolCallParseError, ToolExecutionTimeoutError, ContextWindowOverflowError, OutputValidationError, RetryExceededError, ) from agent_failure_types import AgentRunRecord, AgentState, AgentFailureType logger = logging.getLogger("agent.failure_detector") class AgentFailureDetector: def classify(self, error: Exception) -> str: """ 根据异常类型将异常归一化为 AgentFailureType。 优先使用异常类型判断，再辅以错误文本关键字兜底。 """ if isinstance(error, ToolCallParseError): return AgentFailureType.TOOL_CALL_PARSE_ERROR.value if isinstance(error, ToolExecutionTimeoutError): return AgentFailureType.TOOL_EXECUTION_TIMEOUT.value if isinstance(error, ContextWindowOverflowError): return AgentFailureType.CONTEXT_WINDOW_OVERFLOW.value if isinstance(error, OutputValidationError): return AgentFailureType.OUTPUT_VALIDATION_FAILED.value if isinstance(error, RetryExceededError): return AgentFailureType.RETRY_EXCEEDED.value error_msg = str(error).lower() if "json" in error_msg or "parse" in error_msg: return AgentFailureType.TOOL_CALL_PARSE_ERROR.value if "timeout" in error_msg or "timed out" in error_msg: return AgentFailureType.TOOL_EXECUTION_TIMEOUT.value if "context" in error_msg or "token limit" in error_msg: return AgentFailureType.CONTEXT_WINDOW_OVERFLOW.value return AgentFailureType.UNKNOWN_ERROR.value def watch( self, agent_fn: Callable[..., Any], agent_id: str = "agent-default", task: str = "", **agent_kwargs: Any, ) -> Tuple[Optional[Any], AgentRunRecord]: """ 带监控的 Agent 执行入口。 参数: agent_fn: 实际执行 Agent 逻辑的可调用对象。 agent_id: Agent 实例标识，用于链路追踪。 task: 当前任务描述，便于日志检索。 **agent_kwargs: 传递给 agent_fn 的其余参数。 """ record = AgentRunRecord(agent_id=agent_id, task=task) try: result = agent_fn(**agent_kwargs) record.finish(AgentState.FINISHED) logger.info( "agent_finished agent_id=%s task=%s duration=%.3fs", record.agent_id, record.task, record.duration(), ) return result, record except Exception as e: failure_type = self.classify(e) record.finish(AgentState.FAILED, str(e), failure_type) logger.warning( "agent_failed agent_id=%s task=%s failure_type=%s error=%s", record.agent_id, record.task, failure_type, str(e), ) return None, record

这里有两个设计细节值得说明。

第一，classify方法采用“先精确匹配异常类型，再靠错误文本兜底”的方式。这是因为在真实场景中，很多底层 SDK 不会抛我们自定义的异常，而是抛出json.JSONDecodeError、TimeoutError等基础异常，必须靠文本关键词去归类。

第二，watch方法并不知道 Agent 内部是怎么实现的，它只负责执行和监控。这种包装器模式可以在不改动原有 Agent 代码的情况下，给任意 Agent 加上检测能力，非常实用。

4.4 实现修复引擎

修复引擎负责根据失败类型选择修复动作，并使用带退避的循环重试来恢复任务。这里我不会过度设计，而是用一个可扩展的repair方法演示核心思路。

文件路径：agent_failure_monitor/repair_engine.py

import logging import time from typing import Any, Dict, Optional from agent_failure_types import AgentRunRecord, AgentFailureType, AgentState logger = logging.getLogger("agent.repair_engine") class RepairEngine: def __init__(self, max_repair_attempts: int = 3): self.max_repair_attempts = max_repair_attempts def repair( self, record: AgentRunRecord, agent_fn, **agent_kwargs: Any, ) -> Dict[str, Any]: """ 根据失败类型选择修复策略并重试。 返回结构: { "success": bool, "result": Any, "records": [AgentRunRecord, ...], "repair_actions": [str, ...], "recovered_attempt": Optional[int], } """ failure_type = record.failure_type repair_actions = [] all_records = [record] # 根据失败类型决定日志中的修复动作名 if failure_type == AgentFailureType.TOOL_CALL_PARSE_ERROR.value: repair_actions.append("retry_with_clean_instruction") elif failure_type == AgentFailureType.TOOL_EXECUTION_TIMEOUT.value: repair_actions.append("retry_with_timeout_extension") elif failure_type == AgentFailureType.CONTEXT_WINDOW_OVERFLOW.value: repair_actions.append("compress_context_and_retry") elif failure_type == AgentFailureType.OUTPUT_VALIDATION_FAILED.value: repair_actions.append("switch_to_structured_output") else: repair_actions.append("increase_max_attempts") # 使用指数退避进行重试 for attempt in range(1, self.max_repair_attempts + 1): wait_time = 1 * (2 ** (attempt - 1)) logger.info( "repair_start attempt=%d action=%s wait=%.1fs", attempt, repair_actions[0], wait_time, ) time.sleep(wait_time) result, new_record = self._safe_run( agent_fn, agent_id=record.agent_id, task=record.task, **agent_kwargs, ) all_records.append(new_record) if new_record.state == AgentState.FINISHED: logger.info( "repair_success attempt=%d agent_id=%s", attempt, record.agent_id, ) return { "success": True, "result": result, "records": all_records, "repair_actions": repair_actions, "recovered_attempt": attempt, } logger.error( "repair_failed agent_id=%s failure_type=%s", record.agent_id, failure_type, ) return { "success": False, "result": None, "records": all_records, "repair_actions": repair_actions, "recovered_attempt": None, } def _safe_run( self, agent_fn, agent_id: Optional[str] = None, task: Optional[str] = None, **agent_kwargs: Any, ) -> tuple: """ 借用检测器执行一次带监控的调用，避免在修复过程中丢失异常信息。 """ from failure_detector import AgentFailureDetector detector = AgentFailureDetector() return detector.watch( agent_fn, agent_id=agent_id or "agent-default", task=task or "", **agent_kwargs, )

这里需要注意：真实的修复策略一定比“重试”更复杂。比如 TOOL_CALL_PARSE_ERROR 可能需要重新生成 prompt 或修正输出格式；CONTEXT_WINDOW_OVERFLOW 可能需要压缩历史记录再继续。

本文示例之所以统一走重试，是为了演示修复引擎的框架。真实实现时，应该将if/else分支替换为策略对象或注册表，让每种失败类型有独立修复逻辑。

4.5 模拟 Agent 执行器

为了在本地演示，我写了一个SimulatedAgent，前两次执行总是抛出解析异常，第三次开始成功。

文件路径：agent_failure_monitor/simulated_agent.py

from agent_errors import ToolCallParseError class SimulatedAgent: def __init__(self, seed: int = 7): self.seed = seed self.run_count = 0 def run_once(self, prompt: str = "", **kwargs) -> dict: """ 模拟真实 Agent 的单次执行过程。 当前逻辑: - 第 1 次执行: 抛出 ToolCallParseError - 第 2 次执行: 抛出 ToolCallParseError - 第 3 次执行: 返回成功结果 这代表了一个可以通过有限重试恢复的失败场景。 """ self.run_count += 1 if self.run_count <= 2: raise ToolCallParseError( "failed to parse tool call: expected JSON array " "but got plain text" ) return { "prompt": prompt, "status": "success", "answer": "Agent 执行完成", "attempt": self.run_count, }

实际项目中，这个类的run_once方法内部会调用大模型，解析工具调用，执行外部服务。模拟类只是为了隔离这些外部依赖，让示例在任何机器上都能稳定跑通。

4.6 主程序串联执行

最后是main.py，负责把检测器和修复引擎串起来。

文件路径：agent_failure_monitor/main.py

import logging import sys from failure_detector import AgentFailureDetector from repair_engine import RepairEngine from simulated_agent import SimulatedAgent logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) def main(): agent = SimulatedAgent(seed=7) detector = AgentFailureDetector() repair_engine = RepairEngine(max_repair_attempts=3) agent_kwargs = { "prompt": "查询用户订单状态并生成总结", } # 第一次执行，结果可能是失败 result, record = detector.watch( agent.run_once, agent_id="order-agent-01", task="查询用户订单状态并生成总结", **agent_kwargs, ) print("首次执行结果:", result) print("首次执行状态:", record.state.value) print("首次失败类型:", record.failure_type) print() # 如果失败，进入自动修复流程 if record.state == "failed": repair_result = repair_engine.repair( record, agent.run_once, **agent_kwargs, ) if repair_result["success"]: print( "修复成功，第 {} 次重试后恢复".format( repair_result["recovered_attempt"] ) ) print("最终结果:", repair_result["result"]) else: print("修复失败，建议转人工处理") print() print("===== 所有执行记录 =====") for r in repair_result["records"]: print( "state={} failure_type={} duration={}s error={}".format( r.state.value, r.failure_type, r.duration(), r.error, ) ) if __name__ == "__main__": main()

4.7 运行验证

在项目目录下执行：

python main.py

预期输出类似：

2024-05-20 10:00:01 [WARNING] agent.failure_detector: agent_failed agent_id=order-agent-01 task=查询用户订单状态并生成总结 failure_type=tool_call_parse_error error=failed to parse tool call: expected JSON array but got plain text 首次执行结果: None 首次执行状态: failed 首次失败类型: tool_call_parse_error 2024-05-20 10:00:02 [INFO] agent.repair_engine: repair_start attempt=1 action=retry_with_clean_instruction wait=1.0s 2024-05-20 10:00:02 [WARNING] agent.failure_detector: agent_failed agent_id=order-agent-01 task=查询用户订单状态并生成总结 failure_type=tool_call_parse_error error=failed to parse tool call: expected JSON array but got plain text 2024-05-20 10:00:04 [INFO] agent.repair_engine: repair_start attempt=2 action=retry_with_clean_instruction wait=2.0s 2024-05-20 10:00:04 [INFO] agent.failure_detector: agent_finished agent_id=order-agent-01 task=查询用户订单状态并生成总结 duration=0.001s 修复成功，第 2 次重试后恢复 最终结果: {'prompt': '查询用户订单状态并生成总结', 'status': 'success', 'answer': 'Agent 执行完成', 'attempt': 3} ===== 所有执行记录 ===== state=failed failure_type=tool_call_parse_error duration=... error=... state=failed failure_type=tool_call_parse_error duration=... error=... state=finished failure_type=None duration=... error=None

第一次执行失败后，修复引擎自动做了两次重试，第二次重试成功。AgentRunRecord列表完整保留了每次执行的状态、失败类型和耗时信息，可以用于后续统计分析。

5. 常见问题与排查清单

5.1 常见问题对照表

问题现象	常见原因	解决思路
Agent 输出 JSON 解析失败	模型返回了被截断 JSON 或 Markdown 包裹	使用结构化输出；增加解析容错；失败后给模型补充格式示例
工具调用超时	外部服务不稳定，或单次调用时间过长	设置单次超时上限，采用指数退避重试
上下文窗口溢出	多轮工具结果累积过长	对历史记录做摘要压缩、滑动窗口裁剪
重试多次仍失败	任务本身不可解决，或提示词内部矛盾	及时终止重试，切换到人工介入或备选方案
检测日志不及时	日志异步写入延迟或采样比过高	使用内存缓冲队列 + 独立上报线程
修复后结果仍然异常	修复动作没有真正解决问题	复盘失败样本，将典型 case 加入回归测试集

5.2 排查步骤建议

当线上出现 Agent 相关问题时，我一般按下面顺序排查：

1. 先看执行记录里的state和failure_type，确定失败发生在哪一步。
2. 检查该次执行的error信息，结合上下文判断是模型输出问题还是工具问题。
3. 查看重试日志，确认修复引擎是否生效、重试次数是否合理。
4. 如果重试次数过多，说明修复策略本身需要优化。
5. 最后把失败样本单独保存，离线分析并设计回归用例。

6. 最佳实践与工程建议

6.1 监控数据要做到全链路打通

检测器输出的日志最好带上agent_id、task_id、trace_id等关联字段。否则当系统同时跑多个 Agent 任务时，很难判断某条异常日志属于哪个任务。生产环境中，可以把这些日志投递到 Loki、Elasticsearch 或云厂商日志服务，并配置基于失败率的告警。

6.2 失败类型需要归一化和分级

不同 Agent 模块可能抛出的异常五花八门。如果不对异常做归一化，后续统计和告警会非常困难。建议在检测器层统一把所有异常映射为标准枚举，并给失败类型设置等级。比如TOOL_EXECUTION_TIMEOUT可能只是临时抖动，属于低等级；RETRY_EXCEEDED说明系统已经无法自愈，属于高等级，需要立即通知人为介入。

6.3 修复动作应优先选择无副作用操作

重试是一种无副作用的修复动作，但“重新执行 SQL”“再次调用支付接口”这类操作并不一定安全。在设计修复策略时，必须考虑幂等性和调用方边界。对可能产生副作用的工具，要增加幂等键或人工确认步骤。

6.4 权限和合规边界要前置

为 Agent 配置工具权限时，应遵循最小权限原则。实时检测系统本身只负责监控和触发修复，不应绕过业务权限校验。生产环境中的任何自动重试动作，尤其是涉及写操作、