LLM（大規模言語モデル）を活用したアプリケーション開発が盛んになる中、LangChainやLlamaIndexといった優れたフレームワークが注目されています。これらのツールは、LLMの活用を容易にする様々な機能を提供しており、多くの開発者に支持されています。

しかし、DSPyは、これらのツールとは異なる強みを持っています。特に、プロンプトエンジニアリングやパイプライン構築に特化しており、より効率的で高度なLLMアプリケーションの開発を可能にします。

今回は、DSPyについて解説し、LLMパイプライン構築方法について実装例を交えて紹介していきます。

DSPyとは？

DSPyは、スタンフォード大学が開発した、LLMをより効率的かつ効果的に活用するためのフレームワークです。特に、プロンプトエンジニアリングや、LLMを活用した様々なタスクの実現を容易にすることを目的としています。

DSPyの特徴

• プロンプトの自動生成と最適化: DSPyは、与えられたタスクやデータに基づいて、最適なプロンプトを自動生成し、その性能を向上させるための機能を提供します。
• モジュール化された設計: 様々なLLMタスク（質問応答、テキスト生成など）に対応するモジュールが用意されており、これらのモジュールを組み合わせることで、複雑な処理を構築できます。
• シグネチャの概念: LLMの入力と出力を抽象化し、モジュールの再利用性を高める「シグネチャ」という概念を導入しています。
• Chain-of-ThoughtやReActなどのサポート: これらの手法を簡単に実装するための機能が提供されており、より高度なLLMアプリケーションを開発できます。

DSPyの主な用途

• プロンプトエンジニアリング: プロンプトの設計と改善を効率化
• RAG（Retrieval Augmented Generation）: 外部知識ベースとの連携による情報検索と生成
• LLMパイプラインの構築: 複数のLLMモジュールを組み合わせた複雑な処理の実現

DSPyのメリット

• 開発効率の向上: プロンプトの設計やモデルのチューニングにかかる時間を大幅に削減できます。
• LLMの活用範囲の拡大: 様々なLLMタスクに対応できるため、LLMの活用範囲が広がります。
• 高い柔軟性: モジュール化された設計により、様々なLLMアプリケーションを開発できます。

DSPyの使い方

まずは、pipコマンドを使用してDSPyをインストールします。

$ pip install --upgrade dspy

DSPy環境の構築

OpenAIのgpt-4o-miniを使用する場合は、次のようにします。

import dspy

lm = dspy.LM("openai/gpt-4o-mini", api_key="YOUR_OPENAI_API_KEY")
dspy.configure(lm=lm)

また、Ollamaを利用してローカルPCに構築したモデルを使用する場合は次のようにします。

import dspy

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

これ以外にも、次のようにして様々なモデルを使用することができます。

import dspy

# Anthropicモデル（anthropic/claude-3-opus-20240229）を使用。
# APIキーは環境変数ANTHROPIC_API_KEYを設定することで指定できる。
anthropic_lm = dspy.LM("anthropic/claude-3-opus-20240229")

# また、引数に認証情報を直接渡すこともできる。
anthropic_lm = dspy.LM("anthropic/claude-3-opus-20240229", api_key="YOUR_API_KEY", api_base="https://XXXXX")

# Cohereモデル（cohere/command-nightly）を使用。
# APIキーは環境変数COHERE_API_KEYを設定することで指定できる。
cohere_lm = dspy.LM("cohere/command-nightly")

# Databricksモデル（databricks/databricks-meta-llama-3-1-70b-instruct）を使用。
# 環境変数DATABRICKS_API_KEYとDATABRICKS_API_BASEで認証できる。
# または、Databricksワークスペース上で自動的に認証されます。
databricks_llama3 = dspy.LM("databricks/databricks-meta-llama-3-1-70b-instruct")

基本的なパイプラインの構築

信頼性の高いAIシステムを構築するには、試行錯誤が不可欠です。しかし、プロンプトを直接編集していると、このサイクルが遅れてしまいます。大規模言語モデル、評価指標、パイプラインを変更するたびに、プロンプトの文字列を手作業で調整する必要があるためです。そこでDSPyは、特定の言語モデルやプロンプトに依存しないようにするために開発されています。

