OpenAI API 重试机制怎么写？指数退避、最大重试次数与错误分类

调用大模型 API 时，网络抖动、服务端临时过载、超时等问题随时可能发生。如果没有重试机制，一次偶发错误就会导致用户体验断裂。

这篇文章把重试策略从简单到生产级讲清楚，帮你写出稳健的 API 调用代码。

为什么需要重试

API 调用失败分两类：

类型错误码是否应该重试
临时性错误429（限流）、500/502/503（服务端错误）、超时应该重试
永久性错误400（参数错误）、401（认证失败）、404（模型不存在）不应重试

重试的核心原则：只重试临时性错误，不重试永久性错误。对 400 错误重试 10 次也不会成功，反而浪费时间和配额。

基础重试实现

最简单的重试

import time
from openai import OpenAI, APITimeoutError, APIConnectionError, RateLimitError

client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="YOUR_BASE_URL",
)

def chat_with_retry(messages, model="YOUR_MODEL", max_retries=3):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
            )
            return response
        except (APITimeoutError, APIConnectionError, RateLimitError) as e:
            print(f"第 {attempt + 1} 次失败: {e}")
            if attempt < max_retries - 1:
                time.sleep(2)  # 固定等待 2 秒
            else:
                raise

response = chat_with_retry([
    {"role": "user", "content": "你好"}
])
print(response.choices[0].message.content)

问题很明显：固定等待 2 秒不够灵活。如果服务端需要 10 秒才能恢复，2 秒后重试还是会失败。

指数退避策略

指数退避（Exponential Backoff）是最经典的重试策略：每次重试的等待时间翻倍。

import time
import random
from openai import OpenAI, APITimeoutError, APIConnectionError, RateLimitError

client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="YOUR_BASE_URL",
)

def exponential_backoff_retry(messages, model="YOUR_MODEL", max_retries=5, base_delay=1):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
            )
            return response
        except (APITimeoutError, APIConnectionError, RateLimitError) as e:
            if attempt < max_retries - 1:
                # 指数退避 + 随机抖动
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"第 {attempt + 1} 次失败，{delay:.1f} 秒后重试...")
                time.sleep(delay)
            else:
                raise

response = exponential_backoff_retry([
    {"role": "user", "content": "你好"}
])

等待时间序列：1s → 2s → 4s → 8s → 16s

为什么要加随机抖动？ 如果多个客户端同时失败并同时重试，会造成"重试风暴"，把服务端再次打垮。随机抖动让重试请求分散开。

使用 tenacity 库

生产环境推荐使用 tenacity 库，它提供了更优雅的重试装饰器：

pip install tenacity

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import OpenAI, APITimeoutError, APIConnectionError, RateLimitError

client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="YOUR_BASE_URL",
)

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=60),
    retry=retry_if_exception_type((APITimeoutError, APIConnectionError, RateLimitError)),
    reraise=True,
)
def chat(messages, model="YOUR_MODEL"):
    response = client.chat.completions.create(
        model=model,
        messages=messages,
    )
    return response

response = chat([
    {"role": "user", "content": "你好"}
])
参数说明
stop_after_attempt(5)最多重试 5 次
wait_exponential(multiplier=1, min=1, max=60)指数退避，最短 1 秒，最长 60 秒
retry_if_exception_type(...)只对特定异常重试
reraise=True重试耗尽后抛出原始异常

读取 Retry-After 头

429 响应通常会带 Retry-After 头，告诉你服务端建议的等待时间：

import time
from openai import OpenAI, RateLimitError

client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="YOUR_BASE_URL",
)

def chat_with_retry_after(messages, model="YOUR_MODEL", max_retries=5):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
            )
            return response
        except RateLimitError as e:
            # 尝试读取 Retry-After 头
            retry_after = getattr(e, 'retry_after', None)
            if retry_after:
                delay = float(retry_after)
                print(f"服务端建议等待 {delay} 秒")
            else:
                delay = 2 ** attempt
                print(f"使用指数退避，等待 {delay} 秒")
            
            if attempt < max_retries - 1:
                time.sleep(delay)
            else:
                raise
        except Exception as e:
            if attempt < max_retries - 1:
                delay = 2 ** attempt
                time.sleep(delay)
            else:
                raise

流式输出的重试

