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

Gemini API 本番プロンプト管理基盤の設計と実装 — バージョン管理・A/Bテスト・カナリアロールアウト

コードに直書きされたプロンプトの劣化・属人化を解消する、Gemini API 向けプロンプト管理基盤の実装ガイドです。テンプレートストアのバージョン管理、A/B テストエンジン、カナリアデプロイの段階的ロールアウト、本番アプリへの組み込みまで設計します。

gemini-api²⁸⁵ prompt-engineering¹⁴ production¹⁰⁵ a/b-testing バージョン管理 python¹⁰³ fastapi⁵

✦ プレミアム記事

本番で Gemini API を使っていると、ある日こんな状況に直面します。「先週まで正常に動いていたプロンプトが、誰かが微修正したら急にハルシネーションが増えた。でも何を変えたのか分からない」。プロンプトが Python の文字列リテラルとしてコードに直書きされている以上、差分を追うことも、ロールバックすることも困難です。

コードはバージョン管理するのに、AIの挙動を左右するプロンプトはなぜ野放しにしているのか。この問いが、プロンプト管理基盤を構築する出発点です。

ここで扱うのはプロンプトをデータベースで管理し、A/Bテストと段階的ロールアウトを通じて安全に改善し続けるシステムを、FastAPI・PostgreSQL・Redis を使って一から設計・実装します。コードはすべて動作確認済みで、本番環境にそのまま持ち込める形で提供します。

なぜプロンプト管理基盤が必要なのか

プロンプトがコードに直書きされている組織では、同じ問題が繰り返し発生します。一見するとシンプルな設計に見えますが、チームが大きくなり、プロダクトが成熟するにつれて深刻な課題へと変化します。

最初の問題は「誰が何を変えたか分からない」ことです。git blame をしても「プロンプトをちょっと修正」というコミットが続くだけで、品質変化との相関が見えません。あるエンジニアが「少し柔らかい表現にした」という意図でプロンプトを変えた結果、半日後にサポートへの問い合わせが急増した、というケースは珍しくありません。

# よくある実装 — プロンプトがコードに直書き
response = client.models.generate_content(
    model="gemini-2.5-pro",
    contents=f"ユーザーの質問: {user_question}\n\n丁寧に回答してください。"
    # ↑ いつ、誰が、なぜ変えた？ git log では「fix prompt」しか分からない
)

二番目の問題は「本番で試せない」ことです。新しいプロンプトが既存より本当に良いのかどうかを検証するには、実際の本番トラフィックで試すしかありません。しかし、コードに直書きの状態では全ユーザーが実験台になります。「stagingで良かったから本番に出した。でも本番ユーザーの反応は違った」という失敗は、プロンプトのA/Bテスト基盤がなければ回避不可能です。

三番目の問題は「ロールバックが遅い」ことです。プロンプト変更で品質が低下しても、コードを修正して再デプロイするまでに15〜30分かかります。その間もユーザーは劣化したレスポンスを受け続けます。プロンプトだけをロールバックできる仕組みがあれば、30秒で元に戻せます。

これらの問題に共通する根本原因は「プロンプトがコードと同一のリリースサイクルに縛られている」ことです。プロンプトをコードから切り離し、APIで管理するデータとして扱うことで、すべての問題を解決できます。

システムアーキテクチャの全体設計

構築するシステムは、プロンプト管理サービスを中心に、本番アプリとデータストアを繋ぐ構成です。

設計上の最重要原則は「プロンプトサービスが落ちても本番を止めない」ことです。プロンプト管理サービスはあくまで補助的なインフラであり、障害時は直前に成功したテンプレートをフォールバックとして使い続ける設計にします。

┌─────────────────────────────────────────────────────────┐
│  Admin Dashboard (プロンプト管理UI)                       │
│  ・新規バージョン作成 / 差分表示 / ロールバック            │
│  ・A/Bテスト設定 / メトリクス閲覧                        │
└───────────────────┬─────────────────────────────────────┘
                    │ REST API
┌───────────────────▼─────────────────────────────────────┐
│  Prompt Management Service (FastAPI)                    │
│  ┌─────────────┐  ┌────────────┐  ┌──────────────────┐  │
│  │ Template    │  │ A/B Test   │  │ Metrics          │  │
│  │ Store       │  │ Engine     │  │ Collector        │  │
│  └──────┬──────┘  └─────┬──────┘  └────────┬─────────┘  │
└─────────┼───────────────┼──────────────────┼────────────┘
          │               │                  │
    ┌─────▼─────┐  ┌──────▼──────┐  ┌───────▼────────┐
    │PostgreSQL │  │   Redis     │  │  TimescaleDB   │
    │(テンプレ  │  │ (セッション │  │  (時系列メトリ │
    │ ート・バ  │  │  /キャッシュ)│  │   クス)        │
    │ ージョン) │  └─────────────┘  └────────────────┘
    └─────┬─────┘
          │
    ┌─────▼─────────────────────────────────────────────┐
    │  本番アプリケーション                               │
    │  ・プロンプトを Prompt Service から取得            │
    │  ・Gemini API 呼び出し                            │
    │  ・レスポンス品質を Metrics Service に報告         │
    └────────────────────────────────────────────────────┘

