Gemini API 本番運用ノート — 429・500・503 に静かに耐えるエラーハンドリングとレート制限の設計

import random
import time
import asyncio
from google import genai
 
# Full Jitter 実装
class GeminiRetryClient:
    def __init__(
        self,
        api_key: str,
        max_retries: int = 5,
        base_delay: float = 1.0,
        max_delay: float = 60.0
    ):
        self.client = genai.Client(api_key=api_key)
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.max_delay = max_delay
 
    def _calculate_delay_full_jitter(self, attempt: int) -> float:
        """Full Jitter: sleep = random(0, min(cap, base * 2^attempt))"""
        exp_delay = self.base_delay * (2 ** attempt)
        capped_delay = min(exp_delay, self.max_delay)
        return random.uniform(0, capped_delay)
 
    def _calculate_delay_equal_jitter(self, attempt: int) -> float:
        """Equal Jitter: sleep = base_delay/2 + random(0, base_delay/2)"""
        exp_delay = self.base_delay * (2 ** attempt)
        capped_delay = min(exp_delay, self.max_delay)
        half = capped_delay / 2
        return half + random.uniform(0, half)
 
    def _calculate_delay_decorrelated(self, prev_delay: float) -> float:
        """Decorrelated Jitter: sleep = min(cap, random(base, prev * 3))"""
        return min(self.max_delay, random.uniform(self.base_delay, prev_delay * 3))
 
    async def generate_with_retry(self, model: str, prompt: str) -> str:
        """リトライ付きでコンテンツを生成する"""
        last_exception = None
 
        for attempt in range(self.max_retries + 1):
            try:
                response = self.client.models.generate_content(
                    model=model,
                    contents=prompt
                )
                return response.text
 
            except Exception as e:
                last_exception = e
                status_code = getattr(e, 'status_code', 500)
 
                if not is_retryable(status_code):
                    raise  # リトライ不可のエラーは即座に再送出
 
                if attempt == self.max_retries:
                    break  # 最大リトライ回数に到達
 
                delay = self._calculate_delay_full_jitter(attempt)
                print(f"[Retry {attempt + 1}/{self.max_retries}] "
                      f"{status_code} エラー → {delay:.2f}秒後にリトライ")
                await asyncio.sleep(delay)
 
        raise last_exception  # 全リトライ失敗
 
# 使用例
# client = GeminiRetryClient(api_key="YOUR_API_KEY")
# result = await client.generate_with_retry(
#     model="gemini-2.5-flash",
#     prompt="日本の四季について説明してください"
# )
# print(result)

def get_retry_delay(error, attempt: int, base_delay: float = 1.0) -> float:
    """Retry-Afterヘッダーを優先し、なければFull Jitterを使う"""
    retry_after = getattr(error, 'retry_after', None)
    if retry_after is not None:
        # サーバー指定の待機時間 + 小さなジッター
        return float(retry_after) + random.uniform(0, 1.0)
    # フォールバック: Full Jitter
    return random.uniform(0, min(60.0, base_delay * (2 ** attempt)))

import time
from enum import Enum
from threading import Lock
 
class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"
 
