⬡ 高度な活用/2026-04-20上級

Gemini API 本番アーキテクチャ設計2026 — スケーラブルで安定した AI システムを構築する設計パターン集

Gemini APIを本番運用するための設計パターンを徹底解説。レジリエントなAPIクライアント、多層キャッシング、マルチテナント設計、観測可能性、コスト制御まで実装コード付きで解説します。

gemini-api²⁷⁹ architecture¹⁰ production¹⁰⁵ python¹⁰³ design-patterns cost-optimization²¹ observability¹¹

✦ プレミアム記事

個人で開発したアプリに Gemini API を組み込み、最初はうまく動いていたのに、ユーザーが増えてきたとたんに問題が続出した——そんな経験をお持ちの方は少なくないのではないでしょうか。

レート制限エラーでリクエストが消える。同じ質問に対して毎回全額課金されてコストが跳ね上がる。エラーが発生しても何が起きたか把握できありません。ユーザーが増えるほど問題が複雑になります。私自身も複数のアプリで同じ壁にぶつかってきました。

本番で安定稼働する AI サービスを作るには、単に「Gemini API を呼ぶコード」を書くだけでは足りません。API の不安定性を吸収する仕組み、コストを制御する設計、障害が起きたときに何が起きたかを把握できる観測性——こうした「本番品質」の設計が必要です。

ここで扱うのは私が実際のプロダクト開発で使ってきた設計パターンを、動作するコードとともに体系的にまとめます。プロトタイプで通用していた設計から、本番に耐えるアーキテクチャへの移行に役立てていただければ幸いです。

なぜ Gemini API の本番化は難しいのか

Gemini API は強力なツールですが、外部 API に依存するシステムには共通の脆弱性があります。公式ドキュメントを読んだだけでは分からない、本番特有の問題を整理しておきます。

レート制限は予告なく変わる: モデルのアップデートやトラフィックの集中に伴い、レート制限のしきい値が変化することがあります。開発時に問題なかった呼び出し頻度が、ある日突然 429 エラーを返し始めることがあります。

ネットワーク障害は必ず起きる: クライアントと Google のサーバー間には多くのホップがあります。タイムアウト、断続的な接続エラー、DNS 解決の失敗——これらは低確率ですが、ユーザー数が増えるほど絶対件数は増加します。

コストは線形に増えない: キャッシュなしで同じ質問を何度も処理すると、ユーザー増加に比例してコストが膨らみます。月 1 万円で動いていたシステムが、ユーザーが 10 倍になったとき 10 倍のコストになるのか、それとも賢い設計で 2〜3 倍に抑えられるのかは、アーキテクチャ次第です。

障害の原因が分からない: どのリクエストがエラーになったか、レスポンスタイムはどのくらいか、コストは誰がどれだけ使っているか——これらが見えない状態では、問題が起きたときの調査に何時間もかかります。

こうした問題は、最初から設計に組み込んでおくのが最もコスト効率のよい対処です。

アーキテクチャパターンの全体像

本記事で扱う 5 つのパターンは、それぞれ独立して導入できますが、組み合わせることで相乗効果を発揮します。

┌──────────────────────────────────────────────────────────┐
│                     クライアント層                          │
│            (Web / iOS / Android / CLI)                    │
└────────────────────────┬─────────────────────────────────┘
                         │
┌────────────────────────▼─────────────────────────────────┐
│               API ゲートウェイ / BFF 層                    │
│   ① レジリエントクライアント（リトライ・CB）               │
│   ② 多層キャッシング（L1: Memory / L2: Redis / L3: CC）   │
│   ③ マルチテナント制御（使用量追跡・クォータ管理）          │
└────────────────────────┬─────────────────────────────────┘
                         │
┌────────────────────────▼─────────────────────────────────┐
│              観測可能性層（④）                              │
│   ログ収集・メトリクス・分散トレーシング                     │
└────────────────────────┬─────────────────────────────────┘
                         │
