概述

OpenAI API 的 Rate Limit（速率限制）会限制单位时间内的请求次数和 Token 数量。当超出限制时，API 返回 429 错误。本文介绍错误原因、排查方法和重试策略。

Rate Limit 类型

识别 Rate Limit 错误

import openai

try:
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Hello"}]
    )
except openai.error.RateLimitError as e:
    print(f"Rate Limit Error: {e}")
    print(f" Retry-After header: {e.headers.get('retry-after')}")

指数退避重试

import time
import openai
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60)
)
def call_with_retry(messages, model="gpt-4"):
    try:
        return openai.ChatCompletion.create(
            model=model,
            messages=messages
        )
    except openai.error.RateLimitError as e:
        # 检查 Retry-After
        retry_after = e.headers.get('retry-after', 30)
        print(f"Rate limited, waiting {retry_after}s")
        time.sleep(int(retry_after))
        raise  # 让 tenacity 重试

# 使用
result = call_with_retry([{"role": "user", "content": "Hello"}])

多 Key 轮换

import os
from itertools import cycle

API_KEYS = [
    os.getenv("OPENAI_API_KEY_1"),
    os.getenv("OPENAI_API_KEY_2"),
    os.getenv("OPENAI_API_KEY_3")
]

class KeyManager:
    def __init__(self, keys):
        self.keys = cycle(keys)
        self.current = next(self.keys)
        self.key_usage = {k: 0 for k in keys}
    
    def get_key(self):
        return self.current
    
    def rotate(self):
        self.current = next(self.keys)
        print(f"Rotated to new key")
    
    def record_usage(self, tokens):
        self.key_usage[self.current] += tokens

key_manager = KeyManager(API_KEYS)

def call_with_key_rotation(messages):
    for _ in range(len(API_KEYS)):
        openai.api_key = key_manager.get_key()
        try:
            response = openai.ChatCompletion.create(
                model="gpt-4",
                messages=messages
            )
            key_manager.record_usage(
                response["usage"]["total_tokens"]
            )
            return response
        except openai.error.RateLimitError:
            key_manager.rotate()
    raise Exception("All keys exhausted")

预防措施

1. 请求合并：将多个小请求合并为大请求
2. 缓存结果：对相同/相似请求使用缓存
3. 限流中间件：使用 TokenBucket 或 LeakyBucket 算法
4. 监控告警：设置 TPM 使用率告警

参考资料

• OpenAI Rate Limits
• Tenacity 库