模型路由与负载均衡：多模型实例间如何智能分发请求

一、一个请求发到繁忙的 GPT-4 实例，另一个发给空闲的实例

部署了多个模型实例（如多个 GPT-4 API endpoint、多个本地部署的 Llama 实例），但请求分发是随机的——一个实例在超载，另一个在摸鱼。

传统负载均衡（Round Robin / Least Connections）对 LLM 推理不完全适用，因为：

1. LLM 推理的资源消耗与请求长度相关——不是固定的
2. 不同模型的推理速度差异巨大（GPT-4 vs GPT-4o-mini）
3. 请求有优先级差异——VIP 用户不应排队

智能路由需要在多个维度上做决策：实例负载、模型能力、请求优先级、成本预算。

flowchart TD A[LLM 请求] --> B[智能路由器] B --> C{路由策略} C --> D[能力路由] C --> E[负载感知路由] C --> F[成本感知路由] C --> G[优先级路由] D --> D1[请求复杂度评估] D1 --> D2[匹配合适模型] D2 --> D3[简单→GPT-4o-mini<br/>复杂→GPT-4] E --> E1[实例并发数检测] E1 --> E2[选择最低负载] F --> F1[请求 Token 预算] F1 --> F2[选择成本最优] G --> G1[用户优先级] G1 --> G2[VIP 用户优先<br/>免费用户排队] D3 & E2 & F2 & G2 --> H[加权决策] H --> I[路由到目标实例]

二、路由决策的四个维度

维度一：能力路由

不是所有请求都需要 GPT-4。简单的事实问答（"1+1=?"）用最便宜的模型就行。路由器需要快速评估请求复杂度：

• 请求 token 数量（长 prompt 通常更复杂）
• 是否包含代码/数学（需要强推理）
• 历史同类型请求的平均处理时间

维度二：负载感知路由

不依赖连接数（传统 LB 的指标），而基于正在进行的推理数——因为 LLM 推理的并发窗口有限，一个实例能同时处理的最大请求数是 GPU 显存决定的。

维度三：成本感知路由

在同等能力下选择最便宜的实例。例如两个不同的 API provider 都提供 GPT-4，价格可能不同。

维度四：优先级路由

业务关键请求（支付、客服等）优先于非关键请求（推荐、内容生成）。

三、智能路由器的实现

