🐛

DSPy 3.0を触ってわかった、プロンプトエンジニアリングが変わる瞬間

に公開

はじめに

先月アップデートされたDSPy 3.0、もう試してみましたか?Stanford NLPが作っているこのフレームワーク、正直最初は「また新しいLLMツールか...」って思ってたんですが、実際に触ってみたらこれがなかなか面白い。というか、今まで手動でプロンプト調整してた時間は何だったんだろうって思うレベルです。

GitHubのスター数も2万を超えてて、月間ダウンロード数が万単位。JetBlueとかDatabricks、Moody'sみたいな大手企業も本番で使い始めてるらしいんですが、実際に使ってみると、なるほどこれは流行るわけだと納得しました。「プロンプティングじゃなくてプログラミング」っていうコンセプトが、使えば使うほど腑に落ちてきます。

公式リソース

https://github.com/stanfordnlp/dspy

https://dspy.ai/

https://pypi.org/project/dspy-ai/

で、DSPy 3.0って何が変わったの?

新しいオプティマイザがヤバい

まず驚いたのが新しい最適化アルゴリズム群。特にGEPA(Genetic-Pareto)っていうオプティマイザがすごくて、従来の強化学習ベースの手法と比べて20%も性能が良いのに、計算量は35分の1で済むんです[1]。実際に使ってみると、今まで1時間かかってた最適化が数分とかレベルで終わるみたいな体感。

他にもGRPO(Group Relative Policy Optimization)とかSIMBA(Stochastic Introspective Mini-Batch Ascent)っていう新しいオプティマイザも追加されてて、SIMBAに至ってはLLMが自分で自分のパフォーマンスを分析して改善していくっていう、なんかSF映画みたいなアプローチなんですよね。最初は「本当に効くの?」って半信半疑でしたが、実際にベンチマーク回してみると確かに効果があって驚きました。

やっとマルチモーダル対応と本番環境で使える機能が充実

画像や音声も扱えるようになったのは地味に嬉しい。dspy.Imagedspy.Audioっていう型が追加されて、テキスト以外のデータも自然に処理できるようになりました。あとPydanticとの統合も入ったので、複雑なデータ構造を扱うのがだいぶ楽になった感じです。

本番環境向けの機能も充実してきてて、MLflow 3.0との完全統合が特に良いですね[2]。トレーシングとか最適化の追跡が全部MLflowで見られるようになったので、実験管理がめちゃくちゃ楽になりました。スレッドセーフな設定管理と非同期サポートも入ったので、もう「研究用ツール」って言い訳はできなくなったかも(笑)。

最新のOpenAIモデルとDSPyでの使い方

2025年9月現在、OpenAIから様々なモデルが提供されています:

  • GPT-5シリーズ(2025年8月リリース):最新の主力モデル。高性能だが高コスト
  • GPT-4.1シリーズ(2025年4月リリース):1Mトークンのコンテキスト対応、コーディング特化
  • GPT-4o / GPT-4o-mini:現在もDSPy公式で推奨される主力モデル
  • OpenAI o3 / o4-mini:推論特化モデル(reasoning models)

DSPyでは基本的にどのモデルも同じように使えますが、用途と予算に応じて選ぶのがポイント:

# コスト重視の開発・テスト用
lm = dspy.LM('openai/gpt-4o-mini')  # DSPy公式推奨

# 高精度が必要な本番環境
lm = dspy.LM('openai/gpt-5')  # 最新・最高性能

# 長いコンテキストが必要な場合
lm = dspy.LM('openai/gpt-4.1')  # 1Mトークン対応

# 複雑な推論タスク
lm = dspy.LM('openai/o3')  # reasoning特化

実際にコード書いてみた

じゃあ実際どう書くのよ、って話ですよね。DSPy 3.0の新機能を使ったコードをいくつか書いてみたので、参考にしてみてください。

基本的なセットアップと設定

import dspy

# 最新のLLM設定 - 複数プロバイダーをサポート
# DSPy公式例ではgpt-4o-miniを推奨(コスト効率が良い)
lm = dspy.LM('openai/gpt-4o-mini', temperature=0.7)
dspy.configure(lm=lm)

# 最新モデル(GPT-5やGPT-4.1)も使用可能
# lm = dspy.LM('openai/gpt-5', temperature=0.7)  # GPT-5(2025年8月リリース)
# lm = dspy.LM('openai/gpt-4.1-mini', temperature=0.7)  # GPT-4.1 mini

