Claude API LLM評価パイプライン構築ガイド — Claude-as-Judge・プロンプトA/Bテスト・品質スコアリングの実装パターン

取り組みの背景 — なぜLLM評価パイプラインが必要なのか

Claude APIを使ったアプリケーションが本番環境で稼働し始めると、避けて通れない課題があります。それは「出力品質をどう保証するか」という問題です。

プロンプトを少し変更しただけで、想定外の出力が返ってくることは珍しくありません。モデルのバージョンアップで既存のワークフローが壊れることもあります。手動でのスポットチェックでは、数百パターンのエッジケースを見落とします。

対象読者

Claude APIを使ったアプリケーションを本番運用している開発者
プロンプトの品質管理に課題を感じているチーム
LLMアプリケーションのCI/CDパイプラインに評価を組み込みたいエンジニア

前提環境

Python 3.11以上
anthropic Python SDK（最新版）
TypeScriptでの実装例も併記

Claude-as-Judge パターン — LLMで出力を自動評価する

基本原理

Claude-as-Judgeとは、Claude自身を「評価者」として使い、別のClaude呼び出し（またはLLM出力全般）の品質を判定するパターンです。人間の評価者と高い相関を持ちながら、数千件の評価を数分で完了できるのが最大の利点です。

評価ルーブリックの設計

まず、評価基準（ルーブリック）を明確に定義します。曖昧な基準では評価結果がブレるため、具体的なスコアリング基準が不可欠です。

# evaluation_rubric.py — 評価ルーブリックの定義
from dataclasses import dataclass, field
from typing import Optional
 
@dataclass
class EvalCriterion:
    """単一の評価基準"""
    name: str
    description: str
    scoring_guide: dict[int, str]  # スコア → 説明
    weight: float = 1.0
 
@dataclass
class EvalRubric:
    """評価ルーブリック全体"""
    name: str
    criteria: list[EvalCriterion] = field(default_factory=list)
 
    def to_prompt(self) -> str:
        """ルーブリックをプロンプト文字列に変換"""
        lines = [f"# 評価ルーブリック: {self.name}\n"]
        for c in self.criteria:
            lines.append(f"## {c.name}（重み: {c.weight}）")
            lines.append(f"{c.description}\n")
            for score, desc in sorted(c.scoring_guide.items()):
                lines.append(f"- **{score}点**: {desc}")
            lines.append("")
        return "\n".join(lines)
 
# 使用例: カスタマーサポート回答の評価ルーブリック
support_rubric = EvalRubric(
    name="カスタマーサポート回答品質",
    criteria=[
        EvalCriterion(
            name="正確性",
            description="回答内容が事実に基づいており、誤情報を含んでいないか",
            scoring_guide={
                1: "重大な事実誤認がある",
                2: "軽微な不正確さがある",
                3: "概ね正確だが曖昧な箇所がある",
                4: "正確で具体的な情報を提供している",
                5: "完全に正確で、補足情報も適切",
            },
            weight=2.0,
        ),
        EvalCriterion(
            name="完全性",
            description="質問に対して必要な情報を漏れなく回答しているか",
            scoring_guide={
                1: "質問の主要部分に回答していない",
                2: "部分的にしか回答していない",
                3: "主要な質問には回答しているが補足が不足",
                4: "質問に十分回答し、次のステップも示している",
                5: "質問に完全回答し、関連情報も先回りで提供",
            },
            weight=1.5,
        ),
        EvalCriterion(
            name="トーン",
            description="回答のトーンが適切で、顧客に寄り添っているか",
            scoring_guide={
                1: "冷たい・事務的すぎる",
                2: "やや機械的",
                3: "標準的で問題はない",
                4: "親しみやすく丁寧",
                5: "共感的で安心感を与える",
            },
            weight=1.0,
        ),
    ],
)

Claude-as-Judge の実装

ルーブリックを使ってClaudeに評価させるコア関数を実装します。

# claude_judge.py — Claude-as-Judge 評価エンジン
import anthropic
import json
from typing import Any
 
client = anthropic.Anthropic()  # ANTHROPIC_API_KEY 環境変数を使用
 
