LLM Agent 在本地跑通一个 Demo 往往很简单真正难的是让它连续执行几个小时不出问题。工具调用失败、上下文溢出、循环卡死、API 超时、模型返回格式异常、权限报错——这些故障不是“小概率事件”而是 Agent 进入生产环境后的常态。Real-Time Detection and Repair of LLM Agent Failures 要解决的核心问题就是在 Agent 执行过程中实时感知异常自动完成故障定位、修复决策和执行恢复让整个系统从“容易挂”变成“挂了能自己爬起来”。这套方案的关键不是让模型更强而是把 Agent 执行过程改造为可观测、可控制、可恢复的工程系统。文章会从故障类型与根因分析、系统总体架构、检测层实现、修复层实现、接口接入与批量任务、资源占用观察、常见问题排查和最佳实践几个角度展开核心代码给出可运行的示例。读者可以基于这套结构自建 LLM Agent 检测与修复服务也可以把它接入已有的 LangChain、LlamaIndex 或自研 Agent 框架。适合的读者有三类正在做 Agent 应用开发的 Python 工程师需要把 LLM Agent 接入生产系统的后端或运维同学以及被“Agent 执行中断、日志里又没有有效信息”折磨过的实践者。本文不依赖特定模型OpenAI 兼容接口、本地 Ollama/vLLM 服务都可以使用同样的检测逻辑。1. 核心能力速览先给结论。本文描述的是一套可复用的 LLM Agent 故障检测与自愈参考架构不是某个绑定特定框架的闭源插件。它的核心能力集中在“执行可观测、故障可检测、异常可修复、任务可重放”四个层面。能力项说明面向对象基于 LLM 的自主 Agent 系统支持多工具调用场景故障覆盖范围工具调用失败、LLM 返回异常、上下文溢出、循环执行、超时、权限错误、提示注入风险检测方式日志采集 状态指标 行为规则 异常计数熔断修复策略自动重试、提示词修正、降级、回退、重新规划、人工接管接口能力HTTP JSON 接口提交任务、查询状态、获取修复记录批量任务支持任务队列、并发控制、失败重放部署形态独立 Python 服务可嵌入现有 Agent 框架也可作为旁路监控模型兼容兼容 OpenAI 兼容接口与本地推理服务硬件门槛检测与修复逻辑本身 CPU 即可运行LLM 推理部分取决于所选模型适合场景Agent 生产环境稳定性建设、批量任务自动化、故障复盘与日志归档这里要强调检测与修复引擎本身不承担 LLM 推理它的资源消耗很小。真正决定显存和算力需求的是 Agent 内部调用的基础模型和工具服务。文章不虚构某一款显卡的显存占用数据实际部署时建议按“本地模型版本 推理参数 并发数”三个变量观测。2. LLM Agent 常见故障类型与根因分析要做实时检测第一步不是写代码而是把故障分类清楚。从工程实践看LLM Agent 的故障可以按层划分LLM 层、工具调用层、执行层、安全合规层。每一层有自己的检测特征和修复方向。2.1 LLM 层故障LLM 层的典型故障包括模型返回 JSON 格式非法、工具参数生成错误、模型进入重复输出、上下文窗口溢出、拒绝回答或幻觉式回答。这些故障的共同特点是模型没有抛异常但返回内容不符合 Agent 执行要求。检测层必须对模型输出做结构化校验比如检查必须字段是否存在、参数类型是否正确、输出长度是否异常。修复方向通常是修正提示词后重试或者把非法输出交给解析器重新清洗。2.2 工具调用层故障工具层故障是最常见的故障类型。典型场景包括外部 API 超时、请求参数类型不匹配、鉴权过期、工具返回结构不符合预期、工具本身执行失败。这类故障往往伴随明确异常码检测逻辑相对直接但需要区分“暂时性故障”和“永久性故障”。暂时性故障可以等待后重试永久性故障必须立即修改参数或切换工具否则无限重试只会浪费时间和费用。2.3 执行层故障执行层故障属于过程性异常典型表现包括Agent 在同一个工具上反复调用、步骤推进缓慢、工具调用顺序错误、上下文积累导致任务偏离原始目标。这类故障不能靠单次运行日志发现需要在检测器中维护“执行轨迹摘要”。比如记录最近 N 步的调用函数名和输入参数哈希如果出现循环重复直接判定为循环故障并中断。2.4 安全合规层故障安全层故障容易被忽略但生产环境一旦触发代价很高。典型情况包括用户通过提示注入让 Agent 执行未授权操作、Agent 拿到了超出最小权限范围的文件或接口、工具参数中包含敏感信息被写入日志。这类故障不能等它发生后再修复建议在检测层加入敏感词过滤、权限校验和输出脱敏规则。涉及用户数据、人脸、声音、版权素材等场景必须确认授权和合规边界。3. 实时检测与修复系统架构整个系统可以分成五层Agent 执行层、可观测性采集层、实时检测层、修复决策层、控制面板与 API 层。五层之间通过内部事件通信不要求所有组件运行在同一进程内。Agent 执行层是业务主体。它负责接收任务、调用 LLM、调用外部工具并返回最终结果。检测系统不要修改 Agent 内部每个函数而是通过埋点或日志钩子拿到执行轨迹。可观测性采集层负责收集三类数据运行日志、状态指标、工具调用记录。状态指标包括当前任务阶段、已执行步数、最近一次 LLM 调用耗时、连续失败次数等。实时检测层消费这些数据套用规则引擎和异常检测模型输出故障事件。修复决策层根据故障事件选择合适的修复策略并把修复结果回传执行层。API 层负责对外提供任务提交、状态查询和修复记录查询。这个架构的核心原则是“检测与执行解耦”。检测器不关心 Agent 具体调用了哪个模型只关心行为特征修复器不直接重写 Agent 业务代码而是通过重试、替换参数、切换提示词等标准操作干预。这样做的好处是当业务方调整 Agent 内部工具时检测与修复服务不需要大改。下面用一个最小可运行的状态机示例说明检测流程。这段代码不是完整生产实现但可以直观表达“采样—判断—修复”的循环节奏。# guard_state_machine.py # 最小检测状态机示例运行 - 检测 - 修复 - 重跑 - 完成/终止 from enum import Enum from dataclasses import dataclass, field from typing import Optional class AgentState(str, Enum): RUNNING running DETECTED detected REPAIRING repairing RE_RUN re_run COMPLETED completed TERMINATED terminated dataclass class TaskState: task_id: str state: AgentState AgentState.RUNNING attempts: int 0 max_attempts: int 3 last_error: Optional[str] None trace: list field(default_factorylist) def can_retry(self) - bool: return self.attempts self.max_attempts状态机把 Agent 的一次任务划分为六个状态。检测层发现异常后任务进入 DETECTED修复决策层根据错误类型选择策略进入 REPAIRING如果策略是修正后再跑则回到 RE_RUN如果连续多次失败则置为 TERMINATED。通过状态字段可以方便地上报给外部监控系统。4. 环境准备与前置条件部署这套检测与修复服务本身不复杂但为了完整跑通测试建议准备以下环境。组件建议要求说明操作系统Linux / macOS / WindowsLinux 服务器部署更稳定Python3.10 及以上用虚拟环境隔离依赖LLM 服务OpenAI 兼容接口或本地 Ollama/vLLM检测逻辑与具体模型无关任务队列Redis 或内存队列批量任务场景建议使用 Redis日志存储JSON 文件或 Elasticsearch用于故障复盘外部工具待测 API、数据库、脚本按业务场景接入安装 Python 依赖时建议用虚拟环境。核心依赖包括 FastAPI、uvicorn、pydantic、httpx 或 requests。如果只是做检测层验证不启动 Web 服务那么只需要 httpx 和日志库。# 创建虚拟环境 python3 -m venv .venv source .venv/bin/activate # 安装基础依赖 pip install fastapi uvicorn pydantic requests redis如果 Agent 使用本地模型推理需要额外确认推理服务已经启动并且端口可以被本机访问。比如 Ollama 默认监听 11434vLLM 默认监听 8000。检测服务只需要能向模型服务发送 HTTP 请求即可不需要读取模型权重。5. 检测层实现检测层是整个系统最核心的部分。它的目标是回答三个问题当前任务是否异常、异常属于哪个类型、异常是否达到触发修复的阈值。5.1 异常检测循环一个简单的检测循环可以放在 Agent 执行函数的外层包装执行过程并捕获异常。下面给出一个可运行的 AgentGuard 示例它同时处理工具异常、格式异常和未知异常并带连续失败熔断。# agent_guard.py import time import logging from typing import Any, Callable logger logging.getLogger(agent_guard) class AgentToolError(Exception): 工具调用异常。 class AgentFormatError(Exception): 模型返回格式异常。 class AgentLoopError(Exception): 检测到执行循环。 class AgentRunError(Exception): 任务运行失败。 class AgentCircuitBreak(Exception): 连续失败次数过多熔断。 class AgentGuard: def __init__(self, execute_fn: Callable[[str], Any], max_retry: int 3, circuit_threshold: int 5): self.execute_fn execute_fn self.max_retry max_retry self.circuit_threshold circuit_threshold self.fail_count 0 def run(self, task: str) - Any: for attempt in range(1, self.max_retry 1): try: result self.execute_fn(task) self.fail_count 0 return result except AgentToolError as exc: logger.warning(工具调用失败 attempt%s error%s, attempt, exc) time.sleep(2 ** attempt) except AgentFormatError as exc: logger.warning(模型返回格式异常 attempt%s error%s, attempt, exc) task self.repair_prompt(task, exc) except AgentLoopError as exc: logger.error(检测到执行循环 attempt%s error%s, attempt, exc) raise except Exception as exc: logger.exception(未知异常 attempt%s error%s, attempt, exc) self.fail_count 1 if self.fail_count self.circuit_threshold: raise AgentCircuitBreak(连续失败次数过多熔断) from exc raise AgentRunError(重试次数用尽) def repair_prompt(self, task: str, exc: AgentFormatError) - str: # 示例修复策略在原始任务后追加格式要求 # 实际项目中可以根据 exc 信息重写提示词 return task \n请严格按照 JSON 格式返回不要输出额外说明。这个实现的关键点是工具类异常先等指数退避再重试格式类异常通过追加提示词修正循环异常直接终止。未知异常累加计数器超过阈值触发熔断。这种模式可以防止在某个外部服务长期不可用时反复消耗 LLM 调用费用。5.2 执行轨迹与循环检测要检测循环执行仅靠异常捕获不够。很多循环故障表现为“调用仍然成功但工具参数和结果越来越相似”。需要在执行层维护一个轨迹缓冲区记录最近 N 次工具调用摘要并对比相似度。下面是一个轻量轨迹记录器。# trace_recorder.py from collections import deque from typing import Any class TraceRecorder: def __init__(self, max_len: int 20): self.trace deque(maxlenmax_len) def record(self, tool_name: str, tool_args: dict, result_sign: str): self.trace.append({ tool: tool_name, args: tool_args, sign: result_sign, }) def detect_loop(self, threshold: int 5) - bool: if len(self.trace) threshold * 2: return False recent list(self.trace)[-threshold:] previous list(self.trace)[-(threshold * 2):-threshold] return all(a[tool] b[tool] and a[sign] b[sign] for a, b in zip(recent, previous))轨迹记录器只保存摘要不保存完整输入输出可以减少内存占用和隐私暴露。如果最近 N 次工具调用与前 N 次完全一致检测器可以判定为循环并通知修复引擎切换策略。6. 修复层实现检测到故障事件后修复层要决定“怎么救、救几次、什么时候放弃”。修复策略的选择应该基于故障类型和当前任务状态而不是一味重试。6.1 修复策略表故障类型推荐策略说明工具暂时性超时指数退避重试间隔 2s、4s、8s最多 3 次工具永久性错误参数修正或切换工具检查请求参数是否合法LLM 返回格式非法提示词修正后重跑要求模型按固定 JSON Schema 输出上下文过长截断历史重新规划丢弃低优先级上下文保留任务目标连续失败过多熔断并人工接管避免费用失控和死循环权限不足直接终止并告警不要绕过权限限制6.2 修复策略分发器修复器可以抽象为统一的接口。每个策略实现相同的 handle 方法返回 Decision 表示处理结果重试、重规划、终止。# repair_strategy.py from typing import Callable from dataclasses import dataclass dataclass class RepairDecision: action: str # retry / replan / terminate reason: str class RetryStrategy: def handle(self, task: str, error: Exception, attempt: int) - RepairDecision: if attempt 3: return RepairDecision(replan, 重试次数用尽需要重新规划) return RepairDecision(retry, 暂时性故障继续重试) class PromptRepairStrategy: def handle(self, task: str, error: Exception, attempt: int) - RepairDecision: if attempt 2: return RepairDecision(terminate, 多次格式修复无效) return RepairDecision(retry, 修正提示词后重试) STRATEGY_MAP { AgentToolError: RetryStrategy(), AgentFormatError: PromptRepairStrategy(), } def dispatch_repair(task: str, error: Exception, attempt: int) - RepairDecision: strategy STRATEGY_MAP.get(type(error)) if strategy is None: return RepairDecision(replan, 未知异常建议重新规划) return strategy.handle(task, error, attempt)这里的策略选择逻辑非常透明方便运维在故障复盘时回答“为什么重试了”“为什么终止了”。实际生产环境还可以把策略表放到配置中心通过接口动态调整重试次数而不需要重新发布代码。7. 接口 API 与批量任务检测与修复服务要接入现有系统最好提供 HTTP JSON 接口。下面给出一个基于 FastAPI 的最小示例包含提交任务和查询状态的接口。# api_server.py import uuid from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class TaskSubmit(BaseModel): task_id: str prompt: str max_retry: int 3 class TaskStatus(BaseModel): task_id: str state: str attempts: int error: str tasks {} app.post(/task/submit) def submit_task(req: TaskSubmit): task_id req.task_id or str(uuid.uuid4()) tasks[task_id] TaskStatus(task_idtask_id, staterunning, attempts0) # 实际场景这里把任务交给 AgentGuard 执行并异步更新状态 return {task_id: task_id, state: running} app.get(/task/{task_id}) def query_task(task_id: str): if task_id not in tasks: raise HTTPException(status_code404, detailtask not found) return tasks[task_id] app.get(/repair/records) def repair_records(): # 生产环境建议分页查询这里只返回数量 return {total: len(tasks)}接口设计遵循简单原则提交任务返回 task_id之后通过任务 ID 查询状态。实际项目中建议把 tasks 字典替换为 Redis并让 AgentGuard 在状态变化时写回 Redis。7.1 批量任务与失败重放批量任务是 Agent 生产环境的刚需。比如每天要处理 5000 篇文档的摘要、图片描述、日志分类。批量任务场景下检测与修复系统要额外支持队列、并发限制和失败重放。# batch_queue.py import queue import threading import time class BatchRunner: def __init__(self, worker_num: int 4): self.task_queue queue.Queue() self.workers [] self.worker_num worker_num self.start_workers() def start_workers(self): for _ in range(self.worker_num): t threading.Thread(targetself.consume, daemonTrue) t.start() self.workers.append(t) def submit(self, task: dict): self.task_queue.put(task) def consume(self): while True: task self.task_queue.get() task_id task[task_id] try: print(f处理任务 {task_id}) # 这里调用 AgentGuard.run except Exception as exc: print(f任务 {task_id} 失败: {exc}) finally: self.task_queue.task_done()批量任务的关键是“失败不丢”。建议给每个任务记录状态、重试次数和错误信息并通过定时任务扫描处于失败状态的任务做重放。重放时需要注意幂等性避免同一个外部操作被执行两次。比如调用支付接口、发送邮件这类不可逆操作必须在任务中标记 operation_id由外部服务做去重。8. 资源占用与性能观察检测与修复服务本身是轻量级的但生产环境仍然需要关注几个性能指标。第一LLM 调用次数与费用。Agent 故障检测与修复越激进LLM 调用次数越多。每次重试都会产生新的模型请求如果使用付费 API费用会线性增长。建议在检测器中记录每次 LLM 调用的 token 数和耗时并设置单任务调用上限。第二日志量与轨迹存储量。可观测性采集层如果记录全部输入输出磁盘和内存都会快速膨胀。建议只保存工具参数摘要、返回结果哈希和错误类型。完整数据可以在必要时单独存档。第三检测频率对性能的影响。检测循环如果每 100ms 扫描一次所有任务在任务量大时会带来无谓的 CPU 开销。更稳妥的做法是事件驱动Agent 执行层主动上报状态变更检测器只处理变更事件而不是轮询全部任务。第四本地模型推理场景下的显存和延迟。检测与修复服务可以运行在 CPU 机器上但 Agent 内部的 LLM 推理如果使用本地模型显存需求由模型大小和并发数决定。建议先用单并发测试一轮再逐步提高并发观察推理服务的平均延迟和显存占用。不要盲信网上某个“6G 显存可跑”的说法实际占用要按本机模型版本和序列长度测量。第五进程与端口管理。FastAPI 服务默认端口是 8000如果与 vLLM 或其他服务冲突需要手动指定端口。# 启动检测服务指定端口 uvicorn api_server:app --host 127.0.0.1 --port 90009. 常见问题与排查方法下面是部署和运行这套系统时最容易遇到的问题。排查思路以实际日志为准表格可以作为第一轮定位的参考。问题现象可能原因排查方式解决方案Agent 反复调用同一个工具检测器没有启用循环检测查看执行轨迹记录开启 TraceRecorder 并配置循环阈值修复后仍然失败故障属于永久性错误查看错误码和请求参数切换工具或修改参数不要继续重试Agent 执行被终止提示 agent execution terminated due to errorAgent 层异常未捕获查看完整异常堆栈在外层包装 AgentGuard捕获未知异常接口响应超时外部工具或模型服务响应慢检查上游服务耗时增加超时时间或采用异步任务批量任务卡住单线程消费者处理耗时任务查看队列积压数提高消费者数量或拆分任务粒度API 调用失败鉴权失效或模型服务未启动检查 API Key 和服务端口更新鉴权信息确认模型服务健康上下文溢出多轮工具调用历史过长查看 token 统计定期裁剪历史只保留最近几轮和任务目标本地模型显存不足并发数过高或序列过长查看推理服务日志降低并发、缩短输入长度或换更大显存机型在接入真实业务时还经常遇到“harness 与 agent 的区别”这类框架层问题。简单说harness 是承载 agent 运行的控制逻辑包括任务循环、中断处理、恢复机制agent 是具体承担推理和工具决策的部分。检测与修复系统应当挂在 harness 层而不是 agent 内部。这样无论底层 agent 如何更换上层的观测和恢复机制都能复用。另一个常见提示是“the agent execution provider did not respond in time”。这通常说明执行提供方超时比如 LLM API 或工具服务耗时过长。排查重点在于确认是模型推理慢、网络延迟还是工具本身阻塞。如果是模型推理慢优先优化提示词长度和采样参数如果是工具阻塞需要给工具调用加上明确的超时控制。10. 最佳实践与使用建议第一第一次接入先做小流量验证。不要直接把所有业务任务都交给检测修复系统应该选取 10 到 20 个典型任务观察故障被正确识别和修复的比例确认没有误杀后再扩大范围。第二保留一套最小可运行配置。把 AgentGuard、TraceRecorder、RepairStrategy 这三个核心组件的配置固化下来保证新环境可以快速复现。模型、工具、外部服务都可以变化但这三个组件的最小逻辑不能丢。第三修复操作必须幂等。自动修复越强越要小心副作用。重试一个无副作用的查询接口没有问题但重试一个下单接口可能产生重复订单。建议所有外部操作接口要求操作方传 request_id由服务端去重。第四日志和状态要分层。error 级日志只记录真正的故障事件info 级记录状态变化debug 级记录完整输入输出。不要把整段用户输入都写在 error 日志中既影响排查效率也可能泄露隐私信息。第五涉及人脸、声音、版权素材、个人数据或内部系统权限时自动修复必须设置更保守的边界。比如自动重试不得绕过权限校验禁止通过修改提示词让模型输出未授权内容。安全类故障应当立即终止并通知人工而不是自动“创造条件再试一次”。第六批量任务要加日志和失败重试机制。建议每个任务记录 start_time、end_time、retry_count、error_type并定时清理超过保留周期的任务数据。这样既方便问题复盘也避免存储无限增长。第七发布或商用前做效果复核。检测修复系统本身也可能出错。上线前准备一组带标注的故障样例覆盖超时、格式错误、循环、权限不足等场景用这组样例回归检测准确率再决定是否投入生产。11. 总结Real-Time Detection and Repair of LLM Agent Failures 值得优先验证的部分不是某个花哨的检测模型而是“可靠地拿到执行轨迹”和“正确地复用重试与熔断机制”。很多 Agent 项目不是模型能力不够而是在故障发生后缺乏标准的干预渠道。只要把执行轨迹、异常类型、修复策略和任务状态管理起来稳定性就能提升一大截。最容易踩的坑是过度依赖“让模型自己恢复”。LLM 对自身错误的修正能力有限尤其是 Agent 已经跑偏上下文之后。更稳妥的方式是结合规则引擎先确定故障类别再选择合适的修复动作。先验证循环检测和熔断这两件事再逐步丰富修复策略和批量重放能力。这篇文章给出的代码是通用模板落地时需要按实际项目替换路径、端口、模型名和工具参数。建议收藏备用等下次 Agent 在生产环境再次挂掉时翻出这套故障分类和排查清单会比临时读日志高效很多。