DSPy によるプロンプト最適化

!pip install -qU dspy weave "datasets<4"

import os
from getpass import getpass

api_key = getpass("Enter you OpenAI API key: ")
os.environ["OPENAI_API_KEY"] = api_key

import weave

weave.init(project_name="dspy-bigbench-hard")

class Metadata(weave.Object):
    dataset_address: str = "maveriq/bigbenchhard"
    big_bench_hard_task: str = "causal_judgement"
    num_train_examples: int = 50
    openai_model: str = "gpt-4o-mini"
    openai_max_tokens: int = 2048
    max_bootstrapped_demos: int = 8
    max_labeled_demos: int = 8

metadata = Metadata()

import dspy
from datasets import load_dataset

@weave.op()
def get_dataset(metadata: Metadata):
    # タスクに対応する BIG-Bench Hard データセットを Huggingface Hub から読み込みます
    dataset = load_dataset(metadata.dataset_address, metadata.big_bench_hard_task)[
        "train"
    ]

    # トレーニング用と検証用のデータセットを作成します
    rows = [{"question": data["input"], "answer": data["target"]} for data in dataset]
    train_rows = rows[0 : metadata.num_train_examples]
    val_rows = rows[metadata.num_train_examples :]

    # `dspy.Example` オブジェクトからなるトレーニング用と検証用のサンプルを作成します
    dspy_train_examples = [
        dspy.Example(row).with_inputs("question") for row in train_rows
    ]
    dspy_val_examples = [dspy.Example(row).with_inputs("question") for row in val_rows]

    # データセットを Weave にパブリッシュします。これにより、データをバージョン管理して評価に使用できます
    weave.publish(
        weave.Dataset(
            name=f"bigbenchhard_{metadata.big_bench_hard_task}_train", rows=train_rows
        )
    )
    weave.publish(
        weave.Dataset(
            name=f"bigbenchhard_{metadata.big_bench_hard_task}_val", rows=val_rows
        )
    )

    return dspy_train_examples, dspy_val_examples

dspy_train_examples, dspy_val_examples = get_dataset(metadata)

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

class CausalReasoning(dspy.Signature):
    """You are an expert in causal reasoning. Analyze the given question carefully
    and answer Yes or No. Provide a detailed explanation justifying your answer."""

    question: str = dspy.InputField(desc="The question to be answered")
    answer: str = dspy.OutputField(desc="Yes or No")
    confidence: float = dspy.OutputField(desc="Confidence score between 0 and 1")
    explanation: str = dspy.OutputField(desc="Detailed explanation for the answer")

class CausalReasoningModule(dspy.Module):
    def __init__(self):
        self.prog = dspy.Predict(CausalReasoning)

    @weave.op()
    def forward(self, question: str) -> dict:
        result = self.prog(question=question)
        return {
            "answer": result.answer,
            "confidence": result.confidence,
            "explanation": result.explanation,
        }

import rich

baseline_module = CausalReasoningModule()

prediction = baseline_module(dspy_train_examples[0]["question"])
rich.print(prediction)

@weave.op()
def weave_evaluation_scorer(answer: str, output: dict) -> dict:
    return {"match": int(answer.lower() == output["answer"].lower())}

@weave.op()
def predict(question: str) -> dict:
    return baseline_module(question=question)

validation_dataset = weave.ref(
    f"bigbenchhard_{metadata.big_bench_hard_task}_val:v0"
).get()

evaluation = weave.Evaluation(
    name="baseline_causal_reasoning_module",
    dataset=validation_dataset,
    scorers=[weave_evaluation_scorer],
)

await evaluation.evaluate(predict)

from dspy.teleprompt import BootstrapFewShot

@weave.op()
def get_optimized_program(model: dspy.Module, metadata: Metadata) -> dspy.Module:
    @weave.op()
    def dspy_evaluation_metric(true, prediction, trace=None):
        return prediction["answer"].lower() == true.answer.lower()

    teleprompter = BootstrapFewShot(
        metric=dspy_evaluation_metric,
        max_bootstrapped_demos=metadata.max_bootstrapped_demos,
        max_labeled_demos=metadata.max_labeled_demos,
    )
    return teleprompter.compile(model, trainset=dspy_train_examples)

optimized_module = get_optimized_program(baseline_module, metadata)

@weave.op()
def predict_optimized(question: str) -> dict:
    return optimized_module(question=question)

evaluation = weave.Evaluation(
    name="optimized_causal_reasoning_module",
    dataset=validation_dataset,
    scorers=[weave_evaluation_scorer],
)

await evaluation.evaluate(predict_optimized)