gmf_forge_ai_data.retrieval

 35class BaseRetriever(ABC):
 36    """
 37    Abstract base class for all retriever implementations.
 38    
 39    Retrievers provide various strategies for finding relevant documents
 40    from a vector store or other data source.
 41    
 42    All implementations must provide:
 43    - retrieve(): Main retrieval method
 44    """
 45    
 46    @abstractmethod
 47    def retrieve(
 48        self,
 49        query: RetrievalQuery
 50    ) -> List[SearchResult]:
 51        """
 52        Retrieve relevant documents based on the query.
 53        
 54        Args:
 55            query: RetrievalQuery containing query parameters
 56        
 57        Returns:
 58            List of SearchResult objects, ordered by relevance
 59        
 60        Raises:
 61            ValueError: If required query parameters are missing
 62        """
 63        pass
 64    
 65    def retrieve_text(
 66        self,
 67        text: str,
 68        top_k: int = 5,
 69        filters: Optional[Dict[str, Any]] = None
 70    ) -> List[SearchResult]:
 71        """
 72        Convenience method for text-based retrieval.
 73        
 74        Args:
 75            text: Query text
 76            top_k: Number of results to retrieve
 77            filters: Optional metadata filters
 78        
 79        Returns:
 80            List of SearchResult objects
 81        """
 82        query = RetrievalQuery(
 83            text=text,
 84            top_k=top_k,
 85            filters=filters
 86        )
 87        return self.retrieve(query)
 88    
 89    def retrieve_embedding(
 90        self,
 91        embedding: List[float],
 92        top_k: int = 5,
 93        filters: Optional[Dict[str, Any]] = None
 94    ) -> List[SearchResult]:
 95        """
 96        Convenience method for embedding-based retrieval.
 97        
 98        Args:
 99            embedding: Query embedding vector
100            top_k: Number of results to retrieve
101            filters: Optional metadata filters
102        
103        Returns:
104            List of SearchResult objects
105        """
106        query = RetrievalQuery(
107            embedding=embedding,
108            top_k=top_k,
109            filters=filters
110        )
111        return self.retrieve(query)

16@dataclass
17class RetrievalQuery:
18    """
19    Represents a retrieval query with optional query text and embedding.
20    
21    Attributes:
22        text: Query text (required for keyword/hybrid search)
23        embedding: Query embedding vector (required for vector search)
24        top_k: Number of results to retrieve
25        filters: Metadata filters to apply
26        metadata: Additional query metadata
27    """
28    text: Optional[str] = None
29    embedding: Optional[List[float]] = None
30    top_k: int = 5
31    filters: Optional[Dict[str, Any]] = None
32    metadata: Optional[Dict[str, Any]] = None

17class VectorRetriever(BaseRetriever):
18    """
19    Vector similarity retriever using cosine similarity.
20    
21    Performs pure vector search using embeddings. Requires query embeddings to be
22    provided externally (e.g., using an embeddings provider).
23    
24    Example:
25        ```python
26        from gmf_forge_ai_data.retrieval import VectorRetriever, RetrievalQuery
27        from gmf_forge_ai_data.embeddings import AzureOpenAIEmbeddings
28        
29        # Setup
30        embedder = AzureOpenAIEmbeddings(...)
31        retriever = VectorRetriever(vector_store)
32        
33        # Retrieve
34        query_text = "What is machine learning?"
35        query_embedding = embedder.embed_text(query_text)
36        
37        results = retriever.retrieve_embedding(
38            embedding=query_embedding,
39            top_k=5
40        )
41        ```
42    """
43    
44    def __init__(self, vector_store: BaseVectorStore):
45        """
46        Initialize vector retriever.
47        
48        Args:
49            vector_store: Vector store to search
50        """
51        self.vector_store = vector_store
52    
53    def retrieve(self, query: RetrievalQuery) -> List[SearchResult]:
54        """
55        Retrieve documents using vector similarity search.
56        
57        Args:
58            query: RetrievalQuery with embedding and parameters
59        
60        Returns:
61            List of SearchResult objects ordered by similarity
62        
63        Raises:
64            ValueError: If query.embedding is None
65        """
66        if query.embedding is None:
67            raise ValueError("VectorRetriever requires query.embedding")
68        
69        return self.vector_store.search(
70            query_embedding=query.embedding,
71            top_k=query.top_k,
72            filters=query.filters,
73            search_type="vector"
74        )

 77class KeywordRetriever(BaseRetriever):
 78    """
 79    Keyword/BM25 retriever using text matching.
 80    
 81    Performs traditional keyword-based search without using embeddings.
 82    Uses BM25 for Azure AI Search, Jaccard similarity for in-memory.
 83    
 84    Example:
 85        ```python
 86        from gmf_forge_ai_data.retrieval import KeywordRetriever, RetrievalQuery
 87        
 88        retriever = KeywordRetriever(vector_store)
 89        
 90        results = retriever.retrieve_text(
 91            text="machine learning algorithms",
 92            top_k=5
 93        )
 94        ```
 95    """
 96    
 97    def __init__(self, vector_store: BaseVectorStore):
 98        """
 99        Initialize keyword retriever.
100        
101        Args:
102            vector_store: Vector store to search
103        """
104        self.vector_store = vector_store
105    
106    def retrieve(self, query: RetrievalQuery) -> List[SearchResult]:
107        """
108        Retrieve documents using keyword search.
109        
110        Args:
111            query: RetrievalQuery with text and parameters
112        
113        Returns:
114            List of SearchResult objects ordered by keyword relevance
115        
116        Raises:
117            ValueError: If query.text is None
118        """
119        if query.text is None:
120            raise ValueError("KeywordRetriever requires query.text")
121        
122        return self.vector_store.search(
123            query=query.text,
124            top_k=query.top_k,
125            filters=query.filters,
126            search_type="keyword"
127        )

