STUDY NOTES

第27回: DSPy + GEPA で日本語 RAG を最適化 — MIPROv2 との対比で理解する「テキストフィードバック進化」¶

第26回 (MIPROv2) と同じ RAG パイプライン・同じ JQaRA データセットを、別の optimizer であるGEPA（Genetic-Pareto）で最適化する回。連載は意図的に「同じ問題を 2 つの optimizer で解く比較実験」として構成している。

第26回との差分: 26回の dspy.MIPROv2 は「ベイズ最適化で指示文 + few-shot demo の組合せを探す」古典的最適化器。27回の dspy.GEPA は 「LLM-as-Judge が テキストフィードバック を返し、それを reflection LM が読んで指示文を書き換える」進化的最適化器。

比喩: MIPROv2 は「数値スコアだけ見て試行錯誤」、GEPA は「人間のレビューコメントを LLM が読んで改稿する」。後者のほうが少ないサンプルでも収束しやすいことが多い。

観点	MIPROv2（第26回）	GEPA（第27回）
学習信号	スカラー score（0-1）	score + 自然言語 feedback
候補生成	プロンプト + demo を組合せ列挙	reflection LM が指示文を直接書き換える
選定戦略	ベイズ最適化	パレート最適（複数指標で支配されない候補を残す）
必要 LM	`prompt_model` (smart) + 実行用	`reflection_lm` (smart, 高 temperature) + 実行用
デバッグ容易性	中（指示文 diff は読める）	高（feedback がログに残るので「どう直したか」が読める）

同じ JQaRA・同じ RAGQA モジュールで EM 6.7% → 83.3% を再現するが、GEPA はサンプル効率が高く、過剰適合しにくいのが売り。

全体像¶

27/
├── config.py                      ← 26回と同一（OpenAI/Azure 切替、Smart/Fast LM）
├── dataset_loader.py              ← 26回と同一（JQaRA positives/negatives 分離）
├── embeddings_cache.py            ← 26回と同一（MD5 ハッシュ pickle）
├── rag_module.py                  ← 26回と同一（RewriteQuery → Retrieve → GenerateAnswer）
├── evaluator.py                   ← 26回と同一（EM + recall の comprehensive metric）
├── rag_optimization_gepa.py       ← ★ 唯一 26回と違うファイル。GEPA + Tee ロギング
├── rag_evaluation.py              ← baseline vs optimized 比較（26回と同じ構造、import 先だけ違う）
├── logs/                          ← Tee で書き出した stdout と詳細メトリクスログ
└── artifact/
    ├── rag_gepa_optimized_latest.json → rag_gepa_optimized_<timestamp>_em<NNN>.json
    └── embeddings_cache/

GEPA 最適化の処理フロー:

flowchart TD
    Init[初期 RAGQA Module<br/>Signature の docstring が指示文] --> Pop[候補プール<br/>初期 = 1 個]

    subgraph Loop ["GEPA イテレーション"]
        Pop --> Pick[親候補を選ぶ]
        Pick --> Eval[trainset の minibatch で実行]
        Eval --> Metric[gepa_metric_with_feedback<br/>各サンプルに score + feedback]
        Metric --> Refl[reflection_lm に<br/>feedback 群を渡す]
        Refl --> Propose[reflection_lm が新しい指示文を提案]
        Propose --> Mut[突然変異した子候補]
        Mut --> Evaluate[valset minibatch で score 評価]
        Evaluate --> Pareto[パレート最適性で候補プールに追加 or 棄却]
        Pareto --> Pop
    end

    Pop --> Best[最良候補を選定]
    Best --> Save[artifact/rag_gepa_optimized_<ts>_em<NN>.json]

    style Refl fill:#FFF2E8,stroke:#ED7100
    style Metric fill:#E1F5FE,stroke:#0288D1

GEPA のキモ: 通常の進化的最適化は「ランダム突然変異 + 適応度評価」だが、GEPA は 「テキストフィードバックを reflection LM に読ませて意図的な書き換えを生成」する。勾配ではなく言語による勾配。

