如何用Python+tenacity优化OpenAI API调用的速率限制处理？

批量处理2-3k条长prompt请求时触发RateLimitError且重试无效，核心问题大概率是并发量失控、重试策略不符合OpenAI要求，以及没利用好官方客户端的限流机制。以下是具体的优化方案和问题排查点：

一、先排查现有实现的常见坑 #

1. tenacity重试的错误用法 #

• 没限定重试的错误类型：如果把所有异常都重试，比如无效prompt的错误，只会浪费资源
• 重试间隔太固定：没有指数退避+抖动，容易引发重试风暴，反而加重限流
• 重试次数过多：超过OpenAI的容忍阈值，直接被临时限制请求

2. asyncio并发无限制 #

异步请求如果一次性把2-3k个任务全部丢进事件循环，瞬间打满OpenAI的并发配额，触发限流后重试又继续发，形成恶性循环。

3. 忽略官方客户端的原生支持 #

新版openai-python已经自带了符合OpenAI规范的重试和速率控制，没必要自己从零写tenacity逻辑。

二、具体优化步骤 #

1. 修正tenacity重试策略（如果坚持用） #

只针对RateLimitError重试，用指数退避加抖动，设置合理的最大重试次数：

from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
from openai import RateLimitError

@retry(
    stop=stop_after_attempt(6),  # 最多重试6次，覆盖大部分临时限流场景
    wait=wait_exponential_jitter(initial=1, max=60),  # 间隔从1秒开始指数增长，加抖动避免集中重试
    retry=retry_if_exception_type(RateLimitError),
    reraise=True
)
async def call_openai(client, params):
    return await client.chat.completions.create(**params)

2. 用Semaphore严格控并发 #

根据你的OpenAI账号配额设置并发数，比如gpt-3.5-turbo的默认配额是每分钟3500请求、每分钟90k token，长prompt的话token消耗快，并发数要降到50甚至更低：

import asyncio

async def process_single(client, params, semaphore):
    async with semaphore:
        try:
            return await call_openai(client, params)
        except Exception as e:
            # 记录失败请求，方便后续重跑
            print(f"Failed: {params['messages'][-1]['content'][:50]} | {str(e)}")
            return None

async def process_batch(client, params_list):
    # 限制并发数为50，根据自己的配额调整
    semaphore = asyncio.Semaphore(50)
    tasks = [asyncio.create_task(process_single(client, p, semaphore)) for p in params_list]
    return await asyncio.gather(*tasks, return_exceptions=True)

3. 改用官方客户端的原生重试（推荐） #

新版openai-python会自动识别Retry-After响应头，比自己写的tenacity更准确，初始化时直接配置：

from openai import AsyncOpenAI

client = AsyncOpenAI(
    api_key="your_api_key",
    max_retries=6,  # 自动重试次数
    timeout=30,  # 避免长时间挂起
)

用这个客户端的话，甚至可以不用自己加tenacity装饰器，官方已经处理了大部分限流场景。

4. 长prompt的额外优化 #

长prompt占token多，会快速耗尽每分钟token配额，要：

• 用tiktoken提前计算token数，过滤或拆分超长prompt
• 进一步降低并发数，比如降到30以下

import tiktoken

def count_prompt_tokens(messages, model="gpt-3.5-turbo"):
    encoder = tiktoken.encoding_for_model(model)
    total = 0
    for msg in messages:
        total += len(encoder.encode(msg["content"]))
    return total

# 预处理：过滤token超标的请求，或者拆分
filtered_params = [p for p in params_list if count_prompt_tokens(p["messages"]) < 8000]

5. 断点续传与错误记录 #

把失败的请求保存到文件，比如JSON格式，后续单独处理，避免重复跑成功的请求：

import json

# 处理完后筛选失败的请求
failed_params = [params for params, result in zip(params_list, results) if result is None]
with open("failed_requests.json", "w") as f:
    json.dump(failed_params, f, indent=2)

三、验证与调优 #

1. 先拿100条测试，观察是否触发限流，调整并发数
2. 去OpenAI API面板查看自己的实际配额，根据请求数/分钟、token数/分钟来调整参数
3. 监控成功率，逐步提升批量大小，找到最稳定的配置

内容的提问来源于stack exchange，提问作者Ivo