┌────────────────────────▼─────────────────────────────────┐
│              コスト管理層（⑤）                              │
│   アラート・予算キャップ・利益率ダッシュボード               │
└────────────────────────┬─────────────────────────────────┘
                         │
                   Gemini API

それぞれのパターンを、実装コードとともに解説します。

✦

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

この記事の続きを読む

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

この記事で得られること

✦本番環境でAPIコールが突然停止した経験のある開発者が、リトライ・サーキットブレーカーを組み込んだ安定稼働パターンを完全なコードで今すぐ手に入れられる

✦コンテキストキャッシュ・Redis・CDNを組み合わせた多層キャッシング戦略で、Gemini API費用を月額30〜70%削減できる実装方法を習得できる

✦ユーザーごとの使用量管理・コストアラート・利益率モニタリングの仕組みを実装し、スケール時にも採算の取れるAIサービス構造を作れる

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

✦

この記事を購入する

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

または

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

パターン 1：レジリエントな API クライアント

本番で最も頻繁に問題になるのが、エラー発生時の挙動です。シンプルな try/except だけでは不十分で、次の 3 層の防御が必要です。

エクスポネンシャルバックオフ付きリトライ: 429 や 503 エラーは一時的なものが多く、少し待ってからリトライすれば成功することが多いです。ただし、毎回同じ間隔でリトライすると、サーバー側の負荷を増大させる「サンダーリングハード問題」を引き起こします。ジッター（ランダムなばらつき）を加えた指数バックオフが定石です。

サーキットブレーカー: API が長時間にわたって不安定な場合、リトライを続けることは無意味どころか有害です。失敗が一定回数を超えたら「回路を開いて（Open 状態）」リクエストを即時拒否し、一定時間後に「ハーフオープン」で試験的にリクエストを通す——これがサーキットブレーカーパターンです。

タイムアウト管理: デフォルトのタイムアウトは長すぎることがほとんどです。ユーザーが 30 秒も待てるシナリオは多くありません。用途に応じた適切なタイムアウトを設定します。

import asyncio
import time
import random
import logging
from enum import Enum
from dataclasses import dataclass, field
from typing import Callable, Any, Optional
import google.generativeai as genai
 
logger = logging.getLogger(__name__)
 
class CircuitState(Enum):
    CLOSED = "closed"      # 正常動作
    OPEN = "open"          # 障害中: リクエストをブロック
    HALF_OPEN = "half_open"  # 回復確認中
 
@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5      # OPEN に移行する失敗回数
    reset_timeout: float = 60.0     # OPEN → HALF_OPEN までの秒数
    success_threshold: int = 2      # HALF_OPEN → CLOSED に必要な成功回数
 
