多模型接入这件事,最早我只是想省点钱。GPT-4o 处理复杂推理,Claude 写长文更顺手,DeepSeek 做批量分类便宜到离谱,结果三个平台三套 SDK、三种鉴权方式、三份计费账单,代码里到处是 if provider == "openai" 这种分支。后来项目一多,改一个超时参数要动五个文件,加一个新模型得重新写一遍重试逻辑,实在受不了了,才开始认真做网关层。
这篇东西就是那段时间的完整沉淀。我会从为什么要做统一网关讲起,把协议适配、路由策略、流式处理、错误重试、成本核算这几个核心问题一个个拆开,最后给出一套可以直接抄的落地结构。适合正在被多模型接入折磨的后端同学,也适合想给自己 side project 加个模型切换能力但不知道从哪下手的独立开发者。读完你应该能自己搭一个能跑生产的多模型网关,而不是停留在"调通一个 API"的阶段。
1. 为什么多模型接入迟早要收敛到网关层
1.1 直连各家 SDK 的三种典型崩法
刚开始大家都是直连。OpenAI 用openai这个包,Anthropic 用anthropic,DeepSeek 因为兼容 OpenAI 协议,很多人直接复用openai包改个base_url就上了。看起来挺美,但项目一旦超过两三个模型,问题会集中爆发。
第一种崩法是鉴权与配置散落。每个 SDK 初始化方式不同,OpenAI 是OpenAI(api_key=...),Anthropic 是Anthropic(api_key=...),DeepSeek 又是OpenAI(api_key=..., base_url="https://api.deepseek.com")。密钥管理、超时、代理、重试次数这些参数,每个客户端都要单独配一遍。等到要换一个 key 或者调一次超时,你得翻遍整个代码库。
第二种崩法是错误语义不统一。OpenAI 的限流返回 429 带error.type,Anthropic 的限流也是 429 但结构完全不同,DeepSeek 兼容 OpenAI 格式但偶尔会有自己的错误码。上层业务想做一个"限流就退避重试"的逻辑,得为每家写一套判断。更麻烦的是流式响应中途断流,三家抛出的异常类型都不一样,日志里全是五花八门的堆栈。
第三种崩法是能力差异被硬编码。GPT 支持 function calling,Claude 的 tool use 格式不一样,DeepSeek 的 function calling 又是另一套。你想在业务层做"优先用支持工具的模型",结果发现判断逻辑写死在调用点,加一个新模型就要改一遍。
1.2 网关层到底该承担什么职责
很多人对网关的理解停留在"转发请求",这是不够的。一个能打的多模型网关,至少要承担五件事。
协议归一:把各家的请求/响应格式统一成一套内部结构,业务层只认这一套。路由决策:根据模型能力、成本、当前健康状态决定这次请求发给谁。可靠性兜底:超时、重试、降级、熔断都在网关层做掉,业务层不用关心。可观测:每次调用的模型、耗时、token 数、成本、是否命中缓存,都要能打点。成本控制:预算、限流、缓存命中率这些直接影响账单的东西,必须在网关层统一管。
这五件事里,协议归一和路由决策是核心,其他三个是围绕它们展开的。我见过不少团队只做了协议归一就上线,结果路由还是写死的,等于换了个地方写 if-else,没解决根本问题。
1.3 一个反直觉的结论:网关不是越薄越好
网上有种说法,网关要"薄",只做转发,业务逻辑放上层。这个观点在纯 HTTP 代理场景下成立,但在多模型场景下会害死人。
原因是模型调用有几个特性:延迟高(动辄几秒到几十秒)、成本敏感(每次调用都是真金白银)、失败模式复杂(限流、超时、内容审核、模型过载)。如果网关只做转发,这些逻辑就会渗透到每个业务调用点,重复实现、行为不一致。
我的经验是:网关要"厚"在可靠性上,"薄"在业务语义上。重试、降级、熔断、缓存、计费这些通用能力,网关层做厚;但"这次调用是为了生成营销文案还是代码补全"这种业务语义,网关不该知道。这条边界划清楚,后面架构就不会乱。
2. 协议适配层:把三套 API 揉成一套内部结构
2.1 统一请求体的字段设计
协议适配的第一步是定义一套内部请求结构。这套结构要能表达所有模型的能力,同时不引入任何一家的私有概念。我用的结构大致是这样:
class UnifiedRequest: model: str # 逻辑模型名,如 "fast" / "smart" / "cheap" messages: list[Message] # 统一的消息格式 tools: list[Tool] | None temperature: float max_tokens: int stream: bool metadata: dict # 业务透传,网关不解析关键点是model字段存的是逻辑模型名而不是真实模型名。业务层说"我要一个便宜的模型",网关根据当前配置决定映射到deepseek-chat还是gpt-4o-mini。这样换模型不用改业务代码,改网关配置就行。
messages的格式我统一成 OpenAI 那套role/content结构,因为它是事实标准,DeepSeek 直接兼容,Anthropic 转换起来也不复杂。tools字段用 OpenAI 的 function 格式作为内部标准,转换到 Anthropic 的 tool use 时做一层映射。
2.2 三家请求格式的转换细节
OpenAI 和 DeepSeek 基本可以直接透传,因为 DeepSeek 就是按 OpenAI 协议设计的。真正需要认真处理的是 Anthropic。
Anthropic 的messages里system是独立参数,不在 messages 数组里。转换时要把role == "system"的消息抽出来拼成system字段。另外 Anthropic 要求messages必须 user/assistant 交替,连续两条 user 消息会报错,需要在转换时合并。
工具调用格式差异更大。OpenAI 是:
{"type": "function", "function": {"name": "...", "parameters": {...}}}Anthropic 是:
{"name": "...", "input_schema": {...}}转换函数要处理parameters到input_schema的映射,同时把tool_choice的语义对齐。Anthropic 的tool_choice有auto、any、tool三种,OpenAI 有auto、none、required和指定函数,映射关系要写清楚,否则会出现"明明指定了工具却没用"的问题。
2.3 响应归一与流式事件的统一
非流式响应相对简单,把各家的choices[0].message或content[0].text统一成内部UnifiedResponse就行。真正麻烦的是流式。
OpenAI 的流式是 SSE,每个 chunk 是data: {...},最后以data: [DONE]结束。Anthropic 的流式事件类型多得多,有message_start、content_block_start、content_block_delta、message_delta、message_stop等。DeepSeek 兼容 OpenAI 格式。
我的做法是定义一个内部事件类型,把三家的流式事件都映射过来:
| 内部事件 | OpenAI 对应 | Anthropic 对应 |
|---|---|---|
text_delta | choices[0].delta.content | content_block_delta(text_delta) |
tool_call_start | delta.tool_calls首帧 | content_block_start(tool_use) |
tool_call_delta | delta.tool_calls增量 | content_block_delta(input_json_delta) |
usage | 最后一帧的usage | message_delta的usage |
done | [DONE] | message_stop |
这样业务层只需要处理五种内部事件,不用管底层是哪家。实测下来,这套映射覆盖了 95% 以上的场景,剩下的是各家特有的字段,比如 Anthropic 的stop_reason和 OpenAI 的finish_reason,做个枚举映射即可。
注意:Anthropic 的流式里
usage是在message_delta事件里返回的,而且 input tokens 在message_start里就有,output tokens 要等message_delta。如果你要做实时计费,得把这两个事件都接住,别只接一个。
3. 路由策略:让请求自己找到合适的模型
3.1 逻辑模型到物理模型的映射表
路由的核心是一张映射表。业务层传model="smart",网关查表决定用哪个物理模型。这张表我建议放在配置中心或者数据库里,支持热更新,不要写死在代码里。
一个典型的映射配置长这样:
routes: smart: primary: claude-sonnet-4 fallback: [gpt-4o, deepseek-chat] constraints: max_cost_per_1k: 0.02 require_tools: true fast: primary: gpt-4o-mini fallback: [deepseek-chat] cheap: primary: deepseek-chat fallback: [gpt-4o-mini]primary是首选,fallback是降级链。constraints是约束条件,比如require_tools表示这个逻辑模型必须支持工具调用,路由时会把不支持工具的模型过滤掉。
这张表的好处是,业务层完全不用关心底层是什么模型。产品说"最近 Claude 太贵了,换成 GPT",你改一行配置就行,不用发版。
3.2 基于成本、延迟、健康度的动态选择
静态映射表能解决 80% 的问题,但有些场景需要动态决策。比如同一个逻辑模型有多个候选,你想选当前延迟最低的,或者成本最低的。
我实现过一个简单的打分路由:每个候选模型有一个分数,分数由三部分组成——健康度(最近失败率)、延迟(P95 响应时间)、成本(每千 token 价格)。权重可配,默认健康度 0.5、延迟 0.3、成本 0.2。每次请求选分数最高的。
健康度用滑动窗口统计,最近 100 次调用的失败率。失败包括超时、5xx、限流。延迟用 P95 而不是平均值,因为平均值会被大量快速请求拉低,掩盖长尾问题。成本直接从配置读。
这套打分逻辑不复杂,但效果很明显。有一次某个模型区域节点抖动,失败率飙升,打分路由在几十秒内就把流量切到了备用模型,业务层完全无感。
3.3 降级链与熔断的触发条件
降级链的执行逻辑是:主模型失败 → 尝试第一个 fallback → 再失败 → 尝试第二个。但"失败"的定义要明确,不是所有错误都该降级。
我的分类是:可重试错误(超时、429、5xx)走重试,重试耗尽后降级;不可重试错误(400 参数错误、401 鉴权失败、内容审核拒绝)直接返回,不降级。因为参数错误换哪个模型都一样错,降级只是浪费一次调用。
熔断用经典的断路器模式:连续失败 N 次(默认 5 次)后打开断路器,后续请求直接走 fallback,不再尝试主模型。冷却时间(默认 30 秒)后进入半开状态,放一个请求试探,成功则关闭断路器,失败则继续打开。
这里有个坑:熔断的粒度要按"模型+区域"而不是按"模型"。因为同一个模型在不同区域节点的健康度可能完全不同,按模型熔断会把健康的节点也一起熔断掉。
4. 流式响应与错误重试的工程细节
4.1 流式场景下重试为什么这么难
非流式请求重试很简单,失败了重新发一次就行。流式请求麻烦在于:响应已经开始返回了,你才发现出错。
比如模型返回了前 100 个 token,然后连接断了。这时候你不能简单重试,因为前 100 个 token 已经发给用户了,重试会导致内容重复。但如果不重试,用户就拿到半截内容。
我的处理策略分三种情况:
连接建立阶段失败(还没收到任何数据):直接重试,用户无感。流式传输中途失败:如果已经输出的内容很少(比如少于 50 token),可以重试并丢弃已输出内容,重新开始;如果已经输出很多,重试代价太大,直接返回错误,让业务层决定是否重新发起。正常结束但内容被截断(finish_reason == "length"):这不是错误,是 max_tokens 设小了,返回给业务层处理。
判断"已输出内容多少"需要一个计数器,在流式转发时累加。这个计数器同时用于计费,一举两得。
4.2 指数退避与抖动参数的取值
重试的退避策略我用的是指数退避加抖动。基础公式是delay = base * 2^attempt,base取 0.5 秒,最多重试 3 次,所以延迟是 0.5s、1s、2s。
抖动(jitter)很重要,不加抖动的话,大量请求会在同一时刻重试,形成"重试风暴",把刚恢复的服务再次打垮。抖动我用的是"全抖动"策略,即delay = random(0, base * 2^attempt),这样重试时间完全随机化,分散效果最好。
有个细节:429 限流错误要读响应头里的Retry-After。如果服务端明确告诉你要等 10 秒,你就等 10 秒,别用自己算的退避时间。很多限流场景下,服务端的建议时间比客户端算的更准。
4.3 超时设置:连接、首字节、总时长要分开
超时不能只设一个。我分三层:
连接超时:建立 TCP 连接的时间,默认 5 秒。首字节超时(TTFB):从发请求到收到第一个字节的时间,默认 30 秒。这个对模型调用特别重要,因为模型推理本身就要时间,但超过 30 秒还没开始返回,基本可以判定卡住了。总时长超时:整个请求从开始到结束,默认 120 秒。流式请求要设长一点,因为长文本生成可能要好几分钟。
这三层超时分别对应不同的失败模式。连接超时说明网络有问题,首字节超时说明模型侧卡住,总时长超时说明生成内容太长或者模型在打转。日志里把这三类分开统计,排查问题时能快速定位。
提示:流式请求的总时长超时要设得比非流式长很多。我见过有人统一设 60 秒,结果长文生成任务全部超时,还以为是模型问题,其实是超时设短了。
5. 成本核算与可观测性:让每一分钱都有迹可循
5.1 token 计费的三种口径差异
计费最容易被忽略的是口径差异。OpenAI 的 token 计数用 tiktoken,Anthropic 用自己的 tokenizer,DeepSeek 又是另一套。同一个中文句子,三家算出来的 token 数可能差 20% 以上。
这意味着你不能用一家的 token 数去估算另一家的成本。我的做法是:计费时用各家返回的 usage 字段,不要自己算。三家在响应里都会返回prompt_tokens和completion_tokens,直接读就行。只有在做预算预估(请求发出前)时,才用估算,而且要按模型分别用对应的 tokenizer。
还有个坑:流式响应的 usage 可能不在最后一帧。OpenAI 需要显式传stream_options: {"include_usage": true}才会在流里返回 usage,否则最后一帧没有 usage。Anthropic 的 usage 分散在message_start和message_delta两个事件里。这些细节不处理,计费就会漏。
5.2 缓存命中率对账单的实际影响
缓存是省钱的大头。同样的 prompt 如果命中缓存,成本能降 90% 以上。缓存分两种:精确缓存(prompt 完全一致)和语义缓存(prompt 语义相似)。
精确缓存用 prompt 的 hash 做 key,简单可靠,命中率取决于业务场景。如果是客服问答这种重复率高的场景,命中率能到 30% 以上。语义缓存用 embedding 做相似度匹配,命中率更高但实现复杂,而且有误命中风险(语义相似但答案不同)。
我的建议是先做精确缓存,把temperature=0的请求都缓存起来。这类请求本身就是要确定性输出,缓存完全合理。语义缓存等业务稳定了再考虑,别一上来就搞复杂的。
缓存还有个细节:要按模型分别缓存。同一个 prompt 发给 GPT 和 Claude,结果不一样,不能共用缓存。key 里要带上模型标识。
5.3 关键监控指标与告警阈值
网关要暴露的指标不多,但每个都要有用。我监控这几个:
| 指标 | 含义 | 告警阈值 |
|---|---|---|
request_duration_p95 | P95 响应时间 | 超过 30s |
error_rate | 错误率 | 超过 5% |
fallback_rate | 降级率 | 超过 10% |
cache_hit_rate | 缓存命中率 | 低于 20% |
cost_per_hour | 每小时成本 | 超过预算 1.5 倍 |
token_usage_ratio | 实际 token / 预估 token | 超过 1.3 |
fallback_rate是我最看重的指标。它升高说明主模型有问题,但业务还没完全失败,是"亚健康"状态。这个指标比error_rate更早发现问题,因为降级成功了,错误率不会升高,但成本和质量可能已经受影响。
token_usage_ratio用来发现 prompt 膨胀。如果实际 token 比预估高很多,说明有请求带了超长上下文,可能是 bug 导致的。
6. 一套可直接落地的网关结构
6.1 目录结构与模块划分
我把网关拆成这几个模块,每个模块职责单一:
gateway/ adapters/ # 协议适配 openai.py anthropic.py deepseek.py router/ # 路由决策 static.py # 静态映射 dynamic.py # 动态打分 reliability/ # 可靠性 retry.py circuit.py timeout.py observability/ # 可观测 metrics.py cost.py core/ request.py # UnifiedRequest response.py # UnifiedResponse stream.py # 流式事件adapters里每个文件实现两个方法:to_provider(request)和from_provider(response)。新增一个模型厂商,只要加一个 adapter 文件,其他模块不用动。
router里静态和动态分开,静态是配置驱动的映射,动态是打分逻辑。两者可以组合使用:先用静态表筛出候选,再用动态打分选一个。
6.2 配置热更新的实现方式
配置热更新我用的方案是:配置存数据库,网关启动时加载到内存,同时起一个后台线程每 10 秒轮询一次版本号,版本变了就重新加载。这样改配置不用重启服务。
轮询间隔 10 秒是个折中。太短会增加数据库压力,太长会导致配置生效延迟。如果对实时性要求高,可以用消息队列推送,但大多数场景 10 秒足够了。
加载配置时要做原子替换,不能边加载边用。我的做法是加载到新的配置对象,加载成功后用一次赋值替换旧对象。Python 里赋值是原子的,不会有中间状态。
6.3 从零到跑通的最小验证路径
如果你想快速验证这套结构,我建议按这个顺序来:
- 先实现 OpenAI 和 DeepSeek 两个 adapter,因为协议兼容,工作量小。
- 加一个静态路由,配置两个逻辑模型
fast和smart。 - 实现非流式调用,跑通端到端。
- 加流式支持,重点测 Anthropic 的事件映射。
- 加重试和超时,用 mock 服务模拟失败。
- 最后加计费和监控。
这个顺序的好处是每一步都能独立验证,不会一次性引入太多复杂度。我见过有人一上来就搞动态路由和语义缓存,结果基础链路都没跑通,debug 起来极其痛苦。
注意:验证流式的时候,一定要测"中途断流"的场景。用一个 mock 服务,返回几个 chunk 后主动断开连接,看网关是否正确处理。这个场景在真实环境里不常见但一旦出现就很致命,提前测过心里有底。
7. 踩过的坑与几条硬经验
7.1 那些文档里不会写的适配陷阱
Anthropic 的max_tokens是必填的,OpenAI 是可选的。如果你内部结构里max_tokens允许为空,转发到 Anthropic 时会直接报错。我的处理是给一个默认值 4096,同时在配置里允许按模型覆盖。
DeepSeek 的base_url要带/v1。很多人直接填https://api.deepseek.com,结果 404。正确是https://api.deepseek.com/v1。这个坑我踩过,排查了半小时才发现是路径问题。
OpenAI 的流式 usage 默认不返回。前面提过,要显式传stream_options。这个参数在旧版本 SDK 里可能不支持,要确认 SDK 版本。
Anthropic 的 tool use 结果要作为 user 消息回传,格式是{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "...", "content": "..."}]}。这个格式和 OpenAI 的{"role": "tool", "tool_call_id": "..."}完全不同,转换时容易搞混。
7.2 并发与连接池的调优心得
模型调用是 IO 密集型,连接池大小很关键。我的经验值是:连接池大小 = 预期并发数 × 1.2。比如预期峰值 100 并发,连接池设 120。
但要注意,不同厂商的连接池要分开。因为 OpenAI 的连接池满了不应该影响 Anthropic 的调用。每个 adapter 维护自己的连接池。
还有个细节:流式请求的连接要单独管理。因为流式连接持有时间长,如果和非流式共用连接池,长连接会把池子占满,导致非流式请求拿不到连接。我的做法是流式和非流式用两个独立的池。
7.3 什么时候该放弃自建网关
自建网关不是万能的。如果你的场景满足以下任意一条,我建议直接用现成的方案或者干脆直连:
只用一个模型:那就没必要做网关,直连最省事。调用量极小(每天几百次):网关的维护成本可能超过它省下的钱。团队没有后端运维能力:网关是要长期维护的,出问题要能快速定位,没这个能力会很痛苦。
自建网关的价值在"多模型 + 中大规模 + 有成本优化诉求"这个组合下才体现得明显。我见过一些小团队为了"架构优雅"硬上网关,结果维护成本比省下的钱还多,得不偿失。
7.4 关于模型切换的一个真实教训
最后分享一个教训。有次我们为了省钱,把一个逻辑模型的 primary 从 GPT-4o 换成了 DeepSeek。配置改完,成本确实降了 80%,但一周后客服反馈"回答质量下降"。
排查发现,DeepSeek 在处理某些特定类型的指令时,格式遵循度不如 GPT-4o,导致下游解析失败。问题不在于 DeepSeek 不好,而在于我们切换模型时只看了成本和延迟,没做质量回归。
后来我加了一条规则:任何 primary 模型的变更,必须先在影子流量上跑一周,对比输出质量指标。质量指标可以是格式遵循率、工具调用成功率、人工抽检评分。只有质量指标不下降,才允许正式切换。
这条规则看起来麻烦,但避免了好几次类似的翻车。模型切换不是改个配置那么简单,它本质上是一次 A/B 测试,得用数据说话。
这套网关我们跑了快一年,中间加过新模型、换过路由策略、调过重试参数,整体还算稳。最大的体会是:网关的价值不在于"统一",而在于"可替换"。统一只是手段,真正让你睡得着觉的,是任何一个模型出问题或者涨价时,你能在几分钟内切走,而业务代码一行不用改。这个能力,值得花时间搭。