本番で 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を短くしておくことで、プロンプトの変更・ロールバックが数秒以内に全リクエストに反映されます。
プロンプトテンプレートストアの実装
データベーススキーマの詳細設計
スキーマ設計の核心は「バージョンとデプロイメントの分離」です。バージョンは不変のレコード（一度作ったら変更しない）であり、デプロイメントは「現在どのバージョンがアクティブか」を指す可変のポインタです。この分離により、ロールバックが「デプロイメントのポインタを古いバージョンに向け直す」だけという単純な操作になります。
-- プロンプトテンプレートのバージョン管理テーブル（不変レコード） 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 を本番で使い続けるための基盤です。