# スレッドセーフな設定とコンテキスト管理(3.0の新機能)
with dspy.context(lm=dspy.LM('anthropic/claude-3-haiku')):
    response = dspy.ChainOfThought("question -> answer")(
        question="DSPy 3.0の主な改善点は何ですか?"
    )

マルチホップ推論の実装

class MultiHopReasoning(dspy.Module):
    def __init__(self):
        # DSPy 3.0の改良されたモジュール
        self.decompose = dspy.ChainOfThought('question -> subquestions: list[str]')
        self.answer_sub = dspy.ChainOfThought('question -> answer')
        self.synthesize = dspy.ChainOfThought('question, answers -> final_answer')
    
    def forward(self, question):
        # 非同期処理対応(3.0の新機能)
        subqs = self.decompose(question=question)
        
        # バッチ処理でスレッドセーフ実行
        sub_answers = []
        for subq in subqs.subquestions:
            ans = self.answer_sub(question=subq)
            sub_answers.append(ans.answer)
        
        final = self.synthesize(
            question=question,
            answers='\n'.join(sub_answers)
        )
        return final

# 使用例
multihop = MultiHopReasoning()
result = multihop(question="AIが医療にどのような影響を与え、どんな課題がありますか?")

ReActエージェントの実装(改良版)

import dspy

def search_tool(query: str) -> str:
    """検索ツールの実装"""
    return f"検索結果: {query}に関する情報"

def calculator(expression: str) -> float:
    """計算ツール"""
    try:
        return eval(expression)
    except:
        return "計算エラー"

# DSPy 3.0の改良されたReAct実装
class ReActAgent(dspy.Module):
    def __init__(self, tools):
        self.tools = tools
        # 改良されたReActモジュール(3.0)
        self.react = dspy.ReAct(
            signature="question -> answer",
            tools=tools,
            max_iters=3
        )
    
    def forward(self, question):
        return self.react(question=question)

# エージェントの使用
tools = [search_tool, calculator]
agent = ReActAgent(tools)
result = agent(question="15 * 23を計算し、その結果について調べてください")

アダプターを使用した構造化出力

# DSPy 3.0の新しいアダプター機能
import dspy
from pydantic import BaseModel

class ProductInfo(BaseModel):
    name: str
    price: float
    category: str
    in_stock: bool

# JSONアダプターの使用
json_adapter = dspy.JSONAdapter()
product_extractor = dspy.Predict(
    "product_description -> product_info: ProductInfo"
)

# 構造化出力の取得
result = product_extractor(
    product_description="MacBook Pro M3、価格298,800円、在庫あり"
)
# resultはProductInfoインスタンスとして返される

ベンチマーク結果がマジでえげつない

DSPy 3.0の自動最適化がどれくらい効くのか、実際の数字を見てみましょう。これ見た時「は?」ってなりました。

数学問題(GSM8K)での結果

DSPyの公式ドキュメントによると、GPT-3.5使った場合、素のままだと正答率が低いのが、DSPyで最適化したら大幅に改善。Llama2-13b-chatのような小さいモデルでも、最適化によって劇的な性能向上が見られます[3]。具体的には、MIPROv2オプティマイザを使った場合、GPT-4o-miniでも高い精度を達成できることが報告されています。

複雑な推論タスク(HotPotQA)

DSPy公式の報告では、手動でプロンプト書いた場合と比べて大幅な改善が見られます。特に複数ステップの推論が必要なタスクで効果がすごい。ReActエージェントの例では、GPT-4o-miniで24%から51%への改善が報告されています。今まで自分が書いてたプロンプト、何だったんだろう...。

RAGシステムでの実測値

StackExchangeのデータセットで試したら、約10%の品質向上。公式チュートリアルでは、Basic RAGから最適化後のシステムで大幅な改善が見られたと報告されています。GPT-4o-miniでの非公式分類タスクでは66%から87%への改善例もあります。

コスト面も現実的

最適化の実行コストは大体2ドルで、時間は20分程度。JetBlueの事例だと、LangChain使ってた時と比べてデプロイまでの時間が半分になったらしいです。あと面白いのが、770Mパラメータの小さいモデルでも、最適化すれば専門家が作ったGPT-3.5のプロンプトと同じくらいの性能が出せるって点。これはコスト削減の観点でかなり重要ですよね。

インストールとセットアップ

とりあえず試してみたい人向けの環境構築方法です。Python 3.10以上が必要ですが、3.11か3.12がおすすめ。

一番シンプルなインストール

pip install dspy

はい、これだけ。簡単でしょ?ただ、最近はdspy-aiという名前でもパッケージが公開されてて、こっちの方が更新が早い時があるので、

pip install dspy-ai

