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

Gemini API 本番パフォーマンスチューニング — レイテンシ・スループット・コストの3軸最適化戦略

Gemini API の本番環境でレイテンシ・スループット・コストを同時に最適化する実装パターンを解説。Flex/Priority推論・Context Caching・モデルルーティング・非同期バッチ処理の統合戦略を、動作するコードとベンチマーク結果で示します。

gemini-api²⁷⁹ performance⁴ optimization⁴ production¹⁰⁵ latency² throughput cost⁴ flex-inference context-caching⁴ streaming¹⁸

✦ プレミアム記事

2014年から個人で iOS / Android アプリを開発し、AdMob 中心の収益化を続けてきました。累計 5,000万ダウンロードを超えた壁紙アプリと癒し系アプリの運用に加え、Dolice Labs として Claude Lab・Gemini Lab・Antigravity Lab・Rork Lab の 4 サイトを並行運営しています。記事自動生成・SEO 監査・購入者向け配信のいずれも Gemini API が裏で動いていて、本番運用していると レスポンスの遅さ・月末の請求額・スループットの壁 に同時にぶつかります。

このページは、4 サイト合計で月数十万トークン規模を捌くなかで「効いた手」と「効かなかった手」の差を、実測値と動くコードで残したものです。机上のチューニングではなく、AdMob 収益で支える個人開発スタックで実際に検証した内容だけを書きます。

「速い・安い・大量に捌ける」は同時に実現できるか

Gemini API を本番環境で運用し始めると、最初にぶつかる壁は3つあります。ユーザーが体感するレスポンス速度、月末に届く請求額、そしてトラフィックが増えたときの処理能力です。厄介なのは、この3つがトレードオフの関係にあること。高速なモデルはコストが高く、コストを下げるとレイテンシが犠牲になり、スループットを上げようとすると両方に影響します。

私が複数のプロダクションサービスで Gemini API を運用してきた経験から言えるのは、このトレードオフは設計パターンの選択で大幅に緩和できるということです。ここではレイテンシ・スループット・コストの3軸を同時に最適化する実装戦略を、実測データとコードで示す。

レイテンシの解剖 — 何がレスポンスを遅くしているのか

Gemini API のレスポンス時間を改善するには、まず遅延がどこで発生しているかを正確に把握する必要があります。多くの開発者が「モデルが遅い」と漠然と感じているが、実際にはネットワーク・入力処理・推論・出力生成の各フェーズで異なるボトルネックがあります。

以下のプロファイリングコードで、各フェーズの所要時間を計測できる:

import time
import google.genai as genai
 
client = genai.Client(api_key="YOUR_API_KEY")
 
def profile_request(model: str, prompt: str, **kwargs) -> dict:
    """Gemini API リクエストの各フェーズを計測する"""
    timings = {}
 
    # ネットワーク接続 + リクエスト送信
    t0 = time.perf_counter()
    stream = client.models.generate_content_stream(
        model=model,
        contents=prompt,
        config=genai.types.GenerateContentConfig(**kwargs),
    )
 
    # Time To First Token (TTFT)
    first_chunk = None
    for chunk in stream:
        if first_chunk is None:
            timings["ttft"] = time.perf_counter() - t0
            first_chunk = chunk
            tokens_received = 1
            continue
        tokens_received += 1
 
    timings["total"] = time.perf_counter() - t0
    timings["generation"] = timings["total"] - timings["ttft"]
    timings["tokens"] = tokens_received
    timings["tokens_per_sec"] = tokens_received / timings["generation"] if timings["generation"] > 0 else 0
 
    return timings
 
# 使用例: Flash vs Pro の比較計測
for model in ["gemini-2.5-flash", "gemini-2.5-pro"]:
    result = profile_request(model, "Pythonでバイナリサーチを実装して、計算量を解説してください")
    print(f"\n{model}:")
    print(f"  TTFT: {result['ttft']:.2f}s")
    print(f"  Total: {result['total']:.2f}s")
    print(f"  Tokens/sec: {result['tokens_per_sec']:.1f}")
 
# 期待される出力例:
# gemini-2.5-flash:
#   TTFT: 0.38s
#   Total: 2.14s
#   Tokens/sec: 142.3
# gemini-2.5-pro:
#   TTFT: 1.12s
#   Total: 5.87s
#   Tokens/sec: 68.7

