xbe.at

LLMs and generative AI / RAG and retrieval

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

19 min read, 14 sections

Contents
  1. Traditional RAG Implementation
  2. HyDE Implementation
  3. Vector Similarity Metrics
  4. Document Chunking Strategies
  5. Embedding Cache Implementation
  6. Performance Comparison Framework
  7. Advanced Query Preprocessing
  8. Hybrid RAG-HyDE Architecture
  9. Contextual Re-ranking System
  10. Adaptive Learning Component
  11. Memory-Efficient Document Store
  12. Real-time Performance Monitoring
  13. Results Analysis Framework
  14. Additional Resources

Traditional RAG Implementation

A foundational implementation of Retrieval-Augmented Generation using vector embeddings and cosine similarity for document retrieval. This base implementation demonstrates core RAG concepts including document chunking, embedding generation, and similarity-based retrieval.

import numpy as np
from sentence_transformers import SentenceTransformer
from typing import List, Tuple

class TraditionalRAG:
    def __init__(self, model_name: str = 'all-MiniLM-L6-v2'):
        self.encoder = SentenceTransformer(model_name)
        self.document_store = []
        self.embeddings = None
    
    def add_documents(self, documents: List[str], chunk_size: int = 512):
        # Chunk documents and store
        chunks = []
        for doc in documents:
            chunks.extend([doc[i:i+chunk_size] for i in range(0, len(doc), chunk_size)])
        self.document_store = chunks
        
        # Generate embeddings
        self.embeddings = self.encoder.encode(chunks)
    
    def retrieve(self, query: str, k: int = 3) -> List[Tuple[str, float]]:
        query_embedding = self.encoder.encode(query)
        scores = np.dot(self.embeddings, query_embedding) / (
            np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(query_embedding)
        )
        top_k_idx = np.argsort(scores)[-k:][::-1]
        return [(self.document_store[i], scores[i]) for i in top_k_idx]

# Example usage
docs = [
    "RAG combines retrieval with generation for better responses.",
    "Vector databases enable efficient similarity search.",
]
rag = TraditionalRAG()
rag.add_documents(docs)
results = rag.retrieve("How does RAG work?")

HyDE Implementation

HyDE extends traditional RAG by first generating hypothetical relevant documents using an LLM, then using these for retrieval. This approach bridges the semantic gap between queries and documents through synthetic document generation.

from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

class HyDERetriever:
    def __init__(self, llm_model: str = "gpt2"):
        self.traditional_rag = TraditionalRAG()
        self.tokenizer = AutoTokenizer.from_pretrained(llm_model)
        self.model = AutoModelForCausalLM.from_pretrained(llm_model)
        
    def generate_synthetic_doc(self, query: str) -> str:
        prompt = f"Generate a detailed document that would answer this query: {query}\n\n"
        inputs = self.tokenizer(prompt, return_tensors="pt")
        
        with torch.no_grad():
            outputs = self.model.generate(
                inputs.input_ids,
                max_length=200,
                num_return_sequences=1,
                temperature=0.7
            )
        
        return self.tokenizer.decode(outputs[0], skip_special_tokens=True)
    
    def hyde_retrieve(self, query: str, k: int = 3) -> List[Tuple[str, float]]:
        synthetic_doc = self.generate_synthetic_doc(query)
        # Use synthetic document as query for retrieval
        return self.traditional_rag.retrieve(synthetic_doc, k)

# Example usage
hyde = HyDERetriever()
hyde.traditional_rag.add_documents(docs)
results = hyde.hyde_retrieve("Explain document retrieval methods")

Vector Similarity Metrics

Understanding various similarity metrics is crucial for both Traditional RAG and HyDE implementations. These metrics determine how documents are matched with queries in the embedding space.

import numpy as np
from scipy.spatial.distance import cosine, euclidean
from typing import Callable

class SimilarityMetrics:
    @staticmethod
    def cosine_similarity(v1: np.ndarray, v2: np.ndarray) -> float:
        return 1 - cosine(v1, v2)
    
    @staticmethod
    def euclidean_similarity(v1: np.ndarray, v2: np.ndarray) -> float:
        return 1 / (1 + euclidean(v1, v2))
    
    @staticmethod
    def dot_product_similarity(v1: np.ndarray, v2: np.ndarray) -> float:
        return np.dot(v1, v2)
    
    def compare_metrics(self, v1: np.ndarray, v2: np.ndarray) -> dict:
        metrics = {
            'cosine': self.cosine_similarity,
            'euclidean': self.euclidean_similarity,
            'dot_product': self.dot_product_similarity
        }
        return {name: metric(v1, v2) for name, metric in metrics.items()}

# Example usage
metrics = SimilarityMetrics()
v1 = np.array([1, 0, 1])
v2 = np.array([1, 1, 0])
comparison = metrics.compare_metrics(v1, v2)
print(f"Similarity Scores: {comparison}")

Document Chunking Strategies

