LLM Agent失败实时检测与自动修复：从事件流到多级策略的工程实现

前几天排查一个线上数据采集 Agent 的连环报错时，我盯着日志里反复出现的工具调用失败提示，突然意识到一个问题：我们一直在监控接口的 5xx、数据库的连接数、模型的响应延迟，却很少有人真正监控 Agent 的“行为轨迹”——它做了哪个决策、调用了哪个工具、拿到了什么结果、下一步打算干什么。

那次事故的经过并不复杂：Agent 调用了天气查询工具，因为城市参数格式不对，工具返回异常；Agent 没有感知到这个失败，继续按原计划执行下游任务，导致后续所有数据关联错误。更麻烦的是，监控大盘完全正常，因为模型调用成功、接口返回 200、延迟也不高。真正出错的，是 Agent 内部的决策链路。

这类问题在 LLM Agent 项目中越来越常见。本文围绕LLM Agent Failures（Agent 失败）的实时检测与修复，结合可运行的代码案例，梳理一套从失败分类、实时检测到自动化修复的工程实现方案。适合正在做 Agent 应用落地、想要提升系统稳定性的开发者阅读。读完你可以掌握：如何给 Agent 建立可观测事件流、如何用规则和信号识别失败、如何设计多级修复策略，以及生产环境下需要避开的坑。

1. LLM Agent 失败：到底是什么在出错

1.1 从“调用失败”到“任务失败”

传统的接口监控，关心的是“服务是否返回了正确响应”。但 LLM Agent 不是一次接口调用，它是一条“决策→行动→观察→再决策”的循环链路。一个 Agent 任务通常包含多轮 LLM 推理、多次工具调用、多段上下文记忆。

因此，LLM Agent 的失败不能简单等同于模型报错或服务宕机。它更像是一系列小步骤的累积偏差：模型选择了错误的工具、工具参数格式不合法、拿到工具结果后模型理解错误、上下文太长导致关键信息丢失、又或者在某个循环里反复执行同一个失败动作。

从工程视角看，可以把 Agent 失败分成两类：

• 可观测失败：工具抛异常、JSON 解析失败、超时、状态码非 2xx。这类失败能直接被捕获，也是传统监控最容易覆盖的部分。
• 隐性失败：模型输出了格式合法的结果，但方向是错的；工具调用成功，但参数与用户真实意图不符；Agent 虽然完成了流程，但结果质量不达标。这类失败很难用状态码去判断，往往需要语义层面的检测。

1.2 常见失败类型

结合目前主流的 Agent 设计模式，我通常把失败归纳为以下五类：

需要注意的是，这五类失败经常不是独立发生的。工具调用失败后，模型可能因为错误反馈而生成了更错误的答案，进而导致推理方向失败。这也是为什么“检测”必须放在“修复”之前，并且检测要尽可能早地介入。

1.3 为什么实时检测与修复是刚需

简单来说，Agent 的失败具有级联性和放大性。

一个普通的接口调用失败，影响范围通常局限在一次请求内。而 Agent 的一个错误决策，会沿着任务链路往下传导，影响后续所有步骤。错误被模型重新理解后，还可能演变成全新的、更难排查的问题。如果不能实时检测并在早期修复，最终可能导致：

• token 资源浪费：反复重试、反复生成无效计划。
• 任务结果不可信：用户拿到的是一个“看似完成、实际错误”的答案。
• 排查成本上升：多轮调用日志之间缺少关联，定位根因非常困难。
• 系统风险扩大：Agent 若被赋予执行类权限，错误动作可能影响业务数据。

所以，我们需要一套针对 Agent 运行轨迹的实时检测与修复机制，而不是简单地把 Agent 当成一个黑盒接口来监控。

2. 实时检测：让 Agent 的每一步都留下“事件”

2.1 检测的本质：从黑盒到白盒

要把 Agent 检测好，第一步不是选择模型或算法，而是建立“白盒化”的事件采集能力。也就是说，Agent 的每一步关键动作，都必须产生一条可记录、可追踪的事件。