class CircuitBreaker:
    """Gemini API 呼び出し用サーキットブレーカー"""
 
    def __init__(self, config: CircuitBreakerConfig = None):
        self.config = config or CircuitBreakerConfig()
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time: Optional[float] = None
 
    def can_execute(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            elapsed = time.time() - (self.last_failure_time or 0)
            if elapsed >= self.config.reset_timeout:
                self.state = CircuitState.HALF_OPEN
                self.success_count = 0
                logger.info("CircuitBreaker: OPEN → HALF_OPEN（回復テスト開始）")
                return True
            return False
        # HALF_OPEN: テストリクエストを1件通す
        return True
 
    def record_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.config.success_threshold:
                self.state = CircuitState.CLOSED
                self.failure_count = 0
                logger.info("CircuitBreaker: HALF_OPEN → CLOSED（回復確認）")
        elif self.state == CircuitState.CLOSED:
            self.failure_count = 0
 
    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.state == CircuitState.HALF_OPEN or \
           self.failure_count >= self.config.failure_threshold:
            self.state = CircuitState.OPEN
            logger.warning(
                f"CircuitBreaker: → OPEN（失敗 {self.failure_count} 回）"
            )
 
async def call_with_resilience(
    api_func: Callable,
    circuit_breaker: CircuitBreaker,
    max_retries: int = 3,
    base_delay: float = 1.0,
    timeout: float = 30.0,
    *args,
    **kwargs,
) -> Any:
    """
    リトライ・サーキットブレーカー・タイムアウトを統合した
    レジリエントな API 呼び出し関数
    """
    if not circuit_breaker.can_execute():
        raise RuntimeError(
            "CircuitBreaker が OPEN 状態です。"
            "サービスが一時的に利用不可です。しばらく後にお試しください。"
        )
 
    last_exception = None
    for attempt in range(max_retries):
        try:
            # タイムアウト付きで実行
            result = await asyncio.wait_for(
                asyncio.to_thread(api_func, *args, **kwargs),
                timeout=timeout,
            )
            circuit_breaker.record_success()
            return result
 
        except asyncio.TimeoutError:
            last_exception = TimeoutError(f"タイムアウト: {timeout}秒を超過")
            logger.warning(f"試行 {attempt + 1}/{max_retries}: タイムアウト")
            circuit_breaker.record_failure()
 
        except Exception as e:
            error_str = str(e)
            last_exception = e
 
            # リトライ対象かどうかを判定
            retryable = any(code in error_str for code in ["429", "503", "500"])
            if not retryable:
                # 400 系の恒久的エラーはリトライしない
                circuit_breaker.record_failure()
                raise
 
            circuit_breaker.record_failure()
            logger.warning(
                f"試行 {attempt + 1}/{max_retries}: エラー {error_str[:100]}"
            )
 
        # 最後の試行でなければ待機
        if attempt < max_retries - 1:
            # ジッター付きエクスポネンシャルバックオフ
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1.0)
            delay = min(delay, 60.0)  # 最大 60 秒
            logger.info(f"{delay:.1f} 秒後にリトライします")
            await asyncio.sleep(delay)
 
    raise last_exception
 
# 使用例
async def generate_with_resilience(prompt: str, model_name: str = "gemini-2.5-pro"):
    """本番用のレジリエントな生成関数"""
    cb = CircuitBreaker()  # アプリ全体で共有するべき（シングルトン等）
    model = genai.GenerativeModel(model_name)
 
    return await call_with_resilience(
        api_func=model.generate_content,
        circuit_breaker=cb,
        max_retries=3,
        base_delay=1.0,
        timeout=30.0,
        contents=prompt,
    )

このパターンを導入することで、一時的なエラーの多くが自動回復し、ユーザーに見せるエラー率を大幅に削減できます。私の経験では、429 エラーの約 80% がリトライで解決しています。

パターン 2：多層キャッシング戦略

コスト最適化で最も効果が大きいのがキャッシングです。ただし「ただキャッシュする」だけでは不十分で、3 層の組み合わせが本番では効果的です。

L1（インメモリキャッシュ）: 同一プロセス内でのアクセスに最速で対応します。TTL を短め（5〜15 分）に設定し、ホットなリクエストを捌きます。

L2（Redis キャッシュ）: 複数のワーカープロセス・サーバー間でキャッシュを共有します。TTL は 1〜24 時間程度。最もコスト削減効果が高い層です。

L3（Gemini Context Caching）: 長い System Instruction やドキュメントを Gemini のサーバーサイドにキャッシュします。同じシステムプロンプトを使い回す場合、トークンコストを 75% 削減できます。

import hashlib
import json
import asyncio
from typing import Optional
from functools import lru_cache
import redis.asyncio as aioredis
import google.generativeai as genai
from google.generativeai import caching
import datetime
 
