模型网关实战：一个接口聚合多家大模型API，成本直降87.5%

1. 项目缘起：被多模型 API 协议差异按在地上摩擦的日子

如果你也被"一个业务要接 N 个模型 API，每家的参数、鉴权、错误码还都不一样"折磨过，那这篇内容应该能帮到你。

大概半年前，我负责的业务开始大规模接入各家大模型的对话能力。最开始只有一个场景，量不大，直接调某家头部厂商的接口，简单粗暴。后来业务扩张，对话、摘要、情感分析、代码片段生成、结构化抽取全都要做成内部能力，问题立刻爆发了——每家厂商的接口风格完全不像是一个物种：

• 有的用 Bearer Token，有的要自定义 Header 传 Key；
• 有的报文是 messages 数组加 temperature，有的却是 prompt 加 top_p；
• 有的 429 表示限流，有的同样情况返回 500，还有的在业务错误码里藏一个"限流"；
• 有的默认超时 30 秒，有的默认 10 秒，想改还找不到参数；
• 同一个提示词在不同模型上的输出风格天差地别，prompt 要针对每家单独调；
• 最要命的是计费：输入和输出分开计价，各家对缓存 token 的优惠政策还不一样。

我梳理了一下，发现我实际在维护的不是"一个 AI 能力"，而是七八套相互不兼容的接口。联调、翻文档、处理报错、对账单，这些破事消耗的时间比写业务逻辑还多。跟同事聊了一圈，发现这不是我一个人的问题，很多做 AI 应用落地的团队都在被同类问题消耗。

1.1 当时的痛点清单

如果只总结成一句话，就是：厂商绑定。业务代码里到处是某个供应商 SDK 的痕迹，想换一家模型，几乎等于重写一大片逻辑。具体拆开是这样的：

第一，切换成本极高。每次想对比一家新模型的效果，都要写一套新的调用代码，处理一套新的错误体系，灰度验证一周才能得出结论。对比三家模型的时间，够写两个业务模块了。

第二，可用性被单点绑架。某家厂商晚上高峰期一限流，我们的业务就跟着抖。没有备用通道，只能干等。

第三，成本完全失控。所有请求一股脑发给旗舰模型，每月账单下来的时候，财务问了我三次"为什么这么贵"。

第四，可观测性为零。出了质量问题，只能去翻供应商控制台，连"这次请求花了多少钱、用了哪个模型"这种基础信息都查不到。

1.2 中转层到底解决什么问题

我最后决定自己动手，做一个模型网关，在业务服务和各家模型之间加一层。对外只暴露一个统一风格接口，对内负责协议转换、密钥管理、路由分发、缓存、重试、计费和日志。业务方感知不到背后是哪个模型，也不需要感知。

做完之后，效果确实立得住：一个接口聚合了多家模型，内部所有团队接一次就能用所有供应商；成本经过三个月灰度调优下降了 87.5%。这篇文章把整套方案的设计思路、核心代码、成本测算和踩坑记录整理出来，给同样被多 API 困扰的开发者和团队参考。

这个方案适合谁？后端和全栈开发者，想在不大改业务代码的前提下接入多家模型的团队，以及想认真把模型调用成本降下来的个人项目。不需要你有很深的算法背景，会写接口、用过 Redis 就能跟上后面的内容。

2. 整体设计与技术选型：先想清楚再动手

2.1 架构分四层，边界要清晰

这个中转层本质上是 AI 网关。动手之前我给自己定了五条硬性指标，后面所有设计都围绕这五条展开：

1. 业务侧零改造：对外只暴露一个统一接口，响应结构保持一致，业务方永远不用改代码；
2. 供应商可插拔：新增一家模型厂商只需要写一个适配器，不动路由和业务代码；
3. 智能路由：不同请求自动分发到合适的模型，质量和成本兼顾；
4. 故障转移：主供应商限流或宕机时自动切到备选模型，业务无感；
5. 可观测、可计费：每个请求的成本、耗时、模型、token 消耗全部落日志，方便对账和调优。

架构分成四层，各层的边界一定要清楚，否则后面越写越乱。

接入层是统一入口，对外提供标准接口风格的 /v1/chat/completions，负责鉴权、限流、参数校验。路由层是策略引擎，根据请求内容、用户预设策略和模型健康状态，决定发往哪个供应商的哪个模型。适配层是供应商连接器，每个上游一个 adapter，完成协议转换、签名、超时设置、错误标准化。缓存与存储层用 Redis 存缓存和限流计数，用 MySQL 存请求日志和计费明细。

