Gemini API 503 エラー（Service Unavailable）の原因と解決策 — 対処法とリトライ実装

import google.generativeai as genai
import time
import random
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
def call_gemini_with_retry(prompt: str, max_retries: int = 5) -> str:
    """503エラーに対して指数バックオフでリトライする"""
    model = genai.GenerativeModel("gemini-2.5-flash")
 
    for attempt in range(max_retries):
        try:
            response = model.generate_content(prompt)
            return response.text
 
        except Exception as e:
            error_str = str(e)
 
            # 503エラー（UNAVAILABLE）を検出
            if "503" in error_str or "UNAVAILABLE" in error_str or "overloaded" in error_str.lower():
                if attempt < max_retries - 1:
                    # 指数バックオフ: 2^attempt 秒 + ランダムなジッター
                    wait_seconds = (2 ** attempt) + random.uniform(0, 1)
                    print(f"⚠️  503エラー（試行 {attempt + 1}/{max_retries}）— {wait_seconds:.1f}秒後にリトライします...")
                    time.sleep(wait_seconds)
                    continue
                else:
                    print("❌ 最大リトライ回数に達しました。")
                    raise
 
            # 503以外のエラーはリトライしない
            raise
 
    return ""  # ここには到達しない
 
# 使用例
try:
    result = call_gemini_with_retry("Pythonで指数バックオフを実装する方法を教えてください。")
    print(result)
except Exception as e:
    print(f"最終的なエラー: {e}")

import { GoogleGenerativeAI } from "@google/generative-ai";
 
const genAI = new GoogleGenerativeAI(process.env.GEMINI_API_KEY!);
 
async function callGeminiWithRetry(
  prompt: string,
  maxRetries: number = 5
): Promise<string> {
  const model = genAI.getGenerativeModel({ model: "gemini-2.5-flash" });
 
  for (let attempt = 0; attempt < maxRetries; attempt++) {
    try {
      const result = await model.generateContent(prompt);
      return result.response.text();
    } catch (error: unknown) {
      const errorMessage = error instanceof Error ? error.message : String(error);
      const is503 =
        errorMessage.includes("503") ||
        errorMessage.toLowerCase().includes("unavailable") ||
        errorMessage.toLowerCase().includes("overloaded");
 
      if (is503 && attempt < maxRetries - 1) {
        const waitMs = Math.pow(2, attempt) * 1000 + Math.random() * 1000;
        console.warn(
          `⚠️  503エラー（試行 ${attempt + 1}/${maxRetries}）— ${(waitMs / 1000).toFixed(1)}秒後にリトライ`
        );
        await new Promise((resolve) => setTimeout(resolve, waitMs));
        continue;
      }
      throw error;
    }
  }
  throw new Error("最大リトライ回数に達しました");
}

import vertexai
from vertexai.generative_models import GenerativeModel
 
REGIONS = ["us-central1", "us-east4", "europe-west4"]
 
def call_with_region_fallback(prompt: str, project_id: str) -> str:
    """503が続く場合に別リージョンへフォールバック"""
    for region in REGIONS:
        try:
            vertexai.init(project=project_id, location=region)
            model = GenerativeModel("gemini-2.5-flash")
            response = model.generate_content(prompt)
            print(f"✅ リージョン {region} で成功")
            return response.text
        except Exception as e:
            if "503" in str(e) or "UNAVAILABLE" in str(e):
                print(f"⚠️  リージョン {region} で503エラー — 次のリージョンへ")
                continue
            raise
    raise RuntimeError("すべてのリージョンで503エラーが発生しました")

import unittest
from unittest.mock import patch, MagicMock
 
class TestGeminiRetry(unittest.TestCase):
    def test_retry_on_503(self):
        """503エラーでリトライして最終的に成功するシナリオ"""
        mock_model = MagicMock()
        # 1回目と2回目は503、3回目は成功
        mock_model.generate_content.side_effect = [
            Exception("503 UNAVAILABLE: The model is overloaded."),
            Exception("503 UNAVAILABLE: The model is overloaded."),
            MagicMock(text="成功レスポンス"),
        ]
 
        with patch("google.generativeai.GenerativeModel", return_value=mock_model):
            result = call_gemini_with_retry("テスト", max_retries=3)
            self.assertEqual(result, "成功レスポンス")
            self.assertEqual(mock_model.generate_content.call_count, 3)
            print("✅ テスト合格：503後のリトライが正常に動作しています")
 
if __name__ == "__main__":
    unittest.main()

wait_seconds = min(60, (2 ** attempt) + random.uniform(0, 1))

from google.api_core import retry
 
@retry.Retry(predicate=retry.if_transient_error)
def my_api_call():
    # この関数は一時的なエラー（503含む）で自動リトライされる
    return model.generate_content("テスト")