DSPy × Ollama でゼロコスト・完全ローカルなLLMパイプラインを作る

はじめに

APIコスト不要、データ外部送信なし。DSPyの自動最適化をOllamaのローカルLLMと組み合わせ、プロダクション品質のLLMパイプラインをオンプレミスで構築する方法を解説します。

1. なぜDSPy × Ollamaなのか

LLMを業務システムに組み込む際、クラウドAPIには次のような課題があります。

• コスト
  大量の取引データを処理するとトークン料金がかさむ
• プライバシー
  会計データ・個人情報を外部サーバーに送信するリスク
• 可用性
  API障害やレートリミットに依存する

OllamaはオープンソースのLLMをローカルで実行するランタイムです。DSPyはOllamaと組み合わせることで、APIコストゼロ・完全オフラインで自動最適化済みのLLMパイプラインを構築できます。

DSPy（パイプライン定義 + 自動最適化）
    ↓
LiteLLM（プロバイダー抽象化）
    ↓
Ollama（ローカルLLMサーバー）
    ↓
llama3.2 / gemma3 / qwen2.5 ...（モデル）

2. 環境構築

Ollamaのインストール

# macOS / Linux
curl -fsSL https://ollama.com/install.sh | sh

# macOS（Homebrew）
brew install ollama

Windowsの場合は ollama.com から公式インストーラーをダウンロードします。

Ollamaの起動とモデルのダウンロード

# サーバーを起動
ollama serve

# モデルをpull（別ターミナルで実行）
ollama pull llama3.2        # 3B、軽量・高速
ollama pull gemma3:4b       # 4B、バランス良好
ollama pull qwen2.5:7b      # 7B、日本語が得意

# 起動確認
ollama list

DSPyのインストール

pip install dspy optuna

3. DSPyからOllamaに接続する

DSPyはLiteLLMを介してOllamaと通信します。接続には dspy.LM() を使い、モデル名に ollama/ または ollama_chat/ プレフィックスを付けます。

import dspy

lm = dspy.LM(
    model="ollama_chat/gemma3:4b",      # チャット形式モデル
    api_base="http://localhost:11434",  # OllamaのデフォルトURL
)
dspy.configure(lm=lm)

ollama/ と ollama_chat/ の違い

プレフィックス	対象	推奨モデル
ollama/	テキスト補完形式（旧来のcompletion API）	mistral、旧世代モデル
ollama_chat/	チャット形式（推奨）	llama3系、gemma3、qwen2.5

最近のモデルは ollama_chat/ を使う方が安定します。

接続パラメータ

lm = dspy.LM(
    model="ollama_chat/gemma3:4b",
    api_base="http://localhost:11434",
    temperature=0.0,    # 再現性が必要な場合は0に設定
    max_tokens=1000,    # 最大出力トークン数
    timeout=120,        # タイムアウト（秒）、大きめに設定推奨
)

4. 基本的な動作確認

シンプルな問答

import dspy

lm = dspy.LM("ollama_chat/gemma3:4b", api_base="http://localhost:11434")
dspy.configure(lm=lm)

qa = dspy.Predict("question -> answer")
result = qa(question="消耗品費と備品費の違いを教えてください")
print(result.answer)

ChainOfThoughtで推論を可視化

import dspy

lm = dspy.LM("ollama_chat/gemma3:4b", api_base="http://localhost:11434")
dspy.configure(lm=lm)

cot = dspy.ChainOfThought("question -> answer")
result = cot(question="消耗品費と備品費の違いを教えてください")

print("推論過程:", result.reasoning)
print("回答:", result.answer)

5. 実践例：会計データの勘定科目自動分類

会計の摘要文から勘定科目を自動判定するシステムを実装します。

ファイル構成

project/
├── make_train_data.py               # 学習・最適化スクリプト（まずこれを実行）
├── test.py                          # 推論CLIスクリプト（最適化後に使う）
├── account_classifier_optimized.json  # 最適化後に自動生成される
├── 経理データ・05年12月.csv
├── 経理データ・06年12月.csv
└── 経理データ・07年12月.csv

スクリプトとCSVファイルを同じディレクトリに置きます。make_train_data.py は Path(__file__).resolve().parent でCSVを探すため、実行ディレクトリに依存しません。

make_train_data.py — 学習・最適化スクリプト

