python3 -m venv env
source env/bin/activate
pip install dspy-ai
pip install openai --upgrade
pip install --upgrade notebook ipywidgets

import os
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

['tracking_shuffled_objects_seven_objects', 'salient_translation_error_detection', 'tracking_shuffled_objects_three_objects', 'geometric_shapes', 'object_counting', 'word_sorting', 'logical_deduction_five_objects', 'hyperbaton', 'sports_understanding', 'logical_deduction_seven_objects', 'multistep_arithmetic_two', 'ruin_names', 'causal_judgement', 'logical_deduction_three_objects', 'formal_fallacies', 'snarks', 'boolean_expressions', 'reasoning_about_colored_objects', 'dyck_languages', 'navigate', 'disambiguation_qa', 'temporal_sequences', 'web_of_lies', 'tracking_shuffled_objects_five_objects', 'penguins_in_a_table', 'movie_recommendation', 'date_understanding']

import dspy
from datasets import load_dataset

ds = load_dataset("maveriq/bigbenchhard", "penguins_in_a_table")["train"]
examples = [dspy.Example({"question": r["input"], "answer": r["target"]}).with_inputs("question") for r in ds]
print(f"There are {len(examples)} examples.")
trainset = examples[0:20]
valset = examples[20:]

example = trainset[10]
for k, v in example.items():
    print(f"\n{k.upper()}:\n")
    print(v)

We will use the DSPy OpenAI connector to make calls to gpt-3.5. Note that DSPy caches API calls so that subsequent calls with the same input will read from the cache instead of calling the OpenAI API a second time.

llm = dspy.OpenAI(model="gpt-3.5-turbo-0125", max_tokens=250)
dspy.settings.configure(lm=llm)

We can test that the calls to OpenAI are working:

llm("Testing testing, is anyone out there?")

llm(example.question)

At any point we can look at the last n calls to the llm:

llm.inspect_history(n=2)

Our evaluation metric will check if the llm output contains the correct multiple choice answer. To define an evaluation metric in DSPy we create a function like the example below. The first two inputs should be instances of dspy.Example. The metric function can contain any logic you need to evaluate your task. You can read more about the trace argument in the documentation. It needs to be there, even if not explicitly using it.

import re


def eval_metric(true, prediction, trace=None):
    pred = prediction.answer
    matches = re.findall(r"\([A-Z]\)", pred)
    parsed_answer = matches[-1] if matches else ""
    return parsed_answer == true.answer

We set up an evaluation pipeline:

from dspy.evaluate import Evaluate

evaluate = Evaluate(devset=valset, metric=eval_metric, num_threads=6, display_progress=True, display_table=10)

Here is a simple module in DSPy for basic question and answer.

class BasicQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.prog = dspy.Predict("question -> answer")

    def forward(self, question):
        return self.prog(question=question)


basic_qa = BasicQA()

The forward method calls __call__ similar to how things work in pytorch.

pred = basic_qa(question=example.question)
print("\nQUESTION:\n")
print(example.question)
print("\nANSWER:\n")
print(example.answer)
print("\nPREDICTION:\n")
print(pred.answer)

eval_metric(example, pred)

llm.inspect_history(n=1)

Now we can pass each example question through the LLM in the validation set and check if we get the correct answer:

evaluate(basic_qa)

from dspy.teleprompt import BootstrapFewShotWithRandomSearch

config = dict(max_bootstrapped_demos=2, max_labeled_demos=4, num_candidate_programs=2, num_threads=6)

teleprompter = BootstrapFewShotWithRandomSearch(metric=eval_metric, **config)
optimized_qa = teleprompter.compile(basic_qa, trainset=trainset, valset=valset)

There is a lot of output from the above code block which I am hiding to keep things cleaner. You can now evaluate the optimized model to see if the accuracy has improved.

evaluate(optimized_qa)

llm.inspect_history()

Now we can try a Chain of Thought [@wei2023chainofthought] prompt.

class CoT(dspy.Module):
    def __init__(self):
        super().__init__()
        self.prog = dspy.ChainOfThought("question -> answer")

    def forward(self, question):
        return self.prog(question=question)


cot_qa = CoT()

evaluate(cot_qa)

llm.inspect_history()

Now we will try and optimize our chain of thought program. I am also hiding the output from this cell to keep things cleaner.

tqdm._instances.clear()
config = dict(max_bootstrapped_demos=1, max_labeled_demos=4, num_candidate_programs=4, num_threads=6)
teleprompter = BootstrapFewShotWithRandomSearch(metric=eval_metric, **config)
optimized_cot_qa = teleprompter.compile(cot_qa, trainset=trainset, valset=valset)

evaluate(optimized_cot_qa)

llm.inspect_history(n=1)