多模型路由：LLM版的智能调度中心

可视化演示

通过路由调度流程、负载均衡权重分配和故障转移机制三个维度，动态展示多LLM供应商的智能调度策略，直观理解请求分发与成本优化逻辑。

路由策略设计：订单该派给谁？

想象你经营着一家外卖平台，手底下签约了多家风格迥异的餐厅：A餐厅出餐快但价格贵，B餐厅便宜但偶尔会送错，C餐厅口味稳定但只擅长川菜。每天不同的顾客点不同的菜，你作为调度员，总得定一条规矩来决定“这个订单派给谁”——这就是路由策略。

在多模型路由的世界里，“餐厅”就是各个LLM服务商，“菜”就是用户的请求类型。规则路由是最简单的策略，像一本写好的菜谱：代码生成类的请求走Claude，通用对话走GPT，中文客服走国内的模型。你按照请求的属性或业务标签，手动设定一条条分流的规则。这种策略直接明了，适合业务场景稳定的情况。

但现实世界远比菜谱复杂，用户的请求不会贴着标签进来。于是有了语义路由——它像一位经验老到的点菜员，听了顾客的描述后，能判断出“这其实是道川菜”还是“这应该交给西餐师傅”。语义路由会用嵌入模型（Embedding）把请求内容转成向量，然后与各模型擅长的任务类型做相似度匹配，把请求送往最合适的模型。最后是成本优先策略，它像一个精打细算的采购员，只在衡量成本与质量后做出抉择。

路由策略的朴素逻辑可以概括为一个条件判断的数学模型：

像餐厅会更新菜单一样，路由策略也需要定期审视。当发现某类请求被反复路由到贵的模型，而便宜的模型质量也够用，就该调整规则了。规则永远是死的，业务场景才是活的。

负载均衡与权重：怎么派单才不会“累死”一家店？

外卖平台最怕什么？怕所有订单都涌向人气最高的那家店，其他店冷冷清清。高峰期一到，爆单的爆单、闲逛的闲逛。调度中心必须有一套算法来平衡各家的负荷——这就是负载均衡。

最简单的思路是轮询调度（Round Robin），像打牌发牌一样，一张一张轮流派，A、B、C、A、B、C……每个模型获得机会均等。但现实中各供应商的性能、价格、额度都不同，纯粹的轮流并不公平。于是有了加权轮询（Weighted Round Robin），给每家餐厅设定一个系数——贵的模型小权重以防烧钱，便宜且稳定的模型大权重来扛流量。就像奶茶店按杯量给店员排班一样，能力强的人多值班。更进一步，最少连接算法像一个监工，实时盯着各家餐厅当前的排队人数，谁手上的单最少就把新单派给谁，真正做到“哪闲去哪”。

加权轮询的流量分配可以这样写：

其中，$w_i$ 是第 $i$ 个模型供应商的权重，$r_i$ 是分配给它的流量比例。

负载均衡是路由的“公平秤”，保证每一家模型供应商都不会被突然的流量峰值压垮。但请注意，权重不是拍脑袋定的——它应该基于各供应商的历史表现动态调整，而这正是后面调优环节要做的事。

故障转移机制：餐厅突然关门怎么办？

就餐高峰突然后厨起火，一家接单中的餐厅闭店了。如果一个平台的调度中心对此毫无反应，用户的订单就会悬挂在那边，直到用户失望而去——AI系统也一样。模型供应商也是第三方服务，可能超时、限流、报错甚至宕机。多模型路由必须配备一套故障转移机制，像训练有素的消防演练一样，在一分钟内把损失降到最低。

故障转移的基础是超时重试：调度中心给每次请求设定一个等待上限，比如等待10秒。如果这个模型一直吞吞吐吐不回复，就把这个请求转投给其他模型，并重新计时。注意，超时后直接重试同一个供应商意义不大，所以通常是“换一家再试”。更进一步的保护是熔断——如果某家模型在短时间内连续失败率达到阈值，比如50次请求中40次失败，调度中心就判定这家“状态不佳”，直接切断通向它的流量，让它冷却一段时间。这就像电路里的保险丝，电流异常时先熔断保护整个系统。所谓自动剔除，就是把冷却期的故障节点从候选列表里移除，待恢复健康后再“复职”。