import asyncio import time from dataclasses import dataclass, field from enum import Enum from typing import Optional import random class ModelTier(Enum): """模型能力分级。""" BASIC = "basic" # gpt-4o-mini 级别 ADVANCED = "advanced" # gpt-4o 级别 PREMIUM = "premium" # gpt-4-turbo / claude-3-opus 级别 class RequestPriority(Enum): HIGH = 0 # VIP 用户 / 支付流程 NORMAL = 1 LOW = 2 # 后台任务 / 免费用户 @dataclass class ModelInstance: """模型实例——可以是云端 API endpoint 或本地推理服务。""" instance_id: str model_name: str tier: ModelTier endpoint: str max_concurrency: int # 最大并发推理数 current_load: int = 0 # 当前推理数 cost_per_1k_tokens: float = 0.0 is_healthy: bool = True last_error: Optional[str] = None class SmartRouter: """智能模型路由器。 路由算法： - 先能力匹配（tier 必须 >= 请求要求的 tier） - 再健康过滤（排除不健康的实例） - 多维度加权打分（负载 40% + 成本 30% + 延迟 30%） - 选择得分最低的实例 """ def __init__(self): self._instances: dict[str, ModelInstance] = {} self._latency_history: dict[str, list[float]] = {} self._lock = asyncio.Lock() def register_instance(self, instance: ModelInstance) -> None: self._instances[instance.instance_id] = instance async def route( self, estimated_complexity: float, # 0-1，0=简单, 1=复杂 request_tokens: int, priority: RequestPriority = RequestPriority.NORMAL, max_cost_per_1k: Optional[float] = None, ) -> ModelInstance: """选择最优模型实例。 Args: estimated_complexity: 预估复杂度 request_tokens: 请求 token 数 priority: 请求优先级 max_cost_per_1k: 成本上限（None=无限制） """ required_tier = self._complexity_to_tier(estimated_complexity) async with self._lock: candidates = [] for inst in self._instances.values(): # 过滤条件1: 能力足够 if inst.tier.value < required_tier.value: continue # 过滤条件2: 健康检查 if not inst.is_healthy: continue # 过滤条件3: 有容量 if inst.current_load >= inst.max_concurrency: continue # 过滤条件4: 成本上限 if max_cost_per_1k and inst.cost_per_1k_tokens > max_cost_per_1k: continue candidates.append(inst) if not candidates: raise NoAvailableInstanceError("No instance available") # 打分并选择最优 scores = [] for inst in candidates: score = self._score_instance(inst, priority) scores.append((score, inst)) scores.sort(key=lambda x: x[0]) selected = scores[0][1] # 更新负载 selected.current_load += 1 return selected async def release(self, instance_id: str, latency_ms: float) -> None: """请求完成——释放实例并发槽位。""" async with self._lock: if instance_id in self._instances: inst = self._instances[instance_id] inst.current_load = max(0, inst.current_load - 1) # 记录延迟历史 if instance_id not in self._latency_history: self._latency_history[instance_id] = [] self._latency_history[instance_id].append(latency_ms) # 只保留最近 100 次 if len(self._latency_history[instance_id]) > 100: self._latency_history[instance_id] = self._latency_history[instance_id][-100:] def _score_instance( self, instance: ModelInstance, priority: RequestPriority ) -> float: """多维度加权打分——分数越低越好。""" # 1. 负载得分 (40%) load_ratio = instance.current_load / max(instance.max_concurrency, 1) load_score = load_ratio * 0.4 # 2. 成本得分 (30%) # 将成本归一化到 0-1，假设最贵的 model = $0.03/1k cost_score = (instance.cost_per_1k_tokens / 0.03) * 0.3 # 3. 延迟得分 (30%) # 使用滑动窗口平均延迟 latencies = self._latency_history.get(instance.instance_id, []) if latencies: avg_latency = sum(latencies) / len(latencies) # 归一化：假设 5000ms 为"很慢" latency_score = min(avg_latency / 5000, 1.0) * 0.3 else: latency_score = 0.15 # 无数据时给中间值 total = load_score + cost_score + latency_score # 4. 优先级调整 # high priority: 降低分数（优先选择） if priority == RequestPriority.HIGH: total *= 0.8 elif priority == RequestPriority.LOW: total *= 1.2 return total def _complexity_to_tier(self, complexity: float) -> ModelTier: """复杂度到模型能力的映射。""" if complexity >= 0.7: return ModelTier.PREMIUM elif complexity >= 0.3: return ModelTier.ADVANCED return ModelTier.BASIC async def health_check(self, instance_id: str) -> bool: """健康检查——探测实例是否可达。""" inst = self._instances.get(instance_id) if not inst: return False # 实际实现应发一个轻量探针请求 return inst.is_healthy class NoAvailableInstanceError(Exception): """无可用的模型实例。""" pass

四、路由策略的权衡

负载均衡的精度

current_load是乐观计数器——请求开始时加 1，结束时减 1。如果请求失败而没有调用release，负载计数就会漂移。需要定期健康检查 + 漏桶修复。

冷启动问题

新实例加入时延迟历史为空，路由给它可能不公平（要么全给它要么全不给）。解决：新实例在最初 10 个请求中给较高权重（exploration），快速建立历史。

成本 vs 延迟的交换

有些场景需要最低延迟（实时对话），有些需要最低成本（离线批处理）。max_cost_per_1k参数让调用方声明自己的偏好。

五、总结

智能路由不是替代传统负载均衡，而是在其上叠加能力、负载、成本、优先级四个维度的感知。简单的请求走便宜模型，复杂的推理走强模型，高优先级请求插队——这一切应该对 Agent 透明。路由器往外暴露的是"给我一个最优的模型实例"，内部的打分排序是它的事。