xbe.at

LLMs and generative AI / RAG and retrieval

Transforming Industries with 12 Types of RAG

14 min read, 13 sections

Contents
  1. Basic Naive RAG Implementation
  2. Advanced RAG with Re-ranking
  3. Modular RAG Architecture
  4. Query-Based RAG Implementation
  5. Latent RAG with Multi-Modal Support
  6. Logit-Based RAG with Attention Fusion
  7. Speculative RAG with Performance Optimization
  8. Self-Reflective RAG with Meta-Learning
  9. Branched RAG with Multi-Path Processing
  10. Agentic RAG with Autonomous Decision Making
  11. Adaptive RAG with Dynamic Knowledge Integration
  12. Corrective RAG (CRAG) Implementation
  13. Additional Resources

Basic Naive RAG Implementation

A fundamental implementation of Retrieval-Augmented Generation using vector embeddings and cosine similarity for document retrieval. This approach demonstrates the core concepts of document chunking, embedding generation, and similarity-based retrieval in Python.

import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

class NaiveRAG:
    def __init__(self, chunk_size=100):
        self.chunk_size = chunk_size
        self.vectorizer = TfidfVectorizer()
        self.document_chunks = []
        self.embeddings = None
    
    def chunk_document(self, text):
        words = text.split()
        return [' '.join(words[i:i+self.chunk_size]) 
                for i in range(0, len(words), self.chunk_size)]
    
    def index_documents(self, documents):
        for doc in documents:
            chunks = self.chunk_document(doc)
            self.document_chunks.extend(chunks)
        self.embeddings = self.vectorizer.fit_transform(self.document_chunks)
    
    def retrieve(self, query, k=3):
        query_embedding = self.vectorizer.transform([query])
        similarities = cosine_similarity(query_embedding, self.embeddings)[0]
        top_k_indices = np.argsort(similarities)[-k:][::-1]
        return [self.document_chunks[i] for i in top_k_indices]

# Example usage
documents = [
    "RAG systems combine retrieval with generation...",
    "Vector embeddings represent text as numbers..."
]
rag = NaiveRAG()
rag.index_documents(documents)
results = rag.retrieve("What is RAG?")

Advanced RAG with Re-ranking

This implementation enhances the basic RAG by adding a re-ranking mechanism using cross-attention scores. The system employs a two-stage retrieval process where initial candidates are refined using more sophisticated similarity metrics.

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

class AdvancedRAG:
    def __init__(self):
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.reranker_tokenizer = AutoTokenizer.from_pretrained('cross-encoder/ms-marco-MiniLM-L6')
        self.reranker_model = AutoModel.from_pretrained('cross-encoder/ms-marco-MiniLM-L6')
        
    def encode_documents(self, documents):
        self.documents = documents
        self.embeddings = self.encoder.encode(documents)
        
    def initial_retrieval(self, query, k=10):
        query_embedding = self.encoder.encode(query)
        scores = np.dot(self.embeddings, query_embedding)
        top_k_idx = np.argpartition(scores, -k)[-k:]
        return [(i, self.documents[i]) for i in top_k_idx]
    
    def rerank(self, query, candidates, top_k=3):
        pairs = [(query, doc) for _, doc in candidates]
        inputs = self.reranker_tokenizer(pairs, padding=True, truncation=True,
                                       return_tensors='pt')
        
        with torch.no_grad():
            outputs = self.reranker_model(**inputs)
            scores = outputs.last_hidden_state[:, 0, :].mean(dim=1)
            
        reranked_idx = torch.argsort(scores, descending=True)[:top_k]
        return [candidates[i][1] for i in reranked_idx]

# Example usage
rag = AdvancedRAG()
docs = ["Document about RAG systems...", "Information about retrieval..."]
rag.encode_documents(docs)
initial_results = rag.initial_retrieval("RAG systems")
final_results = rag.rerank("RAG systems", initial_results)

