プロンプトは「資産」です — Antigravityで作る本番級プロンプト管理基盤:バージョニング・A/Bテスト・品質評価の実装パターン

# prompts/summarize_article.yaml
# このファイルがプロンプトの「ソース・オブ・トゥルース」です
id: summarize_article
versions:
  - version: v1
    status: stable            # stable / canary / deprecated
    weight: 0.8               # ルーター層での選択確率
    created_at: 2026-03-15
    author: masaki
    template: |
      あなたはプロの編集者です。以下の記事を3文で要約してください。
      記事: {article}
      要約:
    model: gemini-2.5-pro
    max_tokens: 256
    temperature: 0.3
 
  - version: v2
    status: canary
    weight: 0.2
    created_at: 2026-04-20
    author: masaki
    description: "『プロの編集者』を『熟練のライター』に変更し、箇条書き出力を禁止するガードを追加"
    template: |
      あなたは熟練のライターです。以下の記事を3文で要約してください。
      出力は必ず通常の文章で行い、箇条書きや見出しは使わないでください。
      記事: {article}
      要約:
    model: gemini-2.5-pro
    max_tokens: 256
    temperature: 0.3

# src/prompt_store.py
# プロンプトストアの読み込みと検証を担当するモジュール
from __future__ import annotations
from pathlib import Path
from typing import Literal
import yaml
from pydantic import BaseModel, Field, field_validator
 
class PromptVersion(BaseModel):
    """単一プロンプトバージョンの定義"""
    version: str
    status: Literal["stable", "canary", "deprecated"]
    weight: float = Field(ge=0.0, le=1.0)  # 0〜1 の確率
    created_at: str
    author: str
    template: str
    model: str
    max_tokens: int = Field(gt=0)
    temperature: float = Field(ge=0.0, le=2.0)
    description: str | None = None
 
class PromptDefinition(BaseModel):
    """プロンプト(複数バージョンを持つ論理単位)"""
    id: str
    versions: list[PromptVersion]
 
    @field_validator("versions")
    @classmethod
    def weights_must_sum_to_one(cls, versions: list[PromptVersion]) -> list[PromptVersion]:
        # deprecated を除いた合計が 1.0 でないとルーターが破綻する
        active = [v for v in versions if v.status \!= "deprecated"]
        total = sum(v.weight for v in active)
        if abs(total - 1.0) > 1e-6:
            raise ValueError(
                f"active weight sum must be 1.0, got {total:.4f}. "
                f"Check prompt id={versions[0].version if versions else 'unknown'}"
            )
        return versions
 
class PromptStore:
    """YAMLディレクトリ全体をインメモリにロードするストア"""
 
    def __init__(self, root: Path) -> None:
        self.root = root
        self._prompts: dict[str, PromptDefinition] = {}
        self._load()
 
    def _load(self) -> None:
        for yaml_file in self.root.glob("*.yaml"):
            try:
                data = yaml.safe_load(yaml_file.read_text(encoding="utf-8"))
                prompt = PromptDefinition.model_validate(data)
                self._prompts[prompt.id] = prompt
            except Exception as e:
                # 1つの YAML が壊れていても全体を止めず、警告を出す
                # 本番では監視に通知するフックを入れる
                print(f"[prompt_store] failed to load {yaml_file.name}: {e}")
 
    def get(self, prompt_id: str) -> PromptDefinition:
        if prompt_id not in self._prompts:
            raise KeyError(f"prompt not found: {prompt_id}")
        return self._prompts[prompt_id]
 
    def list_ids(self) -> list[str]:
        return sorted(self._prompts.keys())
 
# 動作確認
if __name__ == "__main__":
    store = PromptStore(Path("prompts"))
    print("loaded:", store.list_ids())
    p = store.get("summarize_article")
    print(f"versions: {[v.version for v in p.versions]}")
    # 期待する出力:
    # loaded: ['summarize_article']
    # versions: ['v1', 'v2']

