STUDY NOTES

第26回: DSPy + MIPROv2 で日本語 RAG パイプライン（JQaRA）を自動最適化 — EM 6.7% → 83.3%¶

第25回の DSPy 入門（チャットボット）の応用編。今度は RAG パイプラインを MIPROv2 で最適化する。題材は JQaRA（Japanese Question Answering with Retrieval Augmentation）データセット。

第25回との差分: 25回は「Predict 1 個の Module」を最適化した。26回は 「クエリリライト Predict → Retrieve → 回答生成 Predict」という 2 段の Predict を含む Moduleを最適化する。MIPROv2 は 両方の Predict を同時に最適化する（指示文と few-shot demo がそれぞれ独立に調整される）。

ベースライン（プロンプトを書かず DSPy のデフォルトを使う）と最適化済みモデルで Exact Match 精度がどう変わるかが見どころ。同梱の artifact/rag_optimized_latest.json で EM 6.7% → 83.3%（+76.7%） という劇的改善が示されている。

概念	DSPy での実装
クエリ最適化	`RewriteQuery` Signature + `dspy.Predict`
Retrieve	`dspy.retrievers.Embeddings(corpus, embedder, k)`
回答生成	`GenerateAnswer` Signature + `dspy.Predict`
評価	EM（生成回答 == 正解）+ 検索 recall の複合メトリクス
最適化器	`MIPROv2(auto="medium")`
OpenAI / Azure 切替	`config.py` の `PROVIDER_NAME` 環境変数
Embedding コスト削減	`embeddings_cache.py` で pickle キャッシュ

全体像¶

26/
├── config.py                       ← OpenAI / Azure 切替 + Smart/Fast LM 設定
├── dataset_loader.py               ← JQaRA を読み込み、正例/負例を分離した Example へ
├── embeddings_cache.py             ← コーパスハッシュをキーに pickle キャッシュ
├── rag_module.py                   ← ★ RewriteQuery → Retrieve → GenerateAnswer の Module
├── evaluator.py                    ← 評価用 metric（EM のみ）+ 最適化用 metric（EM + recall）
├── rag_optimization.py             ← MIPROv2 で最適化 → JSON 保存 → シンボリックリンク更新
├── rag_evaluation.py               ← ベースライン vs 最適化版を比較
└── artifact/
    ├── rag_optimized_latest.json   → rag_optimized_<timestamp>_em<NNN>.json
    └── embeddings_cache/           ← Embedding ベクトルの pickle キャッシュ

最適化フロー:

flowchart TD
    HF[hotchpotch/JQaRA<br/>dev 50問 + test 30問] --> Loader[dataset_loader<br/>positives/negatives 分離]
    Loader --> Examples[DSPy Example<br/>question, answer, positives, negatives]
    Loader --> Corpus[corpus_texts<br/>正例+負例をシャッフル]

    Examples -->|shuffle + 30%-split| Trainset[trainset 15問]
    Examples -->|残り 70%| Valset[valset 35問]

    Corpus --> Cache[embeddings_cache<br/>md5 ハッシュをキーに pickle]
    Cache --> Retriever[dspy.retrievers.Embeddings<br/>k=10]

    Trainset --> MIP[MIPROv2.compile<br/>auto=medium<br/>minibatch=True]
    Valset --> MIP
    Metric[rag_comprehensive_metric<br/>0.5×EM + 0.5×recall] --> MIP
    Retriever --> MIP
    SmartLM[smart_lm: gpt-4.1<br/>prompt_model] --> MIP
    FastLM[fast_lm: gpt-4.1-nano<br/>RAG 実行用] --> MIP

    MIP --> Opt[最適化済み RAGQA Module]

    Opt --> Save[artifact/rag_optimized_<ts>_em<NN>.json]
    Save -->|symlink| Latest[artifact/rag_optimized_latest.json]

推論時の RAGQA 内部フロー:

sequenceDiagram
    participant User
    participant RAG as RAGQA
    participant RW as Predict(RewriteQuery)<br/>(最適化済み prompt)
    participant Ret as dspy.settings.rm<br/>(Embeddings retriever)
    participant Gen as Predict(GenerateAnswer)<br/>(最適化済み prompt)

    User->>RAG: forward(question="...")
    RAG->>RW: rewrite(question=...) → rewritten_query
    RW-->>RAG: 検索向けに整形されたクエリ
    RAG->>Ret: rm(rewritten_query) → top-k passages
    Ret-->>RAG: List[str] (k=10)
    RAG->>Gen: generate(context=..., question=...) → answer
    Gen-->>RAG: 回答テキスト
    RAG-->>User: Prediction(answer, retrieved_passages, rewritten_query)