Modular RAG Architecture

A flexible implementation of RAG that separates concerns into independent modules for retrieval, ranking, and generation. This architecture allows for easy swapping of components and customization based on specific use cases.

from abc import ABC, abstractmethod
from typing import List, Tuple

class Retriever(ABC):
    @abstractmethod
    def retrieve(self, query: str, k: int) -> List[str]:
        pass

class Ranker(ABC):
    @abstractmethod
    def rank(self, query: str, documents: List[str]) -> List[Tuple[str, float]]:
        pass

class Generator(ABC):
    @abstractmethod
    def generate(self, query: str, context: List[str]) -> str:
        pass

class TFIDFRetriever(Retriever):
    def __init__(self):
        self.vectorizer = TfidfVectorizer()
        self.documents = []
        
    def index(self, documents):
        self.documents = documents
        self.embeddings = self.vectorizer.fit_transform(documents)
    
    def retrieve(self, query: str, k: int = 3) -> List[str]:
        query_vector = self.vectorizer.transform([query])
        scores = cosine_similarity(query_vector, self.embeddings)[0]
        top_k = np.argsort(scores)[-k:][::-1]
        return [self.documents[i] for i in top_k]

class ModularRAG:
    def __init__(self, retriever: Retriever, ranker: Ranker, 
                 generator: Generator):
        self.retriever = retriever
        self.ranker = ranker
        self.generator = generator
    
    def process_query(self, query: str) -> str:
        retrieved_docs = self.retriever.retrieve(query, k=5)
        ranked_docs = self.ranker.rank(query, retrieved_docs)
        response = self.generator.generate(query, [doc for doc, _ in ranked_docs])
        return response

Query-Based RAG Implementation

A specialized implementation focusing on query transformation and optimization. This system preprocesses queries using templating and semantic matching to improve retrieval accuracy while maintaining computational efficiency.

import re
from typing import List, Dict
from dataclasses import dataclass
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

@dataclass
class QueryTemplate:
    template: str
    weight: float

class QueryBasedRAG:
    def __init__(self, templates: List[QueryTemplate]):
        self.templates = templates
        self.document_store = {}
        self.embeddings = {}
        
    def expand_query(self, query: str) -> List[str]:
        expanded_queries = []
        for template in self.templates:
            expanded = template.template.format(query=query)
            expanded_queries.append((expanded, template.weight))
        return expanded_queries
    
    def precompute_embeddings(self, model, documents: Dict[str, str]):
        self.document_store = documents
        for doc_id, content in documents.items():
            self.embeddings[doc_id] = model.encode(content)
            
    def retrieve(self, query: str, model, top_k: int = 3) -> List[str]:
        expanded_queries = self.expand_query(query)
        scores = {}
        
        for expanded_query, weight in expanded_queries:
            query_embedding = model.encode(expanded_query)
            for doc_id, doc_embedding in self.embeddings.items():
                similarity = cosine_similarity(
                    [query_embedding], 
                    [doc_embedding]
                )[0][0]
                scores[doc_id] = scores.get(doc_id, 0) + similarity * weight
                
        top_docs = sorted(scores.items(), 
                         key=lambda x: x[1], 
                         reverse=True)[:top_k]
        return [self.document_store[doc_id] for doc_id, _ in top_docs]

# Example usage
templates = [
    QueryTemplate("what is {query}", 1.0),
    QueryTemplate("define {query}", 0.8),
    QueryTemplate("{query} explanation", 0.7)
]

qb_rag = QueryBasedRAG(templates)
documents = {
    "1": "RAG systems combine retrieval with generation...",
    "2": "Query optimization improves search accuracy..."
}

Latent RAG with Multi-Modal Support

This implementation extends RAG to handle multiple modalities by projecting different types of data into a shared latent space. The system can process text, images, and structured data simultaneously.

import torch
import torch.nn as nn
from PIL import Image
from torchvision import transforms
from transformers import CLIPModel, CLIPProcessor