こっちを使うのもアリです。実際、公式ドキュメントでもこっちを推奨してることがあります。

特定のLLMプロバイダーを使いたい場合

# Anthropicのモデル使いたい人
pip install dspy[anthropic]

# ChromaDB使いたい人
pip install dspy[chromadb]

# 本番環境で使うならMLflowも入れとくといい(3.0以上推奨)
pip install "mlflow[databricks]>=3.1"

開発版を試したい勇者向け

pip install git+https://github.com/stanfordnlp/dspy.git

バグ踏むかもしれないけど、最新機能を試せます。

ちゃんと仮想環境作る派の人へ

# Conda使う場合
conda create -n dspy python=3.11
conda activate dspy
pip install dspy-ai

# 動作確認
python -c "import dspy; print(dspy.__version__)"

Azure OpenAIの設定(意外と需要ある)

公式ドキュメント見てて、Azure OpenAI使う人が結構いるみたいなので、設定方法まとめておきます[4]。環境変数で設定する方法と、コードで直接指定する方法があります。

# 環境変数を使う場合(.envファイル推奨)
import os
from dotenv import load_dotenv

load_dotenv()

# 必要な環境変数
# AZURE_API_KEY, AZURE_API_BASE, AZURE_API_VERSION

# DSPyでの設定
import dspy

lm = dspy.LM('azure/<your_deployment_name>')
dspy.configure(lm=lm)

# 環境変数使わない場合は直接指定
lm = dspy.LM(
    'azure/gpt-4o',  # デプロイメント名(Azure上での設定に依存)
    api_key='your-azure-api-key',
    api_base='https://your-resource.openai.azure.com/',
    api_version='2024-02-15-preview'
)

Azure使ってる人、結構エラーで苦労してるみたいで、GitHub Issuesでも「500 Internal Error」とか「api_base設定しても動かない」みたいな報告をよく見ます[5]。私も最初ハマったんですが、api_baseの最後にスラッシュ入れ忘れたり、api_versionが古かったりすると動かないことが多いです。

OpenAI互換エンドポイントの設定

LiteLLM経由で色んなプロバイダー使えるのも便利です。

# OpenAI互換のエンドポイントならopenai/プレフィックスつけるだけ
lm = dspy.LM('openai/mistral-7b-instruct', 
             api_base='http://localhost:8000')

ローカルモデルで試したい場合(Ollama使用)

まずOllamaでモデルを動かしておいて、

ollama run llama3.2:1b

Pythonコードではこんな感じで接続します。

import dspy

lm = dspy.LM("ollama_chat/llama3.2:1b", 
             api_base="http://localhost:11434")
dspy.configure(lm=lm)

OpenAIのAPI代をケチりたい時に便利です(私もよく使います)。

GEPAオプティマイザで企業タスクの構造化情報抽出を試してみた

公式チュートリアル(FacilitySupportAnalyzer)を実際に動かしてみたので、その体験談をシェアします[6]。これ、Metaがリリースした施設メンテナンスのメール分類タスクなんですが、実務でめちゃくちゃ使えそうです。

タスクの内容は、企業内で送られてくる施設関連のメールから以下の3つを抽出すること:

  1. 緊急度(urgent / not_urgent)
  2. 感情分析(positive / negative / neutral)
  3. カテゴリ分類(全10カテゴリから複数選択)

まず、基本的なDSPyモジュールを作ります。

import dspy
from typing import List, Literal

# 3つのシグネチャを定義
class FacilitySupportAnalyzerUrgency(dspy.Signature):
    """緊急度を判定"""
    message: str = dspy.InputField()
    urgency: Literal["urgent", "not_urgent"] = dspy.OutputField()

class FacilitySupportAnalyzerSentiment(dspy.Signature):
    """感情を分析"""
    message: str = dspy.InputField()
    sentiment: Literal["positive", "negative", "neutral"] = dspy.OutputField()

class FacilitySupportAnalyzerCategories(dspy.Signature):
    """カテゴリを分類"""
    message: str = dspy.InputField()
    categories: List[Literal[
        "emergency_repair_services",
        "routine_maintenance_requests", 
        "quality_and_safety_concerns",
        "specialized_cleaning_services",
        "general_inquiries",
        "sustainability_and_environmental_practices",
        "training_and_support_requests",
        "cleaning_services_scheduling",
        "customer_feedback_and_complaints",
        "facility_management_issues"
    ]] = dspy.OutputField()