このプロファイラで分かるのは、TTFT（最初のトークンまでの時間）と生成速度の違いです。Flash モデルは TTFT が Pro の約3分の1で、生成速度も約2倍速い。しかし「常に Flash を使えばいい」という単純な話ではありません。Pro が必要な複雑なタスクで Flash を使うと、出力品質が下がって再試行が増え、結果的に総コストとレイテンシが悪化します。

✦

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

この記事の続きを読む

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

この記事で得られること

✦P95 TTFT を 1.8秒 → 0.6秒（67%削減）に短縮した実運用パイプラインの全構成

✦月額 Gemini API コスト $420 → $180（57%削減）を実現した Context Caching と Model Routing の損益分岐閾値

✦5,000万DL の壁紙アプリ運用と Dolice Labs 4サイト並行運用から導いた、429エラー多発時の AdaptiveRateLimiter チューニング表

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

✦

この記事を購入する

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

または

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

ストリーミングで体感速度を劇的に改善する

ユーザーが体感する「速さ」は、実はレスポンス全体の完了時間ではなく、最初の文字が表示されるまでの時間で決まる。ストリーミングを使えば、TTFT の時点でユーザーに応答を返し始められます。

しかし、ストリーミングの実装には注意すべき落とし穴があります。単純に generate_content_stream を使うだけでは、エラーハンドリングやタイムアウト処理が抜け落ちる:

import asyncio
from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
async def resilient_stream(
    model: str,
    prompt: str,
    timeout_seconds: float = 30.0,
    max_retries: int = 2,
):
    """タイムアウト・リトライ付きのストリーミング生成"""
    last_error = None
 
    for attempt in range(max_retries + 1):
        try:
            buffer = []
            async for chunk in await client.aio.models.generate_content_stream(
                model=model,
                contents=prompt,
                config=types.GenerateContentConfig(
                    temperature=0.7,
                    max_output_tokens=2048,
                ),
            ):
                if chunk.text:
                    buffer.append(chunk.text)
                    yield chunk.text  # クライアントへ即座に送信
 
            return  # 成功したらループを抜ける
 
        except Exception as e:
            last_error = e
            if attempt < max_retries:
                # 指数バックオフ: 1秒, 2秒, 4秒...
                wait = 2 ** attempt
                print(f"⚠ Attempt {attempt+1} failed: {e}. Retry in {wait}s")
                await asyncio.sleep(wait)
            else:
                raise RuntimeError(
                    f"Streaming failed after {max_retries+1} attempts: {last_error}"
                )
 
# FastAPI との統合例
# from fastapi import FastAPI
# from fastapi.responses import StreamingResponse
# app = FastAPI()
#
# @app.get("/chat")
# async def chat(q: str):
#     return StreamingResponse(
#         resilient_stream("gemini-2.5-flash", q),
#         media_type="text/plain",
#     )

なぜリトライ付きストリーミングが必要か。本番環境では、ネットワーク瞬断やサーバー側の一時的な過負荷で接続が切れることがあります。単純なストリーミング実装だと、途中で切れた応答がそのままユーザーに返ってしまう。指数バックオフ付きリトライを入れることで、一時的な障害を透過的に処理できます。

Flex 推論と Priority 推論 — コストとレイテンシの意図的な制御

2026年4月に導入された Flex 推論と Priority 推論は、コストとレイテンシのトレードオフをアプリケーション側で明示的に選択できる仕組みです。

3つの推論ティアの使い分けを整理する:

Standard（デフォルト）: 通常価格・通常レイテンシ。対話型のチャットやリアルタイム応答に最適
Flex: Standard の50%割引。レイテンシは1〜15分の範囲で保証なし。バックグラウンド処理や非対話型タスクに使う
Priority: Standard の1.75〜2倍の価格。最低レイテンシと最高の可用性を保証。決済処理や顧客対応チャットボットなど、ダウンタイムが許されない場面向け

重要なのは、同一アプリケーション内で複数のティアを使い分けるパターンです:

from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
class AdaptiveInference:
    """リクエストの性質に応じて推論ティアを自動選択する"""
 
    def __init__(self):
        self.model = "gemini-2.5-flash"
 
    async def generate(
        self,
        prompt: str,
        priority: str = "auto",
        user_facing: bool = True,
    ) -> str:
        # ティアの自動判定
        if priority == "auto":
            tier = self._select_tier(prompt, user_facing)
        else:
            tier = priority
 
        config_kwargs = {
            "temperature": 0.7,
            "max_output_tokens": 2048,
        }
 
        # Flex / Priority の設定
        # service_tier を httpOptions で渡す
        # （2026年4月時点の Python SDK では
        #   GenerateContentConfig に service_tier を追加）
        if tier == "flex":
            config_kwargs["http_options"] = {"headers": {"X-Goog-Api-Service-Tier": "flex"}}
        elif tier == "priority":
            config_kwargs["http_options"] = {"headers": {"X-Goog-Api-Service-Tier": "priority"}}
 
        response = await client.aio.models.generate_content(
            model=self.model,
            contents=prompt,
            config=types.GenerateContentConfig(**config_kwargs),
        )
        return response.text
 
    def _select_tier(self, prompt: str, user_facing: bool) -> str:
        """リクエスト特性からティアを判定"""
        # ユーザーが画面の前で待っている → Standard or Priority
        if user_facing:
            # 短いプロンプトは Standard で十分速い
            if len(prompt) < 500:
                return "standard"
            # 長文処理でも対話型なら Priority で低レイテンシを保証
            return "priority"
 
        # バックグラウンド処理 → Flex でコスト50%削減
        return "flex"
 
# 使用例
engine = AdaptiveInference()
 
# ユーザー対話: Standard で即応答
# answer = await engine.generate("この関数のバグを教えて", user_facing=True)
 
# バッチレポート生成: Flex で50%コスト削減
# report = await engine.generate(long_document, user_facing=False)
 
# 決済確認メッセージ: Priority で最高信頼性
# confirmation = await engine.generate(payment_prompt, priority="priority")

この設計のポイントは、推論ティアの選択をビジネスロジックから分離していることです。user_facing フラグとプロンプト長だけで判定するシンプルなルールだが、実運用では以下の改善が効果的ですった:

深夜帯（ユーザーが少ない時間）はバッチ処理を Flex に寄せてコスト削減
SLA が厳しい企業向けエンドポイントは常に Priority
社内ツールは Standard 固定（コストと速度のバランスが良い）

Context Caching — 繰り返しコンテキストのコスト削減

Gemini API の Context Caching は、同じシステムプロンプトや参照ドキュメントを繰り返し送信するコストを劇的に削減します。キャッシュされたトークンの入力コストは通常の約25%になります。

しかし、Context Caching には「最低32,768トークン」という下限があります。短いプロンプトでは使えないし、キャッシュ自体にもストレージコストがかかります。損益分岐点を正確に把握しないと、逆にコストが増える:

from google import genai
from google.genai import types
import hashlib
 
client = genai.Client(api_key="YOUR_API_KEY")
 
class SmartCacheManager:
    """Context Caching の損益分岐点を考慮した自動管理"""
 
    # キャッシュストレージ: $1.00 / 1M tokens / hour（2.5 Flash の場合）
    # 通常入力: $0.15 / 1M tokens
    # キャッシュ入力: $0.0375 / 1M tokens（75%割引）
    # 損益分岐: 1時間あたり N 回以上同じコンテキストを送るなら得
    # N = storage_cost / (normal_cost - cached_cost) per token
    # N ≈ $1.00 / ($0.15 - $0.0375) = 8.9 回/時間
 
    BREAK_EVEN_REQUESTS_PER_HOUR = 9  # 1時間に9回以上で元が取れる
 
    def __init__(self):
        self._cache_registry: dict[str, str] = {}  # hash -> cache_name
        self._request_counts: dict[str, list[float]] = {}  # hash -> timestamps
 
    def _content_hash(self, content: str) -> str:
        return hashlib.sha256(content.encode()).hexdigest()[:16]
 
    def should_cache(self, content: str, expected_rpm: float) -> bool:
        """キャッシュすべきかどうかを判定"""
        token_estimate = len(content) / 4  # 粗い推定
        if token_estimate < 32768:
            return False  # 最低トークン数未満
        return expected_rpm * 60 >= self.BREAK_EVEN_REQUESTS_PER_HOUR
 
    async def get_or_create_cache(
        self,
        system_content: str,
        model: str = "gemini-2.5-flash",
        ttl_minutes: int = 60,
    ) -> str | None:
        """キャッシュを取得、なければ作成。損益分岐に達しない場合はNone"""
        content_hash = self._content_hash(system_content)
 
        # 既存キャッシュがあればそのまま返す
        if content_hash in self._cache_registry:
            return self._cache_registry[content_hash]
 
        # 新規キャッシュ作成
        cache = await client.aio.caches.create(
            model=model,
            config=types.CreateCachedContentConfig(
                contents=[
                    types.Content(
                        role="user",
                        parts=[types.Part(text=system_content)],
                    )
                ],
                ttl=f"{ttl_minutes * 60}s",
                display_name=f"ctx-{content_hash}",
            ),
        )
        self._cache_registry[content_hash] = cache.name
        return cache.name
 
    async def generate_with_cache(
        self,
        cache_name: str,
        user_prompt: str,
        model: str = "gemini-2.5-flash",
    ) -> str:
        """キャッシュを使った生成"""
        response = await client.aio.models.generate_content(
            model=model,
            contents=user_prompt,
            config=types.GenerateContentConfig(
                cached_content=cache_name,
                temperature=0.7,
            ),
        )
        return response.text
 
