1. 从单模型硬扛到多模型调度：我为什么折腾这套网关

去年下半年，我手上有个内容生成类的项目，一开始图省事，所有请求都往同一个大模型上怼。刚开始量小，感觉还行，响应质量也说得过去。但跑了两三个月，问题全冒出来了：高峰期延迟飙到十几秒，账单一个月比一个月吓人，而且有些任务明明用轻量模型就能干得不错，我却一直在用最贵的那个在跑。说白了，就是“一个模型打天下”的粗暴打法，成本、延迟、质量三本账全是糊涂账。

后来我花了两周时间，基于蓝耘元生代平台搭了一套多模型调度网关，核心目标就一个：让每一次请求都能被路由到“当下最合适”的模型上，同时把成本、延迟、质量这三个维度全部透明化。这套东西跑到现在差不多半年，整体推理成本降了六成多，P95延迟从十几秒压到了三秒以内，而且质量没有肉眼可见的下降。

这篇文章我会把整套网关的设计思路、核心调度策略、成本核算方法、延迟优化手段，以及我在实操中踩过的坑，全部摊开来讲。适合正在被大模型调用成本困扰的开发者、需要做多模型选型的技术负责人，以及任何想搞清楚“多模型调度到底怎么落地”的人。不管你是刚接触这块，还是已经在做类似的事情，应该都能从里面找到能直接抄作业的东西。

2. 蓝耘元生代平台上的模型接入与统一抽象层设计

2.1 为什么先要做统一抽象层

多模型调度最怕的一件事，就是每接一个新模型就要改一遍业务代码。我一开始就是犯了这个错：第一个模型接进去的时候，业务层直接调了它的SDK，参数格式、返回结构全写死在业务逻辑里。结果接第二个模型的时候，发现两边的请求体结构完全不一样，一个用messages数组，一个用prompt字符串，返回的字段名也对不上。改起来那叫一个痛苦。

所以第二版我做的第一件事，就是抽一层统一模型接口。所有模型，不管底层是哪家、什么版本，对外暴露的请求和响应格式全部统一。业务层只认这一套格式，完全不关心背后是谁在干活。

具体来说，我定义了一个标准的请求结构：

class UnifiedRequest: def __init__(self, messages, max_tokens=1024, temperature=0.7, stream=False, task_type=None, priority="normal"): self.messages = messages # 统一用 messages 数组 self.max_tokens = max_tokens self.temperature = temperature self.stream = stream self.task_type = task_type # 任务类型，调度时用 self.priority = priority # 优先级，影响路由策略

响应结构也统一成一套：

class UnifiedResponse: def __init__(self, content, model_used, latency_ms, prompt_tokens, completion_tokens, cost): self.content = content self.model_used = model_used self.latency_ms = latency_ms self.prompt_tokens = prompt_tokens self.completion_tokens = completion_tokens self.cost = cost

这样一来，业务层拿到的永远是同一套东西，模型换了、加了、删了，业务代码一行不用动。这个抽象层的价值，在后期我频繁调整模型组合的时候体现得淋漓尽致。

2.2 蓝耘元生代上的模型注册与适配器模式

在蓝耘元生代平台上，模型接入本身是比较顺的，平台提供了标准的调用入口。但不同模型之间还是有差异，比如有的支持流式输出，有的对temperature的取值范围要求不同，有的对max_tokens有硬上限。我的做法是给每个模型写一个适配器，适配器负责把统一请求翻译成该模型能懂的格式，再把返回结果翻译回统一响应。

适配器的核心逻辑大概长这样：

class ModelAdapter: def __init__(self, model_name, endpoint, config): self.model_name = model_name self.endpoint = endpoint self.config = config # 包含价格、限流、能力标签等 def transform_request(self, unified_req): # 子类实现具体的转换逻辑 raise NotImplementedError def transform_response(self, raw_resp, latency_ms): # 子类实现具体的转换逻辑 raise NotImplementedError def estimate_cost(self, prompt_tokens, completion_tokens): # 根据配置里的单价计算成本 input_price = self.config["input_price_per_1k"] output_price = self.config["output_price_per_1k"] return (prompt_tokens / 1000) * input_price + \ (completion_tokens / 1000) * output_price

