AI Engineering Academy · レッスン

自動評価ハーネスの構築

テストセットに対してRAGシステム全体を実行し、すべての指標を計算してレポートを生成する再現可能な評価パイプラインを作成し、時間の経過に伴う改善を追跡できるようにします。

レッスン 4/413 ステップ

「自動評価ハーネスの構築」はCoddyKit上の無料AI Engineering Academyレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Engineering Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Engineering Academyコースには全4レッスンが含まれています。

評価ハーネスとは何か

評価ハーネスとは、標準化されたテストセットに対してRAGシステム全体を実行し、すべての指標を計算してレポートを作成する、再現可能な自動パイプラインです。重要なのは再現可能であることです。チャンク分割戦略、埋め込みモデル、プロンプト、LLMのいずれかを変更するたびに同じハーネスを実行し、ベースラインと結果を比較します。これにより、RAG開発は主観的な試行錯誤からデータ駆動型のエンジニアリングへと変わります。

ハーネスのアーキテクチャ

適切に設計された評価ハーネスは、4つの層で構成されます。テストデータ管理(ゴールデンデータセットの読み込みとバージョン管理)、パイプライン実行(各テスト質問をRAGパイプライン全体に通す)、指標計算(検索と生成に関するすべての指標を計算する)、レポート生成(バージョン情報付きで結果を保存し、以前のベースラインとの差分を作成する)です。各層は独立してテストおよび設定できるようにします。

class RAGEvaluationHarness:
    def __init__(self, retriever, llm_client, config):
        self.retriever = retriever
        self.llm_client = llm_client
        self.config = config  # chunk_size, top_k, model, threshold, etc.
        self.results = []

    def run(self, golden_dataset):
        for item in golden_dataset:
            result = self._evaluate_single(item)
            self.results.append(result)
        metrics = self._compute_metrics()
        self._save_report(metrics)
        return metrics

各テストケースを実行する

ゴールデンデータセット内の各質問について、RAGパイプライン全体を実行し、すべての中間出力を記録します。これには、取得したチャンクIDとスコア、整形済みコンテキスト、生成された回答、トークン数が含まれます。中間値を保存することは、失敗のデバッグに不可欠です。質問のスコアが低い場合でも、高コストなパイプラインを再実行せずに、どのチャンクが取得され、なぜ回答が誤っていたのかを正確に調べられます。

import time

def _evaluate_single(self, item):
    start = time.perf_counter()
    query_vector = embed_query(item['question'])
    chunks = self.retriever.retrieve(query_vector, top_k=self.config['top_k'])
    filtered_chunks = filter_by_score(chunks, self.config['threshold'])
    context = format_context(filtered_chunks)
    answer_result = generate_answer(item['question'], context, self.llm_client)
    latency_ms = (time.perf_counter() - start) * 1000

    return {
        'question': item['question'],
        'expected_answer': item['answer'],
        'generated_answer': answer_result['answer'],
        'retrieved_chunk_ids': [c['id'] for c in filtered_chunks],
        'retrieved_scores': [c['score'] for c in filtered_chunks],
        'relevant_chunk_ids': item['relevant_chunk_ids'],
        'context_texts': [c['text'] for c in filtered_chunks],
        'tokens_used': answer_result['tokens_used'],
        'latency_ms': round(latency_ms)
    }

すべての指標を1回の処理で計算する

すべてのテストケースの出力を収集したら、結果全体を1回走査して指標一式を計算します。検索指標(チャンクIDから計算)と生成指標(評価用LLMを呼び出して計算)を分けます。効率を最大化するため、評価用LLMの呼び出しはバッチ処理します。たとえば、忠実性の評価をまとめ、順番に実行するのではなくasyncioを使って並列に送信します。生成指標は100件を超えるテストケースで数分かかる場合があるため、進捗をログに記録します。

