Gemini API AIゲートウェイ設計パターン — レート制限・フォールバック・コスト追跡を統合するプロキシサーバー構築ガイド

取り組みの背景 — なぜ AIゲートウェイが必要なのか

Gemini API を1つのアプリケーションから呼び出すだけなら、SDKを直接使えば十分です。しかし本番環境では、複数のサービスが同時にAPIを呼び出し、レート制限に引っかかり、コストが予測不能に膨らむ問題が避けられません。

AIゲートウェイ（APIプロキシ）は、アプリケーションとGemini APIの間に立つ中間レイヤーです。すべてのAPIリクエストを一箇所で管理することで、以下の課題を解決します。

レート制限の一元管理: 複数サービスからのリクエストを集約し、429エラーを事前に防止
自動フォールバック: Gemini 2.5 Pro がダウンした場合に Flash へ自動切り替え
トークンコスト追跡: リクエストごとのトークン使用量とコストをリアルタイムに記録
モデルルーティング: リクエストの複雑さに応じて Pro / Flash を自動選択

AIゲートウェイのアーキテクチャ設計

全体構成

AIゲートウェイは4つの主要コンポーネントで構成されます。

┌─────────────┐     ┌──────────────────────────────────────┐     ┌─────────────┐
│  Service A  │────▶│          AI Gateway (Proxy)           │────▶│ Gemini 2.5  │
│  Service B  │────▶│  ┌────────┐ ┌──────┐ ┌───────────┐  │────▶│    Pro       │
│  Service C  │────▶│  │Rate    │ │Router│ │Cost       │  │────▶│ Gemini 2.5  │
│  Service D  │────▶│  │Limiter │ │      │ │Tracker    │  │     │    Flash    │
│             │     │  └────────┘ └──────┘ └───────────┘  │     └─────────────┘
└─────────────┘     └──────────────────────────────────────┘

設計原則

ゲートウェイを設計する際の重要な原則は3つあります。

透過性（Transparency）: クライアント側はゲートウェイの存在を意識せず、通常のGemini APIと同じインターフェースで呼び出せること。

耐障害性（Resilience）: 1つのモデルやリージョンが障害を起こしても、自動的に代替パスへフォールバックし、クライアントにエラーを返さないこと。

可観測性（Observability）: すべてのリクエストのレイテンシ、トークン数、コスト、エラー率をリアルタイムに計測・記録すること。

レート制限の実装 — トークンバケットアルゴリズム

Gemini API のレート制限は RPM（Requests Per Minute）と TPM（Tokens Per Minute）の2軸で管理されています。ゲートウェイ側でこれを先回りして制御することで、429エラーの発生を防ぎます。

import time
import asyncio
from dataclasses import dataclass, field
from typing import Optional
 
