プロンプトエンジニアリングはどう変わる？ DSPy / TextGrad による自動最適化の実力検証

1. はじめに：終わらない「プロンプト修正」

本記事をご覧いただきありがとうございます。ドコモアドベントカレンダー22日目の記事になります。

初めまして。NTTドコモ R&D戦略部の武田です。現在は主に対話AIの評価・改善業務に従事しています。日々、AIの応答品質を定量・定性の両面から分析していますが、品質改善のプロセスにおいて必ず直面する「最大のボトルネック」があります。

それが、「プロンプトの微調整」 です。

LLMアプリ開発者の多くがコードの実装よりも、指示文（プロンプト）の調整に膨大な時間を費やしているのが実情ではないでしょうか。

「もう少し丁寧に」と修正すれば要点がぼやけ、「簡潔に」と書き換えれば必要な情報が抜け落ちる——。このトレードオフを解消するために、手作業で表現を少しずつ書き換えては試行するプロセスは、エンジニアリングというよりは「職人芸」に近く、再現性や効率の面で大きな課題があります。

そこで本記事では、このプロセスを自動化する 「プロンプト自動最適化（Automatic Prompt Optimization）」 に焦点を当てます。

主要フレームワークである DSPy と TextGrad の2つを取り上げ、同一の分類タスクに対して「どのように最適化するのか」「結果としてどんなプロンプトが生まれるのか」を比較します。「プロンプトエンジニアリングは今後どう変わるのか？」そのヒントとなる、両ツールの挙動と使い分けの指針について解説します。

                   

【Before】手動での試行錯誤

【After】自動最適化の適用

---

2. タスク定義とベースライン評価

本記事で検証タスクとして設定したのは、判断基準が曖昧になりやすい 「カスタマーサポートの問い合わせ分類」 です。

2.1 タスク概要

入力された問い合わせテキストを、以下の4つの社内管理コード（カテゴリ）に分類します。

• Category_A (Refund): 返金要求、緊急の金銭トラブル
• Category_B (Cancellation): 解約、契約変更、事務手続き
• Category_C (Technical_Issue): システム不具合、バグ報告
• Category_D (Feature_Request): 機能要望、ユーザーの声

2.2 難易度設計

単純なキーワードマッチングでは分類が困難な、「複数の意図が混在するデータ」を意図的に含めました。 たとえば、以下のようなケースです。

入力例: 「不具合が多くてもう使いたくない。解約して返金してほしい。」

この文には「不具合」「解約」「返金」の3要素が含まれています。今回はビジネスルールとして Refund (A) > Cancellation (B) > Technical_Issue (C) の優先順位を正解として定義しました（「返金要求」が含まれる場合は、たとえ原因が不具合であってもカテゴリAを優先する、というルールです）。

2.3 ベースライン評価（初期プロンプト）

まず、最適化を行わない状態（Zero-shot）での分類精度を確認します。 ベースラインとして使用するプロンプトは、カテゴリ定義を箇条書きにしただけのシンプルなものです。

実行コード（eval_baseline.py）:

# eval_baseline.py
"""
ベースライン評価 
"""
from __future__ import annotations
import json
import os
import re  
from typing import List, Dict, Any
from openai import OpenAI

from dotenv import load_dotenv


load_dotenv()

# ===== 設定 =====
SMART_MODEL = os.getenv("SMART_MODEL", "gpt-4.1")

# ===== 定義マップ =====
LABEL_MAP = {
    "Refund": "Category_A",
    "Cancellation": "Category_B",
    "Technical_Issue": "Category_C",
    "Feature_Request": "Category_D"
}

# ===== プロンプト =====
BASE_PROMPT_TEMPLATE = """
あなたはカスタマーサポートの問い合わせを分類するAIです。
以下の「社内管理コード」の定義に基づいて、問い合わせを分類してください。

【定義】
- Category_A: 緊急対応案件（金銭トラブルやクレームなど、優先度「高」の対応）
- Category_B: 事務手続き（契約内容の変更、アカウント更新、ステータス変更）
- Category_C: システム不具合（エラーログの確認が必要なもの）
- Category_D: ユーザーの声（機能への要望や感想）

【注意】
- 判断に迷う場合は、直感で構いません。
- 出力はカテゴリ名のみ（例: Category_A）にしてください。余計な説明は不要です。

【問い合わせ内容】
{text}

【分類】
"""

# ===== ユーティリティ関数 (厳格化) =====
def normalize_label_strict(raw_text: str) -> str:
    """
    LLMの出力から Category_X を厳格に抽出する。
    キーワード検索（例: '返金'が含まれたらAなど）は廃止。
    """
    if not raw_text:
        return ""
  
    clean_text = raw_text.strip()
    match = re.search(r"(Category_[A-D])", clean_text)
  
    if match:
        return match.group(1)
  
    return ""

def load_dataset(path: str) -> List[Dict[str, str]]:
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)

