批量打标脚本用线程池并发丢两千条数据,跑到三百多条开始整片报错:
{
"error": {
"code": 429,
"message": "Resource has been exhausted (e.g. check quota).",
"status": "RESOURCE_EXHAUSTED"
}
}
换一个 key 照样红,问题在我把并发当无限资源。
这里先记一个区分方法:立刻重试就能过,是短时并发撞了 RPM;隔几分钟才缓过来,是撞了 RPD;只有长 prompt 批量跑的时候才红,是 TPM 被吃满。三种配额要用的应对策略不一样,别一律加 sleep。
撞到任意一个维度都返回 429:
免费层和付费层走不同配额池,拿免费额度跑生产任务,几轮就耗光当天额度。
三个维度里 RPD 最难受:撞了当天就回不来,只能等窗口重置,重试写得再聪明也没用。夜间集中跑批量的任务,动手前先把一天要发多少次请求算清楚。
多数 SDK 只抛 message,retryDelay 得看原始响应:
curl -s -X POST \
"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-pro:generateContent" \
-H "x-goog-api-key: $GEMINI_API_KEY" \
-H "Content-Type: application/json" \
-d '{"contents":[{"parts":[{"text":"hi"}]}]}' | jq '.error.details'
这个秒数通常是服务端算出的真实窗口,按它来比拍脑袋定值靠谱。免费额度层的 429 常常连 details 都不返回,那就只能用退避加抖动的兜底策略硬扛:
[
{
"@type": "type.googleapis.com/google.rpc.RetryInfo",
"retryDelay": "17s"
}
]
固定 sleep 最坑:并发打满时所有线程同时醒来再同时撞墙,退避必须带抖动。
import random
def backoff_delay(attempt: int, base: float = 1.0, cap: float = 60.0) -> float:
# 指数增长封顶后乘随机抖动
raw = min(cap, base * (2 ** attempt))
return raw * (0.5 + random.random() * 0.5)
不想手写就用 tenacity:
from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential_jitter
@retry(
retry=retry_if_exception_type(RateLimitError),
wait=wait_exponential_jitter(initial=1, max=60),
stop=stop_after_attempt(6),
)
def call_gemini(payload):
return client.generate(payload)
多 Key 轮询只能兜底,还要配熔断:同一账号下的 Key 共享配额池,只有不同账号或项目的 Key 才分开算。
同一个脚本在公司网络报 403,切手机热点就正常,基本是区域问题。
还有个判断技巧:看出错的是整批请求还是一部分。全部 403 更像区域或权限问题;只有某几个模型 403,说明是那个模型没在你的项目里开权限。云服务器上跑 Gemini 报 403 的概率远高于家用宽带,这是机房出口 IP 的常见待遇,跟代码无关。
国内要稳定调 Gemini,走中转最省事:请求打到中转域名,跨境那段由对方处理,业务代码不用动。小鱼API 这类聚合中转的 https://xyuapi.top/v1 可以直接当 OpenAI 兼容端点接进现有客户端。
401 是身份问题,403 是授权问题。列模型能返回 200,说明 key 身份成立:
curl -s -o /dev/null -w "%{http_code}\n" \
"https://generativelanguage.googleapis.com/v1beta/models" \
-H "x-goog-api-key: $GEMINI_API_KEY"
生成接口却 403,就是项目权限或区域链路问题。挂代理再测同样请求,能通即说明出口 IP 被拦。
400 的 message 经常直接点名哪个字段有问题,只看 code 会绕很远。
先把 message 完整打出来再改代码,比对着文档猜字段名快得多:
curl -s -X POST "$BASE/models/gemini-2.5-pro:generateContent" \
-H "x-goog-api-key: $GEMINI_API_KEY" \
-H "Content-Type: application/json" \
-d @payload.json | jq -r '.error.message'
大小写混用、topK 和 topP 塞错层级、parts 为空数组、maxOutputTokens 越界,这些都是 400 常客。
#错误写法
bad_payload = {
"contents": [{"parts": []}],
"generation_config": { # 应为 generationConfig 驼峰键名
"top_k": 40, # 正确键名是 topK
"temperature": 0.7,
"max_tokens": 999999, # 正确键名是 maxOutputTokens
},
}
#正确写法
good_payload = {
"contents": [{"role": "user", "parts": [{"text": "写一段快排"}]}],
"generationConfig": {
"temperature": 0.7,
"topP": 0.95, # topP 和 topK 建议只调一个
"maxOutputTokens": 2048,
},
}
我习惯发送前 json.dumps 打一遍,肉眼过键名。结构化输出也一样,schema 少一个 required 字段就报 400,并排看最快。
名字是「gemini + 版本 + 变体」结构,如 gemini-2.5-pro、gemini-3-pro-preview、gemini-3.1-pro-preview,精确匹配,漏掉 -preview 或按别家风格去拼都直接 404。
generateContent 一次性返回,streamGenerateContent 才是流式。想要流式却写了前者,客户端会一直等完整响应,像卡死。新模型往往先在 v1beta 上线。
#流式要带 alt=sse,否则是一整块 JSON
curl -N "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-pro:streamGenerateContent?alt=sse" \
-H "x-goog-api-key: $GEMINI_API_KEY" \
-H "Content-Type: application/json" \
-d '{"contents":[{"parts":[{"text":"数到十"}]}]}'
报 404 时先把可用模型列出来,看名字在不在里面,比翻文档快:
curl -s "https://generativelanguage.googleapis.com/v1beta/models" \
-H "x-goog-api-key: $GEMINI_API_KEY" | jq -r '.models[].name'
500 INTERNAL 和 503 UNAVAILABLE 都是服务端状态。同一份 payload 五分钟前能过、现在不行、几分钟后又好了,跟你的代码没关系。
FALLBACK_CHAIN = ["gemini-3.1-pro-preview", "gemini-3-pro-preview", "gemini-2.5-pro"]
def call_with_fallback(payload):
last_error = None
for model in FALLBACK_CHAIN:
try:
return client.generate(model=model, payload=payload)
except UnavailableError as exc:
last_error = exc # 服务端过载,换模型
raise last_error
顺序按任务重要程度排,主模型最强,末端放便宜稳定的收口。走中转时同一个 Key 换模型名就能降级,不用另配凭据;按次计费与字数无关,降级也不跳档。
按次计费的参考价:
| 模型 | 单次价格 |
|---|---|
| gemini-2.5-pro | 0.031元/次 |
| gemini-3-pro-preview | 0.05元/次 |
| gemini-3.1-pro-preview | 0.09元/次 |
| claude-sonnet-4-5-thinking | 0.09元/次 |
| claude-opus-4-5-thinking | 0.12元/次 |
| claude-opus-4-6-thinking | 0.25元/次 |
| claude-opus-4-7-thinking | 0.25元/次 |
| grok-4.1 | 0.05元/次 |
| grok-4.5 | 0.09元/次 |
| deepseek-v4-pro-thinking | 0.09元/次 |
| gpt-5.5 | 0.2元/次 |
| kimi-k2.6 | 0.09元/次 |
原生 Gemini 接口用 x-goog-api-key 头,OpenAI 兼容接口用 Authorization: Bearer。
// 错误写法:原生接口路径上挂 Bearer
{
"url": "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-pro:generateContent",
"headers": { "Authorization": "Bearer sk-xxxx" }
}
// 正确写法:原生接口用 x-goog-api-key
{
"url": "https://generativelanguage.googleapis.com/v1beta/models/gemini-2.5-pro:generateContent",
"headers": { "x-goog-api-key": "AIza...", "Content-Type": "application/json" }
}
// 正确写法:OpenAI 兼容端点用 Bearer
{
"url": "https://xyuapi.top/v1/chat/completions",
"headers": { "Authorization": "Bearer sk-xxxx", "Content-Type": "application/json" }
}
排查 401 先别跑业务请求,用最小脚本验证 key:
import os
import requests
BASE = os.environ.get("GEMINI_BASE_URL", "https://generativelanguage.googleapis.com/v1beta")
KEY = os.environ.get("GEMINI_API_KEY", "").strip() # strip 复制来的空格
assert KEY, "GEMINI_API_KEY 没加载"
resp = requests.get(f"{BASE}/models", headers={"x-goog-api-key": KEY}, timeout=20)
print("HTTP", resp.status_code)
if resp.status_code == 200:
print("key 可用,模型数量:", len(resp.json().get("models", [])))
elif resp.status_code == 401:
print("key 无效或被删,去后台重新生成")
elif resp.status_code == 403:
print("key 有效但权限受限,见 403 节")
else:
print(resp.text[:300])
超时和 connection reset 属于网络层,跨境链路抖动、DNS 慢、代理不稳都会导致。固定失败在相近秒数多半是 timeout 设短了,随机失败多半是链路问题。这类失败要和配额类失败分开统计,别拿着网络问题去调配额。
小图内嵌,大文件上传后走 File API 引用。内嵌字段是 inline_data(REST 里是 inlineData),base64 里不能带 data:image/png;base64, 前缀。
#错误写法:带了 data URI 前缀
bad = {"inline_data": {"mime_type": "image/png", "data": "data:image/png;base64,iVBORw0KGgo..."}}
#正确写法:纯 base64 + 单独 mimeType
import base64
import pathlib
raw = pathlib.Path("chart.png").read_bytes()
good = {
"inline_data": {
"mime_type": "image/png",
"data": base64.b64encode(raw).decode("utf-8"),
}
}
内嵌请求有体积上限,几兆原图很容易失败。传图前压到长边 1568 像素以内、1MB 上下,识别质量几乎没差别。
from PIL import Image
img = Image.open("chart.png")
img.thumbnail((1568, 1568)) # 长边限制在 1568 以内
img.convert("RGB").save("chart_small.jpg", quality=85)
压完看一眼文件大小,超过 1MB 就把 quality 继续往下调,别指望接口替你兜底。
jpg 字节流标成 image/png 这种隐形坑,有时能过有时报错,后缀和真实编码不一致时用程序探测。远程 URL 也不可靠,服务端抓取会被反爬拦掉,自己先下载再上传。PDF 和视频别内嵌,用 File API 拿 file_uri,视频要等状态变 ACTIVE。
有一类「报错」其实是 200 成功,candidates 空或 text 是空串。别查代码,先看两个字段:
def inspect_response(data: dict) -> str:
feedback = data.get("promptFeedback", {})
if feedback.get("blockReason"):
return f"输入被拦截: {feedback['blockReason']}"
candidates = data.get("candidates") or []
if not candidates:
return "无候选结果"
finish = candidates[0].get("finishReason", "STOP")
if finish == "MAX_TOKENS":
return "输出被截断,调大 maxOutputTokens"
if finish in ("SAFETY", "RECITATION"):
return f"生成被拦截: {finish}"
parts = candidates[0].get("content", {}).get("parts", [])
return "".join(p.get("text", "") for p in parts)
档位从 BLOCK_NONE、BLOCK_ONLY_HIGH、BLOCK_MEDIUM_AND_ABOVE 到 BLOCK_LOW_AND_ABOVE,越往右越严,默认偏保守,做内容审核会被误拦。
payload = {
"contents": [{"role": "user", "parts": [{"text": prompt}]}],
"safetySettings": [
{"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_ONLY_HIGH"},
{"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_ONLY_HIGH"},
{"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_ONLY_HIGH"},
{"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_ONLY_HIGH"},
],
}
放开阈值只在可控的批处理里用,面向用户的产品别照抄。拦截不是 bug 而是策略,正确做法是记录并改写 prompt 重试。
改写时别只加一句「请用温和的语言」就重发,这类句式对审核几乎不起作用。实测更有效的做法是把触发词换成同义表述、把指令改写成客观的第三人称描述,必要时拆成两步:先让模型复述需求,再单独执行具体任务。批处理脚本里把被拦的 prompt 和 blockReason 一起落库,攒够样本再分析,比逐条现场猜效率高得多。
| HTTP 码 | 状态名 | 常见原因 | 处理动作 |
|---|---|---|---|
| 400 | INVALID_ARGUMENT | generationConfig 键名或取值不对、parts 为空 | 打印请求体核对键名 |
| 401 | UNAUTHENTICATED / API_KEY_INVALID | key 带空格、被删、环境变量没加载、请求头写错 | 跑 key 自检脚本 |
| 403 | PERMISSION_DENIED | 区域或 IP 限制、项目未启用接口、权限不足 | 列模型验证,必要时走中转 |
| 404 | NOT_FOUND | 模型名写错、漏 -preview、端点路径不对 | 列可用模型比对 |
| 429 | RESOURCE_EXHAUSTED | RPM/TPM/RPD 撞线、并发打满、免费额度用尽 | 读 retryDelay 退避重试 |
| 500 | INTERNAL | 服务端偶发错误 | 退避重试,失败走降级链 |
| 503 | UNAVAILABLE | 服务端过载 | 退避重试加降级 |
400、401、404 重试一百次结果一样,只浪费时间;429、500、503 才值得退避。把判断写成函数,业务里就不会到处散落 try。
还有一类容易误判的:超时和连接被重置。它们不是 HTTP 状态码,抓不到 resp.status_code,得靠 requests 的 RequestException 或 socket.timeout 去捕获。这类失败要跟配额类失败分开统计,否则你会拿着网络问题反复去调配额,越调越乱。重试前还要确认请求本身幂等:纯生成任务重发没有副作用,带外部写入动作的工具调用重发就可能造成重复落库。
import json
import os
import random
import time
import requests
KEY = os.environ.get("GEMINI_API_KEY", "").strip()
BASE = os.environ.get("GEMINI_BASE_URL", "https://generativelanguage.googleapis.com/v1beta")
CHAIN = ["gemini-3.1-pro-preview", "gemini-3-pro-preview", "gemini-2.5-pro"]
RETRYABLE = {429, 500, 503}
class GeminiError(Exception):
def __init__(self, status, message):
super().__init__(f"HTTP {status}: {message}")
self.status = status
self.message = message
def backoff_delay(attempt, base=1.0, cap=60.0):
return min(cap, base * (2 ** attempt)) * (0.5 + random.random() * 0.5)
def parse_response(data):
"""返回 (文本, 备注)"""
feedback = data.get("promptFeedback", {})
if feedback.get("blockReason"):
return "", f"输入被拦截:{feedback['blockReason']}"
candidates = data.get("candidates") or []
if not candidates:
return "", "无候选结果"
finish = candidates[0].get("finishReason", "STOP")
parts = candidates[0].get("content", {}).get("parts", [])
text = "".join(p.get("text", "") for p in parts)
if finish == "MAX_TOKENS":
return text, "输出被截断"
if finish in ("SAFETY", "RECITATION"):
return text, f"生成被拦截:{finish}"
return text, ""
def generate(prompt, model, max_output_tokens=1024, max_attempts=4):
url = f"{BASE}/models/{model}:generateContent"
payload = {
"contents": [{"role": "user", "parts": [{"text": prompt}]}],
"generationConfig": {"temperature": 0.7, "maxOutputTokens": max_output_tokens},
}
last_error = None
for attempt in range(max_attempts):
try:
resp = requests.post(
url,
headers={"x-goog-api-key": KEY, "Content-Type": "application/json"},
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
timeout=60,
)
except requests.RequestException as exc:
last_error = exc
time.sleep(backoff_delay(attempt))
continue
if resp.status_code == 200:
return parse_response(resp.json())
if resp.status_code in RETRYABLE:
last_error = GeminiError(resp.status_code, resp.text[:200])
wait = backoff_delay(attempt)
retry_after = resp.headers.get("Retry-After")
if retry_after and retry_after.isdigit():
wait = max(wait, float(retry_after))
time.sleep(wait)
continue
raise GeminiError(resp.status_code, resp.text[:200]) # 不可重试,立刻抛
raise last_error
def generate_with_fallback(prompt, **kwargs):
last_error = None
for model in CHAIN:
try:
return generate(prompt, model, **kwargs)
except GeminiError as exc:
if exc.status in (404, 500, 503):
last_error = exc
continue
raise
raise last_error
if __name__ == "__main__":
text, note = generate_with_fallback("解释什么是限流")
print(note or "OK")
print(text)
关键点:不可重试的状态码立刻抛出,不在循环里耗;可重试的按退避等待;模型不存在或服务端过载就换下一个模型。跑起来之后 429 和 503 基本不会再打断任务。
处理逻辑写完只算一半,另一半是知道线上究竟在报什么。把每次失败按状态码分桶打点,然后按天看趋势:400 和 401 突然变多,多半是你自己刚改了代码或换了 Key;429 往上走,说明量涨了或者并发开太大;500 和 503 有起伏属于上游正常抖动,只要退避和降级兜住了就不用管。分桶看几天你会发现,问题往往集中在某一个状态码上,比一条条翻日志快得多。
Gemini 这些报错摸清规律都不算玄学,剩下的交给重试和降级。我把上游挂在 https://xyuai.cc 这类聚合中转上,一个 Key 通吃全系列,国内直连不用折腾代理。