1. 从单模型硬扛到多模型调度我为什么折腾这套网关去年下半年我手上有个内容生成类的项目一开始图省事所有请求都往同一个大模型上怼。刚开始量小感觉还行响应质量也说得过去。但跑了两三个月问题全冒出来了高峰期延迟飙到十几秒账单一个月比一个月吓人而且有些任务明明用轻量模型就能干得不错我却一直在用最贵的那个在跑。说白了就是“一个模型打天下”的粗暴打法成本、延迟、质量三本账全是糊涂账。后来我花了两周时间基于蓝耘元生代平台搭了一套多模型调度网关核心目标就一个让每一次请求都能被路由到“当下最合适”的模型上同时把成本、延迟、质量这三个维度全部透明化。这套东西跑到现在差不多半年整体推理成本降了六成多P95延迟从十几秒压到了三秒以内而且质量没有肉眼可见的下降。这篇文章我会把整套网关的设计思路、核心调度策略、成本核算方法、延迟优化手段以及我在实操中踩过的坑全部摊开来讲。适合正在被大模型调用成本困扰的开发者、需要做多模型选型的技术负责人以及任何想搞清楚“多模型调度到底怎么落地”的人。不管你是刚接触这块还是已经在做类似的事情应该都能从里面找到能直接抄作业的东西。2. 蓝耘元生代平台上的模型接入与统一抽象层设计2.1 为什么先要做统一抽象层多模型调度最怕的一件事就是每接一个新模型就要改一遍业务代码。我一开始就是犯了这个错第一个模型接进去的时候业务层直接调了它的SDK参数格式、返回结构全写死在业务逻辑里。结果接第二个模型的时候发现两边的请求体结构完全不一样一个用messages数组一个用prompt字符串返回的字段名也对不上。改起来那叫一个痛苦。所以第二版我做的第一件事就是抽一层统一模型接口。所有模型不管底层是哪家、什么版本对外暴露的请求和响应格式全部统一。业务层只认这一套格式完全不关心背后是谁在干活。具体来说我定义了一个标准的请求结构class UnifiedRequest: def __init__(self, messages, max_tokens1024, temperature0.7, streamFalse, task_typeNone, prioritynormal): 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 模型能力标签体系的建立光有适配器还不够调度的时候得知道每个模型擅长什么。我建了一套能力标签体系给每个模型打上多维度的标签标签维度说明示例值任务类型擅长的任务类别对话、代码、摘要、翻译上下文长度支持的最大上下文4K、8K、32K、128K输出质量综合质量评级高、中、基础响应速度平均首字延迟快1s、中1-3s、慢3s成本等级每千token成本高、中、低稳定性历史可用性高、中这套标签不是拍脑袋定的是我在实际跑了一段时间之后根据每个模型在真实任务上的表现慢慢校准出来的。比如有个模型标称支持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_size100): self.latencies deque(maxlenwindow_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。网关在记录成本的时候会把这两个维度一起存下来。这样我就能生成这样的报表业务方任务类型调用次数总成本平均单次成本业务A摘要12,34045.6元0.0037元业务A对话8,92089.2元0.0100元业务B代码生成3,450156.8元0.0454元业务B翻译6,78023.4元0.0035元有了这张表优化方向就一目了然了。比如业务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_threshold5, recovery_timeout30): 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_retries3, base_delay1): 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延迟从十几秒压到了三秒以内质量方面因为有了兜底机制反而比之前单模型的时候更稳定了。但我想说的是这套东西不是一蹴而就的。我前后迭代了四五个版本踩了无数坑才慢慢打磨到现在这个状态。如果你也想做类似的事情我的建议是先从统一抽象层做起把模型接入这层做扎实后面的调度、成本、质量才有发挥的空间。不要一上来就追求大而全先跑通一个最小闭环再逐步加功能。另外调度策略没有银弹。我见过有人想找一个“完美算法”一劳永逸结果折腾了几个月也没落地。实际的做法应该是先定几个简单的规则跑起来然后根据真实数据不断调整。数据会告诉你哪里需要优化而不是靠拍脑袋。最后分享一个我最近在试的方向基于历史数据的预测性调度。简单说就是根据历史请求的模式提前预判接下来一段时间哪些模型会紧张提前做流量分配。这个还在实验阶段等跑出稳定结果了再跟大家分享。