async def evaluate_with_claude(
    input_text: str,
    output_text: str,
    rubric: EvalRubric,
    reference_answer: str | None = None,
    model: str = "claude-sonnet-4-6",
) -> dict[str, Any]:
    """
    Claude-as-Judgeで出力を評価する
 
    Args:
        input_text: 元の入力（ユーザーの質問等）
        output_text: 評価対象の出力
        rubric: 評価ルーブリック
        reference_answer: 参照回答（オプション）
        model: 評価に使用するモデル
 
    Returns:
        各基準のスコアと総合スコアを含む辞書
    """
    system_prompt = """あなたはLLM出力の品質を評価する専門家です。
与えられたルーブリックに厳密に従い、各基準についてスコアを付けてください。
評価は客観的かつ一貫性を持って行ってください。
 
必ず以下のJSON形式で回答してください:
{
  "scores": {
    "基準名": {"score": 数値, "reasoning": "理由"},
    ...
  },
  "weighted_total": 加重合計スコア,
  "max_possible": 最大可能スコア,
  "overall_feedback": "総合的なフィードバック"
}"""
 
    user_prompt = f"""以下の入力に対する出力を評価してください。
 
{rubric.to_prompt()}
 
## 入力
{input_text}
 
## 評価対象の出力
{output_text}
"""
    if reference_answer:
        user_prompt += f"\n## 参照回答（理想的な回答例）\n{reference_answer}\n"
 
    response = client.messages.create(
        model=model,
        max_tokens=2000,
        system=system_prompt,
        messages=[{"role": "user", "content": user_prompt}],
    )
 
    # JSON部分を抽出してパース
    text = response.content[0].text
    json_start = text.find("{")
    json_end = text.rfind("}") + 1
    result = json.loads(text[json_start:json_end])
 
    return result
 
# 実行例
# result = await evaluate_with_claude(
#     input_text="返品の手続きを教えてください",
#     output_text="返品は購入後30日以内に...",
#     rubric=support_rubric,
# )
# print(f"総合スコア: {result['weighted_total']}/{result['max_possible']}")

期待される出力:

{
  "scores": {
    "正確性": {"score": 4, "reasoning": "返品期限と手順が正確に説明されている"},
    "完全性": {"score": 3, "reasoning": "基本手順は網羅しているが送料負担の説明が不足"},
    "トーン": {"score": 4, "reasoning": "丁寧で親しみやすいトーン"}
  },
  "weighted_total": 16.5,
  "max_possible": 22.5,
  "overall_feedback": "正確で丁寧な回答だが、送料に関する補足が望ましい"
}

評価の一貫性を高めるテクニック

Claude-as-Judgeの評価精度を上げるために、以下のテクニックが有効です。

Few-shot examples: ルーブリックに加え、各スコアレベルの具体的な回答例を提示する
Pairwise comparison: 2つの出力を比較させる方式は、絶対評価より一貫性が高い
複数回評価の中央値: 同じ評価を3回実行し、中央値を採用することでバラつきを抑える

async def evaluate_with_consistency(
    input_text: str,
    output_text: str,
    rubric: EvalRubric,
    n_runs: int = 3,
) -> dict:
    """複数回評価して中央値を取る"""
    import statistics
 
    results = []
    for _ in range(n_runs):
        result = await evaluate_with_claude(input_text, output_text, rubric)
        results.append(result)
 
    # 各基準のスコアの中央値を計算
    median_scores = {}
    for criterion in rubric.criteria:
        scores = [r["scores"][criterion.name]["score"] for r in results]
        median_scores[criterion.name] = statistics.median(scores)
 
    return {
        "median_scores": median_scores,
        "all_runs": results,
        "consistency": _calculate_consistency(results, rubric),
    }

プロンプトA/Bテストフレームワーク

なぜプロンプトのA/Bテストが必要か

「このプロンプトの方が良い気がする」という感覚的な判断は、本番環境では危険です。プロンプトA/Bテストでは、2つ（以上）のプロンプトバリアントを同じテストケース群で実行し、統計的に有意な差があるかを判定します。

テストフレームワークの設計