Efficient document chunking is essential for both Traditional RAG and HyDE to maintain context coherence while managing token limits. This implementation demonstrates various chunking strategies including overlap and semantic boundaries.

import re
from typing import List, Optional

class DocumentChunker:
    def __init__(self, chunk_size: int = 512, overlap: int = 50):
        self.chunk_size = chunk_size
        self.overlap = overlap
    
    def chunk_by_size(self, text: str) -> List[str]:
        chunks = []
        start = 0
        while start < len(text):
            end = start + self.chunk_size
            if end < len(text):
                # Find the last period or newline before chunk_size
                last_break = max(
                    text[start:end].rfind('.'),
                    text[start:end].rfind('\n')
                )
                if last_break != -1:
                    end = start + last_break + 1
            
            chunks.append(text[start:end])
            start = end - self.overlap
        
        return chunks

    def chunk_by_sentence(self, text: str) -> List[str]:
        sentences = re.split(r'(?<=[.!?])\s+', text)
        chunks = []
        current_chunk = []
        current_length = 0
        
        for sentence in sentences:
            sentence_length = len(sentence)
            if current_length + sentence_length > self.chunk_size:
                chunks.append(' '.join(current_chunk))
                current_chunk = []
                current_length = 0
            
            current_chunk.append(sentence)
            current_length += sentence_length
        
        if current_chunk:
            chunks.append(' '.join(current_chunk))
        
        return chunks

# Example usage
chunker = DocumentChunker(chunk_size=100, overlap=20)
text = """This is a long document that needs to be chunked properly.
It contains multiple sentences and paragraphs. We need to ensure
that the chunking maintains semantic coherence while respecting
size limits. This is crucial for both RAG and HyDE implementations."""

size_chunks = chunker.chunk_by_size(text)
sentence_chunks = chunker.chunk_by_sentence(text)

Embedding Cache Implementation

An efficient caching system for document embeddings to optimize performance in both Traditional RAG and HyDE systems. This implementation reduces computational overhead by storing and reusing previously computed embeddings.

import hashlib
from typing import Dict, Any
import numpy as np
import pickle
from datetime import datetime, timedelta

class EmbeddingCache:
    def __init__(self, cache_ttl: int = 3600):
        self.cache: Dict[str, Dict[str, Any]] = {}
        self.cache_ttl = timedelta(seconds=cache_ttl)
    
    def _generate_key(self, text: str) -> str:
        return hashlib.md5(text.encode()).hexdigest()
    
    def get_embedding(self, text: str) -> np.ndarray:
        key = self._generate_key(text)
        if key in self.cache:
            cache_entry = self.cache[key]
            if datetime.now() - cache_entry['timestamp'] < self.cache_ttl:
                return cache_entry['embedding']
            else:
                del self.cache[key]
        return None
    
    def store_embedding(self, text: str, embedding: np.ndarray) -> None:
        key = self._generate_key(text)
        self.cache[key] = {
            'embedding': embedding,
            'timestamp': datetime.now()
        }
    
    def save_cache(self, filepath: str) -> None:
        with open(filepath, 'wb') as f:
            pickle.dump(self.cache, f)
    
    def load_cache(self, filepath: str) -> None:
        try:
            with open(filepath, 'rb') as f:
                self.cache = pickle.load(f)
        except FileNotFoundError:
            self.cache = {}

# Example usage
cache = EmbeddingCache(cache_ttl=3600)
test_embedding = np.random.rand(768)  # Example embedding dimension
cache.store_embedding("test document", test_embedding)
retrieved_embedding = cache.get_embedding("test document")

Performance Comparison Framework

A comprehensive framework for comparing Traditional RAG and HyDE approaches, implementing metrics like precision, recall, and latency measurements. This framework helps in quantifying the effectiveness of both approaches across different scenarios.

import time
from dataclasses import dataclass
from typing import List, Dict, Any
import numpy as np
from sklearn.metrics import precision_score, recall_score, f1_score

@dataclass
class RetrievalMetrics:
    precision: float
    recall: float
    f1: float
    latency: float
    memory_usage: int

