作为一名在 AI 工程领域摸爬滚打多年的技术负责人，我见过太多团队在国内调用 Gemini API 时踩坑——网络超时、Key 管理混乱、成本居高不下。这些问题在我负责的图像识别平台月调用量突破 500 万次后变得尤为突出，直到我们接入 HolySheep 中转服务，问题才迎刃而解。

本文将深入剖析 HolySheep 接入 Gemini Pro 的完整方案，涵盖架构设计、生产级代码实现、性能 benchmark、成本优化策略，以及我踩过的那些坑和对应的解决方案。

一、为什么选择 Gemini Pro 与 HolySheep 的组合

Google Gemini Pro 在多模态理解、长上下文处理（支持 100K token）方面具有显著优势，尤其适合需要同时处理文本、图像、视频的复杂业务场景。然而，国内团队直接调用 Google API 面临两大核心障碍：

HolySheep 的核心价值在于：¥1=$1 无损汇率（官方 ¥7.3=$1，节省超过 85%）、国内直连延迟 <50ms、微信/支付宝充值、以及统一的多模型 API 管理。这让我们在成本和稳定性上都获得了质的飞跃。

二、Gemini Pro vs 主流模型价格对比

模型Input ($/MTok)Output ($/MTok)上下文窗口多模态适合场景
Gemini 2.5 Flash$0.15$2.501M token✅快速响应、实时应用
Gemini 2.5 Pro$1.25$5.001M token✅复杂推理、长文本
GPT-4.1$2.50$8.00128K token✅通用对话、代码
Claude Sonnet 4.5$3.00$15.00200K token✅长文本分析、写作
DeepSeek V3.2$0.27$0.42128K token❌成本敏感、纯文本

从价格数据可以看出，Gemini 2.5 Flash 的 output 价格仅为 Claude Sonnet 4.5 的 1/6，在多模态任务中具有极高的性价比。配合 HolySheep 的汇率优势，实际成本将进一步压缩至原来的 1/7 左右。

三、架构设计与网络方案

3.1 整体架构

┌─────────────────────────────────────────────────────────────────┐
│                        业务层                                     │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐        │
│  │ 图片审核  │  │ 文档理解  │  │ 视频分析  │  │ 智能客服  │        │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘        │
└───────┼─────────────┼─────────────┼─────────────┼───────────────┘
        │             │             │             │
        ▼             ▼             ▼             ▼
┌─────────────────────────────────────────────────────────────────┐
│                      API 网关层                                   │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │  HolySheep SDK (Python/JavaScript/Go)                   │    │
│  │  - 自动重试    - 限流控制    - Key 轮询    - 成本追踪   │    │
│  └─────────────────────────────────────────────────────────┘    │
└───────────────────────────┬─────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────────┐
│                    HolySheep 中转层                              │
│  https://api.holysheep.ai/v1                                    │
│  - 国内专线优化    - 多地域节点    - 智能路由                    │
│  - 实测延迟 <50ms  - 99.9% 可用性                               │
└───────────────────────────┬─────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────────┐
│                    Google Gemini API                             │
│  gemini-2.5-flash / gemini-2.5-pro                              │
└─────────────────────────────────────────────────────────────────┘

3.2 网络延迟实测

我在生产环境中对不同调用路径做了为期一周的延迟监控，数据如下：

四、生产级代码实现

4.1 Python SDK 封装

import requests
import time
import json
from typing import Optional, Dict, Any, Union
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
import hashlib

@dataclass
class HolySheepConfig:
    """HolySheep API 配置"""
    api_key: str
    base_url: str = "https://api.holysheep.ai/v1"
    model: str = "gemini-2.5-flash"
    max_retries: int = 3
    timeout: int = 30
    rate_limit: int = 100  # 每秒请求数