# src/prompt_router.py
# リクエストごとにプロンプトバージョンを選択するルーター
from __future__ import annotations
import hashlib
import random
from dataclasses import dataclass
from src.prompt_store import PromptStore, PromptVersion
 
@dataclass
class RoutingResult:
    version: PromptVersion
    routing_key: str  # ログに記録して追跡する
 
class PromptRouter:
    def __init__(self, store: PromptStore) -> None:
        self.store = store
 
    def select(self, prompt_id: str, user_id: str | None = None) -> RoutingResult:
        """
        user_id が与えられた場合は「スティッキー」ルーティング。
        同じユーザーには常に同じバージョンを返す(学習データの一貫性確保)。
        """
        prompt = self.store.get(prompt_id)
        active = [v for v in prompt.versions if v.status \!= "deprecated"]
        if not active:
            raise RuntimeError(f"no active versions for {prompt_id}")
 
        # スティッキールーティング: user_id のハッシュで 0.0〜1.0 を生成
        if user_id:
            h = hashlib.sha256(f"{prompt_id}:{user_id}".encode()).hexdigest()
            r = int(h[:8], 16) / 0xFFFFFFFF
        else:
            # user_id がない場合(バッチ処理等)はランダム
            r = random.random()
 
        cumulative = 0.0
        for v in active:
            cumulative += v.weight
            if r < cumulative:
                return RoutingResult(version=v, routing_key=f"{prompt_id}:{v.version}")
 
        # 浮動小数点の丸め誤差で最後まで到達する場合のフォールバック
        return RoutingResult(version=active[-1], routing_key=f"{prompt_id}:{active[-1].version}")
 
# 動作確認: 10,000 ユーザーで分布を検証
if __name__ == "__main__":
    from pathlib import Path
    from collections import Counter
    store = PromptStore(Path("prompts"))
    router = PromptRouter(store)
    counts = Counter()
    for i in range(10000):
        result = router.select("summarize_article", user_id=f"user_{i}")
        counts[result.version.version] += 1
    print(counts)
    # 期待する出力(おおむね):
    # Counter({'v1': 8000, 'v2': 2000})  (誤差±1%以内)

# src/evaluator.py
# 応答品質を LLM で 0〜10 点に評価するモジュール
from __future__ import annotations
import json
import os
from dataclasses import dataclass
from typing import Any
import google.generativeai as genai
 
genai.configure(api_key=os.environ["GOOGLE_API_KEY"])  # 環境変数必須
 
JUDGE_PROMPT = """\
あなたはAI応答の品質を評価する厳格な審査員です。
以下の「入力」に対して、「AIの応答」を0〜10点で評価してください。
 
評価基準:
- 関連性 (query との一致度): 0-3点
- 事実性 (捏造のなさ): 0-4点
- 明瞭性 (読みやすさ): 0-3点
 
合計点と、各基準のスコア、簡単な理由を JSON で返してください。
他の文字は一切出力せず、JSONオブジェクトのみを返してください。
 
入力: {query}
AIの応答: {response}
 
出力形式:
{{"total": 8, "relevance": 3, "factuality": 3, "clarity": 2, "reason": "..."}}
"""
 
@dataclass
class EvalResult:
    total: float
    relevance: float
    factuality: float
    clarity: float
    reason: str
 
def evaluate_response(query: str, response: str, judge_model: str = "gemini-2.5-pro") -> EvalResult:
    """LLM-as-a-Judge で応答品質を評価する"""
    model = genai.GenerativeModel(judge_model)
    prompt = JUDGE_PROMPT.format(query=query, response=response)
    try:
        result = model.generate_content(prompt, generation_config={"temperature": 0.0})
        # マークダウンの ``` が混入することがあるのでストリップ
        text = result.text.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()
        data: dict[str, Any] = json.loads(text)
        return EvalResult(
            total=float(data["total"]),
            relevance=float(data["relevance"]),
            factuality=float(data["factuality"]),
            clarity=float(data["clarity"]),
            reason=data.get("reason", ""),
        )
    except json.JSONDecodeError as e:
        # LLMが指示に反してJSON以外を返すことがある。再試行ではなくフォールバックで続行
        return EvalResult(total=0.0, relevance=0.0, factuality=0.0, clarity=0.0,
                          reason=f"judge output parse error: {e}")
    except Exception as e:
        # API エラー等はスコアなしで返し、呼び出し側で分岐できるようにする
        return EvalResult(total=-1.0, relevance=-1.0, factuality=-1.0, clarity=-1.0,
                          reason=f"judge error: {type(e).__name__}: {e}")
 