class MultiModalEncoder(nn.Module):
    def __init__(self, latent_dim=512):
        super().__init__()
        self.clip = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
        self.processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
        self.latent_projection = nn.Linear(768, latent_dim)
        
    def encode_text(self, text: str) -> torch.Tensor:
        inputs = self.processor(text=text, return_tensors="pt", padding=True)
        text_features = self.clip.get_text_features(**inputs)
        return self.latent_projection(text_features)
    
    def encode_image(self, image: Image) -> torch.Tensor:
        inputs = self.processor(images=image, return_tensors="pt")
        image_features = self.clip.get_image_features(**inputs)
        return self.latent_projection(image_features)

class LatentRAG:
    def __init__(self, encoder: MultiModalEncoder):
        self.encoder = encoder
        self.document_embeddings = []
        self.documents = []
        
    def add_document(self, content, modality='text'):
        if modality == 'text':
            embedding = self.encoder.encode_text(content)
        elif modality == 'image':
            embedding = self.encoder.encode_image(content)
        
        self.document_embeddings.append(embedding)
        self.documents.append(content)
        
    def retrieve(self, query, modality='text', k=3):
        if modality == 'text':
            query_embedding = self.encoder.encode_text(query)
        elif modality == 'image':
            query_embedding = self.encoder.encode_image(query)
            
        similarities = torch.cosine_similarity(
            query_embedding.unsqueeze(0),
            torch.stack(self.document_embeddings),
            dim=1
        )
        
        top_k_indices = torch.topk(similarities, k).indices
        return [self.documents[i] for i in top_k_indices]

# Example usage
encoder = MultiModalEncoder()
latent_rag = LatentRAG(encoder)

# Add text document
latent_rag.add_document("RAG system explanation...", modality='text')

# Add image document
image = Image.open("example.jpg")
latent_rag.add_document(image, modality='image')

Logit-Based RAG with Attention Fusion

An advanced implementation that combines retrieved context information at the logit level during generation. This approach enables fine-grained control over how retrieved information influences the output tokens.

import torch
import torch.nn as nn
from transformers import AutoTokenizer, AutoModelForCausalLM

class LogitRAG(nn.Module):
    def __init__(self, model_name='gpt2'):
        super().__init__()
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(model_name)
        self.context_projection = nn.Linear(768, self.model.config.vocab_size)
        
    def forward(self, input_ids, context_embeddings, attention_mask=None):
        # Get base model logits
        outputs = self.model(input_ids, attention_mask=attention_mask)
        base_logits = outputs.logits
        
        # Project context to vocabulary space
        context_logits = self.context_projection(context_embeddings)
        
        # Combine logits using attention-based fusion
        attention_weights = torch.softmax(
            torch.matmul(base_logits, context_logits.transpose(-1, -2)) 
            / torch.sqrt(torch.tensor(768.0)), 
            dim=-1
        )
        
        fused_logits = base_logits + torch.matmul(attention_weights, context_logits)
        return fused_logits
    
    def generate_with_context(self, prompt, context_embeddings, max_length=50):
        input_ids = self.tokenizer.encode(prompt, return_tensors='pt')
        
        for _ in range(max_length):
            outputs = self.forward(input_ids, context_embeddings)
            next_token_logits = outputs[:, -1, :]
            next_token = torch.argmax(next_token_logits, dim=-1)
            
            if next_token.item() == self.tokenizer.eos_token_id:
                break
                
            input_ids = torch.cat([input_ids, next_token.unsqueeze(0)], dim=-1)
            
        return self.tokenizer.decode(input_ids[0])

# Example usage
logit_rag = LogitRAG()
context = torch.randn(1, 5, 768)  # Example context embeddings
generated_text = logit_rag.generate_with_context(
    "The key features of RAG include", 
    context
)

Speculative RAG with Performance Optimization