使用ライブラリ・原理¶

`dspy.GEPA(metric, reflection_lm, auto, ...)` — 進化的・反射的最適化¶

optimizer = dspy.GEPA(
    metric=gepa_metric_with_feedback_logged,
    auto="medium",
    num_threads=4,
    reflection_minibatch_size=3,
    reflection_lm=reflection_lm,        # temperature=1.0 推奨
    candidate_selection_strategy="pareto",
    track_stats=True,
)
optimized_rag = optimizer.compile(rag, trainset=trainset, valset=valset)

主要パラメータ:

パラメータ	役割
`metric`	score + feedback を返す `dspy.Prediction` を返す関数。MIPROv2 とは signature が違うので要注意
`reflection_lm`	feedback を読んで指示文を書き換える LM。temperature=1.0 で多様性を出すのが定石
`auto`	`"light"` / `"medium"` / `"heavy"` の予算プリセット
`reflection_minibatch_size`	1 回の reflection で何サンプルの feedback を読ませるか。3-5 が無難
`candidate_selection_strategy`	`"pareto"`（複数指標で支配されない候補を残す）が推奨
`track_stats`	内部統計を保持。実験分析に必須

`gepa_metric_with_feedback` — GEPA 専用 metric signature¶

def gepa_metric_with_feedback(gold, pred, trace=None, pred_name=None, pred_trace=None):
    score = rag_comprehensive_metric(gold, pred, trace)

    feedback_parts = []
    # 回答評価
    if pred.answer.strip() == gold.answer.strip():
        feedback_parts.append("✓ 完全一致")
    else:
        # 部分一致チェック
        if gold.answer.lower() in pred.answer.lower() or pred.answer.lower() in gold.answer.lower():
            feedback_parts.append(f"△ 部分一致: 期待={gold.answer}, 実際={pred.answer}")
        else:
            feedback_parts.append(f"✗ 不正解: 期待={gold.answer}, 実際={pred.answer}")
    # 検索評価
    if positives:
        recall = ...
        feedback_parts.append("△ 検索改善余地: ...")
    # クエリ改善ヒント
    if recall < 0.5 and hasattr(pred, 'rewritten_query'):
        feedback_parts.append(f"クエリ改善を検討: '{pred.rewritten_query}'")

    feedback = " | ".join(feedback_parts)
    return dspy.Prediction(score=score, feedback=feedback)

MIPROv2 metric (-> float) と GEPA metric (-> Prediction(score, feedback)) は 戻り値が違うことに注意。GEPA は dspy.Prediction でないとエラーになる。

`pred_name` / `pred_trace` — Predictor 単位のフィードバック¶

GEPA metric の引数 pred_name と pred_trace は どの sub-Predictor のために feedback を集めるかを示す。

pred_name="rewrite" → RewriteQuery Predict の improvement のための feedback
pred_name="generate" → GenerateAnswer Predict の improvement のための feedback

これにより GEPA は 複数 Predict をそれぞれ独立に最適化できる（MIPROv2 と同じく）。本サンプルでは pred_name を feedback に含めるだけだが、本格的には pred_name ごとに feedback の内容を切り替えると効く。

`Tee` クラスで stdout を 2 系統に分岐¶

class Tee:
    def __init__(self, file_path, original_stdout):
        self.file = open(file_path, 'w', encoding='utf-8')
        self.stdout = original_stdout
    def write(self, message):
        self.stdout.write(message)
        self.file.write(message)
        self.file.flush()
    ...

sys.stdout = Tee(stdout_path, sys.stdout)

GEPA 最適化は数百回の LLM 呼び出し + DSPy 内部の冗長 print が走るため、stdout を画面とファイル両方に同時保存して後で再現性確保。Unix の tee コマンドの Python 内 PoC。

logging モジュールで metric 評価詳細を記録¶

logger = logging.getLogger("gepa_optimization")
logger.setLevel(logging.INFO)
file_handler = logging.FileHandler(log_path, encoding='utf-8')
...

