Skip to content
🔗 分享本题
查看我的学习进度 →

25 模块 Q3 教学图:设计一个 LLM API 网关(限流 + 路由 + 计费)

🧠 图解记忆:网关统一鉴权、配额、路由和计费,并用降级与观测守住模型依赖的不确定性;点击图片可查看原图。

💡 答案要点

题目理解:

LLM API 网关:
- 限流:防止用户打爆 API 配额
- 路由:多模型选择、成本优化
- 计费:按使用量收费,支持多租户
- 核心挑战:高并发、低延迟、可观测

整体架构:

外部请求

┌──────────────────────────────────────────────────────────┐
│                    API Gateway                           │
│   统一入口 │ TLS 终结 │ 请求日志                          │
└──────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│                Auth & Rate Limit Layer                   │
│   API Key 验证 │ 令牌桶限流 │ 额度扣减                    │
└──────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│                  Model Router                            │
│   意图分类 │ 成本优先/质量优先 │ 模型选择                   │
└──────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│              LLM Provider Adapters                       │
│   OpenAI │ Anthropic │ Azure │ 国内模型 │ 自部署          │
└──────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│                Usage & Billing                           │
│   用量记录 │ 计费规则 │ 成本分析                          │
└──────────────────────────────────────────────────────────┘

限流实现(令牌桶 + 多维度):

展开 Python 代码示例(58 行)
python
import time
import redis
import hashlib

class RateLimiter:
    """多维度限流:用户 + API Key + IP"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
    
    async def check_limit(
        self,
        api_key: str,
        requests_per_minute: int = 60,
        tokens_per_minute: int = 60000,
        expected_tokens: int = 1000,
    ) -> tuple[bool, str]:
        """检查限流,返回 (是否允许, 原因)"""
        
        # 1. 请求频率限流(令牌桶)
        req_key = f"rate:req:{api_key}"
        req_allow = await self.redis.evalsha(
            RATE_LIMIT_SCRIPT,
            1, req_key,
            requests_per_minute, 60,  # 每分钟 N 个请求
            time.time()
        )
        if not req_allow:
            return False, "rate_limit:requests"
        
        # 2. Token 限流
        tok_key = f"rate:tok:{api_key}"
        current_tokens = await self.redis.get(tok_key) or 0
        
        if current_tokens + expected_tokens > tokens_per_minute:
            return False, "rate_limit:tokens"
        
        await self.redis.incrby(tok_key, expected_tokens)
        await self.redis.expire(tok_key, 60)  # 1分钟窗口
        
        return True, "ok"


RATE_LIMIT_SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

local current = redis.call('GET', key) or 0
if current >= limit then
    return 0
end

redis.call('INCR', key)
redis.call('EXPIRE', key, window)
return 1
"""

智能路由(成本 + 质量平衡):

展开 Python 代码示例(61 行)
python
class ModelRouter:
    """模型路由:根据请求特征选择最优模型"""
    
    MODELS = {
        # (max_tokens, quality_score, cost_per_1k)
        "gpt-4o": (128000, 0.95, 0.015),
        "gpt-4o-mini": (128000, 0.85, 0.0015),
        "claude-3.5-sonnet": (200000, 0.93, 0.010),
        "qwen-plus": (131072, 0.80, 0.004),
    }
    
    def route(
        self,
        request: ChatRequest,
        strategy: str = "cost_quality_balance"
    ) -> str:
        """
        路由策略:
        - cost_first: 成本优先
        - quality_first: 质量优先
        - cost_quality_balance: 成本质量平衡
        """
        
        if strategy == "cost_first":
            return self._cheapest_model(request)
        
        if strategy == "quality_first":
            return self._best_quality_model(request)
        
        # 默认:成本质量平衡
        return self._balanced_model(request)
    
    def _balanced_model(self, request: ChatRequest) -> str:
        """综合评分:质量分数 / 成本"""
        # 简单问题 → 便宜模型
        if self._is_simple(request):
            return "gpt-4o-mini"
        
        # 代码相关 → Claude(代码能力强)
        if self._is_code_related(request):
            return "claude-3.5-sonnet"
        
        # 长上下文 → 支持大的模型
        if request.history_length > 20:
            return "claude-3.5-sonnet"
        
        # 中文 + 简单 → 国内模型(成本低)
        if request.language == "zh" and self._is_simple(request):
            return "qwen-plus"
        
        return "gpt-4o"
    
    def _is_simple(self, request: ChatRequest) -> bool:
        """判断是否为简单请求"""
        simple_patterns = ["是什么", "介绍一下", "定义", "时间", "地点"]
        return any(p in request.prompt for p in simple_patterns)
    
    def _is_code_related(self, request: ChatRequest) -> bool:
        """判断是否代码相关"""
        code_patterns = ["代码", "函数", "Python", "Java", "bug", "debug"]
        return any(p in request.prompt.lower() for p in code_patterns)

