这次我们不聊具体的 Agent 框架怎么装、哪个模型跑得更快而是聊一个更容易被忽略、但在生产环境里几乎天天遇到的问题Agent 异常处理的可选性。Agent 这个词在不同语境下指的东西不太一样。在 AI 开发里Agent 通常指能自主调用工具、拆解任务、执行多步流程的智能体在 Java 异步编程里Agent 又可能是任务执行器、数据采集器或者远程服务调用代理。但不管哪种 Agent只要跑在真实环境里就一定会出现异常模型超时、服务端 500、上下文过长、工具调用失败、批量任务中断。真正的问题不是“会不会出错”而是“出错之后怎么选”。本文会把这套“可选性”拆开讲先解释什么是 Agent 异常处理的可选性再梳理 5 种可选的异常处理策略然后分别用 Java CompletableFuture 和 Python asyncio 给出能直接改用的示例代码最后补充接口 API 调用、批量任务和常见排查清单。适合正在做 Agent 开发、异步任务编排、本地服务接入的读者。1. 核心能力速览能力项说明主题定位Agent 运行过程中的异常处理策略选择重点解决“出错了怎么办”适用范围AI Agent 任务链、异步任务编排、接口调用、批量处理可选策略显式抛出、捕获重试、降级回退、忽略放行、人工升级核心价值通过策略组合提高 Agent 任务的成功率和可维护性常用语言JavaCompletableFuture、Pythonasyncio批量任务能力支持通过队列、重试器、补偿任务实现批量异常恢复关键风险超时、模型不稳定、上下文超长、工具调用失败、部分成功这套内容不依赖特定的 Agent 框架不限制模型不管是自研 Agent 还是接入第三方 Agent 服务设计思路都能复用。2. 先说清楚异常处理的“可选性”到底指什么先看一段最原始的 Agent 调用代码。很多初学者是这样写的def run_agent(prompt): result agent_execute(prompt) return result这段代码的问题在于只要底层某一个步骤出错整个 Agent 任务就中断。如果是一条测试链路这样写没问题但如果是在生产环境用户提交了一个长文档解析任务模型在第 3 步超时结果前端直接报错用户不知道发生了什么也不知道要不要重试这就是典型的“不可选”。所谓“可选性”简单说就是同一个异常你能根据场景选择不同的处理路径。维度不可选可选小任务失败任务整体失败直接重试模型超时用户看到报错自动降级到备用模型工具链中断任务卡死跳过当前工具继续后续步骤批量任务部分失败全部失败只重跑失败项关键任务失败静默失败通知人工处理从工程角度看Agent 异常处理需要把“异常”当作业务数据的一部分而不是直接丢弃也不是每次都必须终止任务。真正成熟的 Agent 系统在设计阶段就会把异常分为可恢复、可忽略、不可恢复三类然后分别配置策略。搜索热词里经常出现 CompletableFuture 异步编程异常处理这也是 Agent 异常可选性很重要的一块。异步任务里异常不是立即抛给调用方的它经过了线程切换、任务编排如果你不在每个环节显式处理异常很容易被吞掉或者延迟到无法定位。后面会专门给示例。3. Agent 异常来源分类在做异常处理之前先得知道异常从哪里来。不同来源的异常处理方式完全不同。3.1 底层依赖异常Agent 要调用模型 API、工具服务、数据库、外部 HTTP 接口。任何一个下游服务不稳定都会变成 Agent 的异常。典型错误The agent execution provider did not respond in time意思是执行提供程序没有按时响应。connection reset连接被重置。HTTP 429请求过多。HTTP 500服务内部错误。这类异常策略上优先考虑重试和降级。3.2 模型语义异常不是网络报错而是模型返回的内容本身有问题。比如返回了空内容、JSON 解析失败、ReAct 格式不对、工具调用参数缺字段。这类异常重试不一定有效因为换个输入或者加提示词可能更好更稳妥的做法是校验输出格式必要时用规则解析兜底。3.3 Agent 自身状态异常上下文超长、记忆丢失、状态冲突、任务循环不终止。这类异常和任务设计强相关。常见处理方式包括清空部分历史、设置最大迭代次数、将失败任务标记为“未完成”。3.4 资源约束异常显存不足、内存溢出、线程池耗尽、文件句柄太多。这类异常通常不是简单重试能解决的需要做资源隔离、限制并发或者在更高层面做排障。异常类型代表情况推荐可选策略依赖超时模型服务未按时响应重试、降级语义异常JSON 解析失败、工具参数缺字段修复提示词后重试状态异常上下文过长、任务死循环截断上下文、终止任务资源异常显存不足、线程池耗尽限流、人工介入4. 五种可选异常处理策略4.1 策略一快速失败对于一些不值钱、不重要的任务比如用户上传一张图片后由 Agent 自动生成标题失败就失败直接把错误返回给上层调用方。不需要重试不需要补偿因为重试成本比任务本身价值还高。应用场景非关键路径的日志分析。用户手动触发的简单 Agent 任务。外部接口明确返回 400 这类不可重试错误。代码示例def run_agent_task(prompt): result agent_execute(prompt) if result is None: raise AgentExecutionError(agent execution failed) return result4.2 策略二捕获重试适用于瞬时异常特别是网络抖动、模型服务繁忙、临时限流。重试时要注意两点最大重试次数和退避时间。无限重试会拖垮整个链路。示例import time from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10)) def call_agent_with_retry(prompt): return agent_execute(prompt)一些 Agent 框架自带重试机制第三方库也提供了tenacity、resilience4j之类的重试组件。重试策略适合的网络超时和 5xx 错误不适合 4xx 参数错误。4.3 策略三降级回退当主链路失败时切换到备用链路。最常见的降级方式是换模型主模型超时自动使用一个更轻量的模型继续执行或者主服务不可用自动切换到本地备用服务。示例流程调用主模型等待 30 秒。如果超时记录一次失败。调用备用模型。备用模型也没有通过才返回最终失败。def execute_with_fallback(prompt): try: return call_main_agent(prompt) except AgentTimeoutError: logger.warning(main agent timeout, switch to backup) return call_backup_agent(prompt)4.4 策略四忽略放行有一部分异常对最终结果没有影响或者可以在任务结束后统一补偿。比如一个 Agent 任务里既要做摘要生成又要做关键词抽取关键词抽取失败了摘要生成结果仍然有效那这个异常就可以降级为“忽略”。这种策略要谨慎。一旦决定忽略必须记录日志否则后续排查时找不到问题根因。try: keywords extract_keywords(text) except Exception as e: logger.warning(fextract keywords failed: {e}, ignored) keywords []4.5 策略五人工升级关键业务任务失败后自动处理无法解决那就需要把人加进来。常见做法把失败任务写入专门的“失败队列”。给运维或审核人员发送通知。保留失败现场包括输入参数、异常堆栈、已执行步骤方便人工重放。这种策略最适合交易类、审核类、内容发布类 Agent 任务。AI Agent 可以自动化一部分流程但关键节点的最终决定权应当保留人工权限。5. Java 异步 Agent 任务中的异常处理CompletableFuture 示例Java 开发者在做 Agent 后台任务时经常会用 CompletableFuture 编排异步流程。异步任务的异常处理和同步代码有很大区别异常不是在抛出点立刻处理而是随着任务链传递。这里给出一个完整的、可改用的示例。5.1 场景定义假设有一个文档处理 Agent包含三个步骤解析文档提取文本。调用模型做摘要。保存摘要结果。这三个步骤可以串行也可以部分并行。我们希望第 1 步失败直接结束第 2 步失败自动重试一次再失败则降级到备用模型第 3 步失败写入补偿队列。import java.time.Duration; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; public class AgentTaskExample { public static void main(String[] args) { String docPath /tmp/input.txt; CompletableFuture.supplyAsync(() - parseDocument(docPath)) .thenApplyAsync(AgentTaskExample::summarizeWithRetry) .thenAcceptAsync(AgentTaskExample::saveResult) .exceptionally(throwable - { // 最终兜底任务失败进入人工队列 System.out.println(Agent task failed: throwable.getMessage()); enqueueManualReview(docPath, throwable); return null; }); // 等待异步任务完成避免主线程退出 try { Thread.sleep(10000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } private static String parseDocument(String path) { if (path null || path.isBlank()) { throw new IllegalArgumentException(doc path is empty); } return document content; } private static String summarizeWithRetry(String content) { try { return CompletableFuture.supplyAsync(() - callMainModel(content)) .get(30, TimeUnit.SECONDS); } catch (Exception e) { System.out.println(main model timeout, switch to backup model); return callBackupModel(content); } } private static String callMainModel(String content) { // 模拟主模型调用 return summary from main model; } private static String callBackupModel(String content) { // 模拟备用模型调用 return summary from backup model; } private static void saveResult(String result) { System.out.println(save result: result); } private static void enqueueManualReview(String path, Throwable throwable) { System.out.println(write failed task to manual review queue); } }这段代码体现了两个关键点exceptionally是 CompletableFuture 的全局兜底。不管任务链哪个环节出现未捕获异常最终都会走到这里。第 2 步的summarizeWithRetry内部自己处理了超时和降级所以第 3 步不会感知到第 2 步曾经失败。5.2 注意 CompletableFuture 的异常吞掉问题CompletableFuture 很常见的一个坑是如果业务代码只调用了thenApply没有调用exceptionally也没有在最终阶段获取返回值那么异常会被静默吞掉。等线程池里的任务失败后外层代码看不到任何信号。解决方式有两个在任务链末尾加exceptionally或者whenComplete。保持引用在合适位置调用get()或join()强制获取结果。CompletableFutureString future CompletableFuture.supplyAsync(() - { throw new RuntimeException(agent inner error); }); // 统一处理 future.exceptionally(ex - { System.out.println(caught exception: ex.getMessage()); return fallback; });这就是“可选性”在异步编程里的体现你可以在异常产生时不管也可以选择在链路的任意位置拦截或者在最终统一兜底。选择权在开发手上而不是让异常自己决定任务命运。6. Python AI Agent 中的异常处理示例Python 是 AI Agent 开发的主力语言。asyncio 是异步 Agent 任务最常用的底层库下面用一个自定义 Agent 执行器来演示可选策略的组合。6.1 自定义 Agent 异常类型class AgentError(Exception): Agent 基础异常 class AgentTimeoutError(AgentError): Agent 执行超时 class AgentToolError(AgentError): Agent 工具调用失败 class AgentOutputError(AgentError): Agent 输出格式异常先定义细粒度的异常类型。不要所有异常都用Exception因为后面的策略选择需要依赖异常类型做分支判断。异常类型本身越细可选性越高。6.2 带策略选择的 Agent 执行器import asyncio import logging logger logging.getLogger(agent) class AgentExecutor: def __init__(self, max_retries3, fallback_modelsmall-model): self.max_retries max_retries self.fallback_model fallback_model async def run(self, prompt: str, strategy: str retry) - str: strategy 参数控制异常处理策略 - fail: 快速失败 - retry: 捕获重试 - fallback: 降级回退 - ignore: 忽略异常返回空结果 - escalate: 异常升级到人工队列 if strategy fail: return await self._run_with_fail(prompt) if strategy retry: return await self._run_with_retry(prompt) if strategy fallback: return await self._run_with_fallback(prompt) if strategy ignore: return await self._run_with_ignore(prompt) if strategy escalate: return await self._run_with_escalate(prompt) raise ValueError(funsupported strategy: {strategy}) async def _run_with_fail(self, prompt: str) - str: return await self._call_model(prompt) async def _run_with_retry(self, prompt: str) - str: last_exception None for attempt in range(self.max_retries): try: return await self._call_model(prompt) except AgentTimeoutError as e: last_exception e wait_time 2 ** attempt logger.warning(attempt %s failed, wait %ss, attempt 1, wait_time) await asyncio.sleep(wait_time) raise AgentError(fretry all failed: {last_exception}) async def _run_with_fallback(self, prompt: str) - str: try: return await self._call_model(prompt) except AgentError: logger.warning(main model failed, switch to fallback: %s, self.fallback_model) return await self._call_model(prompt, modelself.fallback_model) async def _run_with_ignore(self, prompt: str) - str: try: return await self._call_model(prompt) except AgentError as e: logger.warning(ignore error: %s, e) return async def _run_with_escalate(self, prompt: str) - str: try: return await self._call_model(prompt) except AgentError as e: logger.error(escalate to manual queue: %s, e) await self._send_to_manual_review(prompt, str(e)) raise async def _call_model(self, prompt: str, model: str main) - str: # 模拟模型调用可以替换成真实的 HTTP 请求 await asyncio.sleep(0.1) return fmodel {model} result async def _send_to_manual_review(self, prompt: str, error: str) - None: # 这里可以写 MQ、写数据库、发通知 logger.info(write to manual review queue: prompt%s, error%s, prompt, error)调用方式async def main(): executor AgentExecutor() result await executor.run(写一段代码, strategyfallback) print(result) asyncio.run(main())这个设计说明一个核心思路同一个 Agent同一段重试逻辑通过strategy参数就可以在不同的任务场景中灵活选择不要求你在每个任务里重新写一套处理逻辑。6.3 批量任务里的策略配合批量任务场景下单独一个任务失败不能影响整个批次。推荐的做法是遍历任务时每个任务单独捕获异常并记录失败状态。async def run_batch(tasks): results [] failed [] for task in tasks: try: result await task.run() results.append(result) except Exception as e: failed.append({task: task, error: str(e)}) results.append(None) return results, failed批量任务要把“失败项”单独收集起来跑完一轮后统一重试。如果全部失败再触发升级策略。7. 接口超时、API 调用与降级链路Agent 开发绕不开 API 调用。AI Agent 要调用模型接口有时候还要通过 API 暴露自己的 Agent 能力。这两条链路都会出现异常处理问题。7.1 调用模型 API 的超时设置很多 Agent 任务的失败并不是模型能力不行而是调用方没有设置超时时间导致请求一直挂着最后整个任务队列被拖死。合理的做法是给每一次模型调用设置明确的超时。Python examplesimport httpx async def call_model_with_timeout(prompt: str): async with httpx.AsyncClient(timeout30.0) as client: response await client.post( https://api.example.com/generate, json{prompt: prompt}, ) response.raise_for_status() return response.json()Java 侧也一样使用HttpClient或者RestTemplate时设置connectTimeout、readTimeout。不要让一个慢请求占用线程池太久。7.2 对外提供 Agent API 时的异常返回如果你的 Agent 服务对外提供 API异常处理要考虑接口调用方的体验。不要把内部异常堆栈直接返回给调用方也不要吞掉异常返回一个假的成功结果。推荐的做法是定义统一的错误结构{ code: 5001, message: agent execution timeout, request_id: 8f2a3e11-9b21-4c55-b40e-c839a3f5c7b9, retryable: true }retryable字段直接告诉调用方这个异常是否值得重试。如果为false说明继续重试没有意义。8. 资源占用与性能观察Agent 异常处理虽然是逻辑层面的设计但也会影响资源占用。这里给出观察方向具体数字需要以实际环境为准。重试策略会延长任务整体执行时间。每多一次重试就多占一次线程池或协程资源同时还会重复调用模型 API增加耗时和费用。批量任务如果失败项很多重试时要控制并发避免所有失败任务同时重新执行造成下游服务被压垮。建议使用信号量限制重试并发数import asyncio semaphore asyncio.Semaphore(5) async def limited_retry(task): async with semaphore: return await task.run_with_retry()本地部署 Agent 时显存占用主要来自模型加载和异常处理本身没有直接关系。但如果任务失败导致模型重新加载资源开销会明显增加。更合理的做法是复用已经加载的模型失败时切换不同提示词而不是销毁模型实例。资源监控建议观察三个指标任务成功率、失败重试比例、任务平均耗时。这三个指标能直接反映异常处理策略是否合理。9. 常见问题与排查方法问题现象可能原因排查方式解决方案Agent 执行超时模型服务未按时响应查看日志中的耗时分布设置超时时间开启重试异常被吞掉任务无输出异步任务缺少兜底处理检查 CompletableFuture 是否挂了 exceptionally在链路末尾增加兜底处理重试导致下游被打爆重试没有设置退避和限流观察重试日志使用指数退避限制并发JSON 解析失败后重试无效模型输出格式不稳定打印原始输出增加输出格式校验和规则兜底批量任务部分失败整体被标记失败批量处理没有隔离异常检查批量循环结构每个任务单独捕获异常失败任务完全没有记录只用 try catch 忽略了异常检查日志输出增加结构化错误日志关键任务失败后无人处理没有升级机制检查任务是否进入人工队列增加人工审核队列和通知10. Agent 异常处理可选性的最佳实践把上面内容整理成一套可以直接落地到项目里的实践清单。10.1 先定义异常类型再写处理逻辑不要到处用except Exception一把抓。先定义好基础异常类再派生出超时、工具调用失败、输出格式异常等子类。异常类型越清晰可选策略的组合空间越大。10.2 给每一个 Agent 任务指定默认策略在 Agent 任务配置里增加一个字段例如error_strategy。默认值可以是retry但不同的业务场景可以覆盖这个字段。这样异常处理不再是一堆散落的 try catch而是一份可配置的策略表。tasks: - name: document_summary error_strategy: fallback timeout: 30 max_retries: 2 - name: keyword_extract error_strategy: ignore timeout: 1010.3 日志必须带上任务上下文记录异常时除了堆栈还要把任务 ID、输入摘要、已执行步骤、模型名写进日志。否则出现问题时你只能看到一个孤零零的堆栈无法定位是哪一步出的问题。logger.error( agent task failed, task_id%s, step%s, error%s, task_id, step, exc_infoe )10.4 关键任务必须保留人工复核入口AI Agent 可以自动完成大部分工作但关键决策节点要保持人对结果的最终控制权。把失败任务和低置信度结果写入人工复核队列这个设计在几乎所有生产级 Agent 系统里都是必要的。10.5 批量任务必须支持断点续跑运行大批量 Agent 任务时如果跑了一半进程崩溃不用每次从头开始。任务状态要有持久化设计已经成功的任务跳过失败的任务重新入队。10.6 关注第三方 Agent 服务的错误语义接入第三方 Agent 服务时要特别关注它返回的错误码和错误消息。有些错误重试有意义有些错误需要改参数还有些错误代表账户额度问题。不做区分直接重试会把临时问题和永久问题混在一起。11. 总结Agent 异常处理的可选性本质上是为了在不可靠的底层依赖之上构建可靠的任务系统。模型会超时、服务会抖动、工具会失败这些是常态不是意外。真正决定一个 Agent 能不能上生产环境的往往不是它能跑通多少条正常路径而是它在异常路径上能否优雅地选择处理方式。这篇文章里给出的 5 种策略——快速失败、捕获重试、降级回退、忽略放行、人工升级——并不是互相排斥的。实际系统通常是多种策略组合使用先重试重试失败后降级降级失败后再升级到人工处理。选择哪一种取决于任务的价值、失败成本、重试代价和实时性要求。建议先做一组小规模故障演练把模型 API 故意停掉观察 Agent 任务链路会走到哪一步把超时时间调短看重试逻辑是否按预期执行跑一批批量任务随机注入失败确认失败项能被正确收集和重跑。这些测试做完比读十篇文章都管用。