Implementation of a speculative RAG system that uses a smaller model to generate initial drafts and a larger model for verification and refinement, significantly reducing latency while maintaining quality.

import time
from typing import Tuple, List
from transformers import AutoModelForCausalLM, AutoTokenizer

class SpeculativeRAG:
    def __init__(self, draft_model_name='gpt2', 
                 verifier_model_name='gpt2-large'):
        self.draft_tokenizer = AutoTokenizer.from_pretrained(draft_model_name)
        self.draft_model = AutoModelForCausalLM.from_pretrained(draft_model_name)
        
        self.verifier_tokenizer = AutoTokenizer.from_pretrained(
            verifier_model_name
        )
        self.verifier_model = AutoModelForCausalLM.from_pretrained(
            verifier_model_name
        )
        
    def generate_draft(self, prompt: str, max_length: int = 50) -> Tuple[str, float]:
        start_time = time.time()
        input_ids = self.draft_tokenizer.encode(prompt, return_tensors='pt')
        
        draft_outputs = self.draft_model.generate(
            input_ids,
            max_length=max_length,
            num_return_sequences=1,
            do_sample=True,
            temperature=0.7
        )
        
        draft_text = self.draft_tokenizer.decode(
            draft_outputs[0], 
            skip_special_tokens=True
        )
        generation_time = time.time() - start_time
        return draft_text, generation_time
    
    def verify_and_refine(self, draft_text: str, 
                         retrieved_context: List[str]) -> Tuple[str, float]:
        start_time = time.time()
        
        # Combine draft with context
        context_string = " ".join(retrieved_context)
        input_text = f"Context: {context_string}\nDraft: {draft_text}\nRefined:"
        
        input_ids = self.verifier_tokenizer.encode(input_text, 
                                                  return_tensors='pt')
        
        refined_outputs = self.verifier_model.generate(
            input_ids,
            max_length=len(input_ids[0]) + 100,
            num_return_sequences=1,
            do_sample=False
        )
        
        refined_text = self.verifier_tokenizer.decode(
            refined_outputs[0], 
            skip_special_tokens=True
        )
        refinement_time = time.time() - start_time
        
        return refined_text.split("Refined:")[-1].strip(), refinement_time

# Example usage
spec_rag = SpeculativeRAG()
context = ["RAG combines retrieval and generation...", 
           "Speculative decoding reduces latency..."]

draft, draft_time = spec_rag.generate_draft(
    "Explain the benefits of speculative RAG"
)
final_text, refine_time = spec_rag.verify_and_refine(draft, context)

print(f"Total latency: {draft_time + refine_time:.2f}s")

Self-Reflective RAG with Meta-Learning

A sophisticated implementation incorporating self-reflection mechanisms that allow the system to evaluate and adjust its retrieval and generation strategies dynamically based on performance metrics and confidence scores.

import torch
import torch.nn as nn
from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class ReflectionMetrics:
    retrieval_confidence: float
    generation_coherence: float
    context_relevance: float

