ChatGPT API调用限额优化实战：从配额管理到分布式调度

1次阅读

没有评论

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

最近在使用 ChatGPT API 开发时，经常遇到 HTTP 429 错误（请求过多），导致业务中断。经过分析，发现官方对 API 调用有以下限制：

每分钟最多 60 次请求（RPM）
每天最多 40000 次请求（RPD）

这些限制对高并发场景下的业务连续性影响很大，特别是当多个服务实例共享同一 API key 时，很容易触发限流。

最简单的实现方式是每个服务实例维护一个本地计数器：

优点：实现简单，无外部依赖
缺点：多实例间无法共享计数，容易超限

我们最终选择了基于 Redis 的分布式方案：

使用 Redis INCR 命令实现原子计数器
Lua 脚本保证操作的原子性
二级缓存（本地内存）减少 Redis 访问

-- rate_limiter.lua
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local expire_time = ARGV[2]
local current = redis.call('GET', key)

if current and tonumber(current) > limit then
    return 0
end

current = redis.call('INCR', key)
if tonumber(current) == 1 then
    redis.call('EXPIRE', key, expire_time)
end

return 1

import redis
import time

class RateLimiter:
    def __init__(self, redis_conn, key_prefix='chatgpt:'):
        self.redis = redis_conn
        self.key_prefix = key_prefix
        # 本地缓存最近 60 秒的计数
        self.local_cache = {}
        self.local_cache_expire = 60

    def is_allowed(self, api_key, limit=60, window=60):
        """检查是否允许调用 API"""
        key = f"{self.key_prefix}{api_key}"
        now = int(time.time())

        # 先检查本地缓存
        if key in self.local_cache and now - self.local_cache[key]['time'] < self.local_cache_expire:
            if self.local_cache[key]['count'] >= limit:
                return False
            self.local_cache[key]['count'] += 1
            return True

        # 调用 Redis Lua 脚本
        allowed = self.redis.eval("""local current = redis.call('GET', KEYS[1])
            if current and tonumber(current) >= tonumber(ARGV[1]) then
                return 0
            end
            redis.call('INCR', KEYS[1])
            if tonumber(redis.call('GET', KEYS[1])) == 1 then
                redis.call('EXPIRE', KEYS[1], ARGV[2])
            end
            return 1""",
            1, key, limit, window
        )

        # 更新本地缓存
        self.local_cache[key] = {'count': 1, 'time': now}
        return bool(allowed)

在集群环境下，我们需要注意：