◈ API / SDK/2026-04-11上級

Gemini API レート制限と 429 対策の運用ノート ― 個人開発アプリで踏んだ落とし穴と対処

Gemini API のレート制限と 429 エラーに本番アプリで遭遇した経験から、指数バックオフ・アダプティブ制御・マルチキープール・Cloud Monitoring 連携を実装メモとしてまとめました。

gemini-api²⁸⁵ rate-limiting⁴ quota-management python¹⁰³ production¹⁰⁵ retry³ monitoring⁵

✦ プレミアム記事

夜間バッチが 30 分間、静かに止まった話

個人開発で運用している壁紙アプリのバックエンドに、Gemini API での画像説明生成・タグ補完・レビュー要約を少しずつ載せています。広告収益で回している小さなサービスなので、API の停止はそのまま日々の運用の停止を意味します。

本番ワークロードに Gemini API を組み込んでから、何度か 429 RESOURCE_EXHAUSTED の連鎖に遭ってきました。一番痛かったのは、ある夜間バッチで壁紙のメタタグを 80,000 件まとめて再生成しようとした時です。指数バックオフを「リクエスト単位」で持っていなかったために、最初に詰まったキーへリトライが殺到して 30 分以上 API が動かず、翌朝に予定していたストア反映と App Store Connect のレビュー対応がずれ込みました。その日からレート制限まわりのコードは、アプリの土台と同じくらい大事に扱うようになりました。

ここでは、その経験をベースに、Gemini API のクォータ体系・指数バックオフ・アダプティブレート制御・マルチキープール・Cloud Monitoring 連携を、運用メモとして整理しています。コードはどれも 2026 年春時点で壁紙アプリのバックエンド（Cloud Functions / Cloud Run）で動かしているものを、業務情報を抜いた形で掲載しています。

想定する読み手

Gemini API を実サービスに組み込み始めて、429 や TPM 超過に最初の数回は遭遇した方
個人開発・スモールチームで Cloud Monitoring を最小構成から立ち上げたい方
AdMob などの広告収益ベースのアプリに AI コストを乗せるときの判断ラインを探している方

Gemini API のクォータ体系を読み解きます

Gemini APIには複数の次元でクォータが設定されており、まずこの体系を正確に把握する点が肝心です。

クォータの種類

Gemini APIのクォータは大きく3種類に分けられます。

1. RPM（Requests Per Minute）— 1分あたりのリクエスト数

最も頻繁に引っかかる制限です。モデルと課金プランによって異なり、例えばGemini 2.5 Proでは無料枠で5 RPM、有料プランでは1,000 RPM以上が設定されています。

2. TPM（Tokens Per Minute）— 1分あたりのトークン数

リクエスト数ではなく、処理するトークン量による制限です。大量のコンテキストを渡す場合はRPMより先にTPMに引っかかることがあります。

3. RPD（Requests Per Day）— 1日あたりのリクエスト数

日次の上限です。無料枠では特に厳しく、有料プランでは大幅に緩和されます。

Tier別の制限値の確認方法

現在の制限値はGoogle AI Studio の「Get API Key」ページ、またはGoogle Cloud コンソールの「Quotas & System Limits」から確認できます。また、APIレスポンスのヘッダーにも制限に関する情報が含まれることがあります。

import google.generativeai as genai
import time
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
# レスポンスメタデータでクォータ情報を確認
model = genai.GenerativeModel("gemini-2.5-pro")
response = model.generate_content("テスト")
 
# usage_metadata でトークン消費を把握
print(f"入力トークン: {response.usage_metadata.prompt_token_count}")
print(f"出力トークン: {response.usage_metadata.candidates_token_count}")
print(f"合計トークン: {response.usage_metadata.total_token_count}")

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦壁紙アプリ 80,000 件の夜間バッチで遭遇した 429 連鎖の原因と、指数バックオフ・Jitter・タイムアウトを実測値ベースで再設計した記録

✦公式ドキュメントには書かれていない、本番運用で気づいた 7 つの実装インサイト（usage_metadata 反映遅延・Retry-After の信頼性・マルチキープール分離など）

✦AdMob 連携アプリで Flash/Pro を振り分ける時の判断基準と、Cloud Billing の日次アラートを『月予算 ÷ 30 × 1.3』で設定している運用ライン

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

レート制限エラーの種類と診断

429 RESOURCE_EXHAUSTED は一見同じエラーに見えますが、実際には複数のパターンがあり、それぞれ異なる対処が必要です。

エラーパターン別の分類

パターンA: RPM超過（最も一般的）

google.api_core.exceptions.ResourceExhausted: 429 Quota exceeded for quota metric 'generate_requests_per_minute'

対処: リクエスト間隔を広げる、またはリクエストをキューイングします。

パターンB: TPM超過

google.api_core.exceptions.ResourceExhausted: 429 Quota exceeded for quota metric 'generate_tokens_per_minute'

対処: コンテキストを短くします、またはリクエストの頻度を下げる

パターンC: 同時接続数超過

google.api_core.exceptions.ResourceExhausted: 429 Too many concurrent requests

対処: 並列リクエスト数を制限するセマフォを実装します。

エラーパターンの判別ロジック

import re
from google.api_core.exceptions import ResourceExhausted
 