每个模型注册的时候，我会在配置里写清楚它的能力标签，比如["长文本", "代码", "推理", "快速"]，还有它的价格、限流阈值、平均延迟。这些元数据是后面调度决策的核心依据。

提示：适配器模式看起来多写了一些代码，但当你手上有五六个模型要管理的时候，这点前期投入会省掉后面无数次的重复劳动。我一开始偷懒没做，后来补的时候花了整整两天重构。

2.3 模型能力标签体系的建立

光有适配器还不够，调度的时候得知道每个模型擅长什么。我建了一套能力标签体系，给每个模型打上多维度的标签：

这套标签不是拍脑袋定的，是我在实际跑了一段时间之后，根据每个模型在真实任务上的表现慢慢校准出来的。比如有个模型标称支持32K上下文，但实测超过16K之后质量下降明显，我就把它的有效上下文标签改成了16K。这种“实测校准”的步骤，是标签体系能不能真正发挥作用的关键。

3. 调度策略的核心：成本、延迟、质量怎么同时算账

3.1 三种调度模式的适用场景

调度策略是整个网关的大脑。我一开始想搞一个“万能调度算法”，后来发现根本不现实，不同场景对成本、延迟、质量的敏感度完全不一样。最后我把它拆成了三种模式，业务方可以根据场景自己选：

成本优先模式：在满足最低质量要求的前提下，优先选最便宜的模型。适合批量处理、离线任务、对实时性要求不高的场景。比如我有个每天跑一次的摘要任务，用这个模式，成本直接降到原来的三分之一。

延迟优先模式：优先选响应最快的模型，质量只要达标就行。适合在线对话、实时交互场景。用户等三秒和等十秒，体验差距是巨大的。

质量优先模式：优先选质量最高的模型，成本和延迟都可以让步。适合关键内容生成、复杂推理任务。

这三种模式不是互斥的，我在网关里做了一个加权评分机制，每个模型根据当前模式得到一个综合分，选分最高的。评分公式大概是这样：

综合分 = w_cost * 成本分 + w_latency * 延迟分 + w_quality * 质量分

三个权重根据模式动态调整。成本优先模式下，w_cost调到0.6，其他两个各0.2；延迟优先模式下，w_latency调到0.6；质量优先模式下，w_quality调到0.6。这个权重不是固定的，我留了配置接口，业务方可以微调。

3.2 基于任务类型的路由规则

光靠加权评分还不够精细。我后来加了一层任务类型路由，在评分之前先做一次粗筛。具体逻辑是：每个请求进来的时候，业务层会带一个task_type标签，比如code_generation、summarization、translation、chat。网关根据这个标签，先筛出所有支持该任务类型的模型，再在候选集里做加权评分。

这样做的好处是避免“用翻译模型去写代码”这种明显不合理的路由。我一开始没做这层，结果有一次一个代码生成请求被路由到了一个擅长对话但代码能力很弱的模型上，输出质量惨不忍睹。加了任务类型路由之后，这类问题基本消失了。

任务类型和模型的映射关系，我维护了一张配置表：

TASK_MODEL_MAP = { "code_generation": ["model_a", "model_c", "model_e"], "summarization": ["model_b", "model_d", "model_a"], "translation": ["model_d", "model_b"], "chat": ["model_a", "model_b", "model_c", "model_d", "model_e"], "reasoning": ["model_c", "model_e"] }

这张表也是动态维护的，新模型接进来、老模型下线，改配置就行，不用动代码。

3.3 动态权重调整与实时反馈闭环