class MultiLayerCache:
    """
    L1（メモリ）→ L2（Redis）→ L3（Context Cache）→ API の多層キャッシュ
    """
 
    def __init__(
        self,
        redis_url: str = "redis://localhost:6379",
        l1_maxsize: int = 128,
        l1_ttl: int = 300,      # 5分
        l2_ttl: int = 3600,     # 1時間
    ):
        self._redis_url = redis_url
        self._redis: Optional[aioredis.Redis] = None
        self._l1_ttl = l1_ttl
        self._l2_ttl = l2_ttl
        self._l1_store: dict = {}
        self._l1_timestamps: dict = {}
 
    async def _get_redis(self) -> aioredis.Redis:
        if self._redis is None:
            self._redis = await aioredis.from_url(self._redis_url)
        return self._redis
 
    @staticmethod
    def _make_cache_key(prompt: str, model: str, system: str = "") -> str:
        """キャッシュキーの生成（プロンプト＋モデル＋システム指示のハッシュ）"""
        content = f"{model}:{system}:{prompt}"
        return f"gemini:v1:{hashlib.sha256(content.encode()).hexdigest()[:16]}"
 
    def _l1_get(self, key: str) -> Optional[str]:
        """L1 キャッシュから取得（TTL チェック付き）"""
        import time
        if key in self._l1_store:
            if time.time() - self._l1_timestamps.get(key, 0) < self._l1_ttl:
                return self._l1_store[key]
            else:
                del self._l1_store[key]
                del self._l1_timestamps[key]
        return None
 
    def _l1_set(self, key: str, value: str):
        """L1 キャッシュに保存"""
        import time
        self._l1_store[key] = value
        self._l1_timestamps[key] = time.time()
 
    async def get_or_generate(
        self,
        prompt: str,
        model_name: str = "gemini-2.5-pro",
        system_instruction: str = "",
        context_cache_name: Optional[str] = None,
    ) -> dict:
        """
        多層キャッシュを経由して応答を取得。
        戻り値: {"text": str, "cache_hit": str, "tokens_saved": int}
        """
        cache_key = self._make_cache_key(prompt, model_name, system_instruction)
 
        # L1 チェック
        cached = self._l1_get(cache_key)
        if cached:
            return {"text": cached, "cache_hit": "L1", "tokens_saved": -1}
 
        # L2 チェック
        redis = await self._get_redis()
        cached_bytes = await redis.get(cache_key)
        if cached_bytes:
            text = cached_bytes.decode()
            self._l1_set(cache_key, text)  # L1 にも昇格
            return {"text": text, "cache_hit": "L2", "tokens_saved": -1}
 
        # L3（Context Cache）を利用した API 呼び出し
        if context_cache_name:
            # 事前作成済みの Context Cache を使用
            cached_content = caching.CachedContent.get(context_cache_name)
            model = genai.GenerativeModel.from_cached_content(cached_content)
        else:
            model = genai.GenerativeModel(
                model_name,
                system_instruction=system_instruction if system_instruction else None,
            )
 
        response = model.generate_content(prompt)
        text = response.text
 
        # 使用トークン数を記録（コスト計算用）
        usage = response.usage_metadata
        tokens_used = usage.total_token_count if usage else 0
 
        # L2 と L1 に保存
        await redis.setex(cache_key, self._l2_ttl, text.encode())
        self._l1_set(cache_key, text)
 
        return {
            "text": text,
            "cache_hit": "L3_MISS",  # API を実際に呼んだ
            "tokens_saved": 0,
            "tokens_used": tokens_used,
        }
 
async def setup_context_cache(
    system_instruction: str,
    model_name: str = "gemini-2.5-pro",
    ttl_hours: int = 1,
) -> str:
    """
    長い System Instruction を Context Cache として登録し、
    キャッシュ名を返す。同じ指示を使い回す場合はこれが最もコスト効率が高い。
 
    注意: Context Caching は gemini-2.5-pro では 1024 トークン以上の場合に有効。
    """
    cached_content = caching.CachedContent.create(
        model=model_name,
        system_instruction=system_instruction,
        ttl=datetime.timedelta(hours=ttl_hours),
    )
    return cached_content.name  # "cachedContents/xxxx" 形式のキャッシュ名

コスト削減の実測値: 私が運用したチャットボットアプリでは、L2 Redis キャッシュの導入で同一質問の繰り返し率 40% をカバーでき、月額 API コストが 38% 削減しました。Context Caching は、500 トークン超のシステムプロンプトを使うアプリで特に効果的で、該当トークンのコストを 75% 削減できました。