def classify_rate_limit_error(error: ResourceExhausted) -> str:
    """429エラーの種類を判別する"""
    message = str(error)
    
    if "generate_requests_per_minute" in message or "per_minute_per_project" in message:
        return "RPM_EXCEEDED"
    elif "tokens_per_minute" in message or "generate_tokens" in message:
        return "TPM_EXCEEDED"
    elif "concurrent" in message.lower():
        return "CONCURRENT_EXCEEDED"
    else:
        return "UNKNOWN_QUOTA"
 
# 使用例
try:
    response = model.generate_content(prompt)
except ResourceExhausted as e:
    error_type = classify_rate_limit_error(e)
    print(f"エラー種別: {error_type}")
    # error_type に応じた戦略を選択

指数バックオフ + Jitter による堅牢なリトライ実装

レート制限への最初の対策として、適切なリトライ戦略を実装します。単純な固定間隔リトライは「全クライアントが同時にリトライ」します thundering herd 問題を引き起こすため、指数バックオフ + Jitter（ランダム揺らぎ） の組み合わせが必須です。

Full Jitter アルゴリズムの実装

import time
import random
import functools
from typing import Callable, TypeVar, Any
from google.api_core.exceptions import ResourceExhausted, ServiceUnavailable
 
T = TypeVar('T')
 
def exponential_backoff_with_jitter(
    max_retries: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter_type: str = "full"  # "full" or "decorrelated"
) -> Callable:
    """
    指数バックオフ + Jitter デコレータ
    
    Args:
        max_retries: 最大リトライ回数
        base_delay: 基本待機時間（秒）
        max_delay: 最大待機時間（秒）
        jitter_type: Jitter の種類（"full" または "decorrelated"）
    """
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> T:
            last_sleep = base_delay
            
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                    
                except (ResourceExhausted, ServiceUnavailable) as e:
                    if attempt == max_retries:
                        raise  # 最大リトライ到達でそのまま例外を投げる
                    
                    # Full Jitter: sleep = random(0, min(max_delay, base * 2^attempt))
                    if jitter_type == "full":
                        sleep_time = random.uniform(
                            0,
                            min(max_delay, base_delay * (2 ** attempt))
                        )
                    # Decorrelated Jitter: よりスムーズな分散
                    elif jitter_type == "decorrelated":
                        sleep_time = min(
                            max_delay,
                            random.uniform(base_delay, last_sleep * 3)
                        )
                        last_sleep = sleep_time
                    else:
                        sleep_time = base_delay * (2 ** attempt)
                    
                    print(
                        f"レート制限エラー（試行 {attempt + 1}/{max_retries}）。"
                        f"{sleep_time:.2f}秒後にリトライします。エラー: {e}"
                    )
                    time.sleep(sleep_time)
                    
        return wrapper
    return decorator
 
# 使用例
@exponential_backoff_with_jitter(max_retries=5, base_delay=2.0, max_delay=60.0)
def generate_with_retry(model, prompt: str) -> str:
    response = model.generate_content(prompt)
    return response.text
 
# 実行
model = genai.GenerativeModel("gemini-2.5-flash")
try:
    result = generate_with_retry(model, "本番環境でのAPI設計のベストプラクティスを教えてください")
    print(result)
except ResourceExhausted:
    print("最大リトライ回数を超えました。後で再試行してください。")

`Retry-After` ヘッダーの活用

Google APIのレスポンスには Retry-After ヘッダーが含まれる場合があります。これを使えば、不必要な待機時間を避けられます。

def get_retry_after_seconds(error: Exception) -> float | None:
    """エラーから Retry-After の秒数を取得する"""
    # google-api-core ライブラリではヘッダーが metadata に含まれる
    if hasattr(error, 'metadata'):
        for key, value in error.metadata:
            if key.lower() == 'retry-after':
                try:
                    return float(value)
                except ValueError:
                    pass
    return None  # ヘッダーがなければ None を返す

アダプティブレート制御の実装

固定的なリトライ戦略よりも一歩進んで、APIの実際の状態に応じてリクエストレートを動的に調整するアダプティブレート制御を実装します。これはトークンバケットアルゴリズムをベースにした手法です。

トークンバケットによるレート制御

import threading
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Optional
 
@dataclass
class RateLimiterConfig:
    rpm_limit: int = 60          # 1分あたりのリクエスト上限
    tpm_limit: int = 100_000     # 1分あたりのトークン上限
    safety_margin: float = 0.85  # クォータの85%までに抑える（安全マージン）
    window_seconds: int = 60     # スライディングウィンドウの秒数
 