一个标准的 Agent 事件，至少应该包含以下信息：

• trace_id：一次完整任务链路的唯一标识。
• step_id：当前步骤的标识，便于区分同一任务里的不同节点。
• agent_id：哪个 Agent 在执行。
• event_type：事件类型，比如decision、tool_call、tool_result、llm_result。
• action：模型选择的动作，比如调用哪个工具。
• payload：关键数据，比如工具参数、模型输出。
• timestamp：事件发生时间。
• metadata：可选字段，比如 token 消耗、模型名称、延迟。

有了这些信息，检测器才能判断一次 Agent 运行“是否正常”，而不是只看到最终结果。

2.2 Agent 运行时的事件采集

下面我们用一个简化但完整的 Python 实现，演示如何为 Agent 增加事件采集能力。这里不依赖任何第三方 Agent 框架，便于理解核心设计。

# 文件路径：agent_events.py import json import time import uuid from dataclasses import dataclass, field, asdict from typing import Any, Dict, Optional @dataclass class AgentEvent: """Agent 运行时产生的结构化事件""" trace_id: str step_id: str agent_id: str event_type: str action: str payload: Dict[str, Any] timestamp: float = field(default_factory=time.time) metadata: Dict[str, Any] = field(default_factory=dict) def to_json(self) -> str: return json.dumps(asdict(self), ensure_ascii=False) class EventBus: """轻量的事件总线，实际项目中可替换为消息队列或日志采集器""" def __init__(self): self._events = [] def emit(self, event: AgentEvent) -> None: self._events.append(event.to_json()) # 在实际项目中，可以在这里将事件发送到 Kafka、写入日志文件或推送到 APM 系统 print(f"[EVENT] {event.to_json()}") def all_events(self): return self._events

这里的EventBus是一个简化版本。真实项目中，emit方法内部一般会做三件事：

1. 把事件写入结构化日志，保留完整轨迹。
2. 把关键事件推送到监控系统，用于实时告警。
3. 把事件写入离线存储，用于后续的训练和评估。

整个 Agent 的每次工具调用、每次模型返回、每次状态变更，都应该调用emit产生一条事件。

2.3 基于规则的实时检测器设计

有了事件流之后，就可以设计检测器了。检测器通常分成三层：

第一层：语法层检测

这一层最直接，主要判断输出和参数是否满足格式要求。例如：

• 工具返回的 JSON 是否可解析。
• 模型输出的字段是否齐全。
• 参数类型是否正确。

第二层：运行层检测

这一层关注 Agent 的运行健康度。例如：

• 单次工具调用是否超时。
• 同一动作是否重复执行超过阈值。
• 整个任务链路是否超过最大步数。

第三层：语义层检测

这一层比较复杂，需要结合文本相似度、关键词匹配或额外的 LLM 评估。例如：

• 工具返回结果中是否包含错误关键字。
• 模型最终答案是否与用户问题主题偏离。
• 关键步骤是否满足业务定义的规则。

下面先实现语法层和运行层的检测器，语义层给出一个可扩展的接口。