一句话概括：业务方眼里只有一个"模型"，真正的切换、容错、省钱都发生在这层里。

2.2 技术选型：Python + FastAPI + Redis + Docker 的理由

技术选型我纠结了好几天，最后确定 Python + FastAPI + Redis + Docker Compose。理由如下。

第一，模型调用是典型的 IO 密集场景，等待网络响应占绝大部分时间。FastAPI 基于 async/await，可以优雅地并发处理大量模型请求，不用维护复杂的线程池。一个进程扛几千并发请求没什么问题。

第二，AI 生态的工具链在 Python 里最全。后面做语义缓存要算 embedding 相似度，做路由要跑小模型分类，Python 里 NumPy、向量库、SDK 的选择面最广，省去跨语言调用的麻烦。

第三，团队都能写 Python，维护门槛低。这种基础设施类项目最怕写成"只有一个人能改"的代码。

存储方面，Redis 承担语义缓存和限流计数两件事。关于语义缓存多说两句：普通缓存是 key 完全一致才命中，但模型场景里用户的问题经常是"换了个说法"，比如"怎么配超时时间"和"超时时间怎么配"，字面完全不同，意思一样。语义缓存会把问题做成向量存进 Redis，新请求来了算相似度，超过阈值就认为是同一个问题，直接返回缓存结果。这个需求 Redis 自带的 Hash 加向量相似度就可以做，不需要引入额外的向量数据库。

部署用 Docker Compose，一个网关服务加一个 Redis 就能跑起来，后面加供应商纯改文件配置。不上 Kubernetes 是因为这个阶段的核心任务是快速验证成本优化方案，单体部署的调试成本最低，等确实需要水平扩展了再迁不迟。

3. 核心实现：一步步把模型网关搭起来

3.1 统一请求模型：业务方只接触一个接口

中转层对外暴露的格式，我选择了业界最通用的 Completions 风格协议，把 messages 数组作为标准输入。业务方只需要传 model（可选）、messages、temperature、max_tokens，网关统一返回 choices、usage 等字段。

json复制{
  "model": "auto",
  "messages": [
    {"role": "system", "content": "你是一名客服助手，回答要简洁专业。"},
    {"role": "user", "content": "我的订单什么时候发货？"}
  ],
  "temperature": 0.3,
  "max_tokens": 500
}

这里有一个隐藏的好处：各家供应商的官方 SDK 几乎都兼容这套风格。业务方如果以后想绕过中转层直连某家模型，只需要改一下 base_url，代码基本不用动。这相当于给未来留了一条退路，反正协议是通用的。

响应格式也要统一。我除了返回标准字段外，还加了一个 x-cache 响应头和 request_id，这样缓存是否命中、这次请求到底用了哪个模型都能追溯。usage 里的 token 消耗按真实模型响应填充，方便计费系统直接读取。

3.2 供应商适配器：把千奇百怪变成千篇一律

适配器是整个网关里最枯燥但最重要的部分。我给每个供应商写一个类，统一实现 chat 方法，内部处理协议转换、鉴权和错误标准化。

python复制class BaseAdapter:
    """所有供应商适配器的基类，统一接口约定"""
    async def chat(self, messages, params):
        raise NotImplementedError

class ProviderAAdapter(BaseAdapter):
    def __init__(self, api_key, endpoint, extra_headers=None):
        self.api_key = api_key
        self.endpoint = endpoint
        self.extra_headers = extra_headers or {}
        self.session = aiohttp.ClientSession()

    async def chat(self, messages, params):
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            **self.extra_headers,
        }
        payload = self.to_provider_payload(messages, params)
        async with self.session.post(
            self.endpoint, json=payload, headers=headers, timeout=60
        ) as resp:
            if resp.status != 200:
                raise await self.normalize_error(resp)
            return self.from_provider_response(await resp.json())

注意，适配器不仅要转协议，更要把错误标准化。每家返回的错误格式千奇百怪，我统一转成内部错误码，这样上层路由只需要关心"这是限流还是超时"，完全不需要理解每家的错误细节。

还要给每个供应商配独立的超时时间。有的供应商长文本生成确实慢，统一 30 秒超时会误杀；有的供应商性能差，给 60 秒就是浪费等待时间。我在配置里用 provider.timeout 字段单独控制，默认 45 秒。

3.3 路由策略：设置一个会思考的"交通警察"