class SelfReflectiveRAG:
    def __init__(self, base_model: nn.Module, learning_rate: float = 1e-4):
        self.base_model = base_model
        self.reflection_network = nn.Sequential(
            nn.Linear(768, 384),
            nn.ReLU(),
            nn.Linear(384, 3)  # 3 metrics output
        )
        self.optimizer = torch.optim.Adam(
            self.reflection_network.parameters(), 
            lr=learning_rate
        )
        self.history: List[Dict] = []
        
    def evaluate_generation(self, 
                          query: str, 
                          retrieved_docs: List[str], 
                          generated_text: str) -> ReflectionMetrics:
        # Encode inputs for reflection
        encoded_query = self.base_model.encode(query)
        encoded_docs = torch.stack([
            self.base_model.encode(doc) for doc in retrieved_docs
        ])
        encoded_generation = self.base_model.encode(generated_text)
        
        # Combine features
        combined_features = torch.cat([
            encoded_query.mean(0),
            encoded_docs.mean(0),
            encoded_generation.mean(0)
        ])
        
        # Get reflection metrics
        metrics = self.reflection_network(combined_features)
        return ReflectionMetrics(
            retrieval_confidence=metrics[0].item(),
            generation_coherence=metrics[1].item(),
            context_relevance=metrics[2].item()
        )
    
    def generate_with_reflection(self, 
                               query: str, 
                               max_attempts: int = 3) -> str:
        best_response = None
        best_metrics = None
        
        for attempt in range(max_attempts):
            # Retrieve documents
            retrieved_docs = self.retrieve_documents(query)
            
            # Generate response
            generated_text = self.generate_response(query, retrieved_docs)
            
            # Evaluate generation
            metrics = self.evaluate_generation(
                query, 
                retrieved_docs, 
                generated_text
            )
            
            # Update best response if needed
            if (best_metrics is None or 
                metrics.generation_coherence > best_metrics.generation_coherence):
                best_response = generated_text
                best_metrics = metrics
            
            # Log attempt for learning
            self.history.append({
                'query': query,
                'retrieved_docs': retrieved_docs,
                'generated_text': generated_text,
                'metrics': metrics
            })
            
            # Break if quality is sufficient
            if metrics.generation_coherence > 0.8:
                break
                
        return best_response
    
    def update_reflection_model(self):
        if len(self.history) < 10:
            return
            
        # Train reflection network using historical data
        for entry in self.history[-10:]:
            self.optimizer.zero_grad()
            
            # Compute loss based on human feedback or other metrics
            loss = self.compute_reflection_loss(entry)
            loss.backward()
            self.optimizer.step()

# Example usage
class DummyBaseModel(nn.Module):
    def encode(self, text: str) -> torch.Tensor:
        return torch.randn(1, 768)
        
    def decode(self, embedding: torch.Tensor) -> str:
        return "Sample generated text"

base_model = DummyBaseModel()
reflective_rag = SelfReflectiveRAG(base_model)
response = reflective_rag.generate_with_reflection(
    "Explain self-reflective RAG systems"
)

Branched RAG with Multi-Path Processing

An implementation that explores multiple retrieval and generation paths simultaneously, then selects the most promising route based on various quality metrics and coherence scores.

import torch
from typing import List, Dict, Tuple
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor

@dataclass
class BranchMetrics:
    relevance_score: float
    coherence_score: float
    novelty_score: float
    
    def total_score(self) -> float:
        return (self.relevance_score * 0.4 + 
                self.coherence_score * 0.4 + 
                self.novelty_score * 0.2)

class BranchedRAG:
    def __init__(self, num_branches: int = 3):
        self.num_branches = num_branches
        self.retriever_variants = self._init_retrievers()
        self.generator_variants = self._init_generators()
        
    def _init_retrievers(self) -> List[callable]:
        return [
            self._semantic_retriever,
            self._keyword_retriever,
            self._hybrid_retriever
        ]
    
    def _init_generators(self) -> List[callable]:
        return [
            self._standard_generator,
            self._diverse_generator,
            self._focused_generator
        ]
        
    def process_query(self, 
                     query: str, 
                     num_workers: int = 3) -> Tuple[str, BranchMetrics]:
        branches: List[Dict] = []
        
        with ThreadPoolExecutor(max_workers=num_workers) as executor:
            # Execute branches in parallel
            future_branches = [
                executor.submit(self._process_branch, query, i)
                for i in range(self.num_branches)
            ]
            
            for future in future_branches:
                branches.append(future.result())
                
        # Select best branch
        best_branch = max(branches, 
                         key=lambda x: x['metrics'].total_score())
        return best_branch['response'], best_branch['metrics']
    
    def _process_branch(self, query: str, branch_id: int) -> Dict:
        # Select retriever and generator for this branch
        retriever = self.retriever_variants[branch_id % len(self.retriever_variants)]
        generator = self.generator_variants[branch_id % len(self.generator_variants)]
        
        # Retrieve documents
        retrieved_docs = retriever(query)
        
        # Generate response
        response = generator(query, retrieved_docs)
        
        # Evaluate branch
        metrics = self._evaluate_branch(query, retrieved_docs, response)
        
        return {
            'branch_id': branch_id,
            'response': response,
            'metrics': metrics
        }
    
    def _evaluate_branch(self, 
                        query: str, 
                        docs: List[str], 
                        response: str) -> BranchMetrics:
        # Implement evaluation logic
        return BranchMetrics(
            relevance_score=self._compute_relevance(query, response),
            coherence_score=self._compute_coherence(response),
            novelty_score=self._compute_novelty(docs, response)
        )