# 使用例: 大量のカスタマーサポート問い合わせを同じナレッジベースで処理
# manager = SmartCacheManager()
# cache = await manager.get_or_create_cache(knowledge_base_text)
# if cache:
#     answer = await manager.generate_with_cache(cache, user_question)
# else:
#     # キャッシュ不要（リクエスト頻度が低い場合）→ 通常リクエスト
#     answer = await client.aio.models.generate_content(...)

損益分岐の計算を自動化しているのがポイントです。「とりあえずキャッシュしておく」と、利用頻度が低いコンテキストでは逆にストレージコストが積み上がる。実測では、1時間に10回以上同じコンテキストを使うケースでのみ、キャッシュが有利だった。

インテリジェント・モデルルーティング — タスク難度に応じたモデル自動選択

すべてのリクエストに同じモデルを使うのは、あらゆる荷物をトレーラーで運ぶようなものです。軽い荷物には軽トラで十分であるように、簡単なタスクには Flash、複雑なタスクには Pro を使い分けることで、コストとレイテンシの両方を最適化できます。

問題は「このリクエストは Flash で十分か、Pro が必要か」の判定です。以下の実装では、タスクの複雑さをヒューリスティック＋軽量分類で判定する:

import re
from dataclasses import dataclass
from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
@dataclass
class RoutingDecision:
    model: str
    reason: str
    estimated_cost_multiplier: float  # Flash基準の倍率
 
class ModelRouter:
    """タスク複雑度に基づくモデルルーティング"""
 
    COMPLEXITY_SIGNALS = {
        "high": [
            r"(?:分析|analyze|compare|比較).*(?:詳細|detail|comprehensive)",
            r"(?:設計|design|architect).*(?:システム|system|production)",
            r"(?:リファクタ|refactor|最適化|optimize).*(?:コード|code)",
            r"(?:数学|math|証明|prove|theorem)",
        ],
        "low": [
            r"(?:翻訳|translate|要約|summar)",
            r"(?:フォーマット|format|変換|convert)",
            r"(?:リスト|list|列挙|enumerate)",
            r"(?:簡単|simple|短く|brief)",
        ],
    }
 
    def route(self, prompt: str, context_tokens: int = 0) -> RoutingDecision:
        complexity = self._estimate_complexity(prompt, context_tokens)
 
        if complexity >= 0.7:
            return RoutingDecision(
                model="gemini-2.5-pro",
                reason=f"High complexity ({complexity:.2f}): Pro model needed",
                estimated_cost_multiplier=8.0,  # Proの入力コストはFlashの約8倍
            )
        elif complexity >= 0.4:
            return RoutingDecision(
                model="gemini-2.5-flash",
                reason=f"Medium complexity ({complexity:.2f}): Flash sufficient",
                estimated_cost_multiplier=1.0,
            )
        else:
            return RoutingDecision(
                model="gemini-2.5-flash",
                reason=f"Low complexity ({complexity:.2f}): Flash with lower tokens",
                estimated_cost_multiplier=1.0,
            )
 
    def _estimate_complexity(self, prompt: str, context_tokens: int) -> float:
        score = 0.5  # ベースライン
 
        # 高複雑度シグナルの検出
        for pattern in self.COMPLEXITY_SIGNALS["high"]:
            if re.search(pattern, prompt, re.IGNORECASE):
                score += 0.15
 
        # 低複雑度シグナルの検出
        for pattern in self.COMPLEXITY_SIGNALS["low"]:
            if re.search(pattern, prompt, re.IGNORECASE):
                score -= 0.15
 
        # コンテキスト長が長い → 複雑なタスクの可能性
        if context_tokens > 50000:
            score += 0.1
        elif context_tokens > 100000:
            score += 0.2
 
        # プロンプト自体が長い → 複雑な指示の可能性
        if len(prompt) > 2000:
            score += 0.1
 
        return max(0.0, min(1.0, score))
 
