最近在做大模型 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 执行循环可以拆成以下几步接收任务。对任务进行规划拆成多个步骤。针对当前步骤生成工具调用指令。执行工具并获取结果。把工具结果放回上下文继续下一步。全部步骤完成后生成最终回复。每一步都可能出现不同类型的失败。例如第 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.pyfrom 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_factorytime.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.pyclass 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 函数并自动完成如下工作创建运行记录。执行 Agent 函数。捕获所有异常。将异常映射为规范化失败类型。输出结构化日志。返回执行结果与运行记录。文件路径agent_failure_monitor/failure_detector.pyimport 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_idagent_id, tasktask) 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.pyimport 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_idrecord.agent_id, taskrecord.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_idagent_id or agent-default, tasktask or , **agent_kwargs, )这里需要注意真实的修复策略一定比“重试”更复杂。比如 TOOL_CALL_PARSE_ERROR 可能需要重新生成 prompt 或修正输出格式CONTEXT_WINDOW_OVERFLOW 可能需要压缩历史记录再继续。本文示例之所以统一走重试是为了演示修复引擎的框架。真实实现时应该将if/else分支替换为策略对象或注册表让每种失败类型有独立修复逻辑。4.5 模拟 Agent 执行器为了在本地演示我写了一个SimulatedAgent前两次执行总是抛出解析异常第三次开始成功。文件路径agent_failure_monitor/simulated_agent.pyfrom 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.pyimport logging import sys from failure_detector import AgentFailureDetector from repair_engine import RepairEngine from simulated_agent import SimulatedAgent logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, handlers[logging.StreamHandler(sys.stdout)], ) def main(): agent SimulatedAgent(seed7) detector AgentFailureDetector() repair_engine RepairEngine(max_repair_attempts3) agent_kwargs { prompt: 查询用户订单状态并生成总结, } # 第一次执行结果可能是失败 result, record detector.watch( agent.run_once, agent_idorder-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_idorder-agent-01 task查询用户订单状态并生成总结 failure_typetool_call_parse_error errorfailed 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 attempt1 actionretry_with_clean_instruction wait1.0s 2024-05-20 10:00:02 [WARNING] agent.failure_detector: agent_failed agent_idorder-agent-01 task查询用户订单状态并生成总结 failure_typetool_call_parse_error errorfailed to parse tool call: expected JSON array but got plain text 2024-05-20 10:00:04 [INFO] agent.repair_engine: repair_start attempt2 actionretry_with_clean_instruction wait2.0s 2024-05-20 10:00:04 [INFO] agent.failure_detector: agent_finished agent_idorder-agent-01 task查询用户订单状态并生成总结 duration0.001s 修复成功第 2 次重试后恢复 最终结果: {prompt: 查询用户订单状态并生成总结, status: success, answer: Agent 执行完成, attempt: 3} 所有执行记录 statefailed failure_typetool_call_parse_error duration... error... statefailed failure_typetool_call_parse_error duration... error... statefinished failure_typeNone duration... errorNone第一次执行失败后修复引擎自动做了两次重试第二次重试成功。AgentRunRecord列表完整保留了每次执行的状态、失败类型和耗时信息可以用于后续统计分析。5. 常见问题与排查清单5.1 常见问题对照表问题现象常见原因解决思路Agent 输出 JSON 解析失败模型返回了被截断 JSON 或 Markdown 包裹使用结构化输出增加解析容错失败后给模型补充格式示例工具调用超时外部服务不稳定或单次调用时间过长设置单次超时上限采用指数退避重试上下文窗口溢出多轮工具结果累积过长对历史记录做摘要压缩、滑动窗口裁剪重试多次仍失败任务本身不可解决或提示词内部矛盾及时终止重试切换到人工介入或备选方案检测日志不及时日志异步写入延迟或采样比过高使用内存缓冲队列 独立上报线程修复后结果仍然异常修复动作没有真正解决问题复盘失败样本将典型 case 加入回归测试集5.2 排查步骤建议当线上出现 Agent 相关问题时我一般按下面顺序排查先看执行记录里的state和failure_type确定失败发生在哪一步。检查该次执行的error信息结合上下文判断是模型输出问题还是工具问题。查看重试日志确认修复引擎是否生效、重试次数是否合理。如果重试次数过多说明修复策略本身需要优化。最后把失败样本单独保存离线分析并设计回归用例。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 配置工具权限时应遵循最小权限原则。实时检测系统本身只负责监控和触发修复不应绕过业务权限校验。生产环境中的任何自动重试动作尤其是涉及写操作、