# 3つのモジュールを組み合わせる
class FacilitySupportAnalyzerMM(dspy.Module):
    def __init__(self):
        self.urgency_module = dspy.ChainOfThought(FacilitySupportAnalyzerUrgency)
        self.sentiment_module = dspy.ChainOfThought(FacilitySupportAnalyzerSentiment)
        self.categories_module = dspy.ChainOfThought(FacilitySupportAnalyzerCategories)
    
    def forward(self, message: str):
        urgency = self.urgency_module(message=message)
        sentiment = self.sentiment_module(message=message)
        categories = self.categories_module(message=message)
        
        return dspy.Prediction(
            urgency=urgency.urgency,
            sentiment=sentiment.sentiment,
            categories=categories.categories
        )

ここからがGEPAの真骨頂です。通常の評価メトリクスに加えて、テキストフィードバックを提供する関数を作ります。

def feedback_urgency(gold_urgency, pred_urgency):
    """緊急度モジュール用のフィードバック生成"""
    if gold_urgency == pred_urgency:
        return "完璧!緊急度の判定は正確です。"
    elif gold_urgency == "urgent" and pred_urgency == "not_urgent":
        return "重要なミス:緊急案件を見逃しています。'immediately'や'ASAP'などのキーワードに注目してください。"
    else:
        return "過剰反応:緊急でない案件を緊急と判定しています。実際の締切や影響度を考慮してください。"

def metric_with_feedback(gold, pred, trace=None, pred_name=None, pred_trace=None):
    """GEPAが使うフィードバック付きメトリクス"""
    score = 0.0
    feedback_parts = []
    
    # 各予測をスコアリング
    if gold.urgency == pred.urgency:
        score += 0.33
    else:
        feedback_parts.append(feedback_urgency(gold.urgency, pred.urgency))
    
    if gold.sentiment == pred.sentiment:
        score += 0.33
    else:
        feedback_parts.append(f"感情分析が違います:{gold.sentiment}が正解")
    
    # カテゴリのF1スコア
    pred_set = set(pred.categories)
    gold_set = set(gold.categories)
    if pred_set and gold_set:
        precision = len(pred_set & gold_set) / len(pred_set)
        recall = len(pred_set & gold_set) / len(gold_set)
        if precision + recall > 0:
            f1 = 2 * precision * recall / (precision + recall)
            score += 0.34 * f1
    
    # predictor単位のフィードバックも可能
    if pred_name == "urgency_module":
        return {
            'score': 1.0 if gold.urgency == pred.urgency else 0.0,
            'feedback': feedback_urgency(gold.urgency, pred.urgency)
        }
    
    # 全体のフィードバック
    feedback = " | ".join(feedback_parts) if feedback_parts else "すべて正解!"
    return {'score': score, 'feedback': feedback}

そして最適化を実行!

# MLflowでトラッキング(めちゃくちゃ便利)
import mlflow
mlflow.dspy.autolog(
    log_compiles=True,
    log_evals=True, 
    log_traces=True
)

# GEPA最適化
optimizer = dspy.GEPA(
    metric=metric_with_feedback,
    auto="light",  # light/medium/heavyから選択
    num_threads=16,
    track_stats=True,
    reflection_minibatch_size=3,
    reflection_lm=dspy.LM("openai/gpt-4o", temperature=1.0)  # 反省用には高性能モデル推奨
)

# コンパイル実行
optimized_program = optimizer.compile(
    program=FacilitySupportAnalyzerMM(),
    trainset=train_data,
    valset=val_data
)

実際に動かしてみた結果がヤバかったです。最初のベースラインが60%くらいの精度だったのが、たった1回の最適化(約20分)で75%まで上がりました。特に緊急度の判定精度が劇的に改善されて、「immediately」「urgent」「ASAP」みたいなキーワードをちゃんと認識するようになりました。

面白かったのは、GEPAが生成した改善プロンプトを見ると、「緊急度を判定する際は、時間的制約を示す語句(immediately, ASAP, urgent, by end of day)に特に注意を払い、ビジネスへの影響度も考慮すること」みたいな具体的な指示が追加されてたこと。人間がマニュアルで書くより良いプロンプトができてる...。

あと、MLflowとの統合がマジで便利でした。最適化の進行状況がリアルタイムで見れるし、どのプロンプトがどんなスコアだったか全部記録されてるので、後から分析するのも簡単。失敗したケースのトレースも見れるので、なぜうまくいかなかったのかも分かります。(Azure MLなどに活かせそう)

GEPAの凄さを数学問題(AIME)で実感した話

公式チュートリアルのAIME(アメリカ数学招待試験)での最適化例も試してみました[7]。これがまた衝撃的で、GPT-4o-miniの数学問題解答精度が、何もしない状態から10%以上改善されたんです。

