失败重试与降级:生产级循环的容错设计
真实的 Agent 活在满是故障的世界里:网络抖一下、API 限流、工具抛异常、参数校验不过。一个"遇错就崩"的 Agent 没法上生产。这节教你一套分类哲学——哪些错该重试、哪些错必须停、怎么优雅降级。
本文导航
- 容错的第一步:分清错误的种类
- 用 pydantic 定义错误类型体系
- 指数退避重试:给网络一点耐心
- 超时分级:连接/读取/总时长
- 错误回填:把失败告诉模型
- 降级链路:优雅的备用方案
- 小结
- 下节预告
第 19 节。上一节我们学会了"停机",这一节处理更现实的问题——运转中出错怎么办。
相信我,不管你的 Agent 写得多优雅,一旦跑起来,错误会像潮水一样涌来:网络超时、428 限流、工具函数 Bug、模型返回 json 解析失败……如果每个错误都直接崩溃,Agent 只配活在 demo 里。生产的标准是:能不崩就不崩,能重试就重试,实在不行优雅降级。
容错的第一步:分清错误的种类
容错设计最忌讳"一锅端"——把所有异常都粗暴重试,或都直接放弃。正确姿势是先分类。我把错误分成两大类,判断标准只有一个:重试有没有可能成功?
| 错误类型 | 例子 | 重试有意义吗 |
|---|---|---|
| 可重试 | 网络超时、连接被重置、429 限流、5xx 服务端错误 | ✅ 有(往往等一下就恢复) |
| 不可重试 | 参数校验失败(pydantic 报错)、401 key 无效、404 资源不存在、模型恶意参数 | ❌ 无(重试一万次也一样) |
可重试的错误,重试。不可重试的错误,别浪费钱,直接走异常流程。这个判断,是整个容错体系的基石。
用 pydantic 定义错误类型体系
为了让错误"可被判断、可被记录",我用 pydantic 给它建一个类型体系(这也符合我们"JSON 一律 pydantic 校验"的铁律)。这样,抛出的每个错误都带有结构化信息,方便后续留痕分析(第 31 节会教怎么把错误写进日志)。
"""errors.py —— 用 pydantic 建立 Agent 容错的错误类型体系(第19节) 运行环境:Python 3.12 + uv 依赖: uv add pydantic """fromenumimportEnumfrompydanticimportBaseModelclassRetryable(str,Enum):"""错误是否可重试"""YES="yes"NO="no"classErrorLevel(str,Enum):"""错误等级:记日志用"""WARNING="warning"# 可重试,重试即可ERROR="error"# 不可重试,但可降级FATAL="fatal"# 致命,必须停机classAgentError(BaseModel):"""统一错误结构"""kind:str# 错误种类,如 network/tool/validationmessage:str# 人类可读的错误信息retryable:Retryable# 能否重试level:ErrorLevel# 错误等级retries_left:int=0# 剩余重试次数(重试后递减)# 几个典型错误的"签名",用一个工厂函数快速构造defmake_network_error(msg:str)->AgentError:returnAgentError(kind="network",message=msg,retryable=Retryable.YES,level=ErrorLevel.WARNING)defmake_validation_error(msg:str)->AgentError:returnAgentError(kind="validation",message=msg,retryable=Retryable.NO,level=ErrorLevel.ERROR)defmake_fatal_error(msg:str)->AgentError:returnAgentError(kind="fatal",message=msg,retryable=Retryable.NO,level=ErrorLevel.FATAL)关键思路:把"可不可重试""什么等级"提前写到错误对象里,后面的重试/降级逻辑就能if err.retryable == Retryable.YES一行判断,不用到处塞 try/except 判断。
指数退避重试:给网络一点耐心
对于可重试错误,最标准的做法是指数退避(Exponential Backoff):第一次失败等base秒,第二次等base*2,第三次等base*4……重试次数越多,等得越久,避免把压力继续压在可能还没恢复的服务上。
"""retry_demo.py —— 指数退避重试(第19节) 运行环境:Python 3.12 + uv 用法: uv run python retry_demo.py """importtime,randomfromerrorsimportAgentError,make_network_error,Retryabledefcall_model():"""模拟一个会偶发抖动的模型调用——30% 概率随机失败"""ifrandom.random()<0.3:raisemake_network_error("Connection reset by peer")return"DeepSeek 返回了正常结果"defretry_with_backoff(func,max_retries=3,base_delay=1.0):"""指数退避重试包装器:base_delay, 2x, 4x"""forattemptinrange(max_retries+1):# 1 次原始 + 3 次重试try:returnfunc()exceptAgentErrorase:ife.retryable!=Retryable.YESorattempt==max_retries:raise# 不可重试或重试耗尽,抛出去delay=base_delay*(2**attempt)# 1s, 2s, 4sprint(f" 第{attempt+1}次失败,{delay:.0f}s 后重试…")time.sleep(delay)if__name__=="__main__":print("开始调用(含指数退避重试)…")result=retry_with_backoff(call_model)print("最终结果:",result)运行输出(30% 失败率下可能看到):
开始调用(含指数退避重试)… 第1次失败,1s 后重试… 第2次失败,2s 后重试… 最终结果: DeepSeek 返回了正常结果退避间隔 1s → 2s → 4s,指数增长。重试不是"疯狂快速点重试",而是"有耐心地等一会再试"——这才是对远端服务的尊重,也更容易成功。
超时分级:连接/读取/总时长
重试之外,超时是另一层保护。一个没设超时的请求,可能卡住几分钟甚至挂死你的整个 Agent。生产级做法是给请求设多级超时:
| 超时级别 | 含义 | 建议值 |
|---|---|---|
| 连接超时 | 建立 TCP 连接的最长时间 | 3-5s |
| 读取超时 | 每读到一段数据的最长间隔 | 30-60s |
| 总超时 | 整个调用(含生成)的最长时长 | 120s+(DeepSeek 生成慢时) |
OpenAI SDK 里可以直接配:
fromopenaiimportOpenAI client=OpenAI(base_url="https://api.deepseek.com",api_key=os.environ["DEEPSEEK_API_KEY"],timeout=60.0,# 总超时max_retries=0,# 关闭 SDK 内建重试,交给我们的重试层统一管)我通常把 SDK 的max_retries设为 0,因为 SDK 内建的重试规则太"黑盒"——它重试什么比例、退避多久,你无法精细控制。把重试权收回自己手里,用自己那套AgentError.retryable判断,才能保证留痕完整(第 31 节)、成本可控。
错误回填:把失败告诉模型
工具执行失败时,有个很容易被忽视但极其有效的技巧:把错误作为工具结果回填给模型,让它自己决定怎么办。
前面第 17 节回填的是"成功结果",这里回填"失败信息"。模型看到"调用 get_weather 报错:参数 city 不能为空",它会自己调整参数重试,或者换一条路走,甚至向用户解释。这比你在代码里写死一堆"如果失败就 XXX"强得多——把"应对失败"的智能也交给模型。
tool_result={"ok":False,"error":"参数 city 不能为空字符串"}messages.append({"role":"tool","tool_call_id":tc.id,"content":json.dumps(tool_result,ensure_ascii=False),# 把结构化错误写进 content})这里的原则是:非致命的失败,尽量作为 Observation 喂回去,而不是直接停机。让 model 自己权衡。只有不可重试的硬错(key 无效、schema 崩了),才真正走降级/停机。
降级链路:优雅的备用方案
重试耗尽或遇到不可重试错误,还有一个动作叫降级(degradation)——不求完美完成,但求优雅收场。我常用的降级链(从重到轻):
具体到 DeepSeek 场景:
- API 主模型失败→ 可降级到同协议的另一模型(OpenAI 兼容的好处,第 10 节讲过 base_url 一换就换模型)。这就是第 65 节《多模型路由》的雏形。
- 工具执行失败→ 回填错误给模型,让它换工具或改参数。
- 全部失败→ 返回"部分成功 + 失败说明",人类接手,绝不能静默失败。
降级的核心思想:与其完美地死掉,不如狼狈地活下来。用户收到一句"我完成了前 2 步,第 3 步因为 X 失败",远比收到一个神秘崩溃强一万倍。
小结
- 先分类再处理:可重试的(网络/限流/5xx)重试,不可重试的(参数/401/404)别浪费钱。
- 用 pydantic 建错误体系:把
retryable、level写进错误对象,重试/降级逻辑一行判断。 - 指数退避是标准重试:1s、2s、4s……越重试越耐心,别疯狂点。
- 超时分三级:连接/读取/总时长分开设,SDK 的 max_retries 关掉交给自己管。
- 失败也能回填给模型:把错误当 Observation 喂回去,让模型自己找退路。
- 降级优于崩溃:有备用方案就用,没有就返回部分结果+说明,绝不静默失败。
下节预告
重试、超时、降级都能处理"偶发故障",但还有个无底洞风险没堵死:模型一旦失控,可能瞬间烧掉巨额 token、卡死系统。下节,我们给 Agent 系上真正的安全带——token 预算熔断、时间熔断、工具调用次数熔断的三重保险。
如果觉得本文对你有帮助,欢迎点赞、收藏、关注三连!
本系列持续更新中,关注不迷路~