class RAGBenchmark:
    def __init__(self, traditional_rag, hyde_retriever):
        self.traditional_rag = traditional_rag
        self.hyde_retriever = hyde_retriever
        self.metrics_history = []
    
    def measure_performance(self, query: str, relevant_docs: List[str]) -> Dict[str, RetrievalMetrics]:
        results = {}
        
        # Traditional RAG evaluation
        start_time = time.time()
        trad_retrieved = self.traditional_rag.retrieve(query)
        trad_latency = time.time() - start_time
        
        # HyDE evaluation
        start_time = time.time()
        hyde_retrieved = self.hyde_retriever.hyde_retrieve(query)
        hyde_latency = time.time() - start_time
        
        # Calculate metrics for both approaches
        results['traditional'] = RetrievalMetrics(
            precision=self._calculate_precision(trad_retrieved, relevant_docs),
            recall=self._calculate_recall(trad_retrieved, relevant_docs),
            f1=self._calculate_f1(trad_retrieved, relevant_docs),
            latency=trad_latency,
            memory_usage=self._get_memory_usage()
        )
        
        results['hyde'] = RetrievalMetrics(
            precision=self._calculate_precision(hyde_retrieved, relevant_docs),
            recall=self._calculate_recall(hyde_retrieved, relevant_docs),
            f1=self._calculate_f1(hyde_retrieved, relevant_docs),
            latency=hyde_latency,
            memory_usage=self._get_memory_usage()
        )
        
        self.metrics_history.append(results)
        return results
    
    def _calculate_precision(self, retrieved: List[tuple], relevant: List[str]) -> float:
        retrieved_docs = [doc for doc, _ in retrieved]
        return len(set(retrieved_docs) & set(relevant)) / len(retrieved_docs)
    
    def _calculate_recall(self, retrieved: List[tuple], relevant: List[str]) -> float:
        retrieved_docs = [doc for doc, _ in retrieved]
        return len(set(retrieved_docs) & set(relevant)) / len(relevant)
    
    def _calculate_f1(self, retrieved: List[tuple], relevant: List[str]) -> float:
        precision = self._calculate_precision(retrieved, relevant)
        recall = self._calculate_recall(retrieved, relevant)
        if precision + recall == 0:
            return 0
        return 2 * (precision * recall) / (precision + recall)
    
    def _get_memory_usage(self) -> int:
        import psutil
        process = psutil.Process()
        return process.memory_info().rss

# Example usage
benchmark = RAGBenchmark(traditional_rag=TraditionalRAG(), hyde_retriever=HyDERetriever())
query = "How does RAG compare to traditional search?"
relevant_docs = ["RAG combines retrieval with generation for better responses."]
metrics = benchmark.measure_performance(query, relevant_docs)

Advanced Query Preprocessing

Implementation of sophisticated query preprocessing techniques that enhance both Traditional RAG and HyDE performance through query expansion, normalization, and semantic analysis.

import nltk
from nltk.corpus import wordnet
from nltk.tokenize import word_tokenize
from nltk.corpus import stopwords
from typing import Set, List

class QueryPreprocessor:
    def __init__(self):
        nltk.download('punkt')
        nltk.download('wordnet')
        nltk.download('stopwords')
        self.stop_words = set(stopwords.words('english'))
        
    def preprocess_query(self, query: str, 
                        expand_synonyms: bool = True,
                        remove_stopwords: bool = True) -> str:
        tokens = word_tokenize(query.lower())
        
        if remove_stopwords:
            tokens = [t for t in tokens if t not in self.stop_words]
            
        if expand_synonyms:
            expanded_tokens = self._expand_with_synonyms(tokens)
            tokens.extend(expanded_tokens)
        
        return ' '.join(set(tokens))
    
    def _expand_with_synonyms(self, tokens: List[str]) -> Set[str]:
        synonyms = set()
        for token in tokens:
            for syn in wordnet.synsets(token):
                for lemma in syn.lemmas():
                    synonyms.add(lemma.name())
        return synonyms
    
    def generate_query_variations(self, query: str) -> List[str]:
        base_query = self.preprocess_query(query)
        variations = [base_query]
        
        # Add question variations
        if not query.strip().endswith('?'):
            variations.append(f"{base_query}?")
        
        # Add declarative variations
        variations.append(f"Information about {base_query}")
        variations.append(f"Tell me about {base_query}")
        
        # Add specific aspect variations
        aspects = ["definition", "example", "comparison", "application"]
        for aspect in aspects:
            variations.append(f"{base_query} {aspect}")
        
        return list(set(variations))

# Example usage
preprocessor = QueryPreprocessor()
query = "machine learning algorithms"
processed_query = preprocessor.preprocess_query(query)
variations = preprocessor.generate_query_variations(query)

print(f"Original query: {query}")
print(f"Processed query: {processed_query}")
print("Query variations:")
for i, var in enumerate(variations, 1):
    print(f"{i}. {var}")

Hybrid RAG-HyDE Architecture

An innovative implementation combining the strengths of both Traditional RAG and HyDE approaches in a single system, with dynamic switching based on query characteristics and performance metrics.

from enum import Enum
from typing import List, Tuple, Optional
import numpy as np

class RetrievalStrategy(Enum):
    TRADITIONAL = "traditional"
    HYDE = "hyde"
    HYBRID = "hybrid"