DSPyは、プロンプトの文字列を直接操作する代わりに、構造化された自然言語でモジュールを定義する、宣言型のプログラミング手法を採用しています。システム内の全てのAIコンポーネントに対して、入出力の形式（Signature）を指定し、モジュールと大規模言語モデルの対応付けを行います。DSPyは、このSignatureを実際のプロンプトに変換したり、生成された出力を解析したりするため、様々なモジュールを組み合わせて、柔軟かつ効率的なAIシステムを構築できます。

DSPyでは、lm(prompt="prompt")のような方法で大規模言語モデルに直接指示を出すことも可能ですが、より構造化された方法としてモジュールが提供されています。

最もシンプルなモジュールはdspy.Predictです。これは、構造化された入出力の形式（Signature）を受け取り、その形式に合わせた処理を行う関数を返します。例えば、質問（文字列）を入力として受け取り、回答を出力するようなモジュールを簡単に定義できます。

qa = dspy.Predict("question: str -> answer: str")
response = qa(question="大規模言語モデル Llamma 3.2 の特徴を教えてください。")

print(response.answer)

Llamma 3.2は、以下の特徴を持つ大規模言語モデルです。
- 高度な自然言語処理能力
- 複雑なテキスト生成と理解能力
- 多様な用途への応用（例：質問答えシステム、文書生成、機械翻訳）
- 大規模なトレーニングデータセットを使用したトレーニング

また、dspy.Signatureは、AIモデルの入出力の構造を定義するためのクラスです。このクラスを使うことで、AIモデルに渡すデータの型や形式を明確に指定し、モデルの振る舞いをより厳密に定義することができます。

from typing import Literal


class SentimentClassifier(dspy.Signature):
    """感情分析のためのクラス"""

    sentence: str = dspy.InputField()
    sentiment: Literal["positive", "negative", "neutral"] = dspy.OutputField()
    confidence: float = dspy.OutputField()


sentence = """
知人女性との不倫問題で役職停止中の国民民主党の玉木代表が存在感を保っている。
党から処分を受けた後も、「一議員」の立場でメディアなどでの発信を続けているためだ。
ただ、党関係者からは「反省の色が見られない」と批判的な見方も出ている。
"""

classifier = dspy.Predict(SentimentClassifier)
result = classifier(sentence=sentence)

print(f"判定結果: {result.sentiment}, 信頼度: {result.confidence}")

判定結果: negative, 信頼度: 0.85

Chain-of-Thought (CoT) プロンプティングを試してみる

Chain-of-Thought (CoT) プロンプティングは、大規模言語モデル（LLM）がより複雑な問題を解くことができるようにする手法です。従来のLLMは、与えられたプロンプトに対して直接的な回答を生成していましたが、CoTでは、問題を解決するための思考過程を言語モデルに明示的に示すことで、より正確で信頼性の高い回答を得られるようにします。

DSPyのdspy.ChainOfThoughtは、このCoTプロンプティングを簡単に実装するためのクラスです。

cot = dspy.ChainOfThought("question -> answer: float")
answer = cot(question="現在、1500円所持しています。このあと、920円の生姜焼き定食を食べると、残りの所持金はいくらになりますか？")

print(answer.reasoning, "\n")
print(f"答え: {answer.answer}円")

生姜焼き定食の価格は 920円です。所持金は現在 1500円です。したがって、残りの所持金を求めるには、次の式を使用します。

所持金 - 生姜焼き定食の価格 = 残りの所持金

この場合、生姜焼き定食の価格は 920円なので、次のようになります。

1500 - 920 = 580

したがって、残りの所持金は 580円です。 

答え: 580.0円

DSPyにはこの他に、Program-of-Thought（PoT）プロンプティングのモジュールdspy.ProgramOfThoughtや、ReActを実装したモジュールdspy.ReActが用意されています。

RAGシステムを構築してみる

RAG（Retrieval-Augmented Generation）とは、大規模言語モデルが生成するテキストに、外部の知識ベースから検索した情報を付与する技術です。これにより、より正確で情報量の多い、そして創造的なテキスト生成が可能になります。

今回は、arXiv論文をデータソースとして、DSPyによるテキスト生成モデルの構築プロセスを解説します。まずは、arXivから論文の内容を取得し、コーパスに分割するコードを作成します。

import requests
from bs4 import BeautifulSoup

