OpenAI API RateLimitError 深度解析与应对策略

from openai import OpenAI
import time

client = OpenAI()

def make_openai_request():
    try:
        # 使用 .with_raw_response 包装 API 调用以获取原始 HTTP 响应
        raw_response = client.chat.completions.with_raw_response.create(
          model="gpt-3.5-turbo",
          messages=[
            {"role": "system", "content": "You are a poetic assistant, skilled in explaining complex programming concepts with creative flair."},
            {"role": "user", "content": "Compose a poem that explains the concept of recursion in programming."}
          ]
        )

        # 解析响应以获取 chat completion 对象
        chat_completion = raw_response.parse()
        # 获取原始 HTTP 响应头部
        response_headers = raw_response.headers

        print("API 调用成功！")
        print("Chat Completion:", chat_completion.choices[0].message)
        print("\n响应头部信息:")
        for header, value in response_headers.items():
            if header.lower().startswith('x-ratelimit'):
                print(f"  {header}: {value}")
        return chat_completion

    except Exception as e:
        print(f"API 调用失败: {e}")
        # 在错误情况下，尝试获取并打印任何可用的响应头部（如果错误对象包含）
        if hasattr(e, 'response') and hasattr(e.response, 'headers'):
            print("\n错误响应头部信息:")
            for header, value in e.response.headers.items():
                if header.lower().startswith('x-ratelimit'):
                    print(f"  {header}: {value}")
        return None

# 执行 API 请求
make_openai_request()

import time
import random
from openai import OpenAI
from openai import RateLimitError

client = OpenAI()

def call_openai_with_retries(messages, max_retries=5):
    retries = 0
    delay = 1 # 初始等待秒数

    while retries < max_retries:
        try:
            completion = client.chat.completions.create(
                model="gpt-3.5-turbo",
                messages=messages
            )
            return completion
        except RateLimitError as e:
            retries += 1
            print(f"RateLimitError 发生，第 {retries} 次重试...")
            if retries == max_retries:
                print("达到最大重试次数，放弃请求。")
                raise e

            # 检查响应头中的 reset 时间，如果可用则优先使用
            # 注意：在 RateLimitError 异常对象中直接获取响应头可能需要更复杂的处理
            # 这里的 delay 是一个通用指数退避策略
            wait_time = delay * (2 ** (retries - 1)) + random.uniform(0, 1) # 指数增长并添加抖动
            print(f"等待 {wait_time:.2f} 秒后重试...")
            time.sleep(wait_time)
        except Exception as e:
            print(f"发生其他错误: {e}")
            raise e
    return None

# 示例使用
try:
    response = call_openai_with_retries([
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello, how are you?"}
    ])
    if response:
        print(response.choices[0].message.content)
except Exception as e:
    print(f"最终未能完成请求: {e}")