说到底这个项目最初只是我自己的一个“偷懒”工具。手头接了好几个开源模型API 也买了不止一家时间一长就发现一个问题模型越多反而越不知道该调哪个。同样的请求有时候大模型效果好但贵得离谱有时候小模型又快又便宜但明显答非所问。Model-Optimizer 就是在这样的背景下被我写出来的——它不是某个具体的模型而是一层夹在“调用方”和“多个模型后端”之间的优化调度组件负责给每一次请求分配合适的模型顺手把推理参数也一起调了。如果你正在维护多个模型接口或者在自托管模型上做应用整天被模型选型、成本控制、延迟波动这些事情折腾这篇文章应该能给你一些参考。我会从设计思路、核心功能、代码实现到上线后的坑逐个讲清楚。不说空话全是我实际踩过的。1. 项目概述为什么需要这么一层优化组件1.1 模型一多决策反而难了很多人以为模型越多效果越好实际用起来完全不是这么回事。单一模型的时候你只需要关心“这个模型行不行”模型多起来以后问题变成“同一个请求到底该给谁”。而且这个决策不是做一次就完了同一个用户在同一个会话里发的两条消息可能一条适合大模型深度推理另一条小模型就能接住。这时候如果你在业务代码里硬编码模型名称比如所有请求都走某个 70B 的模型效果是稳定了但成本和延迟会把你逼疯。我实际测过一个场景同样一段文本总结任务70B 模型平均要 4 到 6 秒才能返回7B 模型只要 1.5 秒左右而两者的输出质量在简单任务上几乎没有差别。如果所有流量都打到大模型上GPU 显存占用和响应时间都会很难看。所以真正的问题不是“哪个模型最强”而是“在可接受的质量范围内怎么让每一次请求都走最合适的路径”。Model-Optimizer 就是想把这套决策逻辑从业务代码里抽出来做成一个独立的、可配置的、带缓存和可观测性的中间层。1.2 Model-Optimizer 的整体设计这个组件的定位很简单它是一个反向代理式的服务接收上游发来的请求经过分类、路由、参数调整之后再把请求转发给真正的模型后端。整体结构分四层入口层接收 HTTP 请求解析出模型要处理的内容、会话上下文、用户标识决策层根据请求特征和配置规则决定“走哪个模型”“用什么参数”“是否命中缓存”执行层把最终的决定落地去调用具体的推理后端比如 vLLM、TGI、llama.cpp 或者云厂商的 API观测层把每次请求的路由原因、耗时、成本、缓存命中情况记录成结构化日志。这个分层最大的好处是每一层都能独立调整。比如今天发现某个模型的推理质量明显下降我可以只改配置规则不用动任何业务代码明天新增了一个更好的模型只需要在后端列表里加一条记录再调整一下路由权重就行。1.3 这层组件适合谁、不适合谁先说不适合的场景如果你只接了一个模型不管它是自托管还是云 API都完全不需要 Model-Optimizer中间多套一层纯属给自己找麻烦。另外如果业务请求对延迟极其敏感每一跳的耗时都很关键那也不要硬上因为这一层会带来少量的额外开销。适合的场景我觉得有三类。第一类是自托管了多个不同规模模型想省 GPU 成本的人第二类是同时接了多个云厂商模型 API想做容灾和成本控制的人第三类是构建复杂 AI 应用需要根据用户输入动态选择模型路线的人。我自己属于第一类和第三类的交集所以越用越顺手。2. 核心功能拆解与设计取舍2.1 请求分类不只按关键词在 Model-Optimizer 里请求分类是最基础的一块。我见过不少人的第一反应是“按关键词匹配”比如输入里出现“代码”就路由给代码模型出现“翻译”就路由给翻译模型。但实际场景里用户的输入往往没有这么明确的关键词。我的做法是给每个请求打一组特征标签。首先是意图类型通过一个轻量分类模型我用的是一 个很小的 300M 左右的嵌入模型加一个线性分类头来识别其次是内容复杂度比如 token 长度、是否包含多轮上下文、是否包含代码块最后是风险敏感度比如是否包含个人身份信息、是否涉及医疗或财务建议这类需要大模型谨慎处理的内容。这组标签会作为后续路由决策的输入。这里有个关键设计分类结果不是非黑即白的而是带权重和置信度的。比如一个请求有 70% 的概率是“翻译”30% 的概率是“改写”路由层会把两层概率都算进去而不是简单粗暴地只走翻译模型。2.2 路由打分与动态权重路由决策是我这个项目里迭代次数最多的模块。最初版本写的是硬规则比如“如果意图是翻译则走模型A”。后来发现硬规则在真实流量面前太脆弱了用户的表述稍微变一下分类结果一变路由质量就会波动。最终我改成了打分制。每个模型后端都有三个维度的分数质量分、成本分、性能分。质量分来自离线评测比如在固定测试集上的 BLEU、准确率指标成本分是每千 token 的单价换算出来的性能分则来自实时监控的响应时间和吞吐量。最终得分是三个维度带权重的加权和权重可以在配置文件里动态调整。举个例子假设有两个模型模型X质量分 0.9、成本分 0.3、性能分 0.6模型Y质量分 0.6、成本分 0.8、性能分 0.9。如果权重设置是质量 60%、成本 30%、性能 10%模型X得分是 0.9×0.6 0.3×0.3 0.6×0.1 0.69模型Y得分是 0.6×0.6 0.8×0.3 0.9×0.1 0.69两边打平。这时候就让配置里的兜底策略来决定比如“平局时偏向更快的一方”。2.3 语义缓存省钱的关键如果说路由决策让请求走得更聪明那语义缓存就是实打实帮你省钱的功能。常规的精确缓存很好理解同样的请求直接返回上一次的结果但真实场景里用户很少会一字不差地问同一个问题更多是“意思相同、说法不同”。Model-Optimizer 里我实现了一层语义缓存。流程是这样的请求进来后先用嵌入模型把输入转成向量然后在缓存库里做相似度检索如果找到了相似度超过阈值的记录并且缓存里的答案新鲜度还在有效期之内就直接返回那条答案不再调用下游模型。这里最需要调的是阈值。阈值设太高比如 0.98几乎只有字面重复才能命中缓存失去意义阈值设太低比如 0.85会频繁出现“看起来像但其实不是同一件事”的错误命中。我最后是在 0.92 到 0.95 之间取的平衡点并且对每个输入都会保留向量特征和原文本方便排查误命中。2.4 推理参数自适应同一个模型在不同请求上的最佳参数其实是不同的。比如“写一首关于夏天的短诗”和“把这段 Python 代码改成 TypeScript”对 temperature 和 max_tokens 的需求完全不同。Model-Optimizer 会在请求进入后端之前动态重写采样参数。我总结了几个常用规则。创意生成类的任务写作、头脑风暴会把 temperature 调到 0.8 到 1.0代码生成和翻译任务会把 temperature 降到 0.1 到 0.3事实性问答保持 0.3 到 0.5。max_tokens 则根据输入长度和任务类型预估比如简单分类任务给 128 就够长文总结至少给 1024。这套逻辑写起来不难但它有一个隐蔽的坑不同推理后端对参数的支持程度不一样。vLLM 支持 top_p、frequency_penalty 这类参数但有些端侧推理框架只支持 temperature 和 max_tokens甚至有的连 temperature 都不支持。所以参数自适应模块必须带一个能力探测机制在启动时检测后端支持的参数列表只传支持的那部分。3. 实操过程落地部署与代码实现3.1 依赖清单与最小环境Model-Optimizer 的代码本身不复杂最重的依赖是嵌入模型和向量检索库。我说一下我实际用的最小环境你按需增减就好Python 3.10 或 3.11建议用 3.11性能差距在 API 服务场景下还是有感的FastAPI 和 uvicorn做 HTTP 入口层sentence-transformers加载嵌入模型做语义向量我用的模型是 bge-small-zh-v1.5体积小效果稳faiss-cpu做向量索引请求量不大用 CPU 版足够httpx做异步转发请求给下游推理后端prometheus-client暴露指标给监控系统。安装命令没什么特殊一条pip install fastapi uvicorn sentence-transformers faiss-cpu httpx prometheus-client就能搞定。需要提醒的是如果机器没有 GPU嵌入模型加载后第一次推理会有点慢建议启动时做一次预热把模型加载和向量索引构建都放进启动事件里。3.2 配置文件怎么写我强烈建议把路由规则和模型列表全部外置成配置文件不要写死在代码里。我自己用的是 YAML 格式因为可读性比 JSON 好而且支持注释方便不同模型负责人各自维护自己的段落。models: - name: fast-chat base_url: http://localhost:8001/v1 api_key: none weight: quality: 0.6 cost: 0.4 speed: 0.3 capabilities: temperature: true top_p: false - name: precise-chat base_url: http://localhost:8002/v1 api_key: none weight: quality: 0.95 cost: 0.2 speed: 0.5 capabilities: temperature: true top_p: true max_tokens_limit: 8192 routing: cache_threshold: 0.93 cache_ttl: 3600 balance_strategy: speed # 平局时偏向更快的模型 params: temp_map: creative: 0.9 code: 0.2 factual: 0.4这个文件里weight是每个模型在质量、成本、速度三个维度上的固定属性分routing里是全局策略。我在实际使用中会经常调整模型权重尤其是某个模型换了版本之后可以做到秒级生效因为配置加载模块会在每次请求时检查文件的更新时间。3.3 核心代码入口与路由路由逻辑是整个项目的主干。我把它写成了一个独立的类这样测试起来方便不会和 API 路由缠在一起。核心思路就是前面说的分类、打分、缓存三步。import numpy as np from sentence_transformers import SentenceTransformer import faiss class Router: def __init__(self, model_cfg, routing_cfg): self.encoder SentenceTransformer(BAAI/bge-small-zh-v1.5) self.models model_cfg[models] self.cache_threshold routing_cfg[cache_threshold] self.cache_ttl routing_cfg[cache_ttl] self.index faiss.IndexFlatIP(512) self.cache_keys [] self.cache_values [] self.cache_time [] def embed(self, text): vec self.encoder.encode(text, normalize_embeddingsTrue) return vec.astype(np.float32) def search_cache(self, query_vec): if len(self.cache_keys) 0: return None scores, idx self.index.search(query_vec.reshape(1, -1), 1) if scores[0][0] self.cache_threshold: cached_id idx[0][0] age time.time() - self.cache_time[cached_id] if age self.cache_ttl: return self.cache_values[cached_id] return None def add_cache(self, query_vec, answer): self.index.add(query_vec.reshape(1, -1)) self.cache_keys.append(query_vec) self.cache_values.append(answer) self.cache_time.append(time.time()) def score_model(self, model, quality_score, cost_score, speed_score): w model[weight] return quality_score * w[quality] cost_score * w[cost] speed_score * w[speed]这里有个细节值得说一下IndexFlatIP是内积索引但我在编码时开了normalize_embeddingsTrue所以内积和余弦相似度是等价的这样在检索时可以直接比较分数不用每次额外算余弦。缓存列表我用的是 Python 列表加 faiss 索引的组合项目刚起步时数据量不大完全够用如果缓存条目超过十万条再考虑上 Redis 或者单独的向量库。3.4 推理后端对接协议路由决定之后剩下的工作就是转发请求。我在 Model-Optimizer 里统一走 OpenAI 兼容协议这样无论后端是 vLLM、TGI 还是云厂商 API都能用同一套代码对接。import httpx async def call_backend(model, body): url model[base_url] /chat/completions headers {Authorization: fBearer {model[api_key]}} async with httpx.AsyncClient(timeout60) as client: resp await client.post(url, jsonbody, headersheaders) resp.raise_for_status() return resp.json()实际使用中我遇到过一个很实际的问题不同后端的 OpenAI 兼容实现有细微差异。比如 vLLM 的/chat/completions接口支持extra_body参数传入best_of、use_beam_search这些高级选项但有些云 API 会直接报参数不支持的错误。所以我做了个capabilities探测在调用时过滤掉后端不支持的参数。我在对接时主要关注三个返回字段choices里的内容、usage里的 token 消耗、以及model字段是否和请求一致。后两个字段是成本核算和质量监控的重要数据建议在转发结果时一并记录下来不要只返回给调用方就完事。4. 上线后的坑排查实录与速查表4.1 缓存命中率始终上不去Model-Optimizer 刚上线那两周语义缓存的命中率一直徘徊在 5% 到 8% 之间低得让人怀疑这个功能是不是白做了。后来我把缓存记录导出分析发现两个问题。第一是我把整个对话历史都放进了缓存键而真实用户请求往往只有前面几轮是重复的整体不重复第二是阈值设得太高很多明显语义相同的改写表达都被过滤掉了。解决办法是只对“当前轮次的用户输入”做缓存匹配并且把系统提示词单独拿出来不参与相似度计算。阈值也从 0.95 降到了 0.93。调整之后同样一批流量命中率提升到了 23% 左右虽然不算特别高但已经能省下不少成本了。另外我加了一个强制跳过机制如果请求里带了cache: false标记就直接绕过缓存查询方便业务方在需要实时数据的时候主动避开缓存。4.2 路由决策“抖”得厉害另一个比较头疼的问题是路由不稳定。同一类请求有时候走大模型有时候走小模型业务方抱怨响应质量忽高忽低。排查下来发现原因在分类模型上输入长度稍微一变分类的置信度就跟着波动导致最终得分在两家模型之间横跳。我给路由加了一个“粘性”机制。针对同一个会话 ID 的连续请求只要会话还没结束就尽量保持上一次的路由选择。这个逻辑听起来简单但它能极大改善用户体验因为同一个对话中如果每轮都换模型上下文会丢失模型对前文的理解会明显变差。实现上就是在路由决策前查一下当前会话最近一次用的模型如果它依然可用就优先沿用。4.3 并发一高就超时项目在测试环境一切正常一上生产就露馅。模型后端的并发能力是有限的当 Model-Optimizer 同时转发大量请求给同一个后端队列很快被打满响应时间直接翻倍。第一次遇到这个问题时我还以为是后端崩了查了半天才发现是接入了大量带缓存的请求同时穿透打垮了下游。处理方法有两个层面。第一层是给每个后端加一个信号量限流比如最多允许同时 16 个并发请求超过的部分在 Model-Optimizer 侧排队等待第二层是给不同后端配置不同的超时时间如果某个后端排队超过设定值就触发熔断临时把流量切换到备用后端。这两招配上以后生产环境的 P99 延迟稳定在了原来的一半左右。4.4 埋点数据对不上账上线初期我们最自信的就是可观测性部分因为日志打得很全。结果对账的时候发现记录的 token 消耗和费用账单对不上差了不少。后来发现原因是有些模型后端在流式输出的时候usage字段是分段返回的如果你在最后一段才去读取usage会丢掉前面几段的 token 计数。这个问题在接入流式响应时经常遇到。解决办法是不要等最终响应里的usage而是在流式过程中对每个 chunk 的choices[0].delta.content做 token 估算最后再用完整的usage做校准。如果你的后端不支持流式返回usage就必须自己做 tokenizer 统计用 tiktoken 或者模型对应的 tokenizer 对最终拼接好的文本重新计算一遍。这套组件在我这边已经跑了小半年后续我还想加两个能力一个是模型版本自动评测定期把暗样本送给各模型打分反向修正路由权重另一个是 cost budget 告警当某个模型当天的消耗超过预设额度时自动把它的路由优先级降到最低。如果你也在折腾多模型接入别急着把所有智能都堆在业务层先想想能不能在“选模型”这件事上把决策做好省下的成本比你想的多得多。