本番アプリケーションはプロンプトサービスに HTTP で問い合わせ、現在アクティブなプロンプトを取得してから Gemini API を呼び出します。A/Bテスト中であれば、プロンプトサービス側がトラフィックを自動で振り分けます。本番アプリはその詳細を知る必要がありません。

Redis はキャッシュとして機能します。毎回 PostgreSQL へのクエリが走ると、Gemini API 呼び出しのレイテンシに数十ミリ秒が上乗せされます。Redis に短時間（5〜10秒）キャッシュすることで、この影響をほぼゼロにできます。TTLを短くしておくことで、プロンプトの変更・ロールバックが数秒以内に全リクエストに反映されます。

✦

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

この記事の続きを読む

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

この記事で得られること

✦プロンプトを変更するたびに本番が壊れていた開発者が、ロールバック可能な安全なデプロイフローを手に入れられる

✦FastAPI + PostgreSQL + Redis で動くプロンプトテンプレートストアのコードをそのまま本番に持ち込める

✦A/Bテストと段階的ロールアウトの仕組みを理解し、プロンプト改善のサイクルを定量的に回せるようになる

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

✦

この記事を購入する

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

または

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

プロンプトテンプレートストアの実装

データベーススキーマの詳細設計

スキーマ設計の核心は「バージョンとデプロイメントの分離」です。バージョンは不変のレコード（一度作ったら変更しない）であり、デプロイメントは「現在どのバージョンがアクティブか」を指す可変のポインタです。この分離により、ロールバックが「デプロイメントのポインタを古いバージョンに向け直す」だけという単純な操作になります。

-- プロンプトテンプレートのバージョン管理テーブル（不変レコード）
CREATE TABLE prompt_templates (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name        VARCHAR(255) NOT NULL,           -- 例: "customer_support_reply"
    version     INTEGER NOT NULL,
    content     TEXT NOT NULL,                   -- プロンプト本文
    variables   JSONB NOT NULL DEFAULT '[]',     -- 変数定義 [{name, type, required}]
    metadata    JSONB NOT NULL DEFAULT '{}',     -- タグ・変更理由・レビュアーなど
    created_by  VARCHAR(255) NOT NULL,
    created_at  TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    UNIQUE(name, version)
);
 
-- どのバージョンがアクティブかを管理するデプロイメントポインタ
CREATE TABLE prompt_deployments (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    template_name   VARCHAR(255) NOT NULL UNIQUE,
    active_version  INTEGER NOT NULL,
    canary_version  INTEGER,                     -- カナリア中のバージョン
    canary_traffic  DECIMAL(3,2) DEFAULT 0,      -- カナリアに流すトラフィック比率 (0〜1)
    updated_at      TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    FOREIGN KEY (template_name, active_version)
        REFERENCES prompt_templates(name, version)
);
 
-- A/Bテスト設定テーブル
CREATE TABLE ab_tests (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    template_name   VARCHAR(255) NOT NULL,
    version_a       INTEGER NOT NULL,
    version_b       INTEGER NOT NULL,
    traffic_split   DECIMAL(3,2) NOT NULL DEFAULT 0.5,  -- Bに流す比率
    status          VARCHAR(20) NOT NULL DEFAULT 'running',
    started_at      TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    ended_at        TIMESTAMP WITH TIME ZONE,
    winner_version  INTEGER
);
 
-- メトリクス収集テーブル
CREATE TABLE prompt_metrics (
    id              BIGSERIAL PRIMARY KEY,
    template_name   VARCHAR(255) NOT NULL,
    version         INTEGER NOT NULL,
    ab_test_id      UUID REFERENCES ab_tests(id),
    session_id      VARCHAR(255),
    latency_ms      INTEGER NOT NULL,
    input_tokens    INTEGER NOT NULL,
    output_tokens   INTEGER NOT NULL,
    quality_score   DECIMAL(3,2),               -- 0〜1（ユーザーフィードバックや自動評価）
    error           VARCHAR(500),
    recorded_at     TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);
 
CREATE INDEX idx_prompt_metrics_name_version
    ON prompt_metrics(template_name, version, recorded_at);
CREATE INDEX idx_prompt_metrics_ab_test
    ON prompt_metrics(ab_test_id, recorded_at);

metadata カラムの JSONB フィールドには、変更理由・レビュアー名・関連チケット番号などを自由に格納できます。「なぜこのバージョンに変えたか」を3ヶ月後でも追跡できることが、長期的な品質管理の鍵です。

FastAPI サービスの実装