路由是整个网关的大脑。我采用两层决策，先用规则做硬判断，规则判断不了的交给小模型分类。

规则层处理那些特征非常明显的请求。我总结了几条核心规则：

• 用户显式指定 model 的请求，直接尊重用户意愿发往指定模型，比如代码生成场景用户就是要能力最强的模型；
• 任务类型为简单问答、关键词提取、短文本改写、格式整理的，默认发往轻量模型；
• 输入或输出 token 预计超过设定阈值的，发往上下文窗口更大的长文本模型；
• 需要稳定结构化 JSON 输出的任务，发往我们验证过格式最稳的模型；
• 同一会话的后续消息，优先跟随会话首次选择的模型，避免回答风格跳变。

规则层判断不了的请求，走模型选择器：用一个很便宜的小模型对请求做三分类——简单、中等、困难——再映射到对应的模型组。这里有个降本技巧，路由用的这个分类模型是网关里成本最低的一档，因为分类任务小模型就够用了，没必要请旗舰模型来判断"该不该用旗舰模型"。

python复制async def route_request(req, profile):
    if req.get("model") in allowed_models:
        return req["model"]

    # 规则路由：任务类型分配合法模型
    task = classify_task(req["messages"])
    route = routing_table(profile).match(task, count_tokens(req["messages"]))

    # 健康状态检查：主模型不可用则走备选
    if route.requires_fallback(health_status):
        route = route.fallback_target()

    return route.model_name

路由服务还要维护一份健康状态表，定期检测每个供应商的延迟和错误率。如果某供应商连续失败超过阈值，自动把流量切到备用方案。这个健康状态表和后面的熔断逻辑是配套的，后面细说。

3.4 语义缓存：命中一次，省一次钱

缓存是省钱的第一大杠杆，我用的是语义缓存。流程是：

1. 请求进入网关，先对 messages 里最关键的当前用户消息做 embedding，得到向量；
2. 在 Redis 里搜索相似向量，相似度大于阈值就说明是同一个问题或者极相似的问题；
3. 命中后做一项条件判断：如果缓存结果的 token 消耗比重新调用便宜，就直接返回缓存；
4. 未命中则正常路由、调用供应商，返回前把答案和问题向量写回缓存。

这里有个很多人踩过的坑：不能把整个对话历史都拿去算 embedding。整段历史向量化之后，稍微换一个话题就被判定成新问题，缓存永远不命中。最佳实践是取当前用户消息，加上最近一轮系统提示，拼接后做 embedding。这样既能保留意图上下文，又不会因为历史过长导致向量漂移。

python复制import numpy as np
from redis import Redis

CACHE_PREFIX = "semantic_cache:v1:"
SIMILARITY_THRESHOLD = 0.96  # 初始值，后续用真实数据校准

def query_semantic_cache(redis: Redis, user_msg: str):
    vec = compute_embedding(user_msg)
    # 请求量上来之后，把这段换成 RedisSearch 或向量数据库的近似检索
    for key in redis.scan_iter(match=CACHE_PREFIX + "*"):
        cached = redis.hgetall(key)
        if cosine_sim(np.array(vec), np.array(cached["embedding"])) >= SIMILARITY_THRESHOLD:
            return json.loads(cached["response"])
    return None

小规模场景直接遍历取余弦相似度没问题；请求量上来之后，建议把缓存检索换成带向量索引的存储，不然 scan 全库的耗时会影响网关延迟。阈值也不要拍脑袋定，先在测试集上人工标注一批"同一个问题换说法"和"不同问题但语义接近"的样本，跑一遍相似度分布，再选一个能把两类完全分开的阈值。

缓存命中后返回的是完整响应，但 usage 信息要重新构造。我返回的 usage 里 prompt_tokens 按真实输入计算，completion_tokens 记 0，然后在响应头里加 x-cache: hit。这样业务方和日志系统都能感知到缓存命中情况，对账的时候也说得清楚。

3.5 故障转移与重试：别让供应商抖动拖垮业务

模型供应商的稳定性参差不齐，晚高峰限流、超时、5xx 都是家常便饭。网关必须有重试和切换策略，我的策略分三层。

第一层是同供应商重试。遇到网络闪断、瞬时 5xx，按指数退避重试一到两次，退避基数 1 秒。第二层是同供应商换模型。比如主模型限流，但同一家厂商还有其他可用模型，切到热备模型再试一次。第三层是跨供应商切换。上面都失败，把请求发给备用供应商的等价模型。