class ArxivCorpusBuilder:
    def __init__(self, url: str) -> None:
        self._url = url

    def split(self, n_parts: int) -> list[str]:
        content = self._get_url_content(self._url)
        return self._split_text_equally(content, n_parts)

    def _get_url_content(self, url: str) -> str:
        response = requests.get(url)
        soup = BeautifulSoup(response.text, "html.parser")
        return soup.get_text()

    def _split_text_equally(self, text: str, n_parts: int) -> list[str]:
        length = len(text)
        part_size, remainder = divmod(length, n_parts)

        sizes = [part_size + 1] * remainder + [part_size] * (n_parts - remainder)

        return [text[sum(sizes[:i]):sum(sizes[:i+1])] for i in range(n_parts)]

次に、このコードを使用して、テキスト生成するコードを書いてみます。使用するモデルは、Ollamaを使用してローカル環境に構築したものを使用します。

まず、dspy.Embedderで、テキストを数値ベクトルに変換する埋め込みモデルを指定します。dspy.retrievers.Embeddingsは、埋め込みモデルを使ってコーパスから質問に関連する文脈を検索する機能を定義しています。それから、質問文に合わせて、検索で見つかった関連する文脈を与えることで、より正確で関連性の高い回答を生成できます。

import dspy

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

embedder = dspy.Embedder("ollama/nomic-embed-text", batch_size=100, api_base="http://localhost:11434", api_key="")
search = dspy.retrievers.Embeddings(
    embedder=embedder,
    corpus=ArxivCorpusBuilder(url="https://arxiv.org/html/2412.21175v1").split(n_parts=100),
    k=3
)

question = "GRB 221009A が平均的なガンマ線バーストとどのように異なるのか？"
respond = dspy.Predict("context, question -> response")
context = search(question).passages
response = respond(context=context, question=question)

print(response.response)

GRB 221009A は、平均的なガンマ線バーストとは大きく異なります。平均的なガンマ線バーストは、通常、低エネルギー部位で発生し、高エネルギー部位では減少しますが、GRB 221009A は、上限を超える高エネルギー部位で発生することが特徴です。この特徴により、GRB 221009A のスペクトルは平均的なガンマ線バーストとは大きく異なります。

生成テキストの評価

従来、自然言語モデルの生成するテキストの評価指標では、生成された文章と正解の文章との単語の一致度（一致率）などを数値化していました。しかし、単語が一致していても、文章全体の意味が異なっている場合があります。

そこで、大規模言語モデルが生成したテキストの質を評価する指標の1つとして、Semantic F1 が提案されています。特に、生成されたテキストが、元の文章や質問に対して、意味的にどれだけ正確に答えられているかを測るために用いられます。

DSPyには、このSemantic F1 を評価するクラスが用意されています。まずは、次のように正解データを用意します。

import dspy

example = {
    "question": "機械学習モデルの一つであるサポートベクタマシンについて簡潔に説明してください。",
    "response": "サポートベクターマシン（SVM） は、教師あり学習のアルゴリズムの一つで、主に分類問題に使われます。様々なデータ（画像、テキストなど）を、あるグループ（クラス）と別のグループ（クラス）に分けたいときに、SVMは非常に有効なツールです。",
}

example = dspy.Example(**example).with_inputs("question")

では、このデータを使用して生成されるテキストを評価してみます。使用するクラスは、dspy.evaluate.SemanticF1 になります。

from dspy.evaluate import SemanticF1

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

# 指標を初期化します。
# decompositional=True は、単語レベルの分解による評価を有効にするオプションです。
metric = SemanticF1(decompositional=True)

# `dspy.Predict` モジュールを使用して、`example` を入力として質問に対する回答を生成します。
qa = dspy.Predict("question: str -> response: str")
pred = qa(**example.inputs())

# 生成された回答と正解の回答を比較し、Semantic F1 スコアを計算
score = metric(example, pred)

print(f"Question: {example.question}")
print(f"Gold Response: {example.response}")
print(f"Predicted Response: {pred.response}\n")
print(f"Semantic F1 Score: {score:.2f}")

Question: 機械学習モデルの一つであるサポートベクタマシンについて簡潔に説明してください。
Gold Response: サポートベクターマシン（SVM） は、教師あり学習のアルゴリズムの一つで、主に分類問題に使われます。様々なデータ（画像、テキストなど）を、あるグループ（クラス）と別のグループ（クラス）に分けたいときに、SVMは非常に有効なツールです。
Predicted Response: サポートベクタマシン（Support Vector Machine，SVM）は、機械学習モデルの一つです。主に二クラス分類問題や非線形関数を表現するために使用されます。 SVM は、データセット内の最も離れた点を「サポートベクトル」として選択し、これらの点を使用して分類境界を定義します。

