用反向代理实现Gemini兼容OpenAI API协议

1. 项目概述：为什么要在 OpenAI API 框架里跑 Gemini？

“Run Gemini using the OpenAI API”——这个标题乍看矛盾，实则直击当前大模型工程落地中最真实、最普遍的痛点：不是模型选得不够好，而是生态适配太割裂。你手头有 Google 的 Gemini（Pro 或 Flash），它在多模态理解、长上下文推理、结构化输出上表现扎实；但你的整个后端服务、前端 SDK、监控告警、重试熔断、Token 统计、成本分摊系统，全都是基于 OpenAI 的chat.completions接口设计的。这时候，硬切 Gemini 原生 API？等于重写 30% 的核心链路，还要额外维护两套鉴权、限流、日志格式、错误码映射逻辑。不现实。

我去年在给一家智能客服 SaaS 做大模型网关升级时就卡在这一步。客户已有 17 个微服务调用openai.ChatCompletion.create()，日均请求 240 万次，所有日志字段、Prometheus 指标标签、AB 测试分流策略都深度绑定model="gpt-4-turbo"这个字符串。突然要接入 Gemini，团队第一反应是“能不能让 Gemini 假装成 OpenAI？”——不是戏谑，而是工程优先级下的务实选择。

答案是肯定的。这不是“绕过限制”，而是构建协议兼容层（Protocol Adapter）：在客户端无感知的前提下，把标准 OpenAI 请求（含messages,temperature,max_tokens,response_format等字段）接收进来，内部完成参数转换、请求体重构、响应格式归一化，再转发给 Gemini 的generateContent接口，最后把 Gemini 的 JSON 响应“翻译”成完全符合 OpenAIchat.completionsOpenAPI Schema 的返回体。整个过程对上游透明，连curl -X POST https://your-api.com/v1/chat/completions的命令都不用改。

关键词“Gemini”“OpenAI API”“protocol adapter”“model gateway”“LLM abstraction layer”全部自然嵌入——这正是本文要解决的核心：如何用最小侵入、最高复用的方式，把 Gemini 融入已有的 OpenAI 生态基建中。适合三类人：正在做模型选型的技术负责人、需要快速切换后端模型的算法工程师、以及想统一管理多个大模型 API 的 DevOps 工程师。它不教你 Gemini 怎么调用，而是告诉你：当你的代码里已经写了 200 处openai.ChatCompletion.create()，怎么让 Gemini 安静地、稳稳地、零修改地接上那根线。

2. 整体架构与方案选型：为什么是反向代理，而不是 SDK 封装或中间件？

2.1 三种常见路径的实操对比

刚接触这个需求时，团队内部讨论过三种主流技术路径：

我们最终选了 C 方案。原因很实在：上线节奏压着，不能让 17 个服务逐一发版。反向代理模式下，运维只需在 Kubernetes 里起一个gemini-proxyDeployment，然后把所有服务的OPENAI_BASE_URL从https://api.openai.com改成http://gemini-proxy:8000，5 分钟内全量生效。而 A 和 B 方案，光协调各服务 owner 做 SDK 替换或中间件注入，排期就得两周。

2.2 网关核心设计原则：不做增强，只做保真

很多团队一上来就想在网关里加“智能路由”“自动降级”“缓存命中”，这是陷阱。我们的设计铁律只有三条：

1. 零语义修改：temperature=0.7必须原样传给 Gemini，不因 Gemini 默认值是 0.3 就偷偷覆盖；max_tokens=1024必须严格对应 Gemini 的maxOutputTokens，不因 Gemini 最大支持 8192 就放宽限制；
2. 错误码归一化：Gemini 返回429 Too Many Requests时，必须转成 OpenAI 格式的{"error": {"message": "...", "type": "rate_limit_exceeded", "param": null, "code": "rate_limit_exceeded"}}，而非直接透传 Google 的429响应体；
3. 流式响应字节级对齐：OpenAI 的 SSE 流每行以data:开头，结尾双换行；Gemini 的流式响应是 JSON 数组，需实时解析candidates[0].content.parts[0].text并按 OpenAI 格式拼装，确保前端EventSource能无缝消费，不丢 chunk、不乱序。

