これはインタラクティブなノートブックです。ローカルで実行することも、以下のリンクを使用することもできます。 DSPy と Weave を使用した LLM ワークフローの最適化
- 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="回答すべき質問")
answer: str = dspy.OutputField(desc="Yes or No")
confidence: float = dspy.OutputField(desc="0から1の間の信頼スコア")
explanation: str = dspy.OutputField(desc="回答の詳細な説明")
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)
DSPy プログラムを評価する
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))
causal reasoning データセットで評価を実行すると、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)