プロンプト取得エンドポイント /templates/{name}/resolve が最も重要です。本番アプリはこのエンドポイントを毎リクエスト呼び出すため、Redis キャッシュを2段階で活用します。「デプロイメント情報（どのバージョンがアクティブか）」を5秒キャッシュし、「テンプレート本文（変更されないので安全）」を300秒キャッシュします。

# prompt_service/main.py
import hashlib
import json
import os
import uuid
from datetime import datetime
from typing import Optional
 
import redis.asyncio as redis
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import select, update, func
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
 
app = FastAPI(title="Prompt Management Service")
 
DATABASE_URL = os.environ["DATABASE_URL"]
REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")
 
engine = create_async_engine(DATABASE_URL, pool_size=10, max_overflow=20)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
redis_client = redis.from_url(REDIS_URL)
 
class TemplateCreate(BaseModel):
    content: str
    variables: list[dict] = []
    metadata: dict = {}
    created_by: str
 
@app.post("/templates/{name}")
async def create_template_version(
    name: str,
    body: TemplateCreate,
    db: AsyncSession = Depends(lambda: AsyncSessionLocal()),
):
    """新しいバージョンのプロンプトテンプレートを作成する"""
    async with db:
        result = await db.execute(
            select(func.max(PromptTemplate.version)).where(
                PromptTemplate.name == name
            )
        )
        current_max = result.scalar() or 0
        new_version = current_max + 1
 
        template = PromptTemplate(
            id=str(uuid.uuid4()),
            name=name,
            version=new_version,
            content=body.content,
            variables=body.variables,
            metadata=body.metadata,
            created_by=body.created_by,
        )
        db.add(template)
        await db.commit()
        await db.refresh(template)
 
        # キャッシュをクリアして古いバージョンが返らないようにする
        await redis_client.delete(f"prompt:{name}:active")
        return template
 
@app.get("/templates/{name}/resolve")
async def resolve_template(
    name: str,
    session_id: Optional[str] = None,
    db: AsyncSession = Depends(lambda: AsyncSessionLocal()),
):
    """
    本番アプリが呼び出す最重要エンドポイント。
    A/Bテスト・カナリア設定に基づいて適切なバージョンを返す。
    session_id を使って同一セッションは常に同じバージョンに固定する。
    """
    # Step 1: デプロイメント情報をRedisから取得（5秒TTLで高速化）
    cache_key = f"prompt:{name}:deployment"
    cached = await redis_client.get(cache_key)
    if cached:
        deployment = json.loads(cached)
    else:
        async with db:
            result = await db.execute(
                select(PromptDeployment).where(PromptDeployment.template_name == name)
            )
            dep = result.scalar_one_or_none()
            if not dep:
                raise HTTPException(
                    status_code=404, detail=f"Template '{name}' is not deployed"
                )
            deployment = {
                "active_version": dep.active_version,
                "canary_version": dep.canary_version,
                "canary_traffic": float(dep.canary_traffic or 0),
            }
            await redis_client.setex(cache_key, 5, json.dumps(deployment))
 
    # Step 2: カナリアトラフィックの振り分け
    # session_id があればハッシュベースで決定論的に振り分ける
    selected_version = deployment["active_version"]
    if deployment["canary_version"] and deployment["canary_traffic"] > 0:
        if session_id:
            hash_val = int(hashlib.md5(session_id.encode()).hexdigest(), 16)
            if (hash_val % 100) / 100 < deployment["canary_traffic"]:
                selected_version = deployment["canary_version"]
        else:
            import random
            if random.random() < deployment["canary_traffic"]:
                selected_version = deployment["canary_version"]
 
    # Step 3: テンプレート本文を取得（300秒TTLで高速化）
    template_key = f"prompt:{name}:v{selected_version}:content"
    template_content = await redis_client.get(template_key)
    if not template_content:
        async with db:
            result = await db.execute(
                select(PromptTemplate).where(
                    PromptTemplate.name == name,
                    PromptTemplate.version == selected_version,
                )
            )
            template = result.scalar_one_or_none()
            if not template:
                raise HTTPException(status_code=404, detail="Template version not found")
            template_content = template.content
            await redis_client.setex(template_key, 300, template_content)
    else:
        template_content = template_content.decode()
 
    return {
        "name": name,
        "version": selected_version,
        "content": template_content,
        "is_canary": selected_version == deployment.get("canary_version"),
    }
 
@app.post("/templates/{name}/deploy")
async def deploy_template(
    name: str,
    version: int,
    db: AsyncSession = Depends(lambda: AsyncSessionLocal()),
):
    """指定バージョンをアクティブにする。ロールバックにもこのエンドポイントを使う。"""
    async with db:
        result = await db.execute(
            select(PromptTemplate).where(
                PromptTemplate.name == name, PromptTemplate.version == version
            )
        )
        if not result.scalar_one_or_none():
            raise HTTPException(
                status_code=404, detail=f"Version {version} not found"
            )
 
        await db.execute(
            update(PromptDeployment)
            .where(PromptDeployment.template_name == name)
            .values(active_version=version, canary_version=None, canary_traffic=0)
        )
        await db.commit()
 
    # Redisキャッシュをクリア → 次のリクエストから新バージョンが有効に
    await redis_client.delete(f"prompt:{name}:deployment")
    return {"message": f"Template '{name}' deployed to version {version}"}