失败重试的概率模型可以用指数退避描述，重试次数与等待时间的关系是：

其中 $t_0$ 是首次等待时间，$t_k$ 是第 $k$ 次重试前的等待时间。

有了这套机制，无论哪家LLM服务商掉链子，你的业务都能像不倒翁一样稳如泰山。故障转移是路由器的“保险丝”，宁可慢一秒，不可崩一片。

效果统计体系：怎么知道调度得好不好？

外卖平台靠什么评估调度员是不是靠谱？当然不是靠感觉，而是看数据：单均配送时长、差评率、超时率。同样，一个多模型路由器是否“聪明”，必须建立在效果统计之上。

首先需要统计成功率——这一百个请求里，有多少个按照预期拿到了完整回复？如果某家供应商的成功率持续走低，那它显然不再是“合格餐厅”。其次是延迟监控，精确到每个请求的响应时间，按模型分组求平均和分位数。一句话总结：成功率是“能不能成”，延迟是“够不够快”。最后是成本核算——这是Token与成本管理篇章最敏锐的探测器：每个模型的单价、每次调用的Token数、一天的累积支出。多模型路由的一大价值就是“好钢用在刀刃上”。

这三个指标组合起来，可以综合打分得到“性价比”指标。用加权评分表示：

其中 Success 是成功率，Latency 是平均延迟，$C$ 是该模型的单位成本，$T_{\text{target}}$ 是目标延迟，$C_{\text{lower}}$ 是成本下界，$\lambda_i$ 是权重系数。

没有统计就没有优化。效果统计是多模型路由的“仪表盘”，让你对每分钱的去向和每秒的延迟都了然于心。

持续调优策略：调度中心如何越用越聪明？

有了统计数据，大厨就能改进菜谱。持续调优是多模型路由的灵魂所在，它不是一次性的静态配置，而是一个不断“反馈-修正”的闭环。

首先是动态调权：根据统计结果，自动微调各供应商的权重。比如统计发现 P3 的成本低且响应快，但当前只分到了 20% 流量，就可以缓缓把它升到 30%，同时压一压贵又慢的 P1 的权重。这种调整不是过山车式急转，而像调温水阀一样一点一点地拧。其次是告警介入：当某家供应商的延迟或错误率越过了红线，系统通过短信、钉钉或邮件通知工程师——毕竟有些决定还得人来做。人工智能负责体力活，人类负责拍板。最后是灰度实验：像美团新店先小范围试营业一样，让新路由规则先只作用于 5% 的流量，评估一段时间再全量上线，把试错成本压到最低。

动态调权的更新公式是一个简单的迭代过程：

其中 $\alpha$ 是学习率（步长），$S_i$ 是当前周期内第 $i$ 个模型的效果得分，$\bar{S}$ 是所有模型的平均分。

多模型路由不是“一锤子买卖”，而是一场永不停歇的马拉松。持续调优让调度系统像一个老练的操盘手，越用越顺手，越用越省钱。

统一入口管理：一个调度台管所有供应商

试想一下，如果你管理的不是一个外卖App，而是同时用十个App各管十家餐厅，那天还没亮你就乱成一团了。多模型路由的中心化架构，天然就是统一入口管理的一把好手。

最直观的好处是API封装：你的业务代码只需要对接一个路由网关的API，由网关内部再调用各家模型的接口。这样如果哪天要换掉底层某家供应商，业务代码零改动。其次是密钥管理：各家模型的API Key被统一存放在路由网关的配置中心里，不会散落在多个服务的环境变量中到处泄露，像把所有贵重钥匙统一锁进一个保险柜里。最后是用量监控：所有流量都过一道闸口，路由层可以顺手记录每个业务方的调用量与花费，像财务对账一样清晰，做到“事前预算、事后追责”。

统一入口还天然承担了降级兜底功能。假设某一家顶级模型完全不可用，网关可以一键把所有它的流量导向备胎模型，而不需要感知到这一点的业务方有任何改动。这就像银行网点虽然关门了，但ATM和手机银行全都正常运转。

统一入口的调用逻辑可以概括为：

即任何一个外部请求先到路由器，由路由器决定并转发给某家供应商，用户只面对一个稳定的API端点。