会計CSVを読み込み、DSPyのMIPROv2オプティマイザーで分類器を最適化します。実行は初回のみで、最適化結果はJSONに保存されます。

import dspy
import csv
import re
from pathlib import Path

# ── Ollama接続 ────────────────────────────────────────────────────────────────
lm = dspy.LM(
    model="ollama_chat/gemma3:4b",
    api_base="http://localhost:11434",
)
dspy.configure(lm=lm)

# ── 勘定科目リスト ────────────────────────────────────────────────────────────
ACCOUNT_CATEGORIES = [
    "消耗品費", "交際費", "サービス利用料", "会議費", "通信費",
    "水道光熱費", "車輛費", "支払手数料", "外注費", "支払家賃",
    "諸会費", "旅費交通費", "福利厚生費", "雑費", "租税公課",
    "損害保険料", "修繕費", "法定福利費", "支払利息", "新聞図書費",
]

# ── Signature（入出力仕様）────────────────────────────────────────────────────
class ClassifyAccount(dspy.Signature):
    """
    支出の摘要と金額から、適切な勘定科目を判定する。
    勘定科目は必ず以下のリストから選ぶこと: {account_list}
    """
    expense_description = dspy.InputField(desc="支出の摘要（取引先・購入品目を含む）")
    amount              = dspy.InputField(desc="取引金額（円）")
    tax_type            = dspy.InputField(desc="消費税区分（内10、内８軽、空白）")
    account_list        = dspy.InputField(desc="選択可能な勘定科目のリスト")
    account_category    = dspy.OutputField(desc="判定した勘定科目名（リストから1つ選択）")
    reason              = dspy.OutputField(desc="判定理由（監査ログ用）")

# ── Module（実行層）───────────────────────────────────────────────────────────
class AccountClassifier(dspy.Module):
    def __init__(self):
        # ChainOfThoughtで推論過程を明示させる
        self.classify = dspy.ChainOfThought(ClassifyAccount)

    def forward(self, expense_description, amount, tax_type=""):
        return self.classify(
            expense_description=expense_description,
            amount=amount,
            tax_type=tax_type,
            account_list="、".join(ACCOUNT_CATEGORIES),
        )

# ── 評価指標 ──────────────────────────────────────────────────────────────────
def account_accuracy(example, prediction, trace=None):
    """正答率（完全一致）"""
    return example.account_category == prediction.account_category

# ── CSVの読み込み ──────────────────────────────────────────────────────────────
BASE_DIR = Path(__file__).resolve().parent

def find_csv_by_year(year_prefix):
    matches = sorted(BASE_DIR.glob(f"*{year_prefix}年12月.csv"))
    if not matches:
        raise FileNotFoundError(f"{year_prefix}年12月のCSVが見つかりません: {BASE_DIR}")
    return matches[0]

def load_training_data(filepath):
    """会計CSVからDSPy用のtrainsetを作成"""
    EXPENSE_ACCOUNTS = set(ACCOUNT_CATEGORIES)
    trainset = []

    with open(filepath, encoding='cp932') as f:
        rows = list(csv.reader(f))

    for row in rows:
        # 番号列が数値の行がデータ行
        if not row or not row[0].strip().isdigit():
            continue

        account = re.sub(r'\s+', '', row[5])   # 全角スペース除去
        if account not in EXPENSE_ACCOUNTS:
            continue

        example = dspy.Example(
            expense_description=row[16].strip(),
            amount=row[13].strip(),
            tax_type=row[14].strip(),
            account_category=account,
        ).with_inputs("expense_description", "amount", "tax_type")

        trainset.append(example)

    return trainset

# 全CSVを読み込んで件数確認
all_data = []
for path in sorted(BASE_DIR.glob("*.csv")):
    all_data.extend(load_training_data(path))
print(f"学習データ件数: {len(all_data)}")

# 年度別にtrain/validationに分割（05年・06年 → train、07年 → validation）
trainset   = load_training_data(find_csv_by_year("05"))
trainset  += load_training_data(find_csv_by_year("06"))
valset     = load_training_data(find_csv_by_year("07"))
print(f"train件数: {len(trainset)}")
print(f"validation件数: {len(valset)}")

# ── MIPROv2で最適化 ───────────────────────────────────────────────────────────
optimizer = dspy.MIPROv2(
    metric=account_accuracy,
    auto="light",   # "light" / "medium" / "heavy" で試行回数を調整
)