class HybridRAGHyDE:
    def __init__(self, confidence_threshold: float = 0.7):
        self.traditional_rag = TraditionalRAG()
        self.hyde_retriever = HyDERetriever()
        self.query_preprocessor = QueryPreprocessor()
        self.confidence_threshold = confidence_threshold
        self.performance_history = []
        
    def retrieve(self, query: str, k: int = 3) -> List[Tuple[str, float]]:
        strategy = self._select_strategy(query)
        processed_query = self.query_preprocessor.preprocess_query(query)
        
        if strategy == RetrievalStrategy.TRADITIONAL:
            results = self.traditional_rag.retrieve(processed_query, k)
        elif strategy == RetrievalStrategy.HYDE:
            results = self.hyde_retriever.hyde_retrieve(processed_query, k)
        else:  # HYBRID
            results = self._hybrid_retrieve(processed_query, k)
            
        self._update_performance_history(strategy, results)
        return results
    
    def _select_strategy(self, query: str) -> RetrievalStrategy:
        # Strategy selection based on query characteristics
        query_length = len(query.split())
        has_technical_terms = self._contains_technical_terms(query)
        is_ambiguous = self._check_query_ambiguity(query)
        
        if query_length <= 3 and not has_technical_terms:
            return RetrievalStrategy.TRADITIONAL
        elif is_ambiguous or has_technical_terms:
            return RetrievalStrategy.HYDE
        else:
            return RetrievalStrategy.HYBRID
    
    def _hybrid_retrieve(self, query: str, k: int) -> List[Tuple[str, float]]:
        trad_results = self.traditional_rag.retrieve(query, k)
        hyde_results = self.hyde_retriever.hyde_retrieve(query, k)
        
        # Combine and re-rank results
        all_results = {}
        for doc, score in trad_results:
            all_results[doc] = score
        
        for doc, score in hyde_results:
            if doc in all_results:
                all_results[doc] = max(all_results[doc], score)
            else:
                all_results[doc] = score
        
        # Sort and return top k results
        sorted_results = sorted(all_results.items(), 
                              key=lambda x: x[1], reverse=True)
        return sorted_results[:k]
    
    def _contains_technical_terms(self, query: str) -> bool:
        # Simplified technical term detection
        technical_keywords = {'algorithm', 'implementation', 'architecture',
                            'framework', 'methodology', 'protocol'}
        return any(term in query.lower() for term in technical_keywords)
    
    def _check_query_ambiguity(self, query: str) -> bool:
        variations = self.query_preprocessor.generate_query_variations(query)
        return len(variations) > 5
    
    def _update_performance_history(self, strategy: RetrievalStrategy,
                                  results: List[Tuple[str, float]]) -> None:
        avg_confidence = np.mean([score for _, score in results])
        self.performance_history.append({
            'strategy': strategy,
            'avg_confidence': avg_confidence,
            'timestamp': time.time()
        })

# Example usage
hybrid_system = HybridRAGHyDE()
query = "Explain the differences between neural networks and decision trees"
results = hybrid_system.retrieve(query)

Contextual Re-ranking System

A sophisticated re-ranking system that evaluates and adjusts document relevance scores based on contextual relationships, semantic similarity, and document freshness for both Traditional RAG and HyDE approaches.

from datetime import datetime
import numpy as np
from typing import List, Tuple, Dict
from dataclasses import dataclass

@dataclass
class DocumentMetadata:
    creation_date: datetime
    last_accessed: datetime
    access_count: int
    source_reliability: float

class ContextualReranker:
    def __init__(self, alpha: float = 0.3, beta: float = 0.2, gamma: float = 0.5):
        self.alpha = alpha  # Semantic similarity weight
        self.beta = beta    # Temporal relevance weight
        self.gamma = gamma  # Context coherence weight
        self.document_metadata: Dict[str, DocumentMetadata] = {}
        
    def rerank(self, query: str, initial_results: List[Tuple[str, float]],
               context: Optional[str] = None) -> List[Tuple[str, float]]:
        reranked_scores = {}
        
        for doc, score in initial_results:
            semantic_score = score
            temporal_score = self._calculate_temporal_score(doc)
            context_score = self._calculate_context_score(doc, context) if context else 0.5
            
            final_score = (
                self.alpha * semantic_score +
                self.beta * temporal_score +
                self.gamma * context_score
            )
            reranked_scores[doc] = final_score
            
        # Update access metadata
        self._update_access_metadata(reranked_scores.keys())
        
        # Sort by final score
        return sorted(reranked_scores.items(), key=lambda x: x[1], reverse=True)
    
    def _calculate_temporal_score(self, document: str) -> float:
        if document not in self.document_metadata:
            return 0.5  # Default score for new documents
            
        metadata = self.document_metadata[document]
        days_old = (datetime.now() - metadata.creation_date).days
        freshness_score = 1.0 / (1.0 + np.log1p(days_old))
        
        # Combine with popularity
        popularity_score = np.log1p(metadata.access_count) / 10.0
        return 0.7 * freshness_score + 0.3 * popularity_score
    
    def _calculate_context_score(self, document: str, context: str) -> float:
        if not context:
            return 0.5
            
        # Simple context matching score
        doc_words = set(document.lower().split())
        context_words = set(context.lower().split())
        overlap = len(doc_words.intersection(context_words))
        return overlap / (len(doc_words) + len(context_words) - overlap)
    
    def _update_access_metadata(self, accessed_docs: List[str]) -> None:
        current_time = datetime.now()
        
        for doc in accessed_docs:
            if doc not in self.document_metadata:
                self.document_metadata[doc] = DocumentMetadata(
                    creation_date=current_time,
                    last_accessed=current_time,
                    access_count=1,
                    source_reliability=0.8
                )
            else:
                metadata = self.document_metadata[doc]
                metadata.last_accessed = current_time
                metadata.access_count += 1