统一入口是多模型路由的“门面担当”，它让复杂的多供应商调度，在用户眼里简单得像调用一个模型。而正是这种“复杂的简单”，才是工程艺术的最高境界。

AI 应用场景

Python 演示

import timeimport randomfrom collections import defaultdict  class ModelProvider:    """模拟一个LLM供应商"""     def __init__(self, name, weight, base_latency=0.1, fail_rate=0.05):        self.name = name        self.weight = weight        self.base_latency = base_latency        self.fail_rate = fail_rate        self.healthy = True        self.cool_until = 0.0        self.requests = 0        self.failures = 0     def invoke(self, prompt):        """模拟调用模型，按概率返回结果或抛异常"""        self.requests += 1        time.sleep(self.base_latency * random.uniform(0.8, 1.2))        if random.random() < self.fail_rate:            self.failures += 1            raise RuntimeError(f"{self.name} 调用失败")        return f"[{self.name}] 已处理: {prompt[:10]}..."  class MultiModelRouter:    """多模型路由器：加权轮询 + 熔断剔除"""     def __init__(self, providers, failure_threshold=3):        self.providers = providers        self.failure_threshold = failure_threshold        self.index = 0        self.circuit_break_count = defaultdict(int)     def pick(self):        """加权轮询挑选健康节点"""        healthy = [p for p in self.providers if p.healthy]        if not healthy:            raise RuntimeError("所有模型均不可用！")        total_weight = sum(p.weight for p in healthy)        r = random.uniform(0, total_weight)        upto = 0.0        for p in healthy:            upto += p.weight            if r <= upto:                return p        return healthy[-1]     def route(self, prompt):        """路由一个请求，处理故障剔除并尝试重试"""        for attempt in range(3):  # 最多尝试3次            provider = self.pick()            try:                result = provider.invoke(prompt)                self.circuit_break_count[provider.name] = 0                return result            except RuntimeError as e:                print(f"  尝试{attempt + 1}失败: {e}")                self.circuit_break_count[provider.name] += 1                if self.circuit_break_count[provider.name] >= self.failure_threshold:                    provider.healthy = False                    provider.cool_until = time.time() + 5  # 熔断5秒                    print(f"  >>> {provider.name} 已被熔断剔除!")         # 触发健康检查：手动恢复冷却结束的节点        self._recover()        raise RuntimeError("所有重试均失败")     def _recover(self):        """恢复冷却到期的节点"""        for p in self.providers:            if not p.healthy and time.time() >= p.cool_until:                p.healthy = True                self.circuit_break_count[p.name] = 0                print(f"  >>> {p.name} 冷却期结束，恢复服务")  if __name__ == "__main__":    providers = [        ModelProvider("GPT-4", weight=50, base_latency=0.2, fail_rate=0.1),        ModelProvider("Claude-3", weight=30, base_latency=0.15, fail_rate=0.05),        ModelProvider("国产模型", weight=20, base_latency=0.05, fail_rate=0.2),    ]    router = MultiModelRouter(providers, failure_threshold=2)     for i in range(10):        print(f"请求 {i+1}:")        try:            resp = router.route(f"第{i+1}个用户问题")            print(f"  成功 -> {resp}")        except RuntimeError as e:            print(f"  失败 -> {e}")        finally:            # 每隔几次请求检查一下恢复状态            if i % 3 == 2:                router._recover()

1. class ModelProvider:定义一个模型供应商类，包含权重、故障率和健康状态
2. self.fail_rate = fail_rate模拟真实供应商的随机故障，让演示动起来
3. def pick(self):实现加权轮询核心算法，健康节点按权重比例被选中
4. healthy = [p for p in self.providers if p.healthy]先过滤出健康节点，病号自动出局
5. r = random.uniform(0, total_weight)随机落点法，权重越大被击中的概率越高
6. self.circuit_break_count[provider.name] += 1连续失败计数，熔断器的计数器
7. provider.healthy = False一旦失败次数超阈值，立即把该节点标记为不健康
8. provider.cool_until = time.time() + 5设置熔断冷却期，5秒内不参与路由
9. self._recover()定期检查冷却到期的节点，让它们自动恢复服务
10. router = MultiModelRouter(providers, failure_threshold=2)组装三家“餐厅”并启动调度

常见误区

面试题