Semantic F1 Score: 0.89

プロンプト最適化の実装例

DSPyには、プロンプトや重みパラメータを調整するための、最適化に関する機能が用意されています。

基本的な手順

1. システムと評価方法を定義する: 最適化を行う前に、対象のシステムと評価方法を明確に定義します。
2. データセットの準備: トレーニングセット、検証セット、テストセットを用意します。トレーニングセットは300例以上あると十分な精度を達成できるが、30例程度でも改善することがある。なお、検証セットはトレーニングセットの一部として使用されます。
3. 初期の最適化を実行: 最初に簡単な最適化を実行し、結果を評価します。結果に満足できない場合は、システムの定義、データ収集、評価指標、最適化アルゴリズムなどを再検討します。
4. 反復的な開発: データ、プログラム構造、アサーション、評価指標、最適化ステップを段階的に改善することで、最適なプログラムを構築します。

最適化アルゴリズム

DSPyの最適化アルゴリズムは、以下の3つの要素を受け取ります。

• DSPyモジュール: 単一モジュールから複雑なマルチモジュールプログラムまで対応します。
• 評価指標: プログラムの出力を評価し、スコアを割り当てる関数です。
• トレーニング入力: 最小限の入力データ（5～10例程度）でも使用可能です。

主な最適化アルゴリズム

• 自動Few-Shot学習:
  • LabeledFewShot: ラベル付きデータからFew-Shot学習用の入出力ペアを作成します。
  • BootstrapFewShot: 教師モジュールを使用して、入出力ペアを生成します。
  • BootstrapFewShotWithRandomSearch: ランダムサーチにより、最適な入出力ペアを探索します。
  • KNNFewShot: k-Nearest Neighborsアルゴリズムを使用して、類似した入出力ペアを検索します。
• 自動インストラクション最適化:
  • COPRO: 山登り法を使用して、プロンプト用のインストラクションを生成・改良します。
  • MIPROv2: データとデモを考慮して、インストラクションとFew-Shotサンプルを生成します。
• 自動ファインチューニング:
  • BootstrapFinetune: プロンプトベースのDSPyモジュールを蒸留し、モデルの重みをファインチューニングします。
• プログラム変換:
  • Ensemble: 複数のDSPyモジュールをアンサンブルし、単一のモジュールとして使用します。

最適化アルゴリズムの選択

最適なアルゴリズムは、タスクやデータ量に応じて選択する必要があります。一般的には、データ量が少なければBootstrapFewShot、データ量が多い場合はBootstrapFewShotWithRandomSearchやMIPROv2が推奨されます。

では、ここではMIPROv2によるプロンプト最適化を試してみます。まずはそのために、最適化対象のモジュールを定義する必要があります。次のコードのように、dspy.Module を継承したクラスを定義します。このクラスのインスタンス変数として保持できるものは、ディープコピー（copy.deepcopy()）可能なものだけになるように注意してください。

import os

import dspy
import ujson
from dspy.retrievers import Embeddings
from dspy.utils import download

class RAG(dspy.Module):
    _searcher = None

    def __init__(self) -> None:
        self._cot = dspy.ChainOfThought("context, question -> response")
        if RAG._searcher is None:
            RAG._searcher = RAG._build_context_searcher()

    def forward(self, question: str) -> str:
        context = RAG._searcher(question).passages
        return self._cot(context=context, question=question)

    @staticmethod
    def _build_context_searcher() -> Embeddings:
        corpus = RAG._load_corpus()
        embedder = dspy.Embedder("ollama/nomic-embed-text", api_base="http://localhost:11434", api_key="")
        return Embeddings(embedder=embedder, corpus=corpus, k=5, brute_force_threshold=20_000)

    @staticmethod
    def _load_corpus(max_characters: int = 6000) -> list[str]:
        if not os.path.exists("ragqa_arena_tech_corpus.jsonl"):
            download("https://huggingface.co/dspy/cache/resolve/main/ragqa_arena_tech_corpus.jsonl")

        with open("ragqa_arena_tech_corpus.jsonl") as f:
            return [ujson.loads(line)["text"][:max_characters] for line in f]

次に、プロンプト最適化用のデータセットを準備します。

import random

