你有过这种经历吗?周末想把自己写的小应用从 GPT 换到 Claude,结果改了一晚上接口,聊天还没跑起来。我在做多模型应用开发的时候,这种经历差不多每周一次:接入的模型越多,接口碎片化问题就越明显——各家给的都是 HTTP API,但鉴权方式不一样、请求体字段不一样、返回结构不一样,连流式输出的字段位置都不一样。这篇文章是我亲身踩坑后的完整梳理,包括具体踩过的坑、最后沉淀下来的统一适配层设计、重试与路由策略,以及至今仍然没解决的边界问题。无论你是在做多模型聚合网关、模型路由,还是单纯想让产品快速切换模型,都可以拿这份经验做参考。
1. 接口碎片化到底是什么:我在同一个周末里调通了三个大模型
先不说理论,说个我自己的例子。当时产品需要同时接 OpenAI、Claude 和 Gemini,第一版我天真地以为"都是 HTTP 接口,改改 URL 就行"。等到真动手,才发现同一个"你好",三家接口长这样:
# OpenAI 系(包括 DeepSeek、Qwen 兼容接口) POST https://api.openai.com/v1/chat/completions Authorization: Bearer sk-xxxx { "model": "gpt-4o", "messages": [{"role": "user", "content": "你好"}] } # Anthropic Claude POST https://api.anthropic.com/v1/messages x-api-key: sk-ant-xxxx anthropic-version: 2023-06-01 { "model": "claude-3-5-sonnet-20241022", "max_tokens": 1024, "messages": [{"role": "user", "content": "你好"}] } # Google Gemini POST https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-pro:generateContent?key=AIza-xxxx { "contents": [{"role": "user", "parts": [{"text": "你好"}]}] }同一个语义,三种协议观感。OpenAI 走 Bearer Token,Claude 要走两个自定义头,Gemini 干脆把密钥放在 URL 查询参数里。请求体更是完全没法互认:OpenAI 用messages一条数组打天下,Claude 把 system 单独拎出来,Gemini 管 user 叫contents、管文本叫parts。这还没算返回结构,OpenAI 在choices[0].message,Claude 在content[0].text,Gemini 在candidates[0].content.parts[0].text。我第一次写解析函数的时候,感觉自己不是在写代码,是在做连连看。
1.1 碎片化到底碎在哪些维度
我后来把这些差异整理了一下,发现其实可以归纳成四个维度:
| 维度 | OpenAI 系 | Anthropic | Gemini |
|---|---|---|---|
| 鉴权方式 | Authorization: Bearer | x-api-key + anthropic-version | URL 查询参数 key |
| 请求结构 | messages 数组 | system + messages 数组 | contents + parts |
| 返回结构 | choices[0].message | content[0].text | candidates[0].content.parts |
| 流式格式 | SSE,delta.content | SSE,多事件类型 | SSE,candidates 整体返回 |
这表格看着简单,实际接入的时候每个格子里还藏着子差异。比如 OpenAI 的 usage 是prompt_tokens/completion_tokens,Claude 是input_tokens/output_tokens,Gemini 是promptTokenCount/candidatesTokenCount。你在做成本统计的时候,光字段映射就能写一屏代码。
1.2 更隐蔽的"伪兼容":OpenAI 兼容接口之间的差异
真正让我意识到问题严重性的,是国内一堆"OpenAI 兼容接口"。它们都宣称"直接兼容 OpenAI 格式",但用起来你会发现:
- 有的不认
max_completion_tokens,只认max_tokens; - 有的对
stream_options直接报 400; - 有的虽然支持 tools,但
tool_choice传具体函数名时会静默忽略,直接按 auto 处理; - 有的返回的
finish_reason只有stop/length,没有tool_calls,但 message 里又带着 tool_calls。
这种"半兼容"比完全不兼容更坑:你按 OpenAI 的姿势写,它能通;一旦上了生产流量,某个参数在某个模型上悄悄失效,表现出来就是偶发性错误,排查起来特别费劲。所以说接口碎片化不只是"三家格式不同"的问题,而是"每一家都在某个子集上不同"的问题——你必须有一套自己的统一层,而不是祈祷各家自觉对齐。
2. 统一抽象层怎么落地:先定义一套业务需要的最小契约
被折磨了一周之后,我决定做统一适配层。这个决定本身不稀奇,稀奇的是很多人做错了方向:一上来就想封装所有能力——视觉、Embedding、函数调用、流式、文件、JSON Mode 全都要统一,结果适配器越写越复杂,每加一个模型都要改核心代码,最后适配层变成了新的碎片。
2.1 别追求"全量统一",追求"业务够用"
我当时的判断标准很简单:我们的产品只依赖四件事——多轮对话、流式输出、函数调用、用量统计。那我就只统一这四件事,其他能力哪个模型有就用哪个,没有就以原始透传的方式暴露。
统一层设计成"业务侧的最小契约",而不是"厂商能力的最大公约数"。这样做的理由很实际:
- 最大公约数方案会逼你兼容三家最弱的那个能力;
- 最小契约方案让每个厂商适配器只做"翻译"这一件事,不做能力补齐;
- 遇到真正需要厂商专属能力(比如 Claude 的 thinking、Gemini 的 grounding)时,可以保留
raw字段透传,不阻碍业务。
用一个不恰当的比喻:适配层是翻译官,不是法官。翻译官的任务是把话说明白,不是逼所有人都说同一种方言。
2.2 用三个数据结构锁死边界
既然要统一,第一步是定标准请求、标准响应、标准流式增量。我用 Pydantic 定义了一套核心结构,所有适配器都基于这套结构做"翻译":
from typing import Optional, Any from pydantic import BaseModel class UnifiedMessage(BaseModel): role: str # user / assistant / system / tool content: Optional[str] = None tool_calls: Optional[list] = None # assistant 消息中发起函数调用 tool_call_id: Optional[str] = None # tool 结果回填时使用 class UnifiedRequest(BaseModel): model: str messages: list[UnifiedMessage] temperature: float = 0.7 max_tokens: int = 1024 stream: bool = True tools: Optional[list] = None class UnifiedDelta(BaseModel): text: str = "" finish_reason: Optional[str] = None tool_calls: Optional[list] = None class UnifiedResponse(BaseModel): id: str choices: list[dict] # 里面放 UnifiedMessage + finish_reason usage: dict = {} raw: dict = {} # 保底保留厂商原始回包,方便排障和透传这里有个容易被忽略的点:raw字段必须保留。你可能觉得"既然统一了,为什么还要原始数据?"实际运营中你会发现,厂商新增了一个字段(比如 Claude 的 stop_reason 里有tool_use,Gemini 的 finishReason 里有MAX_TOKENS),你还没适配完,线上代码可以先透传出来,避免阻塞业务。我后面排查线上问题,几乎每次都靠这个 raw。
2.3 适配器模式的落地:每个厂商只写一段"翻译官"
定义好契约后,我给每个厂商写一个 adapter,统一实现四个方法:
class ProviderAdapter: provider_name: str = "base" def build_request(self, req: UnifiedRequest) -> dict: # 把 UnifiedRequest 翻译成厂商的请求体 ... def parse_response(self, resp: dict) -> UnifiedResponse: # 把厂商回包翻译成 UnifiedResponse ... def parse_stream(self, lines) -> Iterator[UnifiedDelta]: # 逐行解析 SS E 流,产出统一增量 ... def raise_error(self, status: int, body: dict, headers) -> None: # 把厂商错误码映射成统一的异常类型 ...每个厂商一个文件,互不干扰。比如 OpenAI 适配器的build_request就是把 role 原样搬过去,Claude 适配器要把 system 从 messages 里拆出来单独放,Gemini 适配器要把 role 从assistant翻译成model、把内容塞进parts。这些翻译逻辑全是笨功夫,但脱离了业务,单独测试非常容易。
我强烈建议把厂商真实回包存成测试 fixture,然后对parse_response和parse_stream做单元测试。因为你没法保证厂商 API 不升级,fixture 测试能在厂商悄悄改字段的时候第一时间报警。这个习惯后来帮我发现过 Gemini 一次候选字段结构微调,吓得我赶紧看 changelog。
3. 最隐蔽的两个深坑:流式输出和函数调用的跨厂商差异
如果说请求和响应的字段差异是"明伤",那流式输出和函数调用就是"暗伤"。这两个地方踩坑的时候报错都不明显,表现出来全是"生成的文本缺一段""函数参数偶尔解析失败"这类玄学问题。
3.1 同样叫 SSE,事件结构完全不同
流式输出是三家里最让我崩溃的部分。OpenAI 的流式是最典型的:
data: {"choices":[{"delta":{"content":"你好"},"finish_reason":null}]} ... data: [DONE]Claude 的流式则是多事件模型:
event: message_start data: {"type":"message_start","message":{...}} event: content_block_start data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} event: content_block_delta data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"你"}} event: message_stop data: {"type":"message_stop"}如果你用 OpenAI 的解析逻辑去读 Claude 的流,每行都能读到,但一个字都抽不出来。我第一版就是这么干的,结果界面上一个字都不出,还以为网络断了。更恶心的是,Claude 为了保持长连接偶尔会发: ping这种注释行,你把整行丢进json.loads直接抛异常。
Gemini 的流式又有自己的脾气:它走的是streamGenerateContent,正常用 SSE 解析,但某些模型/某些网络环境下,返回的 event 里data是一个聚合了多句话的 JSON,而不是逐 token 增量。如果你的客户端逻辑建立在"一行一个 token"的假设上,遇到聚合包只能拿到一大段,UI 打字机效果会一顿一顿,甚至直接超时。
我当时写了统一的流式归一化函数,核心思路是按厂商特征做分支而非按"SSE 通用格式"处理:
def normalize_stream(self, raw_line: str) -> Optional[UnifiedDelta]: if not raw_line.startswith("data:"): return None # 过滤掉注释行 / 空行 / 心跳包 payload = json.loads(raw_line[5:].strip()) if payload.get("type") == "content_block_delta": # Claude return UnifiedDelta(text=payload["delta"].get("text", "")) if "choices" in payload: # OpenAI delta = payload["choices"][0].get("delta", {}) return UnifiedDelta(text=delta.get("content", "")) if "candidates" in payload: # Gemini parts = payload["candidates"][0]["content"]["parts"] return UnifiedDelta(text=parts[0].get("text", "")) return None依赖厂商的名字判断很丑,但很实用——因为你没法指望三家自己把格式统一。这里额外提醒:HTTP 流式传输中,JSON 可能被 TCP 分段切成半个包,所以不要自己拼字符串之后试图 json.loads 半截数据,一定要用iter_lines这类按行迭代的机制,或者上成熟的 SSE 库。我见过太多人栽在"chunk 没拼完就解析"上。
3.2 函数调用:三种 schema、三种回包,一个解析函数
函数调用(function calling)是碎片化的另一个重灾区。三家都支持,但格式完全不一样:
| 厂商 | 请求里的描述方式 | 回包里函数结果的位置 | 结果怎么回填 |
|---|---|---|---|
| OpenAI | tools: [{type:"function", function:{name, description, parameters}}] | choices[0].message.tool_calls[0].function.arguments | role=tool 消息,携带 tool_call_id |
| Claude | tools: [{name, description, input_schema}] | content 数组里 type=tool_use 的块,input 是对象 | user 消息里放 tool_result content block |
| Gemini | functionDeclarations: [{name, description, parameters}] | parts 数组里 functionCall 块,args 是对象 | user 消息里放 functionResponse part |
光看这张表就知道,适配器里要做的事情不只是改字段名:OpenAI 的arguments是一个 JSON 字符串,你需要字符串解析;Claude 的input和 Gemini 的args本来就是对象,但需要处理多轮对话时把历史消息里的函数块也翻译正确。
我踩的最深的一个坑是:OpenAI 流式模式下,函数调用的 tool_calls 是按 chunk 增量返回的,每个增量只带一部分 index 和字段,需要自己按 index 聚合完整 arguments,而且中间某个 chunk 可能只带{"role":"assistant","content":null}这种占位。如果直接拿每次 delta 塞进最终消息,你会得到半截 JSON 参数,传给对方 API 直接 400。Claude 的流式也要按 content_block index 聚合,Gemini 的流式反而是在单个 chunk 里给出完整的 functionCall。三种聚合方式,最后还是得上统一的 tool_calls 累加器。
参数解析我建议写一个容错函数,只做一件事:把"可能是字符串、也可能是对象"的参数统一成 Python 字典。因为各家不仅格式不同,有时还在 JSON 字符串里放进没转义干净的换行符。
至于tool_choice,也别贪心,统一成三种模式就够了:auto、none、force(name),剩下的厂商专有模式(比如 Claude 的any、OpenAI 的required)尽量在适配层映射,不要在业务代码里到处透传。透传一时爽,三个月后没人敢动这段代码。
4. 稳定性是碎片化的重灾区:错误码、限流和重试的统一
把请求和响应统一好,应用能跑通了,但一上生产你会发现新的问题:这家的限流形态和那家完全不一样,错误码也不能交叉理解。这里我最开始的做法是"哪家报错就在哪家的 SDK 里 catch",结果适配层下面的代码全是try: openai_call except: try: claude_call except: ...,丑得没法看。
4.1 各家 429 的"花样"不一样
限流是每个大模型 API 都会遇到的问题,但表现形式南辕北辙:
| 厂商 | 典型错误 | 是否带 Retry-After | 备注 |
|---|---|---|---|
| OpenAI | 429 rate_limit_exceeded / insufficient_quota | 有时带,单位是秒 | 响应头里有 x-ratelimit-* 系列 |
| Anthropic | 429 rate_limit_error / 529 overloaded_error | 429 带,529 不带 | 529 表示服务过载,重试成功率很高 |
| Gemini | 429 RESOURCE_EXHAUSTED | 不带 | RPD/RPM 配额共享,容易被别的任务挤占 |
| 部分国内兼容接口 | 429 但错误体千奇百怪 | 不一定 | 有的甚至拿 400 表达限流 |
Anthropic 的 529 是个很有意思的存在:它不是 5xx,但真实含义是"服务器过载,请稍后重试"。如果你按标准 5xx 处理,它确实会被重试;但如果你只针对 429 重试,529 就会漏掉,表现为偶发性的"上游失败"。我后来把所有厂商错误统一映射成几类异常:RateLimitError、AuthError、InvalidRequestError、OverloadedError、TimeoutError、StreamError。适配器的raise_error方法负责翻译,业务代码只认这几类,厂商怎么变都不影响上层。
4.2 一套重试策略能不能通吃?能,但要注意细节
重试策略我用的是指数退避加随机抖动。核心代码如下:
import random import time def request_with_retry(send_once, max_retries=4): for attempt in range(max_retries): try: return send_once() except RateLimitError as e: # 优先用服务端给的 Retry-After,没有再用指数退避 delay = getattr(e, "retry_after", None) if not delay: delay = 2 ** attempt + random.uniform(0, 1) time.sleep(min(delay, 30)) except OverloadedError: # 529 这类服务过载,短退避多试一两次 time.sleep(attempt + random.uniform(0, 0.5)) except (TimeoutError, ConnectionError): time.sleep(2 ** attempt + random.uniform(0, 1)) raise RuntimeError("after max retries")三个细节:
Retry-After优先,但必须设上限。有次某家返回 60 秒 Retry-After,如果照做,用户得干等一分钟,还不如把这个请求降级到备用模型。- 抖动必须加,否则多个请求同时失败后同时重试,直接把自己的限流额度打爆。
- 重试次数别太多。超过 4 次对用户体验已经是灾难,不如直接走降级路由。
4.3 流中断、超时和半截响应
流式请求的超时处理和普通请求不一样。普通请求可以用 connect/read 超时兜底,流式请求一旦连接建立,可能在几十秒内没有数据——这不是卡死,是模型在思考。所以要给流式单独设 idle 超时,比如 30 秒没收到任何字节就断开重连。Claude 的: ping心跳在这里反而是好事,它让你能区分"连接活着但没内容"和"连接死了"。
另一个很现实的问题是半截响应。流式过程中如果网络断了,客户端已经展示了一半文本,服务端后端任务已经消耗了 tokens。我的处理方式是:把半截内容标记为interrupted存下来,重试时在 prompt 里附加"请从刚才中断的地方继续",而不是整段重新生成。这样虽然多花点 tokens,但用户体验不至于从一段完整回答变成一段开头。
5. 把碎片化彻底"藏起来":配置化模型路由与自动降级
适配层解决的是"一个模型怎么被稳定调用"的问题,但实际业务里还有"多个模型怎么被编排"的问题。很多团队做完统一适配层就以为结束了,结果上游模型挂了一个,业务照样断。我后来加了一层配置化路由,效果非常明显。
5.1 用 YAML 描述整个模型服务池
配置化的核心思路是:所有模型接入信息都写在配置文件里,而不是散落在代码中。我当时用了一份 YAML:
model_pool: openai_prod: provider: openai model: gpt-4o base_url: https://api.openai.com/v1 api_key_env: OPENAI_API_KEY claude_prod: provider: anthropic model: claude-3-5-sonnet-20241022 api_key_env: ANTHROPIC_API_KEY gemini_prod: provider: gemini model: gemini-1.5-pro api_key_env: GEMINI_API_KEY qwen_aliyun: provider: openai base_url: https://dashscope.aliyuncs.com/compatible-mode/v1 model: qwen-plus api_key_env: DASHSCOPE_API_KEY route_rules: - name: production_tool_call when: {env: prod, need_tool: true} primary: claude_prod fallback: [openai_prod, qwen_aliyun] - name: canary when: {env: canary} primary: qwen_aliyun weight: 0.1注意我故意把 qwen 这类"兼容接口"也放进去了,因为它们虽然不是 OpenAI 官方,但 adapter 是同一个,只是 base_url 不同。接入一个新模型,很多时候就是往 YAML 里加一段配置,然后在 adapter 目录里确认有没有对应 provider。这份配置在启动时校验,环境变量缺失直接报错,避免上线后才发现密钥没配。
5.2 路由策略:灰度、按比例、按能力、按成本
路由不只是"主模型挂了换备用",实际生产里我用的规则大概有这几种:
- 失败降级:主模型抛
OverloadedError或RateLimitError,按 fallback 列表依次尝试,每个备用模型独立记失败次数,连续失败自动摘除。 - 灰度放量:新模型先在 canary 环境按 10% 流量灰度,观察 p95 延迟和生成质量,再逐步加大比例。
- 成本控制:当某供应商日成本到了阈值,自动把低优先级请求切到便宜模型。token 统计来自统一 usage 字段,做起来不费劲。
- 能力路由:请求里带 tools 时走支持函数调用且格式稳定的模型,纯聊天请求走性价比更高的模型。
降级这里有个前提:多个备用模型要真的"能力等价"。不是所有模型都适合处理同一类任务,比如让一个不支持函数调用的模型去处理工具类请求,降级等于降智。所以配置里要带能力标记,路由时先过滤再排序。
5.3 观测比路由更重要
路由做得好不好,全靠观测数据说话。我在适配层统一打了三类日志:厂商原始回包、归一化后的响应、路由决策记录。指标上重点看每个 provider 的成功率、p95 延迟、每千 token 成本、平均响应长度。这些数据一出来,很多问题会自己暴露:
- 某模型 p95 延迟突然翻倍,可能是厂商侧在调参;
- 某模型平均输出长度明显短于其他家,说明不是限流问题而是生成风格问题;
- 某模型 429 占比高,可能是你的 token 消耗量和共享配额不匹配。
没有观测的降级是盲降,你不知道自己在牺牲多少质量换取稳定性。
6. 要不要直接用现成网关?我的取舍与遗留问题
看到这里你可能会问:市面上不是有 LiteLLM、one-api 这类现成网关吗,为什么还要自己写适配层?我确实试过现成方案,下面说下我的取舍,以及这套方案跑了一阵之后仍然没解决干净的边界问题。
6.1 现成网关 vs 自建薄适配层
先给结论:内部工具、快速验证、对延迟不敏感的场景,直接用现成网关是合理的。LiteLLM 确实一次接入几十家模型,而且对外暴露 OpenAI 兼容格式,省掉大量初期工作。one-api 这类工具在国内生态里也很成熟,带密钥管理和额度控制,适合做团队内部的中转层。
我最后放弃现成网关,不是因为它们不好,而是因为我们的业务对两件事有硬要求:
- 流式处理的时序可控。我们需要在流式过程中插入业务逻辑(例如敏感词过滤、关键词高亮、引用标注),现成网关的 OpenAI 兼容输出会吞掉一些厂商事件细节,导致部分功能做不了。
- 深度排障能力。厂商回包的原始细节对排查异常很重要,现成网关往往只暴露归一化字段,遇到诡异问题很难定位是网关的问题还是厂商的问题。
如果你也有类似的定制需求,自建薄适配层是值得的。但要注意控制规模,我见过有人把适配层写得比业务代码还复杂,最后没人敢动——那还不如用现成网关。
6.2 仍然没完全解决的边界问题
有几个问题我至今没有完美答案,写出来供大家参考:
Reasoning 模型的输出换算。OpenAI 的 o 系列带 reasoning_content,Claude 的 thinking 块、Gemini 的 thought 部分,统一契约里没有位置放这些推理过程。我的方案是保留 raw 透传,业务层决定展示还是忽略——但对于"流式输出中先出现推理后出现回答"这类行为,各家的时序完全不同,产品 UI 很难做到一致。
JSON Mode 和结构化输出。各家都有结构化输出能力,但参数完全不同:OpenAI 有 response_format,Gemini 要设置 responseMimeType,Claude 习惯用 tool 强制约束。如果我强行统一,等于把各家最灵活的配置都做死。目前的妥协是:基础文本生成走统一参数,结构化输出走适配器专属的extra_params透传。
多模态输入。图像输入在 OpenAI 是 image_url,在 Claude 是 source block,在 Gemini 是 inlineData,格式差异巨大,而且不同模型接受的分辨率和格式也不一样。我一开始没有把多模态纳入统一契约,现在如果要做,大概率要新增一个 media 抽象层,而不是塞进现有 messages 里。
上下文缓存。Anthropic 有显式 cache_control,OpenAI 是自动缓存,Gemini 也有 prompt caching,但计费方式完全不一致。统一层能统计 token,但没法统计各家缓存命中带来的费用差异。我们目前是分 provider 记账,不做缓存层的统一。
6.3 踩坑后养成的几个习惯
最后分享几个我在做这套东西时沉淀下来的习惯,谈不上多高明,但确实帮我少走了很多弯路:
第一,每个适配器必须带一份真实回包 fixture。没有 fixture 的 parse 函数就是裸奔,厂商悄悄改字段你连个报警的都靠运气。
第二,CI 里跑一轮"人造故障"测试。用一个 mock server 模拟 500、429、超时和半截流,看看路由和重试会不会把请求打挂。这个测试我用脚本模拟 500 并发打了一次,当场发现备用模型切换后上下文没有带上,差点上线事故。
第三,厂商版本要显式固定。Claude 头里的anthropic-version、Gemini 的v1beta路径,都是会变的,固定到你验证过的版本,升级前先跑一遍回归。
第四,新模型接入之后,至少人工跑 20 轮"暗测"。覆盖中文、代码、长文本、多轮、工具调用五种场景,对比输出质量和延迟。光看接口通不通就上线,迟早被线上用户教育。
接口碎片化不是能一劳永逸解决的问题,它更像一个需要持续维护的设计课题。模型厂商在演进,你的业务在演进,适配层也要跟着演进。只要还保留着"每个厂商只翻译一次、每个业务只对接一次"的原则,后面无论再接入多少家模型,这个摊子都不至于乱到哪里去。