Gemini API プロンプト評価・最適化パイプライン構築ガイド — LLM-as-Judge で品質を自動計測する

取り組みの背景 — なぜプロンプト評価が必要なのか

Gemini API を使ったアプリケーションを本番環境で運用していると、ある日突然出力品質が変わったように感じたり、プロンプトの修正が本当に改善につながっているのか確信が持てなかったりすることがあります。

「プロンプトを変えたら良くなった気がする」という直感的な判断だけでは、大規模な本番システムの品質を維持できません。ソフトウェア開発においてテストが不可欠であるように、LLM アプリケーションにもプロンプトの品質を定量的に評価する仕組みが必要です。

ここではGemini API を活用して以下を実現するパイプラインを構築します。

LLM-as-Judge パターン: Gemini 自身を評価者として使い、出力品質を自動スコアリング
A/B テスト基盤: 複数のプロンプトバリアントを統計的に比較
コスト・品質の最適化: トークン使用量と品質スコアのトレードオフを可視化
継続的モニタリング: 品質劣化を自動検知するアラートシステム

対象読者は、Gemini API の基本的な使い方を理解しており、本番環境での品質管理に課題を感じている開発者です。

前提知識・環境準備

必要なもの

Python 3.11 以上
Google AI Studio の API キー（こちらから取得）
基本的なライブラリ: google-genai, pandas, scipy

インストール

pip install google-genai pandas scipy numpy

API クライアントの初期化

import os
from google import genai
 
# APIクライアントの初期化
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# 評価に使用するモデル
EVAL_MODEL = "gemini-2.5-pro"       # 評価者（Judge）用
TARGET_MODEL = "gemini-2.5-flash"   # 評価対象のモデル

ポイントとして、評価者（Judge）には対象モデルと同等以上の能力を持つモデルを使用します。Gemini 2.5 Pro は推論能力が高く、評価タスクに適しています。

LLM-as-Judge パターンの実装

LLM-as-Judge とは、LLM の出力品質を別の LLM（あるいは同じモデル）に評価させるアプローチです。人間の評価と高い相関が報告されており、大量のテストケースを高速に処理できるメリットがあります。

評価基準の定義

まず、評価観点を明確に定義します。曖昧な基準では評価のブレが大きくなるため、具体的なルーブリック（採点基準表）を設計する点が肝心です。

EVALUATION_RUBRIC = """
あなたはAIアシスタントの出力品質を評価する専門家です。
以下の4つの観点で、1〜5のスコアを付けてください。
 
## 評価観点
 
### 1. 正確性 (Accuracy)
- 5: 事実として完全に正確。誤情報なし
- 4: ほぼ正確。軽微な不正確さが1箇所
- 3: 概ね正確だが、やや不正確な記述が複数
- 2: 重要な誤りが含まれる
- 1: 大部分が不正確
 
### 2. 関連性 (Relevance)
- 5: ユーザーの質問に完璧に対応
- 4: ほぼ対応しているが、一部余分または不足
- 3: 部分的に対応
- 2: 大部分が質問と無関係
- 1: 完全に的外れ
 
### 3. 明瞭性 (Clarity)
- 5: 非常に読みやすく、構造が明確
- 4: 読みやすいが、構成に改善の余地あり
- 3: 理解可能だが、分かりにくい部分あり
- 2: 理解しにくい
- 1: ほぼ理解不能
 
### 4. 完全性 (Completeness)
- 5: 必要な情報を漏れなくカバー
- 4: ほぼ完全だが、軽微な不足あり
- 3: 主要な点はカバーしているが、重要な不足あり
- 2: 大部分が不足
- 1: ほとんど情報がない
 
## 出力形式
必ず以下のJSON形式で回答してください:
{
  "accuracy": <1-5>,
  "relevance": <1-5>,
  "clarity": <1-5>,
  "completeness": <1-5>,
  "overall": <1-5の加重平均>,
  "reasoning": "<評価の根拠を2-3文で>"
}
"""

自動評価関数の実装

import json
from google.genai import types
 