# Example usage
branched_rag = BranchedRAG(num_branches=3)
response, metrics = branched_rag.process_query(
    "Explain the advantages of branched RAG"
)

Agentic RAG with Autonomous Decision Making

An implementation that combines RAG with autonomous agent capabilities, allowing the system to make independent decisions about information retrieval and task decomposition while maintaining coherent interaction flow.

import torch
from typing import List, Dict, Optional
from enum import Enum
from dataclasses import dataclass

class AgentAction(Enum):
    RETRIEVE = "retrieve"
    CLARIFY = "clarify"
    GENERATE = "generate"
    REFINE = "refine"

@dataclass
class AgentState:
    context: List[str]
    current_task: str
    confidence: float
    history: List[Dict]

class AgenticRAG:
    def __init__(self, confidence_threshold: float = 0.7):
        self.confidence_threshold = confidence_threshold
        self.state = AgentState([], "", 0.0, [])
        
    def process_query(self, query: str) -> str:
        self.state.current_task = query
        action_sequence = self._plan_actions(query)
        
        final_response = ""
        for action in action_sequence:
            result = self._execute_action(action)
            if result:
                final_response = result
                
        return final_response
    
    def _plan_actions(self, query: str) -> List[AgentAction]:
        # Analyze query and determine necessary actions
        actions = [AgentAction.RETRIEVE]
        
        if self._needs_clarification(query):
            actions.insert(0, AgentAction.CLARIFY)
            
        actions.append(AgentAction.GENERATE)
        
        if self.state.confidence < self.confidence_threshold:
            actions.append(AgentAction.REFINE)
            
        return actions
    
    def _execute_action(self, action: AgentAction) -> Optional[str]:
        if action == AgentAction.RETRIEVE:
            retrieved_docs = self._retrieve_relevant_docs()
            self.state.context.extend(retrieved_docs)
            return None
            
        elif action == AgentAction.CLARIFY:
            clarification = self._generate_clarification()
            self.state.history.append({
                'action': 'clarify',
                'content': clarification
            })
            return clarification
            
        elif action == AgentAction.GENERATE:
            response = self._generate_response()
            self.state.history.append({
                'action': 'generate',
                'content': response
            })
            return response
            
        elif action == AgentAction.REFINE:
            refined = self._refine_response()
            self.state.history.append({
                'action': 'refine',
                'content': refined
            })
            return refined
    
    def _retrieve_relevant_docs(self) -> List[str]:
        # Implement retrieval logic
        return ["Relevant document 1", "Relevant document 2"]
    
    def _generate_clarification(self) -> str:
        # Implement clarification logic
        return "Could you please specify...?"
    
    def _generate_response(self) -> str:
        # Implement response generation
        return "Generated response based on context"
    
    def _refine_response(self) -> str:
        # Implement refinement logic
        return "Refined response"
    
    def _needs_clarification(self, query: str) -> bool:
        # Implement clarification detection
        return len(query.split()) < 3

# Example usage
agent_rag = AgenticRAG()
response = agent_rag.process_query("Explain agentic RAG capabilities")

Adaptive RAG with Dynamic Knowledge Integration