def gepa_metric_with_feedback_logged(gold, pred, trace=None, pred_name=None, pred_trace=None):
    result = gepa_metric_with_feedback(gold, pred, trace, pred_name, pred_trace)
    log_metric_evaluation(gold, pred, trace, pred_name, pred_trace, result)
    return result

ロガーで logs/gepa_optimization_<timestamp>.log に全 metric 評価の入力と結果を残す。GEPA は「reflection LM がどの feedback を読んだか」が肝なので、これがないとデバッグ困難。

ファイル別の役割¶

ファイル	役割	26回からの差分
`config.py`	LM/Embedder 設定	なし
`dataset_loader.py`	JQaRA ロード + positives/negatives 分離	なし
`embeddings_cache.py`	MD5 pickle キャッシュ	なし
`rag_module.py`	RAGQA Module（RewriteQuery + Retrieve + GenerateAnswer）	なし
`evaluator.py`	EM-only metric + comprehensive metric	なし
`rag_optimization_gepa.py`	MIPROv2 → GEPA, train/val 30/70 → 50/50, Tee + logger 追加	大部分書き直し
`rag_evaluation.py`	baseline vs optimized 比較	import 先が `rag_optimization` → `rag_optimization_gepa` （※ 既知不具合あり、後述）

行レベルの工夫（中核ロジックの抜粋）¶

① GEPA 専用 metric の構築 (`rag_optimization_gepa.py:43-105`)¶

def gepa_metric_with_feedback(gold, pred, trace=None, pred_name=None, pred_trace=None):
    score = rag_comprehensive_metric(gold, pred, trace)                       # ①

    feedback_parts = []

    # 回答評価
    if pred.answer.strip() == gold.answer.strip():
        feedback_parts.append("✓ 完全一致")
    else:
        pred_lower = pred.answer.strip().lower()
        gold_lower = gold.answer.strip().lower()
        if gold_lower in pred_lower or pred_lower in gold_lower:              # ②
            feedback_parts.append(f"△ 部分一致: 期待={gold.answer}, 実際={pred.answer}")
        else:
            feedback_parts.append(f"✗ 不正解: 期待={gold.answer}, 実際={pred.answer}")

    # 検索評価
    retrieved = set(pred.retrieved_passages) if hasattr(pred, 'retrieved_passages') else set()
    positives = set(gold.positives) if hasattr(gold, 'positives') and gold.positives else set()

    if positives:
        overlap = len(retrieved & positives)
        max_retrievable = min(len(positives), RETRIEVAL_K)
        recall = overlap / max_retrievable if max_retrievable > 0 else 0

        if recall >= 0.8:
            feedback_parts.append(f"✓ 検索良好: {overlap}/{max_retrievable}個の正解文書")
        elif recall >= 0.5:
            feedback_parts.append(f"△ 検索改善余地: {overlap}/{max_retrievable}個の正解文書")
        else:
            feedback_parts.append(f"✗ 検索不良: {overlap}/{max_retrievable}個の正解文書")

        if recall < 0.5 and hasattr(pred, 'rewritten_query'):
            feedback_parts.append(f"クエリ改善を検討: '{pred.rewritten_query}'")          # ③

    if pred_name:
        feedback_parts.append(f"[{pred_name}]")

    feedback = " | ".join(feedback_parts)

    return dspy.Prediction(score=score, feedback=feedback)                    # ④

行	やってること	なぜそうする
①	26回と同じ comprehensive metric で score 計算	学習信号の数値部分は変えない。GEPA は score + feedback の両方を使う
②	部分一致を「△」として明示	完全一致 0、不一致 0 の二値だと reflection LM が「何が惜しい」を読み取れない。「△ 部分一致」を入れると LLM が「あと一歩」と認識して微調整を提案
③	クエリ改善ヒントを feedback に注入	検索不良時に「現在のリライト後クエリはこれ」と渡すことで、reflection LM が「もっと検索に効くクエリ生成プロンプトに直そう」と気づく
④	`dspy.Prediction(score=..., feedback=...)` で返す	GEPA の signature 要件。`float` を返すと TypeError