静态权重有个问题：模型的实际表现是会变的。今天这个模型快，明天可能因为负载高就慢了；这个月这个模型便宜，下个月可能调价了。所以我加了一个实时反馈闭环，网关会持续收集每个模型的实际延迟、成功率、成本数据，然后定期（我设的是每5分钟）更新一次权重。

具体做法是维护一个滑动窗口，记录每个模型最近N次调用的表现：

class ModelStats: def __init__(self, window_size=100): self.latencies = deque(maxlen=window_size) self.success_count = 0 self.fail_count = 0 self.total_cost = 0.0 def record(self, latency_ms, success, cost): self.latencies.append(latency_ms) if success: self.success_count += 1 else: self.fail_count += 1 self.total_cost += cost def get_avg_latency(self): if not self.latencies: return float('inf') return sum(self.latencies) / len(self.latencies) def get_success_rate(self): total = self.success_count + self.fail_count if total == 0: return 1.0 return self.success_count / total

调度的时候，如果某个模型的成功率跌到阈值以下（我设的是95%），就自动把它从候选集里摘掉，等它恢复再放回来。这个机制帮我避免了好几次因为某个模型临时抽风导致的大面积故障。

注意：滑动窗口的大小需要根据实际流量调整。流量大的时候窗口可以小一点，反应更灵敏；流量小的时候窗口要大一点，避免因为样本太少导致误判。我一开始设的50，后来发现低峰期样本不够，改成了100。

4. 成本透明化：每一分钱花在哪里都要看得见

4.1 Token级成本核算的实现

成本透明化的第一步，是能精确算出每次请求花了多少钱。这个听起来简单，做起来有不少细节。首先你得拿到每次请求的prompt_tokens和completion_tokens，然后根据模型的单价算。但问题是，不同模型的计费方式不一样，有的按输入输出分开算，有的统一算，有的还有阶梯价格。

我的做法是在适配器里统一处理，每个模型适配器负责把自己的计费逻辑翻译成统一的成本数值。核心代码前面已经展示过了，就是estimate_cost那个方法。这里补充一个细节：流式输出的成本核算。流式输出的时候，completion_tokens是逐步产生的，我是在流结束的时候统一结算，而不是每来一个chunk就算一次，避免频繁计算带来的开销。

另外，我还在网关层做了一个成本聚合，按天、按模型、按业务方三个维度汇总。这样月底一看报表，就知道钱花在哪了。

4.2 成本归因：按业务方、按任务类型拆分

光知道总成本还不够，得知道是谁花的、花在什么任务上。我在请求头里加了两个字段：business_unit和task_type。网关在记录成本的时候，会把这两个维度一起存下来。

这样我就能生成这样的报表：

有了这张表，优化方向就一目了然了。比如业务B的代码生成平均单次成本是翻译的13倍，那我就会重点去看代码生成这块能不能用更便宜的模型替代，或者能不能通过优化prompt减少token消耗。

4.3 预算告警与自动降级机制

成本透明化做到位之后，下一步就是预算控制。我给每个业务方设了日预算和月预算，网关会实时监控消耗情况。当消耗达到预算的80%时，发告警；达到100%时，自动触发降级策略。

降级策略分两档：第一档是切换到成本优先模式，把请求往便宜模型上导；第二档是如果成本优先模式还压不住，就对非关键任务做限流。这个机制帮我避免了好几次“月底账单爆炸”的情况。

def check_budget(business_unit, estimated_cost): daily_used = get_daily_cost(business_unit) daily_budget = get_daily_budget(business_unit) if daily_used + estimated_cost > daily_budget: return "exceeded" elif daily_used + estimated_cost > daily_budget * 0.8: return "warning" return "normal"

提示：预算告警的阈值不要设得太死，我一开始设的90%告警，结果经常在半夜触发，没人处理。后来改成80%告警，留出足够的缓冲时间，效果就好多了。

5. 延迟优化：从十几秒到三秒以内的实战路径

