我是在同时接入第三个模型API的时候，终于忍无可忍，决定自己动手写一个统一接口。如果你也正在同时对接两三家不同厂商的AI模型API，大概已经体验过那种“换个厂商就要改一遍业务代码”的滋味。参数名不一样、返回结构不一样、鉴权方式不一样、连报错格式都各说各话，接得越深，维护成本越高。这篇博文是我从零开始接多模型API、最后做成统一网关的完整记录，会讲清楚为什么要做、怎么做、设计时哪些细节最容易被忽略，以及实际运行中踩过的坑。适合正在做AI应用后端、需要对接多家模型供应商的同学参考，也适合想搞明白“统一接口到底统一的是什么”的人。
1. 先搞清楚：多模型接入为什么这么痛
1.1 接口格式参差是最大的坑
我最早接第一个模型API的时候，觉得很简单——填一个API Key，把prompt发过去，拿回文本就行。真正开始接第二个、第三个的时候，才发现事情远没有这么轻松。
不同模型的请求字段习惯完全不一样。有的叫max_tokens，有的叫max_output_tokens；有的直接叫temperature，取值范围是0到1，有的却是0到2，甚至还有别名叫top_k、repetition_penalty这种需要额外映射的参数。再看响应体，普遍的做法是把生成结果放在choices数组里，但具体字段路径、finish_reason的枚举值、usage的字段命名，各家都有各家的写法。更麻烦的是错误码体系，一个模型返回429代表限流，另一个可能返回400，还有的会把限流信息藏在响应体里的某个嵌套字段里，不仔细看文档根本发现不了。
这带来的直接后果是：业务代码里全是if 厂商A then ... elif 厂商B then ...。每新增一家模型供应商，就要把这些分支再复制扩展一遍。时间一长，代码变得又臭又长，改一个公共逻辑要同时顾及好几套数据格式，线上出问题也很难排查到底是哪一层出的错。
1.2 统一接口到底统一什么
在动手写代码之前，我花了比较多的时间想清楚一个问题：统一接口到底是为了统一什么？
很多人以为“统一接口”就是把各家API包一层，让调用方发同样的HTTP请求，拿到同样的JSON返回。这当然没错，但只是最表面的一层。我的理解是，统一接口要统一的是四样东西：请求格式、响应格式、错误格式、流式传输协议。这四样统一了，上层的应用代码才能做到“一次编写，多模型运行”。
在具体选型上，我建议内部统一采用目前行业里使用最广泛的“聊天补全”格式。为什么选它？因为这种格式本身就有比较清晰的约定——请求体里包含model、messages、temperature、max_tokens、stream这些标准字段，响应体里包含id、choices、usage。大量开源的客户端库、工具链和教程都基于这套格式，生态最成熟。让外部厂商来兼容你，远比你挨个去兼容他们省事得多。
另外，统一接口也要划清边界。它不应该关心业务层的意图，比如“这个对话是不是要走RAG检索”“那个请求要不要走敏感词过滤”，这些应该由更上层的业务服务去处理。统一接口的职责就是：接收统一格式的请求，路由到正确的模型供应商，把结果翻译成统一格式返回。
2. 统一网关的核心细节：这些环节最容易被忽略
2.1 请求参数映射与归一化
真正动手做适配层的时候，最烦也最关键的就是参数映射。
每个模型都有自己的“脾气”。有的模型temperature有效范围是0到1，有的一直可以调到2，而且默认值也不一样。有的模型用max_tokens，有的则用max_output_tokens，甚至还有同时支持俩但语义不同的。如果没有统一映射，上层业务传一个temperature=0.7，到不同模型那里代表的“随机性”完全不一样，出来的效果自然也不一样。
我采取的办法是：在网关内部定义一套标准字段和标准取值范围，每个Provider负责把标准值翻译成目标模型的期望值。比如内部统一使用0到1的temperature范围，当适配某个取值范围为0到2的模型时，就做个线性映射，0.7的内部值映射过去就是1.4。如果目标模型不支持某个字段，Provider可以直接忽略或者给个默认值，但网关对外始终暴露同一套语义。
不要把这种映射逻辑散落在业务代码里。它必须收敛在Provider适配层，并且用单元测试保护起来。每接一个新模型，先给映射逻辑补测试，再联调业务，这是我在实践中总结出来的比较可靠的顺序。
2.2 鉴权与密钥管理
模型API的鉴权方式也五花八门，有在Header里带Authorization的，也有要求把API Key放在请求体某个字段的，还有用Query参数传递的。如果实现统一网关，就不能让调用方去关心这些差异——调用方只需要带自己的网关凭证，由网关在转发时动态注入目标模型的密钥。
密钥管理这块，我要特别强调：不要相信网上那些把密钥硬编码在代码里的示例。开发环境可以把密钥放在.env文件里，但生产环境一定要用密钥管理服务或纯环境变量注入，并且要支持多密钥轮换。我在做网关时甚至设计了一个小的密钥池，同一个模型配多把Key，当遇到限流时自动切换，这在生产环境能救命。
还有一个容易踩的坑：日志里泄露密钥。网关在记录请求日志时，会把携带API Key的Header原样打印出来。一旦日志系统被扫描或同步到第三方，等于把钥匙直接交出去了。务必在日志输出前做脱敏处理。
2.3 流式响应与SSE实现
如果说普通请求转发是“传话”，那流式响应就是“直播”。每个模型在流式模式下都会返回一段一段的数据，这通常是基于SSE（Server-Sent Events）实现的。SSE的格式说简单也简单：通过Content-Type: text/event-stream返回，每行以data:开头，两条消息之间用空行分隔。
但统一网关在处理流式时，有几件事必须做好。
第一是逐块翻译。你不能直接把外部模型吐出来的原始chunk原样转发给客户端，因为chunk里面的字段名、结构都是外部模型自己的格式。你要做的是读取chunk，按内部标准格式重新组装，再通过标准SSE事件发出去。
第二是结束标记。流式响应的最后通常会有一个finish_reason或者[DONE]标记。网关必须正确识别并转发这个标记，否则客户端会一直挂等，直到超时。
第三是心跳。有些网络环境对长时间没有数据流动的长连接不友好，中间代理可能会把连接掐掉。稳妥的做法是在流式过程中定时发一个注释行作为心跳，保持连接活跃。
这三件事里任何一件没做好，用户的实际体感就是“网页上字蹦到一半卡住了”或者“转半天圈没反应”。很多初稿的网关都栽在流式这块，我也不例外。
3. 实操：从零实现一套统一接口
3.1 整体架构与工程目录
我先说一下整体结构。一个最小的统一网关，从上到下分三层：路由层、网关服务层、Provider适配层。路由层接收外部请求，做基本校验；网关服务层负责把请求转化成内部标准格式、选择目标Provider、处理错误和重试；Provider适配层负责和具体的模型API打交道。
工程目录我习惯这么组织：
app/ ├── main.py # 网关入口，注册路由 ├── config.py # 环境变量与配置加载 ├── schemas.py # 内部标准协议的数据模型 ├── gateway.py # 核心服务：路由、转换、流式转发 └── providers/ ├── base.py # Provider基类，定义统一抽象 ├── provider_a.py # 适配模型A的Provider └── provider_b.py # 适配模型B的Provider这个结构的好处是每加一个新模型，只需要在providers目录下新增一个文件，继承基类，实现对应方法，然后注册到配置里即可。业务代码不需要任何改动。
3.2 定义内部标准协议
我使用Python来写网关，因为异步支持好、生态成熟、写起来也很快。第一步是定义内部标准的数据模型，用一段示意代码说明：
# app/schemas.py from typing import Optional, List from dataclasses import dataclass, field @dataclass class ChatMessage: role: str # user / assistant / system content: str @dataclass class ChatRequest: model: str # 统一模型名，路由用 messages: List[ChatMessage] temperature: float = 0.7 # 网关内部统一0到1 max_tokens: Optional[int] = None stream: bool = False user_id: Optional[str] = None # 用于审计与成本归因 @dataclass class ChatChunk: id: str role: str content: str finish_reason: Optional[str] = None @dataclass class ChatResponse: id: str content: str finish_reason: Optional[str] = None prompt_tokens: int = 0 completion_tokens: int = 0这套数据模型就是“内部通用语言”。不管外部模型长什么样，进了网关以后，统一按这套模型流转。
3.3 编写Provider基类与具体适配器
Provider基类的作用是定义统一的能力接口，具体实现由各个适配器完成：
# app/providers/base.py from abc import ABC, abstractmethod from typing import AsyncIterator from app.schemas import ChatRequest, ChatResponse, ChatChunk class BaseProvider(ABC): name: str = "" # 模型标识前缀 @abstractmethod async def chat(self, req: ChatRequest) -> ChatResponse: """非流式调用外部模型并返回标准响应""" raise NotImplementedError @abstractmethod async def stream_chat(self, req: ChatRequest) -> AsyncIterator[ChatChunk]: """流式调用外部模型，逐块产出标准chunk""" raise NotImplementedError接下来是具体的Provider适配器。以模型A为例，它要求把请求发到某个接口，并且使用max_output_tokens而不是max_tokens，temperature范围是0到2。那么在适配器里要做两层转换：
# app/providers/provider_a.py from app.schemas import ChatRequest, ChatResponse, ChatChunk from app.providers.base import BaseProvider from typing import AsyncIterator class ProviderA(BaseProvider): name = "provider-a" def _to_external_request(self, req: ChatRequest) -> dict: return { "model": req.model.removeprefix("provider-a/"), "prompt": self._messages_to_prompt(req.messages), "temperature": req.temperature * 2, # 由 0~1 映射到 0~2 "max_output_tokens": req.max_tokens or 1024, "stream": req.stream, } def _from_external_response(self, raw: dict) -> ChatResponse: # 把外部响应字段翻译成内部标准字段 return ChatResponse( id=raw.get("request_id", ""), content=raw["output"]["text"], finish_reason=raw["output"].get("finish_reason"), prompt_tokens=raw.get("usage", {}).get("prompt_tokens", 0), completion_tokens=raw.get("usage", {}).get("completion_tokens", 0), ) async def chat(self, req: ChatRequest) -> ChatResponse: external_req = self._to_external_request(req) # 这里用你熟悉的HTTP客户端发起POST请求 raw = await http_post(self.name, external_req) return self._from_external_response(raw) async def stream_chat(self, req: ChatRequest) -> AsyncIterator[ChatChunk]: external_req = self._to_external_request(req) # 逐行读取外部流式数据，翻译成标准chunk后产出 async for raw_chunk in http_stream(self.name, external_req): yield ChatChunk( id=raw_chunk.get("id", ""), role="assistant", content=raw_chunk["output"]["delta"], finish_reason=raw_chunk["output"].get("finish_reason"), )模型B适配器也是类似的写法，区别只是外部请求的构造方式、鉴权头以及响应字段解析方式不同。这里的关键在于：所有差异都被关在了适配器内部，外部完全感知不到。
3.4 网关路由与流式转发
网关入口做的事情很纯粹：根据请求里的model字段，找到对应的Provider，调它。如果是流式请求，就返回一个SSE流；如果不是，就返回一个标准JSON。
# app/gateway.py from app.schemas import ChatRequest, ChatResponse from app.providers.base import BaseProvider PROVIDER_REGISTRY: dict[str, BaseProvider] = {} def route_provider(model: str) -> BaseProvider: prefix = model.split("/", 1)[0] provider = PROVIDER_REGISTRY.get(prefix) if provider is None: raise ValueError(f"unknown model prefix: {prefix}") return provider当收到流式请求时，网关需要把Provider吐出来的标准chunk，格式化成一个一个SSE事件，然后写回客户端。这里要特别注意，不要让业务层的等待、重试逻辑阻塞流式写入。每次从Provider拿到一个chunk，就立刻把它flush给客户端，不要攒批次。
超时配置也要分层处理：HTTP客户端的连接超时要短，比如5秒；读取超时要长，比如60秒甚至更长，具体看模型流式返回的速度。如果读取超时设得太短，模型生成长文时稍微犹豫一下，连接就被断掉了，用户看到的就是“生成到一半卡死”。
3.5 本地启动与联调验证
写好代码后，先用本地环境做联调。我会启动网关，然后用curl分别调用两个模型的统一接口，验证它们的行为表现完全一致：
curl http://127.0.0.1:8080/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "provider-a/model-a-v1", "messages": [{"role": "user", "content": "你好"}], "stream": false }'换一个模型，只需要改model字段的前缀：
curl http://127.0.0.1:8080/v1/chat/completions \ -H "Content-Type: application/json" \ -d '{ "model": "provider-b/model-b-v2", "messages": [{"role": "user", "content": "你好"}], "stream": true }'如果两个请求拿到的返回结构一致，说明网关的适配工作已经基本到位。再测一遍流式，观察SSE事件序列是否正确结束，就可以进入更真实的场景了。
4. 真实环境下踩过的坑：问题排查速查
4.1 鉴权问题：401/403的经典原因
我在接入第一个模型API时就遇到过诡异的现象：同一个API Key，直接在本地curl能通，通过网关转发却一直返回401。排查了很久才发现，是网关使用的HTTP客户端在构造请求时，把Authorization头放在了自定义Header集合里，被某个中间处理环节吞掉了。
这种问题在网关里其实很常见。排查鉴权类报错时，我一般按这个顺序检查：
第一，确认请求真正到达外部API时Header是否完整。可以临时在网关里打一条带完整请求头的日志，或者用HTTP代理工具转发看一下，确认Authorization有没有被正确透传。
第二，确认API Key本身没有额外的空格或换行符。有些密钥是从浏览器直接复制的，会悄悄带上\n或者空格，导致校验失败。
第三，确认不该透传的自定义Header没有污染外部请求。比如你给调用方发的网关令牌，如果也被转发给了外部API，被误判成无效凭证，同样会返回401。
还有一点经验之谈：不要在网关日志里明文记录API Key和Authorization头。就算只是在本地调试，也尽量只记录后四位，否则日志文件泄露就等于密钥泄露。
4.2 流式响应中断：前端卡住、半包问题的排查
流式中断是接入后期最容易遇到的问题，而且现象五花八门：有时候是“开始正常，中途停了”，有时候是“首次请求就无响应”，有时候是“A模型正常，B模型一直断”。
我遇到过最典型的一个场景是网关部署在某平台之后，所有流式响应都在大约30秒后断开。排查下来发现是负载均衡器的空闲超时设置太短——流式请求如果两个数据块之间间隔超过30秒，连接就被静默掐断了。解决方案是给SSE流加一个周期性心跳行，让连接保持“活跃状态”，同时在代码里把读取超时调大。
另一个常见的坑是网关在处理流式数据时，用了带缓冲的写入方式，数据在缓冲区里攒着没有即时flush。客户端看到的就是“半天没字蹦出来”。解决方法是拿到chunk立即flush，不要让数据在网关里滞留。
排查这类问题，我建议在网关层加一个stream日志，记录每个SSE事件的编号、时间戳和间隔。这样一旦出问题，可以直接看出是外部模型不吐数据了，还是数据卡在网关了。
4.3 上下文长度与Token限制
不同模型的上下文窗口差别很大。有的支持大几十万token，有的只有几万token。如果统一接口不做任何处理，发送给窗口较小的模型时，会直接报输入超长错误。
我在网关里加了一个简单的估算模块：发送前先按内容长度粗估token数，然后再看目标模型的上限，预先留出足够的输出buffer。比如目标模型窗口是8K token，我预留2K给输出，那么输入超过6K就触发告警，超过上限就返回提示，请上层做截断或摘要。
估算token没有完美的公式，但我用了一个比较实用的经验：中文按每个字1.5到2个token估算，英文按4个字符1个token估算。这样虽然不精确，但足够安全。真正精确的统计应该由Provider返回的usage字段来核准，所以我在响应里会透传真实的token统计，方便上层系统做精细控制。
4.4 限流与429的应对策略
模型API的限流策略各有不同，有的是按每分钟请求数（RPM），有的是按每分钟token数（TPM）。统一网关面对这些限流，至少要处理好两件事：合理的重试和优雅的降级。
重试我强烈建议用指数退避加抖动。也就是第一次失败后等1秒，再失败等2秒，4秒、8秒这样翻倍，同时加一个随机扰动，避免多个请求同时重试导致流量洪峰再次触发限流。重试次数不要超过3次，超过就快速失败，把错误信息转成标准格式返回给调用方。
优雅降级则是给整个网关增加一层“Provider健康度”管理。如果某个Provider连续多次返回429或5xx，就把它的状态标记为不健康，后续请求自动路由到备用模型上。这个功能我放在后面的扩展里详细介绍，但它的价值在真实环境里很快就体现出来了。
5. 统一接口上线后，我还会做的几件事
5.1 自动回退：让故障在用户侧几乎无感
网关上线一段时间后，我遇到过一个很难受的场景：某个模型供应商在高峰期不稳定，时不时报错，但用户请求已经打进来了，总不能让他们看到“服务异常”就完事。后来我给网关加了一个回退机制，核心思路不复杂，就是给模型配置一个有序的候选列表，优先用第一个，失败的场景下自动切到下一个。
配置看起来像这样：
routes: - name: chat-default candidates: - provider-a/model-a-v1 - provider-b/model-b-v2 fallback_condition: [500, 502, 503, 429, timeout_error]这个机制部署之后，很多瞬间的故障用户完全感知不到。但回退逻辑一定要设置好阈值，不能让同一次请求内部无限循环切换，否则成本和延迟都会被放大。我的做法是：一次请求最多回退2次，且只在“连接失败、超时、5xx、429”这四类错误下触发。
5.2 用量统计与成本归因
统一网关的另一个好处是，它是所有模型请求的必经之路。做用量统计和成本归因会变得非常自然。
我在网关层给每个请求生成了一个request_id，并把它贯穿到日志和响应体中。调用方可以传user_id，网关负责把model、prompt_tokens、completion_tokens、provider、耗时、是否回退这些结构化日志写到统一的日志管道里。
有了这些数据，后续做成本分析就非常方便。只需把各模型的计价规则写成一个配置表，再用usage里的token数做乘加计算，就能得到每次请求的实际成本，按用户、按模块、按天做聚合分析。没有这一步，模型费用就是一笔糊涂账。
5.3 缓存与并发控制的取舍
最后聊两个容易过度设计的点。
一个是缓存。相同请求命中缓存可以显著省钱，但不是所有场景都适合。比如对话类请求通常带个性化上下文，直接启用缓存会导致结果张冠李戴。我的建议是：缓存只对幂等的无状态请求开放，并且key需要包含model、provider、消息内容的hash。敏感内容默认不落缓存，这条一定要在合规层面想清楚再上。
另一个是并发控制。统一网关在高峰期很容易同时收到大量请求，如果下游某个模型自身限流比较严格，网关必须做并发控制，否则会直接触发429风暴。我用了最简单的信号量方式：为每个Provider维护一个并发上限，超过上限的请求先排队等待。这个机制虽然简单，但特别有效，能避免很多不必要的重试和报错。
我在实际运行这个网关半年后最大的体会是：统一接口最大的价值不在于“省掉几行代码”，而在于把多模型的不确定性隔离在一个明确的边界之内。上游再怎么变，业务侧始终面对的是一套稳定的协议。后续每接一个模型，成本都在逐步递减，而稳定性带来的收益会随着模型数量的增加不断放大。如果你也在做类似的接入工作，建议从最小可用版本开始，先统一一个非流式接口跑通，再逐步加上流式和回退，不要一开始就追求大而全。
