如果你这几年一直在做大模型相关项目,大概率和我有过同样的感觉:每次新模型出来,先不是看效果,而是先看它的 API 长什么样;换成新供应商,就要把请求协议、鉴权头、返回结构全部再调一遍。业务代码里到处都是 if 分支,模块越来越难维护。我最近把几个项目收敛到一起,用一套统一的 SDK 去对接多家大模型服务,包括各家的在线 API 和本地部署的模型服务,业务调用方只认一套接口。这篇文章就完整记录一下这轮多模型统一 API 接入的思路、坑位和可用代码,适合正在做 Agent、评测平台、企业级 AI 中台,或者只想把“多家模型切换”这件事做干净的同学参考。
1. 为什么非要搞“一套 SDK”:碎片化才是最大成本
1.1 大模型 API 碎片化带来的真实成本
业务早期只接一家模型供应商时,大家普遍不会觉得 API 抽象有多重要。反正就是拿现成 SDK,填 key,调chat,拿回字符串。等产品需要第二家、第三家模型的时候,问题就来了。每一家的鉴权头不一样,有的用Authorization: Bearer,有的把 key 直接放进路径参数里;消息结构也不一致,OpenAI 风格把 system 消息和普通消息放在同一个messages数组里,Anthropic 风格则强制要求把 system 单独抽出来,而且 base_url 的路径也不一样,有的是/chat/completions,有的是/messages。
这还只是协议差异,真正隐形的是业务代码里散落的适配逻辑。比如项目经理说“某个任务效果不好,换一个更便宜的模型试试”,你可能需要改动调用层、修改流式解析、重新处理工具调用,工作量大到让人怀疑人生。我在实际项目里见过很多“屎山”代码,其实就是模型供应商快速叠加导致的结果——每个接口都有自己的一套字段,平铺在服务里,最终没人敢动。碎片化的本质不是技术难度,而是多供应商协议不一致带来的持续维护成本。一套稳定统一的 SDK,能让这些差异收敛到“一层适配器”里,业务只面向自己的稳定协议编程,这是问题的核心解法。
1.2 统一抽象的核心思想:把变化点锁在适配器里
很多人觉得统一 SDK 就是做个“万能转发层”,把请求发给哪家模型就返回哪家格式,那就误解了。真正要做的不是把差异抹平,而是把差异关在笼子里。我采用的是经典的适配器模式:对外暴露稳定接口chat、text_embedding、list_models,对内为每个供应商实现一个 adapter。业务层永远不直接感知具体模型厂商,只关心模型别名、消息列表、参数和回调。
这样做的好处是,换模型时业务代码基本不动,改的是注册表和 adapter。如果你采用了模型别名机制,比如定义一个llm-fast别名指向某个具体模型的入口,之后想换成另一家模型,只需要调整注册中心和 provider 的映射,而不是改几十处调用点。同时,统一 SDK 内部可以提供 token 计数、错误分类、限流重试、流式输出归一化等公共能力,这些功能一旦在每个模型封装层各写一遍,不仅浪费,还非常容易各自为战,排查问题时你都不知道谁处理超时谁没有。只用一层抽象,代码量能减少 30% 以上,关键问题是架构清晰很多,后续加新供应商的成本会从几天压缩到几小时。
1.3 判断一下:你现在到底该不该自己写统一 SDK?
也不是所有场景都需要自己造一套轮子。如果业务只用固定一家在线模型,且确定未来不会切走,直接用官方 SDK 更省事;如果团队规模很小,一个产品只做一个对模型格式不敏感的简单问答,自己封装反而增加间接层。我建议出现以下任何一个信号时再动手:第一,代码里出现if provider == "xxx",并且超过三处;第二,同一个业务逻辑需要对比两个以上模型的效果;第三,Agent 场景里需要让工具调用在不同模型间无缝切换;第四,需要使用模型别名做降级和流量切换。满足两条以上,统一 SDK 的收益就已经大于成本了。
2. 统一 API 层的核心设计:先定义协议再写实现
2.1 统一请求与响应模型怎么定义最合适
我见过不少人第一步就去找各家 SDK 的官方 Python/Node 包,然后试图把它们的类型强行统一到一起,结果被各家的历史包袱拖死。更稳妥的做法是,先定义自己的核心数据模型,再让每家 adapter 把自己的请求转换成核心模型。
统一数据模型其实不需要太多字段,消息是核心。我实际使用的最小集合包括:角色role,内容content,可选的消息名name、工具调用信息tool_calls、工具结果对应的 IDtool_call_id。为了兼容多模态输入,content不能只定义为字符串,还要支持内容块数组。响应侧则统一为:模型返回的主文本content、工具调用数组、供应商名、模型名,以及 token 用量字典。有了这套模型,上层 Agent 代码不需要知道各家协议差异,下面我给出一个简化版的 Python 结构,实际工程可以再补充业务字段。
from dataclasses import dataclass, field from typing import Optional @dataclass class UnifiedMessage: role: str # system / user / assistant / tool content: str | list[dict] # 字符串或多模态内容块数组 name: Optional[str] = None tool_call_id: Optional[str] = None tool_calls: Optional[list] = None @dataclass class UnifiedRequest: messages: list[UnifiedMessage] temperature: Optional[float] = None max_tokens: Optional[int] = None tools: Optional[list] = None stream: bool = False @dataclass class UnifiedResponse: content: str = "" tool_calls: list = field(default_factory=list) provider: str = "" model: str = "" usage: dict = field(default_factory=dict)这个模型看起来很基础,但它已经足够支撑绝大多数聊天、Agent、评测和批量生成场景。一些额外参数例如stop、top_p、response_format,如果不想提前建模,可以在请求里放一个extra: dict透传字段,让有需求的调用方直接使用。我的经验是,不要一开始就把所有参数都塞进模型,先支持 80% 场景,其他参数保留透传通道,后续发现有高频需求再升级为正式字段,这样模型不会越来越臃肿。
2.2 内部统一协议选型:为什么倾向采用 OpenAI 兼容风格
在设计内部统一协议时,有一个关键选择:要不要直接用 OpenAI 的 Message 结构作为基准。我的答案是要,但需要做一层“宽容的兼容”。原因是 OpenAI 的消息格式已经事实成为大模型生态的通用语言,很多开源模型框架和部署工具都提供兼容端,甚至许多非 OpenAI 系供应商也直接暴露了同款协议。业务团队大多习惯这种写法,用相同的内部结构能降低学习成本;统一的 HTTP 还是内部 SDK 封装,都更容易被理解。
采用公有协议作为基准也有代价,比如 Claude 系列的消息结构会强制把 system 拆出来,Gemini 的多模态结构有自己独特的内容块字段,只用 OpenAI 协议直接请求会失败。所以我们的内部模型允许system角色存在,但 adapter 转换时要把 system 内容抽出来放到 Anthropic 的 system 参数里或 Gemini 的系统指令字段里。内部统一不是让内部协议绑架所有厂商,而是提供一个稳定的“中转语言”,各家 adapter 负责完成双向翻译。因此,我把这种设计称为“外部宽容,内部统一”,宁可让 adapter 复杂一点,也不让上层业务复杂。
2.3 模型注册中心:用别名代替硬编码供应商加模型名
仅仅统一消息结构还不够,调用方不应该知道“我要调某个供应商的某型号”,而应该告诉 SDK“我要调一个擅长摘要的模型”或“我要用默认最强模型”。这就是模型注册中心要做的事情。注册中心里有三个核心概念:模型别名、供应商配置和 fallback 链。比如把fast-chat映射到一个便宜的模型,把power-reasoning映射到一个推理能力更强的模型,当主力供应商超时或返回限流时,SDK 自动按 fallback 顺序切到备选模型。
我常用 YAML 或数据库表来维护这个注册关系,下面是一个精简的配置片段,真实场景可以加上健康状态、发布批次和权重。
providers: - name: openai_compatible_a driver: openai_compatible base_url: https://api.example-a.com/v1 - name: provider_b driver: anthropic base_url: https://api.example-b.com/v1 models: - alias: fast-chat provider: openai_compatible_a model_NAME: model-fast fallback: - provider: provider_b model_NAME: model-small - alias: strong-chat provider: provider_b model_NAME: model-sonnet fallback: - provider: openai_compatible_a model_NAME: model-max注册中心的引入,让“多模型”真正成为可配置项。上线一个新模型或切换主备流量,只要改配置和编排,不需要重新发布业务代码。对线上运行的系统来说,这带来的改变是灾难性的,尤其你想做多模型对比评测时,同一套数据集只要改model_alias就能跑完所有候选模型,省下的时间和人力非常可观。
3. 手写 Provider 适配器:从同一种请求到不同风格 API
3.1 Provider 接口与错误分类
定义完内部模型后,接下来是最核心的适配器层。每种厂商风格一个类,统一继承抽象基类,暴露同一个chat方法。不过这个chat要同时支持同步和流式,所以返回值设计需要小心。我的做法是:非流式直接返回UnifiedResponse,流式返回一个异步生成器,生成已经切分好的文本增量。上层调用者只需要知道,如果stream=True就 for await 文本增量,否则等一个完整响应。
错误处理也必须前置统一。各家模型服务的错误返回格式差异极大,有的返回 JSON 里的error.message,有的返回纯文本,有的返回一个带status的嵌套结构。我会在底层统一转换成自己的异常体系:AuthenticationError、RateLimitError、ContextLengthError、OverloadedError、InvalidRequestError和BadGatewayError。只有先做了错误归一化,才能在上层实现统一的重试和降级策略,否则每一家模型的异常都要单独 catch,那又是一片混乱。
class BaseProvider(ABC): config: ProviderConfig @abstractmethod async def chat( self, req: UnifiedRequest, stream: bool = False, ): """返回 UnifiedResponse 或异步文本增量生成器""" pass3.2 OpenAI 兼容 Provider 实现细节
OpenAI 兼容风格的适配是最好写的,因为双方结构基本同构。但仍然有几个位置要小心:URL 路径前缀各厂商不同,有的直接https://api.xxx.com/v1/chat/completions,有的要自己拼接chat/completions;鉴权头绝大多数是Authorization: Bearer <key>,但个别网关要求api-key头;返回结构里,非流式响应是choices[0].message.content和choices[0].message.tool_calls,流式响应则要监听choices[0].delta.content和choices[0].delta.tool_calls。
我把 OpenAI 兼容 Provider 的 HTTP 层直接建立在httpx.AsyncClient上,而不是引入官方 openai 库。原因有两个:一是很多兼容端并不过完整 OpenAI SDK 的测试,直接裸 HTTP 更可控;二是后续要对请求做重试、超时和自定义日志时,官方客户端的钩子不一定顺手。实现时,payload 只需做字段过滤,把请求里的None值删掉,因为不少服务端对显式传入的null参数报错,这一点很容易被忽视。
class OpenAICompatibleProvider(BaseProvider): def __init__(self, config: ProviderConfig): self.config = config self._client = httpx.AsyncClient(timeout=config.timeout) async def chat(self, req: UnifiedRequest, stream: bool = False): payload = self._build_payload(req, stream) if stream: return self._stream_chat(payload) resp = await self._client.post( f"{self.config.base_url}/chat/completions", headers={"Authorization": f"Bearer {self.config.api_key}"}, json=payload, ) data = self._parse_response(resp) return self._to_unified_response(data)需要注意max_tokens字段在某些新模型里被改名成max_completion_tokens,而且发送旧的max_tokens参数反而会报不支持。这类兼容细节无法纯粹靠规范文档解决,准确做法是在 Provider 的配置里增加一个request_field_map,允许对单个供应商做参数名映射,这样就不用为每一种兼容端单独 fork 一个类。
3.3 Anthropic 风格与 Gemini 风格适配映射
真正有挑战的是 Anthropic 消息结构和 Google Gemini 结构。Anthropic 风格的核心难点有三个:system 消息要单独提取,不能放到 messages 里;assistant 消息如果包含工具调用,要把工具调用改写成tool_use内容块;上一轮工具执行结果则要放入tool_result内容块,而不是用 OpenAI 风格里的role=tool消息。另一个重要差别是 max_tokens 在 Anthropic API 是必填字段,有的调用方没有设置,转换时如果直接不传就会收到校验错误。
我通常这样转换:先遍历请求 messages,把 system 角色拼接成字符串;遇到 tool 角色时转换成tool_resultblock;遇到 assistant 且带tool_calls时转换成tool_useblock。Gemini 的转换逻辑也类似,只是它的内容块结构叫functionCall和方法参数functionResponse,且角色和消息的组织方式也不相同。这个过程如果没有提前统一消息模型,会在业务层越来越混乱,因为所有转换逻辑都散在调用点。
def convert_to_anthropic(req: UnifiedRequest) -> dict: system_text = [] messages = [] for m in req.messages: if m.role == "system": system_text.append(m.content) elif m.role == "tool": messages.append({ "role": "user", "content": [{ "type": "tool_result", "tool_use_id": m.tool_call_id, "content": m.content, }], }) elif m.role == "assistant": content_blocks = [] if m.content: content_blocks.append({"type": "text", "text": m.content}) for tc in m.tool_calls or []: content_blocks.append({ "type": "tool_use", "id": tc["id"], "name": tc["function"]["name"], "input": json.loads(tc["function"]["arguments"] or "{}"), }) messages.append({"role": "assistant", "content": content_blocks}) else: messages.append({"role": m.role, "content": [{"type": "text", "text": m.content}]}) return { "model": req.model_name, "system": "\n".join(system_text), "messages": messages, "max_tokens": req.max_tokens or 4096, }在具体落地时,不要过度追求“把所有内容都转换成完全相同的文本”。比如有的模型支持图片输入,有的只支持文本,统一 SDK 要做的是格式转换,而不是强行抹平能力边界。当目标模型不支持某类内容时,应该在请求前校验并报错,而不是等模型返回 400 后再让人排查。
3.4 流式输出也要统一成一个 AsyncGenerator
多模型接入最麻烦的具体细节之一是流式协议。OpenAI 兼容端通常返回text/event-stream,每个事件以data:开头,结束标志是data: [DONE];Anthropic 则是先发event:类型行,再发data:内容块;Gemini 又有自己的candidates结构。如果上层每个调用点都写一套流式解析,那代码会非常难看,而且很容易漏处理特殊结束事件。
统一的思路是让 Provider 内部解析完各自 SSE,对外只 yield 纯文本增量。拿 OpenAI 兼容端举例,就是遍历aiter_lines(),过滤以data:开头的行,去掉前缀后解析 JSON,从choices[0].delta.content或choices[0].delta.tool_calls里取内容。Anthropic 端则要判断event: content_block_delta之后的数据行。只要把流式解析锁在 adapter 内,上层 Agent 拿到的是一个统一的AsyncIterator[str],想做打字机效果、Token 统计还是异常中断都会方便得多。
async def _stream_chat(self, payload: dict): async with self._client.stream( "POST", f"{self.config.base_url}/chat/completions", headers={"Authorization": f"Bearer {self.config.api_key}"}, json=payload, ) as resp: if resp.status_code != 200: body = await resp.aread() raise self._raise_error(resp.status_code, body) async for line in resp.aiter_lines(): if not line.startswith("data:"): continue data = line[5:].strip() if not data or data == "[DONE]": continue obj = json.loads(data) delta = obj["choices"][0].get("delta", {}) if delta.get("content"): yield delta["content"] if delta.get("tool_calls"): yield {"tool_calls": delta["tool_calls"]}唯一要提醒的是“空行”和“注释行”处理。有些兼容端会在 SSE 流里发: keep-alive之类的注释,如果直接按data:前缀硬过滤就走不到完整的解析路径,所以我的代码里先判断是否startswith("data:")再处理,其他行全部忽略。
4. 接入层实践:密钥、重试、限流、可观测性,一个不能少
4.1 API Key 管理与多租户隔离
统一 SDK 如果只是内部使用,API key 直接放在环境变量里就可以。一旦 SDK 要提供给多个业务团队当作内部组件使用,key 的管理就必须谨慎。最忌讳的是把各家供应商的 key 直接暴露给前端,或者上传到代码仓库。实践中建议的架构是:SDK 里只接受一个来自配置中心或密钥管理服务的 token,服务端负责把 token 映射到真正的大模型供应商 key,并在后端完成所有转发。
如果要在同一个 SDK 里支持多个业务方,还需要做多租户隔离。比如业务 A 调fast-chat模型,业务 B 也调fast-chat,但两边使用的 key 配额不同,不能共用一个后端凭据。我的办法是给 ProviderConfig 增加一层租户解析,把每个请求上下文里的租户 ID 映射到对应的密钥版本,并支持 key 轮换。这样即使某个业务方把 key 打爆,也不会连累其他业务方。
4.2 超时与重试策略怎么设计才不踩坑
统一 SDK 里最容易出问题的就是无脑重试。我见过不少同学写了retry=3,结果下游模型服务持续 503 时依然傻傻重试,把限流打得更严重;更危险的是写操作类请求,比如一个工具调用需要执行某个外部动作,第一次请求实际已经发出去了,只是响应超时,这时盲目重试就会产生重复副作用。因此重试必须配合错误类型和请求幂等来判断。
我的推荐策略是:只有网络连接错误、5xx、429 限流这三类情况才重试;4xx 业务错误不重试,因为重试大概率还是同样结果。重试间隔采用指数退避加抖动,避免多个请求同一时刻大量重试造成“重试风暴”。比如第一次失败后等 1 秒重试,第二次等待 2 秒,第三次等待 4 秒,每次加一个随机 0 到 0.5 秒的偏移。下面是一段简单的退避逻辑,在实际项目里可以直接扩展。
async def request_with_retry(func, max_retries=3): for attempt in range(max_retries + 1): try: return await func() except (RateLimitError, OverloadedError, BadGatewayError) as exc: if attempt == max_retries: raise delay = 2 ** attempt + random.random() * 0.5 logger.warning("model request failed, retrying in %.2fs: %s", delay, exc) await asyncio.sleep(delay)还有一个容易忽略的点:重试时要把整个provider.chat(...)调用包进去,而不能只重试底层 HTTP 请求。因为在流式场景下,连接可能在中间断开,此时重试只能从 provider 层重新发起完整请求;底层重试无法恢复已经消费了一半的 SSE 流。
4.3 并发控制、限流与排队
不同模型服务的并发额度差别很大,统一 SDK 不能只把请求打出去就不管。最好在 SDK 内内置一个基于asyncio.Semaphore的并发闸门,根据每个 Provider 的max_concurrency配置来限制同时发出的请求数。同时做一些每秒请求数控制,防止跑评测或批量任务时瞬间触发模型服务限流。
实际测试时你会发现,很多模型的限流是分不同维度计算的,有的限制每秒请求数,有的限制每分钟 token 数,还有的限制并发连接数。SDK 至少要能处理“并发限制”和“请求频率限制”。如果已经有网关层,可以只在 SDK 里配置大致并发即可;如果 SDK 直连供应商,建议在重试前先做一次本地限流,比如令牌桶,而不是等 429 回来再被动重试。虽然这只是小细节,但在大批量调用场景下可以明显降低错误率。
4.4 可观测性:结构化日志与调用追踪
统一 SDK 带来的一个额外好处是,可以非常自然地在接入层统一埋点。每个chat调用都应该记录:模型别名、实际请求的供应商和模型名、请求是否流式、等待耗时、首 token 延迟、总耗时、输入 token、输出 token、错误类型。这些信息用结构化 JSON 输出到日志或 trace 平台,出了问题就不是靠猜了。
我用一个简单的装饰器或上下文管理器实现调用追踪,记录进入和离开时的时间差。这里需要注意,token 用量在非流式响应里由供应商返回,但流式响应通常在结束时才有usage字段,有的兼容端甚至不返回,需要自己估算。我的做法是流式场景不依赖服务端 usage,而是用 tokenizer 对最终文本做本地估算,至少能拿到数量级,方便监控趋势。多模型评测、成本核算这类场景,建议还是以非流式返回的 usage 为准,更准确。
5. 模型行为差异与兼容性避坑
5.1 参数方言:temperature、top_p、max_tokens 的隐藏坑
很多人在接入多家模型时,直觉认为字段名字一样,含义就完全相同,实际远没有那么简单。比如temperature,OpenAI 文档里说它是控制随机性的采样温度,但一些推理模型直接把temperature参数忽略甚至报错。再比如top_p,按照 OpenAI 的建议是不要和 temperature 同时调整,但其他模型没有这种限制,你可以同时设置两层采样逻辑。
max_tokens是最大的坑点。早期模型基本都叫max_tokens,新出的推理模型却要求使用max_completion_tokens,如果你发送旧的字段名,不少接口会返回显式错误;反过来,某些兼容旧模型的网关只认识max_tokens,你发max_completion_tokens它也会忽略。所以统一 SDK 在 Provider 内部维护一个参数名映射表是很有必要的。我建议在最底层用“内部标准参数名”,到具体 Provider 时再转换为它们要求的名字,这样核心代码永远不需要感知这些差异。
5.2 function calling / tool calling 格式差异
Agent 项目没法避免 tool calling,而各家 tool calling 的实现差异比聊天接口大一个数量级。OpenAI 风格返回消息里的tool_calls数组,每个元素有id、type、function.name、function.arguments,其中arguments是一个 JSON 字符串;Anthropic 风格则把工具调用作为 assistant 消息内容块里的tool_use对象,参数直接是 JSON 对象;Gemini 的functionCall结构又是另一套 key。
统一 SDK 在收到各家响应后,应该统一转换成一种内部工具调用结构,比如ToolCall(id, name, arguments_dict)。业务层拿到这个结构后可以直接执行函数,而不需要关心底层的 arguments 是 JSON 字符串还是对象。流式场景更加麻烦,OpenAI 兼容端在流式返回工具调用时,会把delta.tool_calls按片段返回,同一个工具调用的index相同,但name和arguments会被拆到多帧里。SDK 需要自己拼接这些碎片,否则上层 Agent 会拿到残缺的 JSON。
我踩过一次印象很深的坑:某个兼容端在流式返回 tool_calls 前先发了一段普通的content字符串,内容还是空文本,我的旧解析代码只处理 content,导致 Agent 完全没感知到工具调用。后来改成不依赖 content,每次 delta 里同时检查 content 和 tool_calls,才解决问题。
5.3 视觉输入、JSON 模式、推理模型的兼容处理
现在的多模态模型输入格式越来越复杂。OpenAI 风格的图片消息是content: [{type: "image_url", image_url: {url: "..."}}],Anthropic 风格是{type: "image", source: {...}},Gemini 则是inline_data或file_data对象。如果统一 SDK 只支持字符串 content,就无法承载图片输入,所以内部UnifiedRequest一定要允许 content 是数组。在 adatper 转换时,需要判断 content block 类型,并按目标 API 的格式重新组装。
JSON 模式也有差异。有的模型支持response_format={type: "json_object"},有的要求你把“请输出 JSON”直接写进 system prompt,还有的使用response_mime_type=application/json。统一 SDK 至少要做到:业务层传入一个结构化的response_format,由 adapter 根据供应商能力决定怎么发。至于新出的推理模型,它们经常不支持传统temperature,内部默认会做自我纠错,不能简单把 temperature 设为 0 来求稳,“把结果调成确定格式”往往是徒劳的,这一点要提前和业务方沟通清楚。
5.4 Token 计算与上下文溢出处理
各家模型的上下文窗口差异很大,tokenizer 也不一样。统一 SDK 需要能够把“字符长度”粗略转换为“token 数量”,再结合注册表里的max_context_tokens判断是否会溢出。系统提示越长,可用上下文越小,有些任务明明内容不多却报超长,多半是消息拼接了多次历史记录,没有做滚动裁剪。
当某个模型返回ContextLengthError时,更好的方案不是用同一个请求重试,而是降级到上下文窗口更大的模型,或自动压缩历史消息。这里的“压缩”需要策略,不能简单截断尾部。我常用的做法是:先去掉最早的非 system 消息;如果还不够,把中间的用户问题压缩成摘要;最后才丢弃工具调用记录。要注意的是,不同模型对相同的文本 token 估计差异可能在 10% 到 20% 之间,如果是接近窗口极限的请求,宁可提前估算并告警,也不要等接口报错。
6. 常见问题与排查技巧实录
6.1 问题速查表
| 现象 | 常见原因 | 解决建议 |
|---|---|---|
| 返回 401 / invalid api key | 配置的密钥错误,或 key 没带对 header | 检查 Provider 鉴权头规则;不要同时使用多个 key |
| 返回 400 model not found | 模型名写错,或该账号没有该模型权限 | 先用供应商控制台查看可用模型列表 |
| 返回 400 context length exceeded | 消息长度超过上下文窗口 | 压缩历史消息或使用窗口更大的模型 fallback |
| 返回 429 rate limit | 并发过高或额度不足 | 本地加信号量限流,按指数退避重试 |
| 返回 503 overloaded | 供应商侧负载高 | 延迟重试,并根据 fallback 链路切换模型 |
| 流式中途断流 | 客户端超时太短,或 SSE 解析错误 | 调大 read timeout;检查是否忽略data: [DONE] |
| 工具调用参数解析失败 | 流式 tool_calls 分片未拼接完整 | 按index聚合流式帧,再统一解析 JSON |
| 同一请求不同模型返回稳定性差异大 | 参数语义不同或模型版本不同 | 设置模型别名,锁定版本,避免无感切换 |
6.2 我实际踩过的坑与应对
先说说“重复扣费”的问题。第一次实现统一重试时,我只在 503 状态下重试底层 HTTP POST,后来排查账单发现,供应商其实已经完成了生成,只是响应在网络上丢了,重试导致同一笔请求被计费两次。现在我在非幂等类请求的重试逻辑里加了“重复请求号”机制,通过供应商支持的request_id或者自定义 header 传入,服务端能做去重的最好;服务端不支持时,只能降级为重试次数设为 1,并在日志里打上高优先级告警让人工判断。这虽然听起来不完美,但至少不会让费用失控。
再说一个流式解析的 bug。我们当时把一个 OpenAI 兼容 Provider 的 base_url 配置成/v1,然后在代码里拼接/chat/completions,能正常工作;后来接入另一个兼容服务时,对方的 base_url 已经写到/v1/chat/completions,我再拼一次就变成了双路径,请求一直 404。排查了整整半小时才发现是路径拼接问题。这个教训让我在适配器里加了一个统一的 URL 拼接函数:如果 base_url 以/chat/completions结尾就直接使用,否则再拼接。类似的还有 API 版本路径。统一 SDK 对 base_url 的解析必须很严格,否则多供应商很容易互相影响。
最后聊聊调试手段。多模型接入时,不同厂商的网关错误信息经常是“半真半假”的,比如有的返回 400 说 model not found,实际原因却是 messages 里包含了一个空的 system 消息。我会在 adapter 里加一个 debug 开关,打开后把最终发给供应商的 payload 结构完整打到日志里,这一步能省掉大量来回试错的时间。对业务方来说,他们只看得到统一后的内容,但定位问题时还是要拿到原始请求体,如果 SDK 不记录原始往返信息,会很被动。
有一件事我觉得比所有框架技巧都重要:保持模型别名稳定。不要在生产代码里直接写死“某个具体的模型版本号”,而应该通过别名指向它。比如内部默认的strong-chat这周指向某个模型,下周供应商发布了更强的新版并经过评测后,你再把别名切到新版本,业务方无需改动代码。这个习惯让我们的模型升级变得很平滑,也避免了线上悄悄变更模型而引发不可控问题。做多模型统一接入,真正的价值不在省那几行请求代码,而在于让团队有了随时切换、比较和降级的能力,这套能力的维护成本会随着模型数量的增长越来越低,而收益会越来越明显。