资讯动态

Claude与Gemini双模型接入实战:五步构建统一API适配层

发布时间:2026/10/21 23:13:35 来源:尧图企业网站定制
先说个我真实的体感大模型接入这件事难点从来不在“发一个 HTTP 请求”而在你同时要接 Claude 和 Gemini 的时候。两家的 API 长得像又都不完全像文档各写各的坑也各埋各的。有人在业务里用 Claude 做深度推理又用 Gemini 做低成本批量最后代码里全是 if provider claude 的分支改一个公共参数要在两个文件里同步改。这篇就把我踩过的坑和沉淀下来的接入套路整理出来五步走目标是让两个模型在你业务里“长得一样”上层调用方只感知一个通用接口底层可以随时加新模型、切换供应商。这篇文章适合谁如果你是后端工程师、AI 应用开发者或者正在做智能体、内容生成、辅助写作这类项目需要把 Claude 和 Gemini 同时接进现有系统那这份实操笔记可以直接当参考。不会涉及复杂的框架选择以 Python 为例讲清楚每一步的取舍你看完能直接抄作业。1. 先看清两个模型的 API 差异才知道要抽象什么1.1 请求格式的基本差异Claude 走的是 Anthropic Messages API端点是/v1/messages。请求体里有一个顶层字段system放系统提示词messages数组里每条消息只有role和content其中 content 可以是一个字符串也可以是内容块数组比如文本块、图像块、工具调用块。这种设计偏“结构化”官方把多模态输入和工具调用都统一成了内容块类型。Gemini 走的是 Google Generative Language API端点是/v1beta/models/{model}:generateContent。请求体用system_instruction放系统提示词contents数组里的 role 只能取user或model内容由parts列表组成。文本就是{text: ...}图片是{inlineData: {mimeType: ..., data: base64}}工具调用则是{functionCall: {...}}。这两套设计谁优谁劣先不评价但它们直接决定了抽象层的难度。最直观的差异是Claude 的 assistant 消息里可以同时有文本和 tool_use 块Gemini 的 model 消息则更强制地区分 functionCall。也就是说同样一次多轮工具调用两边返回的 assistant 消息结构根本不能直接互转。1.2 鉴权与模型命名的差异Claude 的鉴权方式是x-api-key请求头同时要求带anthropic-version版本头比如2023-06-01。Gemini 则把 API Key 放在查询参数?key...上或者用Authorization: Bearer ...的 OAuth 方式。这个差异看着小但如果你要接的是网关统一鉴权就得在适配器层分别处理。模型命名也不一样。Claude 用claude-opus-4-1、claude-sonnet-4-1、claude-haiku-4-5这一套。Gemini 是gemini-2.5-pro、gemini-2.5-flash这种而且支持带-latest后缀的别名。问题来了你的业务层不可能让用户去记两套模型名所以必须在统一配置里做一层“业务模型名”到“供应商模型名”的映射。我自己习惯这样配providers: claude: api_key_env: ANTHROPIC_API_KEY gemini: api_key_env: GEMINI_API_KEY models: fast: provider: gemini model_name: gemini-2.5-flash max_tokens: 4096 reasoning: provider: claude model_name: claude-sonnet-4-1 max_tokens: 8192这样上层永远只传fast或reasoning供应商和版本细节都关在配置里。1.3 限流、计费与响应结构的差异限流维度不同。Claude 针对每分钟请求数RPM和每分钟 token 数TPM分别限流还会区分 prompt token 和 output token。Gemini 的限流也是 RPM 和 TPM但不同模型、不同地域的配额差异很大免费额度和付费额度完全两个世界。做生产系统时不能假设“请求量小就没事”一定要在代码里假设 429 是常态。响应结构上Claude 的顶层返回content数组里面是{type: text, text: ...}结束原因用stop_reason可选值有end_turn、max_tokens、stop_sequence、tool_use。Gemini 的返回是candidates数组candidates[0].content.parts里取文本结束原因在candidates[0].finishReason可选值有STOP、MAX_TOKENS、SAFETY、RECITATION等。这两个响应模型差异很大的地方在于安全结束原因Gemini 会单独给出SAFETY或RECITATIONClaude 则把这类情况混在普通消息里。抽象层如果忽略这些状态线上就会遇到“模型返回了空文本但 HTTP 状态码是 200”的诡异问题。2. 五步集成法的整体思路与前置准备2.1 五步到底在解决什么问题我见过太多项目一上来就对着 Claude 官方文档写代码跑通了再说要接 Gemini结果第二个接入周期比第一个还长。所以我把整个集成过程拆成五步每一步的产出是清晰的需求梳理与 API 差异盘点明确业务要哪些能力纯文本、流式、工具调用、多模态列出两家的 API 差异表。定义统一接口层设计内部的请求/响应数据结构跟任何供应商无关。实现适配器Claude 一个适配器Gemini 一个适配器内部完成参数转换和响应解析。处理流式与工具调用等特殊路径这是最容易翻车的地方单独抽出来讲。生产级打磨重试、限流、超时、降级、日志和成本监控。这五步不是瀑布式的实际做的时候第二、三步经常要来回迭代。你发现统一接口设计得不合理适配器里就得堆 if/else那就回头改接口定义而不是硬扛。2.2 账号与密钥管理接入前先把两家平台的 API Key 申请好。Claude 在控制台里创建 API KeyGemini 在 Google AI Studio 里创建两者都可能需要绑定付费账号才能高配额调用。密钥我建议统一放到环境变量或密钥管理服务里不要写进代码库。我自己的项目里习惯用.env文件管理本地开发线上用密钥服务注入环境变量。export ANTHROPIC_API_KEYsk-ant-... export GEMINI_API_KEYAIza...有一个容易忽略的点Claude 的 Key 是sk-ant-开头Gemini 是AIza开头Git 仓库的密钥扫描规则要注意把这两个都加进去避免误提交。权限方面也别贪大只授予当前项目真正需要的 API 权限。2.3 设计统一接口的请求模型统一接口层长什么样决定了整个接入体验。我建议用数据类定义一个ChatRequest和一个ChatResponse而不是直接透传各家格式。理由很简单透传格式一开始省事但你的业务逻辑会被供应商格式绑架。from dataclasses import dataclass, field from typing import Optional, List, Union, Dict, Any dataclass class ChatMessage: role: str # system | user | assistant | tool content: Union[str, List[Dict[str, Any]]] name: Optional[str] None dataclass class FunctionTool: name: str description: str parameters: Dict[str, Any] dataclass class ChatRequest: model: str messages: List[ChatMessage] max_tokens: int 1024 temperature: Optional[float] 0.7 tools: Optional[List[FunctionTool]] None stream: bool False dataclass class ChatResponse: text: str finish_reason: str input_tokens: int 0 output_tokens: int 0 tool_calls: List[Dict[str, Any]] field(default_factorylist) raw: Optional[Dict[str, Any]] None这里有两个设计细节要强调。第一ChatMessage的role用业务层语义而不是供应商语义。Gemini 里的model在我们的层里统一叫assistant等适配器转换时再映射。第二content允许字符串也允许内容块数组是为了兼容工具调用场景assistant 返回的 content 可以是一个包含tool_use的块这个块在业务层也可以保留但常规文本场景下上层只关心text字段。3. 核心实现适配器模式接入 Claude 和 Gemini3.1 用协议类统一适配器行为适配器层我建议定义一个抽象基类或者 Python 的 Protocol约束所有供应商适配器必须实现同样的方法。目的不是写花架子而是让后续接新模型时新增一个类就行两边改动的 diff 一目了然。from abc import ABC, abstractmethod class LLMProvider(ABC): abstractmethod def chat(self, req: ChatRequest) - ChatResponse: pass abstractmethod def chat_stream(self, req: ChatRequest): pass abstractmethod def count_tokens(self, req: ChatRequest) - tuple: passcount_tokens看着不起眼但生产环境里非常有用。你可以用它估算每一次请求的花费也可以用来在发送前检查上下文是否超限避免发出去被 400 打回来。3.2 Claude 适配器实现拆解Claude 适配器的核心工作有两块请求转换和响应解析。请求转换时要注意几点第一Claude 的system不在messages里面所以遍历req.messages时遇到role system的内容要单独提出来放到顶层system字段。第二Claude 的messages里content如果是内容块数组需要保留原有结构如果是纯字符串直接透传。第三工具调用中涉及tool_result的消息要按 Claude 的规范包成{type: tool_result, tool_use_id: ..., content: ...}这种块。import requests import os class ClaudeAdapter(LLMProvider): API_URL https://api.anthropic.com/v1/messages VERSION 2023-06-01 def __init__(self, api_key: str None): self.api_key api_key or os.getenv(ANTHROPIC_API_KEY) def _to_claude_messages(self, messages: List[ChatMessage]): system_parts [] claude_messages [] for msg in messages: if msg.role system: system_parts.append(msg.content) elif msg.role tool: claude_messages.append({ role: user, content: [{type: tool_result, tool_use_id: msg.name, content: msg.content}] }) else: claude_messages.append({role: msg.role, content: msg.content}) return \n.join(system_parts) or None, claude_messages def chat(self, req: ChatRequest) - ChatResponse: system_text, claude_messages self._to_claude_messages(req.messages) payload { model: req.model, max_tokens: req.max_tokens, messages: claude_messages, } if req.temperature is not None: payload[temperature] req.temperature if system_text: payload[system] system_text if req.tools: payload[tools] [ {name: t.name, description: t.description, input_schema: {type: object, properties: t.parameters}} for t in req.tools ] resp requests.post(self.API_URL, headers{ x-api-key: self.api_key, anthropic-version: self.VERSION, content-type: application/json, }, jsonpayload, timeout60) resp.raise_for_status() data resp.json() text .join(block.get(text, ) for block in data.get(content, []) if block.get(type) text) return ChatResponse( texttext, finish_reasondata.get(stop_reason, ), input_tokensdata.get(usage, {}).get(input_tokens, 0), output_tokensdata.get(usage, {}).get(output_tokens, 0), rawdata, )这套实现的坑在tool_result。Claude 官方规定工具结果必须放在user角色的消息里而且 content 要是块数组不是普通字符串。如果你把tool_result直接拼成字符串API 会报错。我一开始就在这上面吃过亏。另一个细节是max_tokens是必填参数Gemini 可以不填但 Claude 不填直接 400。3.3 Gemini 适配器实现拆解Gemini 适配器的转换逻辑跟 Claude 正好相反。它把system指令放到顶层system_instructionmessages转成contentsrole 从assistant映射到model工具函数则要改成tools下的function_declarations结构。class GeminiAdapter(LLMProvider): API_BASE https://generativelanguage.googleapis.com/v1beta def __init__(self, api_key: str None): self.api_key api_key or os.getenv(GEMINI_API_KEY) def _to_contents(self, messages: List[ChatMessage]): system_instruction None contents [] for msg in messages: if msg.role system: system_instruction {parts: [{text: msg.content}]} elif msg.role assistant: contents.append({role: model, parts: [{text: msg.content}]}) elif msg.role tool: # 工具结果要映射到 user 角色且用 functionResponse 结构 contents.append({ role: user, parts: [{ functionResponse: { name: msg.name, response: {result: msg.content} } }] }) else: contents.append({role: user, parts: [{text: msg.content}]}) return system_instruction, contents def chat(self, req: ChatRequest) - ChatResponse: system_instruction, contents self._to_contents(req.messages) payload {contents: contents} if system_instruction: payload[systemInstruction] system_instruction payload[generationConfig] { maxOutputTokens: req.max_tokens, } if req.temperature is not None: payload[generationConfig][temperature] req.temperature if req.tools: payload[tools] [{function_declarations: [ {name: t.name, description: t.description, parameters: t.parameters} for t in req.tools ]}] url f{self.API_BASE}/models/{req.model}:generateContent resp requests.post(url, params{key: self.api_key}, jsonpayload, timeout60) resp.raise_for_status() data resp.json() try: candidate data[candidates][0] parts candidate[content][parts] text .join(part.get(text, ) for part in parts) tool_calls [part[functionCall] for part in parts if functionCall in part] except (KeyError, IndexError): text, tool_calls , [] finish_reason candidate.get(finishReason, ) usage data.get(usageMetadata, {}) return ChatResponse( texttext, finish_reasonfinish_reason, input_tokensusage.get(promptTokenCount, 0), output_tokensusage.get(candidatesTokenCount, 0), tool_callstool_calls, rawdata, )Gemini 这里有一个特别容易出问题的细节用generateContent时请求体里的 API Key 是放在 URL 查询参数上的。如果你们的网关会对查询参数做日志脱敏一定要确认不会把 key 打到日志里。另外 Gemini 的角色映射里没有system用户侧如果用了 system 消息必须通过systemInstruction字段单独传否则被当成普通用户文本模型行为会整个跑偏。3.4 模型路由与工厂适配器本身不负责选模型。业务层调chat时要通过一个工厂根据配置里的providers和models找到对应的适配器实例。最简单的方式是做一个注册表class ProviderRegistry: def __init__(self): self._providers {} self._model_map {} def register_provider(self, name: str, provider: LLMProvider): self._providers[name] provider def register_model(self, alias: str, provider: str, model_name: str): self._model_map[alias] (provider, model_name) def get(self, alias: str): provider_name, model_name self._model_map[alias] provider self._providers[provider_name] return provider, model_name这里我刻意没有让ChatRequest.model直接对应供应商模型名而是对应业务别名。这样做的好处是切换供应商时不需要动业务代码。比如你哪天发现 Gemini 2.5 Flash 费用涨了想换成别家模型只改配置范围可控。4. 流式输出与工具调用的工程细节4.1 流式输出的归一化流式输出是集成里最容易出 bug 的部分。Claude 的流式响应是 SSE 格式每一行以event:开头data:携带 JSON。事件类型包括message_start、content_block_delta、message_delta等。文本增量在content_block_delta的delta.text字段。Gemini 的流式响应可以通过streamGenerateContent接口拿到内容是 SSE 或 JSON 数组文本增量在candidates[0].content.parts[0].text字段另一个主要区别是 Gemini 流式返回里会周期性重复完整内容块不像 Claude 那样只有增量。两种协议如果直接透传前端流式渲染的代码要写两套必须归一化成一个回调函数只向上层传字符串增量。def chat_stream(self, req: ChatRequest): # 以 Claude 为例归一化后外层只收到 (delta_text, finish_reason) resp requests.post(self.API_URL, headers..., jsonpayload, streamTrue, timeout60) finish_reason for line in resp.iter_lines(decode_unicodeTrue): if not line or not line.startswith(data:): continue data json.loads(line[5:]) if data.get(type) content_block_delta: yield data[delta].get(text, ), None elif data.get(type) message_delta: finish_reason data.get(delta, {}).get(stop_reason, ) yield , finish_reasonGemini 的流式归一化同理。核心原则是上游供应商的协议细节不能泄漏到业务层一旦泄漏你后面加一个新模型前端就要跟着改一次。4.2 工具调用的兼容处理工具调用是智能体应用绕不开的环节。Claude 的工具结果是tool_use/tool_result配对Gemini 是functionCall/functionResponse配对。这两对概念语义几乎一致但字段结构完全不同。在统一接口层我建议把工具调用抽象为tool_calls列表每个元素至少包含id、name、arguments三个字段。适配器负责从各自的响应里提取这三个字段并把后续工具结果消息转换成各自的格式。实际操作时Gemini 的 functionCall 里的args本身就是字典Claude 的是input字段。转换逻辑不复杂但要小心嵌套。还有一种情况是模型连续调用多个工具Claude 会在一次响应里返回多个tool_use块Gemini 也会在一个候选里返回多个functionCall你的循环逻辑要处理列表不要只取第一个。4.3 多模态输入的参数转换如果业务涉及图片输入转换的麻烦更大。Claude 的图片块是{type: image, source: {type: base64, media_type: image/png, data: ...}}。Gemini 的图片块是{inlineData: {mimeType: image/png, data: ...}}。两个结构不兼容但都能接收 base64 数据。我的建议是统一接口的 content 块用自定义格式{ type: image, mime_type: image/png, data: base64 }然后适配器各自转换。这里有个性能坑图片 base64 字符串很长如果每轮对话都带着同一张图token 消耗会显著增加。最好在统一层做一次图片去重或者把图片引用换成 URL 让模型拉取而不是无脑内联 base64。5. 生产级打磨重试、限流、降级与成本控制5.1 重试与退避策略生产环境里没有不报错的 API。我处理的请求里429 和 5xx 是最常见的失败类型但两类要分开对待。429 是限流说明你请求太快必须退避重试5xx 是供应商服务端问题可以适当重试。还有一类是网络层超时要单独设置 connect timeout 和 read timeout不能只用一个总超时。我常用的策略429指数退避初始 1 秒翻倍最多到 30 秒加随机抖动重试 3 次。5xx线性退避间隔 2 秒重试 2 次。网络超时立即失败不重试让上层走降级逻辑。这里的重要心得是重试必须在适配器层做而不是在业务调用层做。否则业务代码会变成一层套一层的重试整个链路反复消费 token账单会很难看。5.2 限流与并发控制虽然供应商有限流但你不能把限流压力全扔给对方。自己这边要做客户端限流否则一有流量高峰429 一来退避重试又加剧请求堆积。可以用信号量控制最大并发数比如 Claude 最多 10 并发Gemini 最多 20 并发。同时做一个简单的令牌桶按 RPM 限制每秒最大请求数。import threading import time class RateLimiter: def __init__(self, max_calls: int, period: float 60.0): self.max_calls max_calls self.period period self.tokens max_calls self.lock threading.Lock() self.last time.time() def acquire(self): with self.lock: now time.time() self.tokens min(self.max_calls, self.tokens (now - self.last) / self.period * self.max_calls) self.last now if self.tokens 1: wait (1 - self.tokens) / (self.max_calls / self.period) time.sleep(wait) self.tokens 0 else: self.tokens - 1这只是个简化版生产环境建议用现成库比如tenacity做重试、limits做限流。但原理就是自己心里要有数两家供应商的配额不互通一个供应商被限流不应该阻塞另一个供应商的请求。所以限流器一定要按供应商实例分开。5.3 超时控制与降级大模型响应慢是常态。Gemini 2.5 Pro 在复杂推理任务上耗时经常超过 30 秒Claude 的 Opus 模型也一样。所以超时要区分不同模型快模型给 60 秒慢模型给 120 秒流式场景则要设置“首包超时”也就是第一次收到增量内容的时间上限避免挂在连接上干等。降级策略也要在设计统一接口时就想好。当 Claude 不可用时要不要用 Gemini 兜底我的建议是分场景如果业务对响应质量要求高宁愿失败也不要降级到弱模型如果业务只是通用问答降级完全可行。把降级开关做成配置项不要写死在代码里。5.4 成本监控与 token 统计成本是生产环境绕不开的问题。两个模型的计费方式都按 token 计但输入和输出的单价不同缓存命中的价格也不同。统一接口返回的input_tokens和output_tokens就是为这个准备的。我建议每次调用之后把模型别名、供应商、输入 token、输出 token、耗时、HTTP 状态码写进日志或者时序数据库然后按小时汇总成本。不用搞太复杂的系统一个结构化日志加一个简单的聚合查询就够了。关键是先把数据埋点做好不然月底账单出来才发现某个模型接口被异常循环调用那时候查日志就晚了。我自己会把 token 用量和业务请求 ID 绑在一起这样排查单个用户的高昂请求时可以直接定位。6. 常见问题与排查技巧实录6.1 鉴权与 401 问题Claude 报 401 时常见原因是环境变量没配对。API Key 里如果有换行符也会导致鉴权失败建议读取后做strip()。Gemini 的 401 往往是 Key 失效或者权限不足检查 Google AI Studio 里的项目和 API Key 是否匹配。还有一种情况是走了代理导致 Host 头不对适配器直接报 SSL 或连接错误看到 401 别急着怀疑 Key先抓包看请求头完整不完整。6.2 上下文过长与请求 400Claude 的context窗口很大但请求体里的max_tokens只控制输出长度两者容易混淆。如果你把输入拼太长超过窗口上限API 会返回 400提示prompt is too long。Gemini 类似超过上限会返回INVALID_ARGUMENT。解决思路是在发送前用count_tokens预估超限就自动裁剪或做 RAG 检索来减少输入。另外有一个细节Gemini API 的maxOutputTokens在某个版本之前叫max_output_tokens传错字段名不会报错但会被静默忽略结果就是输出长度不受控。我遇到过实际输出远超预期的案例排查到最后是字段名拼错。6.3 流式响应解析失败与编码问题流式解析失败最常见的原因是 SSE 的数据跨多行。特别是 Claude 的流式里一个事件的数据可能很大iter_lines按行读取时没问题但如果数据里本身包含换行符就要按事件边界组装而不是每行单独解析。还有一个常见问题是 Unicode 转义JSON 里的中文可能以\uXXXX形式存在要统一用json.loads解析不要手动截字符串。Gemini 流式还有一个坑如果 prompt 触发安全过滤流式返回里可能根本没有candidates字段只有promptFeedback。解析代码如果没有对空candidates做保护会直接抛IndexError导致流式中断。这也是为什么我在前面强烈建议把finish_reason里的SAFETY单独映射成统一状态而不是统一当成普通空文本处理。6.4 工具调用循环终止的兜底工具调用场景最常见的故障不是调用失败而是无限循环。模型不断返回functionCall你的 Agent 不断执行工具、把结果丢回去模型又继续调用直到触达上下文上限或者账单爆炸。处理手段是设置最大工具调用轮次比如 5 轮。到上限后强制要求模型直接给出文本结果不再接受functionCall响应。这个逻辑用在统一接口层不要在单个适配器里做否则两个模型行为不一致。最后分享一个经验多模型接入一定不要图省事直接复用某一家 API 的官方 SDK 当通用层。SDK 耦合了供应商的请求协议和错误模型换一家就要重写。自建一个薄薄的适配器层前期多花半天时间后面每一次模型升级、新增供应商都能省下大量时间。我自己接第一个模型时也没想这么清楚等第二个模型进来才回头抽层改得那个痛苦。如果你现在正准备接 Claude 和 Gemini建议一步到位按这五步走。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