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和contentClaude 的 Messages API 也类似但系统提示是独立的system字段而不是塞在 messages 里且max_tokens是必填。DeepSeek 基本兼容 OpenAI 格式差异较小。适配器要做的就是把这些字段做双向映射。响应结构差异。OpenAI 的返回里choices[0].message.content是主文本usage里有prompt_tokens、completion_tokensClaude 返回的是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-chatsmart映射到更强的版本。3.5 流式透传的实现要点流式透传是网关里最容易出 bug 的地方我单独拎出来说。核心原则是边收边转边发不要攒完再发。如果用httpx的流式接口可以逐行读取上游的 SSE每读到一行就交给适配器转换转换结果立即通过StreamingResponse推给客户端。几个必须注意的点缓冲区处理SSE 的一行可能被 TCP 分包不能假设一次read就是完整一行。要用缓冲累积遇到换行符才处理。异常中断上游中途断开时要确保客户端能收到一个明确的结束信号而不是一直挂着。背压如果客户端消费慢要有机制避免网关内存被撑爆异步生成器天然有背压优势。async def stream_proxy(adapter, unified_req): async with httpx.AsyncClient(timeoutNone) as client: async with client.stream(POST, adapter.url, jsonadapter.build_request(unified_req)) as resp: async for line in resp.aiter_lines(): chunk adapter.parse_stream_chunk(line) if chunk: yield fdata: {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 测试让同一逻辑模型按比例分流到不同上游对比效果和成本。我个人最想加的是按任务自动选模型——业务侧连别名都不写只描述任务网关根据历史数据自动选性价比最高的上游。不过这需要积累足够的调用数据才能做准属于锦上添花。最后分享一个小技巧网关上线初期一定要开一个影子模式把请求同时发给主上游和一个候选上游对比两者结果差异但不影响业务返回。这样能在切换模型前拿到真实数据避免拍脑袋决策。我在切换主力模型时用过这招发现候选模型在某些中文场景下确实更稳果断调整了路由权重。