这三条看似简单，实则踩坑最多。比如 Gemini 的stopSequences参数和 OpenAI 的stop字符串数组语义不完全等价——Gemini 是“遇到任意一个序列即停”，OpenAI 是“遇到完整字符串才停”，我们在网关里做了精确语义对齐：将["\n\n", "END"]拆成两个独立 stop sequence 发送给 Gemini，避免因\n\n被提前截断导致回答不完整。

2.3 为什么不用 Nginx / Envoy 做纯转发？

有人会问：Nginx 不也能做反向代理吗？为什么还要写代码？因为OpenAI 和 Gemini 的请求体结构本质不同：

• OpenAI 请求体：

  { "model": "gpt-4-turbo", "messages": [{"role": "user", "content": "Hello"}], "temperature": 0.7, "response_format": {"type": "json_object"} }

• Gemini 请求体：

  { "contents": [{"role": "user", "parts": [{"text": "Hello"}]}], "generationConfig": { "temperature": 0.7, "responseMimeType": "application/json" } }

Nginx 只能做 URL 和 Header 转发，无法动态修改 JSON body 字段名、嵌套结构、数组/对象互转。比如messages→contents是数组到数组，但messages[0].role映射到contents[0].role是直通的，而messages[0].content（字符串）要拆成contents[0].parts[0].text（对象嵌套）。这种结构化转换必须由代码完成。我们用 Pydantic V2 定义了双向 Schema，确保每个字段都有明确的转换逻辑和默认值兜底，这才是稳定性的根基。

3. 核心细节解析：参数映射、内容转换与流式处理的硬核实现

3.1 请求参数的精准映射表（附转换逻辑说明）

OpenAI 的 12 个常用参数，到 Gemini 的映射不是一一对应，而是存在语义鸿沟。我们整理了生产环境验证过的映射规则，每个都带转换逻辑和注意事项：

这张表不是静态配置，而是网关启动时加载的MappingRule类实例。每个字段转换都封装成方法，例如convert_stop_sequences()会先校验stop是否为 list，再过滤空字符串，最后转成 Gemini 兼容格式。这样做的好处是：当某天 OpenAI 新增logprobs参数，我们只需新增一个转换方法，不影响其他逻辑。

3.2 内容结构转换：从 messages 到 contents 的三次解析

messages数组到contents数组的转换，表面是字段重命名，实则暗藏三重解析逻辑：

第一重：Role 标准化
OpenAI 的systemrole 在 Gemini 1.0 中不被识别，会被忽略。我们的网关在解析时做如下处理：

• 若messages[0].role == "system"且模型为gemini-1.0-pro，则将messages[0].content提取出来，作为首条user消息的前缀，拼接为"【系统指令】{content}\n\n{original_user_content}"；
• 若模型为gemini-1.5-pro，则直接映射role: "system"；
• assistant和user角色直通，不做修改。

提示：Gemini 的systemrole 并非简单前置提示词，它会影响模型的底层行为模式（如是否启用工具调用、是否严格遵循 JSON schema），所以不能简单拼接。我们实测发现，1.5 版本下systemrole 的权重比拼接文本高 3.2 倍（通过 prompt engineering 测试得出），因此必须原生支持。

第二重：Content 多类型拆解
OpenAI 的content字段可以是 string 或 object（含image_url、text、type）。Gemini 要求parts数组中每个元素是text或inlineData（base64 图片）。网关需做：

• 若content是 string →parts = [{"text": content}]；
• 若content是 array → 遍历每个 item：
  • type=="text"→parts.append({"text": item.text})；
  • type=="image_url"→ 异步下载item.image_url.url，转 base64，生成{"inlineData": {"mimeType": "image/jpeg", "data": "base64..."}}；
  • 下载失败时，记录 error 并 fallback 到"图片无法加载，请检查 URL"文本。

我们用httpx.AsyncClient(limits=httpx.Limits(max_connections=100))管理图片下载连接池，实测 100 QPS 下平均下载耗时 120ms（CDN 缓存命中率 89%）。

第三重：Message 合并与截断
Gemini 对单次请求的总 token 有限制（1.5-pro 为 1M）。网关需预估messages总长度：