重要な設計判断: ベースライン (RAGQA() のまま) でも構造は同じ。違うのは「rewrite と generate の指示文 + few-shot demo がMIPROv2 が探した最適なものになっている」だけ。LLM も retriever も変えていない。プロンプトの質だけで EM が 6.7% → 83.3% に上がる。

使用ライブラリ・原理¶

`dspy.retrievers.Embeddings` — DSPy 標準の埋め込み検索器¶

retriever = dspy.retrievers.Embeddings(
    embedder=embedder,
    corpus=corpus_texts,
    k=10,
)
result = retriever(query)  # result.passages: List[str]

ポイント:

コンストラクタ時に corpus 全文の embedding を計算して内部に保持（高コスト）
検索時はクエリも embed → コサイン類似度 top-k を返す
dspy.settings.rm に set すると dspy.Module から dspy.settings.rm(query) で呼べる（global retriever 設定）

`dspy.Embedder` — OpenAI / Azure 互換の embedding ラッパ¶

embedder = dspy.Embedder(model="openai/text-embedding-3-small", api_key=...)
# or
embedder = dspy.Embedder(model="azure/text-embedding-3-small", api_base=..., api_key=..., api_version=...)

LiteLLM 互換 prefix（openai/ / azure/ 等）で provider を切り替えられる。provider 切替が model= 文字列だけで完結するのが DSPy 3.x の良いところ。

Embedding pickle キャッシュ戦略¶

corpus_content = "".join(sorted(corpus_texts))
corpus_hash = hashlib.md5(corpus_content.encode()).hexdigest()[:12]
cache_file = cache_path / f"embeddings_{corpus_hash}.pkl"

要点:

コーパス全文をソート → MD5 ハッシュでキャッシュキー化
ソートする理由は「同じ文書集合なら順序が違ってもキャッシュ HIT させたい」から
ハッシュは MD5 12 桁で衝突無視（数十万 corpus でも実用上問題なし）
dspy.retrievers.Embeddings を空 corpus で作るのではなく、インスタンス化後に .corpus_embeddings 属性を上書きして計算をスキップ

MIPROv2 の `minibatch=True` モード¶

optimizer.compile(rag, trainset=trainset, valset=valset, minibatch=True)

minibatch=True の効果:

評価のたびに valset 全部を回さず、mini-batch で抜き打ち評価 → 探索を加速
ベイズ最適化の各 trial 単価が下がる
最後の選定だけ valset 全部で評価
auto="medium" で typical 6-10 イテレーション = 100-300 LLM 呼び出し（minibatch なしより 3-5 倍速い）

Comprehensive metric vs EM metric の分離¶

# 最適化用（探索を進めやすくする）
def rag_comprehensive_metric(gold, pred, trace=None):
    answer_match = float(pred.answer.strip() == gold.answer.strip())
    retrieved = set(pred.retrieved_passages)
    positives = set(gold.positives) if gold.positives else set()
    max_positives = min(len(positives), RETRIEVAL_K) if positives else 0
    positive_ratio = len(retrieved & positives) / max_positives if max_positives else 0.0
    return 0.5 * answer_match + 0.5 * positive_ratio

# 最終評価用（厳しい）
def exact_match_metric(gold, pred, trace=None):
    return float(pred.answer.strip() == gold.answer.strip())

設計意図:

最適化中は「検索も回答も両方良い」を勾配にすることで、初期段階で EM=0% でも検索が改善すれば部分的に高スコアになる → 学習が進む
最終評価は EM 一本で「ユーザから見た正答率」を測る
これは ML 一般の「学習用 loss と評価 metric を分ける」と同じ発想

ファイル別の役割¶