def evaluate_response(
    user_query: str,
    ai_response: str,
    reference_answer: str | None = None,
) -> dict:
    """
    Gemini をJudgeとして使い、AIレスポンスの品質を自動評価する。
 
    Args:
        user_query: ユーザーの元の質問
        ai_response: 評価対象のAI出力
        reference_answer: 参照回答（ある場合）
 
    Returns:
        各評価観点のスコアと根拠を含む辞書
    """
    judge_prompt = f"""{EVALUATION_RUBRIC}
 
## 評価対象
 
**ユーザーの質問:**
{user_query}
 
**AIの回答:**
{ai_response}
"""
    if reference_answer:
        judge_prompt += f"""
**参照回答（理想的な回答）:**
{reference_answer}
"""
 
    response = client.models.generate_content(
        model=EVAL_MODEL,
        contents=judge_prompt,
        config=types.GenerateContentConfig(
            temperature=0.0,  # 評価の一貫性を最大化
            response_mime_type="application/json",
        ),
    )
 
    try:
        scores = json.loads(response.text)
        # トークン使用量も記録
        scores["usage"] = {
            "prompt_tokens": response.usage_metadata.prompt_token_count,
            "output_tokens": response.usage_metadata.candidates_token_count,
        }
        return scores
    except json.JSONDecodeError:
        return {"error": "JSON解析に失敗", "raw": response.text}
 
# 使用例
result = evaluate_response(
    user_query="Pythonでリスト内包表記はどう書きますか？",
    ai_response="リスト内包表記は [式 for 変数 in イテラブル] の形式で書きます。"
)
print(json.dumps(result, indent=2, ensure_ascii=False))
# 期待する出力:
# {
#   "accuracy": 5,
#   "relevance": 5,
#   "clarity": 4,
#   "completeness": 3,
#   "overall": 4.25,
#   "reasoning": "基本構文は正確だが、条件付き内包表記やネストの例が不足している。"
# }

評価の信頼性を高めるテクニック

LLM-as-Judge は強力ですが、いくつかのバイアスに注意が必要です。

import random
 
def evaluate_with_debiasing(
    user_query: str,
    response_a: str,
    response_b: str,
    num_rounds: int = 3,
) -> dict:
    """
    位置バイアスを軽減するため、回答の提示順を入れ替えて
    複数回評価し、平均スコアを算出する。
    """
    scores_a = []
    scores_b = []
 
    for i in range(num_rounds):
        # ランダムに順序を入れ替え
        if random.random() < 0.5:
            first, second = response_a, response_b
            order = "AB"
        else:
            first, second = response_b, response_a
            order = "BA"
 
        comparison_prompt = f"""
2つのAI回答を比較評価してください。
 
**ユーザーの質問:** {user_query}
 
**回答1:**
{first}
 
**回答2:**
{second}
 
各回答に1〜5のスコアを付け、JSON形式で返してください:
{{"score_1": <int>, "score_2": <int>, "reasoning": "<str>"}}
"""
        response = client.models.generate_content(
            model=EVAL_MODEL,
            contents=comparison_prompt,
            config=types.GenerateContentConfig(
                temperature=0.0,
                response_mime_type="application/json",
            ),
        )
 
        result = json.loads(response.text)
        if order == "AB":
            scores_a.append(result["score_1"])
            scores_b.append(result["score_2"])
        else:
            scores_a.append(result["score_2"])
            scores_b.append(result["score_1"])
 
    return {
        "response_a_avg": sum(scores_a) / len(scores_a),
        "response_b_avg": sum(scores_b) / len(scores_b),
        "response_a_scores": scores_a,
        "response_b_scores": scores_b,
    }

プロンプト A/B テスト基盤の構築

プロンプトの改善が統計的に有意であるかを検証するために、A/B テストの仕組みを構築します。

テストケースとバリアントの管理

from dataclasses import dataclass, field
from datetime import datetime
 
@dataclass
class PromptVariant:
    """プロンプトのバリアント（テスト対象）"""
    name: str
    system_instruction: str
    template: str
    model: str = "gemini-2.5-flash"
    temperature: float = 0.7
    max_tokens: int = 2048
 
@dataclass
class TestCase:
    """評価用のテストケース"""
    query: str
    reference_answer: str | None = None
    category: str = "general"
    metadata: dict = field(default_factory=dict)
 
@dataclass
class EvalResult:
    """評価結果"""
    variant_name: str
    test_case_query: str
    scores: dict
    latency_ms: float
    token_count: int
    cost_usd: float
    timestamp: str = field(
        default_factory=lambda: datetime.now().isoformat()
    )

A/B テストランナー

import time
import pandas as pd
from scipy import stats
 
