在vLLM上建高并发中转站，身份认证和配额限流建议放在中转站网关层实现，不要让vLLM实例直接暴露给客户端。网关层负责校验每个请求的API Key、决定是否放行，并限制同一Key的请求速率和总量。这样做的直接原因是：vLLM本身只负责推理和并发调度，不提供用户维度的多租户配额管理；若把配额逻辑写进多个vLLM实例，会造成分散且难以统一调整。下文给出一个可落地的认证和限流骨架，以及对应的验证命令。
认证与配额限流应放在哪个层
中转站可以基于FastAPI、Go或Nginx+Lua实现。推荐用独立网关进程，把API Key校验、速率限制、配额统计放在请求进入后端之前处理。vLLM实例只保留模型并行和并发上限（如`--max-num-seqs`）这样的服务端保护参数，用户维度限制全部由网关负责。
请求身份认证：API Key校验骨架
下面是一个FastAPI依赖函数的示例，用于从Authorization头中提取Bearer Token或裸Key并执行查询。实际使用时，请将get_user_by_key替换为你的数据库/缓存查询逻辑。
from fastapi import FastAPI, Depends, HTTPException, Header
app = FastAPI()
def get_user_by_key(api_key: str):
# 这里只做演示，实际请查询你的用户表
# 返回 dict 或 None
if api_key == 'test-key':
return {'id': 'user_001', 'rpm_limit': 60}
return None
def verify_api_key(authorization: str = Header(..., alias='Authorization')):
if not authorization:
raise HTTPException(status_code=401, detail='缺少Authorization头')
# 兼容 'Bearer xxxx' 和裸key
if authorization.startswith('Bearer '):
token = authorization[7:]
else:
token = authorization
user = get_user_by_key(token)
if user is None:
raise HTTPException(status_code=401, detail='无效API Key')
return user这个依赖可以挂在任意需要保护的router上。如果中转站同时服务多个上游模型，建议把API Key映射到可访问的模型清单，在同一个函数里提前过滤不可用模型。
配额限流：令牌桶与Redis滑动窗口
限流维度至少包含RPM（每分钟请求数）和TPM（每分钟Token消耗数）。最简单可用Redis滑动窗口或令牌桶。下面用Redis的INCR和EXPIRE实现每分钟的固定窗口限流，适合快速部署，但存在窗口边界突发。如果要求更平滑，可改用令牌桶脚本或Lua实现。
import redis
import time
from fastapi import HTTPException
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
def check_rpm(user_id: str, rpm_limit: int) -> None:
# 以当前分钟为窗口键，比如 user:user_001:202603151530
now = time.strftime('%Y%m%d%H%M')
key = f'rpm:{user_id}:{now}'
count = redis_client.incr(key)
if count == 1:
redis_client.expire(key, 60) # 60秒后自动失效
if count > rpm_limit:
raise HTTPException(status_code=429, detail='每分钟请求数超限')
# 在具体路由中调用
def proxy_route(user=Depends(verify_api_key)):
check_rpm(user['id'], user['rpm_limit'])
# 转发请求到 vLLM如果需要同时限制Token速率，可以在转发请求前预估prompt token，并在拿到vLLM返回的usage后累加以消耗量。不要把TPM限流做成同步等待，否则会拖垮网关。
配额不足时的错误处理
当RPM或TPM超限，应返回HTTP 429，并在响应头中加上Retry-After，告诉客户端多少秒后可以重试。FastAPI的HTTPException默认不会生成Retry-After，你需要在中间件或路由中手动添加。下面是一个能自动附加该头的中间件示例：
from fastapi import Request
from fastapi.responses import JSONResponse
@app.middleware('http')
async def retry_after_middleware(request: Request, call_next):
response = await call_next(request)
if response.status_code == 429:
# 从当前分钟窗口剩余时间估算，或直接给60
response.headers['Retry-After'] = '60'
return response注意，上面的中间件写了一个固定值，实际上应根据窗口重置时间动态计算。可以把窗口键的剩余秒数放在请求上下文中，以便中间件读取。
验证方式和风险边界
写完网关后，用以下两条curl验证认证和限流是否生效。假设网关监听8000端口：
# 不带Key，应返回401
curl -i -X POST http://localhost:8000/v1/chat/completions -H 'Content-Type: application/json' -d '{"model":"qwen2.5-7b-instruct","messages":[{"role":"user","content":"hi"}]}'
# 带有效Key，首次请求应返回200（假设后端vLLM正常）
curl -i -X POST http://localhost:8000/v1/chat/completions -H 'Authorization: Bearer test-key' -H 'Content-Type: application/json' -d '{"model":"qwen2.5-7b-instruct","messages":[{"role":"user","content":"hi"}]}'要验证限流，可将rpm_limit设为1，连续发送两次请求，第二次应返回429并带Retry-After。风险方面：认证和限流依赖外部Redis，如果Redis宕机，网关会失去限流能力。建议给Redis加主从或改用内存限流作为降级方案，但多节点场景下内存限流不能共享计数。另外，限流窗口选择会直接影响高并发下的突发行为，固定窗口实现简单，但可能让最后1秒打满整个窗口配额；按秒的滑动窗口更好，但对Redis的读写有更高要求，需要结合环境确认。