Claude与Gemini双模型集成实战：从密钥管理到生产级稳定落地

做AI应用开发的朋友应该都有这种感觉：把某一个模型API调通，真不难，照着官方文档写个请求，半小时就能出结果。难的是把两个模型放在同一个业务里，还要让它们稳定、省钱、能降级。我接手过不少这类项目，一开始都是在一个模型上跑通demo，等真要上生产时，才发现密钥管理、消息格式、工具调用、流式输出、限流费用这些坑一个都没躲掉。这篇文章就把我同时接入 Claude 和 Gemini 的完整做法拆成五步，每一步都会给出能直接复用的代码和当时踩过的坑。适用于已经跑通过至少一个模型API、正准备把多模型做成生产级服务的后端开发者。

1. 先想清楚：同时接两个模型到底图什么

很多人看到"接入两个模型"的第一反应是：选一个最好的不就行了？我一开始也是这么想的，直到在真实项目里吃了亏。单模型方案看起来简单，但实际撑起一个线上服务之后，会暴露三个绕不开的瓶颈。

1.1 单模型方案的三个真实瓶颈

第一个是单点故障。模型厂商的接口再稳定，也会有抖动、限流、升级窗口。你只有一个模型的时候，它一旦连续5xx或者429，你的业务就跟着一起挂。没有备胎，就没有选择权，服务可用性完全被上游捏着。

第二个是能力边界。Claude 和 Gemini 各有各的强项：Claude 在复杂指令遵循、代码生成、长链条推理上更细腻；Gemini 在超长上下文、多模态理解、响应速度上更有优势。一个业务里往往同时存在"写作润色"和"长文档问答"两种需求，强行用一个模型做所有事，总有一半场景表现打折。

第三个是成本和路由空间。不同模型的输入输出定价不同，有的便宜但速度慢，有的贵但质量高。接两个模型之后，简单问题走快模型，复杂问题走强模型，还能在两个模型之间做 A/B 对比。这不是炫技，是给业务留了优化空间。

1.2 我理解的高效集成，不是跑通demo

标题里说的"高效集成"，我的评价标准有三条：新增一个模型API时业务代码改动要小；某个模型出问题时能快速切换；线上运行时的超时、重试、限流、成本能被统一管理。跑通demo只是起点，demo只证明"能调用"，离"能上线"还差了工程化的那层皮。

所以这篇的五步是这样划分的：第一步管钥匙，第二步管翻译，第三步管对话，第四步管稳定，第五步管花钱。每一步都有自己的独立性，你也可以跳过已经做过的部分，直接看你缺的那一步。

2. 第一步：密钥、端点和最小请求，一小时跑通

先把最基础的环境跑通。这一步没什么高深的，但很多人的第一个坑都栽在认证方式上——Claude 和 Gemini 的认证头完全不一样，照抄另一个模型的请求一定会报错。

2.1 密钥管理：别让API Key出现在代码里

我见过不少团队把 API Key 直接写在业务代码里，甚至提交到仓库，这是迟早要出事的问题。正确做法是密钥走环境变量，本地开发用.env文件，生产环境用配置中心或密钥管理服务。至少做到代码仓库里永远搜不到明文 Key。

一个干净的本地配置长这样：

# .env CLAUDE_API_KEY=sk-ant-xxxx CLAUDE_API_BASE=https://你的Claude接口域名 GEMINI_API_KEY=AIzaxxxx GEMINI_API_BASE=https://你的Gemini接口域名

注意我用的是CLAUDE_API_BASE和GEMINI_API_BASE而不是写死域名。原因很简单：有些项目要走网关转发、有些需要统一走内网代理，把 base URL 抽成配置，换环境就不用改代码。读取时用os.getenv就行，但一定要做缺失校验，启动时发现没有 Key 就直接报错，别等到用户调接口时才暴露。

2.2 认证方式的差异：一个用请求头，一个也建议用请求头

先看两家请求层的差别，这张表我建议收藏：

Gemini 的 Key 放在 URL 查询参数里是老教程里常见的写法，官方 SDK 也兼容，但我强烈建议你用请求头x-goog-api-key。因为放在 URL 里的 Key 会出现在网关日志、代理日志、浏览器历史里，等于裸奔。用请求头至少不会把 Key 打在访问日志的地址栏。

2.3 最小可运行代码：先用requests把链路打通

在引入任何封装之前，先用最原始的requests调通一次。这能帮你确认网络、认证、参数三层都没问题。我贴两个最小函数，能跑通你再往下封装。

