プロンプト、手直す時代を、ぶっ壊す👍——DSPyで始める自動チューニング

$ pip install dspy openai

従来のプロンプトチューニング:
  プロンプトを手で書く → 試す → 結果を見る → 手で直す → 試す → ...

DSPy:
  入出力を定義する → 処理フローを組む → 評価関数を作る → オプティマイザに渡す → 最適化されたプロンプトが出てくる

import dspy

class ReviewAnalysis(dspy.Signature):
    """商品レビューを分析して、感情・評価ポイント・改善提案を出力する"""
    review_text = dspy.InputField(desc="商品レビューの本文")
    sentiment = dspy.OutputField(desc="positive または negative")
    key_points = dspy.OutputField(desc="レビューの主な評価ポイント（箇条書き）")
    suggestion = dspy.OutputField(desc="商品やサービスの改善提案")

class ReviewAnalyzer(dspy.Module):
    def __init__(self):
        super().__init__()
        self.analyze = dspy.ChainOfThought(ReviewAnalysis)

    def forward(self, review_text: str) -> dspy.Prediction:
        return self.analyze(review_text=review_text)

class SentimentJudge(dspy.Signature):
    """レビューの感情を判定する"""
    review_text = dspy.InputField(desc="商品レビューの本文")
    sentiment = dspy.OutputField(desc="positive または negative")
    key_points = dspy.OutputField(desc="レビューの主な評価ポイント（箇条書き）")

class ImprovementAdvisor(dspy.Signature):
    """感情判定の結果をもとに改善提案を行う"""
    review_text = dspy.InputField(desc="商品レビューの本文")
    sentiment = dspy.InputField(desc="判定された感情")
    key_points = dspy.InputField(desc="評価ポイント")
    suggestion = dspy.OutputField(desc="商品やサービスの改善提案")

class TwoStepAnalyzer(dspy.Module):
    def __init__(self):
        super().__init__()
        self.judge = dspy.ChainOfThought(SentimentJudge)
        self.advisor = dspy.Predict(ImprovementAdvisor)

    def forward(self, review_text: str) -> dspy.Prediction:
        judgment = self.judge(review_text=review_text)
        advice = self.advisor(
            review_text=review_text,
            sentiment=judgment.sentiment,
            key_points=judgment.key_points,
        )
        return dspy.Prediction(
            sentiment=judgment.sentiment,
            key_points=judgment.key_points,
            suggestion=advice.suggestion,
        )

import dspy

trainset = [
    dspy.Example(
        review_text="画面がきれいで動作もサクサク。バッテリーも1日持つので満足です。",
        sentiment="positive",
        key_points="画面品質が高い、動作が快適、バッテリー持ちが良い",
    ).with_inputs("review_text"),
    dspy.Example(
        review_text="買って1週間で充電できなくなった。サポートに連絡しても返答が遅い。",
        sentiment="negative",
        key_points="充電不良、サポート対応が遅い",
    ).with_inputs("review_text"),
    dspy.Example(
        review_text="デザインは良いけど重すぎる。長時間持っていると手が疲れる。",
        sentiment="negative",
        key_points="デザインは良い、重量が重い、長時間使用に不向き",
    ).with_inputs("review_text"),
    # ... 実際にはもっと多くのデータを用意する
]

import dspy

class PointsQuality(dspy.Signature):
    """評価ポイントの品質を判定する"""
    review_text = dspy.InputField(desc="元のレビュー本文")
    predicted_points = dspy.InputField(desc="モデルが出力した評価ポイント")
    reference_points = dspy.InputField(desc="正解の評価ポイント")
    score = dspy.OutputField(desc="0から10のスコア", format=int)

points_evaluator = dspy.ChainOfThought(PointsQuality)

def review_metric(example, prediction, trace=None):
    # 感情判定の正誤（完全一致）
    sentiment_correct = (
        example.sentiment.strip().lower() == prediction.sentiment.strip().lower()
    )

    # 評価ポイントの品質をLLMで採点
    with dspy.context(lm=eval_lm):
        quality = points_evaluator(
            review_text=example.review_text,
            predicted_points=prediction.key_points,
            reference_points=example.key_points,
        )

    points_score = min(10, max(0, int(quality.score))) / 10.0

    return (1.0 if sentiment_correct else 0.0) * 0.5 + points_score * 0.5

import dspy

# 推論用のモデル
lm = dspy.LM(model="openai/gpt-4o-mini")
dspy.configure(lm=lm)

# 評価用のモデル（推論用と分ける）
eval_lm = dspy.LM(model="openai/gpt-4o")

# Moduleのインスタンス化
analyzer = TwoStepAnalyzer()

# オプティマイザの設定
optimizer = dspy.MIPROv2(
    metric=review_metric,
    prompt_model=eval_lm,
    auto="light",
    max_bootstrapped_demos=2,
    max_labeled_demos=2,
)

# 最適化の実行
optimized_analyzer = optimizer.compile(
    analyzer,
    trainset=trainset,
)

test_review = "音質は良いけど、ノイズキャンセリングが弱い。通勤電車だと外の音がかなり入ってくる。"

# 最適化前
before = analyzer(review_text=test_review)
print("【最適化前】")
print(f"感情: {before.sentiment}")
print(f"評価ポイント: {before.key_points}")
print(f"改善提案: {before.suggestion}")

print("---")

# 最適化後
after = optimized_analyzer(review_text=test_review)
print("【最適化後】")
print(f"感情: {after.sentiment}")
print(f"評価ポイント: {after.key_points}")
print(f"改善提案: {after.suggestion}")

# 保存
optimized_analyzer.save("artifact/review_analyzer")

# 読み込み
loaded_analyzer = TwoStepAnalyzer()
loaded_analyzer.load("artifact/review_analyzer")