def _compute_metrics(self):
    # Retrieval metrics (no LLM calls needed)
    hit_rates = []
    mrr_scores = []
    for r in self.results:
        retrieved = r['retrieved_chunk_ids']
        relevant = set(r['relevant_chunk_ids'])
        hit = any(rid in relevant for rid in retrieved)
        hit_rates.append(1.0 if hit else 0.0)
        for rank, rid in enumerate(retrieved, 1):
            if rid in relevant:
                mrr_scores.append(1.0 / rank)
                break
        else:
            mrr_scores.append(0.0)

    metrics = {
        'hit_rate_at_5': sum(hit_rates) / len(hit_rates),
        'mrr': sum(mrr_scores) / len(mrr_scores),
        'mean_latency_ms': sum(r['latency_ms'] for r in self.results) / len(self.results),
        'mean_tokens': sum(r['tokens_used'] for r in self.results) / len(self.results)
    }
    return metrics

バージョン情報付きで結果を保存する

設定間で結果を比較できるよう、評価実行のたびにバージョンメタデータとともに保存します。コードのgitコミットハッシュ、設定パラメータ(埋め込みモデル、チャンクサイズ、K、しきい値、LLMモデル)、タイムスタンプ、人間が読める実行内容の説明を含めます。結果はJSON Linesファイルまたはデータベーステーブルに保存します。これにより、システムがどのように進化してきたかを恒久的に記録できます。

import json
import subprocess
from datetime import datetime

def _save_report(self, metrics):
    git_hash = subprocess.check_output(
        ['git', 'rev-parse', '--short', 'HEAD']
    ).decode().strip()

    report = {
        'run_id': datetime.utcnow().strftime('%Y%m%d_%H%M%S'),
        'git_commit': git_hash,
        'config': self.config,
        'metrics': metrics,
        'n_test_cases': len(self.results),
        'timestamp': datetime.utcnow().isoformat()
    }

    with open('eval_history.jsonl', 'a') as f:
        f.write(json.dumps(report) + '\n')
    print(f'Saved evaluation run: {report["run_id"]}')
    print(json.dumps(metrics, indent=2))

ベースラインと比較する

各実行後に以前のベースラインと自動的に比較し、リグレッションを検出します。リグレッションとは、いずれかの指標がしきい値(例:2パーセントポイント)を超えて低下することです。指標の変化を示す差分テーブルを出力します。いずれかの指標に大きなリグレッションがある場合、評価実行はゼロ以外の終了コードで失敗させます。これによりCI/CDパイプラインが変更のデプロイを阻止します。

def compare_to_baseline(current_metrics, baseline_file='best_eval.json'):
    import json
    from pathlib import Path
    if not Path(baseline_file).exists():
        print('No baseline yet. Saving current as baseline.')
        Path(baseline_file).write_text(json.dumps(current_metrics, indent=2))
        return True

    baseline = json.loads(Path(baseline_file).read_text())
    regressions = []
    print('\nMetric comparison (current vs baseline):')
    for metric, current_val in current_metrics.items():
        baseline_val = baseline.get(metric, 0)
        delta = current_val - baseline_val
        status = 'OK' if delta >= -0.02 else 'REGRESSION'
        print(f'  {metric}: {current_val:.3f} vs {baseline_val:.3f} ({delta:+.3f}) {status}')
        if status == 'REGRESSION':
            regressions.append(metric)
    return len(regressions) == 0

CI/CDに統合する

評価ハーネスは、CI/CDパイプラインに統合したときに最も大きな効果を発揮します。チャンク分割ロジック、埋め込みモデルの設定、プロンプトテンプレート、検索パラメータを変更するすべてのプルリクエストで、自動的に実行されるよう設定します。すべての指標が最低しきい値を満たし、メインブランチのベースラインからリグレッションしていない場合にのみ、パイプラインが成功するようにします。これにより、品質が意図せず低下した変更が本番環境にリリースされるのを防げます。

