这两年做 AI 应用，最磨人的不是单个模型的效果，而是项目里模型 API 接得越来越多以后，业务代码里开始到处散落着针对不同提供商的特殊分支。这篇文章想分享的是我从零做一个多模型 API 统一接入层的完整过程：怎么设计协议、怎么写适配器、怎么处理流式输出，以及那些只有跑过生产流量才能发现的坑。适合正在做 AI 应用集成、想给团队省掉重复对接工作的读者，也适合刚拿到“统一模型网关”任务、不知道从哪里下手的同学。
先说结论：统一接口并不是简单做一个“请求转发”，它真正要解决的，是把多个模型服务商的 API 差异关在同一个边界里，让上层只面对一套稳定的请求和响应语义。这里的稳定不仅是字段格式稳定，还包括错误码语义稳定、流式结束语义稳定、用量统计口径稳定。只有把这四件事都定好，统一接口才能真正成为业务敢依赖的基础设施。
1. 为什么每个多模型 API 项目都要先把“统一接口”想清楚
1.1 业务代码里的 if-else 是这样一步步腐化的
一个很常见的场景。第一版系统只接了一个模型服务商 A，业务代码里直接调用 A 的客户端，代码很干净。后来 A 偶尔限流，为了不影响线上体验，增加服务商 B；再过一段时间，某个新场景需要更长的上下文，又想接 C。于是业务代码里开始出现类似的模式：
if provider == "A": result = call_provider_a(...) elif provider == "B": result = call_provider_b(...) elif provider == "C": result = call_provider_c(...)不同服务商的参数名不一样、返回结构不一样、错误码定义不一样、重试策略也不一样。今天 A 的响应格式升级了，你要在业务代码里找所有调 A 的地方；明天 B 的鉴权方式改了，又要动一遍。这种腐化几乎是必然的，因为“模型调用”这件事本身正在变成基础设施，基础设施不应该散落在业务代码里。统一接口的价值，就是把这些差异全部隔离到一个独立层，业务侧只依赖你自己定义的那份协议。
更麻烦的是，这种 if-else 往往还会在客户端、服务端、定时任务里重复出现。你会发现同一段适配逻辑被复制了好几次，每次出现的地方还不完全一致：A 团队处理错误的方式和 B 团队不一样，C 团队的重试次数又不一样。最后线上出了问题，排查时甚至分不清是哪一个副本在处理请求。所以，尽早把多模型 API 的调用收敛到一个统一接口，本质上是在控制技术债的扩散。
1.2 统一接口不光是少写代码，它同时承担路由、兜底、成本和安全
如果以为统一接口只是为了“少写几个适配”，那就低估它了。接入多家模型之后，真正有价值的能力都发生在这个统一层里。路由、兜底、成本、安全这些能力，单独看每一个都能在业务侧用 if-else 实现，但合在一起，只有集中到一个统一入口才能稳定地提供。
- 路由：同样是“聊天问答”，可以让默认流量走服务商 A，当 A 的延迟高于阈值时，自动把新请求切到 B。
- 兜底：A 返回 503 或连续超时，网关自动重试到 B，用户侧感知不到切换。
- 成本：所有请求都经过统一入口，每一笔 token 消耗、单价、总费用都可以记下来，按业务线归因。
- 安全：上游密钥只需要保存在统一接口这一层，业务同学接触不到明文密钥，也不用私下互相拷贝凭据。
这些能力一旦散到业务代码里，每个团队都得各自实现，而且大概率实现得有差异。集中到统一接口之后，配置一次，所有接入方都受益。
2. 统一接口到底统一什么：协议、适配器与注册表
想清楚为什么要做，接下来就是设计。设计的第一步，不是马上写 HTTP 服务，而是先回答一个问题：我们统一的是“传输格式”，还是“语义”？传输格式最简单，把请求体规范和响应体规范定下来就可以了；真正难的是语义，包括什么是完成、什么是错误、什么是流式结束，这些在厂商之间差异很大。
2.1 统一请求协议：最小字段集加上扩展字段
我在项目里定义请求体时，遵循一个原则：核心字段尽量少，厂商独有参数尽量放到扩展字段。统一请求的核心字段只有这些：
- model：逻辑模型名，比如 chat-basic、chat-speed、long-context，不直接暴露厂商模型名。
- messages：标准的 role/content 消息数组。
- temperature、max_tokens：通用采样参数。
- stream：是否流式返回。
- extensions：厂商差异化参数，比如某些平台特有的 response_format、thinking 开关、人工审核参数。
之所以不把所有参数都统一，是因为各家都有各自的私有能力。强行把“私有能力”都抽象成标准字段，会导致协议膨胀，新厂商接入的时候反而不方便。统一接口的价值在于把 80% 的通用场景统一掉，剩下 20% 的私有场景通过扩展字段透传，并且在模型注册表里标记该厂商支持哪些扩展。
2.2 统一响应协议：把“差异”消化在适配器内
响应体我同样收敛成最小集：
- id：统一请求 ID。
- model：回显逻辑模型名。
- choices：包含 message（角色和内容）、finish_reason（stop、length、tool_calls 等）。
- usage：prompt_tokens、completion_tokens、total_tokens。
- error：统一错误结构，包含错误码、错误类型、可读信息。
厂商返回的错误千奇百怪，有的用 HTTP 状态码表达，有的在 body 里带一个内部 code。适配器必须把这些标准化。比如厂商 A 的 429 对应“限流”，厂商 B 的 400 里某个字段对应“上下文过长”，网关把它们分别映射成 rate_limit 和 context_length_exceeded。这样上层业务就不用关心“哪个厂商返回了什么错误码”，只需要统一处理“该重试”还是“该给用户提示”。
2.3 适配器模式：把厂商差异关进一个文件里
实现层面，我强烈推荐用适配器模式，而不是在一个大服务里堆 if-else。每个厂商一个 Adapter 类，对外暴露同样的接口，输入统一请求，输出统一响应。新增厂商时，只需要新增一个文件，注册到工厂里，改动范围非常小。
实现时定义三层结构：统一的请求对象、统一的响应对象、适配器接口。每个适配器只负责两件事：把统一请求翻译成厂商私有请求，再把厂商私有响应翻译回统一响应。翻译这个过程看似简单，实际上有很多细节，比如厂商 A 要求 temperature 不能为 0，需要做数值钳制；厂商 B 的 max_tokens 有最小值限制，需要自动抬起；厂商 C 的时间戳格式不一样，需要按统一标准对齐。
2.4 选型：独立网关服务还是客户端 SDK
很多团队会纠结，统一接口到底做成一个独立 HTTP 服务，还是一个业务进程内的 SDK。我的建议是：如果团队只有一种语言、且对单次调用延迟极度敏感，优先做轻量 SDK；只要存在多语言客户端、需要集中管理密钥、要对成本做统一归因，就优先做独立网关。
独立网关最大的优势是权限收敛：上游密钥不进入业务容器，运维只需要保护网关这一层；同时可以做全量的可观测性和成本账本。缺点是多一跳网络延迟，但可以通过在同一网络环境内部署、使用内网域名访问来把开销压到很低。我实际采用的是“网关为主体、SDK 为薄封装”的组合：业务侧依赖一个小客户端，只负责构造统一请求和解析统一响应，真正调用和治理逻辑都放在网关。
3. 从零实现多模型 API 统一接入：核心步骤与参数设计
这一节是实操部分。我会按一个最小可用版本的顺序来讲：先跑通最简单的非流式转发，再做流式适配，然后加路由、注册表、健康检查。代码示例以 Python 为主，但思路不依赖具体语言框架。
3.1 定义一个不依赖厂商的请求与响应对象
基础数据结构越简单越好，我习惯先用普通类定义，不急着引入重量级框架。因为这些对象会贯穿整个网关代码，如果一开始就绑到某个 Web 框架的模型类上，后面换框架、写 SDK、做测试都会很痛苦。普通类的好处是零依赖、容易序列化、也方便在单元测试里直接构造。我还会为每个对象实现 to_dict 和 from_dict 方法，纯粹用来做序列化，不掺业务逻辑。流程中所有进过网关的数据，都可以统一落地成 JSON 日志。
class NormalizedRequest: def __init__(self, logical_model: str, messages: list, temperature: float = 0.7, max_tokens: int = 2048, stream: bool = False, extensions: dict = None): self.logical_model = logical_model self.messages = messages self.temperature = temperature self.max_tokens = max_tokens self.stream = stream self.extensions = extensions or {}统一响应对象则包含结果和用量。这里有一个容易被忽略的设计点：usage 字段必须同时保留“厂商原始用量”和“归一化后的用量”。因为有些厂商上报的 token 数口径不一样，不做归一化会导致成本对不上，后面排查问题时也缺少原始信息。
class NormalizedResponse: def __init__(self, logical_model: str, content: str, finish_reason: str, usage: dict, raw: dict): self.logical_model = logical_model self.content = content self.finish_reason = finish_reason self.usage = usage # 统一口径后的用量 self.raw = raw # 厂商原始响应3.2 适配器接口与第一版厂商实现
接口抽象成三个方法：请求翻译、非流式调用、流式调用。这样接入一个厂商时，思路非常明确：翻译请求、调厂商、翻译响应。不建议在适配器里写任何业务规则，比如“什么场景走哪家厂商”这种应该交给上层的路由模块，适配器只做格式和语义的转换。如果某个厂商的接口有特殊鉴权头或特殊 endpoint，也放在适配器的初始化参数里，通过 settings 传入，不要写死在代码里。
class BaseAdapter: def __init__(self, settings: dict): self.settings = settings def translate_request(self, norm: NormalizedRequest) -> dict: raise NotImplementedError def translate_response(self, raw: dict, request: NormalizedRequest) -> NormalizedResponse: raise NotImplementedError async def call(self, norm: NormalizedRequest, timeout: float) -> NormalizedResponse: raise NotImplementedError async def stream(self, norm: NormalizedRequest, timeout: float): raise NotImplementedError以厂商 A 为例，它的私有协议里模型名不叫 chat-basic，而叫 model-a-32k，并且 temperature 不支持小数。那适配器的 translate_request 就要做模型名映射和参数钳制。这些逻辑写在一个文件里，后续模型升级，只需要改动这个适配器，不会波及业务。
3.3 流式输出统一：SSE 事件映射与结束判定
流式输出是最容易踩坑的环节。不同厂商的流式事件格式差别很大。有的把增量内容放在 delta 里，有的放在 choices[0].delta.content；有的用固定事件名，有的每条数据自带 finish_reason。统一网关对外建议固定一种 SSE 协议，例如统一事件格式：
{"id": "req_xxx", "choices": [{"index": 0, "delta": {"content": "你好"}, "finish_reason": null}]}当流结束时，网关发送一个 done 事件，并携带累计用量。适配器要做的事情就是逐条转换：收到厂商 A 的原始 chunk，解析出增量文本，拼成统一 chunk；如果遇到厂商结束标记，再根据已经累积的内容生成统一 done 事件。这里有个细节：部分厂商在流式响应结束时不返回 usage，需要网关自己根据输入 token 数和输出字数估算，或者在下一次非流式请求里补查。我在实践里倾向于在流结束后，把估算用量写入日志，不做实时精确计费。
3.4 超时、重试参数与幂等性推导
统一网关还要负责超时和重试，这两项参数如果设计不好，在线上的代价很大。我的默认参数表是这样：
| 阶段 | 默认值 | 说明 |
|---|---|---|
| 连接超时 | 3 秒 | 建立连接的最长等待时间 |
| 读超时（非流式） | 60 秒 | 等待首个响应体的最长时间 |
| 读超时（流式） | 300 秒 | 整个流式过程允许的最长时间 |
| 重试次数 | 2 次 | 有限次重试，避免雪崩 |
| 基础退避 | 500 毫秒 | 指数退避的起始等待时间 |
| 最大退避 | 5 秒 | 防止重试排队太久 |
重试有一个非常容易被忽略的问题：POST 请求不一定是幂等的，盲目重试可能造成用户被重复扣费。我的做法是网关为每个请求生成一个 request_id，并把它附带给上游厂商作为幂等建。在重试时，如果上游返回的是网络层错误（连接重置、读超时），不确定请求是否已经被处理，就直接失败返回，而不是在超时后无限重试。只有明确收到 429、503 这类“服务端明确拒绝”的响应，才按退避策略重试。
3.5 动态路由与成本计算示例
有了注册表和适配器，就可以做路由了。路由规则可以非常灵活，最常用的是“可用性优先”和“成本优先”。比如成本优先路由，假设 A 和 B 处理同样请求的成本不同，A 便宜但慢，B 贵但快。在一个预算周期内，可以按 70/30 的比例把流量切给 A 和 B。假设平均每次请求 A 需要 2000 输入 token 加 500 输出 token，单价分别是每千 token 0.5 和 1.5；B 单价是 A 的 2 倍。A 的单次成本是 2000 / 1000 * 0.5 + 500 / 1000 * 1.5 = 1.75，B 是 3.5。70% 流量走 A、30% 流量走 B 时，平均成本是 0.7 * 1.75 + 0.3 * 3.5 = 2.275。如果不做路由，全部走 B，平均成本就是 3.5。合理性一目了然。
路由实现上，我建议把规则做成可配置项，比如 JSON 规则：策略类型、权重、候选厂商列表、熔断阈值。业务不用改代码，运维同学可以在管理界面调整。系统运行时，网关会自动感知厂商健康状态，不健康厂商的权重会被临时置零。
3.6 模型注册表与健康检查
模型注册表是整个网关的“数据库”。一个逻辑模型可以对应多个厂商模型，每条注册记录至少要包含这些字段。这个表在早期可以先用 YAML 或 JSON 文件维护，等接入模型超过十个、权限维度变多之后，再迁移到配置后台或数据库。把模型定义从代码里抽出来的意义在于，改模型映射、调权重、加候选都不需要发版，线上可以热更新。
| 字段 | 说明 | 示例 |
|---|---|---|
| logical_model | 业务使用的逻辑模型名 | chat-basic |
| candidates | 按优先级排列的厂商模型映射列表 | A: model-a-32k, B: model-b-chat |
| capabilities | 支持的能力标记 | stream, tools, json_mode |
| cost | 价格信息，用于成本路由 | input=0.5, output=1.5 |
| health | 当前健康状态与连续失败次数 | healthy / unhealthy |
健康检查我推荐“被动为主、主动为辅”。只需要在每次请求成功后记录最近成功时间，在失败后递增连续失败次数；连续失败达到 5 次就标记不健康，并从路由候选中剔除；每 30 秒做一次主动最小请求探测，探测成功后恢复。不要用高频主动探测去打扰厂商，也不要在高负载下再叠加无意义的请求。
4. 常见问题与排查技巧实录
这一节把我实际踩过的坑按“现象、原因、解法”整理成快查列表。很多问题看起来是模型问题，其实是统一网关处理不到位。
4.1 现象：接口偶尔超时，但厂商控制台显示请求成功
这种问题在非流式调用里最容易出现。根因往往是响应已经生成了，但网关在读超时时间没到之前没有收到完整响应，导致业务侧提前失败。排查时先对比“厂商侧耗时”和“网关耗时”，如果厂商侧耗时很长，说明确实是模型速度慢，对策是调大读超时，或者换更快的路由候选；如果厂商侧耗时很短但网关侧耗时接近读超时，就要检查是不是代理层、序列化、日志打印拖慢了响应。建议在网关里给“上游响应耗时”和“网关总耗时”打两个独立指标，对比着看。
4.2 现象：流式输出中间突然中断，但没有错误码
流式接口比非流式更难排查，因为它可能在任意一个 chunk 之后连接断开。最常见的两个原因：厂商服务端因为业务侧读太慢而主动断开；或者网关在流过程中做同步持久化、日志上报，阻塞了消费循环。解法是把“读流”和“处理流”完全解耦：读流协程只负责把原始 chunk 放进内存队列，另一个协程负责转换为统一事件并发送。不要在流式循环里夹杂网络请求、数据库写入这类重操作。
4.3 现象：账单对不上，厂商控制台费用高于网关统计
原因通常是重试导致同一请求被计费多次，或者流式响应没有统计 usage。我的经验是网关必须记录“真实调用次数”而不是“业务请求次数”。一个业务请求可能因为熔断、降级调用了两家厂商，产生两次上游消费，但这个次数不应该算到业务头上。成本对账时，以网关流水为准，厂商控制台只做参考；如果相差超过 5%，优先查非流式重试和流式补算逻辑。
4.4 现象：切换模型后，上层出现以前没有的错误
这通常是“能力差异”问题。之前接的厂商支持 JSON 模式、支持工具调用，新接的厂商不支持，但统一协议没有体现。结果上游把 tool_calls 参数透传下去，厂商直接报错。解法就是在模型注册表里加 capabilities 字段，业务通过“查询模型能力”来决定是否启用某个能力，而不是假设所有模型都支持。这一条看似简单，却能让切换模型的返工率大幅下降。
5. 生产落地：从“能跑”到“能上线”的几个经验
很多通用的网关代码写完、本地测试通过后，离生产可用还有很长一段距离。最后分享几个我认为最重要的生产经验。
5.1 先把 Mock 厂商搭好，再用真实厂商
对接外部 API 最怕联调时频率高、难以复现问题。我通常先做一个 FakeAdapter，内置固定延迟、随机错误、计数字段。发到统一接口的请求会被 FakeAdapter 随机返回限流或者超时，从而验证网关的熔断、重试和路由逻辑。这比拿真实 key 反复打断点高效得多。CI 也可以用 Mock 跑一遍回归，确保网关逻辑不破坏。
5.2 新厂商先给 5% 流量，稳了再放量
接入新模型最忌讳“一把切全量”。我建议先在注册表里添加一个新的逻辑模型名，比如 chat-basic-staging，配置路由权重为 5%，观察一段时间。关注错误率、平均延迟、P95/P99 延迟、用户反馈几个指标。如果没有明显劣化，再把权重逐步调到 30%、70%、100%。整个过程完全不发版，只是改配置，出了问题也能立刻回退。
5.3 每一次调用都留一条账本
统一网关天然适合做审计和成本归因。每次请求完成或失败，都要记一行结构化日志，包含 request_id、逻辑模型、厂商、上游模型、HTTP 状态、错误类型、耗时、token 用量、预估费用、是否发生重试和降级。这些日志汇总到面板后，可以很直观地看到每家厂商的可用性和单位成本。这个能力越早做，后面分析模型切换、投诉用户响应慢时就越省力。
5.4 密钥和敏感数据隔离
统一网关集中了多家厂商的密钥，这既是便利也是风险。密钥必须放在环境变量或专门的密钥管理服务里，不能出现在代码仓库里。如果业务有不同的数据安全等级，还需要支持按业务线、项目维度做路由隔离：部分项目只允许调用满足数据驻留要求的厂商，另一些项目可以使用其余候选。这个能力最好在设计协议时就把 project/tenant 字段带上，不然后期补会非常痛苦。
5.5 团队协作：协议文档先行，SDK 同步更新
统一接口一旦被多个业务使用，协议就变成了团队内的小契约。我一般会维护一份协议文档，包括请求示例、响应示例、错误码语义、流式事件格式、模型注册表规范。后端新增适配器时，前端同学只需要更新 SDK 的模型能力枚举，不用理解厂商细节。这样即使负责网关的同学离开，协议文档加上示例代码，也能保证新人快速接手。
写到最后，我个人最深的体会是：统一多模型 API 接入，功夫不在“写请求转发”，而在“把差异关起来”。先用最小协议跑通一个厂商，再逐步加适配器、加路由、加观测，比一开始就想做个万能网关要靠谱得多。如果你正面临模型 API 越接越多、代码越来越散的阶段，从统一协议开始做，大概率是投入产出比最高的选择。
