AI API速率限制(Rate Limit)全面应对策略：从429到丝滑并发的实战之路

import time
import threading

class TokenBucket:
    def __init__(self, rate, capacity):
        self.rate = rate          # 每秒放入的令牌数
        self.capacity = capacity  # 桶的最大容量
        self.tokens = capacity    # 当前令牌数
        self.last_time = time.time()
        self.lock = threading.Lock()

    def consume(self, tokens=1):
        with self.lock:
            now = time.time()
            # 补充令牌
            elapsed = now - self.last_time
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.rate
            )
            self.last_time = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    def wait_and_consume(self, tokens=1):
        # 等待直到有足够的令牌
        while not self.consume(tokens):
            time.sleep(0.1)

# 使用示例：限制为每秒8个请求，允许突发到20个
bucket = TokenBucket(rate=8, capacity=20)

for i in range(50):
    bucket.wait_and_consume()
    # 发送API请求...
    print(f"请求 {i+1} 已发送")

import random
import time
from openai import OpenAI, RateLimitError

client = OpenAI()

def call_with_retry(messages, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=messages
            )
            return response
        except RateLimitError as e:
            # 优先使用API返回的Retry-After
            retry_after = e.response.headers.get("Retry-After")
            if retry_after:
                wait = float(retry_after)
            else:
                # 指数退避 + 随机抖动
                base_wait = 2 ** attempt
                jitter = random.uniform(0, 1) * base_wait
                wait = base_wait + jitter

            print(f"429触发，第{attempt+1}次重试，等待{wait:.1f}秒...")
            time.sleep(wait)

    raise Exception(f"重试{max_retries}次后仍然失败")

# 使用示例
result = call_with_retry([
    {"role": "user", "content": "写一首关于代码的诗"}
])

import itertools
import threading
from openai import OpenAI

class MultiKeyRouter:
    def __init__(self, api_keys):
        self.clients = [OpenAI(api_key=key) for key in api_keys]
        self.key_states = [{"rpm_remaining": 500, "tpm_remaining": 40000} for _ in api_keys]
        self.lock = threading.Lock()
        self.index = 0

    def get_client(self):
        with self.lock:
            # 轮询选择，跳过剩余额度不足的Key
            tried = 0
            while tried < len(self.clients):
                state = self.key_states[self.index]
                if state["rpm_remaining"] > 5:
                    client = self.clients[self.index]
                    state["rpm_remaining"] -= 1
                    self.index = (self.index + 1) % len(self.clients)
                    return client
                self.index = (self.index + 1) % len(self.clients)
                tried += 1
            raise Exception("所有API Key额度已耗尽")

    def update_limits(self, key_index, response_headers):
        # 根据响应头更新剩余额度
        self.key_states[key_index]["rpm_remaining"] = int(
            response_headers.get("x-ratelimit-remaining-requests", 0)
        )

# 使用示例
router = MultiKeyRouter([
    "sk-proj-xxx1",
    "sk-proj-xxx2",
    "sk-proj-xxx3",
])

client = router.get_client()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)

# producer.py - 生产者：将请求推入队列
import redis
import json
import uuid

r = redis.Redis(host='localhost', port=6379, db=0)

def enqueue_request(messages, priority=0):
    # priority越大越优先处理
    task = {
        "id": str(uuid.uuid4()),
        "messages": messages,
        "priority": priority
    }
    # 使用有序集合实现优先级队列
    score = -priority  # Redis ZSET分数越小越靠前
    r.zadd("ai_request_queue", {json.dumps(task): score})
    return task["id"]

# 消费者可以从结果队列获取响应
# r.get(f"result:{task_id}")

# ---
# consumer.py - 消费者：以安全速率消费队列
import time
from openai import OpenAI

client = OpenAI()
bucket = TokenBucket(rate=8, capacity=15)  # 复用前面的令牌桶

def process_queue():
    while True:
        # 取出优先级最高的任务
        result = r.zpopmin("ai_request_queue")
        if not result:
            time.sleep(0.5)
            continue

        task = json.loads(result[0][0])
        bucket.wait_and_consume()  # 限流控制

        try:
            response = client.chat.completions.create(
                model="gpt-4o",
                messages=task["messages"]
            )
            # 将结果写入结果队列
            r.setex(
                f"result:{task['id']}",
                3600,  # 1小时过期
                response.choices[0].message.content
            )
        except Exception as e:
            r.setex(f"error:{task['id']}", 3600, str(e))

# 启动多个消费者worker
import threading
for i in range(3):
    t = threading.Thread(target=process_queue, daemon=True)
    t.start()

from openai import OpenAI, RateLimitError

client = OpenAI()

# 模型降级链：从强到弱
MODEL_CHAIN = ["gpt-4o", "gpt-4o-mini", "gpt-3.5-turbo"]

def classify_complexity(messages):
    # 简单启发式：根据输入长度和关键词判断复杂度
    text = " ".join(m["content"] for m in messages)
    if len(text) > 3000 or "分析" in text or "推理" in text:
        return "high"
    elif len(text) > 500:
        return "medium"
    return "low"

def smart_call(messages):
    complexity = classify_complexity(messages)
    start_idx = {"high": 0, "medium": 1, "low": 2}[complexity]

    for i in range(start_idx, len(MODEL_CHAIN)):
        model = MODEL_CHAIN[i]
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages
            )
            if i > start_idx:
                print(f"已降级到 {model}")
            return response
        except RateLimitError:
            print(f"{model} 触发限流，尝试降级...")
            continue

    raise Exception("所有模型均不可用")

# 简单问题直接用mini，复杂问题才用4o
result = smart_call([
    {"role": "user", "content": "帮我翻译：Hello World"}
])