# 使用例
router = ModelRouter()
 
# 簡単なタスク → Flash
decision = router.route("この英文を日本語に翻訳してください: Hello world")
print(f"Model: {decision.model}, Reason: {decision.reason}")
# 出力: Model: gemini-2.5-flash, Reason: Low complexity (0.35): Flash with lower tokens
 
# 複雑なタスク → Pro
decision = router.route("このマイクロサービスアーキテクチャを分析して、ボトルネックを特定し、詳細な改善案を設計してください", context_tokens=80000)
print(f"Model: {decision.model}, Reason: {decision.reason}")
# 出力: Model: gemini-2.5-pro, Reason: High complexity (0.85): Pro model needed

実運用では、このルーティングでAPI コストが約40%削減された。リクエストの約70%が Flash で処理できる簡単なタスクだったためです。ただし、ルーティングの精度が低いと品質劣化や再試行増加で逆効果になります。定期的にルーティング判定の正答率をモニタリングし、パターンを調整する点が肝心です。

非同期バッチパイプライン — スループットの壁を突破する

Gemini API にはレート制限があります。Tier 1 で 15 RPM（リクエスト/分）、Tier 2 で 2,000 RPM です。単純にループで逐次リクエストを送ると、Tier 1 では1時間に900リクエストしか処理できません。

非同期処理と適応型レート制御を組み合わせることで、レート制限ギリギリまでスループットを引き上げられる:

import asyncio
import time
from collections import deque
from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
class AdaptiveRateLimiter:
    """429エラーに応じてレートを自動調整するリミッター"""
 
    def __init__(self, initial_rpm: int = 1800):
        self.rpm = initial_rpm
        self.min_interval = 60.0 / initial_rpm
        self._timestamps: deque[float] = deque()
        self._lock = asyncio.Lock()
        self._consecutive_429 = 0
 
    async def acquire(self):
        async with self._lock:
            now = time.monotonic()
 
            # 1分以上前のタイムスタンプを除去
            while self._timestamps and now - self._timestamps[0] > 60:
                self._timestamps.popleft()
 
            # レート制限内か確認
            if len(self._timestamps) >= self.rpm:
                wait = 60.0 - (now - self._timestamps[0]) + 0.1
                await asyncio.sleep(wait)
 
            self._timestamps.append(time.monotonic())
 
    def on_429(self):
        """429エラー時にレートを下げる"""
        self._consecutive_429 += 1
        reduction = min(0.5, 0.1 * self._consecutive_429)  # 最大50%削減
        self.rpm = max(10, int(self.rpm * (1 - reduction)))
        self.min_interval = 60.0 / self.rpm
        print(f"⚠ Rate reduced to {self.rpm} RPM (429 count: {self._consecutive_429})")
 
    def on_success(self):
        """成功時にレートを徐々に回復"""
        if self._consecutive_429 > 0:
            self._consecutive_429 = max(0, self._consecutive_429 - 1)
            self.rpm = min(1800, int(self.rpm * 1.05))
            self.min_interval = 60.0 / self.rpm
 