流式输出的重试更复杂，因为可能在传输中途断开：

import time
from openai import OpenAI, APITimeoutError, APIConnectionError, RateLimitError

client = OpenAI(
    api_key="YOUR_API_KEY",
    base_url="YOUR_BASE_URL",
)

def stream_with_retry(messages, model="YOUR_MODEL", max_retries=3):
    for attempt in range(max_retries):
        try:
            stream = client.chat.completions.create(
                model=model,
                messages=messages,
                stream=True,
            )
            
            collected = ""
            for chunk in stream:
                if chunk.choices[0].delta.content:
                    content = chunk.choices[0].delta.content
                    collected += content
                    print(content, end="", flush=True)
            
            return collected
            
        except (APITimeoutError, APIConnectionError) as e:
            print(f"\n[中断] 第 {attempt + 1} 次失败: {e}")
            if attempt < max_retries - 1:
                delay = 2 ** attempt
                print(f"[重试] {delay} 秒后继续...")
                time.sleep(delay)
            else:
                raise

result = stream_with_retry([
    {"role": "user", "content": "写一篇 500 字的文章"}
])

带熔断的重试

如果服务端持续不可用，不断重试没有意义。熔断器（Circuit Breaker）可以在连续失败后暂停请求：

import time
from openai import OpenAI

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = 0
        self.state = "closed"  # closed, open, half-open
    
    def call(self, func, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = "half-open"
                print("[熔断器] 进入半开状态，尝试恢复")
            else:
                raise Exception("[熔断器] 服务不可用，请稍后重试")
        
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
    
    def _on_success(self):
        self.failure_count = 0
        self.state = "closed"
    
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
            print(f"[熔断器] 连续失败 {self.failure_count} 次，进入熔断状态")

# 使用
breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60)
client = OpenAI(api_key="YOUR_API_KEY", base_url="YOUR_BASE_URL")

def make_request():
    return client.chat.completions.create(
        model="YOUR_MODEL",
        messages=[{"role": "user", "content": "你好"}],
    )

try:
    response = breaker.call(make_request)
    print(response.choices[0].message.content)
except Exception as e:
    print(f"请求失败: {e}")
状态说明
closed正常状态，请求正常通过
open熔断状态，直接拒绝请求
half-open半开状态，允许少量请求试探恢复

重试策略对比

策略优点缺点适用场景
固定间隔简单可能造成重试风暴内部测试
指数退避逐步加大间隔可能等待过久通用场景
指数退避+抖动避免重试风暴稍复杂生产环境推荐
Retry-After尊重服务端建议依赖服务端返回429 限流场景
熔断器避免无效重试需要额外状态管理高可用场景

快速排错表

问题可能原因解决方法
重试后仍然失败永久性错误被重试只对 429/5xx/超时重试
重试等待太久指数退避上限太高设置 max 上限（如 60 秒）
多个客户端同时重试没有随机抖动添加 jitter
服务端持续不可用没有熔断机制引入 Circuit Breaker

配置检查清单

检查项建议值
最大重试次数3-5 次
退避基数1 秒
最大等待时间60 秒
是否添加抖动是
是否读取 Retry-After是
是否引入熔断高可用场景必须

重试机制是 API 调用的基本保障。指数退避 + 随机抖动 + Retry-After + 熔断器，这四个组合起来就能覆盖绝大多数生产场景。

# python # 开发语言
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

找人做AI项目，是找个干活的还是找个拿主意的？翻译和律师的区别

avatar 2048 AI社区

AI-Infra-Guard 部署与技能扫描实战：Docker 一键起，附我的一次“漏报”复盘

AI-Infra-Guard（A.I.G）是腾讯朱雀实验室推出的开源AI基础设施安全扫描工具，支持Docker一键部署，覆盖模型、运行时、供应链与平台四大安全面。可对Ollama、vLLM等服务进行指纹识别与CVE比对，支持MCP Server与Agent Skills源码审计，并提供CLI工具集成CI流水线。实测中使用本地小模型导致漏报，凸显大模型能力对安全检测的关键影响。强调仅限内网使用，扫描

avatar 2048 AI社区
cover

大模型里的月亮与浪漫：当千古名篇被压缩进 4096 维高维空间

avatar 2048 AI社区