② GEPA optimizer の組み立て (`rag_optimization_gepa.py:285-301`)¶

optimizer = dspy.GEPA(
    metric=gepa_metric_with_feedback_logged,                                  # ①
    auto="medium",
    num_threads=4,                                                            # ②
    reflection_minibatch_size=3,                                              # ③
    reflection_lm=reflection_lm,                                              # ④
    candidate_selection_strategy="pareto",                                    # ⑤
    track_stats=True,
)
optimized_rag = optimizer.compile(
    rag,
    trainset=trainset,
    valset=valset,
)

行	やってること	なぜそうする
①	metric は logger ラップ版を渡す	metric 評価の入出力を全部 logger に流す。GEPA がどの feedback で何を学んだか後追い可能
②	`num_threads=4` で並列評価	trainset minibatch を 4 並列で実行 → 1 イテレーションあたりの wall time を短縮
③	`reflection_minibatch_size=3` で 3 サンプル分の feedback を 1 reflection に渡す	1 サンプルだけだと reflection LM が過剰反応、多すぎると context が冗長。経験的に 3-5 がスイートスポット
④	`reflection_lm` は temperature=1.0 の smart LM	temperature 高めが命。reflection LM が同じ指示文ばかり提案しないように多様性を出す
⑤	`candidate_selection_strategy="pareto"`	「EM 高い & recall 高い」など複数指標で支配されない候補を残す。単一スコアで選ぶより過剰適合を避けやすい

③ Reflection LM の temperature 設定 (`rag_optimization_gepa.py:256-258`)¶

reflection_lm = configure_lm(SMART_MODEL, temperature=1.0, max_tokens=8192)   # ①
fast_lm = configure_lm(FAST_MODEL, temperature=0.0, max_tokens=4096)          # ②

行	やってること	なぜそうする
①	reflection LM は temperature=1.0, max_tokens=8192	feedback を読んで指示文書き換え案を生成する用。創造性が必要なので高温度、指示文が長くなる可能性に備えて max_tokens 大きめ
②	実行用 fast LM は temperature=0	推論再現性を確保。RAG の応答にバラツキがあると metric が安定しない

④ Tee で stdout を 2 系統保存 (`rag_optimization_gepa.py:23-40, 191-198`)¶

class Tee:
    def __init__(self, file_path, original_stdout):
        self.file = open(file_path, 'w', encoding='utf-8')
        self.stdout = original_stdout

    def write(self, message):
        self.stdout.write(message)                                            # ①
        self.file.write(message)
        self.file.flush()                                                     # ②

    def flush(self):
        self.stdout.flush()
        self.file.flush()

    def close(self):
        self.file.close()

# 使用側
original_stdout = sys.stdout
tee = Tee(stdout_path, original_stdout)
sys.stdout = tee                                                              # ③

行	やってること	なぜそうする
①	console と file 両方に write	画面で進捗を見つつ、後でファイル全体を grep できる
②	毎回 `flush()`	長時間実行なので途中で kill されてもログが残る
③	`sys.stdout = tee` で完全に置換	DSPy 内部の `print()` も全部キャプチャ。`logging` だけだとサードパーティ print を拾えない

学んだこと（要点）¶

GEPA は MIPROv2 と双子。同じ DSPy Module を同じ問題で最適化するが、学習信号がスカラー（MIPROv2）か言語（GEPA）かで性格が違う
テキスト feedback の質が GEPA の収束速度を決める。「✗ 不正解」だけより「✗ 不正解: 期待=X, 実際=Y」、さらに「クエリ改善を検討: '...'」のように 具体的なヒントを入れると reflection LM の改稿精度が上がる
pareto 戦略は過剰適合に強い。単一 score で best を選ぶと train データに最適化されすぎる。複数指標（answer EM と retrieval recall）の Pareto front を維持すると汎化しやすい
reflection_lm の temperature=1.0 が定石。低いと毎回似たような改稿しか出ない → 探索が頭打ち
reflection_minibatch_size=3 は経験的スイートスポット。1 だと過剰反応、10 だと context 冗長
Tee + logger の二重ロギングは GEPA で必須レベル。reflection LM がどの feedback を読んでどう改稿したかを後追いできる
train/val を 50/50 に分けるのは GEPA 特有。MIPROv2 (26回) は 30/70 だったが、GEPA は val も探索ループで使うため val を厚めにする
同じ RAG モジュール + 同じデータで MIPROv2 と GEPA を比較できるのがこのリポジトリ最大の学習価値。自分の問題で両方試して、どちらが速く・高く・安定して収束するかを実測できる
GEPA で得られる artifact JSON は MIPROv2 のそれと同じ形式（指示文 + few-shot demo）。optimizer は違っても保存形式は統一されている