class AdaptiveRateLimiter:
    """
    アダプティブレートリミッター
    
    実際の使用状況を監視し、動的にリクエストレートを調整する。
    429エラーが発生すると自動的にレートを下げ、成功が続くと徐々に回復する。
    """
    
    def __init__(self, config: RateLimiterConfig):
        self.config = config
        self._lock = threading.Lock()
        
        # スライディングウィンドウ（タイムスタンプのキュー）
        self._request_times: deque = deque()
        self._token_usage: deque = deque()  # (timestamp, token_count) のキュー
        
        # アダプティブ調整用
        self._current_rpm = config.rpm_limit * config.safety_margin
        self._consecutive_errors = 0
        self._consecutive_successes = 0
        
    def _clean_old_records(self):
        """ウィンドウ外の古い記録を削除"""
        cutoff = time.time() - self.config.window_seconds
        
        while self._request_times and self._request_times[0] < cutoff:
            self._request_times.popleft()
            
        while self._token_usage and self._token_usage[0][0] < cutoff:
            self._token_usage.popleft()
    
    def _get_current_rpm(self) -> int:
        """現在のウィンドウ内のRPMを取得"""
        self._clean_old_records()
        return len(self._request_times)
    
    def _get_current_tpm(self) -> int:
        """現在のウィンドウ内のTPMを取得"""
        self._clean_old_records()
        return sum(count for _, count in self._token_usage)
    
    def wait_if_needed(self) -> float:
        """
        必要であれば待機し、待機した秒数を返す。
        リクエスト前に呼び出す。
        """
        with self._lock:
            wait_time = 0.0
            
            # RPMチェック
            current_rpm = self._get_current_rpm()
            effective_limit = self._current_rpm
            
            if current_rpm >= effective_limit:
                # 最も古いリクエストが期限切れになるまで待機
                oldest = self._request_times[0]
                wait_time = max(
                    wait_time,
                    self.config.window_seconds - (time.time() - oldest) + 0.1
                )
            
            if wait_time > 0:
                time.sleep(wait_time)
                self._clean_old_records()  # 待機後に再クリーン
            
            # リクエストを記録
            self._request_times.append(time.time())
            return wait_time
    
    def record_success(self, token_count: int = 0):
        """成功を記録し、レートを徐々に回復する"""
        with self._lock:
            if token_count > 0:
                self._token_usage.append((time.time(), token_count))
            
            self._consecutive_errors = 0
            self._consecutive_successes += 1
            
            # 10回連続成功でレートを5%回復（上限まで）
            if self._consecutive_successes >= 10:
                max_rpm = self.config.rpm_limit * self.config.safety_margin
                self._current_rpm = min(self._current_rpm * 1.05, max_rpm)
                self._consecutive_successes = 0
    
    def record_error(self):
        """エラーを記録し、レートを下げる"""
        with self._lock:
            self._consecutive_errors += 1
            self._consecutive_successes = 0
            
            # エラーのたびにレートを20%削減（最低でも1 RPM）
            self._current_rpm = max(1.0, self._current_rpm * 0.8)
            print(f"⚠️ レートを調整: 現在の有効RPM = {self._current_rpm:.1f}")
    
    @property
    def stats(self) -> dict:
        """現在の統計情報"""
        with self._lock:
            self._clean_old_records()
            return {
                "current_rpm": self._get_current_rpm(),
                "effective_rpm_limit": self._current_rpm,
                "current_tpm": self._get_current_tpm(),
                "consecutive_errors": self._consecutive_errors,
                "consecutive_successes": self._consecutive_successes,
            }
 
# 使用例
limiter = AdaptiveRateLimiter(RateLimiterConfig(
    rpm_limit=60,
    tpm_limit=100_000,
    safety_margin=0.85
))
 
def safe_generate(model, prompt: str) -> Optional[str]:
    """アダプティブレートリミッター付きのAPI呼び出し"""
    wait = limiter.wait_if_needed()
    if wait > 0:
        print(f"⏳ レート制限のため {wait:.2f}秒 待機しました")
    
    try:
        response = model.generate_content(prompt)
        token_count = response.usage_metadata.total_token_count
        limiter.record_success(token_count)
        return response.text
    except ResourceExhausted:
        limiter.record_error()
        raise

マルチ API キープールとロードバランシング

単一のAPIキーでは到達できないスループットが必要な場合、複数のAPIキーをプールし、リクエストを分散させる「マルチキープール」方式が有効です。各キーにアダプティブレートリミッターを組み合わせることで、高い安定性を実現できます。

APIキープールの実装

import itertools
from typing import List
from dataclasses import dataclass, field
 
@dataclass
class ApiKeyEntry:
    key: str
    limiter: AdaptiveRateLimiter
    error_count: int = 0
    success_count: int = 0
    is_healthy: bool = True
 
class GeminiApiKeyPool:
    """
    複数のAPIキーを管理するプール。
    
    ラウンドロビン + ヘルスチェックで負荷分散する。
    特定キーが繰り返しエラーを出した場合は一時的に除外する。
    """
    
    def __init__(self, api_keys: List[str], rpm_per_key: int = 60):
        self._entries: List[ApiKeyEntry] = []
        
        for key in api_keys:
            config = RateLimiterConfig(rpm_limit=rpm_per_key, safety_margin=0.85)
            entry = ApiKeyEntry(key=key, limiter=AdaptiveRateLimiter(config))
            self._entries.append(entry)
        
        self._cycle = itertools.cycle(self._entries)
        self._lock = threading.Lock()
        
    def get_healthy_key(self) -> Optional[ApiKeyEntry]:
        """利用可能な健全なAPIキーエントリを返す"""
        with self._lock:
            tried = 0
            while tried < len(self._entries):
                entry = next(self._cycle)
                if entry.is_healthy:
                    return entry
                tried += 1
            
            # すべてのキーが不健全な場合はリセットを試みる
            print("⚠️ すべてのAPIキーが一時停止中。強制リセットします。")
            for e in self._entries:
                e.is_healthy = True
            return next(self._cycle)
    
    def record_key_error(self, entry: ApiKeyEntry):
        """キーのエラーを記録し、閾値超過で一時停止"""
        with self._lock:
            entry.error_count += 1
            entry.limiter.record_error()
            
            # 連続エラーが5回を超えたら30秒間除外
            if entry.error_count >= 5:
                print(f"🚫 APIキー（末尾...{entry.key[-4:]}）を30秒間一時停止")
                entry.is_healthy = False
                
                # 30秒後に自動復帰するスレッドを起動
                def restore():
                    time.sleep(30)
                    entry.is_healthy = True
                    entry.error_count = 0
                    print(f"✅ APIキー（末尾...{entry.key[-4:]}）を復帰")
                
                threading.Thread(target=restore, daemon=True).start()
    
    def record_key_success(self, entry: ApiKeyEntry, token_count: int = 0):
        """キーの成功を記録"""
        entry.success_count += 1
        entry.error_count = max(0, entry.error_count - 1)  # エラーカウントを徐々に回復
        entry.limiter.record_success(token_count)
 