130class HybridRetriever(BaseRetriever):
131    """
132    Hybrid retriever combining vector and keyword search.
133    
134    Combines vector similarity with keyword matching for comprehensive retrieval.
135    Scores are normalized and combined (typically 50/50 weighting).
136    
137    Example:
138        ```python
139        from gmf_forge_ai_data.retrieval import HybridRetriever, RetrievalQuery
140        from gmf_forge_ai_data.embeddings import AzureOpenAIEmbeddings
141        
142        embedder = AzureOpenAIEmbeddings(...)
143        retriever = HybridRetriever(vector_store)
144        
145        query_text = "machine learning"
146        query_embedding = embedder.embed_text(query_text)
147        
148        # Hybrid search using both text and embedding
149        query = RetrievalQuery(
150            text=query_text,
151            embedding=query_embedding,
152            top_k=5
153        )
154        results = retriever.retrieve(query)
155        ```
156    """
157    
158    def __init__(self, vector_store: BaseVectorStore):
159        """
160        Initialize hybrid retriever.
161        
162        Args:
163            vector_store: Vector store to search
164        """
165        self.vector_store = vector_store
166    
167    def retrieve(self, query: RetrievalQuery) -> List[SearchResult]:
168        """
169        Retrieve documents using hybrid search (vector + keyword).
170        
171        Args:
172            query: RetrievalQuery with both text and embedding
173        
174        Returns:
175            List of SearchResult objects ordered by combined relevance
176        
177        Raises:
178            ValueError: If query.text or query.embedding is None
179        """
180        if query.text is None:
181            raise ValueError("HybridRetriever requires query.text")
182        if query.embedding is None:
183            raise ValueError("HybridRetriever requires query.embedding")
184        
185        return self.vector_store.search(
186            query=query.text,
187            query_embedding=query.embedding,
188            top_k=query.top_k,
189            filters=query.filters,
190            search_type="hybrid"
191        )

 16class MMRRetriever(BaseRetriever):
 17    """
 18    Maximal Marginal Relevance (MMR) retriever for diverse results.
 19    
 20    MMR reranks initial retrieval results to balance relevance with diversity:
 21    - High lambda (λ → 1.0): Prioritize relevance
 22    - Low lambda (λ → 0.0): Prioritize diversity
 23    - λ = 0.5: Balanced (default)
 24    
 25    Algorithm:
 26    1. Retrieve initial candidate documents (fetch_k results)
 27    2. Select most relevant document first
 28    3. For each remaining selection:
 29       - Score = λ * relevance - (1-λ) * max_similarity_to_selected
 30       - Choose document with highest score
 31    4. Return top_k diverse results
 32    
 33    Benefits:
 34    - Reduces redundancy in results
 35    - Improves coverage of different aspects
 36    - Useful for exploration and summarization
 37    
 38    Example:
 39        ```python
 40        from gmf_forge_ai_data.retrieval import MMRRetriever, RetrievalQuery
 41        from gmf_forge_ai_data.embeddings import AzureOpenAIEmbeddings
 42        
 43        embedder = AzureOpenAIEmbeddings(...)
 44        retriever = MMRRetriever(
 45            vector_store=vector_store,
 46            lambda_param=0.5,  # Balanced relevance/diversity
 47            fetch_k=20         # Retrieve 20 candidates, return top_k diverse
 48        )
 49        
 50        query_text = "machine learning algorithms"
 51        query_embedding = embedder.embed_text(query_text)
 52        
 53        # Returns 5 diverse results from 20 candidates
 54        results = retriever.retrieve_embedding(
 55            embedding=query_embedding,
 56            top_k=5
 57        )
 58        ```
 59    
 60    References:
 61        Carbonell, J., & Goldstein, J. (1998). The use of MMR, diversity-based
 62        reranking for reordering documents and producing summaries.
 63    """
 64    
 65    def __init__(
 66        self,
 67        vector_store: BaseVectorStore,
 68        lambda_param: float = 0.5,
 69        fetch_k: int = 20
 70    ):
 71        """
 72        Initialize MMR retriever.
 73        
 74        Args:
 75            vector_store: Vector store to search
 76            lambda_param: Balance between relevance (1.0) and diversity (0.0)
 77            fetch_k: Number of initial candidates to fetch (should be >= top_k)
 78        
 79        Raises:
 80            ValueError: If lambda_param not in [0, 1] or fetch_k < 1
 81        """
 82        if not 0 <= lambda_param <= 1:
 83            raise ValueError(f"lambda_param must be in [0, 1], got {lambda_param}")
 84        if fetch_k < 1:
 85            raise ValueError(f"fetch_k must be >= 1, got {fetch_k}")
 86        
 87        self.vector_store = vector_store
 88        self.lambda_param = lambda_param
 89        self.fetch_k = fetch_k
 90    
 91    def retrieve(self, query: RetrievalQuery) -> List[SearchResult]:
 92        """
 93        Retrieve diverse documents using MMR reranking.
 94        
 95        Args:
 96            query: RetrievalQuery with embedding (required)
 97        
 98        Returns:
 99            List of SearchResult objects with diverse, relevant results
100        
101        Raises:
102            ValueError: If query.embedding is None
103        """
104        if query.embedding is None:
105            raise ValueError("MMRRetriever requires query.embedding for diversity calculation")
106        
107        # Fetch initial candidates (more than top_k)
108        fetch_k = max(self.fetch_k, query.top_k)
109        candidates = self.vector_store.search(
110            query_embedding=query.embedding,
111            top_k=fetch_k,
112            filters=query.filters,
113            search_type="vector"
114        )
115        
116        if len(candidates) == 0:
117            return []
118        
119        if len(candidates) <= query.top_k:
120            # Not enough candidates for MMR, return as-is
121            return candidates[:query.top_k]
122        
123        # Extract embeddings and relevance scores
124        query_embedding = np.array(query.embedding, dtype=np.float32)
125        candidate_embeddings = []
126        candidate_scores = []
127        
128        for result in candidates:
129            if result.document.embedding is None:
130                raise ValueError(
131                    f"Document {result.document.id} has no embedding. "
132                    "MMR requires all documents to have embeddings."
133                )
134            candidate_embeddings.append(result.document.embedding)
135            candidate_scores.append(result.score)
136        
137        candidate_embeddings = np.array(candidate_embeddings, dtype=np.float32)
138        candidate_scores = np.array(candidate_scores, dtype=np.float32)
139        
140        # Normalize embeddings for cosine similarity
141        query_norm = query_embedding / (np.linalg.norm(query_embedding) + 1e-10)
142        candidate_norms = candidate_embeddings / (
143            np.linalg.norm(candidate_embeddings, axis=1, keepdims=True) + 1e-10
144        )
145        
146        # MMR selection
147        selected_indices = []
148        selected_embeddings = []
149        
150        for _ in range(min(query.top_k, len(candidates))):
151            if len(selected_indices) == 0:
152                # First selection: most relevant
153                best_idx = int(np.argmax(candidate_scores))
154            else:
155                # Subsequent selections: balance relevance and diversity
156                mmr_scores = []
157                
158                for idx in range(len(candidates)):
159                    if idx in selected_indices:
160                        mmr_scores.append(-np.inf)
161                        continue
162                    
163                    # Relevance score (already normalized 0-1 from vector store)
164                    relevance = candidate_scores[idx]
165                    
166                    # Diversity score: max similarity to selected documents
167                    doc_embedding = candidate_norms[idx]
168                    similarities = [
169                        np.dot(doc_embedding, selected_embeddings[i])
170                        for i in range(len(selected_embeddings))
171                    ]
172                    max_similarity = max(similarities)
173                    
174                    # MMR score: λ * relevance - (1-λ) * max_similarity
175                    mmr_score = (
176                        self.lambda_param * relevance -
177                        (1 - self.lambda_param) * max_similarity
178                    )
179                    mmr_scores.append(mmr_score)
180                
181                best_idx = int(np.argmax(mmr_scores))
182            
183            selected_indices.append(best_idx)
184            selected_embeddings.append(candidate_norms[best_idx])
185        
186        # Build results with updated ranks
187        mmr_results = []
188        for rank, idx in enumerate(selected_indices):
189            result = candidates[idx]
190            mmr_results.append(SearchResult(
191                document=result.document,
192                score=result.score,  # Keep original relevance score
193                rank=rank
194            ))
195        
196        return mmr_results

 16class ParentDocumentRetriever(BaseRetriever):
 17    """
 18    Retriever that searches child chunks but returns parent documents.
 19    
 20    This pattern is useful when:
 21    - Documents are chunked into small pieces for precise embedding
 22    - But you want to return full parent documents for better context
 23    - Multiple chunks from the same parent might match the query
 24    
 25    Architecture:
 26    - Child store: Contains small chunks with embeddings (searchable)
 27    - Parent store: Contains full parent documents (for retrieval)
 28    - Mapping: Children have metadata["parent_id"] pointing to parent
 29    
 30    Workflow:
 31    1. Search child store for relevant chunks
 32    2. Extract parent_ids from child metadata
 33    3. Retrieve parent documents from parent store
 34    4. Deduplicate and rank by best child score
 35    
 36    Example:
 37        ```python
 38        from gmf_forge_ai_data.retrieval import ParentDocumentRetriever
 39        from gmf_forge_ai_data.vector_stores import InMemoryVectorStore, Document
 40        from gmf_forge_ai_data.embeddings import AzureOpenAIEmbeddings
 41        
 42        embedder = AzureOpenAIEmbeddings(...)
 43        
 44        # Setup stores
 45        child_store = InMemoryVectorStore()
 46        parent_store = InMemoryVectorStore()
 47        
 48        # Create parent document
 49        parent = Document(
 50            id="doc_1",
 51            content="Full document with multiple sections...",
 52            embedding=embedder.embed_text("Full document...")
 53        )
 54        parent_store.add_documents([parent])
 55        
 56        # Create child chunks
 57        child1 = Document(
 58            id="doc_1_chunk_0",
 59            content="Introduction section...",
 60            embedding=embedder.embed_text("Introduction..."),
 61            metadata={"parent_id": "doc_1"}  # Link to parent
 62        )
 63        child2 = Document(
 64            id="doc_1_chunk_1",
 65            content="Methods section...",
 66            embedding=embedder.embed_text("Methods..."),
 67            metadata={"parent_id": "doc_1"}
 68        )
 69        child_store.add_documents([child1, child2])
 70        
 71        # Setup retriever
 72        retriever = ParentDocumentRetriever(
 73            child_store=child_store,
 74            parent_store=parent_store,
 75            parent_id_key="parent_id"
 76        )
 77        
 78        # Search chunks, return parent
 79        query_embedding = embedder.embed_text("introduction")
 80        results = retriever.retrieve_embedding(
 81            embedding=query_embedding,
 82            top_k=5
 83        )
 84        # Returns full parent documents, not chunks
 85        ```
 86    
 87    Benefits:
 88    - Precise search (small chunks)
 89    - Rich context (full parents)
 90    - Automatic deduplication
 91    """
 92    
 93    def __init__(
 94        self,
 95        child_store: BaseVectorStore,
 96        parent_store: BaseVectorStore,
 97        parent_id_key: str = "parent_id",
 98        search_type: str = "vector"
 99    ):
