6 Types of RAG: In-Depth Analysis with Python Examples

from transformers import AutoTokenizer, AutoModel
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

# Initializing components
retriever = SentenceTransformer('distilbert-base-nli-mean-tokens')
tokenizer = AutoTokenizer.from_pretrained("gpt2")
model = AutoModel.from_pretrained("gpt2")

# Creating a FAISS index for fast search
documents = ["Document 1", "Document 2", "Document 3"]
embeddings = retriever.encode(documents)
index = faiss.IndexFlatL2(embeddings.shape[1])
index.add(embeddings)

def standard_rag(query):
    # Retrieving relevant documents
    query_vector = retriever.encode([query])
    _, I = index.search(query_vector, k=1)
    retrieved_doc = documents[I[0][0]]
    
    # Generating response
    input_ids = tokenizer.encode(query + " " + retrieved_doc, return_tensors="pt")
    output = model.generate(input_ids, max_length=100)
    response = tokenizer.decode(output[0], skip_special_tokens=True)
    
    return response

# Usage
query = "How does Standard RAG work?"
print(standard_rag(query))

import re

def corrective_rag(query):
    # Getting initial response using Standard RAG
    initial_response = standard_rag(query)
    
    # Function to check the response for errors
    def check_errors(response):
        errors = []
        if len(response.split()) < 10:
            errors.append("Response is too short")
        if not re.search(r'\b(RAG|Retrieval-Augmented Generation)\b', response):
            errors.append("Response lacks key terms")
        return errors
    
    # Function to correct the response
    def correct_response(response, errors):
        if "Response is too short" in errors:
            response += " " + standard_rag("Expand the answer on RAG")
        if "Response lacks key terms" in errors:
            response = "RAG (Retrieval-Augmented Generation) is " + response
        return response
    
    # Iterative correction process
    max_iterations = 3
    for _ in range(max_iterations):
        errors = check_errors(initial_response)
        if not errors:
            break
        initial_response = correct_response(initial_response, errors)
    
    return initial_response

# Usage
query = "Explain the concept of RAG"
print(corrective_rag(query))

import random

def speculative_rag(query):
    # Generating multiple response variants
    variants = [standard_rag(query) for _ in range(3)]
    
    # Function to evaluate variants
    def evaluate_variant(variant):
        score = 0
        if len(variant.split()) > 20:
            score += 1
        if "RAG" in variant:
            score += 2
        if "retrieval" in variant.lower() and "generation" in variant.lower():
            score += 3
        return score
    
    # Evaluating and choosing the best variant
    scores = [evaluate_variant(v) for v in variants]
    best_variant = variants[scores.index(max(scores))]
    
    return best_variant

# Usage
query = "What is Speculative RAG?"
print(speculative_rag(query))

def fusion_rag(query):
    # Simulating responses from different sources
    source1 = standard_rag(query)
    source2 = standard_rag(query + " in detail")
    source3 = standard_rag(query + " with examples")
    
    # Function to extract key phrases
    def extract_key_phrases(text):
        return text.split('.')
    
    # Extracting key phrases from all sources
    phrases = (extract_key_phrases(source1) + 
               extract_key_phrases(source2) + 
               extract_key_phrases(source3))
    
    # Removing duplicates and combining phrases
    unique_phrases = list(set(phrases))
    fused_response = ' '.join(unique_phrases)
    
    return fused_response

# Usage
query = "Explain the concept of Fusion RAG"
print(fusion_rag(query))

class Agent:
    def __init__(self, name, specialization):
        self.name = name
        self.specialization = specialization
    
    def process(self, task):
        return f"Agent {self.name} ({self.specialization}): {standard_rag(task)}"

def agentic_rag(query):
    # Creating agents
    agents = [
        Agent("Researcher", "information retrieval"),
        Agent("Analyst", "data analysis"),
        Agent("Writer", "text generation")
    ]
    
    # Breaking down the task into subtasks
    tasks = [
        "Find the key concepts of RAG",
        "Analyze the advantages of the agentic approach",
        "Write a summary on Agentic RAG"
    ]
    
    # Agents performing their tasks
    results = []
    for agent, task in zip(agents, tasks):
        results.append(agent.process(task))
    
    # Combining the results
    final_response = "\n".join(results)
    
    return final_response

# Usage
query = "Tell me about Agentic RAG"
print(agentic_rag(query))

def self_rag(query):
    # Generating initial response
    initial_response = standard_rag(query)
    
    # Function for self-evaluation
    def self_evaluate(response):
        score = 0
        if len(response.split()) > 50:
            score += 1
        if "RAG" in response and "self-correcting" in response.lower():
            score += 2
        if "example" or "for instance" in response.lower():
            score += 1
        return score
    
    # Self-correction process
    max_iterations = 3
    current_response = initial_response
    for _ in range(max_iterations):
        score = self_evaluate(current_response)
        if score >= 3:
            break
        
        # Generating improved response
        improvement_query = f"Improve this response about Self RAG: {current_response}"
        current_response = standard_rag(improvement_query)
    
    return current_response

# Usage
query = "What is Self RAG?"
print(self_rag(query))