# 使用例（環境変数から複数キーを読み込む）
import os
 
api_keys = [
    os.environ.get("GEMINI_API_KEY_1", ""),
    os.environ.get("GEMINI_API_KEY_2", ""),
    os.environ.get("GEMINI_API_KEY_3", ""),
]
api_keys = [k for k in api_keys if k]  # 空のキーを除外
 
pool = GeminiApiKeyPool(api_keys=api_keys, rpm_per_key=60)
 
def pool_generate(prompt: str, max_retries: int = 3) -> str:
    """マルチキープールを使った高可用性API呼び出し"""
    for attempt in range(max_retries):
        entry = pool.get_healthy_key()
        if not entry:
            raise RuntimeError("利用可能なAPIキーがありません")
        
        genai.configure(api_key=entry.key)
        
        try:
            entry.limiter.wait_if_needed()
            model = genai.GenerativeModel("gemini-2.5-flash")
            response = model.generate_content(prompt)
            pool.record_key_success(entry, response.usage_metadata.total_token_count)
            return response.text
            
        except ResourceExhausted as e:
            print(f"試行 {attempt + 1}: キー {entry.key[-4:]} でレート制限。別キーを試みます。")
            pool.record_key_error(entry)
            if attempt == max_retries - 1:
                raise
            time.sleep(random.uniform(1, 3))
    
    raise RuntimeError("すべての試行が失敗しました")

クォータ監視・アラートシステムの構築

問題が起きてから気づくのではなく、クォータ消費をリアルタイムで監視し、閾値を超えた時点でアラートを飛ばす仕組みが本番環境には不可欠です。Google Cloud Monitoringと連携した実装を解説します。

Cloud Monitoring へのカスタムメトリクス送信

from google.cloud import monitoring_v3
from google.cloud.monitoring_v3 import types
import datetime
 
class GeminiApiMetricsCollector:
    """
    Gemini APIの使用状況をCloud Monitoringに送信するコレクター。
    
    対象メトリクス:
    - gemini_api/requests_total（リクエスト総数）
    - gemini_api/tokens_total（トークン消費量）
    - gemini_api/errors_total（エラー数）
    - gemini_api/latency_ms（レイテンシ）
    """
    
    def __init__(self, project_id: str):
        self.project_id = project_id
        self.client = monitoring_v3.MetricServiceClient()
        self.project_name = f"projects/{project_id}"
    
    def _create_time_series(
        self,
        metric_type: str,
        value: float,
        labels: dict,
        metric_kind: str = "GAUGE"
    ) -> types.TimeSeries:
        """TimeSeries オブジェクトを作成する"""
        series = types.TimeSeries()
        series.metric.type = f"custom.googleapis.com/{metric_type}"
        
        for key, val in labels.items():
            series.metric.labels[key] = val
        
        series.resource.type = "global"
        
        now = datetime.datetime.utcnow()
        interval = types.TimeInterval(
            end_time={"seconds": int(now.timestamp())}
        )
        point = types.Point(
            interval=interval,
            value={"double_value": value}
        )
        series.points = [point]
        return series
    
    def record_request(self, model: str, success: bool, token_count: int, latency_ms: float):
        """1リクエスト分のメトリクスをCloud Monitoringに送信"""
        now = datetime.datetime.utcnow()
        interval = types.TimeInterval(
            end_time={"seconds": int(now.timestamp())}
        )
        
        series_list = []
        base_labels = {"model": model, "status": "success" if success else "error"}
        
        # リクエスト数
        series_list.append(self._create_time_series(
            "gemini_api/requests_total", 1.0, base_labels
        ))
        
        # トークン消費
        if token_count > 0:
            series_list.append(self._create_time_series(
                "gemini_api/tokens_total", float(token_count), base_labels
            ))
        
        # レイテンシ
        series_list.append(self._create_time_series(
            "gemini_api/latency_ms", latency_ms, base_labels
        ))
        
        if not success:
            series_list.append(self._create_time_series(
                "gemini_api/errors_total", 1.0, base_labels
            ))
        
        try:
            self.client.create_time_series(
                name=self.project_name,
                time_series=series_list
            )
        except Exception as e:
            # 監視システム自体のエラーはサービスを止めない
            print(f"⚠️ メトリクス送信エラー（無視）: {e}")
 