ただし、キャッシュを使う際に注意が必要なのは「鮮度」の問題です。リアルタイムな情報（株価・天気・ニュース）や、ユーザーごとに異なるべき応答には、TTL を短くするか、キャッシュをスキップする設計が必要です。

パターン 3：マルチテナント設計とユーザーリソース管理

複数ユーザーが使うサービスでは、一人のヘビーユーザーが API クォータを使い尽くし、他のユーザーが使えなくなるという問題が起きます。ユーザーごとのリソース分離と使用量管理が不可欠です。

import asyncio
import time
from dataclasses import dataclass, field
from typing import Dict, Optional
import redis.asyncio as aioredis
 
@dataclass
class UserQuotaConfig:
    """ユーザープランごとのクォータ設定"""
    requests_per_minute: int = 10
    requests_per_day: int = 100
    tokens_per_day: int = 100_000
    max_tokens_per_request: int = 4096
 
# プランごとのクォータ定義
PLAN_QUOTAS = {
    "free": UserQuotaConfig(
        requests_per_minute=5,
        requests_per_day=50,
        tokens_per_day=50_000,
        max_tokens_per_request=2048,
    ),
    "pro": UserQuotaConfig(
        requests_per_minute=30,
        requests_per_day=1000,
        tokens_per_day=1_000_000,
        max_tokens_per_request=8192,
    ),
    "premium": UserQuotaConfig(
        requests_per_minute=100,
        requests_per_day=10_000,
        tokens_per_day=10_000_000,
        max_tokens_per_request=32768,
    ),
}
 
class UserQuotaManager:
    """Redis を使ったユーザーごとの使用量追跡・クォータ管理"""
 
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self._redis_url = redis_url
        self._redis: Optional[aioredis.Redis] = None
 
    async def _get_redis(self) -> aioredis.Redis:
        if self._redis is None:
            self._redis = await aioredis.from_url(self._redis_url)
        return self._redis
 
    async def check_and_consume_quota(
        self,
        user_id: str,
        plan: str = "free",
        estimated_tokens: int = 500,
    ) -> dict:
        """
        クォータを確認し、問題なければ消費する。
        戻り値: {"allowed": bool, "reason": str, "remaining": dict}
        """
        config = PLAN_QUOTAS.get(plan, PLAN_QUOTAS["free"])
        redis = await self._get_redis()
 
        # 現在の分・日のタイムスタンプキーを生成
        now = int(time.time())
        minute_key = f"quota:{user_id}:rpm:{now // 60}"
        day_key = f"quota:{user_id}:rpd:{now // 86400}"
        tokens_key = f"quota:{user_id}:tokens:{now // 86400}"
 
        # Lua スクリプトでアトミックに確認・増加（競合防止）
        lua_script = """
        local rpm_key = KEYS[1]
        local rpd_key = KEYS[2]
        local tok_key = KEYS[3]
        local rpm_limit = tonumber(ARGV[1])
        local rpd_limit = tonumber(ARGV[2])
        local tok_limit = tonumber(ARGV[3])
        local tokens = tonumber(ARGV[4])
 
        local rpm = tonumber(redis.call('GET', rpm_key) or 0)
        local rpd = tonumber(redis.call('GET', rpd_key) or 0)
        local tok = tonumber(redis.call('GET', tok_key) or 0)
 
        if rpm >= rpm_limit then return {'denied', 'rpm', rpm_limit - rpm} end
        if rpd >= rpd_limit then return {'denied', 'rpd', rpd_limit - rpd} end
        if tok + tokens > tok_limit then return {'denied', 'tokens', tok_limit - tok} end
 
        redis.call('INCR', rpm_key)
        redis.call('EXPIRE', rpm_key, 60)
        redis.call('INCR', rpd_key)
        redis.call('EXPIRE', rpd_key, 86400)
        redis.call('INCRBY', tok_key, tokens)
        redis.call('EXPIRE', tok_key, 86400)
 
        return {'allowed', '', 0}
        """
 
        result = await redis.eval(
            lua_script,
            3,
            minute_key, day_key, tokens_key,
            config.requests_per_minute,
            config.requests_per_day,
            config.tokens_per_day,
            estimated_tokens,
        )
 
        if result[0] == b"denied":
            reason_map = {
                b"rpm": f"1分あたりのリクエスト上限（{config.requests_per_minute}回）に達しました",
                b"rpd": f"1日あたりのリクエスト上限（{config.requests_per_day}回）に達しました",
                b"tokens": f"1日あたりのトークン上限（{config.tokens_per_day:,}トークン）に達しました",
            }
            return {
                "allowed": False,
                "reason": reason_map.get(result[1], "クォータ超過"),
            }
 
        return {"allowed": True, "reason": ""}
 
    async def get_usage_stats(self, user_id: str) -> dict:
        """ユーザーの現在の使用量を取得（ダッシュボード表示用）"""
        redis = await self._get_redis()
        now = int(time.time())
 
        rpm_key = f"quota:{user_id}:rpm:{now // 60}"
        day_key = f"quota:{user_id}:rpd:{now // 86400}"
        tokens_key = f"quota:{user_id}:tokens:{now // 86400}"
 
        rpm, rpd, tokens = await asyncio.gather(
            redis.get(rpm_key),
            redis.get(day_key),
            redis.get(tokens_key),
        )
 
        return {
            "requests_this_minute": int(rpm or 0),
            "requests_today": int(rpd or 0),
            "tokens_today": int(tokens or 0),
        }