• 用tiktoken.encoding_for_model("gpt-4")计算 token 数（Gemini tokenizer 未开源，tiktoken 是业界事实标准）；
• 若prompt_tokens > 0.9 * model_max_tokens，触发截断逻辑：
  • 从最早user消息开始删，保留最后 3 轮对话；
  • 删除时优先删content中的冗余空格、换行，再删整条消息；
  • 截断后插入系统提示：“【注意】因上下文过长，历史消息已被精简，当前仅保留最近 3 轮交互。”

这套逻辑让网关在 token 超限时，不是粗暴报错，而是优雅降级，保障服务可用性。

3.3 流式响应的字节级处理：如何让 EventSource 不崩溃

OpenAI 流式响应是 Server-Sent Events（SSE），每条数据以data:开头，结尾双换行：

data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":"Hello"},"index":0}]} data: {"id":"chatcmpl-xxx","object":"chat.completion.chunk","choices":[{"delta":{"content":" world"},"index":0}]}

Gemini 流式响应是 JSON Lines（NDJSON），每行一个 JSON 对象：

{"candidates":[{"content":{"parts":[{"text":"Hello"}]}}]} {"candidates":[{"content":{"parts":[{"text":" world"}]}}]}

网关必须做三件事：

1. 实时解析 JSON Lines：用async for line in response.aiter_lines()逐行读取，json.loads(line)解析；
2. 提取 text 并构造 SSE chunk：从candidates[0].content.parts[0].text取值，拼装成 OpenAI 格式{"delta":{"content":"Hello"}}；
3. 处理空响应与错误帧：Gemini 有时返回{"error":{"code":429,"message":"Rate limited"}}，需捕获并转成 OpenAI 格式错误 chunk。

关键难点在于流式 chunk 的边界对齐。Gemini 的text字段可能包含换行符\n，如果直接拼进 SSE 的data:行，会导致前端EventSource解析失败（SSE 规范要求data:行内不能有换行）。我们的解决方案是：对text做 JSON 序列化再嵌入，即json.dumps({"delta": {"content": text}})，这样换行符自动转义为\n，保证单行安全。

此外，Gemini 流式响应没有finish_reason字段，直到最后一帧才返回{"finishReason":"STOP"}。网关需维护一个状态机，记录是否收到finishReason，并在最后一帧补全{"finish_reason":"stop"}字段，否则前端 SDK 会一直等待。

4. 实操过程：从零部署一个生产级 Gemini Proxy 网关

4.1 环境准备与依赖安装

我们选用 Python 3.11 + FastAPI + httpx 作为技术栈，原因明确：FastAPI 的异步能力完美匹配流式 IO，httpx 的AsyncClient对流式响应支持最成熟，且社区生态丰富。整个网关代码控制在 800 行以内，便于审计和定制。

基础依赖（requirements.txt）：

fastapi==0.111.0 httpx==0.27.0 pydantic==2.8.2 tiktoken==0.7.0 google-generativeai==0.8.3 uvicorn[standard]==0.30.1

注意：google-generativeaiSDK 是 Google 官方维护，必须用最新版（0.8.3+），旧版本不支持stream=True的异步迭代。tiktoken用于 token 预估，虽非 Gemini 原生 tokenizer，但实测误差 < 3%，可接受。

环境变量配置（.env）：

# Gemini API Key（务必用服务账号密钥，非个人 API Key） GEMINI_API_KEY=your_gemini_api_key_here # 模型映射表（JSON 字符串） MODEL_MAPPING={"gpt-4-turbo": "gemini-1.5-flash", "gpt-3.5-turbo": "gemini-1.0-pro", "my-pro": "gemini-1.5-pro"} # 限流配置（Redis 连接，可选） REDIS_URL=redis://localhost:6379/0 RATE_LIMIT_PER_MINUTE=1000 # 日志级别 LOG_LEVEL=INFO

部署时，GEMINI_API_KEY必须通过 Kubernetes Secret 注入，禁止硬编码。我们用python-decouple库加载.env，确保本地开发和生产环境配置一致。

4.2 核心网关代码详解（含关键注释）

以下是main.py的核心逻辑，已脱敏并添加生产级注释：

