当前位置：首页 > news >正文

Python 如何实现 AI API 的自动重试与故障恢复：从异常捕获到退避策略

news 2026/8/9 4:33:51

在调用 AI 接口时，偶发的网络闪断、上游超时或限流错误很常见。如果程序遇到错误直接中断，用户体验会受到影响。本文介绍一种适合个人项目的自动重试与故障恢复实现方法。

为什么基础异常捕获不够？

很多项目在调用接口时会写这样的代码：

try:response=client.chat.completions.create(...)exceptExceptionasexc:print("请求失败：",exc)

这段代码能防止程序崩溃，但它无法解决实际问题：

临时网络抖动时，请求直接放弃
遇到限流错误时没有等待就再次发起请求，导致失败加剧
无法区分“可以重试的临时错误”和“不能重试的参数错误”

因此，成熟的调用层需要配合自动重试（Retry）与退避策略（Backoff）。

一、明确哪些错误可以重试

不是所有异常都适合重试。盲目重试非但无用，还可能加重上游服务负担。

1. 适合重试的错误

连接中断（APIConnectionError）
请求超时（APITimeoutError）
服务端临时错误（5xx 状态码或APIError）
限流错误（RateLimitError）

2. 不适合重试的错误

鉴权失败（401 错误，Key 无效）
请求参数错误（400 错误，如模型名称写错、消息格式非法）
额度不足

如果请求参数本身存在问题，重试一百次也无法成功，应当直接报错并终止。

二、使用指数退避减少连续冲击

如果请求因为限流（Rate Limit）失败，立即重试往往会再次失败。合理的做法是逐渐延长重试等待时间，也就是指数退避（Exponential Backoff）：

第 1 次失败 → 等待 2 秒 第 2 次失败 → 等待 4 秒 第 3 次失败 → 等待 8 秒

同时可以加入随机抖动（Jitter），避免多个客户端在同一时间同时发起重试，造成惊群效应。

三、Python 实现带重试的请求封装

下面是一个结合了异常过滤与退避等待的完整封装示例：

importtimeimportrandomfromopenaiimportOpenAIfromopenaiimportAPIError,APIConnectionError,APITimeoutError,RateLimitError client=OpenAI(api_key="your-api-key",base_url="https://your-api-domain.com/v1",timeout=20.0)RETRYABLE_ERRORS=(APIConnectionError,APITimeoutError,RateLimitError,)defask_llm_with_retry(prompt:str,model:str="your-model-name",max_retries:int=3)->str:last_exception=Noneforattemptinrange(1,max_retries+1):try:response=client.chat.completions.create(model=model,messages=[{"role":"system","content":"你是一个专业的技术助手。"},{"role":"user","content":prompt}])returnresponse.choices[0].message.contentexceptRETRYABLE_ERRORSasexc:last_exception=excifattempt==max_retries:break# 计算退避时间：2的次方 + 随机抖动sleep_time=(2**attempt)+random.uniform(0,1)print(f"请求临时失败 ({exc})，将在{sleep_time:.2f}秒后进行第{attempt+1}次重试...")time.sleep(sleep_time)exceptExceptionasexc:# 不可恢复的错误直接抛出，不进行重试raiseRuntimeError(f"请求发生不可恢复错误：{exc}")fromexcraiseRuntimeError(f"请求失败，已达到最大重试次数{max_retries}。最后错误：{last_exception}")

四、这段代码的核心设计

1. 明确的重试边界

通过max_retries=3限制最大重试次数，防止程序陷入死循环。

2. 异常精准匹配

只捕获RETRYABLE_ERRORS，像参数错误（BadRequestError）等会直接进入底部的Exception分支并抛出。

3. 指数退避与抖动

2 ** attempt让等待时间随着重试次数递增，random.uniform(0, 1)错开并发请求的时间点。

五、在日志中记录重试轨迹

为了方便后续排查稳定性问题，建议在重试时记录日志：

importlogging logging.basicConfig(level=logging.INFO,format="%(asctime)s | %(levelname)s | %(message)s")# 在捕获到可重试异常时：logging.warning(f"API temporary failure | attempt={attempt}| error={exc}| retry_in={sleep_time:.2f}s")