# ===== メイン評価処理 =====
def evaluate(dataset_path: str = "data/dataset.json"):
    # 1. OpenAIクライアントの初期化
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        raise ValueError("OPENAI_API_KEY が見つかりません。.envを確認してください。")
  
    client = OpenAI(api_key=api_key)

    print(f"\n=== ベースライン評価開始 ===")
    print(f"Model: {SMART_MODEL}")
    print("※ 詳細ログ出力モード\n")
  
    dataset = load_dataset(dataset_path)
    test_data = dataset 
  
    correct = 0
    # 集計用辞書 (キーは Category_A 等のコード)
    per_label_total = {v: 0 for v in LABEL_MAP.values()}
    per_label_correct = {v: 0 for v in LABEL_MAP.values()}

    # ログヘッダー
    print(f"{'Status':<6} | {'Gold Label':<16} -> {'Prediction':<16} | {'Text (Prefix)'}")
    print("-" * 80)

    for idx, item in enumerate(test_data, 1):
        text = item["text"]
        raw_label = item["label"] # "Refund" など
  
        # 正解ラベルをコード (Category_A) に変換
        gold = LABEL_MAP.get(raw_label, "Category_A")
  
        # プロンプト作成
        prompt_content = BASE_PROMPT_TEMPLATE.format(text=text)

        try:
            # 2. OpenAI API呼び出し
            response = client.chat.completions.create(
                model=SMART_MODEL,
                messages=[
                    {"role": "user", "content": prompt_content}
                ],
                temperature=0.0,
                max_tokens=20
            )
  
            # 3. レスポンスからテキスト抽出
            pred_text = response.choices[0].message.content
            # 厳格な正規化関数を通す
            pred = normalize_label_strict(pred_text)
  
        except Exception as e:
            print(f"API Error at {idx}: {e}")
            pred = "Error"

        # 集計処理
        per_label_total[gold] += 1
        is_correct = (pred == gold)
  
        status_icon = "✅ OK" if is_correct else "❌ NG"
        # ログ表示用に改行を除去
        display_text = text.replace("\n", " ")[:30] + "..."

        print(f"{status_icon:<6} | {gold:<16} -> {pred:<16} | {display_text}")

        if is_correct:
            correct += 1
            per_label_correct[gold] += 1

    # 最終結果表示
    acc = correct / len(test_data) * 100
    print("-" * 80)
    print(f"全体正解率: {acc:.2f}% ({correct}/{len(test_data)})")
    print("-" * 80)
  
    # ラベル別精度の表示
    # LABEL_MAP を逆引きして表示名を作る
    REVERSE_MAP = {v: k for k, v in LABEL_MAP.items()}
  
    for code in sorted(LABEL_MAP.values()):
        total = per_label_total[code]
        ok = per_label_correct[code]
        lbl_acc = (ok / total * 100) if total > 0 else 0
        label_name = REVERSE_MAP[code]
        print(f"{code:<12} ({label_name:<15}): {lbl_acc:5.1f}% ({ok}/{total})")

if __name__ == "__main__":
    evaluate()

実行結果：

=== ベースライン評価結果 ===
Model: gpt-4.1

Status | Gold Label       -> Prediction       | Text (Prefix)
--------------------------------------------------------------------------------
✅ OK   | Feature_Request  -> Feature_Request 
      Text: CSVエクスポート機能についてですが、なぜJSON形式が選べないのでしょうか？ 技術的に難しいことではないはずです。至急実装してください。
❌ NG   | Cancellation     -> Technical_Issue 
      Text: 御社のサービスは素晴らしいと思いますが、私の現在のプロジェクト規模には少しオーバースペックでした。来月の更新を停止したいのですが、手続きのページがエラーで開きません。どうにかしてください。
--------------------------------------------------------------------------------
❌ NG   | Technical_Issue  -> Refund
      Text: ログインしようとすると毎回「不明なエラー」と表示されます。これで今月の利用料を取られるのは納得がいきませんが、まずはこの現象を直す方法を教えてください。話はそれからです。
--------------------------------------------------------------------------------
❌ NG   | Feature_Request  -> Refund
      Text: 以前から要望を出しているダークモードの実装はまだですか？夜間の作業で目が疲れて仕方がないので、早急に対応していただきたいです。これがないと継続利用も考え直さざるを得ません。
✅ OK   | Refund           -> Refund
      Text: ３日連続でサーバーがダウンしていて仕事になりません。損害賠償とまでは言いませんが、使えなかった期間分の料金を日割りで差し引く等の対応をするのが筋ではないでしょうか？
... (中略) ...
--------------------------------------------------------------------------------
全体正解率: 76.25% (61/80)
--------------------------------------------------------------------------------
Refund          (Category_A):  88.2% (15/17)
Cancellation    (Category_B):  60.9% (14/23)  <-- 苦手
Technical_Issue (Category_C):  65.0% (13/20)  <-- 苦手
Feature_Request (Category_D):  95.0% (19/20)

2.4 結果分析

評価の結果、正解率は 76.25% となりました。 数値上は悪くない結果ですが、失敗ケース（Error Analysis）を確認すると、LLMの挙動に明確な癖が見えてきます。

1. 特定の単語（「返金」など）に過剰反応している

   • 文脈上は Cancellation（解約）の話題であっても、「返金してほしいくらいですが、今回は結構です」といった表現が含まれると、モデルが単語だけを拾って Refund（A）に誤分類するケースが目立ちました。

2. 複合的な意図（解約×エラー）に対する判断の揺らぎ

   • 「解約しようとしたらエラーになった」というような、本来は事務手続き（B）寄り、あるいは不具合（C）寄りとして扱うべき内容に対し、判断の軸が定まらず分類がブレています。

現状は 「単語レベルの反応はできているが、文脈やルールの優先順位を正しく解釈できていない」 状態です。 この課題に対し、DSPyとTextGradがどのようなアプローチで改善を図るのかを見ていきましょう。

---

3. 検証対象ツールの技術的特徴

本記事で比較検証を行うのは、以下の2つのフレームワークです。

• DSPy：宣言的に定義したLLMパイプラインを、評価指標に基づいて自動最適化するフレームワーク
• TextGrad：自然言語フィードバックを「疑似勾配」として扱い、プロンプトや生成物を最適化するフレームワーク

以降、それぞれの思想と仕組みを、本検証での設定に寄せて解説します。

3.1 DSPy

① フレームワークの思想

DSPyはプロンプトを単なる文字列としてではなく、入出力型を持つモジュールとして扱います。 開発者はPythonコード上で「Signature（入力/出力フィールド）」と「Module（処理の呼び出し方）」を定義します。これに対しDSPyは、使用するオプティマイザ（最適化器）に応じて、「指示文やFew-shot事例といった プロンプトの構成要素をどう組み合わせれば スコアが最大化するか」を自動的に探索します。