class HolySheepGeminiClient:
    """HolySheep Gemini API 客户端 - 生产级实现"""
    
    def __init__(self, config: HolySheepConfig):
        self.config = config
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {config.api_key}",
            "Content-Type": "application/json",
            "X-Holysheep-Model": config.model
        })
        # Token bucket 用于限流
        self._tokens = config.rate_limit
        self._last_refill = time.time()
        self._lock = __import__('threading').Lock()
    
    def _acquire_token(self):
        """令牌桶限流"""
        with self._lock:
            now = time.time()
            # 每秒补充 rate_limit 个令牌
            self._tokens = min(
                self.config.rate_limit,
                self._tokens + (now - self._last_refill) * self.config.rate_limit
            )
            self._last_refill = now
            if self._tokens < 1:
                sleep_time = (1 - self._tokens) / self.config.rate_limit
                time.sleep(sleep_time)
                self._tokens = 0
            else:
                self._tokens -= 1
    
    def chat(
        self,
        messages: list,
        temperature: float = 0.7,
        max_tokens: int = 4096,
        **kwargs
    ) -> Dict[str, Any]:
        """
        发送对话请求到 Gemini
        
        Args:
            messages: 消息列表，格式 [{"role": "user", "content": "..."}]
            temperature: 创造性控制 (0-1)
            max_tokens: 最大输出 token 数
        
        Returns:
            API 响应字典
        """
        self._acquire_token()
        
        # 转换为 Gemini 格式
        contents = self._convert_to_gemini_format(messages)
        
        payload = {
            "contents": contents,
            "generationConfig": {
                "temperature": temperature,
                "maxOutputTokens": max_tokens,
                **kwargs
            }
        }
        
        endpoint = f"{self.config.base_url}/chat/completions"
        
        for attempt in range(self.config.max_retries):
            try:
                response = self.session.post(
                    endpoint,
                    json=payload,
                    timeout=self.config.timeout
                )
                
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 429:
                    # 限流重试，指数退避
                    wait_time = 2 ** attempt
                    print(f"Rate limited, retrying in {wait_time}s...")
                    time.sleep(wait_time)
                else:
                    response.raise_for_status()
                    
            except requests.exceptions.RequestException as e:
                if attempt == self.config.max_retries - 1:
                    raise RuntimeError(f"API call failed after {self.config.max_retries} retries: {e}")
                time.sleep(2 ** attempt)
        
        raise RuntimeError("Max retries exceeded")
    
    def _convert_to_gemini_format(self, messages: list) -> list:
        """将 OpenAI 格式转换为 Gemini 格式"""
        contents = []
        for msg in messages:
            role = "user" if msg["role"] == "user" else "model"
            content = msg["content"]
            
            # 支持多模态内容
            if isinstance(content, str):
                contents.append({"role": role, "parts": [{"text": content}]})
            elif isinstance(content, list):
                parts = []
                for item in content:
                    if item["type"] == "text":
                        parts.append({"text": item["text"]})
                    elif item["type"] == "image_url":
                        parts.append({
                            "inlineData": {
                                "mimeType": item["image_url"]["detail"].get("mime_type", "image/jpeg"),
                                "data": item["image_url"]["url"].split(",")[1] if "," in item["image_url"]["url"] else item["image_url"]["url"]
                            }
                        })
                contents.append({"role": role, "parts": parts})
        
        return contents
    
    def batch_chat(self, requests: list, max_workers: int = 10) -> list:
        """批量并发请求"""
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = [executor.submit(self.chat, **req) for req in requests]
            return [f.result() for f in futures]



使用示例

if __name__ == "__main__":
    config = HolySheepConfig(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        model="gemini-2.5-flash",
        max_retries=3,
        rate_limit=50
    )
    
    client = HolySheepGeminiClient(config)
    
    # 文本对话
    response = client.chat(
        messages=[{"role": "user", "content": "解释什么是 RAG 系统"}],
        temperature=0.7,
        max_tokens=500
    )
    
    print(f"响应: {response['choices'][0]['message']['content']}")
    print(f"使用 token: {response['usage']}")

4.2 Node.js/TypeScript 实现

import axios, { AxiosInstance, AxiosError } from 'axios';

interface GeminiMessage {
  role: 'user' | 'model';
  content: string | Array<{ type: 'text' | 'image_url'; text?: string; image_url?: { url: string } }>;
}

interface GeminiConfig {
  apiKey: string;
  baseUrl?: string;
  model?: string;
  maxRetries?: number;
  timeout?: number;
}

interface RateLimiter {
  tokens: number;
  lastRefill: number;
  readonly rateLimit: number;
}

class HolySheepGeminiClient {
  private client: AxiosInstance;
  private maxRetries: number;
  private rateLimiter: RateLimiter;
  