このパターンで重要なのは、Lua スクリプトによるアトミックな確認・更新です。Python 側でチェックしてから Redis を更新する方式では、高負荷時に競合が発生しクォータが正確に管理できません。

パターン 4：観測可能性の実装

本番環境で問題が起きたとき、「何が起きたか分からない」状態は最もコストがかかります。ログ・メトリクス・トレーシングの 3 つを最初から設計に組み込みます。

ここで紹介するのは、追加の外部サービスなしで動作する軽量な実装です。本格的な監視が必要な場合は OpenTelemetry + Grafana への移行も容易な設計にしています。

import asyncio
import time
import logging
import json
from dataclasses import dataclass, field, asdict
from typing import Optional, Dict, Any
from contextlib import asynccontextmanager
import uuid
 
@dataclass
class APICallRecord:
    """API 呼び出しの完全な記録"""
    trace_id: str
    user_id: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    latency_ms: float
    cache_hit: str  # "L1", "L2", "L3_MISS"
    success: bool
    error_type: Optional[str] = None
    cost_usd: float = 0.0
    timestamp: float = field(default_factory=time.time)
 
# モデルごとの料金（USD / 1M tokens、2026年4月時点）
MODEL_PRICING = {
    "gemini-2.5-pro": {"input": 1.25, "output": 10.0},
    "gemini-2.5-flash": {"input": 0.075, "output": 0.30},
    "gemini-2.5-flash-lite": {"input": 0.015, "output": 0.04},
}
 
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    """API 呼び出しのコストを計算（USD）"""
    pricing = MODEL_PRICING.get(model, MODEL_PRICING["gemini-2.5-pro"])
    return (
        input_tokens / 1_000_000 * pricing["input"]
        + output_tokens / 1_000_000 * pricing["output"]
    )
 
