Gemini API 429、503 怎么解决：查配额、读错误与正确重试

{
  "error": {
    "code": 429,
    "status": "RESOURCE_EXHAUSTED",
    "message": "Quota exceeded",
    "details": [
      {
        "@type": "type.googleapis.com/google.rpc.QuotaFailure",
        "violations": [
          {
            "quotaId": "ExampleRequestsPerMinute",
            "quotaValue": "20"
          }
        ]
      },
      {
        "@type": "type.googleapis.com/google.rpc.RetryInfo",
        "retryDelay": "12s"
      }
    ]
  }
}

import json
import math
import os
import random
import re
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime


def error_policy(status, body):
    error = body.get("error", {}) if isinstance(body, dict) else {}
    details = error.get("details", []) if isinstance(error, dict) else []
    quota_ids, delays = [], []
    details = details if isinstance(details, list) else []
    for item in details:
        if not isinstance(item, dict):
            continue
        if item.get("@type") == "type.googleapis.com/google.rpc.QuotaFailure":
            violations = item.get("violations", [])
            if not isinstance(violations, list):
                violations = []
            for violation in violations:
                quota_ids.append(str(violation.get("quotaId", ""))
                                 if isinstance(violation, dict) else "")
        if item.get("@type") == "type.googleapis.com/google.rpc.RetryInfo":
            value = str(item.get("retryDelay", ""))
            if re.fullmatch(r"\d+(?:\.\d{1,9})?s", value):
                delays.append(float(value[:-1]))
    if status == 429:
        if any("perday" in q.lower() for q in quota_ids):
            return "quota_reset", 0
        if not quota_ids or not all("perminute" in q.lower() for q in quota_ids):
            return "inspect_quota", 0
    elif status not in (500, 502, 503, 504):
        return "fix_request", 0
    return "retry", max(delays, default=0)


def retry_after(value):
    if value is None:
        return 0
    try:
        seconds = float(value)
    except ValueError:
        try:
            seconds = (parsedate_to_datetime(value) - datetime.now(timezone.utc)).total_seconds()
        except (TypeError, ValueError, OverflowError):
            return 0
    return max(0, seconds) if math.isfinite(seconds) else 0


def generate(payload, model="gemini-3.1-pro-preview"):
    waited = 0.0
    for attempt in range(4):
        request = urllib.request.Request(
            f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent",
            data=json.dumps(payload).encode(),
            headers={"Content-Type": "application/json",
                     "x-goog-api-key": os.environ["GEMINI_API_KEY"]},
            method="POST"
        )
        try:
            with urllib.request.urlopen(request, timeout=20) as response:
                return json.load(response)
        except urllib.error.HTTPError as error:
            raw = error.read().decode("utf-8", errors="replace")
            try:
                body = json.loads(raw)
            except json.JSONDecodeError:
                body = {}
            action, hint = error_policy(error.code, body)
            print(f"第{attempt + 1}次，HTTP {error.code}，处理：{action}")
            if action != "retry" or attempt == 3:
                raise RuntimeError(raw) from error
            delay = max(hint, retry_after(error.headers.get("Retry-After")),
                        random.uniform(0, min(8, 2 ** attempt)))
            if waited + delay > 30:
                raise RuntimeError("建议等待时间超出本轮预算，停止发送") from error
        time.sleep(delay)
        waited += delay
    raise RuntimeError("尝试次数已耗尽")


if __name__ == "__main__":
    result = generate({
        "contents": [{"parts": [{"text": "请只回复：连接检查完成"}]}],
        "generationConfig": {"thinkingConfig": {"thinkingLevel": "LOW"}}
    })
    candidates = result.get("candidates", [])
    if not candidates or candidates[0].get("finishReason") != "STOP":
        raise RuntimeError("生成未正常结束，请检查返回正文")
    parts = candidates[0].get("content", {}).get("parts", [])
    text = "".join(p.get("text", "") for p in parts if not p.get("thought"))
    if not text.strip():
        raise RuntimeError("没有最终文本，请检查返回正文")
    print(text)
    print("实际用量：", result.get("usageMetadata", {}))