個人で開発したアプリに 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
それぞれのパターンを、実装コードとともに解説します。
パターン 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 に統一し、表示時のみユーザーのタイムゾーンに変換します。
本番移行チェックリストと判断フレームワーク
5 つのパターンを一度に全部導入する必要はありません。サービスのフェーズに合わせた優先順位があります。
フェーズ 1（プロトタイプ〜最初の 100 ユーザー）
- ✅ パターン 1 のエクスポネンシャルバックオフ（最低限の安定性）
- ✅ 基本的なログ出力（エラー時のスタックトレースを含む）
- ✅ Gemini API のスペンドキャップ設定
フェーズ 2（100〜1,000 ユーザー）
- ✅ パターン 2 の L2 Redis キャッシュ（コスト最適化の効果が出始める）
- ✅ パターン 4 の構造化ログ（問題調査の効率化）
- ✅ パターン 5 の日次コストアラート
フェーズ 3（1,000 ユーザー以上）
- ✅ パターン 1 のサーキットブレーカー
- ✅ パターン 3 のユーザークォータ管理（ヘビーユーザー対策）
- ✅ パターン 2 の Context Caching（長い System Instruction を使う場合）
- ✅ メトリクスダッシュボードの整備
どのフェーズでも「なぜこの設計が必要か」を理解した上で導入することが大切です。コードをコピーするだけでなく、自分のサービスのトラフィックパターンやコスト構造に合わせた調整が必要です。
この記事で紹介したパターンは、いずれも実際のプロダクションで試したものです。完璧なアーキテクチャを最初から作ろうとするより、まず基本を押さえて段階的に改善していく方が、実際には安定した本番運用につながります。
ぜひ自分のサービスに合った形でアレンジして使ってみてください。もし特定のパターンについてもっと詳しく知りたい部分があれば、各トピックの深掘り記事もご参照いただければ幸いです。