# Gemini APIの価格（2026年3月時点の概算）
PRICING = {
    "gemini-2.5-pro": {"input": 1.25 / 1_000_000, "output": 10.0 / 1_000_000},
    "gemini-2.5-flash": {"input": 0.15 / 1_000_000, "output": 0.60 / 1_000_000},
}
 
def estimate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    """トークン使用量からコストを推定"""
    pricing = PRICING.get(model, PRICING["gemini-2.5-flash"])
    return (input_tokens * pricing["input"]) + (output_tokens * pricing["output"])
 
def run_ab_test(
    variants: list[PromptVariant],
    test_cases: list[TestCase],
) -> pd.DataFrame:
    """
    複数のプロンプトバリアントを全テストケースに対して実行し、
    評価結果をDataFrameとして返す。
    """
    results = []
 
    for variant in variants:
        print(f"\n🔄 Testing variant: {variant.name}")
 
        for i, tc in enumerate(test_cases):
            prompt = variant.template.format(query=tc.query)
 
            # レイテンシ計測
            start = time.time()
            response = client.models.generate_content(
                model=variant.model,
                contents=prompt,
                config=types.GenerateContentConfig(
                    system_instruction=variant.system_instruction,
                    temperature=variant.temperature,
                    max_output_tokens=variant.max_tokens,
                ),
            )
            latency_ms = (time.time() - start) * 1000
 
            # 品質評価
            scores = evaluate_response(
                user_query=tc.query,
                ai_response=response.text,
                reference_answer=tc.reference_answer,
            )
 
            # コスト計算
            input_tokens = response.usage_metadata.prompt_token_count
            output_tokens = response.usage_metadata.candidates_token_count
            cost = estimate_cost(variant.model, input_tokens, output_tokens)
 
            results.append(EvalResult(
                variant_name=variant.name,
                test_case_query=tc.query,
                scores=scores,
                latency_ms=latency_ms,
                token_count=input_tokens + output_tokens,
                cost_usd=cost,
            ))
 
            print(f"  [{i+1}/{len(test_cases)}] "
                  f"Score: {scores.get('overall', 'N/A')} | "
                  f"Latency: {latency_ms:.0f}ms | "
                  f"Cost: ${cost:.6f}")
 
    # DataFrameに変換
    df = pd.DataFrame([
        {
            "variant": r.variant_name,
            "query": r.test_case_query,
            "overall_score": r.scores.get("overall", 0),
            "accuracy": r.scores.get("accuracy", 0),
            "relevance": r.scores.get("relevance", 0),
            "clarity": r.scores.get("clarity", 0),
            "completeness": r.scores.get("completeness", 0),
            "latency_ms": r.latency_ms,
            "tokens": r.token_count,
            "cost_usd": r.cost_usd,
        }
        for r in results
    ])
 
    return df

統計的検定と結果分析

def analyze_ab_results(df: pd.DataFrame) -> dict:
    """
    A/Bテスト結果を統計的に分析し、有意差を判定する。
    Welchのt検定を使用（等分散を仮定しない）。
    """
    variants = df["variant"].unique()
    if len(variants) < 2:
        return {"error": "比較には2つ以上のバリアントが必要です"}
 
    analysis = {}
    baseline = variants[0]
    baseline_scores = df[df["variant"] == baseline]["overall_score"]
 
    for variant in variants[1:]:
        variant_scores = df[df["variant"] == variant]["overall_score"]
 
        # Welchのt検定
        t_stat, p_value = stats.ttest_ind(
            variant_scores, baseline_scores, equal_var=False
        )
 
        # 効果量（Cohen's d）
        pooled_std = (
            (baseline_scores.std() ** 2 + variant_scores.std() ** 2) / 2
        ) ** 0.5
        cohens_d = (
            (variant_scores.mean() - baseline_scores.mean()) / pooled_std
            if pooled_std > 0 else 0
        )
 
        # コスト比較
        baseline_cost = df[df["variant"] == baseline]["cost_usd"].sum()
        variant_cost = df[df["variant"] == variant]["cost_usd"].sum()
 
        analysis[f"{baseline}_vs_{variant}"] = {
            "baseline_mean": round(baseline_scores.mean(), 3),
            "variant_mean": round(variant_scores.mean(), 3),
            "improvement": round(variant_scores.mean() - baseline_scores.mean(), 3),
            "p_value": round(p_value, 4),
            "significant": p_value < 0.05,
            "cohens_d": round(cohens_d, 3),
            "effect_size": (
                "large" if abs(cohens_d) > 0.8
                else "medium" if abs(cohens_d) > 0.5
                else "small"
            ),
            "cost_ratio": round(variant_cost / baseline_cost, 2) if baseline_cost > 0 else None,
            "recommendation": _get_recommendation(
                p_value, cohens_d, variant_cost / baseline_cost if baseline_cost > 0 else 1
            ),
        }
 
    return analysis
 
