これはインタラクティブなノートブックです。ローカルで実行するか、以下のリンクをご利用ください。
- DSPy：LLMワークフローの構築と最適化に使用します。
- Weave：LLMワークフローをトラッキングし、プロンプト戦略を評価するために使用します。
- datasets：HuggingFace Hub から BIG-Bench Hard データセットにアクセスするために使用します。
!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
weave.init を呼び出すと、DSPy の関数が自動的にトレースされ、その内容を Weave UI で確認できます。詳しくは、DSPy 向け Weave インテグレーションのドキュメント を参照してください。
import weave
weave.init(project_name="dspy-bigbench-hard")
weave.Object を継承したメタデータクラスを使用します。
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()
オブジェクトのバージョン管理: Metadata オブジェクトは、それらを利用する関数がトレースされると、自動的にバージョン管理され、あわせてトレースされます。
BIG-Bench Hardデータセットを読み込む
weave.Evaluationを使用してプロンプト戦略を評価することもできます。
import dspy
from datasets import load_dataset
@weave.op()
def get_dataset(metadata: Metadata):
# HuggingFace HubからタスクにあったBIG-Bench Hardデータセットを読み込む
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)
dspy.LM を使用し、それをデフォルトに設定するには dspy.configure を使用します。
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,
}
CausalReasoningModule をテストします。
import rich
baseline_module = CausalReasoningModule()
prediction = baseline_module(dspy_train_examples[0]["question"])
rich.print(prediction)
weave.Evaluation で検証セットを評価します。Weave は各例をアプリケーションに渡し、複数のカスタムスコアリング関数で出力を採点します。これにより、アプリケーションのパフォーマンスを把握できるほか、個々の出力やスコアを詳しく確認できる充実した UI も利用できます。
まず、予測した回答が正解と一致するかどうかを判定するスコアリング関数を作成します。Weave のスコアリング関数は、モデルの戻り値を output として受け取り、さらにデータセット例内の一致するキーを追加の引数として受け取ります。ここでは、answer はデータセットから取得され、output は CausalReasoningModule.forward が返す dict です。
@weave.op()
def weave_evaluation_scorer(answer: str, output: dict) -> dict:
return {"match": int(answer.lower() == output["answer"].lower())}
weave.Evaluation から呼び出せるように、モジュールをトレース対象の関数でラップします。ラッパーの引数名は、モデルが受け取るデータセットの列名と一致している必要があります。
@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)
Python スクリプトから実行している場合は、次のコードを使用して評価を実行できます。import asyncio
asyncio.run(evaluation.evaluate(predict))
因果推論データセットで評価を実行するには、OpenAI クレジットが約 $0.24 かかります。
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)
因果推論の評価用データセットを実行すると、OpenAI クレジットが約 $0.04 かかります。
@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)