# 文件路径：detector.py import json import re from typing import Any, Dict, List, Optional class DetectionResult: def __init__(self, is_failure: bool, failure_type: Optional[str] = None, detail: str = ""): self.is_failure = is_failure self.failure_type = failure_type self.detail = detail def __repr__(self): return f"DetectionResult(is_failure={self.is_failure}, failure_type={self.failure_type}, detail={self.detail})" class BaseDetector: """所有检测器的基类""" def detect(self, event: Dict[str, Any]) -> DetectionResult: raise NotImplementedError class ToolResultDetector(BaseDetector): """检测工具返回结果是否异常""" def __init__(self, max_retry_limit: int = 3): self.max_retry_limit = max_retry_limit def detect(self, event: Dict[str, Any]) -> DetectionResult: if event.get("event_type") != "tool_result": return DetectionResult(is_failure=False) payload = event.get("payload", {}) # 1. 检查工具是否明确返回错误 if payload.get("status") == "error": return DetectionResult( is_failure=True, failure_type="tool_error", detail=payload.get("message", "unknown tool error") ) # 2. 检查 HTTP 状态码 http_code = payload.get("http_code") if isinstance(http_code, int) and http_code >= 400: return DetectionResult( is_failure=True, failure_type="http_error", detail=f"HTTP status: {http_code}" ) # 3. 检查工具返回内容是否为合法 JSON raw_output = payload.get("output", "") if isinstance(raw_output, str) and raw_output.strip(): try: json.loads(raw_output) except json.JSONDecodeError as e: return DetectionResult( is_failure=True, failure_type="invalid_json", detail=f"Invalid JSON output: {e}" ) return DetectionResult(is_failure=False) class LoopDetectionDetector(BaseDetector): """检测 Agent 是否出现重复动作（死循环风险）""" def __init__(self, max_repeat_count: int = 3): self.max_repeat_count = max_repeat_count def detect(self, event: Dict[str, Any]) -> DetectionResult: action = event.get("action") if not action: return DetectionResult(is_failure=False) # 调用方负责传递历史动作列表 history = event.get("metadata", {}).get("action_history", []) repeat_count = sum(1 for item in history if item == action) if repeat_count >= self.max_repeat_count: return DetectionResult( is_failure=True, failure_type="loop_detected", detail=f"Action '{action}' repeated {repeat_count + 1} times" ) return DetectionResult(is_failure=False) class DetectionPipeline: """组合多个检测器，按顺序执行""" def __init__(self, detectors: List[BaseDetector]): self.detectors = detectors def run(self, event: Dict[str, Any]) -> DetectionResult: for detector in self.detectors: result = detector.detect(event) if result.is_failure: return result return DetectionResult(is_failure=False)

这里的核心思路是：每个检测器只负责一种失败模式的识别，DetectionPipeline按顺序执行所有检测器，一旦发现失败立即返回结果。这样做的好处是检测逻辑可以灵活扩展——你发现新的失败模式时，只需要新增一个 Detector 类，而不必修改主流程。

2.4 从事件到检测结果的闭环

有了事件流和检测器，还需要把两者串起来。我的做法是：Agent 在执行工具调用后，立即把tool_result事件交给DetectionPipeline检测；检测结果如果是失败，则进入修复模块；如果正常，则继续下一步。

这里有一个容易忽略的点：检测不仅要在工具返回后执行，还要在模型输出后执行。因为模型本身可能生成一个“看起来正常、实际错误”的决策。所以在真实项目中，我会在 LLM 输出解析处也加一道检测。

3. 修复：从重试到自愈的多级策略

3.1 修复不能是简单的“重试一次”

很多项目遇到 Agent 失败时，第一个想法是“加个重试机制”。但粗暴重试往往会带来更多问题：

• 工具不是幂等的，重试导致重复下单。
• 失败原因是模型规划错误，重试一万次结果都一样。
• 重试消耗大量 token，但成功率没有明显提升。

正确的做法，是根据失败类型选择修复策略。我把修复分成五个级别：

设计修复模块时，必须给每个修复策略设置“最大尝试次数”，避免无限循环。

3.2 修复管线的代码实现

下面用 Python 实现一个可配置的修复管线。RepairHandler负责根据检测到的failure_type决定执行哪种修复策略。