class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 30.0,
        half_open_max_calls: int = 3
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls
 
        self._state = CircuitState.CLOSED
        self._failure_count = 0
        self._success_count = 0
        self._last_failure_time = 0.0
        self._half_open_calls = 0
        self._lock = Lock()
 
    @property
    def state(self) -> CircuitState:
        with self._lock:
            if self._state == CircuitState.OPEN:
                # クールダウン期間が経過したらHALF_OPENに遷移
                if time.time() - self._last_failure_time >= self.recovery_timeout:
                    self._state = CircuitState.HALF_OPEN
                    self._half_open_calls = 0
            return self._state
 
    def allow_request(self) -> bool:
        """リクエストを許可するかどうかを判定する"""
        current_state = self.state
        if current_state == CircuitState.CLOSED:
            return True
        elif current_state == CircuitState.HALF_OPEN:
            with self._lock:
                if self._half_open_calls < self.half_open_max_calls:
                    self._half_open_calls += 1
                    return True
            return False
        return False  # OPEN状態
 
    def record_success(self):
        """成功を記録する"""
        with self._lock:
            if self._state == CircuitState.HALF_OPEN:
                self._success_count += 1
                if self._success_count >= self.half_open_max_calls:
                    # 十分な成功 → CLOSEDに復帰
                    self._state = CircuitState.CLOSED
                    self._failure_count = 0
                    self._success_count = 0
            elif self._state == CircuitState.CLOSED:
                self._failure_count = 0
 
    def record_failure(self):
        """失敗を記録する"""
        with self._lock:
            self._failure_count += 1
            self._last_failure_time = time.time()
            if self._state == CircuitState.HALF_OPEN:
                # HALF_OPENでの失敗 → 再びOPEN
                self._state = CircuitState.OPEN
                self._success_count = 0
            elif self._failure_count >= self.failure_threshold:
                self._state = CircuitState.OPEN
 
# 使用例
# breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30.0)
# if breaker.allow_request():
#     try:
#         result = call_gemini_api(prompt)
#         breaker.record_success()
#     except RetryableError:
#         breaker.record_failure()
# else:
#     # フォールバック処理（キャッシュ応答やデフォルトメッセージ）
#     result = get_fallback_response()

import time
import asyncio
from collections import deque
 
class TokenBucket:
    """トークンバケットアルゴリズムによるレート制限"""
 
    def __init__(self, capacity: int, refill_rate: float):
        """
        Args:
            capacity: バケットの最大容量（例: RPM=15なら15）
            refill_rate: 1秒あたりの補充レート（例: RPM=15なら15/60=0.25）
        """
        self.capacity = capacity
        self.refill_rate = refill_rate
        self.tokens = float(capacity)
        self.last_refill = time.monotonic()
        self._lock = asyncio.Lock()
 
    async def acquire(self, tokens: int = 1, timeout: float = 30.0) -> bool:
        """トークンを取得する。利用可能になるまで待機"""
        deadline = time.monotonic() + timeout
 
        while True:
            async with self._lock:
                self._refill()
                if self.tokens >= tokens:
                    self.tokens -= tokens
                    return True
 
            # トークン補充を待つ
            wait_time = (tokens - self.tokens) / self.refill_rate
            if time.monotonic() + wait_time > deadline:
                return False  # タイムアウト
            await asyncio.sleep(min(wait_time, 0.1))
 
    def _refill(self):
        """経過時間に基づいてトークンを補充する"""
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now
 
class GeminiRateLimiter:
    """Gemini API の4次元レート制限を管理する"""
 
    def __init__(self, rpm: int = 15, tpm: int = 250000, rpd: int = 1000):
        self.rpm_bucket = TokenBucket(capacity=rpm, refill_rate=rpm / 60.0)
        self.tpm_bucket = TokenBucket(capacity=tpm, refill_rate=tpm / 60.0)
        self.rpd_bucket = TokenBucket(capacity=rpd, refill_rate=rpd / 86400.0)
        self._daily_count = 0
        self._daily_reset = time.time() + 86400
 
    async def acquire(self, estimated_tokens: int = 1000) -> bool:
        """リクエスト送信前にレート制限をチェックする"""
        # RPM チェック
        rpm_ok = await self.rpm_bucket.acquire(1)
        if not rpm_ok:
            return False
 
        # TPM チェック（入力トークン数の推定値）
        tpm_ok = await self.tpm_bucket.acquire(estimated_tokens)
        if not tpm_ok:
            return False
 
        # RPD チェック
        rpd_ok = await self.rpd_bucket.acquire(1)
        if not rpd_ok:
            return False
 
        return True
 
# 使用例
# limiter = GeminiRateLimiter(rpm=15, tpm=250000, rpd=1000)
#
# async def safe_generate(prompt: str):
#     if await limiter.acquire(estimated_tokens=len(prompt) // 4):
#         return await client.generate_with_retry("gemini-2.5-flash", prompt)
#     else:
#         raise Exception("レート制限待機タイムアウト")

