如果你最近在做 AI 应用,哪怕只是一个简单的客服机器人,应该也有同感:大模型 API 是越用越谨慎。今天用这家,明天换那家,后天又可能想把开源模型部署到自己的服务器上,光是各家 API 的鉴权方式、超时时间、返回格式就够喝一壶。我今年做了一整套多模型混合调用架构之后,最大的感受是——把单个模型的接口调通不难,难的是让同一套业务逻辑可以随时切换、混合、互为兜底,又不至于把代码改得面目全非。这个项目要解决的核心问题,就是对多个大模型 API 做统一管理:所有业务方只认一个入口,底层到底路由给哪家模型,由网关自己决定。这篇文章我会把整套设计思路、关键代码、还有踩过的坑完整记录下来,适合正在做 Agent 应用、RAG 系统或任何需要接入多家模型的工程团队参考。
1. 为什么需要多模型混合调用架构:单一模型的时代已经过去了
很多人刚开始接大模型 API 时,逻辑非常简单:选一家头部模型,注册 Key,把 Prompt 传进去,拿到结果就完事。我刚做第一个 AI 项目时也是这个思路,觉得“选最强的模型就行了”。但真正上了生产环境,你会发现这条路走不通,而且问题不是出在“模型能力不够”,而是业务对模型的诉求本身就是多元的。
1.1 单模型依赖的三个硬伤
第一个硬伤是可用性不稳定。大模型 API 再稳也是别人的服务,别人限流你就得排队,别人升级你就得跟着变。我们团队曾经遇到过一次上游模型服务连续四个小时返回高延迟,那还是在白天业务流量最大的时候。看起来只是一个外部服务抖动,但对于业务来说,等于整个 AI 功能瘫痪了四个小时,任何技术补救都来不及。如果当时代码里只硬编码了一家模型,除了干等没有任何办法。
第二个硬伤是成本完全不可控。头部模型能力强、价格也高。把它们用在所有场景是很浪费的——简单的内容摘要、关键词抽取、意图分类这些任务,用中等能力的模型完全够用,但如果你只接入了一家高端模型,那每一笔请求都得花冤枉钱。等到月底看到账单时再优化,已经迟了。
第三个硬伤是能力边界太明显。没有哪家模型在所有维度上都碾压对手。有的模型擅长代码生成,有的模型在中文长文本理解上表现更好,有的模型工具调用更规范。如果你的架构只绑定了一个模型,就等于把这些能力差异全部挡在了门外。用户说“这个回答不如某某家的好”,你只能苦笑,因为你根本换不了。
1.2 混合调用到底能带来什么价值
把多个大模型 API 混在一个架构里,表面上看是“多接了几个 API”,实际上是拿到了一层完全不同的能力:
- 故障转移:主模型超时或报错时,自动把请求转到备选模型。用户无感知,系统可用性直接从单一依赖变成了“至少有一家能用”。
- 成本优化:按任务复杂度分配模型。简单任务走便宜的小模型,复杂推理才用旗舰模型。我测试过一个翻译场景,用中等模型完成了 80% 的请求,账单直接降了一半。
- 质量择优:同一类任务可以跑多个模型,再由网关选最优结果或做投票融合。虽然会增加一点延迟,但在对质量要求极高的场景下非常值。
- 谈判筹码:架构上不绑定任何一家,任何模型出问题都可以随时替换,不会被供应商的价格策略拿捏。
- 免费额度利用:这也是很多人忽略的一点。不少模型服务商都提供免费调用额度,适合做个人项目和冷启动阶段。把它们纳入统一路由,配好限速和回退策略,免费模型顶不住的时候自动切到付费模型,一年也能省不少。
所以“统一管理多个大模型 API”并不是为了炫技,而是生产环境对稳定性、成本和能力覆盖共同提出的要求。这套东西做好了,相当于给业务装了一个双向切换的阀门,上游随便怎么变,下游稳如老狗。
2. 架构设计:怎么把“路由”从业务代码里抽出来
多模型混合调用架构最核心的设计决策,不是选什么技术栈,而是把“调用哪个模型”这件事从业务代码里抽出来。业务方不需要知道自己调的是 claude 还是文心,更不需要关心今天是不是切换到了备用模型。它们只需要发一个标准的请求进来,网关帮它们决定一切。
2.1 统一抽象层应该放在哪里
我见过很多失败的例子,团队一开始为了省事,直接在业务代码里写工具函数,比如call_openai()、call_claude(),然后在上层写 if-else 判断。刚开始只有两三个模型时还能忍,等模型数量到了五个以上,你会在每一个需要调模型的接口里看到一团乱麻的 switch 语句。改一个模型参数要全局搜索,加一个新模型要动所有调用方。这时再想抽就晚了。
正确的做法是加一个独立的 API 网关中间层,业务方通过 HTTP 或 SDK 调用网关,网关内部再转发到真正的模型服务端。这个中间层承担四个职责:
- 统一接收业务请求,做参数校验和归一化;
- 按照路由策略选择模型,并执行转发;
- 对下游异常做重试、降级和熔断;
- 记录所有请求日志,为计费和质量分析提供数据。
这个位置选好,后续扩展就是配置层面的事,而不是代码层面的事。
2.2 接口规范:兼容 OpenAI 格式,还是自定义格式?
统一网关要面对的第二个问题,是多家模型的请求格式不一样。OpenAI 用的是messages数组,Claude 用的是system加messages,有些国产模型还要额外传temperature和top_p之外的参数。如果网关内部按照每家模型单独写 adapter,那就不叫统一管理了。
这里我强烈建议:网关对外提供的 API 格式采用 OpenAI 的chat/completions风格。原因很简单——它已经是事实上的行业标准,大部分开源工具和框架原生支持,业务方几乎零学习成本就能接入。内部再用 adapter 把 OpenAI 格式转成各家模型自己的格式。虽然 OpenAl 格式本身有很多被人吐槽的地方,但兼容一个广泛接受的“方言”,远比让所有业务方都学习你的“普通话”要容易落地。
2.3 配置驱动:路由规则不写死在代码里
统一管理最大的优势,就是变更时不需要重新发版。路由规则、可用模型列表、各模型的密钥、超时时间、限流阈值,全部应该放在配置文件里,甚至放在配置中心动态更新。我在项目中用的是 YAML 配置,理由很朴素:YAML 支持注释,团队里非工程师也能读懂并能参与修改。每次调整模型切换逻辑,只需要改配置再 reload,不需要动任何代码。
配置文件里我一般分三块:providers声明可用的模型后端,routers定义路由策略的细节,fallbacks声明失败时的转移顺序。这三块拆开,看起来只是文件结构上的小优化,但在实际维护中非常关键——谁负责哪块一目了然,改配置时不容易误伤其他规则。
3. 核心实现:必会的四个功能模块
架构想清楚了,落地就是写代码。下面我挑四个我认为最重要的模块展开讲。它们不要求你一次性写完,但如果目标是生产可用,这四块缺一不可。
3.1 Provider 适配器:把各家 API 差异关进笼子里
Provider 适配器是整个架构里最基础、也最需要耐心的部分。每一个模型服务商都写一个 adapter,所有 adapter 实现同一个抽象接口。接口定义非常关键,我参考过很多方案,最后落地的版本长这样:
class BaseLLMProvider(ABC): provider_name: str model_name: str @abstractmethod async def chat_completion(self, messages: list[dict], **kwargs) -> dict: """输入统一格式的消息,返回统一格式的结果""" @abstractmethod def count_tokens(self, messages: list[dict]) -> int: """估算输入 token 数,用于路由判断和成本预估"""接口只有两个方法,完全够用。为什么要把count_tokens也放进来?因为不同模型的 token 计算方式不完全一样,有的用 tokenizer 估算,有的直接靠字符数近似。路由决策时需要知道请求的 token 规模,如果这里不做适配,后面做“按成本路由”就会出现偏差。适配器内部做的事情不复杂:把统一消息格式翻译成各家的请求体,调用 SDK 或 HTTP 接口,把各家响应重新包装成统一格式返回。你只需要注意各家返回的结构性差异,比如有的把 content 直接放在message["content"]里,有的还要拼接delta增量才能拿到完整文本。
3.2 路由策略:从简单优先到成本最优
路由是整个网关的大脑。最简单的是“优先级路由”,每个模型配一个权重或等级,默认走第一个,失败再走下一个。适合多数入门场景。复杂一点的是“基于任务类型的路由”,比如给每个路由规则定义task_type匹配条件,摘要任务走摘要专用模型,代码生成走代码能力强的模型。再进阶一点的“成本最优路由”,需要预估请求的复杂度,再结合每个模型的定价来选。
我给大家一个很实用的规则组合:先做优先级路由,再做成本兜底。
- 请求进来先看有没有显式的
model_hint,如果调用方指定了模型,尊重它; - 如果没指定,就用任务类型匹配模型组;
- 模型组内部按优先级排序,选当前可用且没有触发熔断的;
- 如果组内所有模型都不可用,降级到备用模型组。
什么叫“备用模型组”?就是和主模型组能力相近、但可能价格更高或者性能略差的模型。宁可让用户用上一个质量稍差的结果,也不能让请求直接报错。这个“保底优先”的思路帮我在线上避免过很多次事故。
3.3 重试、熔断与超时:没有这层,架构等于没搭
很多自建网关只做转发,不做后端保护,结果上游一抖动,整个链路跟着抖动。多模型架构给你提供了多个后端选择,善用重试和熔断才能把这种优势真正发挥出来。
我在网关里对每个模型单独记录连续失败次数和p99 延迟状态,设置两个阈值:失败次数超过 3 次就触发熔断,熔断后 30 秒内不再路由到这个模型,30 秒后放一个探测请求试探。这个设计参考了微服务里的熔断器模式,但在模型网关里更简单:重构时不需要关闭整个服务,只摘掉那一个有问题的模型实例。还有超时控制,每个模型单独设置连接超时和读取超时,比如指令模型给 30 秒,小模型给 15 秒。返回超时的请求不要直接报错,把重试机会留给备用模型。
这里有个关键细节:重试的请求要保证幂等。因为模型 API 不是幂等的,同一个请求重试两次,可能生成不同的文案。如果是文案生成任务,重试还算能接受;如果是在做带业务副作用的 Agent 执行,重试可能会产生重复的操作。所以我在网关里对重试请求打上idempotency_key标记,一部分服务商支持去重,不支持的在日志里做标记,至少方便事后追溯。
3.4 可观测性:日志、指标、计费拆分一起做
没有可观测性的网关就是黑盒。是哪家模型慢、哪家贵、哪个任务频繁失败,这些信息平时看不出来,一旦出问题全都来问你。我在系统里做了一个极简的可观测性方案:所有请求都写一条结构化日志,包含请求 ID、任务类型、命中的 provider、token 数、耗时、成本估算、错误信息。指标层面用计数器统计每个模型的调用量、错误率和 p99 耗时。成本方面更直接,每个 provider 配置一个单价表,网关估算每次调用的费用,按天汇总成成本报表。
这套东西看起来“又多活了”,但恰恰是它让你敢于做自动路由。没有数据支撑前,你不敢把请求自动切到备用模型,因为怕质量不达标看不出来;有了日志和报表后,每天看一眼趋势,哪家效果好、哪家便宜一目了然,路由策略的调整也有了依据。我建议任何做多模型网关的团队,第一天上线就把日志和指标做好,别等着出事故再补。
4. 实操记录:一个最小可用的多模型网关,600 行代码搞定
理论讲再多,不如直接上一份能跑的代码。下面是我自己维护的一个最小实现,基于 Python 3.10 + FastAPI,核心代码不到 600 行,已经把“统一入口 + 路由 + 故障转移 + 日志”全部走通了。你可以直接抄,也可以按自己的业务拆开改。
4.1 准备工作:依赖、模型账号和配置文件
先准备三样东西:
- FastAPI 和 httpx:一个做 HTTP 服务,一个异步调用各家模型 API;
- pydantic:做请求参数校验;
- 至少两个不同服务商的 API Key,比如一家适合闲聊的、一家适合复杂推理的,这样才能体现路由的价值。
配置文件config.yaml长这样:
providers: - name: provider_a api_base: "https://api.example-a.com/v1" api_key_env: "PROVIDER_A_KEY" models: - name: "model-small" cost_per_1k_input: 0.001 cost_per_1k_output: 0.002 - name: "model-large" cost_per_1k_input: 0.003 cost_per_1k_output: 0.006 - name: provider_b api_base: "https://api.example-b.com/v1" api_key_env: "PROVIDER_B_KEY" models: - name: "model-fast" cost_per_1k_input: 0.0005 cost_per_1k_output: 0.001 router: default_policy: "priority" groups: - task_type: "conversation" models: ["provider_a/model-small", "provider_b/model-fast"] - task_type: "reasoning" models: ["provider_a/model-large"] fallbacks: - from_group: "reasoning" to_group: "conversation"把模型密钥放在环境变量里,配置文件只写环境变量名,这是个安全和运维上的好习惯。cost_per_1k字段是我后加的,用来做成本路由和数据统计。
4.2 核心代码:统一入口与 Provider 适配
FastAPI 的服务入口很简单:
from fastapi import FastAPI, Header from pydantic import BaseModel, Field app = FastAPI() class ChatRequest(BaseModel): task_type: str = "conversation" model_hint: str | None = None messages: list[dict] temperature: float | None = 0.7 max_tokens: int | None = 512 @app.post("/v1/chat/completions") async def chat_completions(req: ChatRequest, authorization: str = Header(...)): api_key = authorization.removeprefix("Bearer ") result = await router.route(req, api_key) return result这里的入参格式刻意模仿了 OpenAI 的/v1/chat/completions,所以业务端甚至可以直接用 OpenAI 的 SDK 来调用我们网关。需要自定义的字段就一个task_type,用于告诉路由这个请求属于什么任务类型。
适配器核心代码就拿一个简化的 HTTP 适配器举例:
class OpenAICompatibleProvider(BaseLLMProvider): def __init__(self, name, api_base, api_key, model_name): self.name = name self.api_base = api_base.rstrip("/") self.api_key = api_key self.model_name = model_name async def chat_completion(self, messages, **kwargs): url = f"{self.api_base}/chat/completions" headers = {"Authorization": f"Bearer {self.api_key}"} payload = { "model": self.model_name, "messages": messages, "temperature": kwargs.get("temperature", 0.7), "max_tokens": kwargs.get("max_tokens", 512) } async with httpx.AsyncClient(timeout=30) as client: resp = await client.post(url, json=payload, headers=headers) resp.raise_for_status() data = resp.json() # 统一为 openai 风格返回 return { "choices": [{ "message": {"role": "assistant", "content": data["choices"][0]["message"]["content"]} }] }说实话,现在绝大多数模型服务商都提供 OpenAI 兼容接口,所以 80% 的适配器可以直接用这个模板。真正需要单独写的,是不兼容 OpenAI 格式的那几家,以及需要带特殊参数的场景,那时只需要重写chat_completion里的 URL 和 payload 组装逻辑就行。
4.3 路由与故障转移:保证请求不因单一模型失败而失败
路由器的核心是配置加载和模型选择。我用一个简单但可靠的方式:启动时把配置里的模型全部实例化存进字典,路由时先走策略选择,选中后直接执行。
class Router: def __init__(self, config: dict, providers: dict): self.config = config self.providers = providers async def route(self, req): if req.model_hint: candidates = [self._find_provider(req.model_hint)] else: candidates = self._match_candidates(req.task_type) for provider in candidates: if provider.is_circuit_open(): continue try: return await provider.chat_completion(req.messages, temperature=req.temperature) except Exception as e: log.warning(f"provider {provider.name} failed: {e}") provider.record_failure() # 全部失败时走降级组 if "fallbacks" in self.config: fb_group = self.config["fallbacks"].get(req.task_type) if fb_group: return await self._try_group(fb_group, req) raise HTTPException(502, "all providers failed")这里的代码体现了一个非常重要的设计:失败不是报错,是转移。只要还有备用模型,前端用户就不会感知到任何异常。“宁慢勿挂”是我调试多模型网关时最深的体会,稳定性优先级永远高于单次请求的极致性能。
4.4 本地验证:用两个 mock 服务把链路跑通
现在没有真实密钥也能把链路测通。我写了一个脚本,本地起两个最简单的 mock server,一个永远正常返回“来自 provider A”,另一个前两次调用的响应故意延迟,再模拟一次超时。验证后的现象是:请求进入网关后,第一次失败,马上自动切到另一个 provider,整体感知只是在响应时间上多了几百毫秒,但用户完全拿到了结果,日志里也没有任何暴露的异常。
这一步验证非常有效,因为生产环境里你不会想在模型故障时再来调试路由代码。本地把重试、超时、熔断都模拟一遍,上了生产就安心多了。
5. 常见问题与排查实录:跨模型调用最容易踩的坑
多模型网关写出来容易,调稳才是最花时间的。这半年里我踩过不少坑,挑几个有代表性的列出来,应该能帮你少走很多弯路。
5.1 上下文长度的“虚标”问题
不同家模型对外宣称的上下文长度是一回事,实际能接受的是另一回事。有的模型按总 token 数限制,有的模型又额外限制输出 token。路由决策时往往只看输入 token 估算,结果转到某个上下文能力弱的模型,一上来就报context length exceeded。
我的解决办法是在配置里给每个模型加一个safe_context_limit,比官方值预留 20% 的安全余量。路由时先判断输入 token + 预期最大输出 token 是否在安全线以内,超出就直接跳过这个模型。这比全靠 error 后再重试高效得多。
5.2 各家 tool calling 的格式差异
做 Agent 项目一定要小心这个。OpenAI 的 tool calling 是functions数组加tool_calls响应,另外几家各有自己的方言。统一消息格式封装好对话没问题,一旦涉及 function call,格式转换就不是简单的字段映射,还可能涉及工具描述 schema 的兼容性调整。我的建议是网关里单独做一个tool_adapter,针对每家模型写工具 schema 的转换逻辑,别把工具格式直接写在业务里。另外,工具调的响应在 token 数计算上也常有出入,成本统计时要做特殊标记。
5.3 输出 JSON 不稳定,比想象中更常见
让模型输出 JSON 是 Agent 项目的基础操作,但换了模型之后你会发现,有的模型用markdown block包裹 JSON,有的会在后面加注释,有的直接凭空在末尾多出一个逗号。统一网关里要有一个修复层:先尝试json.loads,失败了再剥离代码块标记,再不行就尝试只提取{}或[]片段,实在提取失败才报错。
这里加一层“宽松JSON解析”绝对值得,它能把跨模型的成功率明显拉高。我自己实测,某些模型直接解析成功率可能只有 60%,加了修复层能到 95% 以上。
5.4 成本报表不准,问题出在 token 估算
不同模型对 token 的计数口径差异很大。我在做成本统计时,一开始直接用请求字符数除以 4 估算每个模型的 token,结果一个月报表出来后和实际账单对不上。后来改成了基于各模型官方 tokenizer 预先计算 token 数,统计才基本一致。
如果嫌官方 tokenizer 太重,也可以采取“每模型一个修正系数”的近似方案:跑一批样例请求,把网关估算值和服务商账单里的真实 token 数对比,算出每个模型的系数。系数法简单可维护,误差控制在 5% 以内,个人项目妥妥够用。
5.5 免费 API 额度的路由坑
说回到免费大模型 API 这个话题。很多免费模型或者限时免费额度都有自己的“隐性门槛”:每秒只能请求一次、高峰期排队、不支持长文本。把这些免费模型接入混合架构没问题,但你要专门给它们配独立的限流器,并且路由优先级要排在付费模型之后。否则免费额度用满后,业务请求会把免费入口堵死,连备用线都跟着卡。我在本地测试时最喜欢用免费额度,但生产环境里始终留着一条纯付费通道作为最后的保底。免费和付费混着用,省下来的钱不是白来的,是用“多一重配置和管理”换来的。
5.6 别忽略配置热更新的一致性
最后提一个运维层面的坑。配置中心热更新时,如果新配置里把某个 provider 的 key 删了,但旧请求还在路由里引用它,就会引发大量 401。我的做法是最小更新:每次配置变更都生成一个版本号,网关确认新配置完全 loaded 后才切换流量,同时保留旧配置 1 分钟作为安全窗口。这一个小小的机制,避免了好几次在发布时因为误操作引发的线上事故。
最后再分享一个我在实际运行中的体会:多模型混合调用架构的成熟度,不是看接了多少家模型,而是看“即使所有模型同时出问题,业务能不能不崩”。刚开始搞这套东西时,我满脑子都在想“怎么选出最好的模型”,后来被现实教育了才明白,在工程世界里,最重要的不是“最优”,而是“总有一条路能走通”。配置化、可观测、自动降级,每加一层防护,系统离“睡个好觉”就进一步。建议你从最小可用的两个模型开始,先把路由和降级跑通,再慢慢迭代策略,这是最稳妥的路。