ChatGPT访问优化实战：解决高并发场景下的API限流与稳定性问题

1次阅读

没有评论

共计 2711 个字符，预计需要花费 7 分钟才能阅读完成。

在高并发场景下调用 ChatGPT API 时，开发者经常会遇到以下典型问题：

429 错误 ：OpenAI 对 API 调用有严格的速率限制，突发流量很容易触发 HTTP 429(Too Many Requests) 响应
响应超时：当服务器负载较高时，API 响应时间会明显延长，导致客户端超时
Token 消耗不可控：长文本交互会快速消耗账号的 Token 配额，难以预测用量
成本激增：失败的请求仍然会计费，重试机制不当会导致费用飙升
服务质量波动：不同时间段的 API 响应质量差异明显，缺乏稳定性保障

针对这些问题，常见的解决方案有以下几种：

简单轮询
实现简单，直接循环调用 API
无法应对突发流量，容易触发限流
资源利用率低
请求队列
将请求排队处理，平滑流量峰值
需要额外存储系统支持
可能引入延迟
熔断器模式
当错误率达到阈值时自动停止请求
防止雪崩效应
需要精细调参
Token 桶算法
精确控制请求速率
支持突发流量
实现复杂度较高

经过实践对比，我们发现 Redis 队列 +Token 桶 + 智能重试 的组合方案效果最佳。

使用 Redis 的 List 数据结构实现生产者 - 消费者模型：

import redis
import json

class ChatGPTQueue:
    def __init__(self, redis_host='localhost', redis_port=6379):
        self.redis = redis.Redis(host=redis_host, port=redis_port)
        self.queue_key = 'chatgpt:request:queue'

    def enqueue(self, request_data):
        """
        将请求加入队列
        :param request_data: 包含 prompt 等参数的 dict
        """
        self.redis.rpush(self.queue_key, json.dumps(request_data))

    def dequeue(self):
        """
        从队列获取请求
        :return: 请求数据 dict 或 None
        """
        result = self.redis.lpop(self.queue_key)
        return json.loads(result) if result else None

采用指数退避算法，并加入随机抖动 (jitter) 防止重试风暴：

import random
import time

def call_with_retry(func, max_retries=5, initial_delay=1):
    """
    带智能重试的 API 调用
    :param func: 要执行的 API 调用函数
    :param max_retries: 最大重试次数
    :param initial_delay: 初始延迟(秒)
    """
    retry_count = 0
    while retry_count <= max_retries:
        try:
            response = func()
            return response
        except Exception as e:
            if getattr(e, 'status_code', None) == 429:
                # 指数退避 + 随机抖动
                delay = initial_delay * (2 ** retry_count) + random.uniform(0, 1)
                time.sleep(min(delay, 60))  # 最大不超过 60 秒
                retry_count += 1
            else:
                raise
    raise Exception(f"API 调用失败，重试 {max_retries} 次后仍不成功")

使用 Redis 实现分布式 Token 桶：

class TokenBucket:
    def __init__(self, capacity, fill_rate, redis_conn):
        """
        :param capacity: 桶容量
        :param fill_rate: 每秒补充的 token 数
        :param redis_conn: Redis 连接
        """
        self.capacity = capacity
        self.fill_rate = fill_rate
        self.redis = redis_conn
        self.key = 'chatgpt:token_bucket'

    def consume(self, tokens=1):
        """
        尝试消费指定数量的 token
        :param tokens: 需要的 token 数
        :return: 是否成功
        """
        now = time.time()
        bucket = self.redis.hgetall(self.key)

        if not bucket:
            # 初始化桶
            bucket = {'tokens': str(self.capacity - tokens), 'timestamp': str(now)}
            self.redis.hmset(self.key, bucket)
            return True

        # 计算应补充的 token
        last_time = float(bucket[b'timestamp'])
        elapsed = now - last_time
        new_tokens = float(bucket[b'tokens']) + elapsed * self.fill_rate
        current_tokens = min(new_tokens, self.capacity)

        if current_tokens >= tokens:
            # 有足够 token
            bucket = {'tokens': str(current_tokens - tokens), 
                'timestamp': str(now)
            }
            self.redis.hmset(self.key, bucket)
            return True
        return False

我们对优化前后进行了对比测试（测试环境：4 核 8G 服务器，100 并发）：