# 文件路径：repair.py import time from typing import Callable, Dict, Optional class RepairRequest: def __init__(self, trace_id: str, action: str, payload: Dict, failure_type: str, detail: str): self.trace_id = trace_id self.action = action self.payload = payload self.failure_type = failure_type self.detail = detail self.retry_count = 0 class RepairResult: def __init__(self, success: bool, output: Optional[Dict] = None, message: str = ""): self.success = success self.output = output self.message = message class RepairStrategy: """修复策略基类""" def can_handle(self, failure_type: str) -> bool: raise NotImplementedError def execute(self, request: RepairRequest, context: Dict) -> RepairResult: raise NotImplementedError class RetryStrategy(RepairStrategy): """L0：瞬时重试策略""" def __init__(self, max_retries: int = 2, delay_seconds: float = 0.5): self.max_retries = max_retries self.delay_seconds = delay_seconds def can_handle(self, failure_type: str) -> bool: return failure_type in ("http_error", "timeout", "rate_limit") def execute(self, request: RepairRequest, context: Dict) -> RepairResult: exec_func = context.get("executor") if not exec_func: return RepairResult(success=False, message="No executor found") while request.retry_count < self.max_retries: request.retry_count += 1 time.sleep(self.delay_seconds * request.retry_count) try: output = exec_func(request) if output and output.get("status") == "success": return RepairResult(success=True, output=output) except Exception as e: last_error = str(e) return RepairResult(success=False, message=f"Retry failed: {last_error}") class FeedbackRepairStrategy(RepairStrategy): """L2：模型反馈修复策略——把错误信息返回给 LLM，让它重新规划""" def __init__(self, max_retries: int = 2): self.max_retries = max_retries def can_handle(self, failure_type: str) -> bool: return failure_type in ("tool_error", "invalid_json", "invalid_parameters", "plan_error") def execute(self, request: RepairRequest, context: Dict) -> RepairResult: llm_reflect_func = context.get("llm_reflect") if not llm_reflect_func: return RepairResult(success=False, message="No reflect function") for _ in range(self.max_retries): new_plan = llm_reflect_func( trace_id=request.trace_id, action=request.action, payload=request.payload, error_detail=request.detail ) if new_plan and new_plan.get("is_valid"): return RepairResult(success=True, output=new_plan) return RepairResult(success=False, message="Feedback repair failed") class TerminateStrategy(RepairStrategy): """L4：终止任务策略""" def can_handle(self, failure_type: str) -> bool: return failure_type in ("loop_detected", "safety_risk") def execute(self, request: RepairRequest, context: Dict) -> RepairResult: # 实际项目中，这里会发送告警、记录日志、通知负责人 return RepairResult( success=False, message=f"Task terminated due to {request.failure_type}: {request.detail}" ) class RepairPipeline: """修复管线：根据失败类型选择修复策略""" def __init__(self): self.strategies = [ RetryStrategy(), FeedbackRepairStrategy(), TerminateStrategy(), ] def register_strategy(self, strategy: RepairStrategy) -> None: self.strategies.append(strategy) def repair(self, request: RepairRequest, context: Dict) -> RepairResult: for strategy in self.strategies: if strategy.can_handle(request.failure_type): return strategy.execute(request, context) return RepairResult(success=False, message=f"No strategy for {request.failure_type}")

这里有一个设计重点：修复策略和检测器一样，都遵循“单一职责 + 可插拔”的原则。新增一种失败场景时，你只需要新增一个RepairStrategy实现类，并注册到RepairPipeline中。

3.3 修复中的关键注意事项

在真实项目中，修复模块需要额外关注以下问题：

幂等性检查。在执行修复动作之前，要确认正在被修复的步骤是否具备幂等性。如果工具调用会写数据库、发消息、扣减库存，那重试之前必须确认上一次调用是否真的失败了。否则会出现“调用其实成功了，但响应超时，重试导致重复写入”的情况。

修复动作本身也要可观测。每次修复动作都要记录日志，包括：失败类型、选择了哪个修复策略、修复前后的状态、耗时等。这些数据可以用来评估修复策略的有效性，甚至在未来做动态策略选择。

不要让 Agent 自己无限修复。建议为整个任务设置一个“总修复次数”或“总延迟时间”的上限。一旦超过上限，不要再尝试修复，而是直接进入降级流程或通知人工处理。

4. 完整实战：构建一个“检测—修复”闭环 Agent

4.1 项目结构

下面我们构建一个简化但完整的示例项目，目录结构如下：