from fastapi import FastAPI, Request, HTTPException, status from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field, ValidationError from typing import List, Dict, Any, Optional, AsyncIterator import httpx import json import asyncio import tiktoken from google.generativeai.types import content_types from google.generativeai import GenerativeModel app = FastAPI(title="Gemini OpenAI Proxy", version="1.0") # 初始化 Gemini 模型客户端（复用连接池） gemini_client = httpx.AsyncClient( base_url="https://generativelanguage.googleapis.com/v1beta", timeout=httpx.Timeout(60.0, connect=10.0), limits=httpx.Limits(max_connections=100, max_keepalive_connections=20) ) # Tiktoken 编码器（用于 token 预估） enc = tiktoken.encoding_for_model("gpt-4") class OpenAIRequest(BaseModel): """OpenAI chat.completions 请求 Schema""" model: str messages: List[Dict[str, Any]] temperature: Optional[float] = Field(default=0.7, ge=0.0, le=1.0) top_p: Optional[float] = Field(default=1.0, ge=0.0, le=1.0) n: Optional[int] = Field(default=1, ge=1, le=8) max_tokens: Optional[int] = Field(default=None, ge=1) stop: Optional[List[str]] = None stream: Optional[bool] = False response_format: Optional[Dict[str, str]] = None tools: Optional[List[Dict[str, Any]]] = None tool_choice: Optional[str] = None class GeminiRequest(BaseModel): """Gemini generateContent 请求 Schema""" contents: List[Dict[str, Any]] generationConfig: Dict[str, Any] tools: Optional[List[Dict[str, Any]]] = None @app.post("/v1/chat/completions") async def chat_completions(request: Request): try: # 1. 解析 OpenAI 请求体（自动校验） raw_body = await request.body() openai_req = OpenAIRequest.model_validate_json(raw_body) # 2. 模型映射（查 MODEL_MAPPING 环境变量） gemini_model = get_gemini_model(openai_req.model) # 3. 构建 Gemini 请求体 gemini_req = build_gemini_request(openai_req, gemini_model) # 4. 调用 Gemini API（流式 or 非流式） if openai_req.stream: return StreamingResponse( gemini_stream_response(gemini_req, gemini_model), media_type="text/event-stream" ) else: return await gemini_non_stream_response(gemini_req, gemini_model) except ValidationError as e: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid OpenAI request: {e}" ) except httpx.HTTPStatusError as e: # 429/500 等错误转为 OpenAI 格式 raise map_gemini_error(e) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Gateway internal error: {e}" ) def get_gemini_model(openai_model: str) -> str: """根据 OpenAI model 名获取 Gemini model 名""" mapping = json.loads(os.getenv("MODEL_MAPPING", "{}")) return mapping.get(openai_model, f"models/{openai_model}") def build_gemini_request(openai_req: OpenAIRequest, gemini_model: str) -> GeminiRequest: """将 OpenAI 请求转换为 Gemini 请求""" # messages → contents 转换（含 role 标准化、content 拆解） contents = convert_messages_to_contents(openai_req.messages) # 构建 generationConfig gen_config = { "temperature": openai_req.temperature, "topP": openai_req.top_p, "candidateCount": openai_req.n, "stopSequences": openai_req.stop or [], } # max_tokens 处理 if openai_req.max_tokens: gen_config["maxOutputTokens"] = openai_req.max_tokens # response_format 处理 if (openai_req.response_format and openai_req.response_format.get("type") == "json_object"): gen_config["responseMimeType"] = "application/json" # tools 处理 tools = None if openai_req.tools: tools = convert_tools_to_gemini(openai_req.tools) return GeminiRequest( contents=contents, generationConfig=gen_config, tools=tools ) async def gemini_stream_response(gemini_req: GeminiRequest, gemini_model: str) -> AsyncIterator[str]: """生成 Gemini 流式响应，并转为 OpenAI SSE 格式""" url = f"https://generativelanguage.googleapis.com/v1beta/{gemini_model}:streamGenerateContent?key={os.getenv('GEMINI_API_KEY')}" async with gemini_client.stream("POST", url, json=gemini_req.model_dump()) as resp: if resp.status_code != 200: error_body = await resp.aread() raise httpx.HTTPStatusError( f"Bad Gemini response: {resp.status_code}", request=resp.request, response=resp ) # 逐行解析 Gemini JSON Lines 流 async for line in resp.aiter_lines(): if not line.strip(): continue try: gemini_chunk = json.loads(line) # 提取 text 并构造 OpenAI chunk text = extract_text_from_gemini_chunk(gemini_chunk) openai_chunk = build_openai_chunk(text, is_final=False) # yield SSE 格式 yield f"data: {json.dumps(openai_chunk)}\n\n" # 检查 finishReason if has_finish_reason(gemini_chunk): final_chunk = build_openai_chunk("", is_final=True, finish_reason="stop") yield f"data: {json.dumps(final_chunk)}\n\n" except json.JSONDecodeError: # Gemini 有时返回非 JSON 行（如 debug info），跳过 continue except Exception as e: yield f"data: {json.dumps({'error': {'message': str(e)}})}\n\n" def extract_text_from_gemini_chunk(chunk: Dict) -> str: """从 Gemini chunk 中安全提取 text""" try: candidates = chunk.get("candidates", []) if not candidates: return "" content = candidates[0].get("content", {}) parts = content.get("parts", []) if not parts: return "" return parts[0].get("text", "") except Exception: return "" def build_openai_chunk(text: str, is_final: bool, finish_reason: str = "null") -> Dict: """构建 OpenAI 格式 chunk""" chunk = { "id": f"chatcmpl-{uuid.uuid4().hex[:8]}", "object": "chat.completion.chunk", "created": int(time.time()), "model": "gemini-proxy", "choices": [{ "index": 0, "delta": {"content": text} if text else {}, "finish_reason": finish_reason if is_final else None }] } return chunk