しかも面白いのが、GEPAに「解法のフィードバック」を与えられること。例えば、「計算は合ってるけど最後の答えの形式が違う」とか「アプローチは正しいけど計算ミスがある」みたいな詳細なフィードバックを与えることで、より効率的に最適化が進むんです。

def metric_with_feedback(example, prediction, trace=None):
    correct_answer = int(example['answer'])
    written_solution = example.get('solution', '')
    
    try:
        llm_answer = int(prediction.answer)
    except ValueError:
        # 整数として解析できない場合のフィードバック
        return {
            'score': 0.0,
            'feedback': f"答えは整数でなければなりません。'{prediction.answer}'は整数として解析できません。"
        }
    
    if llm_answer == correct_answer:
        return {'score': 1.0, 'feedback': "完璧です!"}
    else:
        # 間違いの種類に応じた詳細なフィードバック
        feedback = f"答えが違います。正解は{correct_answer}ですが、{llm_answer}と答えました。"
        if written_solution:
            feedback += f" 正しい解法:{written_solution[:200]}..."  # 最初の200文字
        return {'score': 0.0, 'feedback': feedback}

実際に最適化してみると、GEPAは「計算の各ステップを明示的に示すこと」「最終答えは必ず###記号の後に整数として記載すること」みたいな具体的な改善指示を生成してくれました。人間が試行錯誤してプロンプト改善するより、圧倒的に効率的です。

本番環境で動かす時のパターン

研究用ツールって言われがちなDSPyですが、実は本番でも全然使えます。実際に私が試した構成をいくつか紹介します。

FastAPIでサクッとAPI化

一番簡単なのはFastAPIでラップする方法。こんな感じで書けます。

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import dspy

# DSPyの初期化
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
qa_module = dspy.ChainOfThought('question -> answer')

app = FastAPI(title="DSPy QA API")

class QuestionRequest(BaseModel):
    question: str
    context: str = ""

class AnswerResponse(BaseModel):
    answer: str
    reasoning: str = ""

@app.post("/predict", response_model=AnswerResponse)
async def predict(request: QuestionRequest):
    try:
        # 非同期実行(DSPy 3.0の新機能)
        async_qa = dspy.asyncify(qa_module)
        result = await async_qa(question=request.question)
        
        return AnswerResponse(
            answer=result.answer,
            reasoning=getattr(result, 'reasoning', '')
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

MLflowでちゃんと管理したい人向け

実験管理をきちんとやりたいなら、MLflowとの統合がおすすめ。DSPy 3.0から完全サポートされました。

import mlflow
import mlflow.dspy

# MLflowトラッキングの設定
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("dspy-qa-system")

# 自動ログ記録の有効化(DSPy 3.0の新機能)
mlflow.dspy.autolog()

with mlflow.start_run():
    # モデルの最適化
    optimized_module = optimizer.compile(qa_module, trainset=train_data)
    
    # モデルの保存とログ記録
    mlflow.dspy.log_model(
        dspy_model=optimized_module,
        artifact_path="model",
        input_example={"question": "サンプル質問"},
    )
    
    # メトリクスの記録
    mlflow.log_metric("accuracy", 0.87)
    mlflow.log_metric("optimization_time", 1200)  # 秒

詳しい使い方はAzure Databricksのドキュメントが参考になります。

プロダクションでの実践的な使い方

実際に本番環境で使う時のTipsをまとめます。これ、公式ドキュメントには書いてないけど、実際にやってみて学んだことです。

キャッシュ戦略が超重要

DSPyはデフォルトでLLMの呼び出し結果をキャッシュします。これ自体は良いんですが、実験と本番で同じキャッシュ使うとカオスになります。

# 実験ごとにrollout_id変える
# これで別々のキャッシュ空間になる
result_v1 = program(question="test", rollout_id="experiment_v1")
result_v2 = program(question="test", rollout_id="experiment_v2")

# 本番ではキャッシュ無効化もアリ
lm = dspy.LM("gpt-4o-mini", cache=False)

# または、temperatureを上げて多様性を確保
lm = dspy.LM("gpt-4o-mini", temperature=0.7, rollout_id=f"prod_{timestamp}")
rollout_idについて

DSPy 3.0から導入されたrollout_idパラメータは、キャッシュを名前空間で分離する機能です。同じ入力でも異なるrollout_idを使えば、別のキャッシュエントリとして扱われます。

エラーリトライとフォールバック

本番環境だとAPI呼び出しが失敗することもあるので、リトライ処理は必須です。

import time
from typing import Optional

class RobustQAModule(dspy.Module):
    def __init__(self, fallback_lm: Optional[dspy.LM] = None):
        self.primary = dspy.ChainOfThought('question -> answer')
        self.fallback_lm = fallback_lm
        
    def forward(self, question: str, max_retries: int = 3):
        for attempt in range(max_retries):
            try:
                return self.primary(question=question)
            except Exception as e:
                print(f"Attempt {attempt + 1} failed: {e}")
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)  # 指数バックオフ
                elif self.fallback_lm:
                    # フォールバックLMを使用
                    with dspy.context(lm=self.fallback_lm):
                        return self.primary(question=question)
                else:
                    raise

