五步集成Claude和Gemini：构建高可用LLM网关

大模型接入这件事放在两年前还是烧钱又稀罕的活儿现在基本成了标配。但标配归标配我见过太多团队栽在同一个地方接一个模型跑demo沾沾自喜等要接第二个、第三个的时候直接被各家API的差异化打法整懵。Claude的messages结构、Gemini的contents结构一个用max_tokens限制输出一个用generationConfig里的maxOutputTokens格式对不上逻辑就得写两套。这篇文章不讲虚的直接给你一套能落地的五步集成法让你把Claude和Gemini同时接进项目里该用谁用谁挂了还能互相顶上。如果你正准备在公司内部做AI能力封装或者你的项目已经深度依赖某个模型、想引入第二个模型做兜底这篇内容可以直接当作业抄。下面所有代码基于Python 3.10用httpx异步实现我自己生产环境就是这么跑的。1. 为什么非要把Claude和Gemini一起接进来动手写代码之前先想清楚动机。很多人问我接一个模型难道不够吗这个问题我早些年也纠结过。如果你的场景只是偶尔生成一段文案那单模型确实够。可一旦模型能力进入业务流程比如客服问答、内容审核、数据分析报告单模型依赖就是一个低概率但高杀伤的隐患。模型服务商的整体可用性再高也架不住限流策略变化、网络抖动、接口版本升级这些事。我遇到过一次真实状况凌晨模型接口突然大面积超时智能客服系统全线飘红用户端看到的就是机器人罢工。当时只有一套链路排查到最后只能干等。那次之后我下定决心不管多麻烦都要把备选模型接进来。1.1 单模型依赖是赌运气不是做工程单点依赖的问题在于你无法控制上游的任何一个环节。官方文档写得再稳定实际生产里限流阈值、并发额度、故障恢复时间都是黑盒。你只能通过重试、降级、缓存去缓解但唯一能根治的办法就是第二条独立链路。注意这里说的第二条链路不是简单的换一个模型名而是真正独立的接口、独立的鉴权、独立的配额。如果两条链路其实都依赖同一个上游网关或者同一台出口机器那只是表面上的双路实际还是单点。1.2 双模型接入的三个实际收益把Claude和Gemini同时接入不是炫技也不是多接一个显得厉害它解决的是三个很实际的问题。第一个是容灾。主模型不可用时另一个模型顶上对用户来说只是感觉响应慢了一点而不是服务挂了。这里的关键不是让两个模型都常驻处理请求而是有一套健康检查和故障转移机制能在主链路异常时自动把流量切过去。第二个是成本。模型按token计费长文本、多轮对话、批量任务的开销完全不同。你可以把低价值的任务路由到成本更低的模型把高价值的任务交给更强的模型。路由配置得当月底账单会明显好看。第三个是效果。Claude和Gemini在能力上各有侧重代码生成、复杂结构化输出、长文本总结、多轮对话不同任务上各有胜负。与其在一个模型上死磕不如让它们各自负责擅长的场景。这套五步集成法本质上就是围绕这三个目标设计的。2. 集成方案选型为什么是适配器模式而不是写死两套逻辑动手之前先讲架构选型。你完全可以写两个函数一个claude_chat()一个gemini_chat()然后在业务代码里if/else判断该调用谁。但这样做完以后你会发现每个业务调用方都要知道自己当前用的是哪个模型都要自己处理两家API返回格式的差异一旦新增模型所有调用方都要跟着改。这不是消除复杂度而是把复杂度摊到了每个角落。我这里用的是适配器模式加统一路由定义一套业务侧接口Claude和Gemini各自适配这套接口业务代码只面对一个LLMProvider抽象具体走哪个模型由路由层决定。这样做的收益很直接——新增模型时业务代码一行不用改只是多写一个适配器、注册一条路由而已。2.1 为什么不是直接封装SDK有同学会问官方不是提供SDK吗直接用SDK不行吗。官方SDK确实封装了鉴权、请求、重试这些基础能力但SDK解决的问题是怎么调通一个模型没有解决多个模型怎么统一管理。而且每个模型一套SDK类型、错误处理、日志格式都不一样接入第二家时依然免不了写胶水代码。更重要的是SDK版本升级往往伴随着breaking change一旦哪个模型升级了SDK主版本所有依赖它的业务代码都要重新验证。自己封装一层薄薄的适配器反而把这种影响隔离在adapters目录里。2.2 技术栈准备与目录结构代码基于Python 3.10主要依赖httpx做异步HTTP调用pydantic做配置校验pytest做测试。如果你团队里用的是FastAPI这套结构基本可以直接嵌进去。目录结构类似这样llm_gateway/ ├─ provider.py # 统一接口定义 ├─ router.py # 路由与故障转移 ├─ config.py # 配置管理 ├─ adapters/ │ ├─ __init__.py │ ├─ claude_adapter.py # Claude适配器 │ └─ gemini_adapter.py # Gemini适配器 └─ tests/ ├─ test_adapters.py └─ test_router.py这个结构不算复杂但有一个关键作用把所有模型的差异锁在adapters目录里。接口参数变化、响应格式变化、鉴权方式变化都只影响对应适配器不会污染上层业务。这一步做扎实后面五个步骤才会顺利。3. 五步集成实战从零写一个可用的LLM网关下面进入正题。这五个步骤是我在一套内部知识库问答系统里跑通的方案前后迭代过三轮现在拿出来的是最简且可上生产的版本。每一步我都会给出代码和选型理由。3.1 第一步定义统一模型接口先定义统一的数据结构和接口。核心原则是所有适配器必须长得一样让上层业务不感知具体模型。# provider.py from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import AsyncIterator dataclass class LLMResponse: text: str # 最终生成的文本内容 model: str # 实际执行模型 total_tokens: int 0 # 统一后的token统计用于成本和监控 finish_reason: str # 结束原因stop / max_tokens / safety... raw: dict field(default_factorydict, reprFalse) # 原始响应排障用 class LLMProvider(ABC): abstractmethod async def chat( self, messages: list[dict], max_tokens: int 1024, temperature: float 0.7, **kwargs, ) - LLMResponse: 非流式对话入口。 messages结构统一为: [{role: user, content: 你好}] abstractmethod def stream_chat( self, messages: list[dict], max_tokens: int 1024, temperature: float 0.7, **kwargs, ) - AsyncIterator[str]: 流式对话入口每次产出增量文本片段。每个字段都有存在的理由。text是业务最终要消费的字段model记录实际执行模型方便审计total_tokens用于成本核算和监控告警finish_reason帮助判断输出是被正常停止还是被截断或安全策略拦截raw字段则是排障神器。接口统一用async因为网关通常要并发处理大量请求同步调用会卡住事件循环统一async也方便后续做超时管理和并发控制。messages统一成[{role: user, content: ...}]这个结构是业界最通用的格式之一。Claude原生就是这种结构Gemini转换起来也不难。这里不要为了迁就某个模型的特殊用法而破坏统一格式特殊需求一律通过kwargs透传。3.2 第二步实现Claude适配器接口定义好之后先实现Claude适配器。这里用httpx直接调HTTP接口不走官方SDK好处是依赖少、链路透明、排查问题方便。# adapters/claude_adapter.py import os import httpx from provider import LLMProvider, LLMResponse class ClaudeAdapter(LLMProvider): def __init__(self, api_key: str ): self.api_key api_key or os.environ[CLAUDE_API_KEY] self.base_url os.environ[CLAUDE_API_BASE] self.model os.getenv(CLAUDE_MODEL, claude-3-5-sonnet-latest) self.api_version os.getenv(CLAUDE_API_VERSION, 2023-06-01) def _headers(self) - dict: return { x-api-key: self.api_key, content-type: application/json, # 按官方文档要求携带协议版本头 anthropic-version: self.api_version, } async def chat(self, messages, max_tokens1024, temperature0.7, **kwargs): payload { model: kwargs.get(model, self.model), max_tokens: max_tokens, temperature: temperature, messages: messages, } async with httpx.AsyncClient(timeout60) as client: resp await client.post( f{self.base_url}/messages, headersself._headers(), jsonpayload, ) resp.raise_for_status() data resp.json() # Claude的content是数组遍历取text字段 text .join(block.get(text, ) for block in data.get(content, [])) usage data.get(usage, {}) return LLMResponse( texttext, modeldata.get(model, self.model), total_tokensusage.get(input_tokens, 0) usage.get(output_tokens, 0), finish_reasondata.get(stop_reason, ), rawdata, )这里两个点要注意。第一版本头anthropic-version必须带否则官方会拒绝请求。第二content返回的数组里每一项可能包含text字段也可能是其他类型的块遍历时只取text遇到工具调用等类型就自然跳过这个写法兼容性最好。base_url从环境变量读取不硬编码便于在测试环境、生产环境之间切换。同样API Key也走环境变量避免明文写死在仓库里。这属于底线安全习惯。3.3 第三步实现Gemini适配器Gemini适配器稍复杂一点因为消息格式和Claude不一样。Gemini使用contents加parts的结构role字段只接受user和model需要把Claude格式里的assistant映射成model。# adapters/gemini_adapter.py import os import httpx from provider import LLMProvider, LLMResponse class GeminiAdapter(LLMProvider): def __init__(self, api_key: str ): self.api_key api_key or os.environ[GEMINI_API_KEY] self.base_url os.environ[GEMINI_API_BASE] self.model os.getenv(GEMINI_MODEL, gemini-1.5-pro) staticmethod def _convert_messages(messages: list[dict]) - list[dict]: contents [] for msg in messages: role msg.get(role, user) # Gemini的role只接受user和modelassistant需要映射为model if role assistant: role model contents.append({ role: role, parts: [{text: msg.get(content, )}], }) return contents async def chat(self, messages, max_tokens1024, temperature0.7, **kwargs): payload { model: kwargs.get(model, self.model), contents: self._convert_messages(messages), generationConfig: { maxOutputTokens: max_tokens, temperature: temperature, }, } async with httpx.AsyncClient(timeout60) as client: resp await client.post( f{self.base_url}/models/{self.model}:generateContent, params{key: self.api_key}, headers{content-type: application/json}, jsonpayload, ) resp.raise_for_status() data resp.json() try: text data[candidates][0][content][parts][0][text] finish_reason data[candidates][0].get(finishReason, ) except (KeyError, IndexError): text finish_reason usage data.get(usageMetadata, {}) return LLMResponse( texttext, modelself.model, total_tokensusage.get(totalTokenCount, 0), finish_reasonfinish_reason, rawdata, )这段代码最关键的转换在_convert_messages。很多人在集成Gemini时报400root cause都是角色映射不对把assistant原样传过去Gemini不接受。另外要注意Gemini的API Key是放在查询参数key里传的不是放在Header里。这两种鉴权方式的差异不实际调一遍很容易忽略。如果你需要传递系统提示词Gemini用的是单独的systemInstruction字段不是塞进contents。Claude则是放在messages数组里以system角色区分。这也是两家的差异点后续可以在适配器里统一支持。3.4 第四步统一响应与流式处理两个适配器都返回LLMResponse之后业务层就统一了。但流式输出还没有统一这一步补上。Claude的流式接口走SSE事件流里有content_block_delta类型取delta里的text字段。Gemini的流式接口也支持SSE模式需要加altsse参数事件流里的data字段是JSON解析后按candidates路径取文本。看起来协议不同但封装到适配器里之后上层拿到的都是同一个异步迭代器每次吐一段文本增量。# adapters/claude_adapter.py 内新增 async def stream_chat(self, messages, max_tokens1024, temperature0.7, **kwargs): import json payload { model: kwargs.get(model, self.model), max_tokens: max_tokens, temperature: temperature, messages: messages, stream: True, } async with httpx.AsyncClient(timeout60) as client: async with client.stream( POST, f{self.base_url}/messages, headersself._headers(), jsonpayload, ) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if not line or not line.startswith(data:): continue data json.loads(line[5:]) if data.get(type) content_block_delta: delta data.get(delta, {}) if delta.get(type) text_delta: yield delta.get(text, )Gemini的流式类似只是事件里数据字段的路径不同。封装完成后两个适配器的stream_chat签名完全一致。业务层接入时不需要关心底层是SSE还是JSON数组只需要async for chunk in provider.stream_chat(messages): await websocket.send_text(chunk)这一层统一的价值在接第三个模型时会越来越明显。消息格式转换、鉴权差异、流式协议差异全部被挡在适配器内部。3.5 第五步路由、降级与故障转移接口统一了接下来做路由和故障转移。最基础的需求是主备模式正常情况下走ClaudeClaude异常时切到Gemini。进一步的需求是按任务类型路由比如代码生成走Claude内容总结走Gemini。我实现了一个带熔断器的网关核心逻辑是给每个模型维护健康状态连续失败超过阈值就暂时摘除冷却一段时间后再恢复。# router.py import asyncio import time from typing import Optional from provider import LLMProvider, LLMResponse class CircuitBreaker: def __init__(self, failure_threshold: int 5, recovery_timeout: int 60): self.failure_threshold failure_threshold self.recovery_timeout recovery_timeout self.failure_count 0 self.last_failure_time 0.0 self.is_open False def record_success(self): self.failure_count 0 self.is_open False def record_failure(self): self.failure_count 1 if self.failure_count self.failure_threshold: self.is_open True self.last_failure_time time.time() def can_try(self) - bool: if not self.is_open: return True if time.time() - self.last_failure_time self.recovery_timeout: self.is_open False self.failure_count 0 return True return False class LLMGateway: def __init__(self, primary: LLMProvider, fallback: LLMProvider): self.primary primary self.fallback fallback self.primary_breaker CircuitBreaker() self.fallback_breaker CircuitBreaker() async def chat(self, messages, **kwargs) - LLMResponse: if self.primary_breaker.can_try(): try: resp await self.primary.chat(messages, **kwargs) self.primary_breaker.record_success() return resp except Exception: self.primary_breaker.record_failure() # 主模型不可用或熔断走备选 try: resp await self.fallback.chat(messages, **kwargs) self.fallback_breaker.record_success() return resp except Exception: self.fallback_breaker.record_failure() raise这个熔断器逻辑不复杂但极其实用。它解决了一个核心问题不把一个模型的偶发超时放大成全站不可用。主模型连续五次失败就会进入熔断60秒内不再请求它直接走备选60秒后放一个探测请求试一下恢复正常就自动切回来。比熔断更进一层的是按场景路由。我在实际项目里维护了一个路由配置比如代码生成类任务优先走Claude长文档总结类任务优先走Gemini。实现上只需要在网关层多传一个task_type参数从配置里查出对应的provider实例即可MODEL_ROUTES { code_generation: claude, summary: gemini, chat: claude, reasoning: gemini, }这一步做完双模型接入的主流程就通了。但千万别急着上生产第五步的测试和监控才是保障。3.6 第五步的另外一半测试、监控与性能基线我把测试和监控拆出来单独讲因为它们的重要性不亚于代码本身。很多项目接入模型时的第一个坑就是demo能通一压测就挂。单测要覆盖两个适配器的消息转换逻辑Claude格式转Gemini格式assistant角色映射是否正确空消息如何处理。用mock的方式模拟HTTP响应不需要真实调用API跑得又快又稳定。联调阶段准备一组固定问题两个模型各跑一遍确认都能正常返回。我更推荐准备一个模型能力评测集按业务场景整理20条典型prompt从回答质量、响应延迟、token消耗三个维度记录。这不是为了比拼谁更强而是为了验证路由配置有没有把任务派给真正的优势模型。监控指标至少要有四类请求成功率、平均延迟和p99延迟、token消耗与估算成本、降级触发次数。我在项目里把LLMResponse里的total_tokens和model字段都打了日志这样每一笔请求花在哪个模型上、消耗了多少token都有据可查。降级次数尤其要盯如果每周都有大量降级说明主模型选得不对或者配额给少了。压测这块我建议先压主模型单链路再压故障转移路径。很多系统主链路压得狠但故障转移从来没测过真出事时备选链路反而先崩。模拟故障的方法很简单把主模型的API Key临时改成错的看流量是否自动切到备选以及切换耗时多久。4. 常见问题与排查实录接入过程中一定会遇到各种报错我把高频问题整理成了一张排查表方便你对号入座。问题现象可能原因处理方式401 UnauthorizedAPI Key缺失或错误检查环境变量确认Key未过期403 ForbiddenAPI Key权限不足在开放平台检查模型访问权限400 Bad Request消息格式转换错误重点检查role映射、parts结构404 Not Foundbase_url路径或模型名拼错逐字核对官方文档中的模型标识429 Too Many Requests触发限流指数退避重试分散请求频率超时网络问题或生成耗时过长分离连接超时与读取超时缩短max_tokens4.1 多个模型的认证方式差异Claude和Gemini的鉴权方式完全不同这是排障时最容易被忽略的点。Claude用Header里的x-api-keyGemini把key放在URL查询参数里。如果你把Claude的鉴权方式套到Gemini上会一直报401但代码看起来又没问题因为HTTP请求确实发出去了只是服务端不认。我的建议是每个适配器在构造函数里就固定好鉴权方式不搞统一逻辑。4.2 超时和限流的正确打开方式模型接口的超时要分两种看待。一种是连接超时通常5秒就够因为TLS握手和路由不需要太久。另一种是读取超时这个要根据业务场景设置短则30秒长则120秒。很多人的误区是只设置一个总超时时间结果短文本请求还好长文本生成任务频繁触发超时重试。限流这块重试不是越快越好。遇到429后直接退避重试会让上游限得更狠。我用的是指数退避加抖动第一次等1秒第二次等2秒第三次等4秒同时加一个0到500毫秒的随机抖动避免多个请求在同一时刻集中重试。4.3 finish_reason是排查输出异常的第一入口Claude的stop_reason和Gemini的finishReason本质上都在告诉你这次生成是怎么结束的。end_turn或STOP代表正常结束max_tokens或MAX_TOKENS代表输出被截断说明你要么调大max_tokens要么让模型精简回答少数时候Gemini会返回SAFETY说明输入或输出触发了内容安全策略这时需要调整prompt而不是改代码。这些信息不能只留在raw里要映射到统一响应的finish_reason字段并写入日志。不然用户反馈回答突然断了你连是截断还是安全拦截都分不清。5. 一点额外的心得与后续扩展最后聊几个这套方案落地过程中的教训。第一适配器别写太厚。我一开始想把两个模型的所有参数都封装成统一配置结果适配器里全是if/else参数组合爆炸。后来想明白了适配器只需要管最小公约数业务特有参数通过kwargs透传就行。第二降级不要等故障来了才第一次测。我吃过一次亏主备切换逻辑写完就丢在那边直到某次线上主模型挂了才触发结果备选链路因为环境变量没配好也跟着挂了。后来我养成了习惯每次发版前都把主模型Key改成错的做一次强制切换演练几秒钟的事但能避免关键时刻掉链子。第三关于扩展。这套结构接入第三个模型非常顺写一个适配器注册到路由表难点只在把该模型的特殊参数映射到统一接口上。如果你想做得更完善还可以在网关层加缓存对完全相同的请求直接命中缓存省掉大量重复token费用。我自己习惯把两个模型的评测结果打印出来贴在工位边上谁擅长什么一目了然路由配置自然有依据。希望这套五步法能帮你少踩几个坑快速把Claude和Gemini集成这件事落地。