# prompt_ab_test.py — プロンプトA/Bテストフレームワーク
import anthropic
import asyncio
from dataclasses import dataclass
from datetime import datetime
from scipy import stats
import numpy as np
 
client = anthropic.Anthropic()
 
@dataclass
class PromptVariant:
    """プロンプトバリアント"""
    name: str
    system_prompt: str
    model: str = "claude-sonnet-4-6"
    temperature: float = 0.0
    max_tokens: int = 4096
 
@dataclass
class TestCase:
    """テストケース"""
    id: str
    input_text: str
    reference_answer: str | None = None
    metadata: dict | None = None
 
@dataclass
class ABTestResult:
    """A/Bテスト結果"""
    variant_a_scores: list[float]
    variant_b_scores: list[float]
    t_statistic: float
    p_value: float
    is_significant: bool
    winner: str | None
    effect_size: float
 
async def run_ab_test(
    variant_a: PromptVariant,
    variant_b: PromptVariant,
    test_cases: list[TestCase],
    rubric: EvalRubric,
    significance_level: float = 0.05,
) -> ABTestResult:
    """
    2つのプロンプトバリアントをA/Bテストする
 
    Args:
        variant_a: バリアントA（ベースライン）
        variant_b: バリアントB（チャレンジャー）
        test_cases: テストケース群
        rubric: 評価ルーブリック
        significance_level: 有意水準（デフォルト5%）
    """
    scores_a, scores_b = [], []
 
    for tc in test_cases:
        # 両バリアントで出力を生成
        output_a = await _generate(variant_a, tc.input_text)
        output_b = await _generate(variant_b, tc.input_text)
 
        # Claude-as-Judgeで評価
        eval_a = await evaluate_with_claude(
            tc.input_text, output_a, rubric, tc.reference_answer
        )
        eval_b = await evaluate_with_claude(
            tc.input_text, output_b, rubric, tc.reference_answer
        )
 
        scores_a.append(eval_a["weighted_total"] / eval_a["max_possible"])
        scores_b.append(eval_b["weighted_total"] / eval_b["max_possible"])
 
    # 対応のあるt検定（同じテストケースで比較）
    t_stat, p_value = stats.ttest_rel(scores_a, scores_b)
    is_significant = p_value < significance_level
 
    # 効果量（Cohen's d）
    diff = np.array(scores_a) - np.array(scores_b)
    effect_size = np.mean(diff) / np.std(diff) if np.std(diff) > 0 else 0
 
    winner = None
    if is_significant:
        winner = variant_a.name if np.mean(scores_a) > np.mean(scores_b) else variant_b.name
 
    return ABTestResult(
        variant_a_scores=scores_a,
        variant_b_scores=scores_b,
        t_statistic=t_stat,
        p_value=p_value,
        is_significant=is_significant,
        winner=winner,
        effect_size=effect_size,
    )
 
async def _generate(variant: PromptVariant, input_text: str) -> str:
    """プロンプトバリアントで出力を生成"""
    response = client.messages.create(
        model=variant.model,
        max_tokens=variant.max_tokens,
        temperature=variant.temperature,
        system=variant.system_prompt,
        messages=[{"role": "user", "content": input_text}],
    )
    return response.content[0].text
 
# 期待される出力例:
# ABTestResult(
#     variant_a_scores=[0.82, 0.78, 0.85, ...],
#     variant_b_scores=[0.88, 0.91, 0.87, ...],
#     t_statistic=-3.42,
#     p_value=0.002,
#     is_significant=True,
#     winner="variant_b_detailed_prompt",
#     effect_size=0.65
# )

テストケースの設計戦略

A/Bテストの信頼性は、テストケースの質と量に大きく依存します。

最低30ケース: 統計的検出力を確保するため、最低30件のテストケースを用意する
エッジケースの網羅: 正常系だけでなく、曖昧な入力・長文入力・多言語入力を含める
カテゴリ別バランス: ユースケースごとのテストケース数を均等にする
ゴールデンデータセット: 人間が書いた理想的な回答を参照回答として用意する

