A Coding Info To Assemble Modular And Self-Correcting QA Strategies With DSPy

!pip arrange dspy-ai google-generativeai


import dspy
import google.generativeai as genai
import random
from typing import Document, Non-compulsory


GOOGLE_API_KEY = "Use Your Private API Key"  
genai.configure(api_key=GOOGLE_API_KEY)


dspy.configure(lm=dspy.LM(model="gemini/gemini-1.5-flash", api_key=GOOGLE_API_KEY))

class QuestionAnswering(dspy.Signature):
    """Reply questions based mostly totally on given context with reasoning."""
    context: str = dspy.InputField(desc="Associated context data")
    question: str = dspy.InputField(desc="Question to answer")
    reasoning: str = dspy.OutputField(desc="Step-by-step reasoning")
    reply: str = dspy.OutputField(desc="Closing reply")


class FactualityCheck(dspy.Signature):
    """Affirm if an answer is factually proper given context."""
    context: str = dspy.InputField()
    question: str = dspy.InputField()
    reply: str = dspy.InputField()
    is_correct: bool = dspy.OutputField(desc="True if reply is factually proper")

class AdvancedQA(dspy.Module):
    def __init__(self, max_retries: int = 2):
        great().__init__()
        self.max_retries = max_retries
        self.qa_predictor = dspy.ChainOfThought(QuestionAnswering)
        self.fact_checker = dspy.Predict(FactualityCheck)
       
    def forward(self, context: str, question: str) -> dspy.Prediction:
        prediction = self.qa_predictor(context=context, question=question)
       
        for strive in differ(self.max_retries):
            fact_check = self.fact_checker(
                context=context,
                question=question,
                reply=prediction.reply
            )
           
            if fact_check.is_correct:
                break
               
            refined_context = f"{context}nnPrevious incorrect reply: {prediction.reply}nPlease current a additional appropriate reply."
            prediction = self.qa_predictor(context=refined_context, question=question)
       
        return prediction

class SimpleRAG(dspy.Module):
    def __init__(self, knowledge_base: Document[str]):
        great().__init__()
        self.knowledge_base = knowledge_base
        self.qa_system = AdvancedQA()
       
    def retrieve(self, question: str, top_k: int = 2) -> str:
        # Straightforward keyword-based retrieval (in apply, use vector embeddings)
        scored_docs = []
        question_words = set(question.lower().break up())
       
        for doc in self.knowledge_base:
            doc_words = set(doc.lower().break up())
            ranking = len(question_words.intersection(doc_words))
            scored_docs.append((ranking, doc))
       
        # Return top-k most associated paperwork
        scored_docs.form(reverse=True)
        return "nn".be a part of([doc for _, doc in scored_docs[:top_k]])
   
    def forward(self, question: str) -> dspy.Prediction:
        context = self.retrieve(question)
        return self.qa_system(context=context, question=question)

knowledge_base = [
    “Use Your Context and Knowledge Base Here”
]


training_examples = [
    dspy.Example(
        question="What is the height of the Eiffel Tower?",
        context="The Eiffel Tower is located in Paris, France. It was constructed from 1887 to 1889 and stands 330 meters tall including antennas.",
        answer="330 meters"
    ).with_inputs("question", "context"),
   
    dspy.Example(
        question="Who created Python programming language?",
        context="Python is a high-level programming language created by Guido van Rossum. It was first released in 1991 and emphasizes code readability.",
        answer="Guido van Rossum"
    ).with_inputs("question", "context"),
   
    dspy.Example(
        question="What is machine learning?",
        context="ML focuses on algorithms that can learn from data without being explicitly programmed.",
        answer="Machine learning focuses on algorithms that learn from data without explicit programming."
    ).with_inputs("question", "context")
]

def accuracy_metric(occasion, prediction, trace=None):
    """Straightforward accuracy metric for evaluation"""
    return occasion.reply.lower() in prediction.reply.lower()


print("🚀 Initializing DSPy QA System with Gemini...")
print("📝 Observe: Using Google's Gemini 1.5 Flash (free tier)")
rag_system = SimpleRAG(knowledge_base)


basic_qa = dspy.ChainOfThought(QuestionAnswering)


print("n📊 Sooner than Optimization:")
test_question = "What is the high of the Eiffel Tower?"
test_context = knowledge_base[0]
initial_prediction = basic_qa(context=test_context, question=test_question)
print(f"Q: {test_question}")
print(f"A: {initial_prediction.reply}")
print(f"Reasoning: {initial_prediction.reasoning}")


print("n🔧 Optimizing with BootstrapFewShot...")
optimizer = dspy.BootstrapFewShot(metric=accuracy_metric, max_bootstrapped_demos=2)
optimized_qa = optimizer.compile(basic_qa, trainset=training_examples)


print("n📈 After Optimization:")
optimized_prediction = optimized_qa(context=test_context, question=test_question)
print(f"Q: {test_question}")
print(f"A: {optimized_prediction.reply}")
print(f"Reasoning: {optimized_prediction.reasoning}")

def evaluate_system(qa_module, test_cases):
    """Think about QA system effectivity"""
    proper = 0
    full = len(test_cases)
   
    as an illustration in test_cases:
        prediction = qa_module(context=occasion.context, question=occasion.question)
        if accuracy_metric(occasion, prediction):
            proper += 1
   
    return proper / full


print(f"n📊 Evaluation Outcomes:")
print(f"Main QA Accuracy: {evaluate_system(basic_qa, training_examples):.2%}")
print(f"Optimized QA Accuracy: {evaluate_system(optimized_qa, training_examples):.2%}")


print("n✅ Tutorial Full! Key DSPy Concepts Demonstrated:")
print("1. 🔤 Signatures - Outlined enter/output schemas")
print("2. 🏗️  Modules - Constructed composable QA strategies")
print("3. 🔄 Self-correction - Utilized iterative enchancment")
print("4. 🔍 RAG - Created retrieval-augmented period")
print("5. ⚡ Optimization - Used BootstrapFewShot to reinforce prompts")
print("6. 📊 Evaluation - Measured system effectivity")
print("7. 🆓 Free API - Powered by Google Gemini 1.5 Flash")