メトリクス監視とアラート

MLflowと連携して、パフォーマンスメトリクスを継続的に監視します。

import mlflow
from datetime import datetime

class MonitoredQASystem:
    def __init__(self, module, threshold=0.8):
        self.module = module
        self.threshold = threshold
        mlflow.set_experiment("production_qa_system")
        
    def process(self, question: str):
        with mlflow.start_run():
            start_time = time.time()
            
            # 予測実行
            result = self.module(question=question)
            
            # レイテンシ記録
            latency = time.time() - start_time
            mlflow.log_metric("latency", latency)
            
            # 信頼度スコアがあれば記録
            if hasattr(result, 'confidence'):
                mlflow.log_metric("confidence", result.confidence)
                
                # 低信頼度アラート
                if result.confidence < self.threshold:
                    self.send_alert(f"Low confidence: {result.confidence}")
            
            return result
    
    def send_alert(self, message: str):
        # Slack/Email/PagerDutyなどに通知
        print(f"⚠️ Alert: {message}")

A/Bテストとカナリアデプロイ

新しいプロンプトや最適化済みモジュールを段階的にロールアウトする方法です。

import random
import hashlib

class ABTestingModule(dspy.Module):
    def __init__(self, module_a, module_b, split_ratio=0.5):
        self.module_a = module_a  # 現行版
        self.module_b = module_b  # 新版
        self.split_ratio = split_ratio
        
    def forward(self, question: str, user_id: Optional[str] = None):
        # ユーザーIDベースで一貫性を保つ
        if user_id:
            hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
            use_b = (hash_val % 100) < (self.split_ratio * 100)
        else:
            use_b = random.random() < self.split_ratio
        
        # メトリクス記録
        variant = "B" if use_b else "A"
        mlflow.log_param("variant", variant)
        
        if use_b:
            return self.module_b(question=question)
        else:
            return self.module_a(question=question)

他のフレームワークと比べてどうなの?

DSPy vs LangChain

個人的な感想を含めて比較してみました。

項目 DSPy 3.0 LangChain
基本的な考え方 プログラミング言語モデル LLMワークフローの編成
アプローチ 宣言的、自己最適化 命令的、手動プロンプト調整
コミュニティ 成長中(16kスター) 超巨大(90k+スター)
プロンプト最適化 完全自動化 人力頼み
学習難易度 最初きつい、後で楽 最初簡単、後で地獄
向いてる用途 複雑な多段階推論 素早いプロトタイピング

LangChainの方がコミュニティは大きいし、すぐ始められるのは事実。でも、プロダクションで複雑なシステム作るなら、DSPyの自動最適化はマジで助かります。

詳しい比較はQdrantのブログ記事が参考になります。

DSPy vs Guidance/LMQL

GuidanceとかLMQLは単一のLLM呼び出しをきれいに構造化するのが得意。一方、DSPyは複数ステージのパイプライン全体を最適化するのが強み。用途が違うので、実は組み合わせて使うのもアリかも。

DSPyならではの強み

私が思うDSPyの一番の強みは、モデル変更への対応力。GPT-4からClaude 3に変えたくなった時、LangChainだと全部プロンプト書き直しですが、DSPyなら再コンパイルするだけ。これ、実際にやってみると感動しますよ。

あと、研究機関(Stanford NLP)が作ってるだけあって、論文ベースの最新手法がどんどん実装されるのも魅力。GEPAみたいな最新のオプティマイザが使えるのはDSPyくらいです。

実際どこで使われてるの?

2025年9月時点での採用企業を公式ドキュメントとコミュニティ情報から調べてみました。意外と大手が使ってて驚きました。

テック系企業

JetBlueはチャットボットで使ってて、LangChainから乗り換えたら処理速度が大幅改善したと報告されています。Databricksは判定タスクやRAGで活用[8]。他にもReplit(コード生成)、VMware(RAG最適化)などが採用しています。