# 動作確認
if __name__ == "__main__":
    r = evaluate_response(
        query="今日の東京の天気は?",
        response="今日の東京は晴れ時々曇りで、最高気温は22度です。"
    )
    print(f"total={r.total}, reason={r.reason}")
    # 期待する出力例:
    # total=8.0, reason="関連性は高いが、天気情報の出典が不明で事実性を減点"

# src/executor.py
# 選ばれたプロンプトを実行し、メトリクスを記録するレイヤー
from __future__ import annotations
import time
import json
from dataclasses import dataclass, asdict
from pathlib import Path
import google.generativeai as genai
from src.prompt_router import RoutingResult
 
@dataclass
class ExecutionRecord:
    """1リクエストあたりの観測レコード"""
    prompt_id: str
    version: str
    user_id: str | None
    query: str
    response: str
    input_tokens: int
    output_tokens: int
    latency_ms: float
    error: str | None = None
 
def _estimate_cost_jpy(input_tokens: int, output_tokens: int, model: str) -> float:
    """モデルごとの1Mトークン単価を掛けて日本円に換算(概算)"""
    # 本番では単価テーブルをYAMLに外出しするのがおすすめ
    table = {
        "gemini-2.5-pro": (0.375, 1.125),  # (input/1M, output/1M) USD
        "gemini-2.5-flash": (0.075, 0.225),
    }
    ipm, opm = table.get(model, (0.3, 1.0))
    usd = (input_tokens / 1_000_000) * ipm + (output_tokens / 1_000_000) * opm
    return round(usd * 155, 4)  # 概算USD/JPY = 155
 
class PromptExecutor:
    def __init__(self, log_path: Path = Path("logs/prompt_runs.jsonl")) -> None:
        self.log_path = log_path
        self.log_path.parent.mkdir(parents=True, exist_ok=True)
 
    def run(self, routing: RoutingResult, variables: dict[str, str],
            user_id: str | None = None) -> ExecutionRecord:
        v = routing.version
        # テンプレート変数の欠落チェック(落とし穴2の対策)
        try:
            prompt_text = v.template.format(**variables)
        except KeyError as e:
            raise ValueError(
                f"missing template variable {e} for prompt {routing.routing_key}"
            ) from e
 
        model = genai.GenerativeModel(v.model)
        start = time.perf_counter()
        error: str | None = None
        response_text = ""
        input_tokens = output_tokens = 0
 
        try:
            response = model.generate_content(
                prompt_text,
                generation_config={
                    "temperature": v.temperature,
                    "max_output_tokens": v.max_tokens,
                },
            )
            response_text = response.text
            # google-generativeai の usage_metadata から取得
            um = getattr(response, "usage_metadata", None)
            if um:
                input_tokens = getattr(um, "prompt_token_count", 0)
                output_tokens = getattr(um, "candidates_token_count", 0)
        except Exception as e:
            error = f"{type(e).__name__}: {e}"
 
        latency_ms = (time.perf_counter() - start) * 1000.0
        record = ExecutionRecord(
            prompt_id=routing.routing_key.split(":")[0],
            version=v.version,
            user_id=user_id,
            query=variables.get("article", "")[:100],  # 先頭100文字のみ記録
            response=response_text,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            latency_ms=latency_ms,
            error=error,
        )
        self._persist(record)
        return record
 
    def _persist(self, rec: ExecutionRecord) -> None:
        # JSONL形式でファイルに追記。本番ではBigQuery/Postgresに送る
        with self.log_path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(asdict(rec), ensure_ascii=False) + "\n")
 