5.1 延迟瓶颈的定位方法

延迟优化最怕盲目调优。我一开始就是瞎调，觉得哪里都慢，结果改了一堆东西，效果不明显。后来我静下心来做了分段计时，把一次请求的耗时拆成几段：网关处理时间、模型排队时间、模型推理时间、网络传输时间。拆完之后发现，大头在模型推理时间，占了总耗时的70%以上。

定位到瓶颈之后，优化方向就清晰了：要么换更快的模型，要么减少推理的token量，要么做并行。

5.2 超时熔断与快速失败

延迟优化里最容易被忽略的一环是快速失败。如果一个模型已经明显超时了，还傻等着它返回，只会拖累整体体验。我给每个模型设了超时阈值，超过阈值直接熔断，把请求转到备用模型上。

熔断器的实现我用了一个简单的状态机：

class CircuitBreaker: def __init__(self, failure_threshold=5, recovery_timeout=30): self.failure_count = 0 self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.last_failure_time = None self.state = "closed" # closed / open / half-open def can_proceed(self): if self.state == "closed": return True if self.state == "open": if time.time() - self.last_failure_time > self.recovery_timeout: self.state = "half-open" return True return False return True # half-open 允许试探 def record_failure(self): self.failure_count += 1 self.last_failure_time = time.time() if self.failure_count >= self.failure_threshold: self.state = "open" def record_success(self): self.failure_count = 0 self.state = "closed"

这个熔断器帮我挡掉了很多因为单个模型故障导致的连锁反应。实测下来，加了熔断之后，整体P99延迟下降了将近40%。

5.3 流式输出与首字延迟的取舍

流式输出对用户体验的提升是巨大的。用户不用等整个回答生成完，首字一出来就能看到内容在滚动，感知延迟大幅降低。但流式输出也有代价：它会让成本核算和错误处理变复杂。

我的做法是默认开启流式，但在网关层做了一层缓冲。具体来说，网关会先等首字出来，确认模型正常响应了，再把流式数据透传给客户端。这样既保证了用户体验，又能在模型一开始就出问题的时候快速切换。

首字延迟的优化，我主要做了两件事：一是把请求路由到当前负载最低的模型实例上；二是对prompt做了精简，去掉不必要的系统提示词。这两招加起来，首字延迟从平均2.5秒降到了0.8秒左右。

6. 质量兜底：怎么保证便宜模型不翻车

6.1 输出质量的多维度评估

用便宜模型最大的风险就是质量翻车。我的应对策略是建立一套质量评估机制，对每次输出做多维度打分。评估维度包括：

• 相关性：输出和输入是否相关
• 完整性：是否完整回答了问题
• 格式合规：是否符合要求的输出格式
• 长度合理：是否过长或过短

评估方法我用的是“规则+模型”混合的方式。规则能搞定的用规则，比如格式检查、长度检查；规则搞不定的用一个小模型来做质量打分。这个评估模型本身很便宜，跑一次成本可以忽略不计。

6.2 低质量输出的自动重试与升级

如果评估发现质量不达标，网关会自动触发重试，并且这次重试会升级到更高质量的模型。这个机制我叫做“质量兜底”。

具体逻辑是：第一次用便宜模型，评估不通过；第二次升级到中等模型，再评估；还不通过就上最贵的模型。最多重试两次，两次都不行就返回错误，让业务层处理。

def execute_with_quality_guard(request, candidate_models): for i, model in enumerate(candidate_models): response = call_model(model, request) score = evaluate_quality(response, request) if score >= QUALITY_THRESHOLD: return response # 记录一次质量不达标 record_quality_failure(model, score) raise QualityError("所有候选模型均未达到质量要求")

这个机制的效果很明显：大部分请求第一次就通过了，只有少部分需要升级，整体成本还是比全部用贵模型低很多，但质量有了保障。

6.3 质量与成本的动态平衡点

