Il s’agit d’un notebook interactif. Vous pouvez l’exécuter en local ou utiliser les liens suivants : Installer les dépendances
- DSPy pour construire et optimiser le flux de travail LLM.
- Weave pour suivre le flux de travail LLM et évaluer les stratégies de prompting.
- datasets pour accéder au jeu de données BIG-Bench Hard depuis le Hub Hugging Face.
!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
Activer le suivi avec Weave
weave.init au début de votre code permet de tracer automatiquement vos fonctions DSPy, que vous pouvez ensuite explorer dans l’interface Weave. Pour plus d’informations, voir la documentation de l’intégration Weave pour DSPy.
import weave
weave.init(project_name="dspy-bigbench-hard")
weave.Object pour gérer les métadonnées.
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()
Gestion des versions des objets : les objets Metadata font automatiquement l’objet d’une gestion des versions et sont tracés lorsque les fonctions qui les utilisent sont elles-mêmes tracées.
Charger le jeu de données BIG-Bench Hard
weave.Evaluation pour évaluer votre stratégie de prompting.
import dspy
from datasets import load_dataset
@weave.op()
def get_dataset(metadata: Metadata):
# charger le jeu de données BIG-Bench Hard correspondant à la tâche depuis Huggingface Hub
dataset = load_dataset(metadata.dataset_address, metadata.big_bench_hard_task)[
"train"
]
# créer les jeux de données d'entraînement et de validation
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 :]
# créer les exemples d'entraînement et de validation composés d'objets `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]
# publier les jeux de données sur Weave, ce qui nous permettra de versionner les données et de les utiliser pour l'évaluation
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 pour configurer le modèle de langage et dspy.configure pour le définir comme valeur par défaut.
llm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=llm)
Écrire la signature du raisonnement causal
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, sur un exemple tiré du sous-ensemble de raisonnement causal de BIG-Bench Hard.
import rich
baseline_module = CausalReasoningModule()
prediction = baseline_module(dspy_train_examples[0]["question"])
rich.print(prediction)
Évaluez le programme DSPy
weave.Evaluation, avec une métrique qui compare la réponse prédite à la réponse de référence. Weave prend chaque exemple, le fait passer dans votre application et attribue un score à la sortie à l’aide de plusieurs fonctions de score personnalisées. Cela vous donne une vue des performances de votre application, ainsi qu’une UI riche pour examiner en détail chaque sortie et son score.
Créez d’abord une fonction de score qui détermine si la réponse prédite correspond à la réponse de référence. Les fonctions de score Weave reçoivent la valeur de retour du modèle sous la forme de output, ainsi que toutes les clés correspondantes de l’exemple du jeu de données en arguments supplémentaires. Ici, answer provient du jeu de données et output est le dictionnaire renvoyé par CausalReasoningModule.forward.
@weave.op()
def weave_evaluation_scorer(answer: str, output: dict) -> dict:
return {"match": int(answer.lower() == output["answer"].lower())}
weave.Evaluation peut appeler. Les noms des arguments de cette fonction d’encapsulation doivent correspondre aux noms des colonnes du jeu de données utilisées en entrée par le modèle.
@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)
Si vous exécutez ce code depuis un script Python, vous pouvez utiliser le code suivant pour lancer l’évaluation :import asyncio
asyncio.run(evaluation.evaluate(predict))
Exécuter l’évaluation sur le jeu de données de raisonnement causal coûte environ 0,24 $ en crédits OpenAI.
Optimiser le programme DSPy
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)
L’exécution de l’évaluation sur le jeu de données de raisonnement causal coûte environ 0,04 $ en crédits OpenAI.
@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)