# 動作確認(API key 必須)
if __name__ == "__main__":
    import os
    from pathlib import Path
    from src.prompt_store import PromptStore
    from src.prompt_router import PromptRouter
 
    genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
    store = PromptStore(Path("prompts"))
    router = PromptRouter(store)
    executor = PromptExecutor()
 
    routing = router.select("summarize_article", user_id="user_42")
    record = executor.run(routing, {"article": "昨日、東京で桜が満開を迎えました。"}, user_id="user_42")
    cost = _estimate_cost_jpy(record.input_tokens, record.output_tokens, "gemini-2.5-pro")
    print(f"version={record.version}, latency={record.latency_ms:.0f}ms, cost=¥{cost}")
    # 期待する出力例:
    # version=v1, latency=820ms, cost=¥0.12

# .github/workflows/prompt-eval.yml
name: Prompt Evaluation
on:
  pull_request:
    paths:
      - "prompts/**.yaml"
 
jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
 
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
 
      - name: Install deps
        run: pip install pydantic pyyaml google-generativeai
 
      - name: Run comparative evaluation
        env:
          GOOGLE_API_KEY: ${{ secrets.GOOGLE_API_KEY }}
        run: |
          python scripts/compare_versions.py \
            --prompt-id summarize_article \
            --baseline-ref origin/main \
            --candidate-ref HEAD \
            --sample-size 50 \
            --output eval_report.md
 
      - name: Comment on PR
        uses: marocchino/sticky-pull-request-comment@v2
        with:
          path: eval_report.md
 
      - name: Fail if quality regressed
        run: |
          SCORE_DIFF=$(jq '.diff' eval_report.json)
          if (( $(echo "$SCORE_DIFF < -0.3" | bc -l) )); then
            echo "Quality regressed by $SCORE_DIFF — blocking merge"
            exit 1
          fi

-- Query 1: バージョンごとの直近24時間の平均品質スコアとレイテンシ
SELECT
  prompt_id,
  version,
  COUNT(*) AS requests,
  AVG(eval_total) AS avg_score,
  AVG(latency_ms) AS avg_latency_ms,
  SUM(input_tokens + output_tokens) AS total_tokens
FROM prompt_runs
WHERE ts >= CURRENT_TIMESTAMP() - INTERVAL 24 HOUR
  AND error IS NULL
GROUP BY prompt_id, version
ORDER BY prompt_id, avg_score DESC;

-- Query 2: A/Bテストの統計的有意性チェック(Welch's t-test の簡易版)
-- canary版が本当にstable版より良いかを数値で判定する
WITH stats AS (
  SELECT
    version,
    AVG(eval_total) AS mean,
    STDDEV(eval_total) AS stddev,
    COUNT(*) AS n
  FROM prompt_runs
  WHERE prompt_id = 'summarize_article'
    AND ts >= CURRENT_TIMESTAMP() - INTERVAL 7 DAY
    AND error IS NULL
  GROUP BY version
)
SELECT
  a.version AS version_a, a.mean AS mean_a,
  b.version AS version_b, b.mean AS mean_b,
  (a.mean - b.mean) / SQRT((a.stddev*a.stddev)/a.n + (b.stddev*b.stddev)/b.n) AS t_stat
FROM stats a JOIN stats b ON a.version < b.version;
-- t_stat の絶対値が約2を超えたら95%水準で有意差ありの目安

-- Query 3: エラー発生の時系列 — 新バージョンデプロイ後に失敗率が跳ねていないか
SELECT
  DATE_TRUNC(ts, HOUR) AS hour,
  version,
  COUNT(*) AS total,
  SUM(CASE WHEN error IS NOT NULL THEN 1 ELSE 0 END) AS errors,
  ROUND(SUM(CASE WHEN error IS NOT NULL THEN 1.0 ELSE 0.0 END) * 100 / COUNT(*), 2) AS error_rate_pct
FROM prompt_runs
WHERE ts >= CURRENT_TIMESTAMP() - INTERVAL 48 HOUR
GROUP BY hour, version
ORDER BY hour DESC, version;