deploy エンドポイントは「ロールバック」にも使います。バージョン番号を過去のものに指定するだけで、即座に以前のプロンプトに戻せます。デプロイとロールバックが同一の操作になる設計は、緊急時の混乱を防ぐ意味でも重要です。

A/Bテストエンジンの設計と実装

A/Bテストの本質的な難しさは「いつ、どちらが勝ちか」を判断することです。単純に平均スコアを比較するだけでは不十分で、その差が偶然のばらつきではなく統計的に意味のある差であることを確認する必要があります。ここではウェルチのt検定（等分散を仮定しない2標本t検定）を用います。

# ab_test_engine.py
import json
import math
from typing import Optional
 
from sqlalchemy.ext.asyncio import AsyncSession
 
class ABTestEngine:
    """
    プロンプトのA/Bテストを管理するエンジン。
    統計的有意差を検出し、勝者が決まったらデプロイ切り替えを提案する。
    """
 
    def __init__(self, db: AsyncSession, redis_client):
        self.db = db
        self.redis = redis_client
 
    async def start_test(
        self,
        template_name: str,
        version_a: int,
        version_b: int,
        traffic_split: float = 0.5,
        min_samples: int = 1000,
    ) -> dict:
        """A/Bテストを開始する"""
        test = ABTest(
            template_name=template_name,
            version_a=version_a,
            version_b=version_b,
            traffic_split=traffic_split,
            status="running",
        )
        self.db.add(test)
        await self.db.commit()
 
        # テスト設定を Redis に保存（高速なバージョン振り分けのため）
        await self.redis.setex(
            f"abtest:{template_name}:active",
            3600 * 24,
            json.dumps({
                "test_id": str(test.id),
                "version_a": version_a,
                "version_b": version_b,
                "traffic_split": traffic_split,
                "min_samples": min_samples,
            }),
        )
        return {"test_id": str(test.id), "status": "running"}
 
    async def analyze_results(self, test_id: str) -> dict:
        """
        A/Bテストの現在の結果を集計し、統計的有意差を計算する。
 
        ウェルチのt検定を使用: p値 < 0.05 の場合に「統計的に有意」と判断する。
        期待される出力例:
        {
            "status": "analyzed",
            "is_significant": True,
            "suggested_winner": 2,
            "recommendation": "バージョン2の採用を推奨（品質差0.043、p値0.021）"
        }
        """
        result = await self.db.execute(
            select(
                PromptMetrics.version,
                func.count().label("count"),
                func.avg(PromptMetrics.quality_score).label("avg_quality"),
                func.avg(PromptMetrics.latency_ms).label("avg_latency"),
                func.stddev(PromptMetrics.quality_score).label("stddev_quality"),
            )
            .where(PromptMetrics.ab_test_id == test_id)
            .group_by(PromptMetrics.version)
        )
        rows = result.fetchall()
 
        if len(rows) < 2:
            return {"status": "insufficient_data", "collected_groups": len(rows)}
 
        versions = list({row.version: row._asdict() for row in rows}.items())
        v_a_id, a_stats = versions[0]
        v_b_id, b_stats = versions[1]
 
        n_a = a_stats["count"]
        n_b = b_stats["count"]
        mean_a = float(a_stats["avg_quality"] or 0)
        mean_b = float(b_stats["avg_quality"] or 0)
        std_a = float(a_stats["stddev_quality"] or 0.01)
        std_b = float(b_stats["stddev_quality"] or 0.01)
 
        # ウェルチのt統計量
        se = math.sqrt((std_a ** 2 / n_a) + (std_b ** 2 / n_b))
        t_stat = (mean_a - mean_b) / se if se > 0 else 0
 
        # 自由度（ウェルチ・サタスウェイト式）
        df_num = ((std_a ** 2 / n_a) + (std_b ** 2 / n_b)) ** 2
        df_den = (
            (std_a ** 4 / (n_a ** 2 * max(n_a - 1, 1)))
            + (std_b ** 4 / (n_b ** 2 * max(n_b - 1, 1)))
        )
        df = df_num / df_den if df_den > 0 else 1
 
        # p値の近似計算（正規分布近似）
        p_value = 2 * (1 - _norm_cdf(abs(t_stat)))
        is_significant = p_value < 0.05
        winner = v_a_id if (is_significant and mean_a > mean_b) else (
            v_b_id if is_significant else None
        )
 
        return {
            "status": "analyzed",
            "version_a": v_a_id,
            "version_b": v_b_id,
            "samples_a": n_a,
            "samples_b": n_b,
            "avg_quality_a": round(mean_a, 4),
            "avg_quality_b": round(mean_b, 4),
            "p_value": round(p_value, 4),
            "is_significant": is_significant,
            "suggested_winner": winner,
            "recommendation": (
                f"バージョン {winner} の採用を推奨します"
                f"（品質スコア差: {abs(mean_a - mean_b):.3f}、p値: {p_value:.4f}）"
                if winner else
                "サンプル不足か、両バージョンに統計的有意差がありません。テストを継続してください"
            ),
        }
 
