LLM API网关设计：限流、缓存、故障转移的工程实践
为什么需要LLM API网关 当企业级应用调用LLM API时，直接暴露上游接口会面临三个核心问题：成本失控、延迟波动和服务不可用。一个设计良好的API网关能在这三者之间取得平衡，同时提供统一的鉴权、计费和可观测能力。 本文将从工程实践角度，拆解LLM API网关的三大核心模块。 限流策略设计 多维度限流模型 与传统API不同，LLM调用具有token维度的特性。一个简单的QPS限流无法覆盖真实场景，需要多维度组合： 维度 限流键 典型阈值 适用场景 QPS API Key + 路径 50 req/s 防恶意刷量 TPM API Key + 模型 100K tokens/min 控制成本 并发数 用户ID 5 concurrent 防资源独占 日总量 租户ID 10M tokens/day 预算控制 令牌桶+滑动窗口的混合实现 以下是基于Redis的混合限流实现： import asyncio import time import redis.asyncio as redis class TokenBucketRateLimiter: def __init__(self, redis_client: redis.Redis, capacity: int, refill_rate: float): self.redis = redis_client self.capacity = capacity self.refill_rate = refill_rate # tokens per second async def acquire(self, key: str, tokens: int = 1) -> bool: lua_script = """ local key = KEYS[1] local capacity = tonumber(ARGV[1]) local refill_rate = tonumber(ARGV[2]) local requested = tonumber(ARGV[3]) local now = tonumber(ARGV[4]) local bucket = redis.call('HMGET', key, 'tokens', 'timestamp') local current_tokens = tonumber(bucket[1]) or capacity local last_timestamp = tonumber(bucket[2]) or now -- 补充令牌 local elapsed = now - last_timestamp current_tokens = math.min(capacity, current_tokens + elapsed * refill_rate) if current_tokens < requested then return 0 end current_tokens = current_tokens - requested redis.call('HMSET', key, 'tokens', current_tokens, 'timestamp', now) redis.call('EXPIRE', key, 3600) return 1 """ now = time.time() result = await self.redis.eval( lua_script, 1, f"ratelimit:{key}", self.capacity, self.refill_rate, tokens, now ) return bool(result) 关键设计点：使用Lua脚本保证原子性，避免竞态条件。对于TPM限流，需要在请求完成后根据实际token消耗做补充扣减。 ...