# GitHub Actions workflow (eval.yml)
# on:
#   pull_request:
#     paths:
#       - 'rag/**'
#       - 'prompts/**'
#       - 'config/**'
# jobs:
#   evaluate:
#     runs-on: ubuntu-latest
#     steps:
#       - uses: actions/checkout@v3
#       - name: Install dependencies
#         run: pip install -r requirements.txt
#       - name: Run evaluation harness
#         run: |
#           python eval/run_harness.py \
#             --test-set eval/golden_dataset.json \
#             --config config/rag_config.yaml \
#             --fail-on-regression
#         env:
#           OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}

人間が読みやすいレポートを生成する

生の指標ファイルに加えて、チームがプルリクエストのコメントで確認できる人間が読みやすいHTMLまたはMarkdownのレポートを生成します。すべての指標の概要テーブル、質問・期待される回答・生成された回答・取得されたチャンクを含む失敗テストケースの一覧、過去10回の実行における指標の推移を示すトレンドチャートを含めます。視覚的なレポートにより、技術に詳しくない関係者もシステムが改善しているか理解しやすくなります。

def generate_markdown_report(metrics, failed_cases, run_id):
    lines = [
        f'# RAG Evaluation Report — {run_id}\n',
        '## Summary Metrics',
        '| Metric | Score | Target |',
        '|--------|-------|--------|',
        f'| Hit Rate@5 | {metrics["hit_rate_at_5"]:.1%} | > 80% |',
        f'| MRR | {metrics["mrr"]:.3f} | > 0.70 |',
        f'| Mean Latency | {metrics["mean_latency_ms"]:.0f}ms | < 500ms |',
        '',
        f'## Failed Cases ({len(failed_cases)} failures)'
    ]
    for case in failed_cases[:10]:  # show first 10
        lines += [
            f'**Q:** {case["question"]}',
            f'**Expected:** {case["expected_answer"]}',
            f'**Generated:** {case["generated_answer"]}\n'
        ]
    return '\n'.join(lines)

評価実行ごとのコストを追跡する

評価実行には費用がかかります。埋め込みAPI、LLM API、評価用LLMを呼び出すためです。品質指標とともに各評価実行のコストを追跡します。100件のテストケースを包括的に評価する場合、使用するモデルによって通常$0.50~$2.00かかります。評価用LLMの呼び出しにはより安価なモデル(忠実性の評価にはGPT-4o-miniなど)を使い、生成には高価なモデルを使用します。保存するレポートに実行コストの見積もりを含め、開発サイクルに評価費用を組み込めるようにします。

def estimate_run_cost(results, config):
    # Embedding cost
    embed_tokens = sum(len(r['question'].split()) * 1.3 for r in results)
    embed_cost = (embed_tokens / 1_000_000) * 0.02  # $0.02/1M tokens

    # Generation cost
    total_gen_tokens = sum(r['tokens_used'] for r in results)
    gen_cost = (total_gen_tokens / 1_000_000) * 5.0  # gpt-4o approx

    # Judge cost (faithfulness evals)
    judge_cost = len(results) * 0.001  # ~$0.001 per eval with gpt-4o-mini

    total = embed_cost + gen_cost + judge_cost
    print(f'Evaluation cost estimate: ${total:.2f}')
    print(f'  Embedding: ${embed_cost:.3f}')
    print(f'  Generation: ${gen_cost:.3f}')
    print(f'  Judgment: ${judge_cost:.3f}')
    return total

本番監視のための定期評価

コード変更時のCI/CD評価に加えて、本番環境で定期的にハーネスを実行し、ログから抽出した実際のユーザークエリに対して毎日または毎週テストします。これによりデータドリフトを検出できます。ドキュメントコーパスが変化し、ユーザークエリのパターンが移り変わると、コードを変更していなくてもシステム品質が低下することがあります。毎週、直近のユーザークエリを50件抽出して評価し、品質の概要をチームのSlackチャンネルに自動送信するよう設定します。

# Example scheduled evaluation (cron job or scheduled cloud function)
import random

