こんにちは、CCCCMKホールディングスAIエンジニアの三浦です。
毎日とても暑いですね・・・。体調を崩さないように気を付けないと、と思います。
databricksのDATA+AI Summit2025に参加したとき、ブレークアウトセッションで"DSPy"というAIアプリケーション開発フレームワークを取り扱っているものが結構あることに気づきました。
DSPyは以前このブログの中でも触れたことがあります。
投稿した日付を見ると、ちょうど一年前でした。今回久しぶりにDSPyを使ってみたのですが、当時よりも使いやすくなった印象を受けました。もちろんDSPy自体の機能強化も要因として挙げられますが、それだけではなくMLflowによって内部処理が分かりやすくなったことにも起因していると感じました。
この記事では改めてDSPyについて取り上げてみたいと思います。
DSPy
AgentをはじめとするLLMを活用したアプリケーションの開発の中で、プロンプトのチューニングをする必要性はLLMの性能が高くなっても相変わらず高いと感じています。たとえば各Agentの振る舞いを決めるシステムプロンプトの調整には結構時間がかかりますし、なかなか安定しない難しさがあります。
DSPyはLLMを利用したアプリケーションのフローを機械学習ライブラリPyTorchと同様に"Module"化した処理単位を組み合わせて構築することが出来ます。DSPyの各Moduleはその振る舞いを自然言語を使って指定出来ますが、その内容によってModuleの動作が影響を受けてしまう点はDSPyでも同じです。
DSPyが面白いのはそのチューニングを自然言語の調整に委ねるのではなく、入力と出力のペアで構成される学習データ、最適化する指標、そして最適アルゴリズムによって通常の機械学習モデルと同様の形式で行うことが出来る、という点です。これによってModuleの最適化を方針が曖昧になりがちなプロンプトチューニングから、学習データの増強、精度指標の再考といったより明瞭な作業に置き換えることが可能になります。
DSPyを構成する主要要素
Module
DSPyではModuleを組み合わせて全体の処理フローを作っていきます。DSPyには定義済みのModuleがいくつか用意されていて、それらはChain Of Thought(CoT)やReActといったプロンプトテクニックに対応しています。これらの定義済みのModuleを使ってRAGなどを実行できる独自のModuleを定義することが出来ます。
Signature
DSPyのSignatureはDSPyの定義済みModuleの振る舞いを決定するために使用されます。単純なSignatureは文字列で表現することが出来、たとえば"question->answer"というSignatureはそのModuleが"question"という値を入力として受け取り、その答えを"answer"という値で返す、という振る舞いを表現しています。
より複雑な振る舞いを表現したい場合は独自クラスで表現することが出来ます。このSignatureの書き方でModuleの動作が変わってしまうので、時間をかけて調整したくなる・・・ところですが、DSPyのドキュメントでは明瞭なSignatureを書くことは推奨しているものの、この調整にいきなり取り掛かるべきではない、と書かれています。それよりもデータを集め、DSPyに搭載された最適化処理に委ねることがDSPyで推奨されるアプローチになっています。
Optimizer
与えられたデータと指標に従って、プログラムを構成するModuleに設定されたプロンプトやLLM自体のパラメータの調整、プログラム全体の構成の最適化を行う機能がDSPyにはOptimizerとして搭載されています。ひとまず私がよく使いそうだな・・・と思ったのは以下2つの種類に分類されるOptimizerです。
Automatic Few-Shot Learning
自動的にプロンプトに含める例示を作成してくれるOptimizerです。学習データセットからランダムにチョイスして構築するLabeledFewShot, 教師となるLLMに生成させたサンプルを含めるBootstrapFewShotなどがあります。
Automatic Instruction Optimization
プロンプトに含まれる指示を自動的に調整するOptimizerです。COPROや指示だけでなく例示も生成するMIPROv2があります。
試してみる
実際にDSPyを使って色々試してみました。Azure DatabricksのNotebookで実行しました。
初期設定＆基本形
必要なライブラリをインストールします。mlflowは3.0以降がLLMアプリ開発にとても便利です。
%pip install -U dspy langchain-community mlflow>=3.0
dbutils.library.restartPython()
mlflowの自動ロギングをオンにします。
import mlflow
mlflow.dspy.autolog()
LLMはAzure OpenAIのモデルを使用しました。DSPyでAzure OpenAIのモデルを使う場合、以下の環境変数に必要な情報を設定する必要があります。環境変数の名前はLangChainのものとは異なるので、普段LangChainを使っている場合は注意です。
- AZURE_API_BASE: エンドポイントURL
- AZURE_API_KEY: APIキー
- AZURE_API_VERSION: APIバージョン
DSPyのプログラムで使用するLLMは、以下のように設定することが出来ます。
import dspy llm = dspy.LM("azure/gpt-4o") dspy.configure(lm=llm)
LLMは次のように直接呼び出すことが出来ますが、DSPyの真価はこれより先で触れる、定義済みのModuleによって発揮されます。
llm("こんにちは")
いくつかある定義済みのModuleの中で、もっともベーシックなものがPredictです。以下の例は、最初に文字列によるSignatureでModuleの振る舞いを決めて、Signatureで定めた形で呼び出しています。
predict = dspy.Predict("question->answer") predict(question="日本で一番高い山は？")
出力結果
Prediction(
answer='日本で一番高い山は富士山です。標高は3,776メートルです。'
)
databricksのNotebookで実行すると、MLflow Trace UIが表示され、実際にはどんなプロンプトが渡されているのかを確認することが出来ます。Predictではシステムプロンプトの中にSignatureの情報が自動的に設定されていることが分かります。
次はChain Of Thought(CoT)も試してみます。ChainOfThoughtという定義済みのModuleがあります。
cot = dspy.ChainOfThought("question->answer") cot(question="日本で一番高い山は？")
出力結果
cot = dspy.ChainOfThought("question->answer")
cot(question="日本で一番高い山は？")
Prediction(
reasoning='日本で一番高い山は富士山です。富士山は標高3776メートルで、日本の象徴的な山として知られています。多くの観光客が訪れる人気のスポットであり、登山や写真撮影のために訪れる人が多いです。',
answer='富士山'
)
ChainOfThoughtを使うとSignatureで指定した"answer"だけでなく、"reasoning"という思考プロセスも生成されていることが分かります。
今度はReActを試してみます。ReActはReActというModuleで実装できます。
まずAgentに与えるToolを定義します。こちらは前回と同じく、BingSearchAPIによるWeb検索Toolを実装しました。
from langchain_community.utilities import BingSearchAPIWrapper def web_search(query:str)->str: """ Bing Searchを使って与えられたqueryに関する情報をwebから検索する """ search = BingSearchAPIWrapper(k=5) return search.run(query)
ReActで実装します。今回はModuleの振る舞いを少し詳細に指定したかったため、Signatureを文字列ではなくSignatureクラスを使って指定してみました。
web_search_agent = dspy.ReAct(
dspy.Signature(
"question->answer",
instructions="web_search関数を使ってWeb検索を利用してquestionに対するanswerを生成する"
),
tools=[web_search]
)
result = web_search_agent(question="DSPyとLangGraphの違いをまとめてください。")
resultには3つのキーが含まれていました。
print(result.keys())
['trajectory', 'reasoning', 'answer']
"trajectory"には思考("thought")とTool実行による観測("observation")、それを受けた次の思考の履歴が記録されているようです。"trajectory"を確認すると、"DSPy 特徴 用途", "LangGraph 特徴 用途"という検索ワードで検索ツールを実行していることが分かりました。結果、次のような回答を生成することが出来ました。
'DSPyは、スタンフォード大学のNLPグループによって開発されたRAGアプリケーションを構築・最適化するためのフレームワークで、プロンプトの自動最適化やAIモデルのプログラミングに焦点を当てています。一方、LangGraphはLangChainによって作成されたオープンソースのAIエージェントフレームワークで、複雑な生成AIエージェントのワークフローを管理し、エージェントのワークフロー管理とタスクの自動化に重点を置いています。'
カスタムModuleの作成
DSPyのReActModuleだけでもかなり良い感じのWeb検索Agentを作ることが出来ることが分かりました。しかし欲を言えばもう少し色々な切り口で情報を調べられるといいな、とも思いました。これを自作のModuleで実現してみようと思います。
実現したいフローは以下の通りです。
- ユーザーのクエリを実現するために必要な情報を検索するための検索クエリを様々な視点で複数生成
- 生成した複数のクエリに対し、Web検索を実行しそれぞれ内容を要約
- 最後にそれらを集約し、参照してユーザーのクエリに回答
この3つの処理に対応したModuleを定義済みModuleで作り、それを組み合わせてカスタムModuleを作っていきます。
まず3つの処理を実行するためのSignatureをそれぞれ次のように用意します。
class GenerateSearchQueries(dspy.Signature): """ questionに答えるために必要な情報をWebから検索するための検索用queriesを多角的な視点で最大5個作る""" question: str = dspy.InputField() queries: list[str] = dspy.OutputField() class SummarizeWebSearchResult(dspy.Signature): """ queryでWeb検索した結果を要約する""" query: str = dspy.InputField() summary: str = dspy.OutputField() class GenerateAnswer(dspy.Signature): """questionに対する回答をcontextsで与えられた情報だけを参照してanswerとして生成する""" question: str = dspy.InputField() contexts: list[str] = dspy.InputField() answer: str = dspy.OutputField()
そして次のようにカスタムModuleを組み立ててみました。
class MultipleWebResearch(dspy.Module): def __init__(self): self.queries_generator = dspy.Predict(GenerateSearchQueries) self.web_researcher = dspy.ReAct(SummarizeWebSearchResult, tools=[web_search]) self.answer_generator = dspy.ChainOfThought(GenerateAnswer) def forward(self, question, **kwargs): queries = self.queries_generator(question=question).queries contexts = [self.web_researcher(query=query).summary for query in queries] return self.answer_generator(question=question, contexts=contexts).answer
これを実行してみます。
researcher = MultipleWebResearch() result = researcher(question="DSPyとLangGraphの違いをまとめてください。") print(result)
結果は次のようになりました。
DSPyはLLMを効果的に扱うためのフレームワークで、プロンプトの自動最適化や信号処理技術の統合を可能にします。LangGraphは複雑なワークフローシステムを開発するためのライブラリで、ワークフローをグラフ構造としてモデル化し、状態管理を支援します
信号処理技術の話が出てきました。DSPy(Differentiable Signal Processing in Python)という、今回取り上げているDSPy(Declarative Self-improving Python)とは違う情報がWebに存在するようで、それが少し混在してしまったようです。一方それ以外の内容はReActを使った時とそんなに変わらないかな・・・という印象を受けました。
MLflow Trace UIを確認し、どんな検索クエリが生成されたのかを見てみました。すると以下のクエリが生成されていることが分かりました。
"DSPyとは"と"DSPy 特徴"というクエリは、結構重複する検索結果になるんじゃないかな、と思います。それよりは"DSPy 活用事例", "DSPy メリット"のようなクエリが生成された方が様々な情報が得られそう、と感じました。
そこで検索クエリを生成するdspy.Predict(GenerateSearchQueries)にターゲットを絞り、より望ましいクエリが生成できるよう最適化を実行してみました。
最適化の実行
最適化を実行するためには、データセットと指標が必要になります。データセットは本当は数10件以上集めたほうが良さそうなのですが、今回は3件を以下のように作ってみました。
input_examples = [
"LangGraphについてまとめてマークダウン形式のレポートで出して。",
"MCPについてレポートをまとめ、マークダウン形式で出力して。",
"Agent開発フレームワークの動向を調べて。"
]
output_examples = [
["LangGraphとは","LangGraph 機能","LangGraph 仕組み","LangGraph 活用事例","LangGraph アップデート"],
["MCP 定義","MCP 解説","MCP 応用例","MCP トレンド","MCP 周辺技術"],
["Agent 開発フレームワーク 最新", "オープンソース Agent framework","Agent-based system framework trend"]
]
train_set = [
dspy.Example(question=x, queries=y).with_inputs("question") for (x, y) in zip(input_examples, output_examples)
]
次に最適化の目標となる指標を決めます。このModuleは検索クエリのリストを生成するので、学習データのqueriesに含まれるものと同じクエリをいくつ生成することが出来たのか、という観点で指標を決めてみました。
次のように関数の形で定義します。
def validate_score(example, pred, trace=None): score = 0 appeared = set() for query in pred.queries: if query in example.queries and query not in appeared: score += 1 appeared.add(query) return score / len(example.queries)
最適化を実行してみます。Optimizerは"Automatic Few-Shot Learning"のBootstrapFewShotを使ってみました。
queries_generator = dspy.Predict(GenerateSearchQueries) tp = dspy.BootstrapFewShot(metric=validate_score) optimized_queries_generator = tp.compile(queries_generator, trainset=train_set)
最適化後のModuleってどんな動作をするんだろう・・・とちょっと気になったので、一度使ってみました。どうも学習データの内容が自動的にこれまでの会話履歴に設定され、これによって回答の内容を調整しているみたいです。
ではこの最適化済みのModuleを組み込んだ、修正版のカスタムModuleを組み立て、実行してみます。
class OptimizedMultipleWebResearch(dspy.Module): def __init__(self): self.queries_generator = optimized_queries_generator self.web_researcher = dspy.ReAct(SummarizeWebSearchResult, tools=[web_search]) self.answer_generator = dspy.ChainOfThought(GenerateAnswer) def forward(self, question, **kwargs): queries = self.queries_generator(question=question).queries contexts = [self.web_researcher(query=query).summary for query in queries] return self.answer_generator(question=question, contexts=contexts).answer optimized_researcher = OptimizedMultipleWebResearch() result = optimized_researcher(question="DSPyとLangGraphの違いをまとめてください。") print(result)
出力結果
DSPyはプロンプトの設計と最適化に特化しており、信号処理と機械学習を統合してプロンプトの自動生成や最適化を行います。LangGraphは、複雑なタスクの自動化と状態管理を可能にするワークフローをグラフ構造としてモデル化し、マルチエージェントシステムの構築に優れています。DSPyはプロンプトの最適化に、LangGraphはエージェントの管理とオーケストレーションに特化しています。
最後にDSPyとLangGraphの得意分野についてのまとめが含まれるようになり、いい感じになったんじゃないでしょうか。
まとめ
ということで、今回はLLMアプリケーション開発フレームワーク"DSPy"を使ってLLMアプリケーション開発を想定した色々なケースを試してみました。DSPyが内部で何をしているのかが以前はつかみにくかったのですが、MLflowを組み合わせることでかなり明確に理解できるようになりました。
LangGraphとの使い分けは私の中でもまだ明確に答えが出ていないのですが、専門Agentの開発はDSPyで、専門Agentを組み合わせたMulti-Agentシステムの開発はLangGraphで、といったように使い分けるのがいいんじゃないかな、と思っています。今度こちらも試してみたいと思いました。