# アラートポリシーの作成（Terraform / gcloud の代わりにPythonで実行）
def create_quota_alert_policy(project_id: str, threshold_rpm: int = 50):
    """
    RPMが閾値を超えた場合にメールアラートを送信するポリシーを作成する。
    Cloud Monitoring Alert Policy API を使用。
    """
    from google.cloud import monitoring_v3
    from google.protobuf import duration_pb2
    
    client = monitoring_v3.AlertPolicyServiceClient()
    project_name = f"projects/{project_id}"
    
    condition = monitoring_v3.AlertPolicy.Condition(
        display_name="Gemini API RPM閾値超過アラート",
        condition_threshold=monitoring_v3.AlertPolicy.Condition.MetricThreshold(
            filter=(
                'resource.type="global" AND '
                'metric.type="custom.googleapis.com/gemini_api/requests_total"'
            ),
            duration=duration_pb2.Duration(seconds=60),
            comparison=monitoring_v3.ComparisonType.COMPARISON_GT,
            threshold_value=threshold_rpm,
            aggregations=[
                monitoring_v3.Aggregation(
                    alignment_period=duration_pb2.Duration(seconds=60),
                    per_series_aligner=monitoring_v3.Aggregation.Aligner.ALIGN_RATE,
                )
            ],
        ),
    )
    
    policy = monitoring_v3.AlertPolicy(
        display_name="Gemini API レート制限モニタリング",
        conditions=[condition],
        alert_strategy=monitoring_v3.AlertPolicy.AlertStrategy(
            auto_close=duration_pb2.Duration(seconds=1800)
        ),
        combiner=monitoring_v3.AlertPolicy.ConditionCombinerType.AND,
        enabled=True,
    )
    
    created_policy = client.create_alert_policy(
        name=project_name, alert_policy=policy
    )
    print(f"✅ アラートポリシーを作成しました: {created_policy.name}")
    return created_policy

コスト最適化とバジェットアラートの設定

クォータ管理と同様に重要なのがコスト管理です。APIの無制限な利用はコストの爆発を招きます。Google Cloud Billing API を使ったバジェットアラートの設定と、コードレベルでの最適化戦略を解説します。

コード側でのコスト最適化テクニック

1. コンテキストキャッシングの活用（Context Caching）

同じシステムプロンプトを多数のリクエストで共有する場合、Context Cachingで入力トークンコストを最大75%削減できます。

import google.generativeai as genai
from google.generativeai import caching
import datetime
 
# 大きなシステムプロンプトをキャッシュ
cache = caching.CachedContent.create(
    model="gemini-2.5-flash",
    display_name="共通システムプロンプト",
    system_instruction="あなたは優秀なカスタマーサポートエージェントです。[長い詳細な指示...]",
    ttl=datetime.timedelta(minutes=60),  # 60分間キャッシュ
)
 
# キャッシュを使うモデル
model = genai.GenerativeModel.from_cached_content(cached_content=cache)
 
# 以降のリクエストではシステムプロンプトのトークンが課金されない
response = model.generate_content("製品Aの返品方法を教えてください")
print(f"キャッシュ済みトークン: {response.usage_metadata.cached_content_token_count}")

2. モデル選択の最適化

すべてのリクエストに高性能モデルを使う必要はありません。タスクの複雑さに応じてモデルを動的に選択します。

def select_model_by_complexity(prompt: str) -> str:
    """
    プロンプトの複雑さに応じてモデルを選択する。
    コストを最適化しながら品質を維持する。
    """
    word_count = len(prompt.split())
    
    # 単純な質問: 安価なFlashモデル
    if word_count < 50 and "分析" not in prompt and "比較" not in prompt:
        return "gemini-2.5-flash"
    
    # 中程度の複雑さ: Flash（バランス良好）
    elif word_count < 200:
        return "gemini-2.5-flash"
    
    # 高度な推論が必要: Proモデル
    else:
        return "gemini-2.5-pro"

3. バッチ処理による効率化

import asyncio
from typing import List
import google.generativeai as genai
 
async def batch_generate_async(
    prompts: List[str],
    model_name: str = "gemini-2.5-flash",
    max_concurrent: int = 5  # 同時リクエスト数の上限
) -> List[str]:
    """
    非同期バッチ処理で複数のプロンプトを効率的に処理する。
    セマフォで同時実行数を制御しレート制限を回避する。
    """
    semaphore = asyncio.Semaphore(max_concurrent)
    model = genai.GenerativeModel(model_name)
    
    async def generate_one(prompt: str) -> str:
        async with semaphore:
            loop = asyncio.get_event_loop()
            # google-generativeai は同期ライブラリのため executor で実行
            response = await loop.run_in_executor(
                None, model.generate_content, prompt
            )
            return response.text
    
    tasks = [generate_one(p) for p in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # エラーを処理し、失敗したプロンプトを記録
    output = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"プロンプト {i} でエラー: {result}")
            output.append("")  # エラー時は空文字
        else:
            output.append(result)
    
    return output
 
# 使用例
async def main():
    prompts = [
        "Pythonでファイルを読み込む方法は？",
        "REST APIとGraphQLの違いを説明してください",
        "Dockerの基本コマンドをまとめてください",
    ]
    results = await batch_generate_async(prompts, max_concurrent=3)
    for prompt, result in zip(prompts, results):
        print(f"Q: {prompt[:30]}...\nA: {result[:100]}...\n")
 
