为什么你的 AI 应用需要限流
接 OpenAI 或 Claude API 的时候，很多人吃过 429 Too Many Requests 的亏。用户突发流量打过来，API 额度瞬间烧完，后续请求全部排队等失败。OpenAI 官方文档明确说，rate limit 按 RPM（每分钟请求数）、TPM（每分钟 token 数）、TPD（每天 token 数）多个维度同时限制，哪个先触顶就限哪个。你的应用层如果不做限流保护，等于裸奔。
限流不只是防 429。控制成本同样重要。一个用户疯狂调接口，一天的 API 账单可能比你预期的高三倍。在应用层加一道限流闸门，既保护下游 API，也控制自己的支出。
选 Redis 滑动窗口还是固定窗口
常见的限流算法有四种：固定窗口、滑动窗口、令牌桶、漏桶。固定窗口最简单，但有个问题：在窗口切换的瞬间，可能会有两倍流量涌入。比如每分钟限 100 次，在 0:59 和 1:00 之间各来 100 次，用户实际在 1 秒内发了 200 次请求。
滑动窗口把时间轴切成更细的粒度（比如每秒一个子窗口），统计最近 60 秒内所有子窗口的总和。这样窗口切换时的流量毛刺就平滑掉了。对 AI API 来说，这个精度差异很关键，因为一次 LLM 调用的 token 消耗可能很大。
令牌桶适合需要突发流量的场景（允许短时间超限），漏桶适合严格匀速。AI API 场景下，滑动窗口是性价比最高的选择。
完整实现：Redis + Lua 脚本
下面这个模板可以直接用在生产环境。用 Redis 执行 Lua 脚本保证原子性，避免并发请求竞争导致计数不准。
docker-compose.yml
先把 Redis 跑起来：
version: '3.8'\nservices:\n redis:\n image: redis:7-alpine\n ports:\n - "6379:6379"\n command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru\n restart: unless-stoppedrate_limiter.py
核心限流逻辑：
import time\nimport redis\n\nclass SlidingWindowRateLimiter:\n def __init__(self, redis_client: redis.Redis):\n self.redis = redis_client\n self.script = \"\"\"\n local key = KEYS[1]\n local now = tonumber(ARGV[1])\n local window_ms = tonumber(ARGV[2])\n local max_requests = tonumber(ARGV[3])\n local window_key = key .. ':window'\n \n redis.call('ZREMRANGEBYSCORE', window_key, 0, now - window_ms)\n local current = redis.call('ZCARD', window_key)\n \n if current >= max_requests then\n local oldest = redis.call('ZRANGE', window_key, 0, 0, 'WITHSCORES')\n return {0, oldest[2]}\n end\n \n local member = now .. ':' .. math.random(1000000, 9999999)\n redis.call('ZADD', window_key, now, member)\n redis.call('EXPIRE', window_key, math.ceil(window_ms / 1000) + 1)\n return {1, current + 1}\n \"\"\"\n self.script_sha = self.redis.script_load(self.script)\n \n def check(self, identifier: str, max_requests: int, window_seconds: int = 60):\n key = f\"ratelimit:{identifier}\"\n now_ms = int(time.time() * 1000)\n window_ms = window_seconds * 1000\n \n try:\n result = self.redis.evalsha(\n self.script_sha, 1, key,\n now_ms, window_ms, max_requests\n )\n allowed = bool(result[0])\n count = int(result[1])\n if not allowed:\n oldest_time = int(result[1]) / 1000\n retry_after = oldest_time + window_seconds - time.time()\n return False, count, max(retry_after, 0.1)\n return True, count, 0\n except redis.exceptions.NoScriptError:\n self.script_sha = self.redis.script_load(self.script)\n return self.check(identifier, max_requests, window_seconds)中间件集成（FastAPI 示例）
from fastapi import FastAPI, Request\nfrom fastapi.responses import JSONResponse\nimport redis\nfrom rate_limiter import SlidingWindowRateLimiter\n\napp = FastAPI()\nr = redis.Redis(host='localhost', port=6379, db=0)\nlimiter = SlidingWindowRateLimiter(r)\n\n@app.middleware('http')\nasync def rate_limit_middleware(request: Request, call_next):\n user_id = request.headers.get('X-User-ID', request.client.host)\n \n if '/api/chat' in request.url.path:\n max_req = 20\n else:\n max_req = 100\n \n allowed, count, retry_after = limiter.check(user_id, max_req, 60)\n \n if not allowed:\n return JSONResponse(\n status_code=429,\n content={\n 'error': 'rate_limit_exceeded',\n 'message': f'每分钟限制 {max_req} 次',\n 'retry_after': round(retry_after, 1)\n },\n headers={'Retry-After': str(int(retry_after) + 1)}\n )\n \n response = await call_next(request)\n response.headers['X-RateLimit-Limit'] = str(max_req)\n response.headers['X-RateLimit-Remaining'] = str(max_req - count)\n return response几个实际使用建议
按用户分层限流。免费用户每分钟 5 次，付费用户每分钟 50 次。用 identifier 区分，比如 ratelimit:free:user123 和 ratelimit:pro:user123。
设置好 Retry-After 响应头。OpenAI 的 SDK 会自动读取这个头来决定重试等待时间。你不设的话，客户端可能立即重试，形成恶性循环。
Redis 内存别给太多。限流数据是短命的，256MB 足够支撑几万并发用户。设 maxmemory-policy allkeys-lru 让 Redis 在内存满时自动淘汰最旧的数据，不会因为限流把 Redis 撑爆。
监控限流触发率。如果某个用户频繁触发 429，要么是他用法异常，要么是你的限额设得太低。把触发事件打到日志里，定期看一下。
能直接拿走用的东西
上面的代码加起来不到 100 行，复制粘贴改一下 Redis 连接配置就能跑。FastAPI 中间件的写法换到 Flask 或 Django 也差不多，核心逻辑是通用的。
需要这套东西但不想自己搭的，docker-compose 文件直接用，Redis 跑起来配上 Python 脚本就完事。Lua 脚本已经处理好原子性问题，不用自己操心并发竞争。