# テストケースセットの構築例
test_cases = [
    TestCase(
        id="support-001",
        input_text="注文した商品が届きません。注文番号はORD-12345です。",
        reference_answer="ご不便をおかけして申し訳ございません。注文番号ORD-12345を...",
        metadata={"category": "shipping", "difficulty": "standard"},
    ),
    TestCase(
        id="support-002",
        input_text="先月の請求が二重になっているようです",
        reference_answer=None,  # 参照回答なし → ルーブリック評価のみ
        metadata={"category": "billing", "difficulty": "complex"},
    ),
    # ... 30件以上
]

回帰テスト — プロンプト変更時の品質担保

回帰テストの重要性

プロンプトの改善やモデルバージョンの変更は、意図しない品質低下（リグレッション）を引き起こすことがあります。回帰テストは「現行バージョンと同等以上の品質を維持しているか」を自動で検証する仕組みです。

CI/CDへの組み込み

# regression_test.py — プロンプト回帰テスト
import json
from pathlib import Path
 
BASELINE_PATH = Path("eval/baselines/current.json")
THRESHOLD = 0.95  # ベースラインの95%以上を維持
 
async def run_regression_test(
    variant: PromptVariant,
    test_cases: list[TestCase],
    rubric: EvalRubric,
) -> dict:
    """
    回帰テストを実行し、ベースラインとの比較結果を返す
    """
    # 現在のベースラインを読み込む
    baseline = json.loads(BASELINE_PATH.read_text())
 
    # 新バリアントでスコアを計算
    new_scores = []
    for tc in test_cases:
        output = await _generate(variant, tc.input_text)
        eval_result = await evaluate_with_claude(
            tc.input_text, output, rubric
        )
        normalized = eval_result["weighted_total"] / eval_result["max_possible"]
        new_scores.append({"id": tc.id, "score": normalized})
 
    # ベースラインとの比較
    baseline_avg = sum(b["score"] for b in baseline["scores"]) / len(baseline["scores"])
    new_avg = sum(s["score"] for s in new_scores) / len(new_scores)
    ratio = new_avg / baseline_avg if baseline_avg > 0 else 0
 
    passed = ratio >= THRESHOLD
    regressions = _find_regressions(baseline["scores"], new_scores)
 
    return {
        "passed": passed,
        "baseline_avg": round(baseline_avg, 4),
        "new_avg": round(new_avg, 4),
        "ratio": round(ratio, 4),
        "threshold": THRESHOLD,
        "regressions": regressions,  # スコアが大幅に下がったケース
        "improvements": _find_improvements(baseline["scores"], new_scores),
    }
 
def _find_regressions(baseline_scores, new_scores, drop_threshold=0.15):
    """ベースラインからスコアが大幅に低下したケースを検出"""
    regressions = []
    baseline_map = {s["id"]: s["score"] for s in baseline_scores}
    for ns in new_scores:
        if ns["id"] in baseline_map:
            drop = baseline_map[ns["id"]] - ns["score"]
            if drop > drop_threshold:
                regressions.append({
                    "id": ns["id"],
                    "baseline_score": baseline_map[ns["id"]],
                    "new_score": ns["score"],
                    "drop": round(drop, 4),
                })
    return regressions
 
# GitHub Actions / CI での使用例:
# python -m pytest eval/test_regression.py -v
# → 失敗すればPRをブロック

期待される出力:

{
  "passed": true,
  "baseline_avg": 0.8234,
  "new_avg": 0.8512,
  "ratio": 1.0338,
  "threshold": 0.95,
  "regressions": [],
  "improvements": [
    {"id": "support-015", "baseline_score": 0.72, "new_score": 0.89, "gain": 0.17}
  ]
}

品質モニタリングダッシュボード

本番環境での継続的モニタリング

評価パイプラインは開発時だけでなく、本番環境でも継続的に動かす点が肝心です。以下は、本番リクエストの一部をサンプリングして品質をモニタリングするパターンです。

# quality_monitor.py — 本番品質モニタリング
import random
import logging
from datetime import datetime, timezone
 
logger = logging.getLogger(__name__)
 