python复制async def call_with_failover(messages, params, candidates, budget):
    for idx, target in enumerate(candidates):
        try:
            resp = await adapter_for(target).chat(messages, params)
            return resp
        except RateLimited:
            circuit_breaker(target).record_failure()
            if budget.exceeded(target) or idx >= len(candidates) - 1:
                raise
            continue
        except Timeout:
            circuit_breaker(target).record_failure()
            continue

重试这里有几个关键细节。第一是幂等性，生成类接口重试可能造成重复扣费但结果不同，所以只在"明确失败"时重试，超时场景宁可切换，不要盲目重试同一个供应商。第二是熔断，单个供应商连续 5 个 5xx 就打开熔断 30 秒，期间流量全部切走，防止雪崩。第三是预算保护，网关统计当天每个供应商的消耗金额，超过日预算上限直接拒绝路由，宁可不调用也不能超支。

4. 成本直降 87.5% 的测算与省钱杠杆拆解

4.1 优化前：一个旗舰模型打天下

写代码容易，把钱省下来难。成本优化我花了三个月调优，最后稳定在 87.5% 的降幅。这个数字是在我的业务场景下跑出来的，不保证每个场景都能复现，但省钱的三个杠杆是通用的。

先说优化前的状态。业务刚开始时，内部所有工具和流程全部使用同一款旗舰模型。我拉了上线后第一个月的日志，做了一个简化统计，为了方便理解，下面用"单位成本"而不是真实金额展示：

• 日均请求约 1 万次；
• 平均每次请求消耗 input 加 output 合计 4000 tokens；
• 旗舰模型按量计价，单次请求平均成本 10 个单位；
• 日均成本等于 10000 乘以 10，也就是 100000 个单位。

4.2 优化后：请求去向的流量拆解

我把一个月的请求日志按任务类型和成本做了聚类，发现浪费比想象中严重：

• 约 45% 的请求是高度重复的问题，来自内部知识库问答和对话机器人，经常是换个说法问同一件事；
• 约 40% 的请求属于简单任务，关键词提取、格式整理、短文本分类，旗舰模型的能力被严重浪费；
• 真正需要旗舰模型深度推理的请求，只占 6% 左右；
• 剩下 9% 属于中等复杂度，中档模型完全能搞定。

于是调整路由策略和缓存策略，优化后一个月的统计结果如下，以 100 次请求为例：

优化前 100 次请求全用旗舰模型，成本是 1000 个单位；优化后是 125 个单位，正好是原来的 12.5%，成本下降 87.5%。

4.3 三个省钱杠杆怎么叠加出 87.5%

第一个杠杆是路由降级。把简单任务交给便宜模型，旗舰模型只处理那 6% 真正需要深度推理的请求。很多团队在这个问题上走极端，要么全用旗舰图省事，要么全换便宜模型导致输出质量崩盘。正确做法是按任务类型分档，凡是能接受 80 分输出、不需要深度推理的任务，一律不用旗舰。轻量模型成本往往是旗舰的二十分之一甚至五十分之一，在简单任务上的输出质量差距并不大。

第二个杠杆是缓存去重。模型调用的天然弱点是同样的问题反复花钱。语义缓存上线后，45% 的重复请求成本直接打到零。但要注意，对时效和一致性有严格要求的场景不要用缓存，比如实时行情问答、订单状态查询这类数据会变化的场景，缓存命中反而会造成信息错误。

第三个杠杆是上下文压缩。同样一个对话任务，输入 token 往往占成本的六成以上。我在发给模型的 messages 上做了三层处理：截断过长的历史消息、把早期历史用摘要替代原文、把系统提示词压到最短。实测在保证输出质量的前提下，输入 token 平均减少了约一半。

注意这三个杠杆的效果是乘法叠加的：路由降级把旗舰占比压到 15%，上下文压缩让非缓存请求的 token 成本再减半，缓存再把四成五的请求清零。最终剩余成本大约是 (1 - 0.55) × 0.5 × 0.55 = 12.4%，四舍五入就是表格里的 12.5%，对应 87.5% 的降幅。这就是为什么"直降 87.5%"不是一个拍脑袋的数字，而是三个杠杆乘出来的结果。

4.4 怎么用历史日志粗算自己的优化空间

如果你想估算自己项目的优化空间，先拉一周历史日志，做三个统计：重复问题占比，用相同输入做相似度聚类就能算出来；简单任务占比，按业务类型标签统计；平均输入 token 中可以压缩掉的比例，抽样人工标注。

然后套这个粗算公式：