async def batch_generate(
    prompts: list[str],
    model: str = "gemini-2.5-flash",
    concurrency: int = 10,
    rate_limiter: AdaptiveRateLimiter | None = None,
) -> list[str]:
    """適応型レート制御付きバッチ生成"""
    if rate_limiter is None:
        rate_limiter = AdaptiveRateLimiter()
 
    semaphore = asyncio.Semaphore(concurrency)
    results: list[str | None] = [None] * len(prompts)
 
    async def process_one(index: int, prompt: str):
        async with semaphore:
            await rate_limiter.acquire()
            for attempt in range(3):
                try:
                    response = await client.aio.models.generate_content(
                        model=model,
                        contents=prompt,
                        config=types.GenerateContentConfig(
                            temperature=0.7,
                            max_output_tokens=1024,
                        ),
                    )
                    rate_limiter.on_success()
                    results[index] = response.text
                    return
                except Exception as e:
                    error_str = str(e)
                    if "429" in error_str or "RESOURCE_EXHAUSTED" in error_str:
                        rate_limiter.on_429()
                        await asyncio.sleep(2 ** attempt * 2)
                    else:
                        if attempt == 2:
                            results[index] = f"ERROR: {e}"
                            return
                        await asyncio.sleep(2 ** attempt)
 
    tasks = [process_one(i, p) for i, p in enumerate(prompts)]
    await asyncio.gather(*tasks)
    return results
 
# 使用例: 1000件の商品説明を一括生成
# prompts = [f"商品名「{name}」の魅力的な説明文を50字で" for name in product_names]
# results = await batch_generate(prompts, concurrency=15)
# 期待される処理時間: Tier 2（2000 RPM）で約30秒

この実装で重要なのは、429エラーを「障害」ではなく「フィードバック信号」として扱っている点です。レートを動的に調整することで、API の許容限界ギリギリまでスループットを引き上げながら、エラーによるリクエスト損失を最小化できます。

Thinking Budget の最適化 — 推論コストの見えないコスト

Gemini 2.5 Pro や 3 Pro のような推論モデルでは、「考える」プロセスにもトークンが消費されます。Thinking Budget を適切に設定しないと、簡単なタスクでも大量の推論トークンが使われ、コストとレイテンシが不必要に膨らむ。

from google import genai
from google.genai import types
 
client = genai.Client(api_key="YOUR_API_KEY")
 
async def generate_with_thinking_budget(
    prompt: str,
    task_type: str = "general",
) -> str:
    """タスクタイプに応じてThinking Budgetを最適化"""
 
    # タスク別のThinking Budget設定
    budget_map = {
        "simple_qa": 256,       # 簡単な質問: 最小限の推論
        "translation": 512,     # 翻訳: 軽い推論
        "summarization": 1024,  # 要約: 中程度の推論
        "coding": 4096,         # コーディング: 十分な推論
        "analysis": 8192,       # 分析: 深い推論
        "math": 16384,          # 数学・証明: 最大限の推論
        "general": 2048,        # デフォルト
    }
 
    budget = budget_map.get(task_type, budget_map["general"])
 
    response = await client.aio.models.generate_content(
        model="gemini-2.5-pro",
        contents=prompt,
        config=types.GenerateContentConfig(
            thinking_config=types.ThinkingConfig(
                thinking_budget=budget,
            ),
            temperature=0.7,
        ),
    )
 
    # 実際に使用された思考トークン数をログ出力
    if hasattr(response, "usage_metadata"):
        meta = response.usage_metadata
        thinking_tokens = getattr(meta, "thinking_tokens", "N/A")
        print(f"💭 Thinking tokens used: {thinking_tokens} / budget: {budget}")
 
    return response.text
 
# 使用例: 同じモデルでもタスクに応じてコストが変わる
# 簡単な質問（256トークン推論）→ 約$0.001
# answer = await generate_with_thinking_budget("Pythonのリスト内包表記とは？", "simple_qa")
 
# 数学の証明（16384トークン推論）→ 約$0.05
# proof = await generate_with_thinking_budget("フェルマーの小定理を証明して", "math")

Thinking Budget を 0 に設定すると推論なしモード（thinking_mode: off と同等）になるが、これは品質に大きく影響します。実測では、コーディングタスクで推論なしにすると正答率が約30%低下しました。タスクに合った適切な予算設定が、コストと品質のバランスを決める。

本番モニタリング — 3軸を可視化して継続改善する

最適化は一度やって終わりではありません。トラフィックパターンの変化、モデルの更新、新機能のリリースに応じて、継続的にチューニングする必要があります。以下のメトリクスを常時監視する:

import time
import json
from dataclasses import dataclass, field, asdict
from collections import defaultdict
 
@dataclass
class RequestMetrics:
    model: str = ""
    tier: str = "standard"
    ttft_ms: float = 0
    total_ms: float = 0
    input_tokens: int = 0
    output_tokens: int = 0
    thinking_tokens: int = 0
    cached_tokens: int = 0
    status: str = "success"
    estimated_cost_usd: float = 0
 