agent-demo/ ├── agent_events.py # 事件定义与事件总线 ├── detector.py # 检测器实现 ├── repair.py # 修复策略实现 ├── simple_agent.py # 主程序：模拟 Agent 执行链路 └── requirements.txt # 依赖（示例项目为标准库，无需安装额外依赖）

4.2 主程序：模拟 Agent 执行链路

主程序会模拟一个“天气查询 Agent”：用户询问“北京明天会下雨吗”，Agent 需要调用query_weather(city)工具获取天气数据，再基于结果生成回答。

为了演示完整闭环，我故意让第一次工具调用因参数格式错误而失败（city 参数传入了一个带非法字符的字符串），然后用FeedbackRepairStrategy把错误信息反馈给一个“模拟反思函数”，修正参数后重试成功。

# 文件路径：simple_agent.py import json import random import time from datetime import datetime from agent_events import AgentEvent, EventBus from detector import DetectionPipeline, ToolResultDetector, LoopDetectionDetector from repair import RepairPipeline, RepairRequest # ---------- 模拟工具：天气查询 ---------- def query_weather(city: str) -> dict: """模拟天气查询工具，城市名带非法字符时会返回错误""" # 校验城市参数 for ch in city: if not ('\u4e00' <= ch <= '\u9fff') and ch not in ('市', '省', '区'): return { "status": "error", "message": f"Invalid city name: {city}, 城市名只能包含中文字符" } # 模拟正常返回 weather_data = { "city": city, "date": datetime.now().strftime("%Y-%m-%d"), "condition": "小雨", "temperature": 18, "rain_probability": 85 } return {"status": "success", "data": weather_data} # ---------- 模拟 LLM 生成 ---------- def mock_llm_decision(user_query: str) -> dict: """模拟 LLM 决策：返回一个工具调用计划，首次生成的 city 参数带非法字符""" return { "action": "query_weather", "parameters": { "city": "北京@2024" } } def mock_llm_reflect(trace_id: str, action: str, payload: dict, error_detail: str) -> dict: """模拟反思函数：读取错误信息，修正参数后重新生成计划""" print(f" [REFLECT] 收到错误：{error_detail}") # 简单规则：从错误信息里提取错误城市参数的关键字 # 实际项目中，这里应该把错误信息拼接进 prompt，调用 LLM 生成新计划 error_msg = error_detail if "北京" in str(payload): # 模拟 LLM 修复：去掉非法字符 fixed_params = {"city": "北京"} return { "action": "query_weather", "parameters": fixed_params, "is_valid": True } return {"is_valid": False} # ---------- 执行工具调用 ---------- def execute_tool(action: str, parameters: dict) -> dict: """根据 action 分发到具体工具""" if action == "query_weather": return query_weather(parameters.get("city", "")) return {"status": "error", "message": f"Unknown action: {action}"} # ---------- Agent 主流程 ---------- def run_agent(user_query: str): trace_id = "trace_" + str(int(time.time() * 1000)) event_bus = EventBus() detection_pipeline = DetectionPipeline([ ToolResultDetector(), LoopDetectionDetector(max_repeat_count=2) ]) repair_pipeline = RepairPipeline() step_id = 0 # Step 1: LLM 生成决策 decision = mock_llm_decision(user_query) step_id += 1 event_bus.emit(AgentEvent( trace_id=trace_id, step_id=f"step_{step_id}", agent_id="weather_agent", event_type="decision", action=decision.get("action"), payload={"parameters": decision.get("parameters")} )) # Step 2: 执行工具调用，并检测结果 action = decision.get("action") parameters = decision.get("parameters", {}) action_history = [] for attempt in range(3): # 执行工具 tool_result = execute_tool(action, parameters) action_history.append(action) # 产生事件 step_id += 1 event_bus.emit(AgentEvent( trace_id=trace_id, step_id=f"step_{step_id}", agent_id="weather_agent", event_type="tool_result", action=action, payload=tool_result, metadata={"action_history": action_history} )) # 检测失败 detection_result = detection_pipeline.run({ "event_type": "tool_result", "action": action, "payload": tool_result, "metadata": {"action_history": action_history} }) if not detection_result.is_failure: print("\n== 工具调用成功 ==") print(json.dumps(tool_result, ensure_ascii=False, indent=2)) return tool_result print(f"\n== 检测到失败：{detection_result.failure_type} ==") print(f"失败详情：{detection_result.detail}") # 进入修复管线 repair_request = RepairRequest( trace_id=trace_id, action=action, payload=parameters, failure_type=detection_result.failure_type, detail=detection_result.detail ) repair_result = repair_pipeline.repair(repair_request, context={ "executor": lambda req: execute_tool(req.action, req.payload), "llm_reflect": mock_llm_reflect }) if repair_result.success: print("== 修复成功，使用修复后的参数重新执行 ==") if repair_result.output and "parameters" in repair_result.output: parameters = repair_result.output["parameters"] elif repair_result.output and "data" in repair_result.output: return repair_result.output else: print(f"== 修复失败：{repair_result.message} ==") break return None if __name__ == "__main__": user_query = "北京明天会下雨吗" result = run_agent(user_query) if result is None: print("\n任务失败，进入人工处理流程")