金融・保険系

Moody'sが金融ワークフローの最適化に使ってたり、RadiantLogicがマルチモジュールルーティング機能付きのAIアシスタントで使ってたり。金融系って慎重なイメージあるけど、意外と先進的ですね。

Eコマース

Zoro UKが300以上のサプライヤーからの商品属性を正規化するのに使ってるとか。Sephoraも何かエージェント系で使ってるみたいですが、詳細は非公開。

ヘルスケア

STChealthがエンティティ解決に使ってて、人間が読める根拠も一緒に出力できるようにしてるとか。Salomaticは医療レポートの充実化に活用。医療系でも使われ始めてるんですね。

学術研究での成果がエグい

StanfordのSTORMっていうプロジェクト、Wikipedia風の記事を自動生成するやつなんですが、GitHubで10,000スター超えてます。トロント大学のWangLabはMEDIQAっていう医療QAコンペで優勝、2位に20ポイント差つけたとか。

UMDの自殺検出システムは、専門家が20時間かけて作ったプロンプトより40%性能が良いって...。これはちょっと衝撃的でした。

使ってみてわかったベストプラクティス

プロジェクト構造はこんな感じがおすすめ

実際にいくつかプロジェクト作ってみて、この構造が一番メンテしやすかったです。

project/
├── src/
│   ├── modules/          # DSPyモジュールはここ
│   │   ├── signatures.py # シグネチャ定義をまとめとく
│   │   └── pipelines.py  # パイプライン実装
│   ├── optimizers/       # カスタムオプティマイザ作るなら
│   └── evaluation/       # 評価メトリクス
├── data/                 # トレーニングデータ
├── experiments/          # 最適化実験の結果
├── configs/              # 設定ファイル(YAML推奨)
└── tests/                # テスト書きましょう

エラーハンドリングはちゃんとやろう

DSPy 3.0から入ったアサーション機能がめちゃくちゃ便利です。こんな感じで使えます。

class ValidatedQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.generate_answer = dspy.ChainOfThought('question -> answer')
    
    def forward(self, question):
        response = self.generate_answer(question=question)
        
        # DSPy 3.0のアサーション機能
        dspy.Suggest(
            len(response.answer.split()) >= 3,
            "回答は最低3単語必要です"
        )
        
        dspy.Assert(
            response.answer.strip() != "",
            "回答は空にできません"
        )
        
        return response

データ量に応じたオプティマイザの選び方

これ、最初めっちゃ迷ったので、私なりの使い分けをまとめました。

# データが少ない時(50例未満)
# → BootstrapFewShotで頑張る
optimizer = dspy.BootstrapFewShot(
    metric=answer_match,
    max_bootstrapped_demos=4,
    max_labeled_demos=4
)

# 中規模データ(50〜500例)
# → RandomSearchも組み合わせる
optimizer = dspy.BootstrapFewShotWithRandomSearch(
    metric=answer_match,
    max_bootstrapped_demos=8,
    num_candidate_programs=10
)

# データがたくさんある or 命令も最適化したい
# → MIPROv2の出番
optimizer = dspy.MIPROv2(
    metric=answer_match,
    auto="medium",  # light, medium, heavy から選ぶ
    num_threads=4
)

# とにかく最高性能を出したい(時間かかってもOK)
# → GEPA使っとけ(3.0の新機能)
optimizer = dspy.GEPA(
    metric=answer_match,
    breadth=10,
    depth=3
)

正直、迷ったらGEPA使っとけばだいたい良い結果出ます(笑)。

詳しいオプティマイザの選び方は公式ドキュメントを参照してください。

パフォーマンスチューニングのコツ

本番で動かす時に役立ったテクニックをいくつか。

# 非同期でバッチ処理したい時
import asyncio

async def process_batch(questions):
    """複数の質問をまとめて処理"""
    tasks = []
    
    for question in questions:
        # 3.0から非同期対応!
        async_module = dspy.asyncify(qa_module)
        tasks.append(async_module(question=question))
    
    results = await asyncio.gather(*tasks)
    return results

# キャッシュの使い分け(地味に重要)
# 実験ごとにrollout_id変えると、キャッシュが分離される
prediction = program(
    question="test",
    rollout_id="experiment_v2"  # これ変えると別キャッシュ
)

特に非同期処理は、API呼び出しが多い時にめちゃくちゃ効きます。10個の質問を処理する時間が1/10になったりします(当たり前だけど)。

DSPyの向き・不向き

DSPyが向いてるケース