class PerformanceMonitor:
    """3軸メトリクスの収集と分析"""
 
    # 2026年4月時点の価格（1Mトークンあたり、USD）
    PRICING = {
        "gemini-2.5-flash": {"input": 0.15, "output": 0.60, "cached": 0.0375, "thinking": 0.15},
        "gemini-2.5-pro": {"input": 1.25, "output": 10.00, "cached": 0.3125, "thinking": 1.25},
    }
 
    def __init__(self):
        self.metrics: list[RequestMetrics] = []
 
    def record(self, m: RequestMetrics):
        m.estimated_cost_usd = self._calc_cost(m)
        self.metrics.append(m)
 
    def _calc_cost(self, m: RequestMetrics) -> float:
        prices = self.PRICING.get(m.model, self.PRICING["gemini-2.5-flash"])
        tier_mult = {"standard": 1.0, "flex": 0.5, "priority": 1.875}.get(m.tier, 1.0)
 
        normal_input = m.input_tokens - m.cached_tokens
        cost = (
            (normal_input / 1_000_000) * prices["input"] * tier_mult
            + (m.cached_tokens / 1_000_000) * prices["cached"] * tier_mult
            + (m.output_tokens / 1_000_000) * prices["output"] * tier_mult
            + (m.thinking_tokens / 1_000_000) * prices["thinking"] * tier_mult
        )
        return round(cost, 6)
 
    def summary(self, last_n: int = 100) -> dict:
        recent = self.metrics[-last_n:]
        if not recent:
            return {}
 
        successful = [m for m in recent if m.status == "success"]
        return {
            "total_requests": len(recent),
            "success_rate": f"{len(successful)/len(recent)*100:.1f}%",
            "avg_ttft_ms": f"{sum(m.ttft_ms for m in successful)/max(len(successful),1):.0f}",
            "p95_ttft_ms": f"{sorted(m.ttft_ms for m in successful)[int(len(successful)*0.95)]:.0f}" if successful else "N/A",
            "avg_total_ms": f"{sum(m.total_ms for m in successful)/max(len(successful),1):.0f}",
            "total_cost_usd": f"${sum(m.estimated_cost_usd for m in recent):.4f}",
            "model_distribution": dict(
                defaultdict(int, {m.model: 1 for m in recent})
            ),
        }
 
    def export_jsonl(self, filepath: str):
        with open(filepath, "w") as f:
            for m in self.metrics:
                f.write(json.dumps(asdict(m)) + "\n")
 
# 使用例
# monitor = PerformanceMonitor()
# metrics = RequestMetrics(model="gemini-2.5-flash", ttft_ms=380, ...)
# monitor.record(metrics)
# print(monitor.summary())

公式ドキュメントには書かれていないが運用で気づいた点

ここからは、Gemini API の公式ガイドや料金ページには書かれていないけれど、4 サイト並行運用を 1 年続けて見えてきた非自明な改善ポイントです。机上の最適化と本番運用の差は、こういう小さな前提のズレに集約されます。

1. レイテンシは「平均」ではなく「P95 と P99」で評価する

公式ドキュメントが提示するレイテンシは、ほぼすべて平均値か中央値です。しかし AdMob 連動の壁紙アプリでは、平均が 600ms でも P99 が 4 秒を超えていると、レビュー欄に「重い」「フリーズする」と書かれ、星評価が落ちます。実測ベースで numpy.percentile(latencies, [50, 95, 99]) を毎日記録し、P99 が中央値の 6 倍を超えたら異常と判定する閾値を運用しています。Gemini Flash でも 1 万リクエストに 1 件は 10 秒近くかかるケースがあり、これは Streaming + クライアント側タイムアウト 8 秒で打ち切る方が体感品質が安定します。

2. Context Caching の損益分岐は「文字数」ではなく「保存時間 × 呼び出し回数」

「キャッシュは長文で効く」と書かれていることが多いですが、実際に効くのは「保存しているあいだに何回叩かれるか」です。Dolice Labs では同じシステムプロンプトを 1 日 300 回ほど呼ぶサイトと、20 回しか呼ばないサイトが混在しています。300 回サイトは TTL を 1 時間に固定して常にキャッシュヒットを狙う方が安く、20 回サイトは TTL を 5 分に下げて、本当にバースト的に連続呼び出しが来た瞬間だけキャッシュする方が結果的に安くなりました。cached_tokens_per_hour ÷ 12,000 でブレークイーブンを毎日計算して、TTL を動的に切り替えるだけで月 $30〜$50 のコスト差が出ます。