# Example usage
reranker = ContextualReranker()
initial_results = [
    ("Document about RAG", 0.8),
    ("Document about HyDE", 0.7),
    ("General ML document", 0.6)
]
context = "Comparing retrieval augmented generation approaches"
reranked_results = reranker.rerank(
    query="RAG vs HyDE performance",
    initial_results=initial_results,
    context=context
)

Adaptive Learning Component

Implementation of an adaptive learning system that continuously improves retrieval performance based on user feedback and interaction patterns for both RAG and HyDE approaches.

from collections import defaultdict
import numpy as np
from typing import Dict, List, Tuple, Optional
from sklearn.linear_model import LogisticRegression

class AdaptiveLearningSystem:
    def __init__(self):
        self.feedback_history: Dict[str, List[Tuple[str, bool]]] = defaultdict(list)
        self.feature_extractors = {
            'length': lambda x: len(x.split()),
            'technical_density': self._calculate_technical_density,
            'query_specificity': self._calculate_query_specificity
        }
        self.model = LogisticRegression()
        self.training_data: List[List[float]] = []
        self.training_labels: List[int] = []
        
    def update_from_feedback(self, query: str, document: str, 
                           was_relevant: bool) -> None:
        self.feedback_history[query].append((document, was_relevant))
        
        # Extract features and update training data
        features = self._extract_features(query, document)
        self.training_data.append(features)
        self.training_labels.append(1 if was_relevant else 0)
        
        # Retrain model if enough data
        if len(self.training_labels) >= 10:
            self._train_model()
    
    def predict_relevance(self, query: str, documents: List[str]) -> List[float]:
        if len(self.training_labels) < 10:
            return [0.5] * len(documents)  # Default confidence
            
        features = [self._extract_features(query, doc) for doc in documents]
        return self.model.predict_proba(features)[:, 1]
    
    def _extract_features(self, query: str, document: str) -> List[float]:
        features = []
        for extractor in self.feature_extractors.values():
            features.append(extractor(query))
            features.append(extractor(document))
        return features
    
    def _calculate_technical_density(self, text: str) -> float:
        technical_terms = {
            'algorithm', 'implementation', 'architecture', 'framework',
            'methodology', 'protocol', 'neural', 'embedding', 'vector'
        }
        words = text.lower().split()
        return sum(1 for w in words if w in technical_terms) / len(words)
    
    def _calculate_query_specificity(self, text: str) -> float:
        general_terms = {'what', 'how', 'why', 'when', 'where', 'who'}
        words = set(text.lower().split())
        return 1 - len(words.intersection(general_terms)) / len(words)
    
    def _train_model(self) -> None:
        if len(set(self.training_labels)) < 2:
            return  # Need both positive and negative examples
            
        self.model.fit(self.training_data, self.training_labels)
    
    def get_performance_metrics(self) -> Dict[str, float]:
        if len(self.training_labels) < 10:
            return {'accuracy': None, 'precision': None, 'recall': None}
            
        from sklearn.metrics import accuracy_score, precision_score, recall_score
        predictions = self.model.predict(self.training_data)
        return {
            'accuracy': accuracy_score(self.training_labels, predictions),
            'precision': precision_score(self.training_labels, predictions),
            'recall': recall_score(self.training_labels, predictions)
        }

# Example usage
adaptive_system = AdaptiveLearningSystem()

# Simulate user feedback
query = "Explain RAG architecture"
document = "RAG combines retrieval with generation for better responses"
adaptive_system.update_from_feedback(query, document, True)

# Predict relevance for new documents
new_documents = [
    "Technical overview of retrieval-augmented generation",
    "Basic introduction to machine learning",
    "Detailed comparison of RAG and HyDE approaches"
]
relevance_scores = adaptive_system.predict_relevance(query, new_documents)

Memory-Efficient Document Store

Implementation of a memory-efficient document storage system that optimizes RAM usage while maintaining fast retrieval capabilities for both Traditional RAG and HyDE approaches through intelligent caching and compression.

import lz4.frame
import pickle
from typing import Dict, Optional, List, Tuple
import mmh3  # MurmurHash3 for efficient hashing
from collections import OrderedDict

