import requestsfrom bs4 import BeautifulSoupimport dspy
res = requests.get("https://grugbrain.dev/")soup = BeautifulSoup(res.text, 'html.parser')raw_text = [p.text for p in soup.find_all('p') if p.text]
raw_text[:10]

from openai import OpenAIclient = OpenAI()openai_model_name= "gpt-3.5-turbo"
class BuildMessages:def __init__(self, system_prompt, user_prompt):self.system_prompt = system_promptself.user_prompt = user_promptdef render(self, **kwargs):sys = self.system_prompt.format(**kwargs)user = self.user_prompt.format(**kwargs)return [{"role":"system", "content":sys},{"role":"user", "content":user},]from functools import cache@cachedef translate_grug(grug_text):prompt = BuildMessages("You are an expert in deciphering strange text. The user will provide text written by someone named Grug and you will provide the translation.","""Translate the following text into plain english: '{text}'.
Do not respond with any other text. Only provide that text. Now take a deep breath and begin.""")result = client.chat.completions.create(messages=prompt.render(text=grug_text), model=openai_model_name)return result.choices[0].message.content

dataset = []for grug_text in raw_text[:10]:translated = translate_grug(grug_text)dataset.append({"grug_text":grug_text, "plain_english":translated})

examples = []for row in dataset:examples.append(dspy.Example(grug_text=row["grug_text"], plain_english=row["plain_english"]).with_inputs("plain_english"))

import numpy as npfrom random import shuffle
def split_for_train_test(values, test_size = 1/3.0):shuffle(values)train = int(len(values)-test_size*len(values))print(train)return values[:train], values[train:]train, test = split_for_train_test(examples)

import dspy
class GrugTranslation(dspy.Signature):"Translate plain english to Grug text."plain_english = dspy.InputField()grug_text = dspy.OutputField()

turbo = dspy.OpenAI(model='gpt-3.5-turbo', max_tokens=1000)dspy.settings.configure(lm=turbo)
from dspy.signatures.signature import signature_to_templategrug_translation_as_template = signature_to_template(GrugTranslation)
print(str(grug_translation_as_template))print(grug_translation_as_template.query(examples[0]))
GrugTranslation.signatureGrugTranslation.with_instructions

class CoT(dspy.Module):def __init__(self):super().__init__()self.prog = dspy.ChainOfThought(GrugTranslation)
def forward(self, plain_english):return self.prog(plain_english=plain_english)c = CoT()

c.forward("You should not construct complex systems.")

# https://apps.dtic.mil/sti/tr/pdf/AD0667273.pdfdef automated_readability_index(text):import recharacters = len(re.sub(r'\s+', '', text)) # Count characters (ignoring whitespace)words = len(text.split()) # Count words by splitting the text# Count sentences by finding period, exclamation, or question marksentences = len(re.findall(r'[.!?\n]', text))# small change is to add a new line character as grug doesn't seem to use punctuation.if words == 0 or sentences == 0:# Prevent division by zeroreturn 0# Calculate the Automated Readability Index (ARI)ari = (4.71 * (characters / words)) + (0.5 * (words / sentences)) - 21.43
return round(ari, 2)

for ex in examples:source_ari = automated_readability_index(ex.plain_english)grug_ari = automated_readability_index(ex.grug_text)print(f"ARI {source_ari} => {grug_ari}")

# https://dspy-docs.vercel.app/docs/building-blocks/metrics#intermediate-using-ai-feedback-for-your-metricclass AssessBasedOnQuestion(dspy.Signature):"""Given the assessed text provide a yes or no to the assessment question."""assessed_text = dspy.InputField(format=str)assessment_question = dspy.InputField(format=str)assessment_answer = dspy.OutputField(desc="Yes or No")
example_question_assessment = dspy.Example(assessed_text="This is a test.", assessment_question="Is this a test?", assessment_answer="Yes").with_inputs("assessed_text", "assessment_question")
print(signature_to_template(AssessBasedOnQuestion).query(example_question_assessment))

gpt4T = dspy.OpenAI(model='gpt-4-turbo', max_tokens=500)
def similarity_metric(truth, pred, trace=None):truth_grug_text = truth.grug_textproposed_grug_text = pred.grug_textsimilarity_question = f"""Does the assessed text have the same meaning as the gold_standard text provided?Gold Standard: "{truth_grug_text}"Provide only a yes or no answer."""with dspy.context(lm=gpt4T):assessor = dspy.Predict(AssessBasedOnQuestion)raw_similarity_result = assessor(assessed_text=proposed_grug_text, assessment_question=similarity_question)print(raw_similarity_result) # for debuggingraw_similarity = raw_similarity_result.assessment_answer.lower().strip()same_meaning = raw_similarity == 'yes'return same_meaning

def ari_metric(truth, pred, trace=None):truth_grug_text = truth.grug_textproposed_grug_text = pred.grug_text
gold_ari = automated_readability_index(truth_grug_text)pred_ari = automated_readability_index(proposed_grug_text)print(f"ARI {gold_ari} => {pred_ari}")ari_result = pred_ari <= 7.01return ari_result

def overall_metric(provided_example, predicted, trace=None):similarity = similarity_metric(provided_example, predicted, trace)ari = ari_metric(provided_example, predicted, trace)if similarity and ari:return Truereturn False

from dspy.teleprompt import BootstrapFewShot
config = dict(max_bootstrapped_demos=4, max_labeled_demos=4)optimizer = BootstrapFewShot(metric=overall_metric, **config)optimizer.max_errors = 1 # helpful to debug errors fasteroptimized_cot = optimizer.compile(CoT(), trainset=train, valset=test)

from dspy.evaluate import Evaluate
individual_metrics = [similarity_metric, ari_metric]
for metric in individual_metrics:evaluate = Evaluate(metric=metric, devset=train, num_threads=1, display_progress=True, display_table=5)evaluate(optimized_cot)

optimized_cot.forward("You should not construct complex systems.")

optimized_cot.save(path="/tmp/model.json")