from dataclasses import dataclass
 
@dataclass
class ModelConfig:
    name: str
    priority: int  # 小さいほど優先
    max_rpm: int
    cost_per_1m_input: float
    cost_per_1m_output: float
 
# モデルカスケードの定義（優先度順）
MODEL_CASCADE = [
    ModelConfig(
        name="gemini-3.1-pro",
        priority=1,
        max_rpm=150,
        cost_per_1m_input=2.0,
        cost_per_1m_output=12.0
    ),
    ModelConfig(
        name="gemini-2.5-flash",
        priority=2,
        max_rpm=300,
        cost_per_1m_input=0.15,
        cost_per_1m_output=0.60
    ),
    ModelConfig(
        name="gemini-2.5-flash-lite",
        priority=3,
        max_rpm=300,
        cost_per_1m_input=0.075,
        cost_per_1m_output=0.30
    ),
]
 
class ModelCascadeClient:
    def __init__(self, api_key: str, cascade: list[ModelConfig]):
        self.client = genai.Client(api_key=api_key)
        self.cascade = sorted(cascade, key=lambda m: m.priority)
        self.breakers = {
            model.name: CircuitBreaker(failure_threshold=3, recovery_timeout=60.0)
            for model in cascade
        }
 
    async def generate(self, prompt: str) -> dict:
        """カスケード方式でコンテンツを生成する"""
        errors = []
 
        for model_config in self.cascade:
            breaker = self.breakers[model_config.name]
 
            if not breaker.allow_request():
                errors.append(f"{model_config.name}: サーキットブレーカーOPEN")
                continue
 
            try:
                response = self.client.models.generate_content(
                    model=model_config.name,
                    contents=prompt
                )
                breaker.record_success()
                return {
                    "text": response.text,
                    "model_used": model_config.name,
                    "fallback": model_config.priority > 1
                }
 
            except Exception as e:
                breaker.record_failure()
                errors.append(f"{model_config.name}: {e}")
                continue
 
        # 全モデル失敗 → キャッシュまたはデフォルト応答
        return {
            "text": "申し訳ありません。現在サービスが混雑しています。",
            "model_used": "fallback_cache",
            "fallback": True,
            "errors": errors
        }
 
# 使用例
# cascade_client = ModelCascadeClient(
#     api_key="YOUR_API_KEY",
#     cascade=MODEL_CASCADE
# )
# result = await cascade_client.generate("Pythonの非同期処理を説明してください")
# print(f"使用モデル: {result['model_used']}")
# print(f"フォールバック: {result['fallback']}")
# print(result['text'])

import asyncio
 
async def stream_with_recovery(
    client,
    model: str,
    prompt: str,
    max_retries: int = 3,
    chunk_timeout: float = 30.0
):
    """ストリーミング応答を中断検出付きで受信する"""
    accumulated_text = ""
 
    for attempt in range(max_retries + 1):
        try:
            response = client.models.generate_content_stream(
                model=model,
                contents=prompt
            )
 
            chunk_count = 0
            async for chunk in response:
                if chunk.text:
                    accumulated_text += chunk.text
                    chunk_count += 1
                    yield chunk.text
 
                # 安全フィルターによるブロック検出
                if hasattr(chunk, 'candidates') and chunk.candidates:
                    candidate = chunk.candidates[0]
                    if hasattr(candidate, 'finish_reason'):
                        if candidate.finish_reason == 'SAFETY':
                            raise SafetyFilterError(
                                "安全フィルターにより応答がブロックされました"
                            )
 
            # 正常完了
            return
 
        except SafetyFilterError:
            raise  # 安全フィルターエラーはリトライしない
 
        except (ConnectionError, TimeoutError) as e:
            if attempt < max_retries:
                delay = random.uniform(0, min(60.0, 1.0 * (2 ** attempt)))
                print(f"[Stream Recovery] 接続中断を検出 → "
                      f"{delay:.1f}秒後にリトライ（{attempt + 1}/{max_retries}）")
                await asyncio.sleep(delay)
                # 蓄積済みテキストを考慮して続きから要求
                if accumulated_text:
                    prompt = (
                        f"以下のテキストの続きを書いてください。"
                        f"途中で中断されたため再開します:\n\n"
                        f"---中断箇所---\n"
                        f"{accumulated_text[-500:]}\n"
                        f"---ここから続き---"
                    )
            else:
                raise
 
