在FastAPI中实现每分钟Token速率限制

由于FastAPI多Worker是独立进程，进程内全局变量无法共享，直接用全局变量会导致竞态条件和计数不准。以下是几种无需数据库的可行方案：

1. 基于multiprocessing.Manager的跨进程共享状态 #

利用Python标准库的multiprocessing.Manager创建跨进程共享的状态字典，配合进程锁避免竞态。适合不想引入外部依赖的场景。

实现示例 #

from fastapi import FastAPI
from multiprocessing import Manager, Lock
import time
import asyncio
from concurrent.futures import ThreadPoolExecutor

# 初始化跨进程共享状态和锁
manager = Manager()
token_stats = manager.dict({
    "used": 0,
    "cycle_start": time.time()
})
token_lock = Lock()
executor = ThreadPoolExecutor()
app = FastAPI()

def _sync_check_and_update(request_tokens: int, max_per_minute: int) -> bool:
    """同步函数：处理Token计数和周期重置，必须加锁"""
    with token_lock:
        now = time.time()
        # 进入新的分钟周期，重置计数
        if now - token_stats["cycle_start"] >= 60:
            token_stats["used"] = 0
            token_stats["cycle_start"] = now
        
        # 检查是否超过限制
        if token_stats["used"] + request_tokens <= max_per_minute:
            token_stats["used"] += request_tokens
            return True
        else:
            # 等待到下一个周期再处理
            wait_time = 60 - (now - token_stats["cycle_start"])
            time.sleep(wait_time)
            token_stats["used"] = request_tokens
            token_stats["cycle_start"] = time.time()
            return True

async def check_token_quota(request_tokens: int, max_per_minute: int) -> bool:
    """异步包装：避免同步锁阻塞事件循环"""
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(executor, _sync_check_and_update, request_tokens, max_per_minute)

# 接口示例
@app.post("/chat-completion")
async def chat_completion():
    # 用tiktoken预估本次请求的Token消耗（输入+输出）
    estimated_tokens = 150
    max_token_limit = 1000

    if not await check_token_quota(estimated_tokens, max_token_limit):
        return {"status": "error", "message": "Token quota exceeded"}
    
    # 发起OpenAI并发请求（asyncio.gather）
    # ...
    return {"status": "success"}

2. 基于Redis的原子计数 #

Redis的内存键值存储支持原子操作，天然避免竞态，性能更优，适合高并发场景（Redis属于轻量服务，不算传统数据库）。

实现示例 #

from fastapi import FastAPI
import redis
import asyncio

redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
app = FastAPI()

async def check_token_quota(request_tokens: int, max_per_minute: int) -> bool:
    # 原子增加Token计数
    current_used = redis_client.incrby("openai_token_counter", request_tokens)
    # 首次设置时添加60秒过期时间（自动重置周期）
    if current_used == request_tokens:
        redis_client.expire("openai_token_counter", 60)
    
    if current_used > max_per_minute:
        # 超过限制，回滚计数
        redis_client.decrby("openai_token_counter", request_tokens)
        return False
    return True

# 接口示例
@app.post("/chat-completion")
async def chat_completion():
    estimated_tokens = 150
    max_token_limit = 1000

    if not await check_token_quota(estimated_tokens, max_token_limit):
        return {"status": "error", "message": "Token quota exceeded"}
    
    # 执行OpenAI并发请求
    # ...
    return {"status": "success"}

3. 本地文件+文件锁（轻量小众场景） #

用本地文件存储Token计数和周期时间，配合文件锁实现跨进程同步。适合极低流量、不想依赖外部服务的场景，需注意跨平台兼容性。

核心逻辑 #

• 用fcntl（Linux）或msvcrt（Windows）实现文件独占锁
• 每次读写文件前加锁，更新计数后解锁
• 定期检查周期时间，超过60秒则重置计数

关键注意事项 #

• Token预估准确性：必须用OpenAI官方的tiktoken库计算输入输出的实际Token数，避免预估不足导致实际超限制
• 异步兼容：同步锁或文件操作需放在线程池执行，避免阻塞FastAPI的事件循环
• 降级处理：当Token耗尽时，可返回429 Too Many Requests状态码，或自动等待到下一个周期再处理

内容的提问来源于stack exchange，提问作者monopoly