ファイル	役割
`config.py`	`PROVIDER_NAME=openai/azure` で切替、`SMART_MODEL` / `FAST_MODEL` / `EMBEDDING_MODEL` の env → DSPy LM/Embedder 構築ヘルパ
`dataset_loader.py`	HF datasets で JQaRA をロード → `q_id` でグループ集約 → `label==1` で positives / それ以外 negatives に分離 → Example 化。corpus_texts に全 passages を混在させる
`embeddings_cache.py`	MD5 ハッシュをキーに pickle キャッシュ。corpus が同じなら embedding を再計算しない
`rag_module.py`	中核。`RewriteQuery` + `GenerateAnswer` の 2 Signature + `RAGQA(dspy.Module)` で forward
`evaluator.py`	最適化用 metric（comprehensive）と評価用 metric（EM）を分離。`dspy.Evaluate(num_threads=4)` で並列評価
`rag_optimization.py`	dev 50問→train/val 30/35 分割 + test 30問 → MIPROv2 で最適化 → JSON 保存 + symlink 更新
`rag_evaluation.py`	baseline vs optimized を test 30問で比較。`OPTIMIZED_MODEL_LATEST` をロード

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

① RAGQA Module の `forward` (`rag_module.py:22-49`)¶

class RAGQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.rewrite = dspy.Predict(RewriteQuery)                             # ①
        self.generate = dspy.Predict(GenerateAnswer)

    def forward(self, question: str):
        rewritten = self.rewrite(question=question).rewritten_query           # ②

        result = dspy.settings.rm(rewritten)                                  # ③
        passages = result.passages if hasattr(result, 'passages') else []

        context = "\n".join(passages) if passages else ""                     # ④

        answer = self.generate(context=context, question=question).answer

        return dspy.Prediction(                                               # ⑤
            answer=answer,
            retrieved_passages=passages,
            rewritten_query=rewritten
        )

行	やってること	なぜそうする
①	`Predict` を 2 つ attribute として保持	MIPROv2 はこれらを sub-Predict と認識し、それぞれの prompt と demo を独立に最適化する。`dspy.Module` の attribute traversal が PyTorch の `nn.Module` と同型
②	クエリリライト	「質問文そのまま」より「埋め込み検索向けに整形」したクエリのほうが recall が上がる古典テク。MIPROv2 はこの Predict 内部で「リライトの仕方」を最適化する
③	`dspy.settings.rm(rewritten)` で global retriever 呼び出し	`rag_optimization.py` で `dspy.configure(rm=retriever)` してあるので、ここで呼べる。DSPy の global state パターン（賛否あり）
④	passages を `\n` 連結して context に	検索結果をそのままプロンプトに詰める素朴な方式。passage 間に区切り文字 `\n---\n` を入れる選択肢もあるが、簡素優先
⑤	`Prediction(answer, retrieved_passages, ...)` を返す	metric が `pred.retrieved_passages` を見て recall を計算するので、必ず返す必要あり

② MIPROv2 への compile (`rag_optimization.py:73-87`)¶

optimizer = dspy.MIPROv2(
    metric=rag_comprehensive_metric,                                          # ①
    prompt_model=smart_lm,                                                    # ②
    auto="medium",                                                            # ③
)

optimized_rag = optimizer.compile(
    rag,
    trainset=trainset,
    valset=valset,                                                            # ④
    minibatch=True,                                                           # ⑤
)

行	やってること	なぜそうする
①	最適化中は comprehensive metric（EM + recall）	学習信号を強くするため。EM だけだと初期段階で全部 0 になり gradient が消える
②	`prompt_model=smart_lm`（gpt-4.1）	プロンプト改良案を提案する LM。fast_lm (nano) では指示文の質が出ないので smart 必須
③	`auto="medium"`	25回の `light` より深い最適化。100-300 LLM 呼び出し / 1 回。$5-15 程度の予算
④	`valset=valset` を明示	MIPROv2 が候補を比較するときの基準セット。train は demo 抽出と各 trial の評価に、val は最終比較に使われる
⑤	`minibatch=True`	valset の mini-batch だけで trial を比較 → 探索を加速

③ JQaRA データセットの集約処理 (`dataset_loader.py:51-76`)¶

def aggregate_group(group):
    question = group['question'].iloc[0]
    answers = group['answers'].iloc[0]
    answer = answers[0] if answers else ""                                    # ①

    mask_positive = group['label'] == 1                                       # ②
    positives = group.loc[mask_positive, 'passage'].tolist()
    negatives = group.loc[~mask_positive, 'passage'].tolist()

    np.random.shuffle(positives)                                              # ③
    np.random.shuffle(negatives)

    return pd.Series({
        'question': question,
        'answer': answer,
        'positives': positives,
        'negatives': negatives,
        ...
    })