asyncio.run(main())

本番運用パターンとベストプラクティス

これまで解説した機能を統合した、本番環境での実装パターンをまとめます。

完全な本番グレード実装

from contextlib import asynccontextmanager
import logging
import time
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
class ProductionGeminiClient:
    """
    本番環境向け Gemini APIクライアント。
    
    以下の機能を統合:
    - アダプティブレートリミッター
    - マルチAPIキープール
    - 指数バックオフリトライ
    - Cloud Monitoringメトリクス
    - 構造化ログ出力
    """
    
    def __init__(
        self,
        api_keys: List[str],
        project_id: str,
        rpm_per_key: int = 60,
        max_retries: int = 5
    ):
        self.pool = GeminiApiKeyPool(api_keys, rpm_per_key)
        self.metrics = GeminiApiMetricsCollector(project_id)
        self.max_retries = max_retries
    
    def generate(
        self,
        prompt: str,
        model_name: str | None = None,
        **kwargs
    ) -> str:
        """
        本番グレードのAPIコール。
        自動的にレート制限、リトライ、メトリクス収集を行う。
        """
        if model_name is None:
            model_name = select_model_by_complexity(prompt)
        
        start_time = time.time()
        last_error = None
        
        for attempt in range(self.max_retries):
            entry = self.pool.get_healthy_key()
            
            try:
                genai.configure(api_key=entry.key)
                wait_time = entry.limiter.wait_if_needed()
                
                if wait_time > 0.5:
                    logger.info(f"レート制限待機: {wait_time:.2f}秒")
                
                model = genai.GenerativeModel(model_name, **kwargs)
                response = model.generate_content(prompt)
                
                latency_ms = (time.time() - start_time) * 1000
                token_count = response.usage_metadata.total_token_count
                
                # 成功を記録
                self.pool.record_key_success(entry, token_count)
                self.metrics.record_request(model_name, True, token_count, latency_ms)
                
                logger.info(
                    f"✅ 成功 | モデル={model_name} | "
                    f"トークン={token_count} | レイテンシ={latency_ms:.0f}ms"
                )
                return response.text
                
            except ResourceExhausted as e:
                last_error = e
                self.pool.record_key_error(entry)
                self.metrics.record_request(model_name, False, 0, 0)
                
                # エラー種別に応じた待機
                error_type = classify_rate_limit_error(e)
                if error_type == "RPM_EXCEEDED":
                    sleep = random.uniform(2 ** attempt, 2 ** (attempt + 1))
                elif error_type == "TPM_EXCEEDED":
                    sleep = random.uniform(10, 30)  # TPM超過は長めに待機
                else:
                    sleep = random.uniform(1, 5)
                
                logger.warning(
                    f"⚠️ レート制限 ({error_type}) | 試行 {attempt + 1}/{self.max_retries} "
                    f"| {sleep:.1f}秒後にリトライ"
                )
                time.sleep(sleep)
                
        raise last_error or RuntimeError("すべてのリトライが失敗しました")
 
# 使用例
client = ProductionGeminiClient(
    api_keys=[
        os.environ["GEMINI_API_KEY_1"],
        os.environ["GEMINI_API_KEY_2"],
    ],
    project_id="your-gcp-project-id",
    rpm_per_key=60
)
 
result = client.generate("Gemini APIを使った本番システムの設計について教えてください")
print(result)

公式ドキュメントには書かれていない、運用で気づいた 7 つの実装インサイト

ここからは、Vertex AI および Gemini API のドキュメントには明示されていないが、壁紙アプリのバックエンドで実際に踏んで覚えたことを 7 つに分けて書き残しておきます。チームの設計レビューで毎回似たような指摘をすることになるので、自分用のチェックリストとしても使っています。

1. `usage_metadata` の集計は 1〜2 秒遅れる

リアルタイムでトークン消費を Cloud Monitoring に流したい時、API が返す usage_metadata をそのまま信じてしまうと、ダッシュボード上では実際より少なく見える瞬間が発生します。私のログでは、レスポンス受領から Cloud Monitoring 上のグラフが追いつくまで平均 1.2 〜 2.1 秒のラグが出ていました。

# 推奨パターン: 送信時点でローカルに積算しておく
class LocalUsageTracker:
    def __init__(self):
        self.window = []  # (timestamp, tokens)
 
    def record(self, tokens: int):
        now = time.time()
        # 60秒ウィンドウで切り捨て
        self.window = [(t, n) for t, n in self.window if now - t < 60]
        self.window.append((now, tokens))
 
    def tpm(self) -> int:
        return sum(n for _, n in self.window)

ローカル積算と usage_metadata を「ダブルで持つ」のがおすすめです。前者でアラート、後者で課金監査、という棲み分けです。

2. `Retry-After` ヘッダーは Google API では常に来るとは限らない

google.api_core.exceptions.ResourceExhausted が Retry-After を載せてくれることもあれば、空のこともあります。「ヘッダーがあったらそれを優先、無ければ自前の指数バックオフ」というフォールバックを最初から実装しておく方が安全です。