4.3 运行与预期结果

在命令行执行：

cd agent-demo python simple_agent.py

预期输出如下：

[EVENT] {"trace_id": "trace_1699000000000", "step_id": "step_1", "agent_id": "weather_agent", "event_type": "decision", "action": "query_weather", "payload": {"parameters": {"city": "北京@2024"}}, "timestamp": 1699000000.0, "metadata": {}} [EVENT] {"trace_id": "trace_1699000000000", "step_id": "step_2", "agent_id": "weather_agent", "event_type": "tool_result", "action": "query_weather", "payload": {"status": "error", "message": "Invalid city name: 北京@2024, 城市名只能包含中文字符"}, "timestamp": 1699000000.01, "metadata": {"action_history": ["query_weather"]}} == 检测到失败：tool_error == 失败详情：Invalid city name: 北京@2024, 城市名只能包含中文字符 [REFLECT] 收到错误：Invalid city name: 北京@2024, 城市名只能包含中文字符 == 修复成功，使用修复后的参数重新执行 == [EVENT] {"trace_id": "trace_1699000000000", "step_id": "step_3", "agent_id": "weather_agent", "event_type": "tool_result", "action": "query_weather", "payload": {"status": "success", "data": {"city": "北京", "date": "2024-11-01", "condition": "小雨", "temperature": 18, "rain_probability": 85}}, "timestamp": 1699000000.02, "metadata": {"action_history": ["query_weather", "query_weather"]}} == 工具调用成功 == { "status": "success", "data": { "city": "北京", "date": "2024-11-01", "condition": "小雨", "temperature": 18, "rain_probability": 85 } }

这个示例完整展示了“决策 → 执行 → 事件采集 → 失败检测 → 修复 → 重新执行 → 成功”的闭环链路。

4.4 如何接入真实 LLM 场景

在上面的示例中，mock_llm_decision和mock_llm_reflect是模拟函数。接入真实项目时，你只需要替换这两部分：

• mock_llm_decision：替换为真实 LLM 调用，并加入结构化输出解析（比如 function calling 或 JSON mode）。
• mock_llm_reflect：把error_detail和原始 payload 拼进 prompt，要求模型重新生成参数或调整计划。

例如，mock_llm_reflect的真实实现大致是：

def llm_reflect(trace_id: str, action: str, payload: dict, error_detail: str) -> dict: prompt = f""" 你是一个 Agent 规划器。上一次工具调用失败，失败信息如下： 工具名称: {action} 调用参数: {json.dumps(payload, ensure_ascii=False)} 错误信息: {error_detail} 请重新生成一个可用的工具调用计划，以 JSON 格式返回，包含 action 和 parameters 字段。 """ # 调用真实 LLM，并解析返回的 JSON response = call_llm(prompt) return json.loads(response)