@dataclass
class TokenBucket:
    """トークンバケットによるレート制限"""
    capacity: float          # バケットの最大容量
    refill_rate: float       # 1秒あたりの補充量
    tokens: float = field(init=False)
    last_refill: float = field(init=False)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
 
    def __post_init__(self):
        self.tokens = self.capacity
        self.last_refill = time.monotonic()
 
    async def acquire(self, count: float = 1.0, timeout: float = 30.0) -> bool:
        """トークンを取得する。取得できるまで待機（タイムアウトあり）"""
        deadline = time.monotonic() + timeout
        async with self._lock:
            while True:
                self._refill()
                if self.tokens >= count:
                    self.tokens -= count
                    return True
                # 次のトークン補充まで待機
                wait_time = (count - 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 RateLimiter:
    """RPM + TPM の二重レート制限"""
 
    def __init__(self, rpm: int = 360, tpm: int = 4_000_000):
        # RPM: 1分あたりのリクエスト数
        self.request_bucket = TokenBucket(
            capacity=rpm,
            refill_rate=rpm / 60.0
        )
        # TPM: 1分あたりのトークン数
        self.token_bucket = TokenBucket(
            capacity=tpm,
            refill_rate=tpm / 60.0
        )
 
    async def acquire_request(self, estimated_tokens: int = 1000) -> bool:
        """リクエスト送信前にレート制限を確認"""
        req_ok = await self.request_bucket.acquire(1.0)
        tok_ok = await self.token_bucket.acquire(float(estimated_tokens))
        return req_ok and tok_ok
 
    async def report_actual_tokens(self, actual_tokens: int, estimated_tokens: int):
        """実際のトークン数で補正（推定より少なければ返却）"""
        diff = estimated_tokens - actual_tokens
        if diff > 0:
            self.token_bucket.tokens = min(
                self.token_bucket.capacity,
                self.token_bucket.tokens + diff
            )

ポイントは 推定トークン数で先に確保し、実測値で補正する パターンです。これにより、リクエスト送信前に枠を確保しつつ、実際の使用量が少なかった場合はバケットに返却してスループットを最大化できます。

自動フォールバック — サーキットブレーカーパターン

特定のモデルで連続してエラーが発生した場合、自動的に代替モデルへ切り替えるサーキットブレーカーを実装します。

import enum
from collections import deque
 
class CircuitState(enum.Enum):
    CLOSED = "closed"        # 正常（リクエストを通す）
    OPEN = "open"            # 遮断（リクエストをブロック）
    HALF_OPEN = "half_open"  # 試行中（1件だけ通す）
 
@dataclass
class CircuitBreaker:
    """モデル単位のサーキットブレーカー"""
    failure_threshold: int = 5       # この回数失敗したらOPENに
    recovery_timeout: float = 60.0   # OPEN後、再試行までの秒数
    success_threshold: int = 2       # HALF_OPENで成功したらCLOSEDに
 
    state: CircuitState = field(default=CircuitState.CLOSED, init=False)
    failures: int = field(default=0, init=False)
    successes_in_half_open: int = field(default=0, init=False)
    last_failure_time: float = field(default=0.0, init=False)
 
    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            # recovery_timeout 経過後に HALF_OPEN へ
            if time.monotonic() - self.last_failure_time >= self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.successes_in_half_open = 0
                return True
            return False
        # HALF_OPEN: 1件ずつ試行
        return True
 
    def record_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.successes_in_half_open += 1
            if self.successes_in_half_open >= self.success_threshold:
                self.state = CircuitState.CLOSED
                self.failures = 0
        else:
            self.failures = 0
 
    def record_failure(self):
        self.failures += 1
        self.last_failure_time = time.monotonic()
        if self.failures >= self.failure_threshold:
            self.state = CircuitState.OPEN

サーキットブレーカーには3つの状態があります。CLOSED（正常稼働）でエラーが閾値を超えるとOPEN（遮断）に遷移し、一定時間後にHALF_OPEN（試行）で1件ずつリクエストを通して回復を確認します。

モデルルーティング — リクエストの複雑さに応じた自動選択

すべてのリクエストに Gemini 2.5 Pro を使うのはコストの無駄です。入力の複雑さを判定して、単純なタスクには Flash、複雑なタスクには Pro を自動ルーティングします。

@dataclass
class ModelConfig:
    name: str
    model_id: str
    cost_per_1k_input: float   # USD per 1K input tokens
    cost_per_1k_output: float  # USD per 1K output tokens
    priority: int              # 低いほど優先
 
# Gemini モデル定義（2026年3月時点の料金）
MODELS = {
    "flash": ModelConfig(
        name="Gemini 2.5 Flash",
        model_id="gemini-2.5-flash-preview-05-20",
        cost_per_1k_input=0.00015,
        cost_per_1k_output=0.0006,
        priority=1,
    ),
    "pro": ModelConfig(
        name="Gemini 2.5 Pro",
        model_id="gemini-2.5-pro-preview-05-06",
        cost_per_1k_input=0.00125,
        cost_per_1k_output=0.005,
        priority=2,
    ),
}
 
class ModelRouter:
    """リクエストの複雑さに基づくモデル自動選択"""
 
    # 複雑なタスクを示すキーワード
    COMPLEX_INDICATORS = [
        "analyze", "compare", "evaluate", "design",
        "architecture", "optimize", "debug", "refactor",
        "分析", "比較", "評価", "設計", "最適化", "デバッグ",
    ]
 
    def select_model(
        self,
        prompt: str,
        force_model: Optional[str] = None,
        max_output_tokens: int = 1024,
    ) -> ModelConfig:
        # 明示的なモデル指定がある場合
        if force_model and force_model in MODELS:
            return MODELS[force_model]
 
        complexity_score = self._estimate_complexity(prompt, max_output_tokens)
 
        # スコア 0.6 以上なら Pro、それ未満なら Flash
        if complexity_score >= 0.6:
            return MODELS["pro"]
        return MODELS["flash"]
 
    def _estimate_complexity(self, prompt: str, max_output_tokens: int) -> float:
        score = 0.0
        prompt_lower = prompt.lower()
 
        # キーワードマッチ
        keyword_hits = sum(
            1 for kw in self.COMPLEX_INDICATORS if kw in prompt_lower
        )
        score += min(keyword_hits * 0.15, 0.45)
 
        # 入力長（長いプロンプトは複雑な傾向）
        if len(prompt) > 3000:
            score += 0.2
        elif len(prompt) > 1000:
            score += 0.1
 
        # 出力トークン数が多い場合
        if max_output_tokens > 4096:
            score += 0.2
 
        return min(score, 1.0)

実運用ではキーワードマッチだけでなく、過去のリクエスト履歴からクライアントごとの傾向を学習し、ルーティング精度を向上させることも可能です。

トークンコスト追跡 — リアルタイムダッシュボード対応

ゲートウェイを通過するすべてのリクエストのコストをリアルタイムに記録します。

import json
import datetime
from collections import defaultdict
 
@dataclass
class UsageRecord:
    timestamp: str
    client_id: str
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
    latency_ms: float
    status: str  # "success" | "error" | "fallback"
 
class CostTracker:
    """トークンコストのリアルタイム追跡"""
 
    def __init__(self):
        self.records: list[UsageRecord] = []
        self.daily_totals: dict[str, float] = defaultdict(float)  # date -> USD
        self.client_totals: dict[str, float] = defaultdict(float)  # client -> USD
 
    def record(
        self,
        client_id: str,
        model_config: ModelConfig,
        input_tokens: int,
        output_tokens: int,
        latency_ms: float,
        status: str = "success",
    ) -> UsageRecord:
        cost = (
            (input_tokens / 1000) * model_config.cost_per_1k_input
            + (output_tokens / 1000) * model_config.cost_per_1k_output
        )
        now = datetime.datetime.now(datetime.timezone.utc)
        record = UsageRecord(
            timestamp=now.isoformat(),
            client_id=client_id,
            model=model_config.name,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            cost_usd=round(cost, 6),
            latency_ms=round(latency_ms, 2),
            status=status,
        )
        self.records.append(record)
        self.daily_totals[now.strftime("%Y-%m-%d")] += cost
        self.client_totals[client_id] += cost
        return record
 
    def get_daily_summary(self, date: Optional[str] = None) -> dict:
        target = date or datetime.datetime.now().strftime("%Y-%m-%d")
        day_records = [r for r in self.records if r.timestamp.startswith(target)]
        return {
            "date": target,
            "total_requests": len(day_records),
            "total_cost_usd": round(sum(r.cost_usd for r in day_records), 4),
            "total_input_tokens": sum(r.input_tokens for r in day_records),
            "total_output_tokens": sum(r.output_tokens for r in day_records),
            "avg_latency_ms": round(
                sum(r.latency_ms for r in day_records) / max(len(day_records), 1), 2
            ),
            "error_rate": round(
                sum(1 for r in day_records if r.status == "error")
                / max(len(day_records), 1),
                4,
            ),
        }

CostTracker は、各リクエストのコストをモデルの料金体系に基づいて自動計算し、日次・クライアント別の集計を即座に返せます。Prometheus や Grafana と連携すれば、リアルタイムダッシュボードの構築も容易です。

ゲートウェイ統合 — すべてを組み合わせる

4つのコンポーネントを統合した AIGateway クラスの実装です。

import google.generativeai as genai
 
class AIGateway:
    """Gemini API AIゲートウェイ（統合プロキシ）"""
 
    def __init__(self, api_key: str, rpm: int = 360, tpm: int = 4_000_000):
        genai.configure(api_key=api_key)
        self.router = ModelRouter()
        self.cost_tracker = CostTracker()
        self.rate_limiters: dict[str, RateLimiter] = {
            "flash": RateLimiter(rpm=rpm, tpm=tpm),
            "pro": RateLimiter(rpm=rpm // 2, tpm=tpm // 2),
        }
        self.circuit_breakers: dict[str, CircuitBreaker] = {
            "flash": CircuitBreaker(),
            "pro": CircuitBreaker(),
        }
 
    async def generate(
        self,
        prompt: str,
        client_id: str = "default",
        force_model: Optional[str] = None,
        max_output_tokens: int = 1024,
        temperature: float = 0.7,
    ) -> dict:
        """ゲートウェイ経由でGemini APIを呼び出す"""
        # 1. モデルルーティング
        model_config = self.router.select_model(prompt, force_model, max_output_tokens)
        model_key = "pro" if "Pro" in model_config.name else "flash"
 
        # 2. サーキットブレーカー確認 → フォールバック
        if not self.circuit_breakers[model_key].can_execute():
            fallback_key = "flash" if model_key == "pro" else "pro"
            if self.circuit_breakers[fallback_key].can_execute():
                model_config = MODELS[fallback_key]
                model_key = fallback_key
            else:
                raise Exception("All models are in circuit-open state")
 
        # 3. レート制限チェック
        estimated_tokens = len(prompt) // 4 + max_output_tokens
        if not await self.rate_limiters[model_key].acquire_request(estimated_tokens):
            raise Exception(f"Rate limit exceeded for {model_config.name}")
 
        # 4. API呼び出し
        start_time = time.monotonic()
        try:
            model = genai.GenerativeModel(model_config.model_id)
            response = model.generate_content(
                prompt,
                generation_config=genai.types.GenerationConfig(
                    max_output_tokens=max_output_tokens,
                    temperature=temperature,
                ),
            )
            elapsed_ms = (time.monotonic() - start_time) * 1000
 
            # 5. 成功記録
            input_tokens = response.usage_metadata.prompt_token_count
            output_tokens = response.usage_metadata.candidates_token_count
 
            self.circuit_breakers[model_key].record_success()
            await self.rate_limiters[model_key].report_actual_tokens(
                input_tokens + output_tokens, estimated_tokens
            )
 
            record = self.cost_tracker.record(
                client_id=client_id,
                model_config=model_config,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                latency_ms=elapsed_ms,
            )
 
            return {
                "text": response.text,
                "model": model_config.name,
                "usage": {
                    "input_tokens": input_tokens,
                    "output_tokens": output_tokens,
                    "cost_usd": record.cost_usd,
                },
                "latency_ms": elapsed_ms,
            }
 
        except Exception as e:
            elapsed_ms = (time.monotonic() - start_time) * 1000
            self.circuit_breakers[model_key].record_failure()
 
            # フォールバック試行
            fallback_key = "flash" if model_key == "pro" else "pro"
            if self.circuit_breakers[fallback_key].can_execute():
                return await self._fallback_generate(
                    prompt, fallback_key, client_id, max_output_tokens, temperature
                )
            raise
 
    async def _fallback_generate(
        self, prompt, model_key, client_id, max_output_tokens, temperature
    ):
        """フォールバック先のモデルで再試行"""
        model_config = MODELS[model_key]
        start_time = time.monotonic()
 
        model = genai.GenerativeModel(model_config.model_id)
        response = model.generate_content(
            prompt,
            generation_config=genai.types.GenerationConfig(
                max_output_tokens=max_output_tokens,
                temperature=temperature,
            ),
        )
        elapsed_ms = (time.monotonic() - start_time) * 1000
 
        input_tokens = response.usage_metadata.prompt_token_count
        output_tokens = response.usage_metadata.candidates_token_count
 
        record = self.cost_tracker.record(
            client_id=client_id,
            model_config=model_config,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            latency_ms=elapsed_ms,
            status="fallback",
        )
 
        return {
            "text": response.text,
            "model": f"{model_config.name} (fallback)",
            "usage": {
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
                "cost_usd": record.cost_usd,
            },
            "latency_ms": elapsed_ms,
        }

実際の使用例を示します。

import asyncio
 
# 期待する出力:
# Model: Gemini 2.5 Flash (単純なリクエストはFlashにルーティング)
# Cost: $0.000XXX
# Daily Summary: {'date': '2026-03-26', 'total_requests': 1, ...}
 
async def main():
    gateway = AIGateway(api_key="YOUR_GEMINI_API_KEY")
 
    # 単純なリクエスト → Flash にルーティング
    result = await gateway.generate(
        prompt="Pythonのリスト内包表記について簡潔に説明してください",
        client_id="service-a",
    )
    print(f"Model: {result['model']}")
    print(f"Cost: ${result['usage']['cost_usd']}")
 
    # 複雑なリクエスト → Pro にルーティング
    result = await gateway.generate(
        prompt="以下のマイクロサービスアーキテクチャを分析し、ボトルネックを特定して最適化案を設計してください...",
        client_id="service-b",
    )
    print(f"Model: {result['model']}")
 
    # 日次サマリー
    print(gateway.cost_tracker.get_daily_summary())
 
asyncio.run(main())

本番デプロイのベストプラクティス

環境変数とシークレット管理

APIキーやモデルのエンドポイントは環境変数で管理し、コードにハードコードしない点が肝心です。

import os
 
GATEWAY_CONFIG = {
    "api_key": os.environ["GEMINI_API_KEY"],
    "rpm": int(os.environ.get("GATEWAY_RPM", "360")),
    "tpm": int(os.environ.get("GATEWAY_TPM", "4000000")),
    "log_level": os.environ.get("GATEWAY_LOG_LEVEL", "INFO"),
}

ヘルスチェックエンドポイント

ゲートウェイ自体の健全性を外部から確認できるヘルスチェックも必須です。サーキットブレーカーの状態、レート制限の残量、直近のエラー率を返すエンドポイントを設けましょう。

ログとアラート

すべてのリクエストを構造化ログ（JSON）で出力し、エラー率が閾値を超えた場合にSlackやPagerDutyへ通知する仕組みを組み込みます。CostTracker の日次コストが予算上限に達した場合のアラートも重要です。

まとめ

AIゲートウェイは、Gemini APIを本番運用する上で避けて通れないインフラレイヤーです。本記事で解説した4つのコンポーネント — レート制限（トークンバケット）、自動フォールバック（サーキットブレーカー）、モデルルーティング、コスト追跡 — を組み合わせることで、信頼性とコスト効率を両立した運用基盤を構築できます。

まずは小規模なプロジェクトでゲートウェイを導入し、トラフィックの増加に合わせてスケーリングしていくのが現実的なアプローチです。コスト追跡のデータが蓄積されれば、ルーティングの最適化やプロンプトの改善にも活用でき、継続的な改善サイクルが回り始めます。

本記事のテーマをさらに深く