class SafetyFilterError(Exception):
    pass

import time
from collections import deque
from dataclasses import dataclass, field
 
@dataclass
class APIMetrics:
    """Gemini API の利用状況メトリクスを収集する"""
    window_size: int = 300  # 5分間のウィンドウ
 
    requests: deque = field(default_factory=deque)
    errors: deque = field(default_factory=deque)
    latencies: deque = field(default_factory=deque)
    tokens_used: deque = field(default_factory=deque)
 
    def record_request(self, success: bool, latency_ms: float,
                       tokens: int, model: str, status_code: int = 200):
        """リクエスト結果を記録する"""
        now = time.time()
        self.requests.append((now, model, status_code))
        self.latencies.append((now, latency_ms))
        self.tokens_used.append((now, tokens))
        if not success:
            self.errors.append((now, model, status_code))
        self._cleanup(now)
 
    def _cleanup(self, now: float):
        """ウィンドウ外の古いデータを削除する"""
        cutoff = now - self.window_size
        for q in [self.requests, self.errors, self.latencies, self.tokens_used]:
            while q and q[0][0] < cutoff:
                q.popleft()
 
    def get_error_rate(self) -> float:
        """現在のエラー率を返す（0.0〜1.0）"""
        if not self.requests:
            return 0.0
        return len(self.errors) / len(self.requests)
 
    def get_p95_latency(self) -> float:
        """P95レイテンシを返す（ミリ秒）"""
        if not self.latencies:
            return 0.0
        sorted_latencies = sorted(l[1] for l in self.latencies)
        idx = int(len(sorted_latencies) * 0.95)
        return sorted_latencies[min(idx, len(sorted_latencies) - 1)]
 
    def should_alert(self) -> list[str]:
        """アラート条件をチェックし、発報すべきアラートのリストを返す"""
        alerts = []
        error_rate = self.get_error_rate()
        p95 = self.get_p95_latency()
 
        if error_rate > 0.1:  # エラー率 10% 超過
            alerts.append(
                f"HIGH_ERROR_RATE: {error_rate:.1%} "
                f"(閾値: 10%、直近{self.window_size}秒間)"
            )
        if error_rate > 0.5:  # エラー率 50% 超過
            alerts.append(
                f"CRITICAL_ERROR_RATE: {error_rate:.1%} "
                f"— サービス停止を検討してください"
            )
        if p95 > 10000:  # P95レイテンシ 10秒超過
            alerts.append(
                f"HIGH_LATENCY: P95={p95:.0f}ms "
                f"(閾値: 10,000ms)"
            )
 
        # 特定エラーコードの急増検出
        error_codes = {}
        for _, _, code in self.errors:
            error_codes[code] = error_codes.get(code, 0) + 1
        for code, count in error_codes.items():
            if code == 429 and count > 10:
                alerts.append(
                    f"RATE_LIMIT_SPIKE: 429エラーが{count}件 "
                    f"— レート制限の見直しが必要"
                )
 
        return alerts
 
# 使用例
# metrics = APIMetrics(window_size=300)
#
# # リクエスト実行後に記録
# metrics.record_request(
#     success=True, latency_ms=1250.0, tokens=3500,
#     model="gemini-2.5-flash", status_code=200
# )
#
# # アラートチェック（定期実行）
# alerts = metrics.should_alert()
# for alert in alerts:
#     send_notification(alert)  # Slack / PagerDuty / メール