  constructor(config: GeminiConfig) {
    this.client = axios.create({
      baseURL: config.baseUrl || 'https://api.holysheep.ai/v1',
      timeout: config.timeout || 30000,
      headers: {
        'Authorization': Bearer ${config.apiKey},
        'Content-Type': 'application/json',
        'X-Holysheep-Model': config.model || 'gemini-2.5-flash'
      }
    });
    
    this.maxRetries = config.maxRetries || 3;
    this.rateLimiter = {
      tokens: 100,
      lastRefill: Date.now(),
      rateLimit: 100
    };
    
    // 响应拦截器 - 自动重试
    this.client.interceptors.response.use(
      response => response,
      async (error: AxiosError) => {
        const originalRequest = error.config;
        if (!originalRequest || error.response?.status !== 429) {
          throw error;
        }
        
        // 指数退避重试
        for (let i = 0; i < this.maxRetries; i++) {
          await new Promise(resolve => setTimeout(resolve, Math.pow(2, i) * 1000));
          try {
            return await this.client(originalRequest);
          } catch (retryError) {
            if (i === this.maxRetries - 1) throw retryError;
          }
        }
        throw error;
      }
    );
  }
  
  private async acquireToken(): Promise {
    const now = Date.now();
    const timePassed = (now - this.rateLimiter.lastRefill) / 1000;
    
    // 每秒补充令牌
    this.rateLimiter.tokens = Math.min(
      this.rateLimiter.rateLimit,
      this.rateLimiter.tokens + timePassed * this.rateLimiter.rateLimit
    );
    
    if (this.rateLimiter.tokens < 1) {
      const waitTime = (1 - this.rateLimiter.tokens) / this.rateLimiter.rateLimit * 1000;
      await new Promise(resolve => setTimeout(resolve, waitTime));
      this.rateLimiter.tokens = 0;
    } else {
      this.rateLimiter.tokens -= 1;
    }
    
    this.rateLimiter.lastRefill = now;
  }
  
  private convertToGeminiFormat(messages: GeminiMessage[]): any[] {
    return messages.map(msg => {
      const parts: any[] = [];
      
      if (typeof msg.content === 'string') {
        parts.push({ text: msg.content });
      } else {
        for (const item of msg.content) {
          if (item.type === 'text' && item.text) {
            parts.push({ text: item.text });
          } else if (item.type === 'image_url' && item.image_url) {
            const url = item.image_url.url;
            const base64Match = url.match(/^data:([^;]+);base64,(.+)$/);
            if (base64Match) {
              parts.push({
                inlineData: {
                  mimeType: base64Match[1],
                  data: base64Match[2]
                }
              });
            }
          }
        }
      }
      
      return {
        role: msg.role === 'user' ? 'user' : 'model',
        parts
      };
    });
  }
  
  async chat(
    messages: GeminiMessage[],
    options: {
      temperature?: number;
      maxTokens?: number;
      topP?: number;
      topK?: number;
    } = {}
  ): Promise {
    await this.acquireToken();
    
    const contents = this.convertToGeminiFormat(messages);
    
    const payload = {
      contents,
      generationConfig: {
        temperature: options.temperature ?? 0.7,
        maxOutputTokens: options.maxTokens ?? 4096,
        topP: options.topP,
        topK: options.topK
      }
    };
    
    try {
      const response = await this.client.post('/chat/completions', payload);
      return response.data;
    } catch (error) {
      if (axios.isAxiosError(error)) {
        console.error('HolySheep API Error:', error.response?.data);
      }
      throw error;
    }
  }
  
  // 多模态图像理解
  async analyzeImage(imageBase64: string, prompt: string): Promise {
    const response = await this.chat([
      {
        role: 'user',
        content: [
          { type: 'text', text: prompt },
          { 
            type: 'image_url', 
            image_url: { url: data:image/jpeg;base64,${imageBase64} } 
          }
        ]
      }
    ]);
    
    return response.choices[0].message.content;
  }
  
  // 批量处理
  async batchChat(requests: { messages: GeminiMessage[]; options?: any }[]): Promise {
    return Promise.all(
      requests.map(req => this.chat(req.messages, req.options))
    );
  }
}

// 使用示例
async function main() {
  const client = new HolySheepGeminiClient({
    apiKey: 'YOUR_HOLYSHEEP_API_KEY',
    model: 'gemini-2.5-flash',
    maxRetries: 3
  });
  
  try {
    // 文本对话
    const textResponse = await client.chat([
      { role: 'user', content: '用一句话解释微服务架构' }
    ], { temperature: 0.7, maxTokens: 100 });
    
    console.log('文本响应:', textResponse.choices[0].message.content);
    console.log('Token 使用:', textResponse.usage);
    
    // 图像分析（示例）
    // const imageAnalysis = await client.analyzeImage(
    //   'BASE64_IMAGE_DATA_HERE',
    //   '描述这张图片的内容'
    // );
    
  } catch (error) {
    console.error('请求失败:', error);
  }
}