公式ドキュメントや論文では、LM Assertions（LLMに満たしてほしい制約条件）といった概念も提唱されています。要約すると「開発者が仕様（Signature）を宣言し、プログラム（DSPy）がその仕様を満たすように自己改善する」という設計思想を持っています。

• 公式ドキュメント: dspy-docs.vercel.app
• GitHub: github.com

② 最適化の仕組み（MIPROv2）

今回の検証では、DSPyに同梱されている最適化アルゴリズム MIPROv2 を使用しました。MIPROv2は指示と事例の両方を最適化対象とするため、以下のプロセスを実行します。

1. 事例抽出: データセットから学習に有効な事例をいくつか抽出します。
2. 候補生成: 抽出した事例を用いて、候補プロンプト（指示文） と 候補Few-shotセット を複数パターン生成します。
3. 探索と評価: 組み合わせごとにDevセットで推論・評価を行い、ベイズ最適化の手法を用いて最も精度の高い組み合わせを絞り込みます。

つまり、人間が試行錯誤する「指示と事例の組み合わせ調整」をアルゴリズムが高速に代行する仕組みです。

③ 本検証での設定

今回の実験では、DSPy側は次のように設定しました。

• Signature: SupportClassification(text -> label)
  • 「問い合わせテキストを受け取り、Category_A〜Dを返す」という入出力を定義。
• Module: dspy.ChainOfThought(SupportClassification)
  • 単純な分類ではなく、推論ステップを含ませるCoTモジュールとして定義。
• モデル構成:
  • Task Model: gpt-4.1-mini
  • Prompt Model: gpt-4.1
• 最適化プロセス:
  • 学習データ（Train）20件を使用し、指示文候補×Few-shot事例セットの探索を30試行実施。Devセットでの正解率が最も高い構成を採用しました。

3.2 TextGrad

① フレームワークの思想

TextGradは、LLMシステム全体をPyTorchのような 計算グラフと勾配 に見立てて扱うライブラリです。 テキストをVariable（変数）として定義し、別のLLM（批評モデル）が生成する自然言語フィードバックを 「テキスト版の勾配（pseudo-gradient）」 として逆伝播させます。この勾配情報に基づいて、プロンプトや生成物を直接書き換えて更新します。

元論文では、モデルの出力をLLMがレビューし、「どこが悪かったのか」「どう修正すべきか」をコメントさせ、それを手がかりにシステムを改善する枠組みが提案されています。

• 論文: arxiv.org
• GitHub: github.com

② 最適化の仕組み

本検証におけるTextGradの処理フローは以下の通りです。

1. Forward（順伝播）:
   • 学習対象の system_prompt と問い合わせ文をTaskモデル（gpt-4.1-mini）に入力し、予測結果を得ます。
2. Backward（逆伝播）:
   • Judgeモデル（gpt-4.1）に対し、「現在のsystem prompt」「ユーザー入力」「モデル予測」「正解ラベル」を渡します。
   • 予測が正しければ Correct. を、間違っていれば「なぜ間違えたか」「system promptをどう修正すべきか」という自然言語フィードバックを生成させます。
3. 更新（Update）:
   • 得られたフィードバックを TextualGradientDescent オプティマイザが解釈し、system_prompt の文章自体をより良い内容に書き換えます。

このサイクルを繰り返すことで、例えば「返金関連ワードの優先順位」や「解約と技術不具合の境界線」といった曖昧なルールが、プロンプト内でより明文化されていきます。

③ 本検証での設定

TextGradには以下の役割と設定を与えました。

• 学習対象: システムプロンプト1つのみ（Few-shot事例は使用せず指示文のみを最適化）。
• 評価ロジック: 完全一致ルール（余計な文字が含まれた時点で不正解扱い）。
• 最適化プロセス:
  • 学習データ（Train）20件をミニバッチで回しながら12ステップの更新を実行。各ステップごとにテストデータ（Test）での正解率推移を計測しました。

---

4. 検証結果と考察

本章では、実装した最適化コードを実行し、その結果を比較・分析します。

4.1 実装コード

DSPyの実装 (optimize_dspy.py):

# optimize_dspy.py
"""
DSPyによるプロンプト自動最適化スクリプト (MIPROv2使用)
"""
import json
import random
import os
import dspy
from dspy.teleprompt import MIPROv2
from dspy.evaluate import Evaluate
from dotenv import load_dotenv

load_dotenv()

# ==============================================================================
# 0. 設定定数 (環境変数およびパラメータ)
# ==============================================================================
# APIキーの確認
api_key = os.getenv("OPENAI_API_KEY")
if not api_key:
    raise ValueError("OPENAI_API_KEY が見つかりません。.envを確認してください。")

# モデル名の定義 (.envから取得)
FAST_MODEL_NAME = os.getenv("FAST_MODEL", "gpt-4.1-mini")
SMART_MODEL_NAME = os.getenv("SMART_MODEL", "gpt-4.1")
print(f"Model Config -> Fast: {FAST_MODEL_NAME}, Smart: {SMART_MODEL_NAME}")

# 学習・最適化パラメータ
SEED = 30
TRAIN_SIZE = 20           # 学習データの数
NUM_CANDIDATES = 7        # プロンプト候補の生成数
INIT_TEMPERATURE = 0.5    # 探索時の創造性
MAX_BOOTSTRAPPED_DEMOS = 5 # 事例の最大数
MAX_LABELED_DEMOS = 5     # ラベル付き事例の最大数
NUM_TRIALS = 30           # 試行回数
MINIBATCH_SIZE = 8        # 1回の試行で評価に使うデータ数