这段代码的关键在于：所有转换逻辑都封装在纯函数中（如convert_messages_to_contents），不依赖 FastAPI 的 request/response 对象，便于单元测试。我们为每个转换函数写了 100% 覆盖率的 pytest，例如：

def test_convert_system_role_to_gemini_10(): messages = [ {"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Hello"} ] contents = convert_messages_to_contents(messages, model="gemini-1.0-pro") assert contents[0]["role"] == "user" assert "【系统指令】" in contents[0]["parts"][0]["text"]

4.3 部署与上线 checklist

部署不是uvicorn main:app一跑了之，以下是生产环境必须完成的 7 项检查：

1. HTTPS 终止：网关必须部署在 TLS 终止层之后（如 Nginx、Cloudflare），GEMINI_API_KEY绝对不可暴露在 HTTP 明文请求中；
2. 连接池调优：httpx.AsyncClient的limits参数需根据 QPS 调整。我们线上配置max_connections=200（支撑 500 QPS），max_keepalive_connections=50；
3. Token 预估兜底：tiktoken计算的 prompt token 数，需加 10% buffer。我们实测gpt-4tokenizer 对 Gemini 输入的误差为 +2.3%，所以max_tokens设置为int(tiktoken_count * 1.1)；
4. 错误日志标准化：所有HTTPException必须包含request_id（从X-Request-IDheader 读取或自动生成），便于全链路追踪；
5. 健康检查端点：添加/healthz端点，检查 Gemini API 连通性（HEAD https://generativelanguage.googleapis.com/v1beta/models），K8s liveness probe 必须用此；
6. Rate Limiting：用 Redis 实现分布式限流。我们用aioredis+slowapi，按X-Forwarded-ForIP 限流，防止单个恶意 client 打垮网关；
7. Metrics 暴露：集成 Prometheus，暴露gemini_proxy_requests_total{model, status_code}、gemini_proxy_latency_seconds_bucket等指标，Grafana 看板实时监控。

上线当天，我们做了三轮压测：

• 第一轮：100 QPS 持续 5 分钟，P95 延迟 < 800ms，错误率 0%；
• 第二轮：模拟突发流量（300 QPS 持续 30 秒），自动触发限流，被限流请求 100% 返回429，无雪崩；
• 第三轮：混杂流式/非流式请求（70% stream, 30% non-stream），验证流式 chunk 不丢、不错序。

5. 常见问题与排查技巧实录：那些文档里不会写的坑

5.1 典型问题速查表（按发生频率排序）