dspy-advanced-workflow

DSPy 高度なワークフロー（2026）

このスキルでは、自然言語によるタスクの説明を、最適化済みで保存され、デプロイ可能な DSPy プログラムに変換する7段階のループを実行します。該当するステップを順番に実行してください。プロトタイプでは、ベースラインを検証した時点で停止します。オプティマイザーの実行には、適切に承認された予算と必要性を示す根拠が必要です。ローカルのアーティファクトをエクスポートしても、デプロイが許可されたことにはなりません。

7つのステップ

1. 仕様

ユーザーのタスクを1文で言い換えます。入力、出力、重要な品質指標、および制約（レイテンシ、コスト、ツールへのアクセス、コンテキストサイズ）を特定します。予測器の形式を選択します。

タスクの形式	予測器
単一ステップの構造化 I/O	dspy.Predict / dspy.ChainOfThought
ツール使用 / 複数ステップ	dspy.ReAct
コード実行	dspy.ProgramOfThought
長いコンテキスト / コードベース	dspy.RLM → dspy-rlm-module

2. プログラム

dspy-fundamentals に従って、型付きの dspy.Signature と dspy.Module サブクラスを記述します。プロンプトをハードコードしないでください。GEPA が対象にできるよう、予測器には名前を付けてください。

3. データ

dspy.Example(...).with_inputs(...) を使って、trainset と別個の valset を作成します。GEPA では、trainset のサイズを最大化し、検証セットは下流の動作を表現するのに十分な大きさに保ちます。保留された testset の結果は最後にのみ報告します。dspy-evaluation-harness を参照してください。

4. リッチメトリック

rich_metric(gold, pred, trace=None, pred_name=None, pred_trace=None) を記述し、dspy.Prediction(score=0..1, feedback="natural-language critique") を返すようにします。フィードバックは非常に重要です。GEPA のリフレクション LM が学習するのはこれです。同じフィールドを持つ dict を使うと dspy.Evaluate がクラッシュします。正しく集計できるのは dspy.Prediction のみです。dspy-evaluation-harness を参照してください。

5. ベースライン

evaluator = dspy.Evaluate(devset=valset, metric=rich_metric,
                          num_threads=8, display_progress=True,
                          provide_traceback=True,
                          save_as_json="runs/baseline.json")
baseline = evaluator(program)
print("Baseline:", baseline.score)

6. GEPA で最適化

reflection_lm = dspy.LM("openai/gpt-5", temperature=1.0, max_tokens=32000)
optimizer = dspy.GEPA(
    metric=rich_metric,
    auto="medium",
    reflection_lm=reflection_lm,
    candidate_selection_strategy="pareto",
    track_stats=True,
    track_best_outputs=True,
    log_dir="./gepa_logs",
    num_threads=8,
    seed=0,
)
optimized = optimizer.compile(student=program, trainset=trainset, valset=valset)
print("Optimized:", evaluator(optimized).score)

まず auto="light" を実行して健全性を確認し、最終実行では auto="medium" / "heavy" に進みます。dspy-gepa-optimizer を参照してください。

意図的な複数段階のコンパイルループが必要な場合、DSPy 3.2.x では、ベースラインの GEPA 設定を整えた後に名前付きオプティマイザーを連結する dspy.BetterTogether(metric=..., bootstrap=..., gepa=...) も公開されています。

7. エクスポートとデプロイ

optimized.save("artifacts/program.json", save_program=False)     # state, portable
# or for full deployment artifact:
optimized.save("artifacts/program_dir/", save_program=True)

デプロイ：

• dspy.load("artifacts/program_dir/") で読み込むか、再構築して .load("program.json") を実行します。
• FastAPI / CLI でラップします。
• コストとレイテンシを可観測にするため、track_usage=True を有効にします。
• CI で MLflow（mlflow.dspy.autolog()）または W&B を使ってログを記録します。
• 保存済みプログラムに対して evaluator を実行し、しきい値を下回った場合に CI を失敗させるオフライン回帰テストを維持します。

完全なオーケストレーションテンプレート

"""DSPy end-to-end pipeline — spec → optimize → deploy."""

import dspy
from pathlib import Path

# ----- 1–2. Spec & program (dspy-fundamentals) -----
class MyTask(dspy.Signature):
    """<one-line instruction from the spec>."""
    input_field: str = dspy.InputField()
    output_field: str = dspy.OutputField()

class MyProgram(dspy.Module):
    def __init__(self):
        super().__init__()
        self.step = dspy.ChainOfThought(MyTask)
    def forward(self, **kw):
        return self.step(**kw)

# ----- 3. Data (dspy-evaluation-harness) -----
trainset = [...]   # list[dspy.Example(...).with_inputs(...)]
valset   = [...]

# ----- 4. Rich metric (dspy-evaluation-harness) -----
def rich_metric(gold, pred, trace=None, pred_name=None, pred_trace=None):
    score = ...          # compute 0..1
    feedback = ...       # detailed critique
    return dspy.Prediction(score=score, feedback=feedback)  # NOT a dict

# ----- 5. Baseline -----
dspy.configure(lm=dspy.LM("openai/gpt-4o"), track_usage=True)
evaluator = dspy.Evaluate(devset=valset, metric=rich_metric, num_threads=8,
                          display_progress=True, provide_traceback=True,
                          save_as_json="runs/baseline.json")
program = MyProgram()
print("Baseline:", evaluator(program).score)

# ----- 6. GEPA optimize (dspy-gepa-optimizer) -----
optimizer = dspy.GEPA(
    metric=rich_metric,
    auto="medium",
    reflection_lm=dspy.LM("openai/gpt-5", temperature=1.0, max_tokens=32000),
    candidate_selection_strategy="pareto",
    track_stats=True, track_best_outputs=True,
    log_dir="./gepa_logs", num_threads=8, seed=0,
)
optimized = optimizer.compile(student=program, trainset=trainset, valset=valset)
print("Optimized:", evaluator(optimized).score)

# ----- 7. Export (dspy-fundamentals) -----
Path("artifacts").mkdir(exist_ok=True)
optimized.save("artifacts/program.json", save_program=False)

ガードレール

• 最適化の前にステップ4でメトリックを定義し、タスクに適した有益なフィードバックを使います。
• 最適化の前に必ずベースラインを確立します。ベースラインがなければ、主張もできません。
• 監査できるよう、最適化前と最適化後の両方のメトリックを JSON に保存します。
• 保留テストのスコアが低下した場合は、その結果を保持し、原因を決めつけずにトレーニングと検証のエビデンスを使って診断します。テスト結果に基づく変更を行った後は、新しく未使用の最終ホールドアウトを使うか、それ以降の結果を探索的なものとして明記します。元のテストセットに対して繰り返し調整しないでください。
• 複数段階の再コンパイルを行う前に、module._compiled = True を設定して最適化済みプログラムを固定します。

実行可能なスキャフォールド → example_pipeline.py