""" Hybrid retrieval combining dense semantic search with sparse BM25 keyword matching. Uses Reciprocal Rank Fusion (RRF) to combine results from both approaches. """ from typing import List, Dict, Tuple, Optional import numpy as np from pathlib import Path import sys # Add project root to Python path for imports project_root = Path(__file__).parent.parent.parent / "project-1-technical-rag" sys.path.append(str(project_root)) from src.sparse_retrieval import BM25SparseRetriever from src.fusion import reciprocal_rank_fusion, adaptive_fusion from shared_utils.embeddings.generator import generate_embeddings import faiss class HybridRetriever: """ Hybrid retrieval system combining dense semantic search with sparse BM25. Optimized for technical documentation where both semantic similarity and exact keyword matching are important for retrieval quality. Performance: Sub-second search on 1000+ document corpus """ def __init__( self, dense_weight: float = 0.7, embedding_model: str = "sentence-transformers/multi-qa-MiniLM-L6-cos-v1", use_mps: bool = True, bm25_k1: float = 1.2, bm25_b: float = 0.75, rrf_k: int = 10 ): """ Initialize hybrid retriever with dense and sparse components. Args: dense_weight: Weight for semantic similarity in fusion (0.7 default) embedding_model: Sentence transformer model name use_mps: Use Apple Silicon MPS acceleration for embeddings bm25_k1: BM25 term frequency saturation parameter bm25_b: BM25 document length normalization parameter rrf_k: Reciprocal Rank Fusion constant (1=strong rank preference, 2=moderate) Raises: ValueError: If parameters are invalid """ if not 0 <= dense_weight <= 1: raise ValueError("dense_weight must be between 0 and 1") self.dense_weight = dense_weight self.embedding_model = embedding_model self.use_mps = use_mps self.rrf_k = rrf_k # Initialize sparse retriever self.sparse_retriever = BM25SparseRetriever(k1=bm25_k1, b=bm25_b) # Dense retrieval components (initialized on first index) self.dense_index: Optional[faiss.Index] = None self.chunks: List[Dict] = [] self.embeddings: Optional[np.ndarray] = None def index_documents(self, chunks: List[Dict]) -> None: """ Index documents for both dense and sparse retrieval. Args: chunks: List of chunk dictionaries with 'text' field Raises: ValueError: If chunks is empty or malformed Performance: ~100 chunks/second for complete indexing """ if not chunks: raise ValueError("Cannot index empty chunk list") print(f"Indexing {len(chunks)} chunks for hybrid retrieval...") # Store chunks for retrieval self.chunks = chunks # Index for sparse retrieval print("Building BM25 sparse index...") self.sparse_retriever.index_documents(chunks) # Index for dense retrieval print("Building dense semantic index...") texts = [chunk['text'] for chunk in chunks] # Generate embeddings self.embeddings = generate_embeddings( texts, model_name=self.embedding_model, use_mps=self.use_mps ) # Create FAISS index embedding_dim = self.embeddings.shape[1] self.dense_index = faiss.IndexFlatIP(embedding_dim) # Inner product for cosine similarity # Normalize embeddings for cosine similarity faiss.normalize_L2(self.embeddings) self.dense_index.add(self.embeddings) print(f"Hybrid indexing complete: {len(chunks)} chunks ready for search") def search( self, query: str, top_k: int = 10, dense_top_k: Optional[int] = None, sparse_top_k: Optional[int] = None ) -> List[Tuple[int, float, Dict]]: """ Hybrid search combining dense and sparse retrieval with RRF. Args: query: Search query string top_k: Final number of results to return dense_top_k: Results from dense search (default: 2*top_k) sparse_top_k: Results from sparse search (default: 2*top_k) Returns: List of (chunk_index, rrf_score, chunk_dict) tuples Raises: ValueError: If not indexed or invalid parameters Performance: <200ms for 1000+ document corpus """ if self.dense_index is None: raise ValueError("Must call index_documents() before searching") if not query.strip(): return [] if top_k <= 0: raise ValueError("top_k must be positive") # Set default intermediate result counts if dense_top_k is None: dense_top_k = min(2 * top_k, len(self.chunks)) if sparse_top_k is None: sparse_top_k = min(2 * top_k, len(self.chunks)) # Dense semantic search dense_results = self._dense_search(query, dense_top_k) # Sparse BM25 search sparse_results = self.sparse_retriever.search(query, sparse_top_k) # Combine using Adaptive Fusion (better for small result sets) fused_results = adaptive_fusion( dense_results=dense_results, sparse_results=sparse_results, dense_weight=self.dense_weight, result_size=top_k ) # Prepare final results with chunk content and apply source diversity final_results = [] for chunk_idx, rrf_score in fused_results: chunk_dict = self.chunks[chunk_idx] final_results.append((chunk_idx, rrf_score, chunk_dict)) # Apply source diversity enhancement diverse_results = self._enhance_source_diversity(final_results, top_k) return diverse_results def _dense_search(self, query: str, top_k: int) -> List[Tuple[int, float]]: """ Perform dense semantic search using FAISS. Args: query: Search query top_k: Number of results to return Returns: List of (chunk_index, similarity_score) tuples """ # Generate query embedding query_embedding = generate_embeddings( [query], model_name=self.embedding_model, use_mps=self.use_mps ) # Normalize for cosine similarity faiss.normalize_L2(query_embedding) # Search dense index similarities, indices = self.dense_index.search(query_embedding, top_k) # Convert to required format results = [ (int(indices[0][i]), float(similarities[0][i])) for i in range(len(indices[0])) if indices[0][i] != -1 # Filter out invalid results ] return results def _enhance_source_diversity( self, results: List[Tuple[int, float, Dict]], top_k: int, max_per_source: int = 2 ) -> List[Tuple[int, float, Dict]]: """ Enhance source diversity in retrieval results to prevent over-focusing on single documents. Args: results: List of (chunk_idx, score, chunk_dict) tuples sorted by relevance top_k: Maximum number of results to return max_per_source: Maximum chunks allowed per source document Returns: Diversified results maintaining relevance while improving source coverage """ if not results: return [] source_counts = {} diverse_results = [] # First pass: Add highest scoring results respecting source limits for chunk_idx, score, chunk_dict in results: source = chunk_dict.get('source', 'unknown') current_count = source_counts.get(source, 0) if current_count < max_per_source: diverse_results.append((chunk_idx, score, chunk_dict)) source_counts[source] = current_count + 1 if len(diverse_results) >= top_k: break # Second pass: If we still need more results, relax source constraints if len(diverse_results) < top_k: for chunk_idx, score, chunk_dict in results: if (chunk_idx, score, chunk_dict) not in diverse_results: diverse_results.append((chunk_idx, score, chunk_dict)) if len(diverse_results) >= top_k: break return diverse_results[:top_k] def get_retrieval_stats(self) -> Dict[str, any]: """ Get statistics about the indexed corpus and retrieval performance. Returns: Dictionary with corpus statistics """ if not self.chunks: return {"status": "not_indexed"} return { "status": "indexed", "total_chunks": len(self.chunks), "dense_index_size": self.dense_index.ntotal if self.dense_index else 0, "embedding_dim": self.embeddings.shape[1] if self.embeddings is not None else 0, "sparse_indexed_chunks": len(self.sparse_retriever.chunk_mapping), "dense_weight": self.dense_weight, "sparse_weight": 1.0 - self.dense_weight, "rrf_k": self.rrf_k }