This implementation features dynamic adaptation to user queries and feedback, continuously adjusting its retrieval and generation strategies based on interaction patterns and performance metrics.

import torch
import numpy as np
from typing import List, Dict, Tuple
from collections import defaultdict

class AdaptiveRAG:
    def __init__(self, learning_rate: float = 0.01):
        self.learning_rate = learning_rate
        self.strategy_weights = defaultdict(lambda: 1.0)
        self.performance_history = []
        
    def adapt_retrieval_strategy(self, 
                               query: str, 
                               feedback_score: float) -> None:
        query_features = self._extract_query_features(query)
        
        # Update weights based on feedback
        for feature, value in query_features.items():
            current_weight = self.strategy_weights[feature]
            gradient = value * (feedback_score - current_weight)
            self.strategy_weights[feature] += self.learning_rate * gradient
    
    def _extract_query_features(self, query: str) -> Dict[str, float]:
        features = {
            'length': len(query.split()),
            'complexity': self._compute_complexity(query),
            'domain_specificity': self._compute_domain_specificity(query)
        }
        return features
    
    def retrieve_with_adaptation(self, 
                               query: str, 
                               top_k: int = 5) -> List[str]:
        # Get current strategy weights
        features = self._extract_query_features(query)
        weighted_score = sum(
            self.strategy_weights[f] * v for f, v in features.items()
        )
        
        # Adjust retrieval parameters based on weighted score
        retrieval_params = self._compute_retrieval_params(weighted_score)
        
        # Perform retrieval
        docs = self._retrieve_documents(query, retrieval_params)
        return docs[:top_k]
    
    def _compute_complexity(self, query: str) -> float:
        # Implement query complexity measurement
        return len(set(query.split())) / len(query.split())
    
    def _compute_domain_specificity(self, query: str) -> float:
        # Implement domain specificity measurement
        return 0.5  # Placeholder
    
    def _compute_retrieval_params(self, 
                                weighted_score: float) -> Dict[str, float]:
        return {
            'temperature': max(0.1, min(1.0, weighted_score)),
            'diversity_penalty': max(0.0, min(1.0, 1.0 - weighted_score)),
            'context_window': int(100 * weighted_score)
        }
    
    def _retrieve_documents(self, 
                          query: str, 
                          params: Dict[str, float]) -> List[str]:
        # Implement actual document retrieval
        return ["Retrieved document 1", "Retrieved document 2"]

# Example usage
adaptive_rag = AdaptiveRAG()
retrieved_docs = adaptive_rag.retrieve_with_adaptation(
    "Explain adaptive RAG systems"
)
adaptive_rag.adapt_retrieval_strategy("Explain adaptive RAG systems", 0.8)

Corrective RAG (CRAG) Implementation

A specialized implementation that focuses on error detection and correction during both retrieval and generation phases, using multiple validation layers to ensure accuracy and consistency of the output.

import torch
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass

@dataclass
class CorrectionMetrics:
    factual_errors: int
    consistency_score: float
    semantic_drift: float