class MemoryEfficientStore:
    def __init__(self, max_memory_mb: int = 100):
        self.max_memory = max_memory_mb * 1024 * 1024  # Convert to bytes
        self.compressed_docs: Dict[str, bytes] = {}
        self.doc_metadata: Dict[str, Dict] = {}
        self.cache = OrderedDict()
        self.current_memory = 0
    
    def add_document(self, doc_id: str, content: str, 
                    metadata: Optional[Dict] = None) -> None:
        # Compress document content
        compressed = lz4.frame.compress(content.encode())
        
        # Calculate memory usage
        memory_usage = len(compressed) + len(pickle.dumps(metadata or {}))
        
        # Check if we need to free up space
        while self.current_memory + memory_usage > self.max_memory:
            if not self._free_memory():
                raise MemoryError("Cannot allocate more memory")
        
        self.compressed_docs[doc_id] = compressed
        self.doc_metadata[doc_id] = metadata or {}
        self.current_memory += memory_usage
    
    def get_document(self, doc_id: str) -> Tuple[str, Dict]:
        # Check cache first
        if doc_id in self.cache:
            content, metadata = self.cache[doc_id]
            self.cache.move_to_end(doc_id)  # Move to end (most recently used)
            return content, metadata
        
        # Decompress and cache document
        if doc_id in self.compressed_docs:
            content = lz4.frame.decompress(
                self.compressed_docs[doc_id]
            ).decode()
            metadata = self.doc_metadata[doc_id]
            
            # Add to cache
            self._cache_document(doc_id, content, metadata)
            
            return content, metadata
        
        raise KeyError(f"Document {doc_id} not found")
    
    def _cache_document(self, doc_id: str, content: str, 
                       metadata: Dict) -> None:
        # Implement LRU cache
        if len(self.cache) >= 100:  # Max cache size
            self.cache.popitem(last=False)  # Remove least recently used
        self.cache[doc_id] = (content, metadata)
    
    def _free_memory(self) -> bool:
        if not self.compressed_docs:
            return False
        
        # Remove least recently accessed document
        oldest_doc_id = min(
            self.doc_metadata.keys(),
            key=lambda x: self.doc_metadata[x].get('last_accessed', 0)
        )
        
        memory_freed = (
            len(self.compressed_docs[oldest_doc_id]) + 
            len(pickle.dumps(self.doc_metadata[oldest_doc_id]))
        )
        
        del self.compressed_docs[oldest_doc_id]
        del self.doc_metadata[oldest_doc_id]
        self.current_memory -= memory_freed
        
        return True
    
    def search(self, query: str, k: int = 5) -> List[Tuple[str, float]]:
        results = []
        query_hash = mmh3.hash(query)
        
        for doc_id in self.compressed_docs:
            # Compute similarity using metadata and minimal decompression
            doc_content, _ = self.get_document(doc_id)
            doc_hash = mmh3.hash(doc_content)
            
            # Simple similarity metric based on hash comparison
            similarity = 1.0 / (1.0 + abs(query_hash - doc_hash))
            results.append((doc_id, similarity))
        
        return sorted(results, key=lambda x: x[1], reverse=True)[:k]

# Example usage
doc_store = MemoryEfficientStore(max_memory_mb=50)

# Add some documents
docs = {
    "doc1": "Detailed explanation of RAG architecture and implementation",
    "doc2": "HyDE methodology and practical applications",
    "doc3": "Comparison of various retrieval techniques in modern AI"
}

for doc_id, content in docs.items():
    doc_store.add_document(
        doc_id, 
        content,
        metadata={"created": "2024-01-01", "type": "technical"}
    )

# Retrieve and search
retrieved_doc, metadata = doc_store.get_document("doc1")
search_results = doc_store.search("retrieval techniques")

Real-time Performance Monitoring

Implementation of a comprehensive monitoring system that tracks and analyzes the performance metrics of both Traditional RAG and HyDE approaches in real-time, enabling dynamic optimization and system adaptation.

from datetime import datetime, timedelta
import numpy as np
from typing import Dict, List, Tuple
import json
from collections import deque