# ==============================================================================
# 1. データセットの準備
# ==============================================================================
LABEL_MAP = {
    "Refund": "Category_A",
    "Cancellation": "Category_B",
    "Technical_Issue": "Category_C",
    "Feature_Request": "Category_D"
}

def load_data():
    """dataset.json を読み込み、DSPy用の学習データ形式に変換する"""
    with open("data/dataset.json", "r", encoding="utf-8") as f:
        raw_data = json.load(f)
  
    examples = []
    for d in raw_data:
        mapped_label = LABEL_MAP.get(d["label"], "Category_A")
        examples.append(
            dspy.Example(text=d["text"], label=mapped_label).with_inputs("text")
        )
    # ランダムにシャッフルして分割
    random.seed(SEED)
    random.shuffle(examples)
    # 学習用(TRAIN_SIZE) と 評価用(残り) に分割
    return examples[:TRAIN_SIZE], examples[TRAIN_SIZE:]

# ==============================================================================
# 2. DSPy Signature (タスクの定義書)
# ==============================================================================
class SupportClassification(dspy.Signature):
    """
    あなたはカスタマーサポートの問い合わせを分類するAIです。
    以下の「社内管理コード」の定義に基づいて、問い合わせを分類してください。

    【定義】
    - Category_A: 緊急対応案件（金銭トラブルやクレームなど、優先度「高」の対応）
    - Category_B: 事務手続き（契約内容の変更、アカウント更新、ステータス変更）
    - Category_C: システム不具合（エラーログの確認が必要なもの）
    - Category_D: ユーザーの声（機能への要望や感想）
  
    【注意】
    - 出力はカテゴリ名のみ（例: Category_A）にしてください。
    """
    # 入力フィールド：問い合わせ内容
    text = dspy.InputField(desc="問い合わせ内容")
    # 出力フィールド：分類ラベル
    label = dspy.OutputField(
        desc="分類結果 (Category_A, Category_B, Category_C, Category_D のいずれか)"
    )

# ==============================================================================
# 3. モジュール定義 (AIの思考プロセス)
# ==============================================================================
class ClassifierModule(dspy.Module):
    def __init__(self):
        super().__init__()
        # ChainOfThought「Reasoning (推論)」→「Answer (答え)」の順に出力
        self.prog = dspy.ChainOfThought(SupportClassification)
  
    def forward(self, text):
        # 入力テキストを受け取り、分類結果を返す
        return self.prog(text=text)

# ==============================================================================
# 4. 評価関数 (正解判定ロジック)
# ==============================================================================
def validate_answer(example, pred, trace=None):
    """AIの予測(pred)が正解(example)と一致しているか判定する"""
    if not getattr(pred, "label", None):
        return False
    # 余計な空白を除去して完全一致判定
    return example.label.strip() == pred.label.strip()

# ==============================================================================
# 5. メイン処理
# ==============================================================================
def main():
    print(f"OpenAIに接続中... (Key ends with: ...{api_key[-4:]})")
  
    # --- A. LLMの接続設定 ---
    # 対話モデル (生徒役): 実際にタスクをこなすモデル
    lm = dspy.LM(model=f"openai/{FAST_MODEL_NAME}", api_key=api_key, temperature=0.0)
    # 評価モデル (先生役): 最適化のアドバイスをする賢いモデル
    teacher_lm = dspy.LM(model=f"openai/{SMART_MODEL_NAME}", api_key=api_key, temperature=0.0)
  
    # DSPy全体の設定を適用
    dspy.settings.configure(lm=lm)

    # --- B. データのロード ---
    trainset, testset = load_data()
    print(f"Train (学習用): {len(trainset)}件, Test (評価用): {len(testset)}件")

    # --- C. 最適化前の評価 (Zero-shot Baseline) ---
    print("\n--- 1. 最適化前 (Zero-shot) の評価 ---")
  
    # Evaluateクラスを使ってテストデータ全件を評価
    evaluate = Evaluate(
        devset=testset,
        metric=validate_answer,
        num_threads=4, # 並列処理数
        display_progress=True
    )
  
    uncompiled_program = ClassifierModule()
  
    try:
        base_score = evaluate(uncompiled_program)
        print(f"Base Score: {base_score} (改善前の基準値)")
    except Exception as e:
        print(f"Zero-shot evaluation failed: {e}")
        return

    # --- D. MIPROv2 による最適化実行 ---
    print("\n--- 2. MIPROv2 による最適化を開始 ---")
    print("学習データを分析し、最も効果的な「事例」と「指示」を探します...")

    # オプティマイザの初期化
    teleprompter = MIPROv2(
        metric=validate_answer,   # 評価基準（正解率）
        prompt_model=teacher_lm,  # 評価モデル (先生)
        task_model=lm,            # 対話モデル (生徒)
        num_candidates=NUM_CANDIDATES, # プロンプトの候補をいくつ作るか
        init_temperature=INIT_TEMPERATURE, # 探索時の創造性
        auto=None                 # 自動設定をオフにして手動設定(num_trials)を有効化
    )
  
    # コンパイル（最適化の実行）
    optimized_program = teleprompter.compile(
        uncompiled_program,
        trainset=trainset,         # 学習データ
        max_bootstrapped_demos=MAX_BOOTSTRAPPED_DEMOS, # プロンプトに含める事例の最大数
        max_labeled_demos=MAX_LABELED_DEMOS,       # ラベル付き事例の最大数
        num_trials=NUM_TRIALS,             # 試行回数
        minibatch_size=MINIBATCH_SIZE      # 1回の試行で評価に使うデータ数
    )
  
    # --- E. 最適化後の評価 ---
    print("\n--- 3. 最適化完了! 評価を実行 ---")
    print("AIが生成したプロンプトを使って、再度テストデータを評価します。")
  
    opt_score = evaluate(optimized_program)
    print(f"Optimized Score: {opt_score} (改善後のスコア)")

    # --- F. 結果の確認と保存 ---
    print("\n--- 4. 生成されたプロンプト (抜粋) ---")
    # 実際にどのようなプロンプトが裏で使われているかを表示
    lm.inspect_history(n=1)

    # 次回以降使えるようにJSON形式で保存
    os.makedirs("artifacts_dspy", exist_ok=True)
    optimized_program.save("artifacts_dspy/optimized_dspy_program.json")
    print("\nプログラムを 'artifacts_dspy/optimized_dspy_program.json' に保存しました。")

