做内部工具的时候，我一直在想怎么把 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 里叫maxOutputTokens，top_p在部分接口里可能被忽略，finish_reason还分别叫finish_reason、stop_reason、finishReason。

如果项目一开始就打算支持多家模型，这些差异就是必须付的技术债。欠在业务代码里，后面每加一个模型就要改一遍业务逻辑；欠在调用层里，只需要改适配器。这也是 TokenX 出现的直接原因。

1.2 统一调用层想解决的四个问题

TokenX 的设计目标一开始就只定四条，不贪多：

1. 切换模型不用改业务代码。业务侧只依赖统一接口，到底用哪个模型由路由决定。
2. 路由规则可以动态调整。默认模型、灰度比例、按用户特征分配模型，都要通过配置完成。
3. 错误、用量、成本结构统一。业务不需要分别处理三家 API 的异常类型，也不需要把prompt_tokens、input_tokens、promptTokenCount各写一遍。
4. 方便做模型评测和故障降级。想对比三个模型的输出，只要把路由规则改成按比例分配；主模型不可用，自动走备用模型。

这四条看起来简单，实际做到需要把很多细节磨平。后面的章节基本按这个目标展开。

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和密钥对应的环境变量名，路由规则则描述默认策略、特殊规则和降级顺序。默认流量走 GPT，pro 用户走 Claude，GPT 故障时自动降级到 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_factory=list) 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( text=choice["message"]["content"], finish_reason=choice.get("finish_reason", ""), usage=TokenUsage( usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0), ), model=raw.get("model", ""), provider="gpt", )

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 body

role映射是 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_stop；Gemini 则是返回一个可迭代的 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_name=provider.name()) if text: yield text

extract_text_from_raw是每个 Provider 自己实现的解析函数，但对外暴露的是同一个流式接口。业务侧接收流式输出时，再也不需要关心当前模型是哪个。

4. 生产环境里的隐藏工程细节

4.1 超时、重试和熔断策略

统一调用层真正复杂的地方不在消息转换，而在工程治理。模型接口不是永远稳定的，超时、限流、5xx 都是常态。TokenX 在 Root Config 里单独维护一组超时和重试参数：

不同模型延迟和吞吐特性不一样，用同一套超时参数会互相拖累。比如 Gemini 整体响应通常更快，首包等 20 秒就偏长；但如果你把整体超时设得太短，又可能在长上下文场景误杀正常请求。

重试也不是所有错误都重试。HTTP 400 说明请求体有问题，重试多少次都一样；429 是限流，需要退避后重试；5xx 和网络错误可以重试。我更推荐用指数退避并加抖动，sleep = base * 2^n + random(0, 50ms)，避免所有请求同时打爆上游。

熔断是对重试的补充。如果某个 Provider 最近失败率超过阈值，TokenX 会直接把它标记为不可用，让流量走备用模型，而不是每次都等超时后再重试。这个阈值一般设在 30% 左右，冷却时间 30 秒，具体需要根据业务容忍度调整。

4.2 Token 计费与成本关注

三家 API 返回的 usage 字段命名完全不一样，不归一化的话，成本统计根本没法做。基础对照如下：

TokenX 会把它们统一成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 里是image+source，在 Gemini 里是inlineData。如果业务将来要传图片，TokenX 需要定义统一的ContentBlock，再由各家 Provider 转换成自己的格式。

5.2 返回字段和终止原因不一致

不同模型的终止原因会让业务侧判断“生成是否正常结束”时踩坑。我见过最典型的场景：请求只生成了很短的内容就停了，前端没感知，但日志显示finish_reason是max_tokens，说明输出被长度上限截断了。

TokenX 在适配时会把这三个来源统一成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 常见问题速查表

这张表是我在维护 TokenX 时自己遇到过的真实问题，几乎每个都花过几小时排查。最麻烦的还不是错误本身，而是当你统一了一层接口之后，出错时报出来的堆栈往往只到 Provider 内部，业务侧看不到原始响应，排查时要加一层“透传原始错误摘要”的开关，否则问题会隐藏得很深。

说到最后，分享一个我自己的体会。做这套调用层时，最容易犯的错是追求一次支持所有能力。我一开始把三家 API 的每个参数都映射了一遍，后来发现大部分参数是摆设，真正有价值的是把messages、tools、流式、用量这四类抽象稳定下来，之后再加模型只是一个新 Adapter 的事。如果团队刚起步，也先别急着拆成独立服务，把 TokenX 当成一个包塞进现有应用里，等第二个应用也要接模型时，再考虑把它独立出来。调用层的价值是给业务腾出时间专注产品，而不是自己变成一个需要持续维护的复杂性源头。