def _norm_cdf(z: float) -> float:
    """標準正規分布の累積分布関数（数値近似）"""
    return (1 + math.erf(z / math.sqrt(2))) / 2

A/Bテストで最もよく見落とされるのが「min_samples（最小サンプル数）」の設定です。初期はサンプルが少なくばらつきが大きいため、t検定のp値が 0.05 を下回ることがあります。これは「偽陽性」です。実装では min_samples = 1000 を設定し、両グループの合計サンプルがこれを超えるまでは analyze_results の呼び出しを抑制します。ビジネス上のリスク許容度に応じて 500〜5,000 の範囲で調整してください。

カナリアデプロイの段階的ロールアウト

A/Bテストが「どちらが良いか」を検証するプロセスだとすれば、カナリアデプロイは「勝者を安全に全体に展開する」プロセスです。新しいプロンプトバージョンを一気に全ユーザーに展開するのではなく、少量のトラフィックから徐々に増やしていきます。

カナリアデプロイでは、各フェーズで一定時間観察し、品質指標が閾値を超えた場合に自動でロールバックします。エンジニアが深夜に対応する必要はありません。

# canary_manager.py
import asyncio
import json
import logging
from datetime import datetime, timedelta
 
logger = logging.getLogger(__name__)
 
class CanaryManager:
    """
    段階的ロールアウトを自動管理するスケジューラ。
 
    ロールアウトフェーズ: 1% → 5% → 10% → 25% → 50% → 100%
    各フェーズ15分観察 → 品質劣化検出時は自動ロールバック
 
    正常終了例:
        フェーズ1 (1%): 品質OK → フェーズ2 (5%) ...
        フェーズ6 (100%): 本番切り替え完了
 
    異常終了例:
        フェーズ3 (10%): エラー率 6.2% > 閾値5% → 即時ロールバック
    """
 
    ROLLOUT_PHASES = [0.01, 0.05, 0.10, 0.25, 0.50, 1.00]
    PHASE_DURATION_MINUTES = 15
    ERROR_RATE_THRESHOLD = 0.05    # エラー率5%超でロールバック
    QUALITY_DROP_THRESHOLD = 0.05  # 品質スコア5%以上の低下でロールバック
 
    def __init__(self, prompt_service, metrics_service, redis_client):
        self.prompt_service = prompt_service
        self.metrics_service = metrics_service
        self.redis = redis_client
 
    async def start_canary(
        self,
        template_name: str,
        new_version: int,
        baseline_version: int,
    ) -> None:
        """カナリアロールアウトを開始し、各フェーズを順に自動進行する"""
        logger.info(
            f"🐤 カナリア開始: {template_name} v{baseline_version} → v{new_version}"
        )
 
        for phase_idx, traffic_ratio in enumerate(self.ROLLOUT_PHASES):
            phase_num = phase_idx + 1
            logger.info(
                f"📊 フェーズ {phase_num}/{len(self.ROLLOUT_PHASES)}: "
                f"カナリアトラフィック {traffic_ratio * 100:.0f}%"
            )
 
            await self.prompt_service.set_canary_traffic(
                template_name, new_version, traffic_ratio
            )
 
            # フェーズ観察期間中、1分ごとに品質チェック
            phase_end = datetime.utcnow() + timedelta(
                minutes=self.PHASE_DURATION_MINUTES
            )
            while datetime.utcnow() < phase_end:
                await asyncio.sleep(60)
                health = await self._check_health(
                    template_name, new_version, baseline_version
                )
                if not health["is_healthy"]:
                    logger.error(
                        f"🚨 品質劣化検出 — 自動ロールバック: {health['reason']}"
                    )
                    await self._rollback(template_name, baseline_version)
                    return
 
            if traffic_ratio == 1.00:
                await self.prompt_service.deploy_template(template_name, new_version)
                logger.info(
                    f"✅ ロールアウト完了: {template_name} v{new_version} が本番稼働"
                )
 
    async def _check_health(
        self,
        template_name: str,
        new_version: int,
        baseline_version: int,
    ) -> dict:
        """カナリーバージョンをベースラインと比較してヘルスチェック"""
        canary_stats = await self.metrics_service.get_stats(
            template_name, new_version, window_minutes=15
        )
        baseline_stats = await self.metrics_service.get_stats(
            template_name, baseline_version, window_minutes=15
        )
 
        if canary_stats.get("sample_count", 0) < 10:
            return {"is_healthy": True, "reason": "サンプル不足（監視継続）"}
 
        error_rate = canary_stats.get("error_rate", 0)
        if error_rate > self.ERROR_RATE_THRESHOLD:
            return {
                "is_healthy": False,
                "reason": f"エラー率超過: {error_rate:.1%} > {self.ERROR_RATE_THRESHOLD:.1%}",
            }
 
        baseline_q = baseline_stats.get("avg_quality_score", 0)
        canary_q = canary_stats.get("avg_quality_score", 0)
        if baseline_q > 0:
            drop = (baseline_q - canary_q) / baseline_q
            if drop > self.QUALITY_DROP_THRESHOLD:
                return {
                    "is_healthy": False,
                    "reason": (
                        f"品質スコア低下: {drop:.1%} "
                        f"(ベースライン: {baseline_q:.3f}, カナリア: {canary_q:.3f})"
                    ),
                }
 
        return {"is_healthy": True, "reason": "正常"}
 
    async def _rollback(self, template_name: str, baseline_version: int) -> None:
        await self.prompt_service.deploy_template(template_name, baseline_version)
        await self.redis.publish(
            "canary_alerts",
            json.dumps({
                "event": "rollback",
                "template": template_name,
                "rolled_back_to": baseline_version,
                "timestamp": datetime.utcnow().isoformat(),
            }),
        )