if __name__ == "__main__":
    main()

TextGradの実装 (optimize_textgrad.py):

# optimize_textgrad.py
"""
TextGradによるプロンプト自動最適化スクリプト
"""
from __future__ import annotations

import json
import os
import random
import re
from typing import List, Dict, Tuple

from dotenv import load_dotenv
import textgrad as tg
from textgrad.optimizer import TextualGradientDescent
from textgrad.loss import MultiFieldEvaluation

load_dotenv()

# ==============================================================================
# 0. 設定定数 (環境変数およびパラメータ)
# ==============================================================================
DATA_PATH = "data/dataset.json"

# モデル設定 (.envから取得)
TASK_MODEL = os.getenv("FAST_MODEL", "gpt-4.1-mini")
JUDGE_MODEL = os.getenv("SMART_MODEL", "gpt-4.1")
print(f"Model Config -> Fast: {TASK_MODEL}, Smart: {JUDGE_MODEL}")

# 学習・最適化パラメータ
TEMPERATURE = 0.0
SEED = 30
TRAIN_SIZE = 20
STEPS = 12       # 最適化の更新回数
BATCH_SIZE = 4   # 1回の更新で見るデータ数


# ==============================================================================
# 1. データセットの準備 (独立したユーティリティ関数)
# ==============================================================================
LABEL_MAP = {
    "Refund": "Category_A",
    "Cancellation": "Category_B",
    "Technical_Issue": "Category_C",
    "Feature_Request": "Category_D",
}

def load_data_independent(path: str) -> List[Dict[str, str]]:
    """JSONデータを読み込む"""
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)

def split_dataset_independent(raw: List[Dict[str, str]]) -> Tuple[List[Dict[str, str]], List[Dict[str, str]]]:
    """データを学習用とテスト用に分割する"""
    ex = []
    for d in raw:
        code = LABEL_MAP.get(d["label"], "Category_A")
        ex.append({"text": d["text"], "label": code})
  
    random.seed(SEED)
    random.shuffle(ex)
    return ex[:TRAIN_SIZE], ex[TRAIN_SIZE:]

def build_user_message(text: str) -> str:
    """ユーザー入力のプロンプト形式"""
    return f"""【問い合わせ内容】
{text}

【分類】
"""

def normalize_category_strict(s: str) -> str:
    """
    判定ロジック
    """
    if not s:
        return ""
  
    clean_s = s.strip()
    # 完全一致 (fullmatch) で判定 # 例: "Category_A" -> OK, "Category_A." -> OK, "Answer: Category_A" -> NG
    m = re.fullmatch(r"(Category_[A-D])\.?", clean_s)
    if m:
        return m.group(1)
    return ""

def predict_category(model: tg.BlackboxLLM, text: str) -> str:
    """モデル予測を実行"""
    user_msg = build_user_message(text)
    user_var = tg.Variable(user_msg, requires_grad=False, role_description="user input")
  
    # System Promptは model 初期化時に渡されているため、ここでは user_var のみ渡す
    pred_var = model(user_var)
    return normalize_category_strict(str(pred_var.value))

def accuracy_on(dataset: List[Dict[str, str]], model: tg.BlackboxLLM) -> float:
    """正解率計算"""
    ok = 0
    for d in dataset:
        pred = predict_category(model, d["text"])
        if pred == d["label"]:
            ok += 1
    return ok / max(1, len(dataset))