def sample_production_queries(query_log_file, n=50):
    with open(query_log_file) as f:
        all_queries = [json.loads(line) for line in f]
    sample = random.sample(all_queries, min(n, len(all_queries)))
    # Convert to golden dataset format (without expected answers — use LLM judge)
    return [
        {'question': q['user_question'], 'relevant_chunk_ids': []}
        for q in sample
    ]

# Run weekly evaluation against production queries
if __name__ == '__main__':
    prod_queries = sample_production_queries('/var/log/rag_queries.jsonl')
    harness = RAGEvaluationHarness(retriever, llm_client, config)
    metrics = harness.run(prod_queries)
    send_slack_digest(metrics)

時間経過に伴う指標の推移を可視化する

JSONLファイル内の生の数値は、一目で解釈するのが困難です。過去20回の評価実行について各指標を折れ線グラフにプロットする、シンプルなトレンド可視化を作成します。x軸には実行時刻、y軸には指標スコアを使用します。最低許容しきい値の位置に水平線を引きます。指標がしきい値を下回ると、生データを読み込まなくても問題をすぐに確認できます。MatplotlibやシンプルなWebダッシュボード(Grafana、Streamlit)などのツールが適しています。

import json
import matplotlib.pyplot as plt
from pathlib import Path

def plot_metric_trends(history_file='eval_history.jsonl', metric='hit_rate_at_5'):
    records = [
        json.loads(line)
        for line in Path(history_file).read_text().strip().split('\n')
    ]
    timestamps = [r['timestamp'][:10] for r in records[-20:]]
    scores = [r['metrics'].get(metric, 0) for r in records[-20:]]
    plt.figure(figsize=(10, 4))
    plt.plot(timestamps, scores, marker='o', label=metric)
    plt.axhline(y=0.80, color='r', linestyle='--', label='Min threshold')
    plt.title(f'{metric} over last 20 evaluations')
    plt.xticks(rotation=45)
    plt.tight_layout()
    plt.savefig(f'eval_trend_{metric}.png')
    print(f'Saved trend chart for {metric}')

理解度チェック

このレッスンで学んだAI Engineeringの概念を確認しましょう。

レッスンのまとめ

このレッスンでは、テストデータ管理、パイプライン実行、指標計算、レポート生成を備えた完全な評価ハーネスの構成方法、実行結果をベースラインと比較し、リグレッション時にCI/CDを失敗させる方法、チームで確認できる人間が読みやすいレポートの生成方法、そしてコードを変更せずにデータドリフトを検出する本番監視の定期実行方法を学びました。これで、本番RAGシステムを構築・評価するための基盤が整いました。

無料で開始

AI チューターと学ぶ Python — 無料

ブラウザでリアルコードを書いて実行し、24/7 の AI チューターから瞬時にサポートを受け、ウェブまたはアプリで続きから学習できます。

コース
30
レッスン
120

よくある質問

「自動評価ハーネスの構築」レッスンは無料ですか?

はい。「自動評価ハーネスの構築」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し(組み込みコードエディタと24時間対応のAIチューター)、AI Engineering Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Engineering Academyコースには全4レッスンが含まれています。

「自動評価ハーネスの構築」で何を学びますか?

テストセットに対してRAGシステム全体を実行し、すべての指標を計算してレポートを生成する再現可能な評価パイプラインを作成し、時間の経過に伴う改善を追跡できるようにします。 ブラウザで直接実行するハンズオンコードでAI Engineering Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Engineering Academyを始めるのに経験は必要ですか?

事前経験は必要ありません。CoddyKitのAI Engineering Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。

「自動評価ハーネスの構築」レッスンにはどのくらい時間がかかりますか?

ほとんどのCoddyKitレッスンは約5~10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Engineering Academyレッスンでコードを書いて実行できますか?

はい。すべてのAI Engineering Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. RAGで評価が重要な理由
  2. 検索指標:Hit Rate、MRR、NDCG
  3. 生成指標:忠実性と回答の関連性
  4. 自動評価ハーネスの構築
← AI Engineering Academyに戻る