做内部工具的时候我一直在想怎么把 GPT、Claude、Gemini 三个模型的调用层收到同一个门面后面后来就有了 TokenX 这套东西。它不是什么高深框架就是一层很薄的适配层业务侧发统一的请求TokenX 负责把请求翻译成三家 API 各自的语言再把响应、用量、异常都翻译回一套统一结构。这篇实操清单适合被模型切换折磨过的开发者也适合正在评估自建模型调用层的团队参考。我会把关键取舍、接口设计、代码示例和我在生产环境踩过的坑一条条列出来尽量具体到可以照抄。1. 为什么需要一套调用层1.1 多模型 API 的“表面相同、实则不同”我最早的项目其实是模型评测后台同一个问题要分别发给 GPT、Claude、Gemini然后把答案摆在一起对比。这需求听起来不难真正动手才发现三家 API 根本不是“换个 model 参数”就能完事。GPT 的消息体是messages数组里面每条消息用role区分system/user/assistant。Claude 虽然也用了messages但system通常是独立字段不混在消息数组里。Gemini 的输入是contents内部再分parts系统提示词要单独放到systemInstruction。同一个“系统提示词用户消息”在三种 API 里是三种组装方式。除了消息结构模型能力参数也不一致max_tokens在 Gemini 里叫maxOutputTokenstop_p在部分接口里可能被忽略finish_reason还分别叫finish_reason、stop_reason、finishReason。如果项目一开始就打算支持多家模型这些差异就是必须付的技术债。欠在业务代码里后面每加一个模型就要改一遍业务逻辑欠在调用层里只需要改适配器。这也是 TokenX 出现的直接原因。1.2 统一调用层想解决的四个问题TokenX 的设计目标一开始就只定四条不贪多切换模型不用改业务代码。业务侧只依赖统一接口到底用哪个模型由路由决定。路由规则可以动态调整。默认模型、灰度比例、按用户特征分配模型都要通过配置完成。错误、用量、成本结构统一。业务不需要分别处理三家 API 的异常类型也不需要把prompt_tokens、input_tokens、promptTokenCount各写一遍。方便做模型评测和故障降级。想对比三个模型的输出只要把路由规则改成按比例分配主模型不可用自动走备用模型。这四条看起来简单实际做到需要把很多细节磨平。后面的章节基本按这个目标展开。2. TokenX 的核心设计与选型2.1 Provider 抽象是地基“业务侧不感知具体模型”这句话好说落地就是一张协议。先用数据类定义统一请求和响应然后每个模型服务实现一个 Provider对外暴露相同的chat和stream_chat方法。抽象时要注意统一接口里只放业务真正需要的字段。我最初也试图把三家 API 的所有参数都塞进去结果presence_penalty、top_k、candidate_count这些字段在不同接口里语义还有差别硬对齐反而会让调用方迷茫。最后 TokenX 只保留这几类messages、model、temperature、max_tokens、tools、stream。够用了。2.2 配置驱动而不是代码驱动在业务代码里写if model gpt: ... elif model claude: ...是最先要消灭的写法。TokenX 会把模型映射和路由规则放到独立配置里代码只负责读配置并执行。我用的配置大概长这样providers: gpt: model: gpt-x api_key_env: GPT_API_KEY claude: model: claude-x api_key_env: CLAUDE_API_KEY gemini: model: gemini-x api_key_env: GEMINI_API_KEY router: default: gpt rules: - name: pro-user match: plan: pro provider: claude fallback: - claude - gemini这里每个 Provider 显式声明自己的model和密钥对应的环境变量名路由规则则描述默认策略、特殊规则和降级顺序。默认流量走 GPTpro 用户走 ClaudeGPT 故障时自动降级到 Claude再不行用 Gemini。密钥一定不要写进配置文件全部从环境变量读。配置里记环境变量名部署时由部署平台注入真实密钥。这是我一直坚持的底线。2.3 为什么没有直接用现成网关当时也研究过现成的开源模型网关最后没有直接用原因很实际网关往往把多租户、审计、报表、统一计费这些治理能力作为卖点部署和维护成本都不低。我们只是想为内部工具做一层隔离直接上网关反而显得头重脚轻。但我也反对为造轮子而造轮子。TokenX 底层直接用成熟的httpx做异步 HTTP 客户端超时和重试用现成工具库处理自己只写模型之间的格式转换。这样既保留了控制权又不用手写底层网络细节。3. 从 0 到 1 实现 TokenX步骤与代码3.1 基础结构统一请求与响应第一步是把统一协议定下来。我用的是 Python dataclass没有引入重型 schema 库因为调用层需要保持轻量。核心结构是这样from dataclasses import dataclass, field from abc import ABC, abstractmethod from typing import AsyncIterator, Optional dataclass class ChatMessage: role: str # system / user / assistant / tool content: str dataclass class ChatRequest: messages: list[ChatMessage] model: str temperature: Optional[float] None max_tokens: Optional[int] None tools: list[dict] field(default_factorylist) stream: bool False dataclass class TokenUsage: prompt_tokens: int completion_tokens: int dataclass class ChatResponse: text: str finish_reason: str usage: TokenUsage model: str provider: str ChatMessage.role先沿用三家通用的语义content先只支持字符串。将来如果要支持图片输入可以再把content扩展成结构体现在先不过度设计。ChatRequest.model是请求时显式指定模型用的正常不走配置文件时也可以动态覆盖这在模型对比场景很有用。然后是 Provider 基类class BaseProvider(ABC): abstractmethod async def chat(self, req: ChatRequest) - ChatResponse: ... abstractmethod async def stream_chat(self, req: ChatRequest) - AsyncIterator[str]: ...chat是普通补全stream_chat是流式补全。接口定义得越简单新 Provider 接入就越容易。3.2 适配三家 Provider 的关键写法每家 Provider 的核心工作只有两件事把ChatRequest转成上游 API 要求的请求体再把上游响应解析成统一的ChatResponse。GPT Provider 相对直接因为消息结构几乎可以复用class GPTProvider(BaseProvider): def _build_body(self, req: ChatRequest) - dict: body { model: req.model, messages: [{role: m.role, content: m.content} for m in req.messages], } if req.temperature is not None: body[temperature] req.temperature if req.max_tokens is not None: body[max_tokens] req.max_tokens return body def _parse_response(self, raw: dict) - ChatResponse: choice raw[choices][0] usage raw.get(usage, {}) return ChatResponse( textchoice[message][content], finish_reasonchoice.get(finish_reason, ), usageTokenUsage( usage.get(prompt_tokens, 0), usage.get(completion_tokens, 0), ), modelraw.get(model, ), providergpt, )Claude Provider 最需要注意的是system字段不能混在消息数组里必须拆出来class ClaudeProvider(BaseProvider): def _build_body(self, req: ChatRequest) - dict: system_parts [m.content for m in req.messages if m.role system] messages [ {role: m.role, content: m.content} for m in req.messages if m.role ! system ] body { model: req.model, max_tokens: req.max_tokens or 1024, messages: messages, } if system_parts: body[system] \n.join(system_parts) return body这里max_tokens对 Claude 是必填字段调用方没传时得给一个默认值否则接口直接报错。Gemini Provider 的转换更绕一点因为历史消息里的assistant角色要映射成model角色class GeminiProvider(BaseProvider): def _build_body(self, req: ChatRequest) - dict: contents [] system_text for m in req.messages: if m.role system: system_text m.content else: mapped_role user if m.role user else model contents.append({ role: mapped_role, parts: [{text: m.content}], }) gen_conf {} if req.temperature is not None: gen_conf[temperature] req.temperature if req.max_tokens is not None: gen_conf[maxOutputTokens] req.max_tokens body {contents: contents, generationConfig: gen_conf} if system_text: body[systemInstruction] {parts: [{text: system_text}]} return bodyrole映射是 Gemini 最容易踩的细节忘了把assistant改成model接口会报角色错误。这三段代码已经构成了 TokenX 最核心的部分后面无论是加工具调用、图片输入还是改成流式都是在这些转换函数里做加法。3.3 动态切换与降级路由Provider 写好之后Router 负责按配置选择谁来处理请求。最简单的路由实现就是读配置、匹配规则、回退class Router: def __init__(self, config: dict, providers: dict[str, BaseProvider]): self.providers providers self.default config[router][default] self.rules config[router].get(rules, []) self.fallbacks config[router].get(fallback, []) async def chat(self, req: ChatRequest): provider_name self._route(req) try: return await self.providers[provider_name].chat(req) except ProviderUnavailable: for fallback in self.fallbacks: try: resp await self.providers[fallback].chat(req) resp.provider fallback return resp except ProviderUnavailable: continue raise注意降级后一定要在统一响应里带上实际使用的 provider 和 model否则前端、日志、甚至计费都会搞混。这个字段在评测场景尤其重要不然同一批请求到底是谁答的根本分不清楚。3.4 流式响应如何统一流式是最难对齐的部分。GPT 的流式走 SSE以data:开头结束标记是data: [DONE]Claude 的事件类型里有content_block_delta和message_stopGemini 则是返回一个可迭代的 JSON 序列文本在candidates[0].content.parts[0].text。TokenX 的做法是统一成AsyncIterator[str]也就是只往业务侧吐文本增量不做完整 JSON 透传。async def stream_chat(self, req: ChatRequest): provider self.router.route(req) async for raw in provider.stream_chat(req): text extract_text_from_raw(raw, provider_nameprovider.name()) if text: yield textextract_text_from_raw是每个 Provider 自己实现的解析函数但对外暴露的是同一个流式接口。业务侧接收流式输出时再也不需要关心当前模型是哪个。4. 生产环境里的隐藏工程细节4.1 超时、重试和熔断策略统一调用层真正复杂的地方不在消息转换而在工程治理。模型接口不是永远稳定的超时、限流、5xx 都是常态。TokenX 在 Root Config 里单独维护一组超时和重试参数Provider首包超时整体超时建议重试条件GPT20 秒60 秒429、5xx、网络错误Claude20 秒60 秒429、5xx、网络错误Gemini15 秒45 秒429、5xx、网络错误不同模型延迟和吞吐特性不一样用同一套超时参数会互相拖累。比如 Gemini 整体响应通常更快首包等 20 秒就偏长但如果你把整体超时设得太短又可能在长上下文场景误杀正常请求。重试也不是所有错误都重试。HTTP 400 说明请求体有问题重试多少次都一样429 是限流需要退避后重试5xx 和网络错误可以重试。我更推荐用指数退避并加抖动sleep base * 2^n random(0, 50ms)避免所有请求同时打爆上游。熔断是对重试的补充。如果某个 Provider 最近失败率超过阈值TokenX 会直接把它标记为不可用让流量走备用模型而不是每次都等超时后再重试。这个阈值一般设在 30% 左右冷却时间 30 秒具体需要根据业务容忍度调整。4.2 Token 计费与成本关注三家 API 返回的 usage 字段命名完全不一样不归一化的话成本统计根本没法做。基础对照如下Provider输入 tokens 字段输出 tokens 字段GPTprompt_tokenscompletion_tokensClaudeinput_tokensoutput_tokensGeminipromptTokenCountcandidatesTokenCountTokenX 会把它们统一成TokenUsage(prompt_tokens, completion_tokens)。单看总量还不够输入和输出的计费单价通常不同而且输出 tokens 因为要占缓存成本往往更高。我会维护一张示例计价表放在配置里不硬编码pricing: gpt: input_per_mtok: 0.01 output_per_mtok: 0.03 claude: input_per_mtok: 0.02 output_per_mtok: 0.05 gemini: input_per_mtok: 0.001 output_per_mtok: 0.002具体数字随产品和活动可能变化单位也可能不是每百万 tokens所以 TokenX 里做成本计算时会先统一换算成“每百万 tokens”的价格。这样每次请求日志里除了 tokens还能带上估算成本月底复盘模型性价比时直接看数据。4.3 可观测性与链路追踪调用层不解决“谁调了、调了哪个模型、花了多少钱、是否成功”这些问题就不算生产可用。我给 TokenX 加了统一的请求 ID每次调用生成request_id从入口透传到 Provider SDK、日志和异常里。日志统一写成 JSON 行方便采集到日志系统里聚合{ request_id: 550e8400-e29b-41d4-a716-446655440000, provider: gpt, model: gpt-x, latency_ms: 321, prompt_tokens: 1240, completion_tokens: 312, estimated_cost: 0.013, error: null }除了请求日志还需要监控错误率、p95 延迟、限流次数。如果调用层是作为独立服务部署可以再挂一层中间件做 TPS 统计和告警如果只是打成库塞进应用里那至少要把日志和错误上报接好。不要等到用户反馈“模型怎么不动了”才去排查。5. 切换模型时最容易踩的坑5.1 Prompt 格式差异兼容先说最常见的 system prompt 问题。很多业务约定俗成把系统提示词放在第一条system消息里这在 GPT 上没问题但切到 Claude 和 Gemini 后如果适配层没有把system从消息数组里拆出来轻则行为异常重则直接报参数错误。然后是工具调用格式。GPT 的工具叫tools内部用 JSON Schema 描述Claude 的工具叫tools但字段略有不同Gemini 叫functionDeclarations。同一套业务函数要想同时跑在三家模型上适配层必须能把统一的 tools 结构分别转换过去不能只传原样。多模态输入也容易炸。图片在 GPT 里是image_url在 Claude 里是imagesource在 Gemini 里是inlineData。如果业务将来要传图片TokenX 需要定义统一的ContentBlock再由各家 Provider 转换成自己的格式。5.2 返回字段和终止原因不一致不同模型的终止原因会让业务侧判断“生成是否正常结束”时踩坑。我见过最典型的场景请求只生成了很短的内容就停了前端没感知但日志显示finish_reason是max_tokens说明输出被长度上限截断了。语义GPTClaudeGemini正常结束stopend_turnSTOP达到 max tokenslengthmax_tokensMAX_TOKENS内容被过滤content_filterstop_sequence等SAFETYTokenX 在适配时会把这三个来源统一成stop、max_tokens、filtered业务只管这三种语义。如果需要更原生信息可以再单独放一个raw_finish_reason字段但默认不要暴露给业务。5.3 上下文长度与 Token 估算切到不同模型后最大上下文长度可能差别很大。有些模型支持很长的窗口有些模型窗口小同一个用户会话切过去直接被拒。这时候按字符数截断一定不准因为中文、英文、代码的 token 占比差异很大。我通常会在调用层里给每个 Provider 维护一个max_context_tokens配置并在发送前对消息做预算裁剪。粗略实现可以是这样def trim_to_budget(messages, budget, estimate_fn): remaining budget trimmed [] for msg in reversed(messages): count estimate_fn(msg.content) if count remaining: break trimmed.insert(0, msg) remaining - count return trimmed这只是一个兜底策略不能替代正式的 tokenizer。但至少比“按条数截断”更接近合理行为。要精确估算可以直接用各模型配套的 tokenizer或者在请求返回后记录真实 usage反过来校准估算参数。5.4 常见问题速查表现象可能原因处理方式切到 Claude 报错说 max_tokens 缺失适配层没给默认值在统一请求里补默认max_tokensGemini 返回角色错误assistant没映射成model在 GeminiProvider 里做 role 映射模型输出半句就停下stop_reason/finish_reason是max_tokens增加max_tokens或启用自动截断续写流式首包出现奇怪前缀SSE 解析时把data:前缀当正文了按行解析并去掉 SSE 包装重试后同一请求被计费两次普通重试不幂等重放导致重复消费只在连接错误和 429/5xx 时重试并用退避usage 字段全是 0流式接口部分场景不返回 usage如果需要精确用量单独查计费接口或改用非流式这张表是我在维护 TokenX 时自己遇到过的真实问题几乎每个都花过几小时排查。最麻烦的还不是错误本身而是当你统一了一层接口之后出错时报出来的堆栈往往只到 Provider 内部业务侧看不到原始响应排查时要加一层“透传原始错误摘要”的开关否则问题会隐藏得很深。说到最后分享一个我自己的体会。做这套调用层时最容易犯的错是追求一次支持所有能力。我一开始把三家 API 的每个参数都映射了一遍后来发现大部分参数是摆设真正有价值的是把messages、tools、流式、用量这四类抽象稳定下来之后再加模型只是一个新 Adapter 的事。如果团队刚起步也先别急着拆成独立服务把 TokenX 当成一个包塞进现有应用里等第二个应用也要接模型时再考虑把它独立出来。调用层的价值是给业务腾出时间专注产品而不是自己变成一个需要持续维护的复杂性源头。