这里不绑定具体的 LLM SDK，因为不同项目的模型接入方式差异很大。核心思路是：让模型根据反馈重新规划一次，而不是简单地重试原参数。

5. 常见问题与排查思路

5.1 常见问题速查表

5.2 排查 Agent 失败问题的清单

当线上 Agent 出现异常时，我建议按下面的顺序排查：

1. 先看 trace。找到对应任务的trace_id，把整条事件链路拉出来。确认失败发生在哪个 step。
2. 判断失败类型。是工具返回错误、参数格式错误，还是模型规划错误？不要把所有问题都归为“模型问题”。
3. 确认是否幂等。如果失败步骤可能产生副作用，先确认副作用是否已经发生。
4. 检查修复策略是否生效。修复日志里有没有对应记录？修复成功后，是参数被修正了，还是仅仅重试碰巧成功了？
5. 评估是否需要人工介入。如果一条任务在自动修复后仍然失败，要触发人工告警，而不是静默重试。

6. 最佳实践与工程建议

6.1 检测优先，修复其次

很多团队在建设 Agent 稳定性时，一上来就搞自动修复，这是不合理的。如果连失败都检测不到，修复就无从谈起。建议先花时间把事件采集和检测规则做扎实，再逐步迭代修复策略。

6.2 修复策略要可配置、可回滚

不要把所有修复逻辑写死在代码里。推荐用配置中心或配置文件管理修复策略的参数，比如最大重试次数、重试间隔、哪些工具禁止自动重试等。这样即使线上出问题，也能快速调整，不需要发版。

6.3 让 Agent 的每一步都有“副作用意识”

Agent 调用工具时，要明确区分“读操作”和“写操作”：

• 读操作：可以放心重试。
• 写操作：重试前必须做幂等校验，或者在业务层面保证幂等性。
• 高风险操作：例如删除、转账、变更状态，建议默认关闭自动修复，改为人工确认。

6.4 定期评估修复方案的效果

修复策略不是越多越好。可以给每次修复打上标签，统计每个策略的修复成功率、修复耗时、修复后任务最终成功率。用数据淘汰无效的修复策略，保留真正有价值的策略。

6.5 注意反馈信息的长度

在使用 L2 模型反馈修复时，如果错误信息太长，会占用大量上下文空间。建议对错误信息做摘要，只保留关键字段，比如：

• 错误类型码
• 涉及的工具名
• 最核心的错误描述
• 原始参数中的关键字段

这样可以减少 token 消耗，也能让模型更容易抓住重点。

6.6 安全边界：不要让修复动作变成攻击入口

Agent 自动修复意味着程序会根据错误信息自动调整参数和行动。如果此时攻击者注入了一段恶意指令，修复环节可能会放大风险。建议：

• 对工具参数做白名单校验。
• 对修复后的计划做一次安全规则扫描。
• 高风险工具不允许进入自动修复流程。

7. 总结与下一步

本文围绕 LLM Agent 失败这个主题，从概念、失败类型、事件采集、实时检测、多级修复到完整示例，梳理了一整套工程化实现思路，核心可以总结为几点：

• Agent 失败不能只靠状态码判断，必须从决策、工具、上下文、输出格式等多个维度采集事件并检测。
• 检测器要按“语法层→运行层→语义层”分层设计，每一层只负责一种失败模式。
• 修复策略要按失败类型动态选择，严格限制重试次数，并对写操作保持警惕。
• 检测和修复的每一个动作都要留下结构化日志，这是后续评估和优化的基础。

如果你正在建设自己的 Agent 应用，可以照着本文的代码搭一个最小可用的检测与修复框架，然后逐步补充真实 LLM 调用和工具接口。

下一步建议做三件事：第一，为你的 Agent 设计一套完整的事件字段规范；第二，用“失败注入”的方式主动制造几种典型失败，验证检测器能否正确识别；第三，统计一周内所有线上 Agent 任务的失败率和修复成功率，用真实数据驱动策略优化。把这四步走完，你的 Agent 稳定性体系基本就立住了。