# ==============================================================================
# 2. メイン処理
# ==============================================================================
def main():
    api_key = os.getenv("OPENAI_API_KEY")
    if not api_key:
        raise ValueError("OPENAI_API_KEY が見つかりません")

    random.seed(SEED)
    print(f"OpenAIに接続中...")

    # エンジン定義 (環境変数から読み込んだモデル名を使用)
    task_engine = tg.get_engine(TASK_MODEL, temperature=TEMPERATURE)
    judge_engine = tg.get_engine(JUDGE_MODEL, temperature=TEMPERATURE)

    # 先生役 (Backward Engine) をセット
    tg.set_backward_engine(judge_engine, override=True)
  
    # データ準備
    train, test = split_dataset_independent(load_data_independent(DATA_PATH))
    print(f"Train (学習用): {len(train)}件 / Test (評価用): {len(test)}件")

    # ---------------------------------------------------------
    # 初期システムプロンプト 
    # ---------------------------------------------------------
    initial_prompt_text = """あなたはカスタマーサポートの問い合わせを分類するAIです。
以下の「社内管理コード」の定義に基づいて、問い合わせを分類してください。

【定義】
- Category_A: 緊急対応案件（金銭トラブルやクレームなど、優先度「高」の対応）
- Category_B: 事務手続き（契約内容の変更、アカウント更新、ステータス変更）
- Category_C: システム不具合（エラーログの確認が必要なもの）
- Category_D: ユーザーの声（機能への要望や感想）

【注意】
- 出力はカテゴリ名のみ（例: Category_A）にしてください。"""

    # 変数化 (学習対象)
    system_prompt = tg.Variable(
        initial_prompt_text,
        requires_grad=True,
        role_description="system prompt for classification"
    )

    # モデル構築
    model = tg.BlackboxLLM(task_engine, system_prompt)

    # 評価者への指示
    evaluation_instruction = tg.Variable(
        """You are a strict evaluator for a Japanese customer support classification task.
You will be given:
(1) the system prompt (in Japanese),
(2) the user inquiry (in Japanese),
(3) the model prediction,
(4) the ground-truth category.

If the prediction is correct: output only "Correct."
If incorrect (including formatting errors): analyze WHY the error occurred based on the system prompt's definition or instructions.
Then, provide short, actionable feedback to improve the SYSTEM PROMPT.
For example, suggest adding specific exclusion rules or formatting constraints.
""",
        requires_grad=False,
        role_description="evaluation instruction"
    )

    # 評価関数の定義
    evaluator = MultiFieldEvaluation(
        evaluation_instruction=evaluation_instruction,
        role_descriptions=[
            "system prompt (to be improved)",
            "user inquiry",
            "model prediction",
            "ground truth"
        ],
        engine=judge_engine
    )

    # オプティマイザ
    optimizer = TextualGradientDescent(parameters=[system_prompt])

    # --- 1. Baseline 評価 ---
    print("\nベースライン評価 (Zero-shot) を実行中...")
    base_acc = accuracy_on(test, model)
    print(f"[初期状態] テスト正解率: {base_acc*100:.2f}%")

    # --- [追加] ベストスコア初期化 ---
    # ベースラインを暫定1位として記録しておく
    best_acc = base_acc
    best_prompt_text = str(system_prompt.value)
    # -------------------------------

    # --- 2. 最適化ループ ---
    print("\n最適化を開始します...")
  
    for step in range(1, STEPS + 1):
        batch = random.sample(train, k=min(BATCH_SIZE, len(train)))
        optimizer.zero_grad()
  
        for d in batch:
            # 入力と正解を変数化
            user_msg = build_user_message(d["text"])
            user_var = tg.Variable(user_msg, requires_grad=False, role_description="user inquiry")
            gold_var = tg.Variable(d["label"], requires_grad=False, role_description="ground truth")
      
            # Forward
            pred_var = model(user_var)
      
            # Loss計算 (批判生成)
            loss = evaluator(
                inputs=[
                    system_prompt,
                    user_var,
                    pred_var,
                    gold_var,
                ]
            )
            # Backward
            loss.backward()

        # Update
        optimizer.step()

        # 途中評価
        test_acc = accuracy_on(test, model)
        print(f"[ステップ {step:02d}] テスト正解率: {test_acc*100:.2f}%")

        # --- ベストスコア更新チェック ---
        if test_acc > best_acc:
            best_acc = test_acc
            best_prompt_text = str(system_prompt.value)
            print(f"  ★ ベストスコア更新! ({best_acc*100:.2f}%) - 現在の状態をメモリに保存しました。")
        # -----------------------------------

    # --- 3. 結果保存 ---
    os.makedirs("artifacts_textgrad", exist_ok=True)
    out_path = "artifacts_textgrad/optimized_textgrad_prompt.txt"
    with open(out_path, "w", encoding="utf-8") as f:
        f.write(str(system_prompt.value))

    print("\n==== 完了 ====")
    print(f"最適化されたシステムプロンプトを保存しました: {out_path}")
    print("\n--- 最終的なシステムプロンプト ---\n")
    print(system_prompt.value)

    # --- ベストプロンプトのファイル保存 ---
    # 最終的なプロンプトとは別に、最も精度が高かったプロンプトを保存する
    best_out_path = "artifacts_textgrad/optimized_textgrad_prompt_best.txt"
    with open(best_out_path, "w", encoding="utf-8") as f:
        f.write(best_prompt_text)
  
    print("\n--- [INFO] ベストチェックポイント ---")
    print(f"ベストシステムプロンプトを保存しました (スコア: {best_acc*100:.2f}%): {best_out_path}")
    # ----------------------------------------

if __name__ == "__main__":
    main()

4.2 DSPy実行結果

optimize_dspy.py の実行ログを分析し、DSPyの挙動を確認します。

ログからは、MIPROv2 オプティマイザが数分間の探索を経て、「最適な事例（Few-shot）」と「指示（Instruction）」の組み合わせを発見するプロセスが読み取れました。

① スコア改善の推移

まず定量的な成果を確認します。人間が作成した定義のみの状態（Zero-shot）と比較して、DSPyによる最適化後の正解率は 約6.7ポイント向上 しました。

評価フェーズ	正解数 / データ数	正解率 (Accuracy)	備考
Before (Zero-shot)	47 / 60	78.3%	人間が書いた定義のみ
Training Best	14 / 16	87.5%	最適化中の最高スコア
After (Optimized)	51 / 60	85.0%	+6.7% 改善

② 最適化プロセスのログ（抜粋）

MIPROv2 が裏側で実行している処理は、ログから詳細に追跡できます。 初期の単純な指示（Instruction 0）からスタートし、徐々に複雑な指示を生成・テストしています。

--- 1. 最適化前 (Zero-shot) の評価 ---
Average Metric: 47.00 / 60 (78.3%)

--- 2. MIPROv2 による最適化を開始 ---
...
INFO: Proposed Instructions for Predictor 0:
INFO: 0: あなたは...（シンプルな定義）
INFO: 5: あなたは大手カスタマーサポートセンターのAIオペレーターとして...分類ミスが発生すると...信頼失墜につながるリスクがあります...
...

--- 3. 最適な組み合わせの探索 ---
# Trial 37: Full Evaluation
INFO: Returning best identified program with score 87.5!