class PerformanceMonitor:
    def __init__(self, window_size: int = 1000):
        self.metrics_window = deque(maxlen=window_size)
        self.alert_thresholds = {
            'latency': 1.0,  # seconds
            'relevance_score': 0.7,
            'memory_usage': 1024 * 1024 * 1024  # 1GB
        }
        self.performance_summary = {
            'traditional_rag': {'success_rate': 0, 'avg_latency': 0},
            'hyde': {'success_rate': 0, 'avg_latency': 0}
        }
    
    def record_query(self, query_data: Dict) -> None:
        query_data['timestamp'] = datetime.now()
        self.metrics_window.append(query_data)
        self._update_performance_summary()
        self._check_alerts(query_data)
    
    def get_performance_report(self, 
                             time_window: timedelta = timedelta(hours=1)
                             ) -> Dict:
        current_time = datetime.now()
        relevant_metrics = [
            m for m in self.metrics_window
            if current_time - m['timestamp'] <= time_window
        ]
        
        if not relevant_metrics:
            return {"error": "No data available for specified time window"}
        
        return {
            'summary': self._calculate_summary(relevant_metrics),
            'trends': self._analyze_trends(relevant_metrics),
            'recommendations': self._generate_recommendations(relevant_metrics)
        }
    
    def _update_performance_summary(self) -> None:
        for approach in ['traditional_rag', 'hyde']:
            relevant_queries = [
                m for m in self.metrics_window
                if m['approach'] == approach
            ]
            
            if relevant_queries:
                success_rate = sum(
                    1 for q in relevant_queries
                    if q['relevance_score'] >= self.alert_thresholds['relevance_score']
                ) / len(relevant_queries)
                
                avg_latency = np.mean([q['latency'] for q in relevant_queries])
                
                self.performance_summary[approach] = {
                    'success_rate': success_rate,
                    'avg_latency': avg_latency
                }
    
    def _calculate_summary(self, metrics: List[Dict]) -> Dict:
        return {
            'total_queries': len(metrics),
            'avg_latency': np.mean([m['latency'] for m in metrics]),
            'avg_relevance': np.mean([m['relevance_score'] for m in metrics]),
            'memory_usage': np.mean([m['memory_usage'] for m in metrics]),
            'success_rate': sum(
                1 for m in metrics
                if m['relevance_score'] >= self.alert_thresholds['relevance_score']
            ) / len(metrics)
        }
    
    def _analyze_trends(self, metrics: List[Dict]) -> Dict:
        timestamps = [m['timestamp'] for m in metrics]
        latencies = [m['latency'] for m in metrics]
        relevance_scores = [m['relevance_score'] for m in metrics]
        
        return {
            'latency_trend': self._calculate_trend(latencies),
            'relevance_trend': self._calculate_trend(relevance_scores),
            'query_volume': self._calculate_query_volume(timestamps)
        }
    
    def _calculate_trend(self, values: List[float]) -> str:
        if len(values) < 2:
            return "insufficient_data"
            
        slope = np.polyfit(range(len(values)), values, 1)[0]
        if slope > 0.01:
            return "increasing"
        elif slope < -0.01:
            return "decreasing"
        return "stable"
    
    def _calculate_query_volume(self, timestamps: List[datetime]) -> Dict:
        hours = [t.hour for t in timestamps]
        volume_by_hour = {h: hours.count(h) for h in range(24)}
        return volume_by_hour
    
    def _check_alerts(self, query_data: Dict) -> None:
        alerts = []
        
        if query_data['latency'] > self.alert_thresholds['latency']:
            alerts.append({
                'type': 'high_latency',
                'value': query_data['latency'],
                'threshold': self.alert_thresholds['latency']
            })
        
        if query_data['memory_usage'] > self.alert_thresholds['memory_usage']:
            alerts.append({
                'type': 'high_memory_usage',
                'value': query_data['memory_usage'],
                'threshold': self.alert_thresholds['memory_usage']
            })
        
        if alerts:
            self._handle_alerts(alerts, query_data)
    
    def _handle_alerts(self, alerts: List[Dict], query_data: Dict) -> None:
        # Log alerts and take appropriate action
        for alert in alerts:
            print(f"ALERT: {alert['type']} - "
                  f"Value: {alert['value']}, "
                  f"Threshold: {alert['threshold']}")

# Example usage
monitor = PerformanceMonitor()

# Record some sample queries
sample_query = {
    'approach': 'traditional_rag',
    'query': 'RAG vs HyDE comparison',
    'latency': 0.5,
    'relevance_score': 0.85,
    'memory_usage': 512 * 1024 * 1024  # 512MB
}
monitor.record_query(sample_query)

# Get performance report
report = monitor.get_performance_report(time_window=timedelta(minutes=30))
print(json.dumps(report, indent=2, default=str))

Results Analysis Framework

A comprehensive framework for analyzing and comparing the results of Traditional RAG and HyDE approaches, incorporating multiple evaluation metrics and statistical analysis to provide actionable insights.

import numpy as np
from scipy import stats
from typing import List, Dict, Tuple
from dataclasses import dataclass
from sklearn.metrics import normalized_mutual_info_score

@dataclass
class EvaluationResult:
    precision: float
    recall: float
    f1_score: float
    latency: float
    semantic_similarity: float
    confidence: float