main();

4.3 Key 管理与轮询策略

import threading
from typing import List, Optional
from collections import defaultdict
import time

class KeyManager:
    """多 Key 管理与轮询策略 - 支持按量限流"""
    
    def __init__(self, keys: List[str], rpm: int = 60, rpd: Optional[int] = None):
        """
        Args:
            keys: API Key 列表
            rpm: 每分钟请求数限制（单 Key）
            rpd: 每日请求数限制（单 Key），可选
        """
        self.keys = keys
        self.current_index = 0
        self.rpm = rpm
        self.rpd = rpd
        
        # 按 Key 统计
        self.key_requests = defaultdict(list)  # {key: [timestamp1, timestamp2, ...]}
        self.key_daily_counts = defaultdict(int)  # {key: count}
        self.key_last_reset = defaultdict(lambda: time.time())
        
        self._lock = threading.Lock()
    
    def get_next_key(self) -> str:
        """获取下一个可用的 Key（令牌桶 + 每日限流）"""
        with self._lock:
            current_time = time.time()
            keys_checked = 0
            start_index = self.current_index
            
            while keys_checked < len(self.keys):
                key = self.keys[self.current_index]
                
                # 重置每分钟计数（每 60 秒）
                if current_time - self.key_last_reset[key] > 60:
                    self.key_requests[key] = []
                    self.key_last_reset[key] = current_time
                
                # 检查每日限额
                if self.rpd and self.key_daily_counts[key] >= self.rpd:
                    self.current_index = (self.current_index + 1) % len(self.keys)
                    keys_checked += 1
                    continue
                
                # 检查每分钟限额
                recent_requests = [
                    t for t in self.key_requests[key] 
                    if current_time - t < 60
                ]
                
                if len(recent_requests) < self.rpm:
                    # 使用这个 Key
                    self.key_requests[key].append(current_time)
                    self.key_daily_counts[key] += 1
                    return key
                
                self.current_index = (self.current_index + 1) % len(self.keys)
                keys_checked += 1
            
            # 所有 Key 都受限，等待
            wait_time = 60 - (current_time - self.key_last_reset[self.keys[start_index]])
            print(f"All keys rate limited, waiting {wait_time:.1f}s...")
            time.sleep(max(wait_time, 1))
            return self.get_next_key()
    
    def release_key(self, key: str, success: bool = True):
        """释放 Key（用于回退计数）"""
        with self._lock:
            if not success and self.key_requests[key]:
                # 不减少计数，但记录失败
                pass
    
    def get_stats(self) -> dict:
        """获取 Key 使用统计"""
        with self._lock:
            stats = {}
            current_time = time.time()
            for key in self.keys:
                recent = [
                    t for t in self.key_requests[key] 
                    if current_time - t < 60
                ]
                stats[key[:8] + '...'] = {
                    'rpm': len(recent),
                    'daily': self.key_daily_counts[key],
                    'limit_rpm': self.rpm,
                    'limit_daily': self.rpd
                }
            return stats



使用示例

if __name__ == "__main__":
    # 管理 3 个 Key，每个 Key 每分钟限制 100 次，每天限制 50000 次
    manager = KeyManager(
        keys=[
            "YOUR_HOLYSHEEP_API_KEY_1",
            "YOUR_HOLYSHEEP_API_KEY_2", 
            "YOUR_HOLYSHEEP_API_KEY_3"
        ],
        rpm=100,
        rpd=50000
    )
    
    # 高并发场景下自动轮询
    def make_request():
        key = manager.get_next_key()
        print(f"Using key: {key[:8]}...")
        # 实际请求逻辑...
        return key
    
    # 多线程测试
    import concurrent.futures
    
    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
        futures = [executor.submit(make_request) for _ in range(50)]
        results = [f.result() for f in futures]
    
    print("\n使用统计:", manager.get_stats())

五、性能优化与并发控制

5.1 连接池配置

# requests 连接池配置
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_optimized_session() -> requests.Session:
    """创建优化过的 requests session"""
    session = requests.Session()
    
    # 连接池配置
    adapter = HTTPAdapter(
        pool_connections=100,      # 连接池数量
        pool_maxsize=100,           # 每个池最大连接数
        max_retries=Retry(
            total=3,
            backoff_factor=0.5,
            status_forcelist=[500, 502, 503, 504]
        ),
        pool_block=False
    )
    
    session.mount('https://', adapter)
    session.headers.update({
        'Connection': 'keep-alive',
        'Accept-Encoding': 'gzip, deflate'
    })
    
    return session


