写在前面：我在去年 11 月接手了一个日均 2000 万次 LLM 调用的客服系统，那次 Claude Sonnet 4.5 区域故障给我们带来了 47 分钟的 P0 事故——没有自动切换，所有请求全挂在 502 上。后来我把团队基于 HolySheep AI 跑通的延迟+成本双权重路由架构完整拆出来，事故复盘里直接降到 800ms 切走。这篇文章把生产代码、benchmark、月度账单、社区评价一次性给你讲透。
一、为什么 LLM 路由比传统 LB 更难
传统负载均衡看的是 QPS 和连接数，但 LLM 路由有三个新维度：
- 延迟高度长尾：单次推理 80ms — 8s 都有可能，P99 是 P50 的 50 倍
- 价格量级差极大：GPT-4.1 输出 $8/MTok，DeepSeek V3.2 仅 $0.42/MTok，相差 19 倍
- 故障模式隐蔽：限流返回 429 但响应 200ms，OOM 直接 stream 中断无错误码
单看延迟会把所有请求打到 Sonnet 4.5（贵 30 倍），单看成本会全走 DeepSeek（复杂推理质量崩）。我们需要一个加权评分机制。
二、核心架构：四层调度模型
"""
架构分层：
L1. Endpoint Registry — 维护所有可用模型节点
L2. Health Checker — 后台异步探活 + 熔断
L3. Latency Tracker — EWMA 滑动窗口记录延迟
L4. Weighted Router — 综合延迟/成本/健康度打分
"""
import asyncio, time, statistics, logging
from dataclasses import dataclass, field
from collections import deque
from enum import Enum
from typing import List, Tuple
import httpx
logging.basicConfig(level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger('llm-router')
class Health(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
@dataclass
class LLMEndpoint:
name: str
base_url: str
api_key: str
model: str
output_price_per_mtok: float # 美元/百万 token
latency_samples: deque = field(default_factory=lambda: deque(maxlen=200))
health: Health = Health.HEALTHY
consecutive_failures: int = 0
breaker_until: float = 0.0
# 权重三因子（可在运行时调整）
w_latency: float = 0.55
w_cost: float = 0.35
w_health: float = 0.10
def is_available(self) -> bool:
return (self.health != Health.UNHEALTHY
and time.time() > self.breaker_until)
def avg_latency(self) -> float:
return statistics.mean(self.latency_samples) if self.latency_samples else 9999.0
三、延迟与成本权重算法
我们对每个 endpoint 算一个 0~1 之间的归一化评分，分数越低越优先：
def score(self,
max_latency_ms: float = 500.0,
max_price: float = 15.0) -> float:
# 延迟分：EWMA 平均延迟 / 基准上限
lat_n = min(self.avg_latency() / max_latency_ms, 1.0)
# 价格分：当前价格 / 上限（最贵模型作基准）
cost_n = min(self.output_price_per_mtok / max_price, 1.0)
# 健康分：UNHEALTHY 直接拉满
h_n = {Health.HEALTHY: 0.0,
Health.DEGRADED: 0.5,
Health.UNHEALTHY: 1.0}[self.health]
return (self.w_latency * lat_n
+ self.w_cost