多模型调度网关实战：成本、延迟、质量三维路由与语义缓存

1. 从单模型到多模型为什么我决定自建调度网关去年下半年我负责的一个智能问答类项目遇到了一个很典型的问题所有请求都打给同一个大模型结果就是成本失控、高峰期延迟飙升、部分场景回答质量还不稳定。最要命的是老板问我“到底哪个模型在什么场景下表现最好、花了多少钱”我拿不出任何数据。那段时间我试过几种方案。最粗暴的是在业务代码里写 if-else根据请求类型手动切换模型。代码很快就变成了意大利面新增一个模型要改五六个地方测试覆盖不过来。后来想用现成的API网关做转发但那些网关对模型调用的语义支持太弱没法按 token 数计费、没法做语义缓存、更没法根据实时延迟动态路由。折腾了两周之后我决定自己搭一个多模型调度网关。核心目标就三个成本透明、延迟透明、质量透明。成本透明是指每次调用都能追溯到具体模型和 token 消耗延迟透明是指能实时看到每个模型在不同场景下的响应时间分布质量透明是指能通过反馈机制持续评估各模型在具体任务上的表现。这个网关我最终基于蓝耘元生代平台搭建跑通之后效果超出预期。下面我把整个设计思路、核心实现和踩过的坑完整分享出来适合正在做多模型接入、成本优化或者AI应用架构的同行参考。即使你目前只用了一个模型这套思路也能帮你提前把可观测性做好后面扩展时不会手忙脚乱。2. 整体架构设计调度网关到底该管什么2.1 核心需求拆解与边界划定在动手写代码之前我先花了一天时间把需求理清楚。一个多模型调度网关本质上要解决四个层面的问题第一层是接入层。不同模型的API协议不一样有的用OpenAI兼容格式有的用自定义HTTP接口参数命名和返回结构都有差异。网关要做的第一件事就是把这些差异屏蔽掉对上提供统一的调用接口。第二层是路由层。这是网关的核心价值所在。路由策略不能是简单的轮询或者随机而应该基于请求的特征比如任务类型、输入长度、是否需要多轮对话和模型的实时状态延迟、错误率、当前配额余量来做决策。第三层是可观测层。每次调用都要记录完整的上下文请求ID、模型名称、输入输出token数、首token延迟、总延迟、是否命中缓存、用户反馈等。这些数据要能实时聚合也要能按时间窗口回溯。第四层是控制层。包括限流、熔断、降级、A/B测试分流等。当某个模型出现大面积超时或者错误率飙升时网关要能自动把它从可用池里摘掉等恢复后再重新加入。我给自己定的边界是网关只负责调度和观测不负责模型本身的推理优化也不做复杂的业务逻辑。这样职责清晰后续维护成本低。2.2 为什么选择蓝耘元生代作为底座选蓝耘元生代作为底座主要看中三点。一是它对多模型接入的支持比较友好提供了统一的鉴权和配额管理省去了我自己维护多套API Key的麻烦。二是它的调用日志和用量统计可以直接对接我不用从零搭建数据采集管道。三是它的弹性伸缩能力在流量波动时不用我手动干预。当然这不是说其他平台做不到而是当时我的时间预算有限需要一个能快速跑通MVP的方案。如果你用的是其他云平台或者自建推理集群下面的设计思路同样适用只是接入层的适配代码需要相应调整。2.3 网关的整体数据流整个网关的数据流是这样的客户端请求先到网关的接入层接入层做参数校验和标准化然后交给路由层。路由层根据当前的路由策略选择一个目标模型同时检查该模型是否在熔断状态。如果命中语义缓存直接返回缓存结果不调用模型。否则发起模型调用记录调用前后的时间戳和token消耗。调用完成后把结果返回给客户端同时异步写入观测数据。如果调用失败触发重试或降级逻辑。这个流程看起来简单但每个环节都有细节需要处理。比如语义缓存的命中判断、流式响应的延迟测量、重试时的幂等性保证等。下面我会逐个展开。3. 核心模块实现路由、缓存与可观测性3.1 统一接入层屏蔽模型差异的关键设计接入层的核心任务是定义一套内部统一的请求和响应格式。我参考了OpenAI的Chat Completions格式因为它的结构比较通用大部分模型都能映射过来。内部请求对象包含这些字段request_id全局唯一标识用于链路追踪task_type任务类型比如问答、摘要、翻译、代码生成messages对话消息列表每条包含role和contentmax_tokens最大输出token数temperature采样温度stream是否流式返回metadata业务侧附加信息比如用户ID、会话ID响应对象则包含request_id对应请求的IDmodel_name实际调用的模型content生成的文本prompt_tokens输入token数completion_tokens输出token数first_token_latency_ms首token延迟total_latency_ms总延迟cached是否命中缓存每个模型适配器负责把内部格式转换成目标模型的API格式再把返回结果转换回内部格式。适配器用策略模式实现新增模型时只需要写一个适配器类注册到工厂里就行。注意token数的计算不要依赖模型返回的usage字段因为有些模型不返回或者返回不准。我建议在网关侧用统一的tokenizer做一次预计算虽然会多消耗一点CPU但数据一致性有保障。3.2 路由策略成本、延迟、质量的三维权衡路由策略是整个网关最核心也最复杂的部分。我最终实现的是一个加权评分模型每个候选模型会得到一个综合分数分数最高的被选中。评分公式如下score w_cost * cost_score w_latency * latency_score w_quality * quality_score其中cost_score、latency_score、quality_score都是归一化到0到1之间的值。权重w_cost、w_latency、w_quality可以根据业务场景动态调整。比如对成本敏感的场景把w_cost调高对实时性要求高的场景把w_latency调高。cost_score的计算方式是先算出该模型在当前请求预估token数下的费用然后取所有候选模型中费用的倒数再归一化。latency_score用的是该模型过去5分钟内的P95延迟同样取倒数归一化。quality_score则来自人工反馈和自动评估的综合得分初始时所有模型给相同的默认值随着反馈数据积累逐渐分化。这里有个细节quality_score的更新不能太激进。我一开始用最近10次调用的平均分结果发现波动太大路由决策频繁跳变。后来改成指数移动平均平滑系数取0.1稳定了很多。3.3 语义缓存不只是字符串匹配语义缓存是我认为投入产出比最高的一个模块。普通的KV缓存只能命中完全相同的请求但实际业务中大量请求是语义相似的。比如“今天天气怎么样”和“今天天气如何”字符串不同但语义相同。我的实现方案是对请求的messages内容做embedding然后在向量数据库里做相似度检索。如果相似度超过阈值我设的是0.92就认为命中缓存直接返回之前的结果。阈值不能设太低否则会把不同意图的请求误判为相同也不能设太高否则命中率上不去。0.92是我实测下来比较平衡的值。缓存的有效期也要考虑。对于事实类问答缓存可以设长一点比如1小时对于时效性强的请求比如新闻摘要缓存有效期要短甚至不缓存。我在metadata里加了一个cache_ttl字段由业务侧指定。实操心得语义缓存的embedding计算本身也有成本。如果请求量很大建议用一个轻量级的embedding模型或者对短文本做截断。我试过用完整的对话历史做embedding延迟增加了80ms后来改成只对最后一条用户消息做embedding延迟降到15ms以内命中率只下降了不到5%。3.4 可观测性让每一分钱和每一毫秒都有迹可循可观测性模块我分了三个维度来建设。成本维度每次调用记录prompt_tokens和completion_tokens乘以该模型的单价得到本次调用的费用。这些数据按小时、按天、按模型、按业务方聚合写入时序数据库。我用的是一张简单的宽表字段包括时间戳、模型名、业务方、token数、费用。查询时用SQL聚合就行不需要上复杂的OLAP引擎。延迟维度记录首token延迟和总延迟。首token延迟对用户体验影响最大尤其是流式场景。总延迟则影响吞吐量。这两个指标都按P50、P95、P99分位数统计。我用的是滑动窗口算法窗口大小5分钟每10秒更新一次。质量维度质量评估分两种。一种是显式反馈比如用户点赞点踩另一种是隐式反馈比如用户是否重新生成了回答、是否在回答后继续追问。显式反馈准确但稀疏隐式反馈密集但有噪声。我把两者加权结合显式反馈权重0.7隐式反馈权重0.3。这三个维度的数据最终汇聚到一个看板上可以按模型、按时间段、按业务方筛选。看板是我用Grafana搭的数据源直接连时序数据库。这样老板再问我“哪个模型划算”我直接截图发过去就行。4. 实操过程从零搭建网关的完整步骤4.1 环境准备与依赖安装我用的技术栈是Python 3.11 FastAPI Redis PostgreSQL Milvus。FastAPI负责HTTP服务Redis做限流和短期缓存PostgreSQL存调用日志和配置Milvus存语义缓存的向量。安装依赖的命令如下pip install fastapi uvicorn redis psycopg2-binary pymilvus httpx numpy如果你不想用Milvus可以用FAISS做本地向量检索省去一个中间件。但FAISS不支持分布式多实例部署时缓存不共享。我一开始用FAISS后来流量上来后换成了Milvus。数据库建表语句我放在这里主要是三张表call_logs记录每次调用model_configs存模型配置routing_weights存路由权重。CREATE TABLE call_logs ( id BIGSERIAL PRIMARY KEY, request_id VARCHAR(64) UNIQUE, model_name VARCHAR(64), task_type VARCHAR(32), prompt_tokens INT, completion_tokens INT, cost DECIMAL(10,6), first_token_latency_ms INT, total_latency_ms INT, cached BOOLEAN, success BOOLEAN, created_at TIMESTAMP DEFAULT NOW() ); CREATE TABLE model_configs ( model_name VARCHAR(64) PRIMARY KEY, api_endpoint VARCHAR(256), api_key_encrypted VARCHAR(512), input_price_per_1k DECIMAL(10,6), output_price_per_1k DECIMAL(10,6), max_context_tokens INT, enabled BOOLEAN DEFAULT TRUE ); CREATE TABLE routing_weights ( task_type VARCHAR(32) PRIMARY KEY, w_cost DECIMAL(3,2), w_latency DECIMAL(3,2), w_quality DECIMAL(3,2) );4.2 模型适配器的编写与注册适配器我定义了一个基类所有具体模型的适配器都继承它from abc import ABC, abstractmethod class BaseModelAdapter(ABC): abstractmethod async def call(self, request: InternalRequest) - InternalResponse: pass abstractmethod def estimate_cost(self, request: InternalRequest) - float: pass以某个OpenAI兼容格式的模型为例适配器的核心逻辑是构造请求体、发送HTTP请求、解析响应。这里要注意超时设置我设的是连接超时5秒读取超时60秒。流式请求的读取超时要更长因为模型可能生成很长的内容。适配器写好后注册到工厂ADAPTER_REGISTRY {} def register_adapter(model_name): def decorator(cls): ADAPTER_REGISTRY[model_name] cls return cls return decorator register_adapter(model_a) class ModelAAdapter(BaseModelAdapter): ...这样新增模型时只需要加一个类不用改路由层的代码。4.3 路由决策的代码实现路由决策的核心函数大概长这样async def select_model(request: InternalRequest) - str: candidates get_enabled_models(request.task_type) weights get_routing_weights(request.task_type) scores {} for model in candidates: if is_circuit_open(model): continue cost estimate_cost(model, request) latency get_p95_latency(model) quality get_quality_score(model, request.task_type) cost_score 1.0 / (cost 1e-6) latency_score 1.0 / (latency 1e-6) quality_score quality scores[model] ( weights.w_cost * normalize(cost_score, scores) weights.w_latency * normalize(latency_score, scores) weights.w_quality * quality_score ) return max(scores, keyscores.get)归一化函数normalize需要等所有候选模型的分数都算出来后再做所以实际代码里是两轮循环。第一轮算原始分数第二轮归一化并加权求和。注意当所有模型都熔断时要有一个兜底策略。我的做法是选一个默认模型强制调用同时打日志告警。这个默认模型通常选最稳定的那个哪怕贵一点。4.4 流式响应的延迟测量流式响应的延迟测量比非流式麻烦。首token延迟是从发起请求到收到第一个chunk的时间总延迟是从发起请求到收到最后一个chunk的时间。我在适配器里用异步迭代器处理流式响应每收到一个chunk就记录时间戳。async def call_stream(self, request): start time.monotonic() first_token_time None async for chunk in self._stream_request(request): if first_token_time is None: first_token_time time.monotonic() yield chunk end time.monotonic() first_token_latency (first_token_time - start) * 1000 total_latency (end - start) * 1000 await self._log_latency(request.request_id, first_token_latency, total_latency)这里有个坑如果客户端提前断开连接异步迭代器会抛出异常日志就写不进去了。我的处理是在finally块里写日志保证即使异常也能记录。5. 常见问题与排查技巧实录5.1 路由抖动为什么模型选择总是跳来跳去这是我最开始遇到的最头疼的问题。表现是同一个任务类型的请求一会儿走模型A一会儿走模型B导致成本和质量都不稳定。排查下来发现两个原因。一是延迟数据更新太频繁每来一个请求就更新一次P95导致分数波动大。二是质量分数初始值相同但一旦某个模型被选中几次后分数略有变化就会导致它被更频繁地选中形成正反馈。解决办法延迟数据改成滑动窗口每30秒更新一次而不是每次请求都更新。质量分数用指数移动平均并且设置一个最小更新间隔比如每100次调用才更新一次。另外给每个模型加一个探索因子即使分数低也有小概率被选中避免陷入局部最优。5.2 缓存穿透相似度阈值设错了有多可怕我一开始把语义缓存的相似度阈值设成了0.85想着命中率高一点。结果上线第一天就出问题了用户问“如何重置密码”系统返回了“如何修改密码”的答案。虽然语义相近但操作步骤完全不同。后来我把阈值调到0.92并且加了一个规则对于操作类、步骤类的问题缓存命中后还要做一次意图分类校验只有意图完全一致才返回缓存。这个校验用一个轻量级分类模型做延迟增加不到10ms但避免了大部分误命中。实操心得语义缓存不是万能的。对于事实类、常识类问题缓存效果好对于操作类、时效类问题要么不缓存要么加校验。我现在的策略是只有task_type为“faq”和“knowledge”的请求才走语义缓存其他类型直接跳过。5.3 成本统计对不上token计算的那些坑成本统计对不上账是我被财务追问最多的问题。排查后发现几个原因第一有些模型返回的usage字段里prompt_tokens不包含系统提示词的token。但计费时是包含的。所以我在网关侧统一用tokenizer重新计算不信任模型返回的usage。第二流式响应的completion_tokens在最后一个chunk才返回如果客户端提前断开这个数据就丢了。我的处理是在网关侧累计每个chunk的token数不依赖模型的最终统计。第三不同模型的tokenizer不一样同样的文本token数可能差20%以上。所以成本估算时要用对应模型的tokenizer不能用统一的。下面这张表是我整理的常见问题速查问题现象可能原因排查方法解决方案路由频繁跳变延迟数据更新过快查看路由日志中分数变化滑动窗口最小更新间隔缓存误命中相似度阈值过低抽查缓存命中记录提高阈值意图校验成本对不上token计算口径不一致对比网关和模型返回的usage网关侧统一重算流式延迟缺失客户端提前断开检查日志写入时机finally块写日志熔断后无可用模型熔断阈值过严查看熔断触发记录设置兜底模型告警5.4 熔断与降级的边界条件熔断逻辑我踩过一个坑一开始设置的是错误率超过50%就熔断结果某个模型因为一次网络抖动连续5个请求失败错误率瞬间到100%直接被熔断。但其实第6个请求就恢复了。后来我改成基于滑动窗口的错误率窗口大小20个请求错误率超过60%才熔断并且熔断后每隔30秒放一个探测请求过去成功了就恢复。这样既避免了误熔断又能快速恢复。降级策略也要提前设计好。我的降级顺序是先重试同一个模型最多2次再切换到备用模型最后返回兜底话术。兜底话术虽然体验不好但比直接报错强。6. 运行效果与后续优化方向这套网关上线跑了三个月效果比我预期的好。成本方面通过路由策略把简单任务分流到低价模型整体费用下降了约35%。延迟方面P95延迟从原来的2.8秒降到了1.6秒主要是因为语义缓存命中率达到了22%这部分请求直接返回不消耗模型推理时间。质量方面通过持续的质量反馈高价值任务逐渐集中到表现最好的两个模型上用户点赞率提升了12个百分点。后续我打算在几个方向继续优化。一是引入更细粒度的质量评估比如按知识领域分类而不是笼统地按任务类型。二是把路由策略从静态权重改成基于强化学习的动态调整让系统自己学习最优的权重组合。三是把网关的能力开放出去让业务方可以自定义路由规则比如指定某些请求必须走特定模型。不过这些都是后话了。当前这套方案已经足够支撑日均百万级的调用量而且维护成本很低。如果你也在做多模型接入建议先从可观测性做起把成本、延迟、质量的数据采集做好再逐步引入路由和缓存。数据是决策的基础没有数据再复杂的路由策略都是拍脑袋。最后分享一个我在实际运维中总结的小技巧每周花十分钟看一下路由权重的变化趋势。如果某个模型的权重持续下降说明它可能出了问题或者有更优的替代品出现了。这个习惯帮我提前发现了好几次模型性能退化的情况比等用户投诉再处理主动得多。