Gemini API のエラーを status で見分ける — 429・SAFETY・トークン超過を本番で止めない実装

import google.generativeai as genai
 
# 方法1: 直接設定（開発用のみ推奨）
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
# 方法2: 環境変数から読み込み（本番環境推奨）
import os
api_key = os.getenv("GEMINI_API_KEY")
genai.configure(api_key=api_key)

export GEMINI_API_KEY="your_actual_api_key"

import google.generativeai as genai
from google.auth import default
 
# GCPの認証を自動取得
credentials, project_id = default()
genai.configure(credentials=credentials)

import time
import google.generativeai as genai
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.0-flash")
 
def generate_with_retry(prompt, max_retries=5):
    """指数バックオフを使ったリトライロジック"""
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt)
            return response.text
        except Exception as e:
            if "429" in str(e):
                # 待機時間: 1秒、2秒、4秒、8秒、16秒...
                wait_time = 2 ** attempt
                print(f"レート制限に達しました。{wait_time}秒待機します...")
                time.sleep(wait_time)
            else:
                raise
 
# 使用例
result = generate_with_retry("日本の首都は？")
print(result)

import { GoogleGenerativeAI } from "@google/generative-ai";
 
const genAI = new GoogleGenerativeAI("YOUR_GEMINI_API_KEY");
const model = genAI.getGenerativeModel({ model: "gemini-2.0-flash" });
 
async function generateWithRetry(prompt: string, maxRetries = 5): Promise<string> {
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const response = await model.generateContent(prompt);
      return response.response.text();
    } catch (error) {
      if (error instanceof Error && error.message.includes("429")) {
        const waitTime = Math.pow(2, attempt) * 1000; // ミリ秒に変換
        console.log(`レート制限に達しました。${waitTime / 1000}秒待機します...`);
        await new Promise((resolve) => setTimeout(resolve, waitTime));
      } else {
        throw error;
      }
    }
  }
  throw new Error("最大リトライ回数に達しました");
}
 
// 使用例
const result = await generateWithRetry("日本の首都は？");
console.log(result);

{
  "error": {
    "code": 429,
    "message": "Resource has been exhausted (e.g. check quota).",
    "status": "RESOURCE_EXHAUSTED"
  }
}

displayName: Gemini API quota alert
conditions:
  - displayName: RPM usage > 80%
    conditionThreshold:
      filter: |
        resource.type="api"
        AND metric.type="serviceruntime.googleapis.com/api/consumer/quota_used_count"
        AND resource.service="generativelanguage.googleapis.com"
      threshold_value: 0.8
      comparison: COMPARISON_GT

import time
 
def generate_with_500_retry(prompt, max_retries=3):
    """500/503エラーに対応したリトライロジック"""
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt)
            return response.text
        except Exception as e:
            error_code = str(e)
            if "500" in error_code or "503" in error_code:
                wait_time = 5 * (attempt + 1)  # 5秒、10秒、15秒
                print(f"サーバーエラーが発生しました。{wait_time}秒後に再試行します...")
                time.sleep(wait_time)
            else:
                raise

import google.generativeai as genai
import os
 
genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
model = genai.GenerativeModel("gemini-2.0-flash")
 
def count_tokens(prompt: str) -> int:
    # 送信前にトークン数を確認する
    return model.count_tokens(prompt).total_tokens
 
def safe_generate(prompt: str, max_chars: int = 30000) -> str:
    if len(prompt) > max_chars:
        # 末尾を切るより中間を省く方が、前後の文脈が保たれます
        half = max_chars // 2
        prompt = prompt[:half] + "\n...[中略]...\n" + prompt[-half:]
    return model.generate_content(prompt).text

response = model.generate_content("...")
print(response.text)  # AttributeError: 'NoneType' object has no attribute 'text'

def safe_get_text(response) -> str | None:
    # candidates が空なら入力段階でブロックされている
    if not response.candidates:
        print(f"入力がブロックされました: {response.prompt_feedback}")
        return None
 
    candidate = response.candidates[0]
    reason = candidate.finish_reason.name
 
    if reason == "SAFETY":
        print(f"安全フィルターでブロック: {candidate.safety_ratings}")
        return None
    if reason == "RECITATION":
        print("著作権コンテンツの検出によりブロック")
        return None
    if reason == "MAX_TOKENS":
        # 途中までのテキストは取り出せる
        print("max_output_tokens に達しました。上限を増やしてください")
 
    try:
        return candidate.content.parts[0].text
    except (IndexError, AttributeError):
        return None

import time
import random
import google.generativeai as genai
from google.api_core import exceptions as gax
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
model = genai.GenerativeModel("gemini-2.0-flash")
 
# 待てば回復する見込みのある状態
RETRYABLE = (gax.ResourceExhausted, gax.ServiceUnavailable, gax.InternalServerError)
# 直さない限り回復しない状態
FATAL = (gax.InvalidArgument, gax.PermissionDenied, gax.Unauthenticated)
 
def call_gemini(prompt, max_retries=5, base=1.0):
    for attempt in range(max_retries):
        try:
            return model.generate_content(prompt)
        except RETRYABLE as e:
            # 指数バックオフ + ジッター（同時リトライの集中を避ける）
            wait = base * (2 ** attempt) + random.uniform(0, base)
            print(f"[retry {attempt + 1}] {type(e).__name__}: {wait:.1f}s 待機")
            time.sleep(wait)
        except FATAL as e:
            # ここで待つのは時間の無駄。呼び出し側へ即返す
            raise RuntimeError(f"修正が必要なエラー: {type(e).__name__}: {e}") from e
    raise RuntimeError("最大リトライ回数を超過しました")

import logging
 
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
 
try:
    response = model.generate_content("テスト")
    logger.info(f"成功: {response.text}")
except Exception as e:
    logger.error(f"エラーが発生しました: {e}")