code复制优化后成本占比 ≈ 非缓存占比 × 压缩后 token 占比 ×
（轻量任务占比 × 轻量模型单价/旗舰单价
 + 中量任务占比 × 中档模型单价/旗舰单价
 + 困难任务占比 × 1）

有一个必须提醒的细节：公式里的"压缩后 token 占比"要按"输入可压缩、输出不可压缩"的混合口径来算。比如输入输出各占一半，输入压掉一半，整体 token 只降到原来的 75%，不能按 50% 算，否则会高估收益。

5. 上线后踩过的坑与排查实录

5.1 缓存误命中的事故：相似不等于相同

语义缓存上线第一周就收到投诉，有业务方问"这个项目的预算还剩多少"，结果返回了某篇文章的答案，驴唇不对马嘴。

排查后确认是相似度阈值太低。0.92 在"同一问题换说法"的场景下够用，但"预算还剩多少"和"预算怎么算"在向量空间里距离很近，被误判成同一个问题。

解决办法是双管齐下：把相似度阈值提到 0.96，同时加一条硬校验——缓存命中要求最后一句用户消息的字符差异不超过 30%。这两个条件同时满足才返回缓存。阈值到底设多少，要靠历史数据人工抽检，不能拍脑袋。

5.2 路由规则误伤输出质量

最初我把"文案改写"统一路由到轻量模型，结果质检同事反馈复杂长文案的改写质量明显下滑。原因是"文案改写"这个标签太粗，既有 50 字的摘要整理，也有两千字的结构化重写。

后来我把轻量模型的路由限定为"改写长度小于 200 字、不需要保持特定语气"的任务，复杂的全部走中档模型。路由规则的粒度决定了用户体验，宁可多留一些请求给贵模型，也不能让质量翻车。

5.3 对账对不上：计费口径误区

日志里记录的成本和供应商后台的账单总是差 5% 左右，排查了一圈发现两个原因。

第一个是优惠额度处理不一致。日志里按刊例价记账，账单扣的是优惠后的真实金额。第二个是重试导致的对不上，日志里显示 100 次调用，供应商账单显示 105 次——部分请求在超时后，供应商那边已经处理成功了，我们重试造成重复扣费。

这两个问题的最终解法是统一计费口径：以供应商账单为准，日志只做成本估算，同时在日志里记录 request_id 和供应商返回的调用 ID，方便逐笔对齐。

5.4 供应商限流引发的连锁反应

某天晚上，供应商出了大促活动，其他团队疯狂调用，我们的网关大量触发限流。当时重试配置不够精细，网关在一个小时内把所有备用供应商的热量也打满了，整条链路都快了。

后来我加了全局令牌桶加按供应商分桶限流，并在网关入口做了并发上限控制。核心思路是：供应商的容量是有限的，网关必须主动管理流量，而不是等对方限流了才被动反应。

5.5 常见问题速查表

6. 后续扩展与个人体会

6.1 本地模型混合调度：把简单任务成本再打到零

网关第一版跑通之后，我又接入了本地开源模型做混合调度。现在简单任务优先走本地模型，成本几乎为零，数据也不用出内网；复杂度高的任务才动态转发到云端模型。这一步相当于在 87.5% 降本的基础上又压了一截，不过本地模型需要 GPU 资源，适合本来就有推理机器的团队。

6.2 多租户资源池化

中转层天然适合做成"一个入口给多个内部团队共用"的模型资源池。每个租户分配独立的 key、预算和路由策略，管理员可以看到每个租户的消耗排行。我们内部现在就是这么用的，谁家烧了多少一眼便知。这个模式还可以顺便做权限隔离：有的团队只允许用内部模型，有的团队可以访问云端旗舰模型。

6.3 几句实在话

做这种网关，最困难的不是写适配器，也不是调缓存，而是定路由策略。策略定得松，成本压不下来；定得紧，业务方天天抱怨输出质量不行。最好的办法是先把日志和计费做扎实，让每个请求花了多少钱、用了什么模型、结果如何都可回溯，然后再拿数据迭代路由规则。我见过不少团队上来就写路由规则，结果灰度一个月翻车。

如果你也正在被多家模型 API 的差异和账单折磨，从一个小规模网关开始，先把请求日志打全，再上语义缓存和路由，最后动重试和熔断。一个接口聚合多家模型，本质上把"被厂商绑定"变成了"被自己掌控"，这个掌控感建立起来之后，后面做任何模型层面的升级都会从容很多。