100        """
101        Initialize parent document retriever.
102        
103        Args:
104            child_store: Vector store containing child chunks with embeddings
105            parent_store: Vector store containing parent documents
106            parent_id_key: Metadata key in children pointing to parent ID
107            search_type: Search type for child store ("vector", "keyword", "hybrid")
108        
109        Raises:
110            ValueError: If search_type is invalid
111        """
112        if search_type not in ["vector", "keyword", "hybrid"]:
113            raise ValueError(f"Invalid search_type: {search_type}")
114        
115        self.child_store = child_store
116        self.parent_store = parent_store
117        self.parent_id_key = parent_id_key
118        self.search_type = search_type
119    
120    def retrieve(self, query: RetrievalQuery) -> List[SearchResult]:
121        """
122        Retrieve parent documents by searching child chunks.
123        
124        Args:
125            query: RetrievalQuery with appropriate parameters for search_type
126        
127        Returns:
128            List of SearchResult with parent documents, deduplicated and ranked
129        
130        Raises:
131            ValueError: If required query parameters are missing
132        """
133        # Validate query based on search type
134        if self.search_type in ["vector", "hybrid"] and query.embedding is None:
135            raise ValueError(f"{self.search_type} search requires query.embedding")
136        if self.search_type in ["keyword", "hybrid"] and query.text is None:
137            raise ValueError(f"{self.search_type} search requires query.text")
138        
139        # Search child store (fetch more candidates to account for deduplication)
140        fetch_k = query.top_k * 3  # Heuristic: fetch 3x to handle duplicates
141        child_results = self.child_store.search(
142            query=query.text,
143            query_embedding=query.embedding,
144            top_k=fetch_k,
145            filters=query.filters,
146            search_type=self.search_type
147        )
148        
149        if len(child_results) == 0:
150            return []
151        
152        # Extract parent IDs and track best scores
153        parent_scores: Dict[str, float] = {}  # parent_id -> best_score
154        parent_order: OrderedDict[str, None] = OrderedDict()  # Track first occurrence
155        
156        for child_result in child_results:
157            parent_id = child_result.document.metadata.get(self.parent_id_key)
158            
159            if parent_id is None:
160                # Skip children without parent link
161                continue
162            
163            # Track best score for each parent (in case multiple children match)
164            if parent_id not in parent_scores:
165                parent_scores[parent_id] = child_result.score
166                parent_order[parent_id] = None  # Track insertion order
167            else:
168                # Update if this child has better score
169                parent_scores[parent_id] = max(
170                    parent_scores[parent_id],
171                    child_result.score
172                )
173        
174        if len(parent_scores) == 0:
175            return []
176        
177        # Retrieve parent documents
178        parent_results = []
179        for rank, parent_id in enumerate(parent_order.keys()):
180            if rank >= query.top_k:
181                break
182            
183            parent_doc = self.parent_store.get_document(parent_id)
184            if parent_doc is None:
185                # Parent not found in store, skip
186                continue
187            
188            parent_results.append(SearchResult(
189                document=parent_doc,
190                score=parent_scores[parent_id],  # Best child score
191                rank=rank
192            ))
193        
194        return parent_results

 17class EnsembleRetriever(BaseRetriever):
 18    """
 19    Ensemble retriever combining multiple retrieval strategies.
 20    
 21    Combines results from multiple retrievers using score fusion techniques:
 22    - Reciprocal Rank Fusion (RRF): Robust, rank-based fusion
 23    - Weighted Average: Score-based fusion with configurable weights
 24    - Max Score: Conservative, consensus-based fusion
 25    
 26    Benefits:
 27    - Improved recall (combine different strategies)
 28    - Robustness (less sensitive to individual retriever failures)
 29    - Flexibility (combine vector, keyword, hybrid, MMR, etc.)
 30    
 31    Example:
 32        ```python
 33        from gmf_forge_ai_data.retrieval import (
 34            EnsembleRetriever,
 35            VectorRetriever,
 36            KeywordRetriever,
 37            MMRRetriever
 38        )
 39        from gmf_forge_ai_data.embeddings import AzureOpenAIEmbeddings
 40        
 41        embedder = AzureOpenAIEmbeddings(...)
 42        
 43        # Create multiple retrievers
 44        vector_retriever = VectorRetriever(vector_store)
 45        keyword_retriever = KeywordRetriever(vector_store)
 46        mmr_retriever = MMRRetriever(vector_store, lambda_param=0.7)
 47        
 48        # Combine with RRF fusion
 49        ensemble = EnsembleRetriever(
 50            retrievers=[vector_retriever, keyword_retriever, mmr_retriever],
 51            weights=[0.5, 0.3, 0.2],  # Optional weights
 52            fusion_strategy="rrf"  # Reciprocal Rank Fusion
 53        )
 54        
 55        # Query requires parameters for all retrievers
 56        query_text = "machine learning"
 57        query_embedding = embedder.embed_text(query_text)
 58        
 59        query = RetrievalQuery(
 60            text=query_text,  # For keyword retriever
 61            embedding=query_embedding,  # For vector/MMR retrievers
 62            top_k=5
 63        )
 64        
 65        results = ensemble.retrieve(query)
 66        # Returns fused results from all retrievers
 67        ```
 68    
 69    Fusion Strategies:
 70    
 71    1. **rrf** (Reciprocal Rank Fusion):
 72       - score(doc) = Σ weights[i] / (k + rank_i)
 73       - k = 60 (default)
 74       - Robust to different score scales
 75       - Rank-based, not score-based
 76    
 77    2. **weighted_avg** (Weighted Average):
 78       - Normalize scores to [0, 1]
 79       - score(doc) = Σ weights[i] * normalized_score_i
 80       - Score-based fusion
 81    
 82    3. **max_score** (Maximum Score):
 83       - score(doc) = max(normalized_scores)
 84       - Conservative, requires consensus
 85    
 86    References:
 87        Cormack, G. V., Clarke, C. L., & Buettcher, S. (2009).
 88        Reciprocal rank fusion outperforms condorcet and individual
 89        ranklists. ACM SIGIR.
 90    """
 91    
 92    def __init__(
 93        self,
 94        retrievers: List[BaseRetriever],
 95        weights: List[float] = None,
 96        fusion_strategy: str = "rrf",
 97        rrf_k: int = 60
 98    ):
 99        """
100        Initialize ensemble retriever.
101        
102        Args:
103            retrievers: List of retrievers to combine
104            weights: Optional weights for each retriever (default: equal weights)
105            fusion_strategy: Fusion method - "rrf", "weighted_avg", or "max_score"
106            rrf_k: k parameter for RRF (default: 60)
107        
108        Raises:
109            ValueError: If retrievers list is empty or params are invalid
110        """
111        if not retrievers:
112            raise ValueError("Must provide at least one retriever")
113        
114        if fusion_strategy not in ["rrf", "weighted_avg", "max_score"]:
115            raise ValueError(
116                f"Invalid fusion_strategy: {fusion_strategy}. "
117                "Must be 'rrf', 'weighted_avg', or 'max_score'"
118            )
119        
120        if weights is None:
121            # Equal weights
122            weights = [1.0 / len(retrievers)] * len(retrievers)
123        else:
124            if len(weights) != len(retrievers):
125                raise ValueError(
126                    f"Number of weights ({len(weights)}) must match "
127                    f"number of retrievers ({len(retrievers)})"
128                )
129            # Normalize weights to sum to 1.0
130            total = sum(weights)
131            weights = [w / total for w in weights]
132        
133        self.retrievers = retrievers
134        self.weights = weights
135        self.fusion_strategy = fusion_strategy
136        self.rrf_k = rrf_k
137    
138    def retrieve(self, query: RetrievalQuery) -> List[SearchResult]:
139        """
140        Retrieve documents using ensemble fusion.
141        
142        Args:
143            query: RetrievalQuery with parameters for all retrievers
144        
145        Returns:
146            List of SearchResult with fused scores and ranks
147        
148        Note:
149            All retrievers are called with the same query. Ensure the query
150            contains all required parameters (text, embedding) for your retrievers.
151        """
152        # Retrieve from all retrievers
153        all_results: List[List[SearchResult]] = []
154        
155        for retriever in self.retrievers:
156            try:
157                results = retriever.retrieve(query)
158                all_results.append(results)
159            except ValueError as e:
160                # Retriever might not support this query type, skip it
161                all_results.append([])
162        
163        if not any(all_results):
164            return []
165        
166        # Apply fusion strategy
167        if self.fusion_strategy == "rrf":
168            fused_results = self._reciprocal_rank_fusion(all_results)
169        elif self.fusion_strategy == "weighted_avg":
170            fused_results = self._weighted_average_fusion(all_results)
171        else:  # max_score
172            fused_results = self._max_score_fusion(all_results)
173        
174        # Sort by fused score and take top_k
175        fused_results.sort(key=lambda x: x.score, reverse=True)
176        top_results = fused_results[:query.top_k]
177        
178        # Update ranks
179        for rank, result in enumerate(top_results):
180            result.rank = rank
181        
182        return top_results
183    
184    def _reciprocal_rank_fusion(
185        self,
186        all_results: List[List[SearchResult]]
187    ) -> List[SearchResult]:
188        """
189        Fuse results using Reciprocal Rank Fusion (RRF).
190        
191        RRF score for document d:
192            score(d) = Σ weight_i / (k + rank_i(d))
193        
194        Where rank_i(d) is the rank of d in retriever i's results.
195        """
196        doc_scores: Dict[str, float] = defaultdict(float)
197        doc_objects: Dict[str, Document] = {}
198        
199        for retriever_idx, results in enumerate(all_results):
200            weight = self.weights[retriever_idx]
201            
202            for result in results:
203                doc_id = result.document.id
204                
205                # RRF score contribution
206                rrf_score = weight / (self.rrf_k + result.rank)
207                doc_scores[doc_id] += rrf_score
208                
209                # Keep document object (use first occurrence)
210                if doc_id not in doc_objects:
211                    doc_objects[doc_id] = result.document
212        
213        # Build SearchResult objects
214        fused_results = [
215            SearchResult(
216                document=doc_objects[doc_id],
217                score=score,
218                rank=0  # Will be set later
219            )
220            for doc_id, score in doc_scores.items()
221        ]
222        
223        return fused_results
224    
225    def _weighted_average_fusion(
226        self,
227        all_results: List[List[SearchResult]]
228    ) -> List[SearchResult]:
229        """
230        Fuse results using weighted average of normalized scores.
231        
232        Scores are normalized to [0, 1] within each retriever, then averaged.
233        """
234        doc_scores: Dict[str, float] = defaultdict(float)
235        doc_objects: Dict[str, Document] = {}
236        
237        for retriever_idx, results in enumerate(all_results):
238            if not results:
239                continue
240            
241            weight = self.weights[retriever_idx]
242            
243            # Normalize scores to [0, 1]
244            scores = np.array([r.score for r in results])
245            min_score = scores.min()
246            max_score = scores.max()
247            
248            if max_score > min_score:
249                normalized_scores = (scores - min_score) / (max_score - min_score)
250            else:
251                normalized_scores = np.ones_like(scores)
252            
253            for result, norm_score in zip(results, normalized_scores):
254                doc_id = result.document.id
255                doc_scores[doc_id] += weight * norm_score
256                
257                if doc_id not in doc_objects:
258                    doc_objects[doc_id] = result.document
259        
260        # Build SearchResult objects
261        fused_results = [
262            SearchResult(
263                document=doc_objects[doc_id],
264                score=score,
265                rank=0
266            )
267            for doc_id, score in doc_scores.items()
268        ]
269        
270        return fused_results
271    
272    def _max_score_fusion(
273        self,
274        all_results: List[List[SearchResult]]
275    ) -> List[SearchResult]:
276        """
277        Fuse results using maximum normalized score across retrievers.
278        
279        Conservative strategy: document needs high score from at least one retriever.
280        """
281        doc_scores: Dict[str, float] = defaultdict(float)
282        doc_objects: Dict[str, Document] = {}
283        
284        for retriever_idx, results in enumerate(all_results):
285            if not results:
286                continue
287            
288            # Normalize scores to [0, 1]
289            scores = np.array([r.score for r in results])
290            min_score = scores.min()
291            max_score = scores.max()
292            
293            if max_score > min_score:
294                normalized_scores = (scores - min_score) / (max_score - min_score)
295            else:
296                normalized_scores = np.ones_like(scores)
297            
298            for result, norm_score in zip(results, normalized_scores):
299                doc_id = result.document.id
300                
301                # Take maximum normalized score
302                doc_scores[doc_id] = max(doc_scores[doc_id], norm_score)
303                
304                if doc_id not in doc_objects:
305                    doc_objects[doc_id] = result.document
306        
307        # Build SearchResult objects
308        fused_results = [
309            SearchResult(
310                document=doc_objects[doc_id],
311                score=score,
312                rank=0
313            )
314            for doc_id, score in doc_scores.items()
315        ]
316        
317        return fused_results