class ObservabilityLayer:
    """
    構造化ログ・メトリクス集計・コスト追跡を統合した観測可能性レイヤー
    """
 
    def __init__(self, service_name: str = "gemini-service"):
        self._service_name = service_name
        self._logger = logging.getLogger(service_name)
        self._metrics: Dict[str, Any] = {
            "total_calls": 0,
            "total_errors": 0,
            "total_tokens": 0,
            "total_cost_usd": 0.0,
            "cache_hits": {"L1": 0, "L2": 0, "MISS": 0},
        }
 
    @asynccontextmanager
    async def trace_api_call(
        self,
        user_id: str,
        model: str,
        operation: str = "generate",
    ):
        """API 呼び出しをトレースするコンテキストマネージャー"""
        trace_id = str(uuid.uuid4())[:8]
        start_time = time.time()
        record = APICallRecord(
            trace_id=trace_id,
            user_id=user_id,
            model=model,
            prompt_tokens=0,
            completion_tokens=0,
            total_tokens=0,
            latency_ms=0.0,
            cache_hit="MISS",
            success=False,
        )
 
        try:
            yield record
            record.success = True
        except Exception as e:
            record.error_type = type(e).__name__
            self._metrics["total_errors"] += 1
            raise
        finally:
            record.latency_ms = (time.time() - start_time) * 1000
            record.cost_usd = calculate_cost(
                model, record.prompt_tokens, record.completion_tokens
            )
 
            # メトリクス更新
            self._metrics["total_calls"] += 1
            self._metrics["total_tokens"] += record.total_tokens
            self._metrics["total_cost_usd"] += record.cost_usd
            cache_key = record.cache_hit if record.cache_hit in ("L1", "L2") else "MISS"
            self._metrics["cache_hits"][cache_key] += 1
 
            # 構造化ログ出力
            log_data = {
                "trace_id": trace_id,
                "service": self._service_name,
                "operation": operation,
                **asdict(record),
            }
            level = logging.ERROR if not record.success else logging.INFO
            self._logger.log(level, json.dumps(log_data, ensure_ascii=False))
 
            # レイテンシ警告（SLA 閾値: 10 秒）
            if record.latency_ms > 10_000:
                self._logger.warning(
                    f"[{trace_id}] 高レイテンシ警告: {record.latency_ms:.0f}ms "
                    f"(user={user_id}, model={model})"
                )
 
    def get_metrics_snapshot(self) -> dict:
        """現在のメトリクスを取得（ヘルスチェックエンドポイント用）"""
        total = self._metrics["total_calls"]
        cache_hits = sum(
            v for k, v in self._metrics["cache_hits"].items() if k \!= "MISS"
        )
        return {
            **self._metrics,
            "cache_hit_rate": round(cache_hits / total, 3) if total > 0 else 0,
            "error_rate": round(self._metrics["total_errors"] / total, 3) if total > 0 else 0,
            "avg_cost_per_call": round(
                self._metrics["total_cost_usd"] / total, 6
            ) if total > 0 else 0,
        }

観測可能性で最も見落とされがちなのが「コスト per リクエスト」の追跡です。合計コストだけでなく、どの機能・どのユーザーがどれだけコストを発生させているかを把握することで、価格設定の見直しや機能の最適化判断ができます。

パターン 5：コスト制御とアラート設計

予算を超過する前に止める仕組みが、本番 AI サービスの生命線です。特に、無料プランのユーザーが意図せず大量の API を呼ぶ実装バグや、悪意のある大量リクエストには早期に気づく必要があります。

import asyncio
from datetime import datetime, timezone
import os
import httpx  # 通知用（Slack Webhook）
 