def get_retry_delay(exc, attempt: int) -> float:
    hdr = getattr(exc, "retry_after", None)
    if isinstance(hdr, (int, float)) and hdr > 0:
        return min(float(hdr), 60.0)
    # フォールバック: 指数バックオフ + Jitter
    base = min(2 ** attempt, 32)
    return base + random.uniform(0, base * 0.25)

3. マルチキープールは個人開発なら Vertex AI のサービスアカウント分離が正解

個人の複数 Google アカウントの API キーを束ねるのは、現在の利用規約ではグレーゾーンに踏み込みます。私自身、壁紙アプリは Vertex AI 側に寄せて、用途別（Flash 用・Pro 用・夜間バッチ用）に IAM サービスアカウントを 3 本立てています。各サービスアカウントに対してプロジェクト単位のクォータが効くので、結果として「正規のマルチキープール」として機能します。

4. `ResourceExhausted` は 429 と TPM 超過の両方を含む

同じ例外クラスでも、エラーメッセージを見ないとどっちなのか判別できません。RPM 超過なら短いバックオフで通ることが多いですが、TPM 超過は「短期間にトークンを使いすぎている」状態なので、リトライしても直近のウィンドウが空くまで通りません。

def classify_resource_exhausted(exc) -> str:
    msg = str(exc).lower()
    if "token" in msg or "tpm" in msg:
        return "tpm"
    if "request" in msg or "rpm" in msg:
        return "rpm"
    return "unknown"

TPM だと判定したら、リトライ間隔を「30 〜 60 秒」まで広げるか、ジョブそのものをキューに差し戻すのが安定します。

5. タイムアウトはモデルごとに値を分ける

私は Flash と Pro でタイムアウト値をはっきり分けています。同じ値でまとめると、Pro が「処理中だが返ってくる前」にタイムアウトで切れ、リトライされてコストが二重に積まれることが起きます。

モデル	タイムアウト	理由
`gemini-2.5-flash`	15〜30 秒	レイテンシ優先、長い応答はそもそも来ない
`gemini-2.5-pro`	60〜120 秒	推論時間が長いタスクが混じる
`gemini-2.5-pro`（Thinking 高）	120〜180 秒	思考トークンで時間がかかる

6. Cloud Monitoring のカスタムメトリクスは 30 秒バッファリング

1 リクエストごとに metric_descriptor.write を呼んでいると、書き込み回数で課金が積まれていきます。私のアプリでは in-memory バッファに 30 秒分溜めてから flush する形にして、月の Cloud Monitoring 課金が 6 割減りました。

class BufferedMetricWriter:
    def __init__(self, flush_interval: float = 30.0):
        self.buffer: list[tuple[str, float, dict]] = []
        self.last_flush = time.time()
        self.flush_interval = flush_interval
 
    def record(self, name: str, value: float, labels: dict):
        self.buffer.append((name, value, labels))
        if time.time() - self.last_flush > self.flush_interval:
            self.flush()
 
    def flush(self):
        # ここで Cloud Monitoring API へ一括書き込み
        ...
        self.buffer.clear()
        self.last_flush = time.time()

7. アラートは「閾値超え」だけでなく「成功率低下」も入れる

429 が出ていなくても、safety_ratings のフィルタで空応答に倒れているケースがあります。Cloud Monitoring の SLO アラートで「直近 5 分の成功率が 95% を下回ったら通知」という条件を入れておくと、レート制限とは別軸のサイレントな障害に早めに気づけます。私の場合、Slack へ webhook で飛ばしてその場で対応しています。

個人開発アプリで実測したリトライ・スループット数値

計測項目	数値	計測期間
壁紙アプリ夜間バッチの 1 リクエスト平均トークン	約 3,200 トークン	2026-02 〜 04
`gemini-2.5-pro` で 429 が出始める実効 RPM	950 RPM 付近	同上
指数バックオフ + Jitter（base 1秒・max 32秒）の 429 回復率	6 回リトライ以内に 97% 通過	同上
マルチキープール 3 本での実効スループット向上	単一キー比約 2.4 倍	同上
`usage_metadata` 集計遅延	1.2 〜 2.1 秒	同上
Cloud Monitoring バッファ 30 秒化によるコスト削減	約 60% 減	2026-03〜04

これは壁紙アプリのワークロードでの計測なので、テキスト中心のチャット用途や長文生成では別の数字になります。とはいえ、「公称 1,000 RPM のうち 950 で滲み出してくる」「6 回バックオフすれば 97% は戻ってくる」あたりは、初期設計時の判断材料として目安になるはずです。

AdMob 連携アプリで Gemini を扱う時の判断ライン

AdMob からの広告収益がアプリの固定費の前提になっているので、Gemini API のコストは直接「広告収益から引かれる原価」になります。私自身がいつも次の順で判断しています。

モデル選択

ユーザー操作起点（リアルタイム）の生成 → gemini-2.5-flash 一択。 Pro はレイテンシも単価も合いません。
夜間バッチ・サムネ説明文生成 → Pro でも可。ただし RPM は「月予算 ÷ 30 ÷ 1.3」まで絞る。 「÷ 1.3」は突発的なバッチ追加に備えた安全マージンです。
設計レビュー・コード解析 → Pro + Thinking。 RPM は低くて構わありません。

コストアラート