def _get_recommendation(p_value: float, cohens_d: float, cost_ratio: float) -> str:
    """分析結果に基づいた推奨アクションを生成"""
    if p_value >= 0.05:
        return "有意差なし。サンプルサイズを増やすか、プロンプトの変更を大きくしてください。"
    if cohens_d > 0.5 and cost_ratio <= 1.2:
        return "✅ 採用推奨。品質が大幅に改善し、コスト増加も許容範囲内です。"
    if cohens_d > 0.2 and cost_ratio <= 1.0:
        return "✅ 採用推奨。品質改善があり、コスト増加もありません。"
    if cohens_d > 0.5 and cost_ratio > 1.5:
        return "⚠️ 品質は改善しますが、コストが大幅に増加します。ROIを慎重に検討してください。"
    return "△ 軽微な改善のみ。他のアプローチを検討してください。"

実行例

# バリアントの定義
variant_a = PromptVariant(
    name="baseline",
    system_instruction="あなたは親切なAIアシスタントです。",
    template="以下の質問に回答してください。\n\n{query}",
)
 
variant_b = PromptVariant(
    name="structured",
    system_instruction=(
        "あなたは技術的な質問に正確に回答する専門家です。"
        "回答はまず結論を述べ、次に詳細な説明を加えてください。"
    ),
    template=(
        "以下の技術的な質問に回答してください。\n"
        "ステップバイステップで考え、コード例があれば含めてください。\n\n"
        "質問: {query}"
    ),
)
 
# テストケースの定義
test_cases = [
    TestCase(
        query="Pythonで非同期HTTPリクエストを送る方法は？",
        category="python",
    ),
    TestCase(
        query="Gemini APIでストリーミングレスポンスを受け取るには？",
        category="gemini",
    ),
    TestCase(
        query="REST APIの認証方式にはどのような種類がありますか？",
        category="api",
    ),
    # ... 本番では30件以上を推奨
]
 
# A/Bテスト実行
df = run_ab_test([variant_a, variant_b], test_cases)
 
# 結果分析
analysis = analyze_ab_results(df)
print(json.dumps(analysis, indent=2, ensure_ascii=False))
# 期待する出力:
# {
#   "baseline_vs_structured": {
#     "baseline_mean": 3.667,
#     "variant_mean": 4.333,
#     "improvement": 0.667,
#     "p_value": 0.0312,
#     "significant": true,
#     "cohens_d": 0.89,
#     "effect_size": "large",
#     "cost_ratio": 1.15,
#     "recommendation": "✅ 採用推奨。品質が大幅に改善し、コスト増加も許容範囲内です。"
#   }
# }

コスト・品質トレードオフの可視化

本番環境では、品質だけでなくコストとレイテンシのバランスが重要です。以下の関数で、モデルとプロンプトの組み合わせごとにパレート最適なポイントを特定します。

def compute_pareto_frontier(df: pd.DataFrame) -> pd.DataFrame:
    """
    品質スコアとコストのパレートフロンティアを計算する。
    パレート最適な組み合わせのみを抽出する。
    """
    summary = df.groupby("variant").agg(
        avg_score=("overall_score", "mean"),
        avg_cost=("cost_usd", "mean"),
        avg_latency=("latency_ms", "mean"),
        total_tokens=("tokens", "sum"),
    ).reset_index()
 
    # パレート最適なポイントを抽出
    # （同じコストでより高品質、または同じ品質でより低コスト）
    pareto = []
    for _, row in summary.iterrows():
        dominated = False
        for _, other in summary.iterrows():
            if (other["avg_score"] >= row["avg_score"] and
                other["avg_cost"] <= row["avg_cost"] and
                (other["avg_score"] > row["avg_score"] or
                 other["avg_cost"] < row["avg_cost"])):
                dominated = True
                break
        if not dominated:
            pareto.append(row)
 
    return pd.DataFrame(pareto)
 