result = df.groupby('q_id', as_index=False).apply(
    aggregate_group, include_groups=False
)

行	やってること	なぜそうする
①	answers は配列だが最初の 1 つだけ採用	JQaRA の `answers` は複数候補（alias）。EM で評価するため代表 1 つに絞る。本来は「いずれかと match」のほうが正しいが簡素化
②	`label == 1` を positives、それ以外を negatives	JQaRA の各 passage には正解を含むかの label が付いている。最適化用 metric はこれで recall を測る
③	positives / negatives をシャッフル	順序バイアスを除去。あとで `corpus_texts.extend(positives + negatives)` するときに位置情報を漏らさない

④ Embedding キャッシュの hash 戦略 (`embeddings_cache.py:36-39`)¶

corpus_content = "".join(sorted(corpus_texts))                                # ①
corpus_hash = hashlib.md5(corpus_content.encode()).hexdigest()[:12]
cache_file = cache_path / f"embeddings_{corpus_hash}.pkl"

行	やってること	なぜそうする
①	corpus をソートしてから連結	同じ文書集合 = 同じハッシュにしたい。`shuffle` で順序が違っても同じキャッシュをヒットさせるため。`sorted()` でメモリ確保が一時的に倍になる点は注意

学んだこと（要点）¶

DSPy で RAG を組むと、各サブ Predict が独立に最適化される。検索クエリリライトと回答生成の prompt が別々の最適解になる
最適化用 metric と評価用 metric を分けるのは ML の基本。RAG は「EM だけ」だと初期スコアが 0 でつぶれるので、recall を半分混ぜると勾配が立つ
EM 6.7% → 83.3% という改善幅は説得力がある。プロンプトを職人技で書くより、MIPROv2 に任せたほうが日本語 RAG の最終精度は出る
Embedding キャッシュは絶対に必要。corpus が変わらない開発ループ中、毎回 OpenAI に embedding を投げると数千円コースになる
MIPROv2(auto="medium") の出力は 1 つの JSON ファイル。git にコミットして CI で性能リグレッション検知できる
dspy.configure(rm=...) で global retrieverを設定するパターンは「グラフの中から dspy.settings.rm で呼べる」便利さと「test 並列化で衝突する」リスクが両立。dspy.context(rm=...) で context 管理するのが正攻法
JQaRA は label 付き = 検索 recall を測れる良いデータセット。1 質問 50-100 passages の中に正例 1-5 個が混じる構造で、retrieve + answer の両方を評価可能
provider 切替を env だけで完結させる config.py のパターンは production 移植時に便利。PROVIDER_NAME=openai/azure の 1 行で全 LLM/embedder の prefix を変える
minibatch=True の効果は劇的。False で 1 時間以上かかる最適化が 10-15 分で終わる（品質はわずかに低下）

拡張アイデア¶

Cohere Rerank との組み合わせ — dspy.retrievers.Embeddings で top-30 取って、Cohere rerank-v3 で top-10 に絞る 2 段検索に変える。recall を保ちつつ precision を上げる
BM25 とのハイブリッド検索 — rank_bm25 で BM25 score を出し、cosine sim と Reciprocal Rank Fusion でハイブリッド化。日本語 RAG では BM25 が驚くほど強い
auto="heavy" で精度を限界まで — medium 83.3% を heavy（500-1000 LLM 呼び出し、$50 級）でどこまで伸ばせるか
chunk size の感度分析 — 現状 1 passage = 1 chunk だが、複数の隣接 passage を結合した chunk で再評価
GEPA への移植 — 第27回で扱う GEPA optimizer に差し替えて、MIPROv2 との比較。GEPA は LLM-as-Judge を使った reflective optimization で MIPROv2 より少サンプルで効くケースがある
複数 answer alias の正規化 — JQaRA は answers: List[str] で alias を持つ。answer in gold.answers を metric に使えば EM がもっと改善する可能性
質問を category 別に分割して最適化 — JQaRA の q_id には category 情報あり。category 別に specialized model を作って ensemble する

現代版に移植するなら¶

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

# 26/.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.py
op run --env-file=.env.op -- uv run python rag_evaluation.py

Azure 版なら AZURE_OPENAI_API_KEY=op://Personal/azure-openai/credential 等の参照に置き換え。

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

