Agent 开发做到后面你会发现真正耗时间的不是写提示词、搭流程而是处理各种“意外”。模型 Provider 可能突然不给响应、工具调用可能返回错误码、Agent 输出可能不符合 JSON 格式、迭代循环可能一直走不出去。这篇内容把 Agent 异常处理的三种方式完整拆一遍入口代码捕获、Agent 内部自愈、外层编排兜底。每一步都给出可落地的代码思路和常见报错的排查路径。先给结论不要只写一套 try/catch 就指望把所有异常挡住。Agent 异常处理应该是多层防线从内到外分别是 Agent 自身容错、调用方硬捕获、执行编排层兜底。本文会先梳理 Agent 运行过程中常见的异常类型再逐一说明三种方式的实现思路和适用场景最后补充接口 API 场景、批量任务、性能开销和排查清单。适合正在做 Agent 应用开发、或者准备把 Agent 接进生产系统的开发者。1. 为什么 Agent 异常处理是开发中最高频的问题Agent 程序比普通接口服务多了一层不确定性普通接口的输入输出是稳定结构而 Agent 需要调用大模型再由模型决定下一步调用什么工具、生成什么格式的结果。大模型本身有概率性输出可能漂移服务可能超时工具链路可能中途失败。这些异常不是“意外踩到”而是 Agent 开发中的常态。从实际运行来看Agent 类的异常大致可以分为以下几类异常类型出现位置常见表现影响范围LLM Provider 超时模型调用环节provider did not respond in time任务直接中断输出解析失败模型返回结果解析JSON 解析报错、字段缺失后续流程卡死工具调用异常工具执行环节HTTP 500、参数错误、依赖服务不可用结果不可信上下文超限Prompt 组装环节token 超长、模型拒绝处理调用失败循环次数超限Agent 循环推理无限重试、无法收敛资源耗尽并发冲突批量任务场景线程阻塞、任务队列堆积吞吐量下降看清这些异常类型后再去设计处理方式就不会只写一个“大 try/catch 套所有逻辑”了。正确的做法是每一类异常都有对应的处理策略并且这些策略要分层次铺开。2. 三种异常处理方式概览先做一个整体对比再分章节展开。处理方式实现位置核心思路优点缺点方式一代码层硬捕获Agent 入口调用处try/except、异步异常回调、超时控制简单直接能兜住所有未处理异常只能事后处理无法提高任务成功率方式二Agent 内部自愈Agent 执行器内部提示词异常规则、输出修复、模型降级、重试能显著提高单次任务成功率实现复杂度高容易造成内部重试风暴方式三外层编排兜底Agent 外部执行器看门狗、超时终止、钩子函数、重试队列适合生产环境可控性强需要额外维护任务队列和监控逻辑三种方式不是互斥关系。实际项目中更常见的组合是Agent 内部先自愈内部解决不了就抛异常调用方用代码层硬捕获接住同时外层编排器负责超时终止和任务重投递。这样既照顾了成功率也保证了系统不会因为单个 Agent 运行异常被拖垮。3. 环境准备与前置条件在演示代码之前先给出一个通用的环境准备清单。如果你的 Agent 项目已经能跑起来可以直接跳到后面的代码部分。操作系统Windows / macOS / Linux 均可建议 Linux 服务器做生产部署。语言环境Python 3.10 或 Java 17具体版本取决于你使用的 Agent 框架。Agent 框架LangChain、LlamaIndex、AutoGen 或自研 Agent 骨架。下面的示例都是通用伪码需要按实际框架的 API 调整。依赖包大模型 SDK如 openai、JSON 解析库、日志库loguru / logback。模型访问准备一个可用的模型 API Key先在小流量测试环境验证不要直接在生产环境跑调试。# Python 环境通用安装命令按项目实际替换 pip install openai langchain pydantic loguru环境准备的核心目的是让异常处理代码可以被稳定复现。如果连一个最小可运行的 Agent 都没有后面所有异常处理都是空谈。建议先搭一个最简 Agent一个模型调用 一个工具函数 一个循环指令跑通后再逐步加入异常处理逻辑。4. 方式一代码层硬捕获这是最基础、也最容易被低估的一层。很多人觉得 try/except 太简单不值得单独写但在 Agent 场景里入口硬捕获是所有异常处理失效时的最后一道防线。先看一个通用 Python 示例。try: result agent.run(帮我查询订单状态) except ProviderTimeout as e: log.error(provider timeout: {}, e) result 服务繁忙请稍后重试 except ToolExecutionError as e: log.error(tool execution failed: {}, e) result 工具调用失败已完成降级处理 except OutputParsingError as e: log.error(output parse failed: {}, e) result 结果解析失败请检查模型输出格式 except Exception as e: log.exception(unexpected agent error) result 系统异常请联系管理员这里的核心不是 try/except 语法本身而是异常类型要细分。如果只写一个except Exception所有异常都走同一个分支排查问题时你根本不知道是模型超时还是工具调用失败。建议的做法是把 Agent 运行过程中可能出现的异常按来源归类然后分别捕获、分别记录、分别处理。4.1 异步任务的异常处理Agent 调用经常是异步的尤其是处理长任务时同步阻塞等待一个思考过程很浪费资源。如果是 Python asyncio 场景可以给任务加超时控制import asyncio try: result await asyncio.wait_for(agent.arun(task), timeout180) except asyncio.TimeoutError: log.error(agent task timeout) result fallback_result()如果是 Java 项目CompletableFuture的异常处理是重点。热搜词里也出现了completablefuture异步编程异常处理这里单独给一个示例CompletableFutureString future CompletableFuture .supplyAsync(() - agent.run(task)) .orTimeout(120, TimeUnit.SECONDS) .exceptionally(ex - { log.error(agent execute failed, ex); return fallbackResult; });orTimeout能保证任务不会无限等待exceptionally负责捕获链路上抛出的所有异常。相比try/catch包住整段同步代码这种方式更符合异步调用的习惯也更容易接入监控体系。4.2 捕获到异常之后做什么代码层硬捕获不等于“捕获了就返回兜底文案”。正确顺序是记录完整异常堆栈和关键上下文。判断是否需要重试重试次数是否达到上限。如果重试失败再走降级逻辑。将错误信息结构化返回给上层调用方。很多线上问题都是“异常被吞了”。比如外层套了 try/except 但只返回null后续步骤拿到null继续执行报错信息完全丢失。代码层硬捕获的关键是把异常变成可观测的数据而不是让它无声消失。5. 方式二Agent 内部自愈第二部分要展开 Agent 内部的自愈能力。这层处理得好可以大幅减少外部重试的次数。5.1 提示词注入异常处理规则Agent 本身是由大模型驱动的你可以通过在 system prompt 中加入异常处理规则让模型在遇到失败时主动调整策略。这是一种成本最低的自愈方式。你是任务规划 Agent。请严格遵循以下规则 1. 如果某个工具调用失败先检查参数是否正确之后可重试一次。 2. 如果重试仍失败请更换工具或直接基于已知信息回答。 3. 如果连续 3 次失败停止执行并向上层报告具体错误原因。这类提示词看起来简单但实际效果往往不错。因为大模型具备一定的推理能力给它明确的失败处理规则后它会在执行链路里主动做调整而不是直接把异常抛出来。在项目早期先用提示词规则兜住一部分常见异常比立刻写复杂的代码逻辑更快见效。5.2 输出解析失败的自动修复Agent 输出解析失败是很常见的异常。模型可能生成了多余的说明文字导致 JSON 解析失败也可能漏掉了必填字段。常见的处理方式是把解析错误信息回传给模型让它重新输出。def parse_output_with_repair(agent, raw_output): try: return json.loads(raw_output) except json.JSONDecodeError as e: repair_prompt f 上次输出解析失败错误信息{e} 请重新输出只输出合法的 JSON不要添加任何解释文字。 上次输出内容是{raw_output} new_output agent.invoke(repair_prompt) return json.loads(new_output)修复过程也需要限制次数否则模型可能连续多次输出不合格格式导致额外 token 消耗。建议最多修复 2 次仍然失败就让上层走硬捕获逻辑。5.3 模型降级与重试Agent 调用的模型服务可能因为限流、临时故障等原因出现不稳定。这里可以设计一条模型链路主模型失败时自动降级到备用模型。def call_llm_with_fallback(prompt, model_chain): last_error None for model in model_chain: try: return model.invoke(prompt) except (RateLimitError, ConnectionError, TimeoutError) as e: last_error e log.warning(model call failed, try next. error: {}, e) continue raise RuntimeError(fall models failed: {last_error})降级顺序要提前配置好。比如主模型用高精度模型备用模型用低成本快速模型。还要注意降级不只是换一个模型接口还要看上下文格式是否兼容避免备份模型不支持某些工具调用格式。5.4 循环次数与停止条件Agent 的循环推理必须设置上限。不设上限的 Agent 可能在同一个失败节点上反复重试直到资源耗尽。常见的做法有设置max_iterations限制最大循环轮数。在每一轮循环前检查已执行轮数超过阈值直接终止。结合 token 消耗统计估算每一轮可能产生的 token 数量提前设定预算。agent Agent( max_iterations5, max_tool_calls10, on_llm_errorhandle_llm_error, on_chain_errorhandle_chain_error, )on_llm_error和on_chain_error是框架层面的回调钩子可以在异常发生时执行自定义逻辑。把重试、降级、日志上报都放到钩子里Agent 主流程会干净很多。6. 方式三外层编排兜底第三部分是外层编排层的兜底。这层的作用不是提高单次任务的成功率而是保证整个系统不会因为一个 Agent 任务卡住而崩掉。6.1 执行器超时终止即使 Agent 内部设置了循环上限外部还是应该再加一道超时保护。比如用线程池执行 Agent 任务并设置future.result(timeout)from concurrent.futures import ThreadPoolExecutor, TimeoutError executor ThreadPoolExecutor(max_workers4) future executor.submit(agent.run, task) try: result future.result(timeout300) except TimeoutError: future.cancel() log.error(agent task exceeded 300 seconds, terminated) notify_alert(task_id)注意future.cancel()只能取消尚未执行的任务。如果 Agent 已经在执行中取消操作不会立即中断运行还需要配合 Agent 内部的停止标识。更稳妥的做法是使用进程级隔离将 Agent 任务放到独立进程外层通过进程终止来控制。6.2 心跳看门狗看门狗的核心思路是Agent 在执行过程中定期上报心跳外层监听器如果一段时间内没收到心跳就判定任务卡死触发恢复操作。def watchdog(agent, timeout60): last_heartbeat agent.last_heartbeat_time if time.time() - last_heartbeat timeout: log.error(agent heartbeat timeout, restart task) agent.stop() retry_queue.put(agent.task_info)看门狗模式适合长耗时、多步骤 Agent 任务。它本身不处理异常只负责发现“卡住”的信号然后交给重试队列做恢复。6.3 重试队列与死信队列生产环境中一次失败的 Agent 任务往往需要重新投递。可以设计一个简单的任务队列失败任务带着重试次数重新排队。# 伪代码示例失败任务进入重试队列 retry_queue.put(task_info) def worker(): while True: task retry_queue.get() try: run_agent(task) except Exception as e: if task.retry_count 3: task.retry_count 1 retry_queue.put(task) else: dead_letter_queue.put(task) log.error(task failed after retries: {}, task.task_id)重试队列需要关注幂等性。如果一个 Agent 任务被重复执行会产生副作用比如发送邮件、扣减库存、调用外部支付接口那么重试前必须做幂等校验否则会造成重复操作。7. 接口 API 与批量任务场景把 Agent 封装成接口服务时异常处理需要考虑服务边界。比如 FastAPI 同步接口里直接跑 Agent请求可能会长时间占用连接。这种情况下建议把 Agent 执行改成异步任务模式。from fastapi import FastAPI, BackgroundTasks app FastAPI() app.post(/agent/task) async def create_task(task: TaskInfo, background_tasks: BackgroundTasks): task_id save_task(task) background_tasks.add_task(run_agent_task, task_id) return {task_id: task_id, status: pending}客户端拿到task_id后通过轮询或回调获取最终结果。这样即使 Agent 执行耗时很长接口服务也不会因为超时把所有连接占满。批量任务场景下异常处理还要考虑队列深度和并发控制。不能因为单个任务失败就暂停整个队列也不能让失败任务无限重试挤占队列资源。建议的配置是单个任务最多重试 2 到 3 次重试之间设置退避时间失败超过上限进入死信队列由人工或补偿任务处理。8. 异常处理对资源与性能的影响异常处理不是无代价的。每一层重试、降级、日志记录都会消耗系统资源。重点观察这几个指标重试导致的 token 额外消耗大模型调用重试意味着更多的请求量和 token 消耗成本会明显上升。日志膨胀异常堆栈如果全部打印单条异常可能产生几 KB 到几十 KB 日志长期运行需要做日志采样或压缩。超时时间设置的影响超时设置过短容易导致频繁重试过长则会让任务堆积在内存或队列中。并发上限Executor 线程数、异步任务数量都要限制否则一个高峰就会拖垮进程。异常处理的设计原则是“确定性优先”。重试要有上限等待要有超时兜底要有降级方案。不要试图用无限重试解决模型稳定性问题那只会把成本吃掉。9. 常见问题与排查方法问题现象可能原因排查方式解决方案provider did not respond in timeProvider 服务响应超时或网络波动查看模型调用耗时、错误码和响应日志增加超时设置、重试策略、模型降级Agent 无限循环缺少停止条件或循环条件太弱查看执行日志中的迭代轮数配置 max_iterations、加看门狗输出 JSON 解析失败模型输出格式漂移打印原始输出检查是否混入解释文字增加输出修复步骤限制修复次数工具调用返回 500下游工具服务不稳定查看下游服务日志和响应体对工具调用做重试与熔断批量任务卡住并发线程过多或死锁检查队列深度、线程状态限制并发数加心跳监控接口调用超时Agent 同步执行时间过长检查接口耗时分布改为异步任务 轮询结果日志无异常但任务失败异常被吞掉检查 except 分支是否记录日志统一异常上报禁止静默捕获重试导致成本飙升重试无上限或失败策略过激进查看 token 消耗统计限制重试次数设置退避时间排查时最有效的方法是给每个任务分配唯一的task_id日志、异常上报、队列消息都带上这个 ID。这样即使异常链路过长也能快速串联出完整路径。10. 最佳实践与使用建议结合前面的内容整理一套可以直接落地的 Agent 异常处理最佳实践三层防线都要有。Agent 内部先自愈内部解决不了抛异常调用方硬捕获外层编排器兜底。少一层生产环境都容易出问题。异常信息统一结构化。建议定义统一的异常对象至少包含code、message、task_id、stack_trace四个字段。方便后续接入监控告警系统。设置合理的重试策略。重试次数一般不超过 3 次重试之间要有退避时间避免同时重试造成雪崩。重试必须考虑幂等。涉及外部服务的副作用操作重试前要检查执行状态防止重复执行。日志要记录关键输入输出。模型调用失败时只记录错误码是不够的至少记录请求摘要和响应片段。不在不对等的场景下滥用降级。模型降级前要确认备用模型的功能边界避免降级后无法调用工具。对用户只返回安全信息。代码异常、堆栈、内部服务地址都不应该暴露给最终用户统一转换为友好提示。涉及自动执行、消息发送、数据修改等场景要保留人工审核和终止任务的能力。失败原因要有记录防止自动决策造成不可控影响。11. 总结与下一步Agent 异常处理和传统后端异常处理最大的区别在于不确定性强重试成本高失败链路长。因此不能只依赖某一种手段。先从 Agent 内部做起用提示词规则和输出修复兜住常见问题再在入口处加硬捕获和超时控制最后把任务放进有重试和死信队列的编排体系里。三层都补上之后Agent 服务的稳定性才会有真正的保障。这篇内容建议收藏备用作为 Agent 异常处理的对照清单。下一步可以做三件事整理自己项目中所有可能出现的 Agent 异常类型建立统一错误码给每个 Agent 任务加上task_id和全链路日志接入监控告警对重试率和失败率设置阈值。完成这三步你的 Agent 服务才算真正能上生产。