从零接入多模型 API一个统一接口的实战指南年初我们团队接到一个需求给内部工具接上多家大模型服务商的 API让业务方可以根据模型能力、价格、稳定性自由切换提示词背后的引擎。一开始大家都觉得简单——不就是调 HTTP 接口吗结果真做起来才发现每家服务的鉴权方式不同、请求体结构不同、流式输出的格式不同、错误码含义也不同光是适配层就写了上千行重复代码。后来我们痛定思痛花了两周时间从零设计了一套统一接口把多家模型的接入成本压到了“新增一家只需写一个适配器”。这篇文章就把整套方案的思考过程和落地细节完整写出来适合正打算接入多个模型 API、或者已经在维护一堆烟囱式调用的团队参考。1. 为什么需要统一接口三家厂商标配的“地狱式差异”1.1 不统一的代价到底有多大先说一个真实场景。当时我们第一个版本是业务方直接各自调用模型 API代码散落在不同服务里。表面上看起来很快但实际维护起来非常痛苦鉴权方式各不相同。有的用 HTTP Header 里的静态密钥有的要用动态签名有的还要先换取临时凭证有效期只有几十分钟。请求体差异巨大。同样是“多轮对话”一家用的是messages数组带role字段另一家把历史对话拼成一个带分隔符的字符串还有一家要求每条消息都带上timestamp。响应结构完全不对齐。有的接口直接返回choices[0].message.content有的需要从outputs[0].text取还有的会把内容包在data.answer里。错误处理更是灾难。一家限流返回 429另一家限流返回 200 但响应体里写个code: 4021还有一家直接在消息里用自然语言提示“您的并发数超了”。刚开始只有一两家模型时这种差异还能靠“写 if 分支”硬扛。等接到第五六家每次新增一个供应商代码里就要多出三四处 if测试用例越来越长出问题却越来越难查。这时候我们意识到真正该做的不是继续适配而是要截断混乱的源头在上游定义一个稳定契约在每家模型前面加一个适配器把差异全部挡在业务层之外。1.2 统一接口应该覆盖哪些层设计统一接口时一个常见的错误是“什么都想统一”。比如有人想把所有模型的能力差异也抹平——这基本是做不到的。我们把统一接口的职责边界划分得很清楚统一的是“调用方式”包括请求入口、鉴权方式、参数结构、返回结构、错误码。不统一的是“模型能力”比如某个模型支持 200K 上下文另一个只支持 32K某个模型原生支持工具调用另一个不支持。这些差异通过能力检测和运行时降级处理而不是在接口层假装它们一样。另外一个关键是“按场景设计契约而不是按厂商设计”。我们当时面向业务方提供的不是“聊天”这一个接口而是按用途拆出来的几组能力文本对话、上下文补全、多轮摘要、分类打分、工具调用。每组能力都有自己的参数模型。这样业务方只关心“我要做什么”而不需要知道“背后是哪家模型”。2. 接口契约设计先把数据结构和边界定义清楚2.1 统一请求模型的设计思路写代码之前我们先用文档把请求契约定了下来。核心是四个字段组request_id每次请求的唯一 ID用于链路追踪和问题排查。messages统一的消息数组结构每一条包含rolesystem/user/assistant、content、可选的name和tool_call_id。params一组调用参数只放所有模型都支持或语义一致的字段例如temperature、max_tokens、top_p、stop、stream。extensions透传字段专门放某家模型独有的参数比如厂商 A 的thinking_mode、厂商 B 的seed。这个字段的设计很关键——它允许我们在不破坏统一契约的前提下保留各家特色能力。我们内部约定params里的字段必须经过适配器翻译而extensions里的字段原样透传。这样既照顾了通用性又不会因为厂商特性而把统一接口膨胀成一堆可选参数的泥潭。还有个容易被忽略的细节超时和重试策略不在请求模型里而是放在配置中心。因为不同模型服务商的响应速度差异很大有的 500ms 就能返回有的要 30 秒还在等首包。如果超时时间写死成同一个值要么弱的服务永远超时要么强的服务为了等慢的模型浪费资源。2.2 统一响应模型成功时和失败时都要稳定响应模型我们分了两个层次。成功的响应统一是{ request_id: req_abc123, usage: {prompt_tokens: 120, completion_tokens: 80, total_tokens: 200}, output: { content: 这是模型返回的文本, finish_reason: stop, tool_calls: [] } }而失败的响应统一是{ request_id: req_abc123, error: { code: UPSTREAM_RATE_LIMITED, message: 上游模型服务限流建议降低并发, retryable: true } }这里最值得强调的是error.retryable这个字段。不同厂商对“可重试错误”和“不可重试错误”的定义完全不同有的 429 重试有效有的 429 重试只会加重限流有的 401 可能是密钥过期不可重试有的 401 只是当前节点未同步需要换节点重试。如果让业务方自己判断要不要重试错误处理逻辑会满天飞。我们在适配器里就把每家错误翻译成“是否可重试”业务层只认这个布尔值。2.3 供应商适配器接口只暴露六个方法我们把每个模型服务商的适配器收敛成六个方法新接入厂家时只需要实现这六个方法class ModelAdapter(ABC): abstractmethod def check_auth(self, config: dict) - AuthResult: ... abstractmethod def prepare_request(self, req: UnifiedRequest) - Tuple[str, dict, dict]: ... abstractmethod def parse_response(self, raw: dict) - UnifiedResponse: ... abstractmethod def parse_stream_chunk(self, chunk: bytes) - StreamEvent: ... abstractmethod def parse_error(self, status_code: int, raw: dict) - UnifiedError: ... abstractmethod def count_tokens(self, messages: list) - int: ...prepare_request返回三元组URL、请求头、请求体parse_response和parse_stream_chunk负责把各家原始响应转成统一结构parse_error负责错误映射。这套接口一旦稳定新增供应商的工作量就从“改业务代码”变成了“写一个新适配器文件”而且可以独立测试。3. 从零搭建调用链路隐藏在网络细节里的那些坑3.1 项目结构与依赖选型我们最终选择的项目结构非常简单llm_gateway/ ├── adapters/ │ ├── base.py # 抽象基类 │ ├── vendor_a.py # 适配厂商A │ ├── vendor_b.py # 适配厂商B │ └── vendor_c.py # 适配厂商C ├── core/ │ ├── request.py # 统一请求模型 │ ├── response.py # 统一响应模型 │ ├── router.py # 路由与负载均衡 │ └── errors.py # 统一异常体系 ├── utils/ │ ├── retry_policy.py # 重试策略 │ └── logger.py # 结构化日志 └── tests/依赖选型上我们没有引入重量级框架就是一个httpx做异步 HTTP 客户端加上pydantic做数据模型校验。为什么选httpx而不是requests因为我们要同时支持流式响应和异步并发requests的流式处理在长连接场景下表现不够好而httpx原生支持 HTTP/2 和异步连接池管理也更透明。如果你用的是同步框架比如 Flask也别担心httpx可以跑在同步模式只是我们内部全部走异步。一个很容易被忽略的问题是 HTTP 客户端的连接池设置。多家厂商共用一个 Client 实例时如果连接池太小某个模型服务出现慢响应会把连接全部占满导致其他模型也跟着超时。我们在生产环境把连接池上限设为 200并且对每家厂商单独维护一个 Client 实例。这样厂商 A 的网络阻塞不会拖垮厂商 B 的请求。3.2 鉴权差异处理动态密钥和静态密钥并存鉴权是接入时最先遇到的坑。我们遇到的三种典型方式静态 Header 密钥最简单Authorization: Bearer xxx。动态签名需要按时间戳和随机数生成签名还要把签名放到特定 Header 里。临时凭证先调用一个换 token 的接口拿到凭证后再访问模型接口凭证过期后还要自动刷新。我们的适配器把鉴权过程封装成了三个阶段初始化阶段构建认证上下文、请求阶段注入请求头、失效处理阶段遇到 401 时刷新凭证并重试一次。这里的核心是“不要在每个请求里去碰密钥而是把密钥管理交给适配器的 session 层”。还有一个小细节很多模型的密钥是有权限范围的有的只能调用对话模型有的只能调用 embedding 模型。如果混用会出现“昨天还能用今天突然 403”的诡异问题。接入前一定要确认密钥的权限范围并且在不同环境生产/测试使用不同的密钥避免测试流量污染生产额度和监控数据。3.3 参数映射层模型特有参数的处理策略参数映射是统一接口里最容易设计过度的部分。我们的策略很简单常见的参数temperature、top_p、max_tokens、stop进入params由适配器负责翻译成各家的字段名不常见的参数一律放extensions原样透传。但这里有个例外情况需要我们特别注意max_tokens的边界值。不同厂商对max_tokens的理解有细微差别有的厂商传max_tokens4096表示最多输出 4096 个 token有的厂商把这个参数叫max_new_tokens并且“输出 token 数不包含提示词部分”。如果不做归一化用户设置同样的值在不同模型上得到的结果长度可能差异很大。我们的做法是在适配器里维护一个“参数边界表”记录每家模型参数的合法范围、别名、单位并在调用前做校验和自动纠偏。比如厂商 B 最大只支持 2048我们就自动把用户传的 4096 截断到 2048而不是报错——因为业务方关心的是“能生成多长”不是“API 收不收这个值”。还有一个我们踩过的坑stop参数。有的模型支持最多 4 个停止词有的支持 8 个如果把超过数量限制的停止词直接传过去某些服务会直接报错。解决方法是在适配器里做“停止词截断 日志警告”同时把完整停止词列表保存在请求上下文中以便排查生成结果时判断是否被意外截断。3.4 响应归一化的隐藏风险字段命名、空值与类型不一致响应归一化看似简单实际处处是坑。首先是字段命名差异同一个字段厂商 A 叫finish_reason厂商 B 叫stop_reason厂商 C 把它放在output.finish_status。这些映射在适配器里解决就好关键是映射表要独立维护不要和业务逻辑混在一起。更大的坑是“缺失字段”的处理。厂商 A 在非流式模式下一定会返回 token 用量统计厂商 B 默认不返回需要加参数usage: true厂商 C 在部分请求下不返回finish_reason。如果统一响应里一上来就取raw[usage][prompt_tokens]就会因为 KeyError 炸掉。我们的原则是“宁可给默认值不能抛异常”。def _safe_extract_usage(raw: dict) - dict: usage raw.get(usage) or raw.get(token_usage) or {} return { prompt_tokens: usage.get(prompt_tokens, 0), completion_tokens: usage.get(completion_tokens, 0), total_tokens: usage.get(total_tokens, 0), }另外一个不容易发现的坑是“内容为空但接口返回成功”。我们的适配器层约定了一个规则如果content字段是空字符串、None、或者内容里只有换行符且finish_reason不是tool_calls一律视为异常响应。因为这个场景通常意味着模型触发了输出安全过滤如果不显式处理业务方拿到的就是一个“自信的空响应”很容易误判为模型没有知识。4. 流式输出与工具调用最容易翻车的两部分4.1 流式响应的格式差异有的按行有的按块还有的不标准如果你只是做同步调用上面那些基本够用。但一旦要接流式输出复杂程度会直接上一个台阶。我们最初实现流式适配时发现三家厂商的流式格式竟然完全不同厂商 A标准 SSE按行分割每行是data: {json}结束标志是data: [DONE]。厂商 B自定义二进制分帧每帧 4 字节长度前缀 JSON 载荷。厂商 C支持 SSE但一次事件里可能包含多个增量而且增量 JSON 的 key 在不同事件里还不一样。如果你在 Gateway 层直接透传原始流下游每个业务方都得写一套解析逻辑那就等于把适配成本转移给了业务方。我们的方案是在 Gateway 内部把各家流式响应统一转成一种简单的增量事件格式{ type: delta, content: 这是本次增量, finish_reason: null }转成统一格式之后业务方只用一个循环就能处理所有模型的流式输出。为了把各家流式输出安全地转成统一格式我们在适配器内部维护了一个状态机每次读取到一个 chunk 就进入状态机处理输出对应的事件。这个状态机需要处理跨 chunk 的半截 JSON、半截行尾符、以及突然中断的流。4.2 状态机设计跨 chunk 拼包和意外中断我做流式适配时踩得最深的坑是“跨 chunk 的 JSON 截断”。SSE 按行传输但一次网络读取可能只拿到半行比如数据量大时一个几千字节的 JSON 被拆成三四段到达。如果简单按\n分割某些段会解析失败。最稳妥的做法是使用httpx的iter_lines()它在底层已经处理了行缓冲。如果用的是原生的iter_raw()就必须自己维护一个以\n结尾的 buffer。厂家 B 的“二进制分帧”更麻烦。它的每个数据帧是“4 字节长度 内容”所以解析逻辑很容易写成死循环或内存泄漏。我们的实现是async def parse_vendor_b_stream(stream): buffer b while True: chunk await stream.read(4096) if not chunk: break buffer chunk while len(buffer) 4: length int.from_bytes(buffer[:4], big) if len(buffer) 4 length: break package buffer[4:4 length] yield json.loads(package) buffer buffer[4 length:]这里最关键的是“外层 while 处理网络流内层 while 处理帧缓冲区”两者必须分开。否则一旦出现半帧数据会把整个连接卡死。流式中断也是必须处理的场景客户端断开、模型服务崩溃、网络抖动都可能导致流中断。我们在 Gateway 层做了一件事——当流意外中断时把已经累积的内容包装成一个partial响应返回同时记录日志。业务方可以决定是继续用部分结果还是用之前的完整结果做兜底。这个设计帮我避免了一次生产事故当时某个模型的流在生成中直接断掉如果没有 partial 机制整个请求会一直挂着直到超时。4.3 工具调用函数调用的归一化各家都有自己的“语法”工具调用是这两年模型 API 差异最大的部分。厂商 A 的请求格式是把函数列表放在tools字段里每个函数有name、description、parameters厂商 B 用的是functions字段且参数的 JSON Schema 格式要求更严格厂商 C 甚至要求先注册工具再通过工具 ID 引用。我们在统一接口里做了一个折中的设计统一请求侧只暴露一个tools字段格式按通用 JSON Schema 定义。适配器负责把统一格式转换成各家风格的请求。统一响应侧也只暴露一个tool_calls数组每个元素包含id、name、arguments字符串形式的 JSON。最麻烦的是厂商 C 的“先注册再使用”流程。我们的适配器在首次调用时自动完成工具注册并把返回的工具 ID 缓存到本地等第二次调用时直接引用。如果服务端工具 ID 过期适配器捕获到错误后自动重新注册并重试一次。这套逻辑虽然只对这个厂商有用但放对位置后业务方完全无感知这也是统一接口的核心价值复杂逻辑下沉到适配层业务方永远只面对简单的统一模型。5. 生产环境必须处理的四个问题超时、限流、成本与灰度5.1 超时与重试策略不同厂商要分开配置很多团队的默认配置是“连接超时 5 秒、读超时 30 秒、重试 2 次”但这在一套网关里是远远不够的。我们在配置中心给每个厂商分别维护三组超时参数连接超时厂商 A 在海外网络不稳定连接超时设为 8 秒厂商 B 在国内直连设为 3 秒。首包超时有的模型流式响应首包很慢需要在服务端排队我们要把首包超时设得比普通请求长。整体超时非流式请求必须有一个总超时防止模型服务挂死导致连接被长时间占用。重试策略我们总结了一个比较实用的原则“不重试不可重试的错误对可重试的错误要用退避而不是立即重试。”对于限流429和 5xx 错误我们采用“指数退避 抖动”基础间隔 500ms最多重试 3 次。对于 4xx 错误参数错误、鉴权失败不重试直接返回错误。另外重要的一点是“重试必须换请求实例”有些场景下上一次请求已经消耗了部分 token 或者在服务端产生了部分结果直接重试可能导致计费两次。我们的做法是重试时生成新的request_id并在日志中关联新旧 ID。5.2 限流与并发控制别把上游打崩接入多家模型后限流问题会从“上游限我”变成“互相拖累”。如果一个服务霸占了所有连接另一个服务的高优先级请求只能排队。我们引入了两层限流第一层是“上游配额保护限流”每个厂商对应一个信号量semaphore控制同时进行的请求数。比如厂商 A 的账号并发限制是 10我们在代码里把信号量设置为 8留一点缓冲。如果连续出现 429 响应还会动态降低信号量从 8 降到 6等一段时间稳定后再慢慢恢复。第二层是“下游优先级调度”按业务方的重要性分配权重。关键业务限流阈值高非关键业务限流阈值低。这里要特别提醒一句不要把限流做在 Gateway 层就完事还要把Retry-After响应头原样传给调用方。厂商返回的Retry-After是上游的预估恢复时间如果不透传业务方会盲目重试导致上游限流越来越严重。5.3 成本追踪与用量统计统一接口最容易忽视的价值统一接口还有一个隐形好处成本核算变得极其容易。因为所有请求都经过 Gateway我们可以统一从响应里提取usage数据写入时序数据库再按模型、厂商、业务方、时间维度做聚合。在第一版里我们没有做这个功能结果月底对账时发现厂商账单比预估高出不少。后来我们补上了用量统计每次请求完成无论是成功还是部分成功都会记录 prompt_tokens、completion_tokens、latency 和错误码。一个星期后我们很快就定位到某个业务方把max_tokens设置得过大导致大量 token 被浪费在“重复生成”上。还有一点值得提醒token 计费方法在厂商间并不统一。有的按字符数计费有的按 token 数计费而且各家 tokenizer 不一样。如果你直接比较两家 API 的“每个 token 价格”是没有意义的要换算成“同一文本的实际花费”才有可比性。我们在用量统计里保存了原始文本长度和各家 token 数这样对账时可以精确核算。5.4 模型灰度切换与 A/B 测试如何安全地换引擎统一接口真正体现价值的一个场景是“模型切换”。以前要换模型需要改业务代码、重新发布现在只需要在配置中心改一条路由规则把某个业务方的流量从模型 A 切到模型 B。为了保险我们还实现了一个简单的灰度模型按业务方维度配置“流量分桶”比如 10% 的请求走新模型90% 走旧模型。每笔请求在 Gateway 侧根据request_id哈希分桶保证同一个会话的所有请求一直落在同一个模型上这样多轮对话不会出现“每轮换个模型”的尴尬。灰度期间持续监控新旧模型的效果指标延迟、错误率、生成长度、业务方反馈。切换时还有两个容易踩的坑一个是“同样参数在不同模型上输出分布差异很大”你从模型 A 切到模型 B 后同样的 prompt 可能生成不同风格的内容需要业务方接受这个变化另一个是“模型版本会变”即使是同一个厂商模型服务端也可能在你不注意的时候更新了模型版本导致行为变化。我们的处理是在网关日志里记录模型版本标识每次排查问题时先从版本差异入手。6. 实测踩坑记录错误码映射、HTTP 调优与日志追踪6.1 错误码映射表把“妖魔鬼怪”翻译成人话接入过程中的大量错误码差异让我们不得不维护一张“错误映射表”。下表是我们演化后的最终版本场景厂商 A 的原始表现厂商 B 的原始表现统一错误码是否可重试请求参数有误HTTP 400 code: invalid_parameterHTTP 400 message: Bad requestINVALID_REQUEST否鉴权失败HTTP 401HTTP 403 code: access_deniedAUTH_FAILED否并发超限HTTP 429 Retry-After: 3HTTP 200 code: 4021RATE_LIMITED是上游内部错误HTTP 500流式中断无 HTTP 状态UPSTREAM_SERVER_ERROR是模型不可用HTTP 503 code: overloadedHTTP 200 content: MODEL_UNAVAILABLE是内容安全拦截HTTP 200 空内容 finish_reason: content_filterHTTP 400 code: safetyCONTENT_FILTERED否这张表最初是我们自己用的后来直接做成配置项放进代码。每接入一家新厂商只要更新映射表错误处理逻辑完全不用动。特别提醒如果厂商返回“HTTP 200 但内容为空”你一定要注意服务端可能把它当成一次成功请求并且照常计费。我们的做法是把这类情况标记为UPSTREAM_EMPTY_RESPONSE并在日志里打出完整请求体方便定位是内容安全过滤还是模型生成异常。6.2 HTTP 客户端调优连接池、DNS 缓存与 Keep-Alive统一网关对 HTTP 客户端的要求比普通业务要高因为所有外部流量都在这里汇聚。我们实际调整过的几个参数连接池大小按连接数上限设置为 200每个厂商维护独立 Client。Keep-Alive开启并设置keepalive_expiry30s避免频繁建立连接。DNS 缓存httpx默认每次请求都做 DNS 解析在高并发下会有额外延迟。我们启用了本地 DNS 缓存把解析结果缓存 60 秒。缓存时间别太长因为厂商可能切换后端 IP。HTTP/2能开启就开启。尤其是流式输出场景HTTP/2 的多路复用特性让我们体验到显著改善。还有一个经验是“不要把整个响应缓存在内存里再统一处理”。有些适配器拿到原始响应后习惯content await resp.json()如果是流式响应这会直接毁掉流式体验。流式响应必须边读边解析非流式响应则要设置合理的响应大小上限防止厂商异常返回超大 JSON 把内存打爆。6.3 日志与链路追踪没有日志就没法排障多模型网关的排障难度比单模型调用高出一个量级。为了能快速定位“是哪一次请求、走了哪个厂商、在哪个环节出的问题”我们给每个请求分配了一个全局request_id并在日志里统一输出[request_idreq_abc123] [vendorvendor_b] [modelmodel_b_xl] [attempt2] [eventprepare_request] [paramstemperature:0.8,max_tokens:2048] [eventupstream_response] [status_code429] [retry_after3] [eventretry_scheduled] [backoff1500ms] [eventfinal_response] [codeRATE_LIMITED] [retryabletrue]这个结构化日志的价值在灰度切换和故障排查时体现得淋漓尽致。比如两个业务方同时报“响应变慢了”通过日志里的 vendor 字段能快速判断是不是某个厂商整体变慢通过attempt能看到是否有大量重试发生通过params能发现是不是某个特殊参数导致了慢查询。除了日志我们还为每个请求记录“每一跳的时间戳”接收入口时间、适配器准备请求时间、上游响应首包时间、完整响应时间。这些指标落到时序数据库做成 P50/P95 延迟监控。有一次我们发现 P95 延迟暴涨看了每个环节的时间戳后才发现不是模型响应慢而是适配器里一个正则表达式在大响应体上消耗了太多时间——这类问题如果不埋点几乎不可能被发现。6.4 最后补充几条小经验如果让我带着现在的认知重新做一遍有几件事我会在第一天就做好统一接口的版本从一开始就要带上比如/api/v1/chat。不要以为内部接口不需要版本控制一旦接入的模型数量多了接口演进是必然的。适配器代码要允许独立发布。我们后来把适配器拆分成了独立包用显式版本依赖管理新增或回退某个厂商适配器时不需要重启整个网关。每个厂商的适配器必须写最小可用测试用例至少覆盖正常响应、空内容、流式中断、限流、鉴权失败这五类场景。这些测试是你新增模型时最可靠的“复读机”。所有模型 API 的密钥、基础 URL 一定放配置中心不要放在代码仓库里。一旦密钥泄露逐个换密钥是非常痛苦的过程。个人经验上我踩过的最深的一个坑是“运营商网络层对 SSE 连接的缓冲”。当流式响应跑在反向代理后面时代理层可能会缓冲响应导致业务方迟迟收不到第一个字节。排查那次问题时我们在链路里隐蔽地测试了直连、绕过代理等方案最终确认问题出在代理对Content-Type: text/event-stream的处理上。如果你也维护统一网关尽量让网关直连模型厂商避免经过不信任的中间代理层。统一接口这件事本质上是在“快速接入更多模型”和“保持业务侧稳定”之间找平衡。它不能帮你解决模型本身的能力问题但能把“换模型”“多加一个供应商”变成一件低风险、低成本的日常操作。只要把契约定清楚、适配器拆干净、错误映射做准确这套架构会让你非常省心。