拡張アイデア¶

MIPROv2 と GEPA の同条件比較 — 同じ seed・同じデータ・同じ予算 (auto="medium") で両方を回し、EM スコア vs LLM 呼び出し回数のグラフを作る。サンプル効率の差が見える
feedback の表現を変えて A/B — 「✗/△/✓ + 数値」 vs 「自然言語のみ」 vs 「JSON 構造化 feedback」で reflection LM の挙動がどう変わるかを実験
pred_name 別 feedback — 現状は [rewrite] [generate] をラベルとして付けるだけだが、rewrite には「検索クエリの質」のフィードバックだけ、generate には「回答スタイル」のフィードバックだけを渡すように分岐すると、各 Predictor の最適化が直交化する
auto="heavy" で限界実験 — medium 83.3% を heavy でどこまで伸ばせるか。reflection LM 呼び出しが 1000 回級になるのでコスト要確認
reflection_lm を Claude Opus に — 同じ DSPy なので model="anthropic/claude-opus-4-7" に変えるだけで試せる。reflection の質が劇的に変わる可能性
メトリクスに「コーパス取得文書の多様性」を追加 — len(set([p.split('\n')[0] for p in pred.retrieved_passages])) でタイトル多様性を測る。同じトピックばかり引いていないかをチェック
学習過程の可視化 — track_stats=True で得られる統計を matplotlib でプロット。「世代ごとの best score 推移」「候補プールの Pareto frontier の推移」を見るとデバッグが格段に楽

現代版に移植するなら¶

1. API キー設定は `.env.op` + `op run` に切り替える（CLAUDE.md ルール 8）¶

連載 README は cp .env.sample .env 方式だが、グローバルルールに反するので使用禁止。代わりに:

# 27/.env.op
PROVIDER_NAME=openai
OPENAI_API_KEY=op://Personal/openai-api-key/credential
SMART_MODEL=gpt-4.1
FAST_MODEL=gpt-4.1-nano
EMBEDDING_MODEL=text-embedding-3-small

起動:

op run --env-file=.env.op -- uv run python rag_optimization_gepa.py
op run --env-file=.env.op -- uv run python rag_evaluation.py

Azure 版なら AZURE_OPENAI_API_KEY=op://Personal/azure-openai/credential に置換。

2. `rag_evaluation.py` の import 不具合修正¶

現状:

from rag_optimization import OPTIMIZED_MODEL_LATEST  # ← 存在しないファイル

修正案 1（軽量）: 文字列リテラルで直書き

GEPA_OPTIMIZED_MODEL_LATEST = "artifact/rag_gepa_optimized_latest.json"

修正案 2（import 整理）:

from rag_optimization_gepa import GEPA_OPTIMIZED_MODEL_LATEST

リポジトリのコードは rag_optimization.py が存在しない（26回からのコピペ間違い）ので修正必須。

3. `gpt-4.1` / `gpt-4.1-nano` の見直し¶

26回と同じく、2026 年現在は gpt-5 / gpt-5-nano 系も選択肢。reflection_lm は smart LM の質が露骨に効くので、SMART_MODEL=gpt-5 に上げる効果が大きい。

4. reflection_lm を Claude Opus に差し替え¶

DSPy は LiteLLM 経由で provider 中立。reflection だけ Anthropic にする:

reflection_lm = dspy.LM(
    model="anthropic/claude-opus-4-7",
    api_key=os.environ["ANTHROPIC_API_KEY"],
    temperature=1.0,
    max_tokens=8192,
)

Claude Opus は reflective reasoning が強いので、GEPA の reflection 用には実験する価値あり。

5. logger を rich に統合¶

logging.StreamHandler を RichHandler に置き換えると、長時間実行ログが色付き + 階層表示で読みやすくなる:

from rich.logging import RichHandler
console_handler = RichHandler(rich_tracebacks=True)

6. `track_stats=True` の保存¶

GEPA は track_stats=True で内部統計を保持するが、本サンプルではその統計を JSON ダンプしていない。optimized_rag.detailed_results や optimizer.history を追加で保存すると、後で世代ごとのスコア推移を分析できる。

既知の不具合・注意点¶

rag_evaluation.py の import が壊れている: from rag_optimization import OPTIMIZED_MODEL_LATEST だが rag_optimization.py は 27/ に存在しない（26/ にしかない）。rag_optimization_gepa.py から GEPA_OPTIMIZED_MODEL_LATEST を import するか、直書きする必要あり。README で「すぐに評価を試せます」と言っているが、実は壊れている
Tee クラスが context manager になっていない: __enter__ / __exit__ がなく with Tee(...) as tee: できない。例外発生時に cleanup_logging の finally で close されるが、with のほうが ergonomic
Tee.write が flush を毎回呼ぶ: 大量 print 時に I/O ボトルネック。バッファリングして N 行に 1 回 flush するほうが速い（ただし途中 kill 耐性とトレードオフ）
logger.handlers[:] のループで close: cleanup_logging で handler を全て close している。GEPA 失敗時に途中ログが flush されない可能性は低いが、finally で再度 logger を取り直すと安全
reflection_lm の cost が大きい: auto="medium" でも reflection LM (smart) の呼び出しが 100 回級。1 回の最適化で $5-15 程度は覚悟
os.symlink は Windows で要管理者権限: 26回と同じ問題
JQaRA の answers 配列の最初しか使わない: 26回と同じ問題。alias を活かすと EM が改善する可能性
pred_name フィードバックが平凡: [rewrite] [generate] を末尾につけているだけで、各 Predict の改善に特化した feedback になっていない。本気で性能を上げるなら if pred_name == "rewrite": ... で feedback 文を切り替えるべき
logger と Tee の重複: gepa_metric_with_feedback_logged の中で logger.info(...) していて、Tee 経由でも一部の print が log ファイルに行く。log ファイルが 2 系統あり、内容が部分的に重複する

記事参照¶

Software Design 2025 年 12 月号連載第27回「DSPy + GEPA」
関連: 第26回 STUDY_NOTES — 同じ RAG パイプラインを MIPROv2 で最適化。直接比較に最適
関連: 第25回 STUDY_NOTES — DSPy + MIPROv2 入門。Predict 1 個での最適化
Obsidian ノート: ~/yamamoto_obsidian/05_personal/05-4_bookshelf/software_design_2025_12.md
GEPA 論文: "GEPA: Reflective Prompt Evolution Can Outperform Reinforcement Learning"（2024 年）
公式 DSPy GEPA ドキュメント: https://dspy.ai/api/optimizers/GEPA/
JQaRA データセット: https://huggingface.co/datasets/hotchpotch/JQaRA

作成: 2026-05-17 ／最終更新: 2026-06-10

STUDY NOTES

第27回: DSPy + GEPA で日本語 RAG を最適化 — MIPROv2 との対比で理解する「テキストフィードバック進化」¶

全体像¶

使用ライブラリ・原理¶

dspy.GEPA(metric, reflection_lm, auto, ...) — 進化的・反射的最適化¶

gepa_metric_with_feedback — GEPA 専用 metric signature¶

pred_name / pred_trace — Predictor 単位のフィードバック¶

Tee クラスで stdout を 2 系統に分岐¶