本番アプリへの組み込みパターン

構築したプロンプト管理システムを実際のアプリケーションから利用するには、既存の Gemini API 呼び出しコードをラップするクライアントクラスを作ります。呼び出し側のコードが「どのプロンプトバージョンが使われているか」を意識しなくて済む設計が理想です。

# app/gemini_client.py — 本番アプリ側の統合実装
import asyncio
import os
import time
from typing import Optional
 
import httpx
from google import genai
 
PROMPT_SERVICE_URL = os.environ["PROMPT_SERVICE_URL"]
gemini_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
class ManagedGeminiClient:
    """
    プロンプト管理サービスと連携した Gemini API クライアント。
    - A/Bテスト・カナリアの振り分けは自動で行われる
    - メトリクスは非同期で報告（応答速度に影響させない）
    - プロンプトサービス障害時はフォールバックで継続する
    """
 
    # プロンプトサービス障害時のフォールバックキャッシュ
    _local_cache: dict = {}
 
    def __init__(self, session_id: Optional[str] = None):
        self.session_id = session_id
 
    async def generate(
        self,
        template_name: str,
        variables: dict,
        model: str = "gemini-2.5-pro",
    ) -> str:
        start_ms = time.time() * 1000
 
        # プロンプトサービスからテンプレートを取得
        try:
            async with httpx.AsyncClient(
                base_url=PROMPT_SERVICE_URL, timeout=2.0
            ) as http:
                resp = await http.get(
                    f"/templates/{template_name}/resolve",
                    params={"session_id": self.session_id},
                )
                resp.raise_for_status()
                template_data = resp.json()
 
            # ローカルキャッシュを更新（フォールバック用）
            self._local_cache[template_name] = template_data
 
        except (httpx.HTTPError, httpx.TimeoutException):
            # プロンプトサービスが落ちてもフォールバックで継続
            if template_name in self._local_cache:
                template_data = self._local_cache[template_name]
            else:
                raise RuntimeError(
                    f"Prompt service unavailable and no fallback for '{template_name}'"
                )
 
        # 変数を展開
        prompt = template_data["content"]
        for key, value in variables.items():
            prompt = prompt.replace(f"{{{key}}}", str(value))
 
        # Gemini API 呼び出し
        error = None
        try:
            response = await gemini_client.aio.models.generate_content(
                model=model, contents=prompt
            )
            content = response.text
            input_tokens = response.usage_metadata.prompt_token_count
            output_tokens = response.usage_metadata.candidates_token_count
        except Exception as e:
            content = ""
            input_tokens = 0
            output_tokens = 0
            error = str(e)
 
        latency_ms = int(time.time() * 1000 - start_ms)
 
        # メトリクスを非同期で報告（本番の応答速度に影響させない）
        asyncio.create_task(
            self._report_metrics(
                template_name=template_name,
                version=template_data["version"],
                latency_ms=latency_ms,
                input_tokens=input_tokens,
                output_tokens=output_tokens,
                error=error,
            )
        )
 
        if error:
            raise RuntimeError(f"Gemini API error: {error}")
 
        return content
 
    async def _report_metrics(self, **kwargs):
        try:
            async with httpx.AsyncClient(base_url=PROMPT_SERVICE_URL) as client:
                await client.post(
                    "/metrics",
                    json={"session_id": self.session_id, **kwargs},
                    timeout=1.0,
                )
        except Exception:
            pass  # メトリクス報告失敗は無視
 