对于 asyncio 场景

import aiohttp

async def create_aiohttp_session() -> aiohttp.ClientSession:
    """创建 aiohttp 连接池"""
    connector = aiohttp.TCPConnector(
        limit=100,              # 全局连接数限制
        limit_per_host=50,      # 单 host 连接数
        ttl_dns_cache=300,      # DNS 缓存时间
        enable_cleanup_closed=True,
        force_close=False       # 保持连接
    )
    
    timeout = aiohttp.ClientTimeout(
        total=30,
        connect=10,
        sock_read=20
    )
    
    return aiohttp.ClientSession(
        connector=connector,
        timeout=timeout,
        headers={'Connection': 'keep-alive'}
    )

5.2 性能 benchmark 数据

以下是我在生产环境中实测的性能数据（测试环境：8 核 CPU，16GB 内存，甘肃节点）：

六、成本优化策略

我在实际项目中总结出以下成本优化经验：

6.1 模型选择策略

"""
智能模型路由 - 根据任务复杂度自动选择最优模型
"""
from enum import Enum
from typing import Optional
import re

class TaskComplexity(Enum):
    SIMPLE = "simple"       # 简单问答、翻译
    MEDIUM = "medium"       # 摘要、分类
    COMPLEX = "complex"     # 推理、分析
    MULTIMODAL = "multimodal"  # 图像/视频理解

class ModelRouter:
    """任务复杂度分析与模型路由"""
    
    # 价格参考 ($/MTok output)
    MODEL_PRICES = {
        "gemini-2.5-flash": 2.50,
        "gemini-2.5-pro": 5.00,
        "deepseek-v3.2": 0.42,
        "claude-sonnet-4.5": 15.00,
        "gpt-4.1": 8.00
    }
    
    def analyze_complexity(self, prompt: str, has_multimodal: bool = False) -> TaskComplexity:
        """简单启发式分析任务复杂度"""
        word_count = len(prompt.split())
        has_code = bool(re.search(r'```|def |class |function ', prompt))
        has_math = bool(re.search(r'\d+[\+\-\*/]\d+|calculate|solve', prompt, re.I))
        
        if has_multimodal:
            return TaskComplexity.MULTIMODAL
        if word_count > 500 or has_math:
            return TaskComplexity.COMPLEX
        if word_count > 100 or has_code:
            return TaskComplexity.MEDIUM
        return TaskComplexity.SIMPLE
    
    def select_model(
        self, 
        complexity: TaskComplexity,
        prefer_low_cost: bool = True
    ) -> str:
        """根据复杂度选择最优模型"""
        
        if complexity == TaskComplexity.SIMPLE:
            # 简单任务用 Flash 或 DeepSeek
            if prefer_low_cost:
                return "gemini-2.5-flash"  # $2.50/MTok
            return "deepseek-v3.2"  # $0.42/MTok
        
        elif complexity == TaskComplexity.MEDIUM:
            return "gemini-2.5-flash"
        
        elif complexity == TaskComplexity.COMPLEX:
            return "gemini-2.5-pro"  # 更强的推理能力
        
        elif complexity == TaskComplexity.MULTIMODAL:
            # 多模态必须用 Gemini
            return "gemini-2.5-flash"
        
        return "gemini-2.5-flash"
    
    def estimate_cost(
        self, 
        input_tokens: int, 
        output_tokens: int, 
        model: str
    ) -> float:
        """估算请求成本（美元）"""
        # Gemini Flash 价格
        input_price = 0.15 / 1_000_000  # $0.15/M
        output_price = self.MODEL_PRICES.get(model, 2.50) / 1_000_000
        
        cost = input_tokens * input_price + output_tokens * output_price
        return cost
    
    def optimize_prompt(self, prompt: str) -> str:
        """简化 prompt 以减少 token 消耗"""
        # 移除冗余空格、换行
        optimized = ' '.join(prompt.split())
        # 截断过长前缀
        if len(optimized) > 2000:
            optimized = optimized[:2000] + "..."
        return optimized



使用示例

router = ModelRouter()


批量处理不同任务