--- 4. 生成されたプロンプト (抜粋) ---
[[ ## reasoning ## ]]
The inquiry is about canceling the service... The user explicitly states no refund is needed... fitting the definition of administrative procedures.

③ 生成されたプロンプト

DSPyが最終的に採用したプロンプトには、2つの大きな特徴が見られました。それは「 リスクを意識させるペルソナ指示 」と「 解約と返金の境界線を教える事例（Few-shot） 」の組み合わせです。

• ペルソナ指示： 採用された指示文（Instruction 5）は、単にタスクを説明するだけでなく、AIに対して 「責任」 を説いています。

  Instruction 5 (抜粋): "あなたは大手カスタマーサポートセンターのAIオペレーターとして... 分類ミスが発生すると、金銭トラブルや重大なクレームへの対応が遅れ、顧客満足度の低下や企業の信頼失墜につながるリスクがあります。...慎重かつ論理的に分類してください。"

このように「間違えた場合のリスク」を言語化することで、LLMの注意力を高め、安易なキーワード反応（返金＝Aなど）を抑制する効果が働いたと考えられます。

• 事例： プロンプトに含まれるFew-shot事例には、「解約と返金の切り分け」に関する明確な推論が含まれていました。

[[ ## text ## ]]
今月中にプロジェクトが終了するため...返金などは不要ですが...解約処理だけ行ってください。

[[ ## reasoning ## ]]
...The user explicitly states no refund is needed... This is a procedural request... fitting the definition of administrative procedures.
（ユーザーは返金不要と明言しており...これは事務手続きに該当します）

[[ ## label ## ]]
Category_B

DSPyは、「返金」という単語が含まれていてもCategory_A（返金要求）ではない、という論理構造を含んだ事例をピンポイントで選択し、プロンプトに埋め込むことで精度を向上させました。

4.3 TextGrad実行結果

続いて optimize_textgrad.py の実行ログを分析します。DSPyが「事例（Few-shot）＋探索」で正解に近づくのに対し、TextGradは システムプロンプト（定義文）そのものを段階的に書き換える アプローチをとります。ログを追うと、あたかも人間が“業務マニュアルを改訂していく”かのような挙動が見て取れました。

① スコア改善の推移

まず定量的な推移を確認します。TextGradはZero-shotの状態（正解率 78.33% ）からスタートし、中盤で最高精度（約15ポイント向上）に達しましたが、後半では不安定な挙動を示しました。

評価フェーズ	正解率 (Accuracy)	備考
Before (Zero-shot)	78.33%	初期の定義文のみ
Best (途中ステップ最高)	93.33%（Step 08, 10）	ルールが噛み合い最高値を記録
After（最終ステップ）	68.33%（Step 12）	過剰なルール追加により失速

特筆すべき点は、TextGradにおいては「最終ステップ＝最良」とは限らないことです。今回は Step 8 でピーク（93.33%）に達した にもかかわらず、Step 9で急落し、Step 10で持ち直すなど、更新を続けることで逆に精度が不安定になる現象が見られました。これは、学習率の調整や更新回数の制限、early stopping（早期終了）の導入が必要な典型的なパターンと言えます。

② 最適化プロセスのログ（抜粋）

実行ログは以下の通りです。StepごとにTest Accuracyが大きく変動しており、「良いルール」と「悪い修正」が行き来する 様子が明確に表れています。

Evaluating Baseline (Zero-shot)...
[Before] Test Accuracy: 78.33%

Starting Optimization...
[Step 01] Test Accuracy: 70.00%
[Step 02] Test Accuracy: 81.67%
  ★ ベストスコア更新!
[Step 03] Test Accuracy: 83.33%
...
[Step 07] Test Accuracy: 88.33%
[Step 08] Test Accuracy: 93.33% <-- ピーク到達
  ★ ベストスコア更新!
[Step 09] Test Accuracy: 75.00% <-- 急落（改悪）
[Step 10] Test Accuracy: 93.33% <-- 再浮上
[Step 11] Test Accuracy: 75.00%
[Step 12] Test Accuracy: 70.00%

③ 生成されたプロンプト（最適解）

以下は探索過程で最も高い精度（93.33%）を記録した、最適なプロンプト の抜粋です。

このプロンプトを確認すると、TextGradが単なる「言い回しの修正」ではなく、「判断アルゴリズムの言語化」 を行っていることがわかります。特に、人間が判断に迷うポイント（複合的な意図や、感情的な文脈）に対する処理手順が、明確なルールとして定義されています。

【最重要原則：現時点の明確な要求のみを分類根拠とする】
- 分類は必ず「現時点で明確に表明された要求・要望・依頼」のみを根拠とし、下記「分類根拠に含めない要素」に該当する内容は一切分類根拠にしない。
- 「主旨・意図」とは「ユーザーが現時点で最も求めていること、明確に要求していること」を指す。主旨抽出が曖昧な場合は、下記「主旨抽出手順」に従うこと。

【分類根拠に含めない要素（共通ルール）】
- 仮定・将来的な可能性（例：「〜かもしれません」「〜予定はありますか？」のみ）
- 背景説明・感情的表現・脅し表現（例：「〜気分」「〜と考えています」「もし改善されなければ解約も検討」等）
- 丁寧語・間接表現・辞退された要求・副次的な言及・明確な要求ではない内容
- 外部知識・推測・問い合わせ文に含まれない情報
- 要望の理由や背景（例：「著作権保護のため」「セキュリティ強化のため」等）は分類根拠に含めない

【主旨抽出手順（複数話題・曖昧な場合の判断フロー）】
1. 問い合わせ文中の「明確な要求・依頼・要望」をすべて抽出する。
2. 上記「分類根拠に含めない要素」に該当する部分を除外する。
3. 残った要求が複数ある場合は、最も明確かつ具体的な現時点の要求を優先する。
4. 複数の要求が「AができなければB」等の条件付き・代替・相互排他的な場合は、現時点で実行可能な最上位の要求（通常は第一選択肢）を主分類根拠とし、該当しない場合は次の選択肢を考慮する。
5. それでも判断がつかない場合は、定義順（A→B→C→D）の下位カテゴリを選択。
6. 明確な要求が全く読み取れない場合のみ「Uncategorized」とする。

【出力フォーマット厳守ルール】
- 出力は必ず「Category_A」「Category_B」「Category_C」「Category_D」「Uncategorized」のいずれかを、半角英字・大文字・1行のみ、余計な空白・説明・日本語・句読点・改行なしで出力すること。
- **出力例: Category_C**
- NG例（絶対に出力しないこと）: 「Category_Aです」「Category_B 」「category_c」「Category_C\n」「 Category_D」「Uncategorized.」など
- 出力直前に、余計な空白・全角文字・改行・句読点・不可視文字・ゼロ幅スペース・Unicodeバリアントが混入していないか必ずダブルチェックし、エラーがあれば必ず再出力すること。

【分類カテゴリ定義・具体例・否定例】
- Category_A: 緊急対応案件（例：金銭トラブル、重大なクレーム、サービス停止、不正アクセス・情報漏洩等の実際のセキュリティインシデント報告や即時対応要求、誇大広告・詐欺を理由とした返金要求、金銭的補償の明確な要求）
  - 含まない例：単なる手続き依頼や要望のみ、将来的な返金・解約・補償・クレーム等の可能性を示唆するのみ、辞退・保留・代替要求（例：「返金は不要」「本来なら返金を求めたいが今回は不要」等）
  - 【注意】「緊急」「至急」「必須」「重要」等の表現があっても、主な要求が技術的サポートやシステム不具合の解決であればCategory_Aには分類しない。金銭的要求が明確な場合のみCategory_A。
  - 【例】「広告と違うので返金してほしい」「不正請求なので課金を取り消してほしい」「納得がいかないので返金してほしい」
  - 【否定例】「もし改善されなければ解約も検討します」（現時点で明確な要求がなければ分類根拠としない）

…＜略＞…

生成されたプロンプトには、以下の3つの特徴的な進化が見られました。

1. 除外条件（Exclusion Criteria）の明文化
   • 「感情表現」や「将来の仮定」はノイズであると定義し、分類根拠から除外するルールを追加しました。
2. 判断フロー（Procedure）の構築
   • 「主旨抽出手順」としてまず全要素を抽出し、次にノイズを除去し、最後に優先順位を適用するという、ステップバイステップのアルゴリズム を定義しました。
3. フォーマットの厳格化
   • 曖昧な回答を防ぐため出力形式を厳密に規定し、エラーチェックの手順まで含めました。

TextGradは「 どう考えれば間違えないか 」という手順書を自ら書き上げたと言えます。これにより、曖昧な問い合わせに対するLLMの判断ブレを大幅に抑制することに成功しました。

---

5. 結論と選定指針

本記事ではプロンプトエンジニアリングの自動化を担う2つの主要ツールを比較検証しました。

結論として、両者はアプローチが根本的に異なります。DSPy はプログラムとして構成要素（事例や指示）の組み合わせを探索するのに対し、TextGrad は人間のように言葉で定義文（Instruction）を書き換えることで改善を図ります。

5.1 DSPyの拡張性とオプティマイザの選択

DSPyの最大の特徴は、目的に応じて オプティマイザ（最適化アルゴリズム） を使い分けられる点です。本検証では指示と事例の両方を探索する MIPROv2 を使用しましたが、DSPyには他にも多様なアプローチが用意されています。

• BootstrapFewShot: 指示文は固定し、効果的なFew-shot事例の組み合わせのみを探索する。
• COPRO / OPRO: 事例ではなく、指示文（Instruction）の言い回し自体を最適化する。
• MIPROv2: 事例と指示文の両方を、ベイズ最適化を用いて包括的に探索する（今回の検証で使用）。

開発者は、タスクの性質や利用可能なデータ量に合わせて、これらを選択・適用できます。

5.2 ツール選定の指針

検証結果と各ツールの特性を踏まえ、推奨されるユースケースを以下に整理しました。

ツール	最適化の特性	推奨ユースケース
DSPy	構成要素の探索（オプティマイザに依存）	本番運用・システム開発：開発者がモジュールやオプティマイザを定義し、再現性の高いパイプラインを構築したい場合に向きます。特に、言葉で説明しにくいニュアンスを「良質な事例（Few-shot）」の組み合わせで吸収させたいタスクに強みを発揮します。
TextGrad	定義の言語化（Instructionの更新）	分析・要件定義・プロトタイピング：LLMがなぜ間違えるのかを言語化し、仕様の抜け漏れや曖昧さを発見したい場合に向きます。「人間が読んでメンテナンス可能なマニュアル」を育て上げたいフェーズで特に有効です。

5.3 最後に：「職人芸」からの脱却

これまで「職人芸」に依存しがちだったプロンプト調整は、こうしたツールを活用することで、より再現性の高い「エンジニアリング（工学）」へと進化しています。

• DSPy で、データに基づいた最適なパラメータ（事例・指示）を探索する。
• TextGrad で、モデルの失敗理由を言語化させ、仕様書（プロンプト）の品質を高める。

開発者はタスクのフェーズに合わせて適切な道具を選定することで、不毛な「言い回し調整」から解放され、本来注力すべき「本質的な課題解決」に時間を割けるようになるはずです。

もし「プロンプト調整に疲れたな...」と感じたら、ぜひ一度、これらのツールに頼ってみてください。AIがAIを育てる新しい開発体験は、きっと皆さんのエンジニアリングに新しい風を吹き込んでくれるはずです。

---

【お知らせ】

昨年度のアドベントカレンダーでは、「同期現象の数理モデルをPythonで実装してみた」 というテーマで執筆しました。もしよろしければ、こちらも合わせてご覧いただけると嬉しいです！

nttdocomo-developers.jp

最後までお読みいただき、ありがとうございました！