optimized_classifier = optimizer.compile(
    AccountClassifier(),
    trainset=trainset,
    valset=valset,
)

# 最適化済みモジュールを保存
optimized_classifier.save("account_classifier_optimized.json")
print("最適化完了")

コードのポイント

パート	役割
ClassifyAccount（Signature）	入出力の仕様を宣言。HOW（プロンプト文面）ではなくWHAT（何を入れて何を出すか）を定義する
AccountClassifier（Module）	ChainOfThoughtでSignatureを実行。推論ステップを出力させることで監査ログにも使える
find_csv_by_year()	年号プレフィックスでCSVを自動検索。ファイル名が変わっても柔軟に対応できる
account_accuracy	正解ラベルと予測の完全一致で正答率を計算。オプティマイザーの目的関数になる
MIPROv2(auto="light")	少数ショット例の合成と指示文の自動改善を行う。lightは最小試行回数でローカル環境向け
.save()	最適化後のプロンプトとデモをJSONに保存。次回は.load()で即利用可能

6. 推論・テスト用CLIスクリプト

make_train_data.py で生成した account_classifier_optimized.json をロードし、摘要と金額を渡して勘定科目を1件判定します。

test.py

# 使い方:
#   python test.py "ノートパソコン" "150000" --tax-type "内10"

import argparse
from pathlib import Path

import dspy

ACCOUNT_CATEGORIES = [
    "消耗品費", "交際費", "サービス利用料", "会議費", "通信費",
    "水道光熱費", "車輛費", "支払手数料", "外注費", "支払家賃",
    "諸会費", "旅費交通費", "福利厚生費", "雑費", "租税公課",
    "損害保険料", "修繕費", "法定福利費", "支払利息", "新聞図書費",
]

MODEL_PATH = Path(__file__).resolve().parent / "account_classifier_optimized.json"


class ClassifyAccount(dspy.Signature):
    """
    支出の摘要と金額から、適切な勘定科目を判定する。
    勘定科目は必ず以下のリストから選ぶこと: {account_list}
    """
    expense_description = dspy.InputField(desc="支出の摘要（取引先・購入品目を含む）")
    amount              = dspy.InputField(desc="取引金額（円）")
    tax_type            = dspy.InputField(desc="消費税区分（内10、内８軽、空白）")
    account_list        = dspy.InputField(desc="選択可能な勘定科目のリスト")
    account_category    = dspy.OutputField(desc="判定した勘定科目名（リストから1つ選択）")
    reason              = dspy.OutputField(desc="判定理由（監査ログ用）")


class AccountClassifier(dspy.Module):
    def __init__(self):
        self.classify = dspy.ChainOfThought(ClassifyAccount)

    def forward(self, expense_description, amount, tax_type=""):
        return self.classify(
            expense_description=expense_description,
            amount=amount,
            tax_type=tax_type,
            account_list="、".join(ACCOUNT_CATEGORIES),
        )


def build_parser():
    parser = argparse.ArgumentParser(description="最適化済み勘定科目分類器を試す")
    parser.add_argument("expense_description", help="支出の摘要")
    parser.add_argument("amount", help="取引金額（円）")
    parser.add_argument("--tax-type", default="", help="消費税区分。例: 内10, 内８軽")
    parser.add_argument(
        "--model",
        default="ollama_chat/gemma3:4b",
        help="使用する Ollama モデル名",
    )
    parser.add_argument(
        "--api-base",
        default="http://localhost:11434",
        help="Ollama API の URL",
    )
    return parser


def main():
    args = build_parser().parse_args()

    if not MODEL_PATH.exists():
        raise FileNotFoundError(f"最適化済みモデルが見つかりません: {MODEL_PATH}")

    lm = dspy.LM(model=args.model, api_base=args.api_base)
    dspy.configure(lm=lm)

    classifier = AccountClassifier()
    classifier.load(str(MODEL_PATH))

    prediction = classifier(
        expense_description=args.expense_description,
        amount=args.amount,
        tax_type=args.tax_type,
    )

    print(f"摘要: {args.expense_description}")
    print(f"金額: {args.amount}")
    print(f"税区分: {args.tax_type or '(空白)'}")
    print(f"勘定科目: {prediction.account_category}")
    print(f"理由: {prediction.reason}")