class ResultsAnalyzer:
    def __init__(self):
        self.traditional_results: List[EvaluationResult] = []
        self.hyde_results: List[EvaluationResult] = []
        
    def add_evaluation(self, result: EvaluationResult, 
                      is_traditional: bool) -> None:
        if is_traditional:
            self.traditional_results.append(result)
        else:
            self.hyde_results.append(result)
    
    def compare_approaches(self) -> Dict:
        if not self.traditional_results or not self.hyde_results:
            return {"error": "Insufficient data for comparison"}
        
        comparison = {
            'performance_metrics': self._compare_performance_metrics(),
            'statistical_analysis': self._perform_statistical_analysis(),
            'efficiency_analysis': self._analyze_efficiency(),
            'recommendations': self._generate_recommendations()
        }
        
        return comparison
    
    def _compare_performance_metrics(self) -> Dict:
        trad_metrics = self._calculate_average_metrics(self.traditional_results)
        hyde_metrics = self._calculate_average_metrics(self.hyde_results)
        
        return {
            'traditional_rag': trad_metrics,
            'hyde': hyde_metrics,
            'relative_improvement': self._calculate_relative_improvement(
                trad_metrics, hyde_metrics
            )
        }
    
    def _calculate_average_metrics(self, 
                                 results: List[EvaluationResult]) -> Dict:
        return {
            'precision': np.mean([r.precision for r in results]),
            'recall': np.mean([r.recall for r in results]),
            'f1_score': np.mean([r.f1_score for r in results]),
            'latency': np.mean([r.latency for r in results]),
            'semantic_similarity': np.mean([r.semantic_similarity for r in results]),
            'confidence': np.mean([r.confidence for r in results])
        }
    
    def _calculate_relative_improvement(self, trad_metrics: Dict,
                                     hyde_metrics: Dict) -> Dict:
        improvements = {}
        for metric in trad_metrics:
            if trad_metrics[metric] > 0:
                rel_imp = ((hyde_metrics[metric] - trad_metrics[metric]) / 
                          trad_metrics[metric] * 100)
                improvements[metric] = f"{rel_imp:.2f}%"
        return improvements
    
    def _perform_statistical_analysis(self) -> Dict:
        analysis = {}
        
        # Prepare data for analysis
        metrics = ['precision', 'recall', 'f1_score', 'latency',
                  'semantic_similarity', 'confidence']
        
        for metric in metrics:
            trad_values = [getattr(r, metric) for r in self.traditional_results]
            hyde_values = [getattr(r, metric) for r in self.hyde_results]
            
            # Perform t-test
            t_stat, p_value = stats.ttest_ind(trad_values, hyde_values)
            
            # Calculate effect size (Cohen's d)
            effect_size = (np.mean(hyde_values) - np.mean(trad_values)) / (
                np.sqrt((np.var(hyde_values) + np.var(trad_values)) / 2)
            )
            
            analysis[metric] = {
                't_statistic': t_stat,
                'p_value': p_value,
                'effect_size': effect_size,
                'significant_difference': p_value < 0.05
            }
            
        return analysis
    
    def _analyze_efficiency(self) -> Dict:
        trad_latencies = [r.latency for r in self.traditional_results]
        hyde_latencies = [r.latency for r in self.hyde_results]
        
        return {
            'traditional_rag': {
                'avg_latency': np.mean(trad_latencies),
                'latency_std': np.std(trad_latencies),
                'latency_95th': np.percentile(trad_latencies, 95)
            },
            'hyde': {
                'avg_latency': np.mean(hyde_latencies),
                'latency_std': np.std(hyde_latencies),
                'latency_95th': np.percentile(hyde_latencies, 95)
            }
        }
    
    def _generate_recommendations(self) -> List[str]:
        recommendations = []
        performance_comparison = self._compare_performance_metrics()
        
        # Analyze performance differences
        for metric, improvement in (
            performance_comparison['relative_improvement'].items()
        ):
            imp_value = float(improvement.strip('%'))
            if abs(imp_value) > 10:
                better_approach = (
                    "HyDE" if imp_value > 0 else "Traditional RAG"
                )
                recommendations.append(
                    f"Consider using {better_approach} for tasks where "
                    f"{metric} is critical ({abs(imp_value):.1f}% difference)"
                )
        
        return recommendations

# Example usage
analyzer = ResultsAnalyzer()

# Add sample evaluation results
trad_result = EvaluationResult(
    precision=0.85,
    recall=0.78,
    f1_score=0.81,
    latency=0.5,
    semantic_similarity=0.92,
    confidence=0.88
)

hyde_result = EvaluationResult(
    precision=0.89,
    recall=0.82,
    f1_score=0.85,
    latency=0.7,
    semantic_similarity=0.95,
    confidence=0.91
)

analyzer.add_evaluation(trad_result, is_traditional=True)
analyzer.add_evaluation(hyde_result, is_traditional=False)

# Get comparison results
comparison_results = analyzer.compare_approaches()

Additional Resources

Note: As mentioned, these are example resources. Please verify URLs and search for current research papers on these topics.

More in RAG and retrieval

  1. The Limitations of Retrieval-Augmented Generation (RAG) Systems8 min

    Retrieval-Augmented Generation (RAG) systems are not the plug-and-play solution many expect. While they offer exciting possibilities for utilizing Large...

  2. Retrieval Augmented Generation (RAG) in Large Language Models16 min

    A comprehensive implementation of core RAG system components including document loading, text chunking, and embedding generation using modern Python...

  3. Transforming Industries with 12 Types of RAG14 min

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

  4. Real-Time GraphRAG Application with LangChain Neo4j and GPT8 min

    GraphRAG (Graph Retrieval-Augmented Generation) is an innovative approach that combines graph databases, natural language processing, and large language...

  5. Tree-RAG (T-RAG), an enhanced RAG technique with Python10 min

    Tree-RAG (T-RAG) is an enhanced Retrieval-Augmented Generation (RAG) technique that leverages tree structures to improve information retrieval and...

  6. Re-ranking in Retrieval-Augmented Generation8 min

    BM25 serves as a foundational retrieval algorithm in RAG systems, implementing a probabilistic scoring function that ranks documents based on query term...