2026 年現在は gpt-5 / gpt-5-nano 系も選択肢。SMART_MODEL を上げる効果は最適化品質に直結（指示文の質が変わる）、FAST_MODEL を上げる効果は実行時精度に直結。両方上げると最大効果だが、コストも比例。

3. `dataset_loader.py:43` の `split=f"{dataset_split}[:{num_records}]"` パターン¶

これは HF datasets の slice syntax だが、評価で常に同じ先頭 N 件を取るので bias がかかる。shuffle().select(range(N)) のほうが公平。再現性は random_seed で確保する。

4. EM だけの評価は厳しすぎ¶

exact_match_metric は文字列完全一致なので「東京都」vs「東京」で 0 点。正解 alias を持つ JQaRA の特性を活かして:

def exact_match_metric(gold, pred, trace=None):
    pred_answer = pred.answer.strip()
    # 元の answers 配列をいずれかと match で OK にする
    if hasattr(gold, 'answers'):
        return float(any(pred_answer == a.strip() for a in gold.answers))
    return float(pred_answer == gold.answer.strip())

ただし dataset_loader 側で answers を Example に渡していないので、そちらも修正が必要。

5. `print` → `logging`¶

dataset_loader.py / embeddings_cache.py 等の進捗 print を logging に。CI 実行時の log level 制御が効く。

6. `numpy.random.seed` から `numpy.random.Generator` へ¶

dataset_loader.py:33 の np.random.seed(random_seed) は process global state を汚染する。rng = np.random.default_rng(random_seed) + rng.shuffle(positives) のほうが local。

既知の不具合・注意点¶

JQaRA answers の最初の 1 つしか使わない: 上記の通り。alias を活かせていない
rag_module.py:32 の dspy.settings.rm: global state。test 並列化（num_threads=4）と相性が悪い。並列 trial が同じ rm を共有することで微妙な race condition の余地
embeddings_cache.py の corpus ソート時メモリ: 数万 passages で "".join(sorted(corpus_texts)) すると一時的に corpus サイズ × 2 のメモリ。大規模 corpus では hashlib に逐次 feed するほうが堅い
os.symlink は Windows で要権限: rag_optimization.py:110 の os.symlink(model_filename, OPTIMIZED_MODEL_LATEST) は Windows で開発者モード必須。shutil.copy のほうが portable
numpy.random のシード固定が dataset_loader 側のみ: rag_optimization.py:32 で random.seed(seed) してから random.shuffle(examples) しているが、numpy.random.shuffle を使う dataset_loader 側は seed 固定済みなので OK。ただし import 順序や reload で混乱する余地あり
Train/Val 30%:70% のコメントが README と食い違う: README は「Train/Val 50:50」と書いているが、コード (rag_optimization.py:34) は split = int(len(examples) * 0.3) で 30%:70% 分割。実態はトレーニング15問・検証35問
exact_match_metric の正規化なし: 全角半角・空白・句読点の差で false negative。unicodedata.normalize("NFKC", ...) を入れると数 % 改善する可能性
OPTIMIZED_MODEL_LATEST の symlink が壊れたまま load しようとすると失敗: rag_evaluation.py が optimized.load(OPTIMIZED_MODEL_LATEST) するので、symlink が dangling だと FileNotFoundError。最適化が中途半端に失敗すると起きる

記事参照¶

Software Design 2025 年 11 月号（推定）連載第26回「DSPy + MIPROv2 で日本語 RAG 最適化」
関連: 第25回 STUDY_NOTES — DSPy + MIPROv2 入門。Predict 1 個での最適化
関連: 第27回 STUDY_NOTES — 同じ RAG パイプラインを GEPA で最適化（MIPROv2 との比較）
JQaRA データセット: https://huggingface.co/datasets/hotchpotch/JQaRA
公式 MIPROv2 ドキュメント: https://dspy.ai/api/optimizers/MIPROv2/
公式 DSPy retrievers: https://dspy.ai/api/retrievers/

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

STUDY NOTES

第26回: DSPy + MIPROv2 で日本語 RAG パイプライン（JQaRA）を自動最適化 — EM 6.7% → 83.3%¶

全体像¶

使用ライブラリ・原理¶

dspy.retrievers.Embeddings — DSPy 標準の埋め込み検索器¶

dspy.Embedder — OpenAI / Azure 互換の embedding ラッパ¶

Embedding pickle キャッシュ戦略¶

MIPROv2 の minibatch=True モード¶