# 使用例: 複数モデル×プロンプトの組み合わせを比較
variants_multi = [
    PromptVariant(name="flash-simple", model="gemini-2.5-flash",
                  system_instruction="簡潔に回答してください。",
                  template="{query}"),
    PromptVariant(name="flash-detailed", model="gemini-2.5-flash",
                  system_instruction="詳細に回答してください。",
                  template="以下について詳しく説明してください。\n{query}"),
    PromptVariant(name="pro-simple", model="gemini-2.5-pro",
                  system_instruction="簡潔に回答してください。",
                  template="{query}"),
    PromptVariant(name="pro-detailed", model="gemini-2.5-pro",
                  system_instruction="詳細に回答してください。",
                  template="以下について詳しく説明してください。\n{query}"),
]

継続的モニタリングの実装

プロンプト評価は一度きりではなく、継続的に実施することで品質の劣化を早期に検知できます。

from datetime import datetime, timedelta
 
class QualityMonitor:
    """
    本番環境でのプロンプト品質を継続的にモニタリングするクラス。
    品質スコアの移動平均が閾値を下回った場合にアラートを発生させる。
    """
 
    def __init__(
        self,
        alert_threshold: float = 3.5,
        window_size: int = 50,
        min_samples: int = 10,
    ):
        self.alert_threshold = alert_threshold
        self.window_size = window_size
        self.min_samples = min_samples
        self.scores: list[dict] = []
 
    def record(self, query: str, response: str, metadata: dict = None):
        """レスポンスを記録し、自動評価を実行"""
        scores = evaluate_response(user_query=query, ai_response=response)
        entry = {
            "timestamp": datetime.now().isoformat(),
            "query": query,
            "scores": scores,
            "metadata": metadata or {},
        }
        self.scores.append(entry)
        self._check_alerts()
        return scores
 
    def _check_alerts(self):
        """移動平均を計算し、閾値を下回っていないか確認"""
        if len(self.scores) < self.min_samples:
            return
 
        recent = self.scores[-self.window_size:]
        avg = sum(
            s["scores"].get("overall", 0) for s in recent
        ) / len(recent)
 
        if avg < self.alert_threshold:
            self._send_alert(avg, len(recent))
 
    def _send_alert(self, avg_score: float, sample_size: int):
        """品質低下アラートを送信（Slack, Email等に接続可能）"""
        print(f"⚠️ 品質アラート: 直近{sample_size}件の平均スコアが"
              f"{avg_score:.2f}に低下（閾値: {self.alert_threshold}）")
 
    def get_daily_report(self) -> dict:
        """日次の品質レポートを生成"""
        today = datetime.now().date()
        today_scores = [
            s for s in self.scores
            if datetime.fromisoformat(s["timestamp"]).date() == today
        ]
 
        if not today_scores:
            return {"date": str(today), "message": "本日のデータなし"}
 
        overall_scores = [
            s["scores"].get("overall", 0) for s in today_scores
        ]
 
        return {
            "date": str(today),
            "total_evaluations": len(today_scores),
            "average_score": round(sum(overall_scores) / len(overall_scores), 2),
            "min_score": min(overall_scores),
            "max_score": max(overall_scores),
            "below_threshold": sum(
                1 for s in overall_scores if s < self.alert_threshold
            ),
        }
 
# 使用例
monitor = QualityMonitor(alert_threshold=3.5, window_size=50)
 
# 本番環境のリクエスト処理時にサンプリングして評価
# （全リクエストを評価するとコストが高いため、5-10%をサンプリング推奨）
import random
SAMPLING_RATE = 0.05  # 5%
 
def handle_request(query: str) -> str:
    """本番リクエストのハンドラー（評価サンプリング付き）"""
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=query,
    )
 
    # サンプリングで品質モニタリング
    if random.random() < SAMPLING_RATE:
        monitor.record(query=query, response=response.text)
 
    return response.text

まとめ

ここではGemini API を活用したプロンプト評価・最適化パイプラインの構築方法を解説しました。

LLM-as-Judge パターンは、プロンプトの品質を定量的に計測するための強力なアプローチです。人間の評価を完全に代替するものではありませんが、開発サイクルの中で迅速なフィードバックを得る仕組みとして非常に有用です。

本記事で紹介した A/B テスト基盤と継続的モニタリングを組み合わせれば、「なんとなく良くなった気がする」という曖昧な判断から脱却し、データドリブンなプロンプト最適化が実現できます。

まずは小さなテストケースセットから始めて、チームのワークフローに評価パイプラインを組み込んでいくことをお勧めします。プロンプトの品質管理は、LLM アプリケーションの競争力を左右する重要な投資です。