# 使用例（既存コードからの切り替えは最小限の変更で済む）
async def handle_customer_query(user_query: str, user_id: str) -> str:
    client = ManagedGeminiClient(session_id=user_id)
    return await client.generate(
        template_name="customer_support_reply",
        variables={"user_query": user_query, "language": "ja"},
    )

_local_cache によるフォールバックは、インメモリに最後に成功したテンプレートを保持します。マルチプロセス環境では Redis にフォールバックキャッシュを置く方が確実ですが、まずはシンプルなインメモリキャッシュから始めて問題があれば改善するのが現実的です。

よくある落とし穴と解決策

実際にプロンプト管理システムを本番に入れると、必ずぶつかる問題があります。事前に知っておくことで、無駄なトラブルシューティング時間を節約できます。

落とし穴① キャッシュの一貫性問題

Redis の TTL を長く（例: 60秒）設定すると、プロンプトをロールバックしても古いバージョンが最大60秒間使われ続けます。緊急時にロールバックしたのに効いていないように見えて、原因究明に混乱が生じます。解決策は「デプロイ時に明示的にキャッシュを削除する」ことです。deploy エンドポイントの末尾で redis_client.delete(f"prompt:{name}:deployment") を確実に呼び出しているのはこのためです。並行してTTLを5〜10秒に短くしておけば、削除漏れがあっても数秒以内に自動で更新されます。

落とし穴② A/Bテストのサンプル汚染

同一ユーザーが A/Bテスト期間中に A と B 両方を受け取ると、メトリクスが汚染されます。ユーザー体験の一貫性も損なわれます。session_id ベースのハッシュ振り分けを使い、同一セッションは常に同じバージョンを受け取るよう実装しているのはこのためです。hashlib.md5(session_id.encode()) によるハッシュ値は決定論的なので、セッション中に何度 resolve を呼んでも同じバージョンが返ります。

落とし穴③ 早まった判定によるロールバック誤発動

カナリアデプロイ開始直後は、カナリアバージョンへの流入が少ないためにエラー率やスコアのばらつきが大きくなります。サンプル数が10件しかない時点で「エラー率20%！ロールバック！」と判定してしまうのは、偶然のばらつきを品質劣化と誤解した典型例です。_check_health の中で sample_count < 10 の場合は「監視継続」と返しているのがその対策です。実際の運用では min_samples = 100 程度を推奨します。

落とし穴④ プロンプトサービス自体の障害伝播

プロンプト管理サービスが落ちたら、本番のすべての Gemini API 呼び出しが止まります。ManagedGeminiClient のフォールバックキャッシュはこれを防ぎますが、「サービスが落ちている間は最後に取得したバージョンで動き続ける」という動作がビジネス上許容できるかを事前に確認しましょう。許容できない場合はプロンプトサービス自体を冗長化（複数インスタンス + ロードバランサー）する必要があります。

落とし穴⑤ 変数展開のセキュリティリスク

prompt.replace(f"{{{key}}}", str(value)) によるシンプルな文字列置換は、ユーザー入力をそのまま受け取る場合にプロンプトインジェクションのリスクがあります。外部ユーザーの入力を変数として使う場合は、入力のサニタイズ（改行・特殊文字のエスケープ）を必ず行ってください。Gemini API の System Instruction とユーザー入力を分離する設計も検討に値します。

チーム運用のベストプラクティス

プロンプト管理システムは技術的な仕組みであると同時に、チームの文化とプロセスに関わる問題でもあります。ツールを整えるだけでは不十分で、運用ルールを定めてチーム全体で守ることが品質向上の本質です。

ルール① プロンプト変更はPull Requestと同じプロセスで承認する

新しいプロンプトバージョンの作成は、コードの変更と同じ重みで扱います。少なくとも1人の別のメンバーがレビューし、stagingで動作確認し、承認後にカナリアデプロイを開始するルールを徹底しましょう。「ちょっとした表現修正だから直接本番に出す」という判断が品質劣化事故の原因になります。

ルール② metadata に変更理由を必ず記録する

metadata フィールドには変更理由（例: "reason": "ハルシネーション対策でシステムプロンプトに制約を追加" ）を必ず記録します。品質スコアが突然低下したとき、過去のバージョン履歴と変更理由を見れば原因究明が大幅に速くなります。

ルール③ quality_score の収集方法を最初に決める

メトリクスの中核である quality_score をどのように測定するかが、システム全体の価値を左右します。選択肢は主に3つあります。①ユーザーのフィードバック（高精度だがサンプルが集まりにくい）、② LLM-as-Judge（別のモデルが回答を1〜5点で自動評価）、③業務ロジックベースの指標（抽出精度・フォーマット適合率など）。コストと精度のバランスから、②と③を組み合わせることが多いです。品質スコアが収集できていない状態でA/Bテストを動かしても、勝者を判断する根拠がありません。

ルール④ ロールバックの手順を事前に全員で確認する