import os import requests def call_claude(prompt: str, model: str = "claude-sonnet-4-0") -> str: url = f"{os.environ['CLAUDE_API_BASE']}/v1/messages" headers = { "x-api-key": os.environ["CLAUDE_API_KEY"], "anthropic-version": "2023-06-01", "content-type": "application/json", } payload = { "model": model, "max_tokens": 1024, "temperature": 0.3, "messages": [{"role": "user", "content": prompt}], } resp = requests.post(url, headers=headers, json=payload, timeout=(10, 60)) resp.raise_for_status() data = resp.json() return "".join( block.get("text", "") for block in data.get("content", []) if block.get("type") == "text" ) def call_gemini(prompt: str, model: str = "gemini-2.0-flash") -> str: url = f"{os.environ['GEMINI_API_BASE']}/v1beta/models/{model}:generateContent" headers = { "x-goog-api-key": os.environ["GEMINI_API_KEY"], "content-type": "application/json", } payload = { "contents": [{"role": "user", "parts": [{"text": prompt}]}], "generationConfig": {"temperature": 0.3, "maxOutputTokens": 1024}, } resp = requests.post(url, headers=headers, json=payload, timeout=(10, 60)) resp.raise_for_status() data = resp.json() try: return data["candidates"][0]["content"]["parts"][0]["text"] except (KeyError, IndexError): raise RuntimeError(f"Gemini返回了意外结构: {data}")

这里埋了一个差异点：Claude 的响应正文是content数组，里面可能有多个 block，文本块要用type == "text"过滤；Gemini 的正文在candidates[0].content.parts，而且一旦命中安全过滤，这个字段可能直接不存在。这个差异后面会专门说，但你现在至少知道了：两家返回结构不是同一个套路。

第一次跑通时，先用print(resp.status_code)和print(resp.text)看原始返回，不要只盯着最终文本。我踩过一个坑：Claude 漏了anthropic-version头，返回 400，错误信息藏在响应体的error.message里，不打印原始响应根本不知道原因。

3. 第二步：设计统一请求层，别再为每家API写一套代码

跑通两个模型之后，最诱人的事情是什么？是在业务代码里直接写两套调用逻辑——如果判断走 Claude 就调call_claude，走 Gemini 就调call_gemini。短期能跑，但后面每个新功能都要改两遍，每次模型厂商更新协议你就要改两个地方。你需要的是一层"翻译官"，把业务侧的通用意图翻译成两家各自的协议。

3.1 业务侧只认一套消息结构

我建议先定义一个最小通用数据结构，业务代码只跟它打交道。

from dataclasses import dataclass, field @dataclass class ChatMessage: role: str # user / assistant / tool content: str tool_call_id: str = "" # tool结果回传时用 name: str = "" # 工具名 tool_calls: list = field(default_factory=list) # 模型发起的工具调用 @dataclass class ChatResult: text: str finish_reason: str usage: dict raw: dict

role只保留三种：user、assistant、tool。你不需要把 Claude 的assistant和 Gemini 的model暴露给业务方，这是翻译层的事。tool_calls用于承载模型要求调用工具的请求，后面第三步会用到。

3.2 两个Adapter：把通用消息翻译成各自格式

有了数据结构，再定义一个抽象接口和两个实现。核心是让业务侧调用provider.chat(messages)时完全感知不到底层是哪个模型。

