上周深夜，我正在跑一个批量文本生成任务，突然收到这个报错：
429 Too Many Requests
{
"error": {
"code": 429,
"message": "Resource has been exhausted (e.g. check quota).",
"status": "RESOURCE_EXHAUSTED",
"details": [{
"@type": "type.googleapis.com/google.rpc.ErrorInfo",
"reason": "RATE_LIMIT_EXCEEDED",
"metadata": {
"quotaLimit": "100",
"quotaUsed": "100",
"window": "60s"
}
}]
}
}任务直接卡住了，所有请求都返回 429。那一刻我才意识到，配额管理不是配置完就完事的事儿，而是需要系统化设计的一门学问。今天这篇文章，我会把我踩过的坑和解决方案完整分享给你。
一、Gemini API 配额体系详解
在开始写代码之前，先理解 Gemini 的配额机制至关重要。Google Gemini API 采用双重限制策略：
- RPM（Requests Per Minute）：每分钟请求数限制
- TPM（Tokens Per Minute）：每分钟令牌数限制
- DPM（Daily Project Limit）：每日项目总调用限制
对于通过 HolySheep AI 接入的 Gemini API，这些限制会根据你的套餐级别有所不同。我测试过多个套餐，基础版 RPM 为 60，高级版可达 500+，国内直连延迟稳定在 <50ms，完全不用担心网络抖动触发超时。
二、基础调用：带频率控制的 Python 封装
先上一段我目前在生产环境使用的代码，基于 HolySheep API 的 Gemini 调用封装：
import time
import requests
from threading import Semaphore
from datetime import datetime, timedelta
class GeminiAPIClient:
"""带频率控制的 Gemini API 客户端"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1",
rpm_limit: int = 60, tpm_limit: int = 60000):
self.api_key = api_key
self.base_url = base_url
self.rpm_limit = rpm_limit
self.tpm_limit = tpm_limit
# 信号量控制并发请求数
self._semaphore = Semaphore(rpm_limit // 10)
self._request_times = []
self._token_usage = []
def _clean_old_records(self):
"""清理60秒前的记录"""
now = datetime.now()
cutoff = now - timedelta(seconds=60)
self._request_times = [
t for t in self._request_times
if t > cutoff
]
self._token_usage = [
(t, tokens) for t, tokens in self._token_usage
if t > cutoff
]
def _wait_if_needed(self, estimated_tokens: int = 1000):
"""检查并等待直到可以发送请求"""
self._clean_old_records()
# 检查 RPM 限制
if len(self._request_times) >= self.rpm_limit:
oldest = self._request_times[0]
wait_time = 60 - (datetime.now() - oldest).total_seconds()
if wait_time > 0:
print(f"[速率限制] 等待 {wait_time:.1f} 秒...")
time.sleep(wait_time)
self._clean_old_records()
# 检查 TPM 限制
current_tpm = sum(tokens for _, tokens in self._token_usage)
if current_tpm + estimated_tokens > self.tpm_limit:
oldest = self._token_usage[0][0]
wait_time = 60 - (datetime.now() - oldest).total_seconds()
if wait_time > 0:
print(f"[令牌限制] 等待 {wait_time:.1f} 秒...")
time.sleep(wait_time)
self._token_usage = [
(t, tokens) for t, tokens in self._token_usage
if (datetime.now() - t).total_seconds() < 60
]
def generate_content(self, model: str, prompt: str,
max_tokens: int = 2048) -> dict:
"""生成内容，自动处理频率限制"""
self._wait_if_needed(estimated_tokens=max_tokens)
url = f"{self.base_url}/models/{model}:generateContent"
headers = {
"Content-Type": "application/json",
"x-goog-api-key": self.api_key, # 兼容 Google 原生格式
"Authorization": f"Bearer {self.api_key}"
}
payload = {
"contents": [{"parts": [{"text": prompt}]}],
"generationConfig": {
"maxOutputTokens": max_tokens,
"temperature": 0.7,
"topP": 0.9
}
}
try:
with self._semaphore:
response = requests.post(url, json=payload, headers=headers, timeout=30)
# 记录本次请求
self._request_times.append(datetime.now())
if response.status_code == 200:
data = response.json()
# 估算 token 使用量
usage = data.get("usageMetadata", {})
tokens = usage.get("totalTokenCount", max_tokens)
self._token_usage.append((datetime.now(), tokens))
return data
elif response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", 60))
print(f"[429 限流] 将在 {retry_after} 秒后重试...")
time.sleep(retry_after)
return self.generate_content(model, prompt, max_tokens)
elif response.status_code == 401:
raise PermissionError("API Key 无效或已过期，请检查配置")
else:
raise Exception(f"API 调用失败: {response.status_code} - {response.text}")
except requests.exceptions.Timeout:
print("[超时] 请求超时，3秒后重试...")
time.sleep(3)
return self.generate_content(model, prompt, max_tokens)
使用示例
client = GeminiAPIClient(
api_key="YOUR_HOLYSHEEP_API_KEY", # 替换为你的 HolySheep API Key
rpm_limit=100,
tpm_limit=80000
)
result = client.generate_content(
model="gemini-2.0-flash",
prompt="用100字介绍人工智能的发展历史"
)
print(result["candidates"][0]["content"]["parts"][0]["text"])三、批量任务：异步队列 + 智能限流
当你需要处理大量请求时（比如批量翻译、批量摘要），单线程顺序调用效率太低。我设计了一个异步队列方案，支持优先级和智能限流：
import asyncio
import aiohttp
from dataclasses import dataclass, field
from typing import List, Optional, Callable
from collections import deque
import time
@dataclass
class Task:
"""异步任务单元"""
id: str
prompt: str
max_tokens: int = 2048
priority: int = 0 # 0=低, 1=中, 2=高
retries: int = 3
result: Optional[dict] = None
error: Optional[str] = None
class AsyncGeminiQueue:
"""异步 Gemini 请求队列"""
def __init__(self, api_key: str, rpm_limit: int = 100,
tpm_limit: int = 80000, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.rpm_limit = rpm_limit
self.tpm_limit = tpm_limit
self._request_interval = 60.0 / rpm_limit
self._token_budget = tpm_limit
self._last_reset = time.time()
# 低优先级队列
self.low_priority: deque = deque()
self.medium_priority: deque = deque()
self.high_priority: deque = deque()
self._semaphore: Optional[asyncio.Semaphore] = None
async def _reset_token_budget_if_needed(self):
"""每分钟重置 token 配额"""
current_time = time.time()
if current_time - self._last_reset >= 60:
self._token_budget = self.tpm_limit
self._last_reset = current_time
async def _wait_for_token_budget(self, tokens_needed: int):
"""等待足够的 token 配额"""
while self._token_budget < tokens_needed:
await asyncio.sleep(1)
await self._reset_token_budget_if_needed()
self._token_budget -= tokens_needed
async def _execute_single_request(self, session: aiohttp.ClientSession,
task: Task) -> dict:
"""执行单个请求"""
url = f"{self.base_url}/models/gemini-2.0-flash:generateContent"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.api_key}"
}
payload = {
"contents": [{"parts": [{"text": task.prompt}]}],
"generationConfig": {"maxOutputTokens": task.max_tokens}
}
async with self._semaphore:
await asyncio.sleep(self._request_interval)
await self._wait_for_token_budget(task.max_tokens)
try:
async with session.post(url, json=payload, headers=headers,
timeout=aiohttp.ClientTimeout(total=30)) as resp:
if resp.status == 200:
data = await resp.json()
usage = data.get("usageMetadata", {})
prompt_tokens = usage.get("promptTokenCount", 0)
self._token_budget -= prompt_tokens # 扣减输入 token
task.result = data
return data
elif resp.status == 429:
retry_after = int(resp.headers.get("Retry-After", 5))
print(f"[任务 {task.id}] 触发限流，等待 {retry_after} 秒...")
await asyncio.sleep(retry_after)
return await self._execute_single_request(session, task)
elif resp.status == 401:
task.error = "认证失败：API Key 无效"
return None
else:
error_text = await resp.text()
task.error = f"HTTP {resp.status}: {error_text}"
return None
except asyncio.TimeoutError:
task.error = "请求超时"
return None
except Exception as e:
task.error = str(e)
return None
async def process_batch(self, tasks: List[Task],
concurrency: int = 10) -> List[Task]:
"""批量处理任务列表"""
self._semaphore = asyncio.Semaphore(concurrency)
# 按优先级分组
priority_groups = [
(self.high_priority, [t for t in tasks if t.priority == 2]),
(self.medium_priority, [t for t in tasks if t.priority == 1]),
(self.low_priority, [t for t in tasks if t.priority == 0])
]
async with aiohttp.ClientSession() as session:
for queue, group in priority_groups:
if not group:
continue
print(f"开始处理 {len(group)} 个任务 (优先级: {queue == self.high_priority and '高' or queue == self.medium_priority and '中' or '低'})...")
# 使用 gather 并发执行
coroutines = [
self._execute_single_request(session, task)
for task in group
]
await asyncio.gather(*coroutines, return_exceptions=True)
return tasks
def get_stats(self) -> dict:
"""获取统计信息"""
all_tasks = list(self.high_priority) + list(self.medium_priority) + list(self.low_priority)
return {
"total": len(all_tasks),
"completed": sum(1 for t in all_tasks if t.result),
"failed": sum(1 for t in all_tasks if t.error),
"pending": sum(1 for t in all_tasks if not t.result and not t.error),
"token_budget_remaining": self._token_budget
}
使用示例
async def main():
# 初始化队列（通过 HolySheep API，延迟 <50ms）
queue = AsyncGeminiQueue(
api_key="YOUR_HOLYSHEEP_API_KEY",
rpm_limit=100,
tpm_limit=80000
)
# 创建测试任务
tasks = [
Task(id=f"task_{i}", prompt=f"翻译第 {i} 段文字",
max_tokens=500, priority=i % 3)
for i in range(50)
]
results = await queue.process_batch(tasks, concurrency=10)
stats = queue.get_stats()
print(f"处理完成！成功: {stats['completed']}, 失败: {stats['failed']}")
asyncio.run(main())四、监控与告警：实时配额追踪
光有控制还不够，我建议你加入实时监控。以下是一个简单的配额监控模块：
import requests
from datetime import datetime, timedelta
from typing import Dict, List
import time
class GeminiQuotaMonitor:
"""Gemini API 配额监控器"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self._usage_history: List[Dict] = []
self._alert_thresholds = {
"rpm_warning": 0.8, # RPM 80% 告警
"rpm_critical": 0.95, # RPM 95% 告警
"tpm_warning": 0.7, # TPM 70% 告警
"tpm_critical": 0.9 # TPM 90% 告警
}
def check_current_usage(self) -> Dict:
"""检查当前配额使用情况（通过调用 API 获取）"""
# 发起一个最小请求来获取 usage metadata
url = f"{self.base_url}/models/gemini-2.0-flash:generateContent"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"contents": [{"parts": [{"text": "test"}]}],
"generationConfig": {"maxOutputTokens": 1}
}
try:
response = requests.post(url, json=payload, headers=headers, timeout=10)
if response.status_code == 200:
data = response.json()
usage = data.get("usageMetadata", {})
record = {
"timestamp": datetime.now(),
"prompt_tokens": usage.get("promptTokenCount", 0),
"completion_tokens": usage.get("candidatesTokenCount", 0),
"total_tokens": usage.get("totalTokenCount", 0)
}
self._usage_history.append(record)
# 只保留最近 60 分钟记录
cutoff = datetime.now() - timedelta(minutes=60)
self._usage_history = [
r for r in self._usage_history
if r["timestamp"] > cutoff
]
return record
else:
return {"error": f"HTTP {response.status_code}"}
except Exception as e:
return {"error": str(e)}
def get_rate_limit_status(self, rpm_limit: int = 60,
tpm_limit: int = 60000) -> Dict:
"""计算速率限制状态"""
now = datetime.now()
minute_ago = now - timedelta(minutes=1)
# 计算最近 1 分钟的请求数和 token 数
recent_records = [
r for r in self._usage_history
if r["timestamp"] > minute_ago
]
requests_in_last_minute = len(recent_records)
tokens_in_last_minute = sum(
r["total_tokens"] for r in recent_records
)
rpm_ratio = requests_in_last_minute / rpm_limit
tpm_ratio = tokens_in_last_minute / tpm_limit
status = "正常"
alerts = []
if rpm_ratio >= self._alert_thresholds["rpm_critical"]:
status = "危险"
alerts.append(f"🔥 RPM 使用率: {rpm_ratio*100:.1f}% (接近上限)")
elif rpm_ratio >= self._alert_thresholds["rpm_warning"]:
status = "警告"
alerts.append(f"⚠️ RPM 使用率: {rpm_ratio*100:.1f}%")
if tpm_ratio >= self._alert_thresholds["tpm_critical"]:
status = "危险" if status == "危险" else "警告"
alerts.append(f"🔥 TPM 使用率: {tpm_ratio*100:.1f}% (接近上限)")
elif tpm_ratio >= self._alert_thresholds["tpm_warning"]:
alerts.append(f"⚠️ TPM 使用率: {tpm_ratio*100:.1f}%")
return {
"status": status,
"rpm": {
"used": requests_in_last_minute,
"limit": rpm_limit,
"ratio": rpm_ratio,
"remaining": rpm_limit - requests_in_last_minute
},
"tpm": {
"used": tokens_in_last_minute,
"limit": tpm_limit,
"ratio": tpm_ratio,
"remaining": tpm_limit - tokens_in_last_minute
},
"alerts": alerts,
"recommendation": self._get_recommendation(rpm_ratio, tpm_ratio)
}
def _get_recommendation(self, rpm_ratio: float, tpm_ratio: float) -> str:
"""根据使用情况给出建议"""
if rpm_ratio > 0.9:
return "建议立即降低请求频率或升级套餐，可考虑使用 HolySheep AI 高級套餐获得更高配额"
elif rpm_ratio > 0.7:
return "建议监控请求模式，考虑实现请求队列"
elif tpm_ratio > 0.8:
return "TPM 使用较高，可尝试减少单次请求的 max_tokens 参数"
else:
return "配额使用正常，继续保持"
def run_monitoring_loop(self, interval: int = 30,
rpm_limit: int = 60,
tpm_limit: int = 60000):
"""持续监控循环"""
print("🚀 开始配额监控...")
while True:
self.check_current_usage()
status = self.get_rate_limit_status(rpm_limit, tpm_limit)
print(f"\n[{datetime.now().strftime('%H:%M:%S')}] 状态: {status['status']}")
print(f" RPM: {status['rpm']['used']}/{status['rpm']['limit']} ({status['rpm']['ratio']*100:.1f}%)")
print(f" TPM: {status['tpm']['used']}/{status['tpm']['limit']} ({status['tpm']['ratio']*100:.1f}%)")
if status['alerts']:
print(" 告警:")
for alert in status['alerts']:
print(f" - {alert}")
print(f" 建议: {status['recommendation']}")
time.sleep(interval)
启动监控
monitor = GeminiQuotaMonitor(api_key="YOUR_HOLYSHEEP_API_KEY")
monitor.run_monitoring_loop(interval=30)五、价格对比：HolySheep AI vs 官方 Gemini API
说到配额管理，不得不提成本控制。我对 Gemini 2.0 Flash 的价格做了详细对比：
- Google 官方：$2.50 / 1M Output Tokens，汇率按官方 ¥7.3=$1 折算
- HolySheep AI：同样是 $2.50 / 1M Output Tokens，但汇率按 ¥1=$1 计算，相当于 节省 85%+
以我上个月的实际使用量为例：
# 假设月消耗 500 万 Output Tokens
Google 官方价格
official_cost = (5000000 / 1000000) * 2.50 # $12.50
official_cost_cny = official_cost * 7.3 # ¥91.25
HolySheep AI 价格（汇率 ¥1=$1）
holysheep_cost = (5000000 / 1000000) * 2.50 # $12.50
实际支付 ¥12.50
print(f"官方费用: ¥{official_cost_cny:.2f}")
print(f"HolySheep: ¥{holysheep_cost:.2f}")
print(f"节省: ¥{official_cost_cny - holysheep_cost:.2f} ({((official_cost_cny - holysheep_cost) / official_cost_cny * 100):.1f}%)")实际测试下来，通过 HolySheep AI 接入，延迟稳定在 40-50ms，支持微信/支付宝充值，即时到账，这对国内开发者来说体验提升非常明显。
常见报错排查
我整理了使用 Gemini API 时最常见的 5 个错误及其解决方案：
错误 1：429 Too Many Requests
# ❌ 错误写法：无限重试，不做退避
while True:
response = requests.post(url, ...)
if response.status_code != 429:
break
✅ 正确写法：指数退避 + 最大重试次数
def call_with_retry(url, payload, headers, max_retries=5):
for attempt in range(max_retries):
response = requests.post(url, json=payload, headers=headers)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# 指数退避：1s, 2s, 4s, 8s, 16s...
wait_time = 2 ** attempt + random.uniform(0, 1)
print(f"触发限流，等待 {wait_time:.1f}s...")
time.sleep(wait_time)
else:
raise Exception(f"请求失败: {response.status_code}")
raise Exception("达到最大重试次数")错误 2：401 Unauthorized
# ❌ 常见原因：API Key 配置错误或已过期
headers = {
"Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY" # 空格丢失
}
✅ 正确写法：确保格式正确
def validate_api_key(api_key: str) -> bool:
if not api_key or len(api_key) < 20:
return False
# 检查是否是 HolySheep API Key 格式
if not api_key.startswith("sk-"):
return False
# 验证连接
url = "https://api.holysheep.ai/v1/models"
headers = {"Authorization": f"Bearer {api_key}"}
try:
response = requests.get(url, headers=headers, timeout=10)
return response.status_code == 200
except:
return False
使用
if not validate_api_key("YOUR_HOLYSHEEP_API_KEY"):
raise ValueError("API Key 无效，请到 https://www.holysheep.ai/register 重新获取")错误 3：ConnectionError / Timeout
# ❌ 没有设置合理的超时
response = requests.post(url, json=payload, headers=headers) # 无限等待
✅ 设置超时 + 自动重试 + 降级策略
def call_with_fallback(prompt: str, api_key: str):
# 主服务器（HolySheep 国内直连）
primary_url = "https://api.holysheep.ai/v1/models/gemini-2.0-flash:generateContent"
# 备用服务器列表
fallback_urls = [
"https://api.holysheep.ai/v1/models/gemini-2.0-flash:generateContent", # 同一域名不同节点
]
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {"contents": [{"parts": [{"text": prompt}]}]}
for url in [primary_url] + fallback_urls:
try:
response = requests.post(
url,
json=payload,
headers=headers,
timeout=(5, 30) # (连接超时, 读取超时)
)
if response.status_code == 200:
return response.json()
except requests.exceptions.Timeout:
print(f"超时，尝试下一个节点...")
continue
except requests.exceptions.ConnectionError:
print(f"连接失败，尝试下一个节点...")
continue
# 所有节点都失败时的降级处理
return {"error": "所有 API 节点均不可用，请检查网络或稍后重试"}错误 4：400 Bad Request - Invalid Input
# ❌ 常见错误：prompt 格式不规范
payload = {
"contents": "你叫什么名字" # 字符串格式错误
}
✅ 正确格式：嵌套的 parts 数组
payload = {
"contents": [
{
"parts": [
{"text": "你叫什么名字？"}
]
}
]
}
更健壮的构建方式
def build_content(prompt: str, system_instruction: str = None) -> dict:
content = {
"contents": [
{
"parts": [{"text": prompt}]
}
]
}
if system_instruction:
content["systemInstruction"] = {
"parts": [{"text": system_instruction}]
}
return content
payload = build_content("总结这篇文章的内容", "你是一个专业的文本摘要助手")错误 5：503 Service Unavailable
# ❌ 没有处理服务不可用的情况
response = requests.post(url, ...)
result = response.json() # 如果服务挂了，这里会报错
✅ 添加服务可用性检查 + 优雅降级
def check_service_health(api_key: str) -> bool:
"""检查 API 服务健康状态"""
url = "https://api.holysheep.ai/v1/models"
headers = {"Authorization": f"Bearer {api_key}"}
try:
response = requests.get(url, headers=headers, timeout=5)
return response.status_code == 200
except:
return False
def call_with_health_check(prompt: str, api_key: str):
if not check_service_health(api_key):
return {
"status": "degraded",
"message": "服务暂时不可用，请稍后重试",
"fallback_available": True,
"fallback_model": "gemini-1.5-flash" # 使用轻量模型降级
}
# 正常调用逻辑
...总结
Gemini API 的配额管理核心就三点：理解限制规则 → 实现智能限流 → 做好监控告警。
在实际项目中，我建议：
- 生产环境必做请求队列和重试机制
- 关键业务加上超时和降级策略
- 监控仪表盘要实时显示 RPM/TPM 使用率
- 成本敏感型场景优先选择 HolySheheep AI，汇率优势 + 国内直连 + 微信/支付宝充值，调试体验完全不在一个层次
如果你对具体实现有疑问，或者想了解如何设计更复杂的限流算法（比如令牌桶 vs 漏桶），欢迎在评论区交流。