class QualityMonitor:
    """本番環境での品質モニタリング"""
 
    def __init__(
        self,
        rubric: EvalRubric,
        sample_rate: float = 0.05,  # 5%サンプリング
        alert_threshold: float = 0.6,  # 60%未満でアラート
    ):
        self.rubric = rubric
        self.sample_rate = sample_rate
        self.alert_threshold = alert_threshold
        self._scores_buffer: list[dict] = []
 
    async def maybe_evaluate(
        self, input_text: str, output_text: str
    ) -> dict | None:
        """サンプリングレートに基づいて評価を実行"""
        if random.random() > self.sample_rate:
            return None  # スキップ
 
        result = await evaluate_with_claude(
            input_text, output_text, self.rubric
        )
        normalized = result["weighted_total"] / result["max_possible"]
 
        record = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "score": normalized,
            "details": result,
        }
        self._scores_buffer.append(record)
 
        # アラート判定
        if normalized < self.alert_threshold:
            logger.warning(
                f"品質アラート: スコア {normalized:.2f} "
                f"(閾値: {self.alert_threshold})"
            )
            await self._send_alert(record)
 
        return record
 
    async def get_daily_report(self) -> dict:
        """日次品質レポートを生成"""
        if not self._scores_buffer:
            return {"status": "no_data"}
 
        scores = [r["score"] for r in self._scores_buffer]
        return {
            "date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
            "total_evaluated": len(scores),
            "avg_score": round(sum(scores) / len(scores), 4),
            "min_score": round(min(scores), 4),
            "max_score": round(max(scores), 4),
            "below_threshold": sum(1 for s in scores if s < self.alert_threshold),
            "p50": round(sorted(scores)[len(scores) // 2], 4),
            "p90": round(sorted(scores)[int(len(scores) * 0.9)], 4),
        }
 
    async def _send_alert(self, record: dict):
        """品質アラートを送信（Slack/PagerDuty等に接続）"""
        # 実装例: Slack webhook に送信
        pass

TypeScript版の実装

TypeScript SDKを使った実装例も示します。

// eval-pipeline.ts — TypeScript版評価パイプライン
import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
interface EvalResult {
  scores: Record<string, { score: number; reasoning: string }>;
  weightedTotal: number;
  maxPossible: number;
  overallFeedback: string;
}
 
async function evaluateOutput(
  input: string,
  output: string,
  rubricPrompt: string,
  model: string = "claude-sonnet-4-6"
): Promise<EvalResult> {
  const response = await client.messages.create({
    model,
    max_tokens: 2000,
    system: `あなたはLLM出力の品質評価者です。JSON形式で回答してください。`,
    messages: [
      {
        role: "user",
        content: `${rubricPrompt}\n\n## 入力\n${input}\n\n## 出力\n${output}`,
      },
    ],
  });
 
  const text =
    response.content[0].type === "text" ? response.content[0].text : "";
  const jsonMatch = text.match(/\{[\s\S]*\}/);
  if (!jsonMatch) throw new Error("JSON parse failed");
 
  return JSON.parse(jsonMatch[0]) as EvalResult;
}
 
// Batch API を使って大量評価を効率化
async function batchEvaluate(
  cases: Array<{ input: string; output: string }>,
  rubricPrompt: string
): Promise<EvalResult[]> {
  // Batch APIで並列処理（コスト50%削減）
  const results = await Promise.all(
    cases.map((c) => evaluateOutput(c.input, c.output, rubricPrompt))
  );
  return results;
}

まとめ

Claude-as-Judgeパターンは、ルーブリック設計と一貫性テクニックを組み合わせることで、人間の評価に匹敵する品質判定を自動化できます。プロンプトA/Bテストフレームワークは、統計的有意性に基づいた意思決定を可能にし、「感覚的な改善」から「データドリブンな最適化」へとプロンプト管理を進化させます。そして回帰テストと品質モニタリングは、本番環境での品質担保の最後のピースです。

これらを組み合わせた評価パイプラインは、Claude APIを使ったアプリケーションの品質を体系的に管理するための基盤となります。まずは小さなテストスイートから始めて、徐々にカバレッジを拡大していくのがおすすめです。