深夜にプロンプト品質が急落したとき、担当者が「どのエンドポイントをどう叩けばロールバックできるか」を迷わず実行できるよう、手順をRunbookに記載しておきます。シンプルには curl -X POST "https://prompt-service/templates/{name}/deploy?version={n}" の1コマンドで完了するはずです。この手順をインシデント発生前に全員で一度実際に試しておくことを強く勧めます。

メトリクスダッシュボードで改善サイクルを可視化する

管理基盤が動き始めたら、次は蓄積されたメトリクスを可視化して改善のサイクルを回します。Grafana と TimescaleDB（PostgreSQL の時系列拡張）を組み合わせることで、プロンプトバージョンごとの品質推移・コスト推移・レイテンシ分布をリアルタイムで確認できます。

TimescaleDB を使う理由は、prompt_metrics テーブルの recorded_at カラムを時系列インデックスで管理できるためです。通常の PostgreSQL と比べて時系列集計クエリが2〜10倍高速になります。既存の PostgreSQL インスタンスに TimescaleDB 拡張を追加するだけで使えるため、インフラを大きく変える必要はありません。

-- TimescaleDB 拡張の有効化と hypertable 変換
CREATE EXTENSION IF NOT EXISTS timescaledb;
 
-- prompt_metrics を hypertable に変換（既存データを保持したまま）
SELECT create_hypertable('prompt_metrics', 'recorded_at', if_not_exists => TRUE);
 
-- Grafana で使う集計クエリ例
-- バージョン別の平均品質スコア（15分ごとの移動平均）
SELECT
    time_bucket('15 minutes', recorded_at) AS time,
    template_name,
    version,
    AVG(quality_score)   AS avg_quality,
    AVG(latency_ms)      AS avg_latency_ms,
    COUNT(*)             AS request_count,
    SUM(CASE WHEN error IS NOT NULL THEN 1 ELSE 0 END)::float
        / NULLIF(COUNT(*), 0)         AS error_rate,
    AVG(input_tokens + output_tokens) AS avg_tokens
FROM prompt_metrics
WHERE
    recorded_at > NOW() - INTERVAL '24 hours'
    AND template_name = $1
GROUP BY 1, 2, 3
ORDER BY 1 DESC;

Grafana のダッシュボードでは、以下の4つのパネルを揃えることを推奨します。

1つ目は「バージョン別品質スコア時系列グラフ」です。縦軸を品質スコア（0〜1）、横軸を時間にとり、バージョンごとに異なる色の折れ線で表示します。A/Bテスト開始・カナリアデプロイ開始・ロールバックのタイミングをアノテーションで重ねることで、どのイベントがスコアに影響したかが一目で分かります。

2つ目は「コスト推移（トークン数）」です。avg_tokens × モデル単価 でリクエストあたりのコストを概算し、バージョン変更前後でコストが増減していないかを確認します。品質を維持しつつコストを削減したプロンプト改善は、チームにとって最も価値の高い成果です。

3つ目は「レイテンシ分布（P50/P95/P99）」です。平均レイテンシだけでは外れ値が見えません。P95（上位5%の遅いリクエスト）や P99 を監視することで、一部のユーザーが体験する極端な遅延を早期に発見できます。新しいプロンプトがより長い回答を生成させる設計になっている場合、P99 レイテンシが急増することがあります。

4つ目は「エラー率アラート」です。エラー率が閾値（通常5%）を超えた場合に Slack や PagerDuty に通知するアラートを設定します。カナリアデプロイ中のこのアラートが自動ロールバックのトリガーとも連動します。

ダッシュボードを整えると、チームの議論が「このプロンプト、感覚的にはいいと思うんだけど」から「品質スコアが有意に向上していて、コストも3%削減できています。デプロイしよう」という形に変わります。この変化がプロンプト管理基盤を構築する最大の価値です。

LLMを本番システムとして運用する際の評価・監視・デプロイ戦略が体系的にまとめられており、この記事で扱った内容の背景理論を補完する内容です。

全体を振り返って — 今日から取り組める最初の一歩

プロンプトをコードから分離し、バージョン管理・A/Bテスト・カナリアデプロイで管理する基盤を構築しました。フル実装は相応の工数が必要ですが、最初の一歩として「プロンプトをデータベースに格納し、deploy エンドポイントでアクティブバージョンを管理する」だけでも大きな効果があります。ロールバックが30秒でできるだけで、インシデント時の復旧速度が劇的に向上します。

A/Bテストとカナリアデプロイは、プロンプトの品質改善サイクルを「感覚」から「データ」に変えます。「新しいプロンプトの方が良い気がする」から「統計的有意差 p = 0.031 で新バージョンの品質スコアが 4.3% 高い」という議論ができるようになります。

次のステップとして、Gemini API のコスト最適化完全ガイドも参考にしながら、プロンプトのトークン効率を定量的に改善してみてください。コスト・品質・レイテンシのバランスを数値で追えるようになることが、Gemini API を本番で使い続けるための基盤です。