1. 多模型接入的现实困境与网关思路
1.1 为什么单模型直连越来越不够用
过去两年，我陆续把手上几个项目从"只调一家模型"改成了"多模型混用"。原因很朴素：不同任务对模型的要求差异太大。写代码补全，某些模型在长上下文里更稳；做中文长文摘要，另一些模型对语感把握更好；批量做结构化抽取，价格便宜的模型完全够用，没必要上最贵的那档。再加上各家时不时出现的限流、区域可用性波动、版本迭代，单点直连的脆弱性会被无限放大。
最开始的土办法是"哪里需要就在哪里写一段请求代码"。结果就是项目里散落着七八处 API 调用，每家的鉴权方式、请求体结构、返回格式、错误码都不一样。改一个超时参数要翻五个文件，加一个新模型要复制粘贴一大坨。这种状态撑不过三个月就会失控。
于是"网关"这个概念就自然浮现了。所谓多模型 API 网关，本质是在你的业务代码和各家模型服务之间插一层统一代理：业务侧只认一套接口规范，网关负责把请求翻译成各家能听懂的样子，再把返回结果翻译回统一格式。它解决的不是"能不能调通"，而是"能不能长期、低成本、可维护地调"。
1.2 网关到底该承担哪些职责
很多人一上来就把网关想得很重，恨不得做成一个平台。我的经验是，先把职责边界划清楚，再决定实现复杂度。一个务实的多模型网关，核心职责其实就四件事：
- 协议归一：把 OpenAI 风格的
/v1/chat/completions作为内部标准，其他模型（Claude、DeepSeek 等）通过适配器转换请求与响应。 - 鉴权与密钥管理：业务侧只拿网关签发的内部 key，真实的上游密钥集中在网关侧，避免泄露和轮换困难。
- 路由与降级：根据模型名、任务类型、成本预算选择上游；某个上游失败时自动切到备用。
- 可观测：记录每次调用的模型、耗时、token 用量、错误类型，为成本核算和问题排查提供依据。
这四件事里，协议归一和路由是刚需，可观测是长期价值最高的，密钥管理则是安全底线。至于限流、缓存、内容审核这些，属于"有了更好"，可以后置。
1.3 为什么选 OpenAI 风格作为内部标准
这里有个关键决策：内部统一接口用谁的风格？我试过自定义一套"最干净"的协议，也试过直接对齐 OpenAI 格式，最后选了后者。理由很实际：
第一，生态惯性。市面上绝大多数 SDK、客户端库、开源工具默认就支持 OpenAI 格式，你只要让网关兼容它，这些工具几乎零改动就能接进来。第二，文档成本低。团队成员大多熟悉这套字段（model、messages、temperature、stream），培训成本几乎为零。第三，适配层好写。Claude 的 Messages API 和 DeepSeek 的接口都能较自然地映射到这套结构上，转换逻辑不复杂。
提示：把 OpenAI 格式当"内部普通话"，不代表要绑定某一家。它只是一套字段约定，网关背后接谁完全由你决定。
2. 核心架构拆解与关键设计取舍
2.1 整体分层：接入层、适配层、路由层
我最终落地的架构分三层，从外到内依次是接入层、路由层、适配层。
接入层负责对外暴露统一的 HTTP 接口，处理内部 key 校验、请求体解析、流式响应的透传。这一层要尽量薄，不做业务逻辑，只做"收进来、发出去"。
路由层是大脑，决定这次请求发给谁。它的输入是请求里的model字段加上一些元信息（比如任务标签、租户 ID），输出是一个具体上游的配置。路由策略可以很简单——按模型名映射；也可以很复杂——按成本、延迟、健康度动态打分。
适配层是手脚，每个上游一个适配器。适配器干两件事：把统一请求转成上游格式，把上游响应转回统一格式。流式场景下还要处理 SSE 事件的逐块转换。
这种分层的最大好处是变更隔离。某家模型改了接口，只需要动它对应的适配器；想加新模型，写个新适配器注册进去就行，路由层和接入层基本不用碰。
2.2 适配器模式：把差异关进笼子
适配器是整个网关里最需要耐心的部分。不同模型的差异主要体现在几个地方，我逐个说。
请求体结构差异。OpenAI 用messages数组，每条消息有role和content；Claude 的 Messages API 也类似，但系统提示是独立的system字段而不是塞在 messages 里，且max_tokens是必填。DeepSeek 基本兼容 OpenAI 格式，差异较小。适配器要做的就是把这些字段做双向映射。
响应结构差异。OpenAI 的返回里choices[0].message.content是主文本，usage里有prompt_tokens、completion_tokens；Claude 返回的是content数组，可能包含多个 block，用量字段叫input_tokens、output_tokens。适配器要把它们统一成 OpenAI 风格，业务侧才不用关心底层是谁。
流式协议差异。这是最容易踩坑的地方。OpenAI 的流式是data: {...}的 SSE，最后以data: [DONE]结束；Claude 的流式事件类型更多（message_start、content_block_delta、message_stop等），需要把增量文本从delta.text里抠出来，再包装成 OpenAI 的 chunk 格式。如果这里处理不干净，前端就会出现"文字重复"或"卡住不结束"的现象。
下面是一个适配器接口的简化示意，用 Python 写：
class BaseAdapter: def build_request(self, unified_req: dict) -> dict: """把统一请求转成上游请求体""" raise NotImplementedError def parse_response(self, raw: dict) -> dict: """把上游响应转回统一格式""" raise NotImplementedError def parse_stream_chunk(self, raw_line: str) -> dict | None: """把上游流式片段转成统一 chunk""" raise NotImplementedError每个上游继承这个基类，实现三个方法。路由层只认BaseAdapter，不认具体实现，这就是"把差异关进笼子"。
2.3 路由策略：从静态映射到动态打分
路由策略我经历了三个阶段，可以给不同规模的团队参考。
阶段一：静态映射。维护一张表，model字段直接对应上游配置。比如gpt-4o走 A 上游，claude-3-5-sonnet走 B 上游，deepseek-chat走 C 上游。简单直接，适合模型数量少、流量稳定的场景。
阶段二：别名 + 权重。引入逻辑模型名，比如业务侧统一写fast和smart，网关内部把fast映射到几个便宜模型并按权重分流，smart映射到几个强模型。这样业务侧不用关心具体版本，切换模型只改网关配置。
阶段三：动态打分。给每个上游维护健康度、近期延迟、错误率、剩余配额等指标，请求进来时实时算一个分数，选最优的。这一阶段复杂度陡增，除非流量很大或对成本极度敏感，否则不必急着上。
我的建议是：从阶段一直接跳到阶段二，阶段三按需。阶段二的别名机制性价比最高，既解耦了业务和具体模型，又不用维护复杂的打分逻辑。
2.4 密钥与配额：安全底线不能省
上游密钥绝对不能下发到业务侧或前端。我见过有团队图省事，把上游 key 直接写进客户端，结果 key 泄露被刷爆。正确做法是网关持有真实密钥，业务侧只拿内部签发的 key，网关校验内部 key 后再用真实密钥请求上游。
内部 key 的管理也有讲究。至少要支持按租户或按项目签发，每个 key 绑定配额（比如每天多少 token、每分钟多少请求）。这样即使某个 key 泄露，损失也可控。配额统计可以放在网关内存里做近似限流，精确统计则落到数据库或 Redis。
注意：密钥轮换要设计成"不停机"的。上游密钥更新时，网关应该能热加载新密钥，旧密钥保留一个过渡期，避免正在进行的请求失败。
3. 实操落地：从零搭一个可用的网关
3.1 技术选型与目录结构
技术栈上，我选了自己最顺手的组合：Python + FastAPI 做接入层，httpx做异步上游请求，Redis 做配额和健康度存储。选 FastAPI 是因为它原生支持异步和 SSE 流式响应，写起来干净；httpx的异步客户端对并发请求友好，比requests更适合网关这种 IO 密集场景。
目录结构大致这样组织：
gateway/ main.py # 接入层，路由注册 router.py # 路由层，模型选择逻辑 adapters/ base.py # 适配器基类 openai.py # OpenAI 及兼容上游 claude.py # Claude 适配器 deepseek.py # DeepSeek 适配器 config/ models.yaml # 模型映射与上游配置 utils/ quota.py # 配额校验 metrics.py # 指标记录配置和代码分离很重要。models.yaml里描述所有上游和映射关系，改配置不用改代码，重启或热加载即可生效。
3.2 统一请求与响应格式定义
统一请求我基本照搬 OpenAI 的字段，只做少量扩展。核心字段包括model、messages、temperature、max_tokens、stream，另外加一个可选的task_tag用于路由打标。
统一响应分两种。非流式返回一个标准对象：
{ "id": "chatcmpl-xxx", "object": "chat.completion", "model": "smart", "choices": [ {"index": 0, "message": {"role": "assistant", "content": "..."}, "finish_reason": "stop"} ], "usage": {"prompt_tokens": 120, "completion_tokens": 80, "total_tokens": 200} }流式则返回一系列 chunk，每个 chunk 的choices[0].delta.content是增量文本，最后以data: [DONE]收尾。业务侧无论底层接的是谁，看到的都是这套结构。
3.3 Claude 适配器的关键转换细节
Claude 的适配是几个里最需要小心的。请求侧，要把统一请求里的 system 消息抽出来放到顶层system字段，messages里只保留 user 和 assistant 的轮次。max_tokens必须给值，如果业务侧没传，适配器要给个合理默认（比如 4096），否则上游直接报错。
响应侧，Claude 返回的content是个数组，可能包含text类型的 block。适配器要把所有 text block 拼起来作为最终内容。用量字段input_tokens、output_tokens映射到统一的prompt_tokens、completion_tokens。
流式是最麻烦的。Claude 的事件流里，文本增量在content_block_delta事件的delta.text里。适配器要监听这个事件，把文本包装成 OpenAI 风格的 chunk 推给业务侧。同时要处理message_stop事件，在它到来时发送[DONE]。我踩过的坑是：早期没处理ping事件，导致某些客户端解析异常，后来加了事件类型过滤才稳定。
3.4 DeepSeek 适配器：兼容但不完全等同
DeepSeek 的接口和 OpenAI 高度兼容，适配器可以复用大部分逻辑，但不能直接照搬。差异点主要在模型名和部分参数支持上。比如某些参数在 DeepSeek 上不支持或行为不同，适配器要做过滤或转换。
我的做法是让 DeepSeek 适配器继承 OpenAI 适配器，只覆写有差异的方法。这样代码复用率高，维护成本低。模型名映射放在配置里，比如业务侧的fast映射到deepseek-chat，smart映射到更强的版本。
3.5 流式透传的实现要点
流式透传是网关里最容易出 bug 的地方，我单独拎出来说。核心原则是：边收边转边发，不要攒完再发。如果用httpx的流式接口，可以逐行读取上游的 SSE，每读到一行就交给适配器转换，转换结果立即通过StreamingResponse推给客户端。
几个必须注意的点：
- 缓冲区处理：SSE 的一行可能被 TCP 分包，不能假设一次
read就是完整一行。要用缓冲累积，遇到换行符才处理。 - 异常中断：上游中途断开时，要确保客户端能收到一个明确的结束信号，而不是一直挂着。
- 背压：如果客户端消费慢，要有机制避免网关内存被撑爆，异步生成器天然有背压优势。
async def stream_proxy(adapter, unified_req): async with httpx.AsyncClient(timeout=None) as client: async with client.stream("POST", adapter.url, json=adapter.build_request(unified_req)) as resp: async for line in resp.aiter_lines(): chunk = adapter.parse_stream_chunk(line) if chunk: yield f"data: {json.dumps(chunk)}\n\n" yield "data: [DONE]\n\n"这段代码看着简单，但aiter_lines已经帮你处理了分包问题，实际生产里还要加上错误捕获和日志。
3.6 配置驱动的模型映射
models.yaml是整个网关的"通讯录"，我一般这么写：
logical_models: fast: primary: deepseek-chat fallback: gpt-4o-mini smart: primary: claude-3-5-sonnet fallback: gpt-4o upstreams: deepseek-chat: adapter: deepseek base_url: https://api.deepseek.com api_key_env: DEEPSEEK_KEY claude-3-5-sonnet: adapter: claude base_url: https://api.anthropic.com api_key_env: CLAUDE_KEY gpt-4o: adapter: openai base_url: https://api.openai.com api_key_env: OPENAI_KEY业务侧只写model: fast或model: smart，网关查表决定实际走谁。密钥从环境变量读，不落配置文件。要加新模型，加一段配置加一个适配器即可。
4. 常见问题与排查技巧实录
4.1 流式响应中断与重复
这是最高频的问题。表现是前端文字打到一半停了，或者同一段文字出现两遍。排查思路：
先看网关日志里上游是否正常返回了结束事件。如果上游正常但客户端异常，多半是适配器转换时漏了结束信号，或者[DONE]发早了。重复问题通常是适配器把同一个增量事件处理了两次，检查事件类型判断逻辑。
我整理了一张速查表：
| 现象 | 可能原因 | 排查方向 |
|---|---|---|
| 文字打一半停住 | 结束信号未透传 | 检查message_stop处理 |
| 文字重复 | 增量事件重复处理 | 检查事件类型过滤 |
| 客户端解析报错 | 非标准 SSE 行 | 检查是否混入 ping 等事件 |
| 长时间无响应 | 上游超时未设 | 检查 httpx timeout 配置 |
4.2 参数不兼容导致的报错
不同模型对参数的支持程度不一样。比如某些模型不接受temperature的极端值，某些模型对max_tokens有上限。适配器要做参数校验和裁剪，把不支持的参数过滤掉，把超限的值截断到合法范围。我一般会在适配器里维护一张"参数支持表"，请求进来先过一遍。
提示：不要假设所有模型都支持全部参数。宁可适配器多做一层校验，也不要让上游报错直接透传给业务侧。
4.3 配额统计不准
配额统计不准通常有两个原因：一是流式场景下用量在最后一个 chunk 才返回，如果中途断开就统计不到；二是并发请求下计数有竞态。解决办法是流式场景在结束时补一次用量记录，并发计数用 Redis 的原子操作。
4.4 上游限流与降级
上游限流是常态，尤其是便宜模型。网关要能识别限流错误码（通常是 429），触发降级逻辑切到备用上游。降级要设阈值，比如连续失败三次才切，避免偶发错误导致频繁切换。切换后要有个恢复探测机制，定期试探主上游是否恢复。
4.5 日志与可观测的取舍
日志不能什么都记，也不能什么都不记。我的做法是：每次调用记录一条结构化日志，包含请求 ID、逻辑模型、实际上游、耗时、token 用量、状态码。请求和响应的完整内容只在调试模式下记录，生产环境默认不记，避免存储爆炸和隐私风险。
5. 成本控制与性能优化的实战经验
5.1 用别名机制做成本分层
成本控制最有效的手段不是砍功能，而是分层。把任务按重要性分成几档，每档对应一个逻辑模型别名。比如批量摘要、分类这种任务走fast，复杂推理走smart。实测下来，光这一招就能把整体成本压下来一大截，因为大量简单任务根本不需要强模型。
5.2 缓存重复请求
很多场景下请求是重复的，比如同一段文本被多次摘要。在网关层加一层基于请求内容哈希的缓存，命中就直接返回，既省钱又快。缓存要注意设置合理的过期时间，以及区分不同模型的结果（同一个请求走不同模型结果不同，缓存 key 要带上逻辑模型名）。
5.3 并发与连接复用
网关是 IO 密集型服务，连接复用能显著降低延迟。httpx.AsyncClient要复用而不是每次请求新建，连接池大小根据上游并发限制调整。我一般把连接池上限设成上游允许并发数的 80% 左右，留点余量。
5.4 超时与重试的平衡
超时设太短会误杀正常请求，设太长会拖垮网关。我的经验是：连接超时 5 秒，读取超时按任务类型区分，普通对话 60 秒，长文生成 180 秒。重试只对幂等且明确可重试的错误（如 429、502）做，且最多重试一次，避免放大上游压力。
6. 后续可扩展的方向
网关跑稳之后，能扩展的方向不少。比如加一层语义缓存，用向量相似度判断请求是否等价；比如接入更多上游，把本地部署的模型也纳入统一路由；比如做 A/B 测试，让同一逻辑模型按比例分流到不同上游，对比效果和成本。
我个人最想加的是"按任务自动选模型"——业务侧连别名都不写，只描述任务，网关根据历史数据自动选性价比最高的上游。不过这需要积累足够的调用数据才能做准，属于锦上添花。
最后分享一个小技巧：网关上线初期，一定要开一个"影子模式"，把请求同时发给主上游和一个候选上游，对比两者结果差异，但不影响业务返回。这样能在切换模型前拿到真实数据，避免拍脑袋决策。我在切换主力模型时用过这招，发现候选模型在某些中文场景下确实更稳，果断调整了路由权重。
与多重继承实战指南)