get_ipython().system('git clone https://huggingface.co/arnavs11/DSPy_TweetGen_Cache')
get_ipython().run_line_magic('cd', 'DSPy_TweetGen_Cache/')
get_ipython().system('git checkout master')
get_ipython().run_line_magic('cd', '..')
import os
repo_clone_path = '/content/DSPy_TweetGen_Cache'

if not os.access('/content', os.W_OK):
    repo_clone_path = os.path.join(os.getcwd(), 'DSPy_TweetGen_Cache')

os.environ["DSP_NOTEBOOK_CACHEDIR"] = repo_clone_path


get_ipython().run_line_magic('load_ext', 'autoreload')
get_ipython().run_line_magic('autoreload', '2')

import sys
import os
import regex as re
import json

try: # When on google Colab, let's clone the notebook so we download the cache.
    import google.colab
    repo_path = 'dspy'
    
    get_ipython().system('git -C $repo_path pull origin || git clone https://github.com/stanfordnlp/dspy $repo_path')
except:
    repo_path = '.'

if repo_path not in sys.path:
    sys.path.append(repo_path)


import pkg_resources # Install the package if it's not installed
if not "dspy-ai" in {pkg.key for pkg in pkg_resources.working_set}:
    get_ipython().system('pip install -U pip')
    get_ipython().system('pip install dspy-ai')
    get_ipython().system('pip install openai~=0.28.1')
    get_ipython().system('pip install -e $repo_path')

import dspy
from dspy.predict import Retry
from dspy.datasets import HotPotQA
from dspy.teleprompt import BootstrapFewShotWithRandomSearch
from dsp.utils import deduplicate
from dspy.evaluate.evaluate import Evaluate
from dspy.primitives.assertions import assert_transform_module, backtrack_handler


colbertv2_wiki17_abstracts = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
dspy.settings.configure(rm=colbertv2_wiki17_abstracts)
turbo = dspy.OpenAI(model='gpt-3.5-turbo', max_tokens=500)
dspy.settings.configure(lm=turbo, trace=[], temperature=0.7)


dataset = HotPotQA(train_seed=1, train_size=300, eval_seed=2023, dev_size=300, test_size=0, keep_details=True)
trainset = [x.with_inputs('question', 'answer') for x in dataset.train]
devset = [x.with_inputs('question', 'answer') for x in dataset.dev]




class GenerateSearchQuery(dspy.Signature):
    """Write a simple search query that will help answer a complex question."""
    context = dspy.InputField(desc="may contain relevant facts")
    question = dspy.InputField()
    query = dspy.OutputField()

class GenerateTweet(dspy.Signature):
    """Generate an engaging tweet that effectively answers a question staying faithful to the context, is less than 280 characters, and has no hashtags."""
    question = dspy.InputField()
    context = dspy.InputField(desc="may contain relevant facts")
    tweet = dspy.OutputField()

class Tweeter(dspy.Module):
    def __init__(self):
        super().__init__()
        self.generate_tweet = dspy.ChainOfThought(GenerateTweet)

    def forward(self, question, answer):
        context = []
        max_hops=2
        passages_per_hop=3
        generate_query = [dspy.ChainOfThought(GenerateSearchQuery) for _ in range(max_hops)]
        retrieve = dspy.Retrieve(k=passages_per_hop)
        for hop in range(max_hops):
            query = generate_query[hop](context=context, question=question).query
            passages = retrieve(query).passages
            context = deduplicate(context + passages)
        generated_tweet = self.generate_tweet(question=question, context=context).tweet
        return dspy.Prediction(generated_tweet=generated_tweet, context=context)
    
tweeter = Tweeter()



def has_no_hashtags(text):
    return len(re.findall(r"#\w+", text)) == 0

def is_within_length_limit(text, length_limit=280):
    return len(text) <= length_limit

def is_assessment_yes(assessment_answer):
    """Check if the first word of the assessment answer is 'yes'."""
    return assessment_answer.split()[0].lower() == 'yes'

def has_correct_answer(text, answer):
    return answer in text


class AssessTweet(dspy.Signature):
    """Assess the quality of a tweet along the specified dimension."""

    context = dspy.InputField(desc='ignore if N/A')
    assessed_text = dspy.InputField()
    assessment_question = dspy.InputField()
    assessment_answer = dspy.OutputField(desc="Yes or No")

def no_hashtags_metric(gold, pred, trace=None):
    tweet = pred.generated_tweet
    no_hashtags = has_no_hashtags(tweet)
    score = no_hashtags
    return score

def is_correct_metric(gold, pred, trace=None):
    answer, tweet = gold.answer, pred.generated_tweet
    correct = has_correct_answer(tweet, answer)
    score = correct
    return score

def within_length_metric(gold, pred, trace=None):
    tweet = pred.generated_tweet
    within_length_limit = is_within_length_limit(tweet, 280)
    score = within_length_limit
    return score

def engaging_metric(gold, pred, trace=None):
    tweet = pred.generated_tweet
    engaging = "Does the assessed text make for a self-contained, engaging tweet? Say no if it is not engaging."
    engaging = dspy.Predict(AssessTweet)(context='N/A', assessed_text=tweet, assessment_question=engaging)
    engaging = engaging.assessment_answer.split()[0].lower() == 'yes'
    score = engaging
    return score

def faithful_metric(gold, pred, trace=None):
    context, tweet = pred.context, pred.generated_tweet
    faithful = "Is the assessed text grounded in the context? Say no if it includes significant facts not in the context."   
    faithful = dspy.Predict(AssessTweet)(context=context, assessed_text=tweet, assessment_question=faithful)
    faithful = faithful.assessment_answer.split()[0].lower() == 'yes'
    score = faithful
    return score



def overall_metric(gold, pred, trace=None):
    answer, context, tweet = gold.answer, pred.context, pred.generated_tweet
    no_hashtags = has_no_hashtags(tweet)
    within_length_limit = is_within_length_limit(tweet, 280)
    correct = has_correct_answer(tweet, answer)
    engaging = "Does the assessed text make for a self-contained, engaging tweet? Say no if it is not engaging."
    faithful = "Is the assessed text grounded in the context? Say no if it includes significant facts not in the context."   
    faithful = dspy.Predict(AssessTweet)(context=context, assessed_text=tweet, assessment_question=faithful)
    engaging = 
dspy.Predict(AssessTweet)