3. Flex 推論は「夜間 + 翌朝確認」ワークフローで効かせる

公式の Flex 推論は「最大 15 分」と書かれていますが、私の運用では深夜帯（日本時間 02:00〜05:00）に投入すると 95% のリクエストが 3 分以内に完了します。AdMob のレポート集計や、Dolice Labs の SEO 監査バッチをこの時間帯に Flex で流し、翌朝 7 時のニューステッカー生成タスクが結果を読み込むパイプラインに統合したところ、推論コストが Standard 比で 約 40% 削減 されました。日中帯に Flex を使うと 8〜12 分かかる場合があるため、ユーザーが待つフローでは使いません。

4. モデルルーティングは「タスク種別」より「期待出力長」で判断する

最初は Pro / Flash の振り分けを「複雑度」で判定していましたが、誤判定が多く再試行率が 8% に達しました。試行錯誤の末、max_output_tokens の見積もりだけでルーティングする方式に切り替えたところ、再試行率が 1.2% まで下がりました。出力が 200 トークン未満で済むタスク（タグ抽出・分類・要約）はすべて Flash、出力が 800 トークン超のタスク（記事本文生成・コードレビュー）だけ Pro。複雑度の判定よりも、出力長の見積もりの方がブレが小さく、結果としてコスト削減と品質安定の両立になります。

5. 監視は「コスト × エラー率 × 体感速度」を 1 枚のダッシュボードに集約する

3 軸を別々のダッシュボードで見ると、改善判断が遅れます。Cloudflare Workers Analytics と Gemini API の使用量ログを統合した自前のダッシュボードを作り、P95 TTFT × 月次コスト × 5xx エラー率 を 1 つのスコアにまとめて毎朝 7 時に Slack に流すようにしてから、ボトルネックの優先度判断が体感で 3 倍速くなりました。スコアは (2000 - p95_ms) × 0.4 + (50 - monthly_cost_usd/10) × 0.4 + (5 - error_rate_pct) × 4 のような重み付き加算で、自分の感覚に合うように毎月チューニングしています。

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

落とし穴1: ストリーミングで TTFT は速いのに総応答時間が遅い

原因: max_output_tokens を大きく設定しすぎています。モデルは指定上限近くまでトークンを生成しようとする傾向があります。必要な出力長に応じて適切に制限すること。短い回答が期待されるタスクでは max_output_tokens: 256 に設定するだけでレスポンス時間が3〜5倍速くなります。

落とし穴2: Context Caching を入れたのにコストが下がらない

原因: キャッシュの TTL が短すぎるか、リクエスト頻度が損益分岐点に達していません。1時間あたり9回未満のリクエスト頻度では、キャッシュのストレージコストが削減分を上回る。まずプロファイリングで実際の利用頻度を測定し、分岐点を超えるコンテキストだけをキャッシュすること。

落とし穴3: Flex 推論のレスポンスが遅すぎてタイムアウトする

原因: Flex 推論のレイテンシは1〜15分の範囲で変動します。短いタイムアウト（30秒等）を設定していると高頻度で失敗します。Flex を使う場合は、クライアントのタイムアウトを最低15分に設定し、バックグラウンドジョブとして処理する設計にすること。ユーザーが画面の前で待つケースでは Flex を絶対に使いません。

落とし穴4: モデルルーティングで品質が不安定

原因: ルーティング判定の精度が低く、Pro が必要なタスクを Flash に振っています。対処法は、ルーティング判定のログを取り、Flash で処理した結果のユーザー満足度（再試行率、フィードバック評価）を追跡すること。再試行率が5%を超える判定パターンは、Pro にルーティングするよう閾値を調整します。

落とし穴5: 非同期バッチで429エラーが大量発生する

原因: 並行数（concurrency）を Tier のレート制限に対して高く設定しすぎています。Tier 1（15 RPM）では concurrency を 2〜3 に、Tier 2（2000 RPM）では 20〜30 に留める。前述の AdaptiveRateLimiter のように、429エラーをフィードバック信号として動的に調整するのが最善です。