大模型接入这件事，放在两年前还是烧钱又稀罕的活儿，现在基本成了标配。但标配归标配，我见过太多团队栽在同一个地方：接一个模型跑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_factory=dict, repr=False) # 原始响应，排障用 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_tokens=1024, temperature=0.7, **kwargs): payload = { "model": kwargs.get("model", self.model), "max_tokens": max_tokens, "temperature": temperature, "messages": messages, } async with httpx.AsyncClient(timeout=60) as client: resp = await client.post( f"{self.base_url}/messages", headers=self._headers(), json=payload, ) 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( text=text, model=data.get("model", self.model), total_tokens=usage.get("input_tokens", 0) + usage.get("output_tokens", 0), finish_reason=data.get("stop_reason", ""), raw=data, )这里两个点要注意。第一，版本头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和model，assistant需要映射为model if role == "assistant": role = "model" contents.append({ "role": role, "parts": [{"text": msg.get("content", "")}], }) return contents async def chat(self, messages, max_tokens=1024, temperature=0.7, **kwargs): payload = { "model": kwargs.get("model", self.model), "contents": self._convert_messages(messages), "generationConfig": { "maxOutputTokens": max_tokens, "temperature": temperature, }, } async with httpx.AsyncClient(timeout=60) as client: resp = await client.post( f"{self.base_url}/models/{self.model}:generateContent", params={"key": self.api_key}, headers={"content-type": "application/json"}, json=payload, ) 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( text=text, model=self.model, total_tokens=usage.get("totalTokenCount", 0), finish_reason=finish_reason, raw=data, )这段代码最关键的转换在_convert_messages。很多人在集成Gemini时报400，root 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模式，需要加alt=sse参数，事件流里的data字段是JSON，解析后按candidates路径取文本。看起来协议不同，但封装到适配器里之后，上层拿到的都是同一个异步迭代器，每次吐一段文本增量。# adapters/claude_adapter.py 内新增 async def stream_chat(self, messages, max_tokens=1024, temperature=0.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(timeout=60) as client: async with client.stream( "POST", f"{self.base_url}/messages", headers=self._headers(), json=payload, ) 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 第五步：路由、降级与故障转移接口统一了，接下来做路由和故障转移。最基础的需求是主备模式：正常情况下走Claude，Claude异常时切到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-key，Gemini把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集成这件事落地。
大模型接入这件事，放在两年前还是烧钱又稀罕的活儿，现在基本成了标配。但标配归标配，我见过太多团队栽在同一个地方：接一个模型跑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_factory=dict, repr=False) # 原始响应，排障用 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_tokens=1024, temperature=0.7, **kwargs): payload = { "model": kwargs.get("model", self.model), "max_tokens": max_tokens, "temperature": temperature, "messages": messages, } async with httpx.AsyncClient(timeout=60) as client: resp = await client.post( f"{self.base_url}/messages", headers=self._headers(), json=payload, ) 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( text=text, model=data.get("model", self.model), total_tokens=usage.get("input_tokens", 0) + usage.get("output_tokens", 0), finish_reason=data.get("stop_reason", ""), raw=data, )这里两个点要注意。第一，版本头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和model，assistant需要映射为model if role == "assistant": role = "model" contents.append({ "role": role, "parts": [{"text": msg.get("content", "")}], }) return contents async def chat(self, messages, max_tokens=1024, temperature=0.7, **kwargs): payload = { "model": kwargs.get("model", self.model), "contents": self._convert_messages(messages), "generationConfig": { "maxOutputTokens": max_tokens, "temperature": temperature, }, } async with httpx.AsyncClient(timeout=60) as client: resp = await client.post( f"{self.base_url}/models/{self.model}:generateContent", params={"key": self.api_key}, headers={"content-type": "application/json"}, json=payload, ) 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( text=text, model=self.model, total_tokens=usage.get("totalTokenCount", 0), finish_reason=finish_reason, raw=data, )这段代码最关键的转换在_convert_messages。很多人在集成Gemini时报400，root 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模式，需要加alt=sse参数，事件流里的data字段是JSON，解析后按candidates路径取文本。看起来协议不同，但封装到适配器里之后，上层拿到的都是同一个异步迭代器，每次吐一段文本增量。# adapters/claude_adapter.py 内新增 async def stream_chat(self, messages, max_tokens=1024, temperature=0.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(timeout=60) as client: async with client.stream( "POST", f"{self.base_url}/messages", headers=self._headers(), json=payload, ) 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 第五步：路由、降级与故障转移接口统一了，接下来做路由和故障转移。最基础的需求是主备模式：正常情况下走Claude，Claude异常时切到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-key，Gemini把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集成这件事落地。