from abc import ABC, abstractmethod class ChatProvider(ABC): @abstractmethod def chat(self, messages, system=None, tools=None, **kwargs) -> ChatResult: pass class ClaudeProvider(ChatProvider): def __init__(self, api_key, base_url, model): self.api_key = api_key self.base_url = base_url self.model = model def chat(self, messages, system=None, tools=None, **kwargs): headers = { "x-api-key": self.api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", } payload = { "model": self.model, "max_tokens": kwargs.get("max_tokens", 1024), "temperature": kwargs.get("temperature", 0.3), "messages": self._to_claude_messages(messages), } if system: payload["system"] = system if tools: payload["tools"] = tools # 这里可以先省略流式，第四步再补 resp = requests.post( f"{self.base_url}/v1/messages", headers=headers, json=payload, timeout=(10, 60), ) resp.raise_for_status() data = resp.json() text = "".join( block.get("text", "") for block in data.get("content", []) if block.get("type") == "text" ) return ChatResult( text=text, finish_reason=data.get("stop_reason", ""), usage=data.get("usage", {}), raw=data, )

Gemini 的版本除了把消息映射成contents，还要注意systemInstruction和generationConfig的位置。核心是_to_gemini_contents和_to_claude_messages这两个转换函数，它们把同一种ChatMessage列表转成两家各自的格式。比如 Claude 收到role="tool"的消息时要包成tool_result块，Gemini 则要转成functionResponse块——这就是翻译层的意义，业务代码永远不接触这些差异。

3.3 参数映射表：同一个参数，两家叫法和范围都不同

翻译层就是把这张表写死的代码。这里有一个经验：不要试图把两家所有参数都统一，只映射业务实际会用到的几个就够。像top_p这种几乎不动的参数，完全可以透传，强行统一反而会让接口变臃肿。我见过有人为了让两家参数完全一致，搞出一套又臭又长的配置对象，后来自己都搞不清哪个参数传给谁。

4. 第三步：多轮对话和工具调用，最容易翻车的地方

第一步第二步跑通的是单轮请求。一旦进入真实业务，你会发现多轮对话和工具调用才是真正拉开差距的地方。Claude 和 Gemini 在这两块有本质性的协议差异，稍不注意就会产生"前面聊得好好的，加个工具就全乱套"的问题。

4.1 消息历史和角色映射的细节

先看多轮消息的角色映射：

这里有两个具体的坑。第一，Gemini 多次连续的相同角色消息会被拒绝，比如你压入两条user消息中间没有model回复，接口会报错。你需要在上游做一次合并：连续同角色就拼成一条。第二，Claude 对消息顺序很敏感，首条消息不能是assistant，user和assistant最好交替出现，如果历史记录里出现两个连续的assistant，部分版本会返回校验错误。这些问题不是模型"脾气大"，是各自的消息结构设计如此，翻译层里做好规整就行。

4.2 系统提示词别放错位置

系统提示词是另一个高频踩坑点。Claude 直接放在请求体顶层：

{ "system": "你是客服助手，回答要简洁", "messages": [...] }

Gemini 则是一个独立对象：

{ "systemInstruction": { "parts": [ { "text": "你是客服助手，回答要简洁" } ] } }

我遇到过一种情况：把 Gemini 的systemInstruction塞进contents里假装成一条user消息，结果模型把系统规则也当作用户的内容去理解，回答完全跑偏。系统提示词就是系统提示词，别拿历史消息去伪装它。

4.3 工具调用的协议差异：一边用块，一边用functionCall

工具调用（也叫函数调用、function calling）是两边差异最大的地方。Claude 的思路是把工具调用表达为content数组里的tool_use块，模型输出一个带id的工具请求；工具执行完之后，你把结果包装成tool_result块，放进role=user的消息里还给模型。

Gemini 则是把工具声明放在tools[0].functionDeclarations里，模型返回的parts[0]里直接出现functionCall对象，带name和args字典；执行完的结果用functionResponse包起来，同样放进role=user的 parts 里。

下面是一个兼容两家的工具执行循环，业务侧只需要注册一个executor函数：

def run_with_tools(provider, messages, tools, executor, max_steps=5): for _ in range(max_steps): result = provider.chat(messages, tools=tools) if not result.tool_calls: return result.text # 模型要求调用工具，先把助手消息放进历史 messages.append(ChatMessage( role="assistant", content=result.text, tool_calls=result.tool_calls, )) # 逐个执行工具，把结果作为tool消息回传 for call in result.tool_calls: tool_result = executor(call["name"], call["args"]) messages.append(ChatMessage( role="tool", content=tool_result, tool_call_id=call["id"], name=call["name"], )) raise RuntimeError("工具调用超过最大轮数，可能存在死循环")

这个循环的关键是ChatMessage里的tool_calls和tool_call_id。Claude 的tool_use有id，Gemini 的functionCall严格来说没有显式 id，但你可以在翻译层为它生成一个临时 id 来对齐模型，这样上层逻辑只需要认tool_call_id。

4.4 工具调用最容易翻车的小细节

第一，Claude 可以在一次响应里返回多个tool_use块，Gemini 通常一次只返回一个functionCall，但你的解析逻辑不能假设只有一个。第二，Claude 的tool_result必须挂在role=user的消息里，而且 content 可以是字符串也可以是数组，如果你把工具的二进制结果（比如图片）塞进去，要按它定义的 block 格式来。第三，Gemini 的functionCall如果参数里有嵌套 JSON，它在传输时会展开成 JSON 对象，你解析时不要强行字符串化。

我给一个实用建议：先把无工具的对话跑通，再把工具定义加上。工具调用一旦出问题，排查链路比普通对话长得多，分阶段调试能帮你快速定位是协议问题还是业务逻辑问题。

5. 第四步：流式、重试与超时，别让线上请求裸奔

密钥有了，翻译层有了，对话和工具也通了。这时候离线上就剩最后一公里：稳定性。我见过太多项目在本地跑得好好的，一上生产就经常卡死、超时、白屏，原因基本都出在流式处理、重试策略和超时控制这三个地方。

5.1 流式返回的SSE解析：两家格式不一样

流式输出用的是 SSE（Server-Sent Events），每一行以data:开头，空行分隔事件。但两家的事件内容完全不同。Claude 的流式是一系列带type字段的事件，文本增量藏在content_block_delta里：

import json def iter_sse_events(resp): for line in resp.iter_lines(): if not line: continue text = line.decode("utf-8") if isinstance(line, bytes) else line if not text.startswith("data:"): continue yield json.loads(text[5:].strip()) def stream_claude(resp): for event in iter_sse_events(resp): if event.get("type") == "content_block_delta": delta = event.get("delta", {}) if delta.get("type") == "text_delta": yield delta.get("text", "")

Gemini 用streamGenerateContent?alt=sse时，每个data:事件就是一个完整的响应对象，文本增量同样在candidates[0].content.parts[0].text。所以你可以在翻译层里把两家都统一成yield text_chunk的迭代器，上层只管渲染。

用requests做流式请求时，记得设置stream=True，并且不要用timeout=(10, 60)这种笼统的读超时卡住整个流。流式场景下更合理的做法是设置一个较长的读超时，或者干脆不设读超时，依赖应用层的响应超时兜底——因为模型生成长文本时，两个数据包之间的间隔可能超过常规读超时。

5.2 重试策略：哪些错误值得重试，哪些千万不要

我处理重试的原则很简单：网络错误、5xx、429 可以重试；4xx 认证失败、参数错误绝对不能重试，重试一万次结果也一样。Claude 和 Gemini 在限流时都会返回 429，响应头里可能带retry-after，但不同时候格式不一样，你要写一个兼容解析。

重试要有退避，别用固定间隔。一个简单可靠的指数退避公式是sleep = min(base * 2 ** attempt, 30) + random.uniform(0, 0.5)，base 从 1 秒起步，最多重试 3 次。加random的原因是防止多个并发请求同时重试，造成"重试风暴"——大家都按同一套时间表重试，上游被打得更惨。

import time import random def retry_request(fn, retries=3): delay = 1.0 for attempt in range(retries + 1): try: return fn() except (requests.Timeout, requests.ConnectionError) as e: if attempt == retries: raise except requests.HTTPError as e: if e.response.status_code not in (429, 500, 502, 503, 504): raise if attempt == retries: raise time.sleep(min(delay * (2 ** attempt), 30) + random.uniform(0, 0.5))

5.3 空响应和截断：被忽略的线上杀手

流式和重试之外，还有一个非常隐蔽的问题：模型返回了 200，但正文里没有你要的文本。Gemini 在触发安全过滤时，finishReason会是SAFETY，candidates里可能只有空的content，甚至整个数组为空。如果解析逻辑不够健壮，前端拿到的就是一把空白。

Claude 也有类似情况：stop_reason为max_tokens时，说明输出被截断了，内容可能只有半句话。这两种情况都要在参数解析时暴露给业务方，而不是静默地返回空字符串，否则排查问题时你会误以为模型"变傻了"。

我在统一接口里加了一个约定：finish_reason一定要透传到上层。前端至少能提示"回答已截断"或"内容被安全策略拦截"，比展示一个空白页强一百倍。

6. 第五步：配额、限流和成本，长期稳定跑下去的关键

到了第五步，重点是让两个模型在线上长期稳定地跑，而不是偶尔爆出 429 或者月底账单吓人一跳。这里涉及配额管理、成本估算和降级策略三件事。

6.1 配额管理：RPM和TPM是两把锁

模型厂商的配额一般有两维：RPM（每分钟请求数）和 TPM（每分钟 token 数）。你在控制台申请到的数字不是理论值，线上请求一超过就会被限流。很多团队只在被 429 打脸后被动重试，其实更健康的做法是在客户端主动控制并发。

一个简单的做法是引入信号量，把并发请求数限制在安全线以下：

import threading class RateLimiter: def __init__(self, max_concurrency): self._sem = threading.Semaphore(max_concurrency) def __enter__(self): self._sem.acquire() def __exit__(self, *args): self._sem.release()

具体并发数怎么定？我一般看两个指标：你的业务峰值 QPS 和模型配额。如果配额是每分钟 1000 次请求，你的服务有 10 个实例，那每个实例的信号量设到 50 以下比较安全，留出余量给突发流量。别把配额定死到 1000 就按 1000 跑，你还要考虑其他调用方共不共享这个配额。

6.2 费用怎么算：别被表面低价骗了

大模型的计费是按 token 走的，输入和输出分开计价，而且上下文越长、工具定义越多，每次请求的输入 token 也越贵。一个常见的误解是只看生成质量，不看输入 token 的消耗。

我建议上线前先做一个成本估算：

单次请求成本 = 输入token数 × 输入单价 + 输出token数 × 输出单价

假设某模型输入单价是每百万 token 50 元，输出是每百万 150 元（数字仅为示意，以厂商定价页为准）。一次请求消耗 3000 输入 token + 800 输出 token，成本就是 3000/1,000,000×50 + 800/1,000,000×150 = 0.15 + 0.12 = 0.27 元。如果业务每天有 10 万次这样的请求，日成本就是 2.7 万——这时候你就会理解为什么要在客户端做上下文裁剪。

上下文裁剪常用的三个手段：滑动窗口只保留最近 N 轮对话；老消息先做摘要再塞回上下文；system 提示词和工具定义只保留当前会话真正需要的部分。这些优化每一条都能实实在在省钱，而且还能降低延迟。

6.3 熔断与降级：两个模型最大的价值是互为备份

两个模型都接好了，如果只是傻傻地一人一半流量，价值只发挥了一半。更聪明的用法是让它们互为备份，并且能自动熔断。

熔断器逻辑不复杂：记录最近几次调用的失败率，失败率超过阈值就暂时"熔断"，把流量切到另一个模型，过一段时间再放少量探测流量试恢复。

class CircuitBreaker: def __init__(self, failure_threshold=5, reset_timeout=60): self.failures = 0 self.threshold = failure_threshold self.reset_timeout = reset_timeout self.opened_at = None def record_failure(self): self.failures += 1 if self.failures >= self.threshold: self.opened_at = time.time() def record_success(self): if not self.opened_at: self.failures = 0 def is_open(self): if self.opened_at is None: return False if time.time() - self.opened_at >= self.reset_timeout: self.opened_at = None self.failures = 0 return False return True

降级策略的伪代码思路是：优先调主模型，主模型熔断就换备模型，两个都熔断就走兜底缓存或错误提示。这样单模型抖动就影响不到用户了，顶多是延迟稍微高一点。

7. 集成完成后的验证、灰度与切换经验

五个步骤走完，你的双模型服务已经能稳定跑了。但真正上线之前，我强烈建议再做一轮"验收"，不然你都不知道两个模型在真实业务里的表现到底差多少。

7.1 建一个固定评测集，别靠感觉判断好坏

我会准备一份包含 50~100 条真实业务请求的评测集，覆盖几类典型场景：短问答、长文本摘要、代码生成、工具调用、敏感内容拒绝。两个模型都跑一遍，记录响应文本、延迟、token 消耗和失败原因。评价指标可以分成硬指标（延迟、失败率、token消耗）和软指标（回答质量、格式正确率）。

这一步最容易被跳过，但也是最重要的。不用评测集的时候，"模型A比模型B好用"完全靠个人体感，一旦上线前有人问"到底哪个模型做这个任务更合适"，你拿不出数据，就只能拍脑袋。

7.2 灰度切换：改配置比改代码快

双模型的路由规则不要写死在代码里，我建议放在配置中心或环境变量里。比如DEFAULT_PROVIDER=claude、FALLBACK_PROVIDER=gemini，再配合一个PROVIDER_ROUTING配置，可以按百分比或按用户维度分流。这样灰度测试新模型时，只需要改配置，不用重新发布。

我做过一个比较顺滑的做法：线上 90% 流量走 Claude，10% 走 Gemini，跑了一周，对比两者的错误率和用户反馈，然后根据数据把主模型切换到表现更好的那一个。整个过程业务代码零改动。

7.3 上线之后还要盯什么

上线不是终点。模型厂商侧的模型版本会更新，API 行为也可能在不知不觉中变化。我建议定期重放评测集，比如每周跑一次，如果发现同一个 case 的结果明显变差，就要去翻是不是上游升级了模型或调整了默认参数。另外，请求日志里一定要保留model、finish_reason、prompt_tokens、completion_tokens、latency_ms、http_status这些字段，账单对不上、延迟变高、错误率突增的时候，这些日志就是你定位问题的第一手材料。

按这套五步走下来，我个人的体感是：最花时间的不是"接入"，而是"处理不一致"——同一套业务参数要在两套协议里翻译，同一个工具调用要在两种格式里转换。统一请求层建议用最小的接口去隐藏差异，但不要过度抹平差异；保留透传能力，保留raw原始返回，这样出了问题还能看到模型到底回了什么。希望这篇能帮你把这两个模型真正融入自己的系统，而不是停在"能调通"的层面。