class CorrectiveRAG:
    def __init__(self, error_threshold: float = 0.3):
        self.error_threshold = error_threshold
        self.fact_checker = self._init_fact_checker()
        self.consistency_validator = self._init_consistency_validator()
        
    def process_with_correction(self, query: str) -> Tuple[str, CorrectionMetrics]:
        # Initial retrieval and generation
        retrieved_docs = self._retrieve_documents(query)
        initial_response = self._generate_response(query, retrieved_docs)
        
        # Correction phase
        corrected_response, metrics = self._apply_corrections(
            query,
            initial_response,
            retrieved_docs
        )
        
        return corrected_response, metrics
    
    def _apply_corrections(self, 
                          query: str,
                          response: str,
                          context: List[str]) -> Tuple[str, CorrectionMetrics]:
        # Check for factual errors
        factual_errors = self._check_facts(response, context)
        
        # Validate consistency
        consistency_score = self._check_consistency(response, context)
        
        # Measure semantic drift
        semantic_drift = self._measure_drift(query, response)
        
        metrics = CorrectionMetrics(
            factual_errors=len(factual_errors),
            consistency_score=consistency_score,
            semantic_drift=semantic_drift
        )
        
        if metrics.factual_errors > 0 or metrics.consistency_score < self.error_threshold:
            response = self._generate_correction(response, factual_errors)
            
        return response, metrics
    
    def _check_facts(self, response: str, context: List[str]) -> List[Dict]:
        errors = []
        statements = self._extract_statements(response)
        
        for statement in statements:
            if not self._verify_statement(statement, context):
                errors.append({
                    'statement': statement,
                    'type': 'factual',
                    'evidence': self._find_contradicting_evidence(statement, context)
                })
                
        return errors
    
    def _check_consistency(self, response: str, context: List[str]) -> float:
        # Implement consistency checking
        embeddings = self._get_embeddings(response)
        context_embeddings = self._get_embeddings(" ".join(context))
        
        return float(torch.cosine_similarity(
            embeddings.mean(0),
            context_embeddings.mean(0),
            dim=0
        ))
    
    def _measure_drift(self, query: str, response: str) -> float:
        # Implement semantic drift measurement
        query_embedding = self._get_embeddings(query)
        response_embedding = self._get_embeddings(response)
        
        return float(torch.cosine_similarity(
            query_embedding.mean(0),
            response_embedding.mean(0),
            dim=0
        ))
    
    def _generate_correction(self, 
                           original: str,
                           errors: List[Dict]) -> str:
        # Implement correction generation
        corrected = original
        for error in errors:
            correction = self._get_correction_for_error(error)
            corrected = self._apply_correction(corrected, correction)
            
        return corrected

# Example usage
crag = CorrectiveRAG()
response, metrics = crag.process_with_correction(
    "Explain how CRAG systems work"
)

Additional Resources

  • Retrieval-Augmented Generation for Large Language Models: A Survey
    • arxiv.org/abs/2312.10997
  • Self-RAG: Learning to Retrieve, Generate, and Critique through Self-Reflection
    • arxiv.org/abs/2310.11511
  • Benchmarking Retrieval-Augmented Generation for Medicine
    • arxiv.org/abs/2312.00656
  • RAG vs Fine-tuning: A Case Study on Legal Question Answering
    • Search on Google Scholar: "RAG legal question answering comparison"
  • Best Practices for RAG Applications: A Comprehensive Guide
    • Visit docs.anthropic.com for detailed implementation guides

More in RAG and retrieval

  1. Understanding RAG vs. Agentic RAG with Python11 min

    Retrieval-Augmented Generation (RAG) is a powerful technique that combines the strengths of large language models with external knowledge retrieval. It...

  2. Retrieval-Augmented Generation Python Practical Examples10 min

    Retrieval-Augmented Generation (RAG) combines traditional language models with an external knowledge base to generate more accurate and factual responses....

  3. Unveiling Inconsistencies in Retrieval-Augmented Language Models with EoR Using Python9 min

    Retrieval-Augmented Language Models (RALMs) combine the power of large language models with external knowledge retrieval. This approach allows models to...

  4. Retrieval-Augmented Generation (RAG) vs. HyDE Revolutionizing Text Generation19 min

    A foundational implementation of Retrieval-Augmented Generation using vector embeddings and cosine similarity for document retrieval. This base...

  5. Visualizing High-Dimensional Data with UMAP in Python9 min

    Large Language Models are AI systems trained on vast amounts of text data to understand and generate human-like text. They form the foundation for many...

  6. Retrieval Techniques in RAG Sparse and Dense Vector Tools9 min

    Retrieval-Augmented Generation (RAG) combines information retrieval with text generation. This presentation focuses on two key retrieval techniques: sparse...