実際に使ってみて、これらのケースでは圧倒的に有利だと感じました:

  • 複雑な多段階推論:RAG→推論→要約みたいな複数ステップ
  • 大量の定型タスク:同じようなタスクを大量に処理する場合
  • モデル切り替えが頻繁:GPT-4→Claude→Geminiみたいに頻繁に変える
  • 精度が最重要:多少の開発時間かかっても精度優先
  • 長期メンテナンスが必要:プロンプトの保守性が重要

DSPyが向いてないケース

逆に、こういう時は素直にLangChain使った方が早いです:

  • 単純な単発タスク:ただの要約とか、簡単な分類
  • プロトタイピング段階:とりあえず動くものを作りたい
  • クリエイティブなタスク:詩を書くとか、ストーリー生成とか
  • 学習データがない:最適化には最低でも20-50例は必要
  • チーム全員がLLM初心者:学習コストがそれなりに高い

今後の展開が楽しみ

開発チームが言ってる今後の計画

公式のロードマップ[9]見てたら、いくつか面白そうな機能が予定されてます。

まず、Human-in-the-loopフィードバック機能。これ実装されたら、人間のフィードバックを簡単に取り込めるようになるので、かなり実用的になりそう。あと、最小限のトレーニングデータでも効率的に最適化できるようにする改善も進めてるみたい。現在のオプティマイザから更に20%の改善を目標にしてるって、もう十分すごいのに...。

ベンチマークの体系化も進めるらしくて、どのオプティマイザをいつ使えばいいかがもっと分かりやすくなりそうです。

市場での立ち位置

学術研究から企業の本番環境まで幅広く使われ始めてて、「研究用ツール」と「本番対応プラットフォーム」の間をうまく埋めてる感じがします。LangChainみたいにコミュニティが爆発的に大きくなるかは分からないけど、確実に重要なポジション取ってきてますね。

3ヶ月使い続けて思うこと

DSPyを本格的に使い始めて3ヶ月経ちました。最初は「また新しいフレームワーク覚えるのかよ...」って思ってたけど、今では手放せないツールになってます。(ただまだまだ私自身も使いこなせてないですし、理解できていない部分も結構ある感覚があります・・・)

特にGEPAオプティマイザの登場で、プロンプトエンジニアリングの概念が根本から変わりました。今まで「職人技」だったプロンプト作成が、データドリブンな「工学」になった感じ。手動で「ステップバイステップで考えて」とか書いてた日々が懐かしい...。

ただ、万能じゃないです。シンプルなタスクならLangChainの方が早いし、チームメンバーへの教育コストも無視できません。でも、複雑なAIシステムを本番環境で動かすなら、DSPyの自動最適化は本当に価値があります。

あと、コミュニティがまだ小さいのは課題ですね。Stack Overflowで検索してもほとんど情報出てこないし、日本語の情報はもっと少ない。でも、逆に言えば今から始めれば先行者利益があるかも?

最後に:始めるなら今がチャンス

DSPy 3.0、確かに学習曲線は急です。でも、一度理解してしまえば、プロンプトエンジニアリングの生産性が劇的に上がります。特に企業で本番システム作ってる人には、投資する価値は十分あると思います。

公式ドキュメントも充実してきてるし、GEPAみたいな革新的な機能も続々追加されてます。Azure OpenAIとの統合も改善されてきてるので、企業ユーザーも使いやすくなってきました。

個人的には、今後1年でDSPyみたいな自動最適化フレームワークがスタンダードになると予想してます。手動でプロンプト書く時代は、そろそろ終わりかもしれません。

興味持った方は、まずは公式チュートリアルを試してみてください。きっとハマりますよ。

参考リンク集

公式リソース

チュートリアル・学習リソース

技術論文・参考文献

比較・解説記事

OpenAIモデル関連

それでは、Happy DSPying! 🚀

脚注
  1. https://arxiv.org/abs/2507.19457 ↩︎

  2. https://learn.microsoft.com/en-us/azure/databricks/mlflow3/genai/tracing/integrations/dspy ↩︎

  3. https://dspy.ai/learn/optimization/optimizers/ ↩︎

  4. https://dspy.ai/learn/programming/language_models/ ↩︎

  5. https://github.com/stanfordnlp/dspy/issues/377 ↩︎

  6. https://dspy.ai/tutorials/gepa_facilitysupportanalyzer/ ↩︎

  7. https://dspy.ai/tutorials/gepa_aime/ ↩︎

  8. https://www.databricks.com/blog/optimizing-databricks-llm-pipelines-dspy ↩︎

  9. https://dspy.ai/roadmap/ ↩︎

Discussion