AI Agent 开发越深入异常处理就越不像普通后端那样简单。模型可能超时、工具调用可能失败、中间链路可能崩掉甚至整个执行引擎都不给反馈。这次的笔记围绕 Agent 开发里的异常处理梳理三种最常用的处理方式同步兜底、基于 CompletableFuture 的异步异常处理以及 Agent 运行时的自动重试与降级。如果你正在做 AI Agent 项目或者刚接触 Agent 框架建议先记住一个结论Agent 的异常处理不是只靠 try/catch而是要把“模型请求失败”和“业务流程失败”分开处理。下面按可落地的顺序拆开讲。1. Agent 异常处理核心能力速览先说三种方式分别解决什么问题方便你在实际代码里做取舍。处理方式适用场景核心思路优点缺点方式一同步 try/catch 与 fail-fast 校验简单 Agent、单轮工具调用、参数校验在调用点捕获异常快速失败代码直观、调试简单无法覆盖异步链路长任务容易漏异常方式二CompletableFuture 异步异常处理多工具并行、跨服务调用、长链路 Agent用 exceptionally/handle 把异常纳入异步链路异步任务可控链路清晰学习成本略高需要处理回调地狱方式三Agent 运行时级容错生产环境、Agent 框架、长时间运行任务超时控制、重试、降级、人工兜底稳定性高适合批量任务需要额外配置可能掩盖真实错误从实际项目看方式一是基础方式二是主力方式三是生产环境的最后防线。下面会分别给出代码示例和验证方法。2. 适用场景与使用边界先明确一点不是所有 Agent 项目都需要把异常处理做得非常重。如果你只是在本地跑一个简单问答 Agent方式一就够用。如果你的 Agent 会调用多个工具并且用户可能通过接口或批量任务触发那方式二和方式三必须上。这个工具组合适合这几类人正在用 Java 做 Agent 框架开发的工程师需要处理 CompletableFuture 异步编程中的异常。使用 LangChain、LangGraph、CrewAI 等 Python Agent 框架但发现默认配置下模型超时或工具调用失败时没有自动恢复能力的人。需要把 Agent 封装成 API 服务再对接批量处理任务的开发团队。使用边界也要说清楚Agent 异常处理不能替代业务校验输入参数还是要在进入 Agent 前检查。如果模型本身不稳定异常处理只能降低失败率不能保证结果一定正确。涉及用户隐私、版权素材、人脸或声音等敏感数据时异常日志里不要打印敏感信息。不要把这套机制用于规避安全限制或绕过系统权限越权或违规操作本身就是不可接受的。合规层面和 Agent 框架、异步编程相关的代码示例都是开发常用模式不涉及敏感操作但在生产环境中需要关注日志脱敏和访问控制。3. Agent 异常处理前置条件与通用环境由于这篇内容不是某个固定开源项目的部署教程而是针对 Agent 开发的通用方案所以环境不是唯一固定的。建议按下面两个轨Prepare3.1 Python Agent 开发环境如果你用 Python 做 Agent 开发例如 LangChain 或自研 Agent# 建议 Python 3.10 python -m venv agent_env source agent_env/bin/activate pip install langchain langgraph openai httpx注意具体版本要看官方文档不同版本之间的 API 差异比较大。不要直接复制旧项目的 requirements。3.2 Java Agent 开发环境如果你用 Java 做 Agent并且需要处理异步调用JDK 11 或更高版本推荐 JDK 17。Maven 或 Gradle 构建工具。常见的 HTTP 客户端依赖例如 OkHttp 或 HttpClient。如果接入大模型 API需要准备模型服务地址和密钥。一个最小 Maven 依赖示例dependency groupIdcom.squareup.okhttp3/groupId artifactIdokhttp/artifactId version4.12.0/version /dependency dependency groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId version2.15.2/version /dependency这里的版本号只是一个常规选择实际项目请以官方当前稳定版本为准。3.3 通用初始化检查在写异常处理代码之前先检查这几项模型服务是否可达用 curl 或测试脚本请求一次确认能返回正常结果。超时时间是否配置HTTP 客户端和模型 SDK 通常有默认超时但生产环境建议显式设置。端口是否被占用如果 Agent 服务需要监听端口启动前检查端口。日志系统是否可用异常处理非常依赖日志建议提前配置 JSON 日志或结构化日志。4. 方式一同步 try/catch 与 fail-fast 校验这是最基础的异常处理方式也是很多 Agent 项目的起点。核心逻辑是在 Agent 调用模型、解析输出、执行工具时遇到异常立刻抛出或记录不让错误继续往下传递。4.1 Python 示例import json from openai import OpenAI client OpenAI(timeout10) def call_agent_with_check(prompt: str) - dict: # 第一步fail-fast 校验不要在 Agent 内部才发现参数为空 if not prompt or len(prompt.strip()) 0: raise ValueError(prompt cannot be empty) try: response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.2 ) content response.choices[0].message.content # 关键解析模型返回的结构时也要捕获异常 return json.loads(content) except json.JSONDecodeError as e: raise RuntimeError(fmodel output is not valid json: {e}) from e except Exception as e: raise RuntimeError(fcall agent failed: {e}) from e这段代码有两个关键点先做参数校验快速失败。把模型返回内容解析 JSON 的异常单独捕获避免和网络异常混在一起。4.2 Java 示例public class AgentExecutor { public AgentResult execute(String prompt) { if (prompt null || prompt.isBlank()) { throw new IllegalArgumentException(prompt cannot be blank); } try { String raw callModel(prompt); return parseResult(raw); } catch (IOException e) { throw new AgentExecutionException(model call failed, e); } catch (JsonParseException e) { throw new AgentExecutionException(model output parse failed, e); } } private String callModel(String prompt) throws IOException { // 实际项目中替换为 HTTP 调用 return {\answer\: \ok\}; } private AgentResult parseResult(String raw) throws JsonParseException { // JSON 解析逻辑 return new AgentResult(raw); } }4.3 边界与问题方式一在单轮调用时非常清晰但在真实 Agent 项目里模型调用常常发生在异步线程池、消息队列或批量任务里。这时 try/catch 只能捕获当前线程的异常无法捕获异步线程或回调链路里的异常。一旦 Agent 的执行跨越多个线程就必须引入方式二。当异常没有被捕获时Agent 执行 Provider 可能会出现类似“the agent execution provider did not respond in time”的错误。这个问题通常不是某个代码块没写 try/catch而是异步执行链路里超时被吞掉或者任务执行线程池满了导致请求排队。5. 方式二异步链路异常处理CompletableFuture 实践热词里出现的 CompletableFuture 异步编程异常处理正好对应 Agent 开发中的多工具并行、多模型调用场景。Agent 经常需要同时查库存、查订单、调搜索这些操作如果串行执行耗时太长如果并行执行就必须处理异步异常。5.1 CompletableFuture 三种处理异常的回调Java 中 CompletableFuture 提供了三个核心方法exceptionally只处理异常返回一个默认值。whenComplete无论成功还是失败都能拿到结果和异常但不改变结果。handle无论成功还是失败都能拿到结果和异常并且可以返回新的结果类型。下面是一个 Agent 调用多个工具的例子import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; public class AsyncAgent { private final ToolClient searchClient; private final ToolClient orderClient; public AsyncAgent(ToolClient searchClient, ToolClient orderClient) { this.searchClient searchClient; this.orderClient orderClient; } public CompletableFutureAgentResult executeAsync(String query) { CompletableFutureString searchFuture CompletableFuture .supplyAsync(() - searchClient.call(query)) .exceptionally(ex - { // 搜索失败时不阻断整个 Agent返回兜底内容 System.out.println(search failed: ex.getMessage()); return search unavailable; }) .orTimeout(5, TimeUnit.SECONDS); CompletableFutureString orderFuture CompletableFuture .supplyAsync(() - orderClient.call(query)) .exceptionally(ex - { System.out.println(order service failed: ex.getMessage()); return order unavailable; }) .orTimeout(8, TimeUnit.SECONDS); return searchFuture .thenCombine(orderFuture, (searchResult, orderResult) - { String answer search searchResult , order orderResult; return new AgentResult(answer); }) .handle((result, ex) - { if (ex ! null) { return new AgentResult(fallback answer due to async failure); } return result; }); } }这段代码包含三层处理每个并行任务用exceptionally提供了局部兜底。orTimeout设置了每个异步任务的最大等待时间。最后handle统一兜底防止合并阶段再抛异常。5.2 Python asyncio 的对应实现Python 里常见的异常处理方式和 CompletableFuture 类似但结构不同import asyncio async def call_search(query: str) - str: # 模拟搜索工具 await asyncio.sleep(1) return search ok async def call_order(query: str) - str: # 模拟订单服务失败 raise RuntimeError(order service timeout) async def run_agent(query: str) - str: search_task asyncio.create_task(call_search(query)) order_task asyncio.create_task(call_order(query)) results await asyncio.gather( search_task, order_task, return_exceptionsTrue # 关键不要默认抛出 ) parsed [] for r in results: if isinstance(r, Exception): parsed.append(service unavailable) else: parsed.append(r) return fsearch{parsed[0]}, order{parsed[1]}return_exceptionsTrue是异步 Agent 里非常重要的一个参数。默认情况下asyncio.gather会在第一个异常时立刻抛错其他任务可能还挂在后台不便于统一处理。设置return_exceptionsTrue后每个任务的结果都会返回异常本身作为结果返回。5.3 异步异常处理的一些坑异常被吞掉。exceptionally返回默认值后原始异常只在日志里出现调用方感知不到。生产环境建议把异常计数上报到指标系统。超时没有设置。只要有一个工具服务一直不返回整个 Agent 链路就会卡住。建议给每个异步任务都加上超时。多个任务互相依赖时异常处理顺序要提前设计。比如 A 失败后 B 是否需要继续执行这取决于业务需求。批量 Agent 任务中CompletableFuture默认使用 ForkJoinPool线程数有限。如果批量任务非常多建议显式传入自定义线程池。6. 方式三Agent 运行时级容错超时、重试与降级方式一到方式二能处理大部分编程层面的异常但 Agent 还是一个不确定性很高的系统。模型可能长时间不返回或者返回格式不对工具可能连续失败。生产环境里常见的问题就是Agent 框架的执行 Provider 没有在预期时间内响应。错误信息通常类似The agent execution provider did not respond in time. This may indicate the model service is overloaded or the configuration is incorrect.这种问题的处理不能只靠代码里的 try/catch需要 Agent 运行时提供更完整的容错机制。6.1 设置全链路超时在 Agent 入口处设置一个总超时时间比在每一个模型调用点各自设置超时更可靠。import asyncio async def run_with_timeout(agent_task, timeout: float): try: return await asyncio.wait_for(agent_task, timeouttimeout) except asyncio.TimeoutError: return { status: timeout, fallback: agent execution timeout, please try again }这个外层超时能兜住很多内部异常如果内部某个工具卡住但内部没做超时外层超时依然能兜底。6.2 自动重试与退避模型 API 的失败通常有瞬时性重试是提高成功率最有效的手段。重试策略需要做指数退避import time def call_with_retry(fn, max_retries3): for i in range(max_retries): try: return fn() except Exception as e: if i max_retries - 1: raise e wait_time 2 ** i print(fretry {i 1}, wait {wait_time}s) time.sleep(wait_time)需要考虑一个问题不是所有异常都适合重试。鉴权失败、参数格式错误这类错误重试只会浪费时间。建议按异常类型区分例如超时和限流可以重试参数错误直接失败。6.3 降级策略当模型不响应或工具全部失败时Agent 不能直接抛错误给用户。合理做法是提供降级回答。def agent_fallback(question: str) - str: return 当前服务繁忙暂时无法处理你的问题。请稍后再试。更复杂的降级策略可以是这样主模型超时切换到备用模型。主模型解析失败使用简单的模板回答。工具服务不可用返回缓存数据。所有能力都不可用返回人工服务入口。6.4 日志与可观测性Agent 容错机制最容易踩的坑是“异常被自动恢复后开发完全不知道”。如果没有日志和监控重试和降级会掩盖真实故障。建议关键节点输出结构化日志{ event: agent_exception, stage: tool_call, tool: order_service, error_type: TimeoutError, retry_count: 2, fallback_used: true, timestamp: 2025-01-01T10:00:00Z }这份日志能帮助确认异常是否高频出现重试是否有效降级是否被触发如果全部静默处理Agent 最终可能总是返回错误或假答案但系统中看不到任何异常记录。7. Agent 服务接口 API 与批量任务的异常设计Agent 一旦封装成服务异常处理就不再只是代码内部的事。接口需要通过错误码和消息告诉调用方发生了什么而批量任务需要独立的失败重试机制。7.1 API 统一错误结构先设计一个统一的响应结构{ code: 0, message: success, data: { answer: ... } }异常时{ code: 50001, message: agent execution timeout, data: null }这里不建议把内部堆栈直接返回给调用方而是记录在服务端日志里。调用方只需要知道错误类型和是否可重试。7.2 FastAPI 示例from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class AgentRequest(BaseModel): prompt: str timeout: float 30.0 class AgentResponse(BaseModel): code: int message: str data: dict | None None app.post(/agent/run, response_modelAgentResponse) async def run_agent(req: AgentRequest): try: result await run_agent_task(req.prompt, req.timeout) return AgentResponse(code0, messagesuccess, data{answer: result}) except TimeoutError: return AgentResponse(code50001, messageagent execution timeout, dataNone) except Exception: return AgentResponse(code50002, messageagent internal error, dataNone)7.3 批量任务错误队列批量 Agent 任务比单次调用更容易失败因为同样的工具链在不同输入下可能触发不同问题。建议在任务队列里加入失败重试、死信队列、人工复核三件套。最简单的批量任务状态设计状态含义处理策略pending等待执行分配执行running执行中超时自动失败success执行成功正常输出failed执行失败进入重试队列dead重试仍失败人工处理在代码中可以这样设计task_status { task_id: task_001, status: pending, retry_count: 0, last_error: None } def process_task(task): try: result agent.run(task[prompt]) task[status] success task[result] result except Exception as e: task[status] failed task[last_error] str(e) task[retry_count] 1 if task[retry_count] 3: queue.put(task) else: task[status] dead实际项目里这种状态通常放在 Redis 或数据库里而不是内存中。使用内存状态时服务一重启任务队列就丢了长耗时批量任务可能全部卡死。8. 功能验证与效果观测无论采用哪种异常处理方式都要实际验证异常发生时系统是否按预期工作。建议按下面的测试用例执行8.1 验证方式一直接调用失败方法测试目的确认同步调用出现异常时能快速返回错误。操作步骤准备一个空 prompt。调用 Agent。观察是否抛出ValueError或IllegalArgumentException。判断标准调用方拿到明确错误而不是卡住或返回空结果。8.2 验证方式二模拟一个异步任务失败操作步骤在 CompletableFuture 链路中让orderClient.call抛出异常。调用executeAsync。观察最终结果是否包含降级文案。判断标准异步链路没有中断最终返回兜底结果。8.3 验证方式三超时触发降级操作步骤把模型调用的超时时间设置为 1 秒。调用一个实际需要 3 秒的慢服务。观察 Agent 是否返回降级回答。判断标准外层超时触发没有一直等待。8.4 观察系统指标重点观察以下指标Agent 请求成功率。工具调用失败率。平均等待时间。超时任务数量。重试次数分布。降级回答占比。如果降级回答占比超过 5%说明上游服务可能不稳定需要排查模型或工具服务。9. Agent 异常处理常见问题与排查方法问题现象可能原因排查方式解决方案Agent 一直不返回结果缺少超时控制检查日志查看任务卡在哪个阶段给模型调用和工具调用加超时看到 “execution provider did not respond in time”模型服务超时或 SDK 配置错误先手动 curl 模型接口确认可达性检查模型服务地址、密钥、超时时间异步任务异常没被捕获使用了exceptionally但异常发生在合并阶段检查链路上是否有handle或whenComplete在最终阶段补充统一异常兜底批量任务卡在 running 状态没有超时机制任务一直等待查看任务队列中的执行时间增加任务超时和失败自动转移重试导致接口被连续打爆重试策略没有退避查看失败任务的日志间隔改用指数退避或加入随机抖动降级回答占比过高上游服务整体不稳定检查工具调用失败率优先恢复上游服务不要只依赖降级日志中看不到异常异常被框架吞掉或日志级别太高检查日志配置把异常事件单独输出到独立日志文件排查时不要只盯着 Agent 框架内部要从整个链路排查模型服务 → 工具服务 → 中间件 → Agent 业务逻辑 → 调用方。10. 最佳实践与后续扩展Agent 异常处理本质上是在做一个系统设计决策哪些错误可以重试哪些错误应该快速失败哪些错误需要降级哪些错误需要人工介入。我从实际项目里总结出下面几条经验按优先级排列入口总超时必须有。无论内部怎么重试总超时要保证任务一定结束。异常类型要细分。网络超时、限流、鉴权失败、解析失败的语义完全不同不能统一处理。重试必须有退避和次数上限批量任务时尤其重要。降级答案必须标记。如果一个回答是降级内容最好在日志和结果结构里增加fallback: true标记避免调用方误以为这是正常结果。日志要有足够的信息。至少包含阶段、异常类型、重试次数、任务 ID、输入摘要。在线服务要阻止外部大量无效请求。前置校验、限流、熔断不要把所有压力打给模型接口。后期扩展方向可以考虑把异常处理抽象成 Agent 中间件统一注入所有 Agent 任务避免每个调用点重复写 try/catch。引入人工审核队列。当重试和降级都失败时把任务送入人工队列而不是默默失败。建立 Agent 运行监控看板实时展示超时次数、重试成功次数、降级次数快速定位不稳定环节。回到开头那个问题Agent 异常处理是不是只靠 try/catch 就够了答案很明确不够。同步异常处理只能解决最基础的问题异步链路需要 CompletableFuture 这类异步异常处理机制生产环境还需要运行时级超时、重试、降级和可观测能力作为兜底。最好的做法是把这三层异常处理当成一套组合拳入口做校验和超时中间并行任务用异步异常回调框架层提供重试和降级。先跑通一个最小示例再逐步补监控和批量任务这样即使模型服务不稳定Agent 服务也能保持可用。