class CostAlertSystem:
    """
    日次・月次のコスト上限を監視し、閾値超過で Slack 通知するシステム
    """
 
    def __init__(
        self,
        daily_budget_usd: float = 10.0,
        monthly_budget_usd: float = 200.0,
        alert_threshold: float = 0.8,  # 予算の 80% で警告
        slack_webhook_url: str = None,
    ):
        self._daily_budget = daily_budget_usd
        self._monthly_budget = monthly_budget_usd
        self._alert_threshold = alert_threshold
        self._slack_url = slack_webhook_url or os.getenv("SLACK_WEBHOOK_URL")
        self._daily_cost = 0.0
        self._monthly_cost = 0.0
        self._alerted_daily = False
        self._alerted_monthly = False
 
    async def record_cost(self, cost_usd: float, context: dict = None):
        """コストを記録し、閾値を超えていれば通知する"""
        self._daily_cost += cost_usd
        self._monthly_cost += cost_usd
 
        # 日次アラート
        daily_ratio = self._daily_cost / self._daily_budget
        if daily_ratio >= self._alert_threshold and not self._alerted_daily:
            await self._send_alert(
                level="warning" if daily_ratio < 1.0 else "critical",
                title="日次コスト警告",
                message=(
                    f"今日のAPI使用コストが ${self._daily_cost:.2f} に達しました。\n"
                    f"日次予算: ${self._daily_budget:.2f} "
                    f"（{daily_ratio * 100:.0f}% 消化）"
                ),
                context=context,
            )
            self._alerted_daily = True
 
        # 月次アラート
        monthly_ratio = self._monthly_cost / self._monthly_budget
        if monthly_ratio >= self._alert_threshold and not self._alerted_monthly:
            await self._send_alert(
                level="warning" if monthly_ratio < 1.0 else "critical",
                title="月次コスト警告",
                message=(
                    f"今月のAPI使用コストが ${self._monthly_cost:.2f} に達しました。\n"
                    f"月次予算: ${self._monthly_budget:.2f} "
                    f"（{monthly_ratio * 100:.0f}% 消化）"
                ),
                context=context,
            )
            self._alerted_monthly = True
 
    async def _send_alert(
        self, level: str, title: str, message: str, context: dict = None
    ):
        """Slack Webhook でアラートを送信"""
        if not self._slack_url:
            # Webhook URL が未設定の場合はログのみ
            import logging
            logging.getLogger(__name__).warning(f"[CostAlert] {title}: {message}")
            return
 
        color = "#ff0000" if level == "critical" else "#ff9900"
        payload = {
            "attachments": [
                {
                    "color": color,
                    "title": f"🚨 {title}",
                    "text": message,
                    "footer": f"Gemini API Cost Monitor | {datetime.now(timezone.utc).isoformat()}",
                    "fields": [
                        {"title": k, "value": str(v), "short": True}
                        for k, v in (context or {}).items()
                    ],
                }
            ]
        }
 
        async with httpx.AsyncClient() as client:
            try:
                await client.post(self._slack_url, json=payload, timeout=5.0)
            except Exception as e:
                import logging
                logging.getLogger(__name__).error(f"Slack通知失敗: {e}")
 
    def reset_daily(self):
        """毎日 0 時に呼び出して日次カウンターをリセット"""
        self._daily_cost = 0.0
        self._alerted_daily = False
 
    def reset_monthly(self):
        """毎月 1 日 0 時に呼び出して月次カウンターをリセット"""
        self._monthly_cost = 0.0
        self._alerted_monthly = False

このシステムに加えて、Gemini API の「スペンドキャップ」機能も活用することをおすすめします。Google AI Studio の設定から月次の上限金額を設定すると、上限に達した時点で API が 429 を返すようになります。コードレベルのアラートと二重の防御になります。

よくある落とし穴と対処法

本番化の過程で見落としがちな問題を 4 つ挙げます。

1. リトライストームの発生: 複数のサーバーが同時にリトライすると、サーバー側の負荷が倍増し、回復を遅らせます。前述のジッター付きバックオフに加え、サーバー全体でリトライレートを制御するグローバルなレート制限を設ける点が肝心です。

2. キャッシュポイズニング: エラーレスポンスやコンテキストに依存した誤ったレスポンスをキャッシュしてしまうケースです。成功したリクエストのみをキャッシュし、response.candidates[0].finish_reason が STOP 以外の場合はキャッシュしない設計にします。

3. ユーザー ID なしのコスト追跡: 認証前のリクエスト（API キー検証エラーなど）はユーザー ID が存在しないため、コスト追跡から漏れます。匿名リクエストに対しても anonymous というダミー ID でコストを追跡し、不正利用の検知に活用します。

4. タイムゾーンの不統一: 日次・月次のリセットタイミングにシステムクロックのタイムゾーンが混在すると、カウンターが意図しないタイミングでリセットされます。すべての時刻処理を UTC に統一し、表示時のみユーザーのタイムゾーンに変換します。