tasks = [
    {"prompt": "解释什么是 REST API", "type": "simple"},
    {"prompt": "分析这段代码并找出 bug", "type": "medium"},
    {"prompt": "解读这张图表的趋势", "image": True, "type": "multimodal"},
]

for task in tasks:
    complexity = router.analyze_complexity(
        task["prompt"], 
        has_multimodal=task.get("image", False)
    )
    model = router.select_model(complexity)
    cost = router.estimate_cost(1000, 500, model)
    
    print(f"任务: {task['type']}")
    print(f"  复杂度: {complexity.value}")
    print(f"  推荐模型: {model}")
    print(f"  预估成本: ${cost:.6f}")

6.2 成本节省计算

假设你的业务场景：月调用量 100 万次，平均每次 1000 input tokens、500 output tokens。

七、常见报错排查

7.1 错误案例与解决方案

错误 1：401 Unauthorized - Invalid API Key

# 错误现象
{
  "error": {
    "message": "Invalid API Key provided",
    "type": "invalid_request_error",
    "code": "invalid_api_key"
  }
}


原因分析


1. Key 格式错误或拼写问题


2. Key 未激活或已过期


3. 请求头 Authorization 格式错误



解决方案


1. 检查 Key 是否正确复制（不含空格、前缀）


2. 在 HolySheep 控制台验证 Key 状态


3. 确保使用 Bearer token 格式



✅ 正确写法

headers = {
    "Authorization": f"Bearer {api_key}",  # 注意 Bearer + 空格
    "Content-Type": "application/json"
}


❌ 错误写法

headers = {
    "Authorization": api_key,  # 缺少 Bearer
    # 或
    "Authorization": f"Bearer {api_key}",  # 缺少空格
}

错误 2：429 Rate Limit Exceeded

# 错误现象
{
  "error": {
    "message": "Rate limit exceeded for model gemini-2.5-flash",
    "type": "rate_limit_error",
    "code": "rate_limit_exceeded"
  }
}


原因分析


1. 单 Key QPM 超过限制


2. 账户总 QPM 超过套餐限制


3. 并发请求过多



解决方案 - 实现智能限流


import time
import threading
from collections import deque

class TokenBucketRateLimiter:
    """令牌桶限流器"""
    
    def __init__(self, rate: int, per_seconds: float):
        """
        Args:
            rate: 每段时间内允许的请求数
            per_seconds: 时间窗口（秒）
        """
        self.rate = rate
        self.per_seconds = per_seconds
        self.allowance = rate
        self.last_check = time.time()
        self._lock = threading.Lock()
    
    def acquire(self, blocking: bool = True) -> bool:
        """获取令牌"""
        with self._lock:
            current = time.time()
            time_passed = current - self.last_check
            self.last_check = current
            
            # 补充令牌
            self.allowance += time_passed * (self.rate / self.per_seconds)
            if self.allowance > self.rate:
                self.allowance = self.rate
            
            if self.allowance < 1:
                if not blocking:
                    return False
                # 阻塞等待
                wait_time = (1 - self.allowance) * (self.per_seconds / self.rate)
                time.sleep(wait_time)
                self.allowance = 0
                return True
            
            self.allowance -= 1
            return True


使用示例

limiter = TokenBucketRateLimiter(rate=50, per_seconds=1)  # 50 QPS

def make_request():
    limiter.acquire()  # 自动限流
    # 实际请求逻辑
    pass

错误 3：400 Bad Request - Invalid JSON or Content Format

# 错误现象
{
  "error": {
    "message": "Invalid JSON payload",
    "type": "invalid_request_error",
    "code": "json_decode_error"
  }
}


或


{
  "error": {
    "message": "Invalid value for contents: Expected a non-empty list",
    "type": "invalid_request_error",
    "code": "invalid_value"
  }
}


原因分析


1. JSON 格式不完整（缺少引号、逗号等）


2. 消息内容为空或格式不符合要求


3. 参数类型错误（如传入字符串而非数组）



解决方案 - 严格校验请求格式


def validate_gemini_request(payload: dict) -> tuple[bool, Optional[str]]:
    """校验 Gemini API 请求格式"""
    
    # 检查必要字段
    if "contents" not in payload:
        return False, "Missing required field: contents"
    
    contents = payload["contents"]
    
    if not isinstance(contents, list):
        return False, "contents must be a list"
    
    if len(contents) == 0:
        return False, "contents cannot be empty"
    
    # 检查每个 content 块

相关资源
相关文章