计费系统(用量记录 + 扣费):

展开 Python 代码示例(52 行)
python
class BillingService:
    """按量计费服务"""
    
    def __init__(self, db: Database, llm_router: ModelRouter):
        self.db = db
        self.router = llm_router
    
    async def record_usage(self, api_key: str, request: ChatRequest, response: ChatResponse):
        """记录用量并扣费"""
        
        # 计算费用
        model = request.model or self.router.route(request)
        _, quality, cost_per_1k = self.router.MODELS[model]
        
        input_tokens = response.usage.input_tokens
        output_tokens = response.usage.output_tokens
        total_tokens = input_tokens + output_tokens
        
        cost = (total_tokens / 1000) * cost_per_1k
        
        # 写入用量记录
        await self.db.execute("""
            INSERT INTO usage_logs (api_key, model, input_tokens, output_tokens, cost, created_at)
            VALUES ($1, $2, $3, $4, $5, NOW())
        """, api_key, model, input_tokens, output_tokens, cost)
        
        # 扣减账户余额
        await self.db.execute("""
            UPDATE accounts
            SET balance = balance - $1
            WHERE api_key = $2 AND balance >= $1
        """, cost, api_key)
    
    async def get_cost_breakdown(self, api_key: str, start_date: datetime, end_date: datetime) -> dict:
        """获取成本分析"""
        rows = await self.db.fetch("""
            SELECT 
                model,
                SUM(input_tokens) as total_input,
                SUM(output_tokens) as total_output,
                SUM(cost) as total_cost,
                COUNT(*) as request_count
            FROM usage_logs
            WHERE api_key = $1 AND created_at BETWEEN $2 AND $3
            GROUP BY model
        """, api_key, start_date, end_date)
        
        return {
            "total_cost": sum(r["total_cost"] for r in rows),
            "by_model": [dict(r) for r in rows],
            "avg_cost_per_request": sum(r["total_cost"] for r in rows) / max(sum(r["request_count"] for r in rows), 1)
        }

多租户计费方案:

方案说明适用场景
预付费套餐买 token 包,用完为止个人用户
后付费月结按月结算,月底扣费企业客户
信用额度设置信用额度,超额熔断大客户
用量分层阶梯定价,用量越大越便宜大客户

面试话术:

"LLM API 网关的三大核心能力:限流、路由、计费。限流用令牌桶实现,Redis+Lua 脚本原子操作,支持请求数和 Token 数双维度限流。路由的核心是'让对的模型处理对的任务'——简单问答用 mini 模型(省 10 倍成本),代码生成用 Claude(代码能力强),长上下文用 Claude(200K)。计费用预写日志(WAL)+ 异步扣费,保证数据一致性。面试能说清楚这三件套的实现细节,说明你有 API 服务端开发经验。"

📚 参考:LiteLLM Proxy(网关:路由/限流/计费参考实现)