if __name__ == "__main__":
    main()

実行例

$ python test.py "カード／AMAZON.CO.JP:HDMIケーブル" "2574" --tax-type "内10"
摘要: カード／AMAZON.CO.JP:HDMIケーブル
金額: 2574
税区分: 内10
勘定科目: 消耗品費
理由: HDMIケーブルは事業用の消耗品に該当するため

$ python test.py "ノートパソコン" "150000" --tax-type "内10"
摘要: ノートパソコン
金額: 150000
税区分: 内10
勘定科目: 消耗品費
理由: ノートパソコンは事業用の消耗品として計上されるため

デフォルトでは gemma3:4b を使いますが、--model オプションでモデルを切り替えられます。

python test.py "クラウドストレージ料金" "3300" --tax-type "内10" --model "ollama_chat/qwen2.5:7b"

7. モデル選定ガイド

日本語の会計テキストを扱う場合のモデル選択指針です。

モデル	サイズ	日本語品質	速度	VRAM目安	推奨用途
llama3.2	3B	△	◎	2GB	開発・動作確認
llama3.1:8b	8B	○	○	6GB	バランス重視
gemma3:4b	4B	○	○	4GB	バランス重視
qwen2.5:7b	7B	◎	○	5GB	日本語優先
qwen2.5:14b	14B	◎	△	10GB	精度優先
mistral:7b	7B	○	○	5GB	汎用

日本語の会計テキスト分類には qwen2.5:7b が特に有効です。日本語コーパスを多く学習しており、専門用語の理解精度が高い傾向があります。

# qwen2.5を使う場合
lm = dspy.LM(
    "ollama_chat/qwen2.5:7b",
    api_base="http://localhost:11434",
    temperature=0.0,
)
dspy.configure(lm=lm)

8. トラブルシューティング

接続エラー: Connection refused

# Ollamaが起動しているか確認
curl http://localhost:11434/api/tags

# 起動していない場合
ollama serve

出力が構造化されない（account_category が空になる）

DSPyはLLMに対してJSON形式での出力を要求しますが、小さなモデルでは崩れることがあります。

# Ollamaのnative JSON modeを有効にする
lm = dspy.LM(
    "ollama_chat/llama3.2",
    api_base="http://localhost:11434",
    # LiteLLM経由でOllamaのformatオプションを渡す
    extra_body={"format": "json"},
)

または、より大きなモデルに切り替えることで解決することが多いです。

最適化が遅い

MIPROv2 はデフォルトで多数のLLM呼び出しを行います。ローカル環境では auto="light" から始めてください。

optimizer = dspy.MIPROv2(metric=account_accuracy, auto="light")

タイムアウトが発生する

大きなモデルは推論に時間がかかります。タイムアウトを延ばしてください。

lm = dspy.LM(
    "ollama_chat/qwen2.5:14b",
    api_base="http://localhost:11434",
    timeout=300,   # 5分
)

9. まとめ

DSPy × Ollamaの組み合わせで実現できること：

項目	内容
コスト	OpenAI APIのトークン料金が不要
プライバシー	会計データ・個人情報が外部に出ない
可用性	API障害・レートリミットに依存しない
自動最適化	BootstrapFewShot・MIPROv2がローカルで動く
モデル切替	dspy.configure(lm=...) の1行で切り替え可能

# 完全な初期化コード（コピーして使える）
import dspy

lm = dspy.LM(
    model="ollama_chat/qwen2.5:7b",
    api_base="http://localhost:11434",
    temperature=0.0,
    timeout=120,
)
dspy.configure(lm=lm)

プライバシーが求められる業務データ（会計・医療・法務など）をLLMで処理したい場合、DSPy × Ollamaは実用的な選択肢です。DSPyの宣言的なプログラミングモデルにより、モデルをクラウドAPIに切り替える際もコードの変更は最小限で済みます。

参考リソース

リソース	URL
DSPy 公式サイト	https://dspy.ai
DSPy GitHub	https://github.com/stanfordnlp/dspy
Ollama 公式サイト	https://ollama.com
Ollama GitHub	https://github.com/ollama/ollama
LiteLLM（DSPy内部で使用）	https://docs.litellm.ai
利用可能なOllamaモデル一覧	https://ollama.com/library