def load_dataset() -> tuple[list[dspy.Example], list[dspy.Example], list[dspy.Example]]:
    if not os.path.exists("ragqa_arena_tech_examples.jsonl"):
        download("https://huggingface.co/dspy/cache/resolve/main/ragqa_arena_tech_examples.jsonl")

    with open("ragqa_arena_tech_examples.jsonl") as f:
        data = [ujson.loads(line) for line in f]
        data = [dspy.Example(**d).with_inputs("question") for d in data]

    random.Random(0).shuffle(data)
    return data[:200], data[200:300], data[300:1000]

ここまで準備ができたら、利用するLLMを定義します。今回は、Llama 3.2 を使用します。また、データセットの準備と、評価方法も定義しておきます。

from dspy.evaluate import SemanticF1

# 大規模言語モデルにはローカル環境に構築した Llama-3.2 を使用
lm = dspy.LM("ollama_chat/llama3.2", api_base="http://localhost:11434", api_key="", max_tokens=3000)
dspy.configure(lm=lm)

# プロンプト最適化用のデータセットを読み込む
trainset, devset, _ = load_dataset()

# 評価指標と評価用のモジュールを定義
metric = SemanticF1(decompositional=True)
evaluate = dspy.Evaluate(
    devset=devset,
    metric=metric,
    num_threads=12,
    display_progress=True,
    display_table=False
)

まず、最適化前のベースラインのモジュールに対して、評価してみましょう。

baseline_rag = RAG()

# ベースラインの評価
evaluate(baseline_rag)

Average Metric: 65.59 / 100 (65.6%): 100%|███████████████████████████████████████████| 100/100 [00:01<00:00, 58.35it/s]
2025/01/01 19:03:02 INFO dspy.evaluate.evaluate: Average Metric: 65.58829046770224 / 100 (65.6%)

では、次のようにして、定義したモジュールに対してプロンプト最適化を試してみます。実行環境にもよりますが、それなりに時間がかかるので気長にお待ち下さい。

# プロンプト最適化のアルゴリズムとして、MIPROv2 を使用する 
mipro = dspy.MIPROv2(metric=metric, auto="light", num_threads=12)
optimized_rag = mipro.compile(baseline_rag, trainset=trainset, requires_permission_to_run=False)

# 最適化完了後の評価
evaluate(optimized_rag)

Average Metric: 65.59 / 100 (65.6%): 100%|███████████████████████████████████████████| 100/100 [00:07<00:00, 13.97it/s]
2025/01/01 19:45:02 INFO dspy.evaluate.evaluate: Average Metric: 65.58829046770224 / 100 (65.6%)

なお、最適化後のモジュールの保存と、保存したモジュールの利用方法は次のようにします。

# モジュールの保存
optimized_rag.save("optimized_rag.json")

# 保存したモジュールの読み込み
loaded_rag = RAG()
loaded_rag.load("optimized_rag.json")

# 保存したモジュールを使用
loaded_rag(question="cmd+tab does not work on hidden or minimized windows")

以上、DSPyの使用方法について簡単に紹介しました。今回紹介した機能以外にも、DSPyは多岐にわたる機能が用意されています。より詳細な情報については、公式ドキュメントやチュートリアルを参照いただくと、より深くDSPyの理解を深めることができるかと思います。

まとめ

今回は、LLM開発を効率化するツールとして注目されているDSPyについて解説しました。特に、プロンプトエンジニアリングやパイプライン構築において、DSPyがどのように役立つのかを具体的に紹介しました。

DSPyは、LLM開発の分野において、まだまだ発展途上のツールです。しかし、その高い柔軟性と拡張性から、今後のさらなる発展が期待されます。今回紹介した内容を参考に、DSPyを活用したLLM開発に挑戦してみてください。

More Informations:

• arXiv:2309.03409, Chengrun Yang et al., 「Large Language Models as Optimizers」, https://arxiv.org/abs/2309.03409
• arXiv:2310.16730, Dong-Ki Kim et al., 「MultiPrompter: Cooperative Prompt Optimization with Multi-Agent Reinforcement Learning」, https://arxiv.org/abs/2310.16730
• arXiv:2406.11695, Krista Opsahl-Ong et al., 「Optimizing Instructions and Demonstrations for Multi-Stage Language Model Programs」, https://arxiv.org/abs/2406.11695
• arXiv:2412.15298, Bhaskarjit Sarmah et al., 「A Comparative Study of DSPy Teleprompter Algorithms for Aligning Large Language Models Evaluation Metrics to Human Evaluation」, https://arxiv.org/abs/2412.15298