质量兜底不是免费的，每次重试都是额外的成本。所以这里有一个平衡点的问题：质量阈值设高了，重试多，成本上去了；设低了，质量没保障。

我的做法是让业务方自己设这个阈值，并且提供一个“质量-成本”曲线参考。具体来说，我会跑一批样本，统计不同质量阈值下的重试率和总成本，画出一条曲线，业务方看着曲线来定阈值。

实测下来，大部分场景下质量阈值设在0.75左右是比较平衡的，重试率大概在8%到12%之间，总成本比全用贵模型低50%以上。

7. 踩坑实录：那些让我半夜爬起来改代码的问题

7.1 模型限流导致的雪崩

有一次晚上，一个业务方突然放量，请求量翻了十倍。网关按照正常逻辑把请求分发到各个模型，结果好几个模型都触发了限流，返回429。更糟糕的是，我的重试逻辑没有做退避，一收到429就立刻重试，结果把限流搞得更严重，形成了雪崩。

后来我加了两层保护：一是指数退避重试，每次重试的间隔翻倍；二是限流感知路由，网关会记录每个模型的限流状态，被限流的模型在一段时间内不再接收新请求。

def retry_with_backoff(func, max_retries=3, base_delay=1): for attempt in range(max_retries): try: return func() except RateLimitError: if attempt == max_retries - 1: raise delay = base_delay * (2 ** attempt) time.sleep(delay)

这个坑让我明白一个道理：重试逻辑一定要配合退避，否则重试就是火上浇油。

7.2 流式响应的成本统计偏差

前面提到流式输出的成本核算，我一开始是在每个chunk到达的时候累加token数，结果发现统计出来的completion_tokens和实际对不上，偏差有时候能到20%。后来查了半天才发现，有些模型的流式返回里，token的计数方式和非流式不一样，有的把特殊token也算进去了，有的不算。

解决办法是以模型返回的最终usage字段为准，流式过程中只做预估，最终结算用模型给的准确值。如果模型不返回usage，那就用一个tokenizer自己算，但要注意用对应模型的tokenizer，不能混用。

7.3 多模型版本升级带来的兼容性问题

有个模型我用了半年，一直很稳定。结果有一天它悄悄升级了版本，返回格式变了，我这边解析直接报错。更坑的是，它没有版本号字段，我根本不知道它升级了。

后来我做了两件事：一是在适配器里加防御性解析，对可能变化的字段做兼容处理；二是建立模型健康检查机制，定期发一个测试请求，验证返回格式是否符合预期，不符合就告警。

提示：不要假设模型的行为是永远不变的。任何外部依赖都要做防御性编程，这是血泪教训。

8. 这套网关跑了大半年，我的一些真实体会

这套多模型调度网关从搭起来到现在，差不多跑了半年多。整体效果是符合预期的：成本降了六成多，P95延迟从十几秒压到了三秒以内，质量方面因为有了兜底机制，反而比之前单模型的时候更稳定了。

但我想说的是，这套东西不是一蹴而就的。我前后迭代了四五个版本，踩了无数坑，才慢慢打磨到现在这个状态。如果你也想做类似的事情，我的建议是先从统一抽象层做起，把模型接入这层做扎实，后面的调度、成本、质量才有发挥的空间。不要一上来就追求大而全，先跑通一个最小闭环，再逐步加功能。

另外，调度策略没有银弹。我见过有人想找一个“完美算法”一劳永逸，结果折腾了几个月也没落地。实际的做法应该是：先定几个简单的规则跑起来，然后根据真实数据不断调整。数据会告诉你哪里需要优化，而不是靠拍脑袋。

最后分享一个我最近在试的方向：基于历史数据的预测性调度。简单说就是根据历史请求的模式，提前预判接下来一段时间哪些模型会紧张，提前做流量分配。这个还在实验阶段，等跑出稳定结果了再跟大家分享。