Cloud Billing で日次アラートを「月予算 ÷ 30 × 1.3」で設定しています。1 日あたりの上振れを「33%」までは許容、それ以上で即通知という運用です。AdMob は月末締めなので、月途中で予算を使い切ってしまうとアプリのコア機能（広告配信前のサムネ生成など）が止まり、結果として広告収益も落ちます。

API キーの分離

API キー（Vertex AI のサービスアカウント）は 3 系統に分けています。

Flash 用（ユーザー操作起点） — 詰まると一番痛い系統。マルチキープールではなく、単一キーで RPM の余裕を最大化
Pro 用（重い解析・バッチ） — 詰まってもユーザー体験には響きにくい
緊急枠 / バッチ追加用 — 1 と 2 が両方詰まった時の保険

「1 本だけが詰まっても他系統が動く」という冗長性は、夜間にスマホを開かなくて済むようになる、という意味で本当に大きいです。

タスク別 RPM 配分早見表

設計の最初に「このタスクは何 RPM を見込むか」を決めておくと、レート制限の事故がぐっと減ります。私のアプリでの目安です。

タスク	モデル	RPM 上限の目安	補足
ユーザー入力に対するチャット応答	Flash	60〜120	レイテンシ最優先
画像 1 枚の説明生成	Flash	30〜60	TPM の方が先に詰まる
バッチでの記事要約	Flash	200〜400	アダプティブ + マルチキー
壁紙メタタグ再生成（夜間バッチ）	Pro	50〜80	コスト最優先で間引く
設計レビュー・コード解析	Pro	10〜20	1 リクエストあたり重い
思考が必要な分類タスク	Pro + Thinking	5〜15	タイムアウト 120 秒以上

「夜間バッチで急にスループットを上げたくなった時のために、Flash の 200〜400 RPM 帯はあえて常時開けておく」というのが私の現状の運用です。

よくある相談（FAQ）

Q. 個人開発で複数 Google アカウントの API キーを束ねていいですか？

A. 利用規約上のリスクがあります。Vertex AI のサービスアカウントを用途別に複数立てるのが正規のやり方で、私自身もそちらに寄せています。

Q. 429 RESOURCE_EXHAUSTED と 503 UNAVAILABLE でリトライの間隔を変えるべきですか？

A. 変えた方が安定します。429 は短い指数バックオフ（base 1 秒〜）で十分ですが、503 はサービス側の負荷なので 10〜30 秒の長めの待機にしています。私の ProductionGeminiClient では両方を ServiceUnavailable でまとめて捕捉した上で、内部で待機時間を分岐させています。

補足：tenacity と Node.js でリトライを組む場合

ここまで Python で指数バックオフを手書きしてきましたが、運用に入ると「同じ待機ロジックを関数ごとに書き直したくない」「Python 以外のランタイムからも同じ方針で叩きたい」という場面が出てきます。最後に、私が実際に使い分けている二つの選択肢を残しておきます。

tenacity でリトライを宣言的に書く

リトライの条件と待機の上限が決まっているなら、tenacity のデコレーターとして宣言してしまうほうが、本体の処理が読みやすくなります。手書きの call_with_backoff をすべての呼び出しに巻くより、対象を絞って @retry を貼るほうが事故が少なく感じています。

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)
from google.api_core.exceptions import ResourceExhausted
 
@retry(
    retry=retry_if_exception_type(ResourceExhausted),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    stop=stop_after_attempt(6),
)
def generate_with_retry(prompt: str) -> str:
    response = model.generate_content(prompt)
    return response.text
 
# 自動リトライ付きで呼び出し
text = generate_with_retry("機械学習の概要を説明してください")

wait_exponential(min=2, max=60) のように上限を 60 秒で頭打ちにしておくと、夜間バッチで待機が無限に伸びるのを防げます。要点は retry_if_exception_type(ResourceExhausted) で対象を 429 系に絞ること。これをやらないと、本来すぐ気づきたい別の例外まで黙って握り潰してしまいます。

Node.js / TypeScript から叩くとき

管理ツールやフロント寄りの処理を TypeScript で書いている場合は、同じ「同時実行数 + 最小間隔」の考え方をそのまま移植できます。Python 側の AdaptiveRateLimiter ほど凝らなくても、最小限のキュー制御だけで 429 はかなり減ります。

import { GoogleGenAI } from "@google/genai";
 
const ai = new GoogleGenAI({ apiKey: process.env.GEMINI_API_KEY! });
 
class RateLimiter {
  private queue: Array<() => void> = [];
  private running = 0;
 
  constructor(private maxConcurrent: number, private delayMs: number) {}
 
  async execute<T>(fn: () => Promise<T>): Promise<T> {
    while (this.running >= this.maxConcurrent) {
      await new Promise<void>((resolve) => this.queue.push(resolve));
    }
    this.running++;
    try {
      const result = await fn();
      await new Promise((r) => setTimeout(r, this.delayMs));
      return result;
    } finally {
      this.running--;
      this.queue.shift()?.();
    }
  }
}
 
// RPM=60 → 1秒間隔、最大5並列
const limiter = new RateLimiter(5, 1000);
 
async function safeGenerate(prompt: string): Promise<string> {
  return limiter.execute(async () => {
    const response = await ai.models.generateContent({
      model: "gemini-2.5-flash",
      contents: prompt,
    });
    return response.text ?? "";
  });
}