Coverage for src / qdrant_loader_mcp_server / search / components / result_combiner.py: 89%
170 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-18 04:51 +0000
1"""Result combination and ranking logic for hybrid search."""
3from typing import Any
5from ...utils.logging import LoggingConfig
6from ..hybrid.components.scoring import HybridScorer
7from ..nlp.spacy_analyzer import SpaCyQueryAnalyzer
8from .combining import (
9 boost_score_with_metadata,
10 flatten_metadata_components,
11 should_skip_result,
12)
13from .metadata_extractor import MetadataExtractor
14from .search_result_models import HybridSearchResult, create_hybrid_search_result
16WRRF_CONSTANT = 60
19class ResultCombiner:
20 """Combines and ranks search results from multiple sources."""
22 def __init__(
23 self,
24 vector_weight: float = 0.6,
25 keyword_weight: float = 0.3,
26 metadata_weight: float = 0.1,
27 min_score: float = 0.3,
28 spacy_analyzer: SpaCyQueryAnalyzer | None = None,
29 ):
30 """Initialize the result combiner.
32 Args:
33 vector_weight: Weight for vector search scores (0-1)
34 keyword_weight: Weight for keyword search scores (0-1)
35 metadata_weight: Weight for metadata-based scoring (0-1)
36 min_score: Minimum combined score threshold
37 spacy_analyzer: Optional spaCy analyzer for semantic boosting
38 """
39 self.vector_weight = vector_weight
40 self.keyword_weight = keyword_weight
41 self.metadata_weight = metadata_weight
42 self.min_score = min_score
43 self.spacy_analyzer = spacy_analyzer
44 self.logger = LoggingConfig.get_logger(__name__)
46 self.metadata_extractor = MetadataExtractor()
47 # Internal scorer to centralize weighting logic (behavior-preserving)
48 self._scorer = HybridScorer(
49 vector_weight=self.vector_weight,
50 keyword_weight=self.keyword_weight,
51 metadata_weight=self.metadata_weight,
52 )
54 def merge_results_with_wrrf(
55 self,
56 vector_results: list[dict[str, Any]],
57 keyword_results: list[dict[str, Any]],
58 ) -> dict:
59 """
60 Merge and rerank results using Weighted Recipocal Rerank Fusion from vector (dense) and keyword (sparse) search.
61 """
62 combined_dict = {}
63 # Process vector results
64 for rank, result in enumerate(vector_results, 1):
65 text = result["text"]
66 if text not in combined_dict:
67 metadata = result["metadata"]
68 combined_dict[text] = {
69 "text": text,
70 "metadata": metadata,
71 "source_type": result["source_type"],
72 "vector_score": result["score"],
73 "keyword_score": 0.0,
74 # 🔧 CRITICAL FIX: Include all root-level fields from search services
75 "title": result.get("title", ""),
76 "url": result.get("url", ""),
77 "document_id": result.get("document_id", ""),
78 "source": result.get("source", ""),
79 "created_at": result.get("created_at", ""),
80 "updated_at": result.get("updated_at", ""),
81 "wrrf_score": self._scorer.vector_weight
82 * (1 / (rank + WRRF_CONSTANT)),
83 }
85 # Process keyword results
86 for rank, result in enumerate(keyword_results, 1):
87 text = result["text"]
88 if text in combined_dict:
89 combined_dict[text]["keyword_score"] = result["score"]
90 # Sum
91 combined_dict[text]["wrrf_score"] += self._scorer.keyword_weight * (
92 1 / (rank + WRRF_CONSTANT)
93 )
94 else:
95 metadata = result["metadata"]
96 combined_dict[text] = {
97 "text": text,
98 "metadata": metadata,
99 "source_type": result["source_type"],
100 "vector_score": 0.0,
101 "keyword_score": result["score"],
102 "title": result.get("title", ""),
103 "url": result.get("url", ""),
104 "document_id": result.get("document_id", ""),
105 "source": result.get("source", ""),
106 "created_at": result.get("created_at", ""),
107 "updated_at": result.get("updated_at", ""),
108 "wrrf_score": self._scorer.keyword_weight
109 * (1 / (rank + WRRF_CONSTANT)),
110 }
111 return combined_dict
113 def extract_chunk_title(
114 self, info: dict, metadata: dict, chunk_index: int, total_chunks: int
115 ) -> str:
116 # Extract fields from both direct payload fields and nested metadata
117 # Use direct fields from Qdrant payload when available, fallback to metadata
118 title = info.get("title", "") or metadata.get("title", "")
120 # Extract rich metadata from nested metadata object
121 file_name = metadata.get("file_name", "")
122 metadata.get("file_type", "")
124 # Enhanced title generation using actual Qdrant structure
125 # Priority: root title > nested section_title > file_name + chunk info > source
126 root_title = info.get(
127 "title", ""
128 ) # e.g., "Stratégie commerciale MYA.pdf - Chunk 2"
129 nested_title = metadata.get("title", "") # e.g., "Preamble (Part 2)"
130 section_title = metadata.get("section_title", "")
132 if root_title:
133 title = root_title
134 elif nested_title:
135 title = nested_title
136 elif section_title:
137 title = section_title
138 elif file_name:
139 title = file_name
140 # Add chunk info if available from nested metadata
141 sub_chunk_index = metadata.get("sub_chunk_index")
142 total_sub_chunks = metadata.get("total_sub_chunks")
143 if sub_chunk_index is not None and total_sub_chunks is not None:
144 title += f" - Chunk {int(sub_chunk_index) + 1}/{total_sub_chunks}"
145 elif chunk_index is not None and total_chunks is not None:
146 title += f" - Chunk {int(chunk_index) + 1}/{total_chunks}"
147 else:
148 source = info.get("source", "") or metadata.get("source", "")
149 if source:
150 # Extract filename from path-like sources
151 import os
153 title = (
154 os.path.basename(source)
155 if "/" in source or "\\" in source
156 else source
157 )
158 else:
159 title = "Untitled"
160 return title
162 def merge_rich_and_enhanced_metadata(
163 self,
164 info: dict,
165 metadata: dict,
166 metadata_components: dict,
167 chunk_index: int,
168 total_chunks: int,
169 ) -> dict:
170 # Create enhanced metadata dict with rich Qdrant fields
171 enhanced_metadata = {
172 # Core fields from root level of Qdrant payload
173 "source_url": info.get("url", ""),
174 "document_id": info.get("document_id", ""),
175 "created_at": info.get("created_at", ""),
176 "last_modified": info.get("updated_at", ""),
177 "repo_name": info.get("source", ""),
178 # Project scoping is stored at the root as 'source'
179 "project_id": info.get("source", ""),
180 # Construct file path from nested metadata
181 "file_path": (
182 metadata.get("file_directory", "").rstrip("/")
183 + "/"
184 + metadata.get("file_name", "")
185 if metadata.get("file_name") and metadata.get("file_directory")
186 else metadata.get("file_name", "")
187 ),
188 }
190 # Add rich metadata from nested metadata object (confirmed structure)
191 rich_metadata_fields = {
192 "original_filename": metadata.get("file_name"),
193 "file_size": metadata.get("file_size"),
194 "original_file_type": metadata.get("file_type")
195 or metadata.get("original_file_type"),
196 "word_count": metadata.get("word_count"),
197 "char_count": metadata.get("character_count")
198 or metadata.get("char_count")
199 or metadata.get("line_count"),
200 "chunk_index": metadata.get("sub_chunk_index", chunk_index),
201 "total_chunks": metadata.get("total_sub_chunks", total_chunks),
202 "chunking_strategy": metadata.get("chunking_strategy")
203 or metadata.get("conversion_method"),
204 # Project fields now come from root payload; avoid overriding with nested metadata
205 "collection_name": metadata.get("collection_name"),
206 # Additional rich fields from actual Qdrant structure
207 "section_title": metadata.get("section_title"),
208 "parent_section": metadata.get("parent_section"),
209 "file_encoding": metadata.get("file_encoding"),
210 "conversion_failed": metadata.get("conversion_failed", False),
211 "is_excel_sheet": metadata.get("is_excel_sheet", False),
212 }
214 # Only add non-None values to avoid conflicts
215 for key, value in rich_metadata_fields.items():
216 if value is not None:
217 enhanced_metadata[key] = value
219 # Merge with flattened metadata components (flattened takes precedence for conflicts)
220 flattened_components = flatten_metadata_components(metadata_components)
221 enhanced_metadata.update(flattened_components)
223 return enhanced_metadata
225 def is_result_filtered(self, use_wrrf: bool, wrrf_score: float, chunk_score: float):
226 # Scale minimum threshold
227 wrrf_min_score = self.min_score * (
228 (self._scorer.vector_weight + self._scorer.keyword_weight)
229 / (WRRF_CONSTANT + 1)
230 )
231 # Filter low wrrf
232 if use_wrrf and wrrf_score <= wrrf_min_score:
233 return True
235 # Fallback to standard filter
236 if not use_wrrf and chunk_score <= self.min_score:
237 return True
238 return False
240 async def combine_results(
241 self,
242 vector_results: list[dict[str, Any]],
243 keyword_results: list[dict[str, Any]],
244 query_context: dict[str, Any],
245 limit: int,
246 source_types: list[str] | None = None,
247 project_ids: list[str] | None = None,
248 ) -> list[HybridSearchResult]:
249 """Combine and rerank results using Weighted Recipocal Rerank Fusion from vector (dense) and keyword (sparse) search.
251 Args:
252 vector_results: Results from vector search
253 keyword_results: Results from keyword search
254 query_context: Query analysis context
255 limit: Maximum number of results to return
256 source_types: Optional source type filters
257 project_ids: Optional project ID filters
259 Returns:
260 List of combined and ranked HybridSearchResult objects
261 """
262 combined_dict = self.merge_results_with_wrrf(
263 vector_results=vector_results, keyword_results=keyword_results
264 )
266 # Calculate combined scores and create results
267 combined_results = []
269 # Extract intent-specific filtering configuration
270 search_intent = query_context.get("search_intent")
271 adaptive_config = query_context.get("adaptive_config")
272 result_filters = adaptive_config.result_filters if adaptive_config else {}
274 # Naive WRRF trigger
275 use_wrrf = len(combined_dict.keys()) >= 10
277 for text, info in combined_dict.items():
278 # Skip if source type doesn't match filter
279 if source_types and info["source_type"] not in source_types:
280 continue
281 # Apply intent-specific result filtering
282 metadata = info["metadata"]
283 if search_intent and result_filters:
284 if should_skip_result(metadata, result_filters, query_context):
285 continue
287 wrrf_score = info["wrrf_score"]
288 # Fallback to standard weighting scoring
289 chunk_score = (info["keyword_score"] * self._scorer.keyword_weight) + (
290 info["vector_score"] * self._scorer.vector_weight
291 )
293 # Filter based on WRRF or standard scores and weighting
294 if self.is_result_filtered(use_wrrf, wrrf_score, chunk_score):
295 continue
297 score = wrrf_score if use_wrrf else chunk_score
299 # Extract all metadata components
300 metadata_components = self.metadata_extractor.extract_all_metadata(metadata)
302 # TODO: Evaluate metadata score boosting with WRRF and in general - Boost score with metadata
303 boosted_score = boost_score_with_metadata(
304 score,
305 metadata,
306 query_context,
307 spacy_analyzer=self.spacy_analyzer,
308 )
309 chunk_index = metadata.get("chunk_index")
310 total_chunks = metadata.get("total_chunks")
312 title = self.extract_chunk_title(
313 info=info,
314 metadata=metadata,
315 chunk_index=chunk_index,
316 total_chunks=total_chunks,
317 )
318 enhanced_metadata = self.merge_rich_and_enhanced_metadata(
319 info=info,
320 metadata=metadata,
321 metadata_components=metadata_components,
322 chunk_index=chunk_index,
323 total_chunks=total_chunks,
324 )
326 # NOTE: No additional fallback; root payload project_id is authoritative
328 # Create HybridSearchResult using factory function
329 hybrid_result = create_hybrid_search_result(
330 score=boosted_score,
331 text=text,
332 source_type=info["source_type"],
333 source_title=title,
334 vector_score=info["vector_score"],
335 keyword_score=info["keyword_score"],
336 **enhanced_metadata,
337 )
339 combined_results.append(hybrid_result)
341 # Sort by combined score
342 combined_results.sort(key=lambda x: x.score, reverse=True)
343 # Apply diversity filtering for exploratory intents
344 if adaptive_config and adaptive_config.diversity_factor > 0.0:
345 try:
346 from ..hybrid.components.diversity import apply_diversity_filtering
348 diverse_results = apply_diversity_filtering(
349 combined_results, adaptive_config.diversity_factor, limit
350 )
351 self.logger.debug(
352 "Applied diversity filtering",
353 original_count=len(combined_results),
354 diverse_count=len(diverse_results),
355 diversity_factor=adaptive_config.diversity_factor,
356 )
357 return diverse_results
358 except Exception:
359 # Fallback to original top-N behavior if import or filtering fails
360 pass
362 return combined_results[:limit]
364 # The following methods are thin wrappers delegating to combining/* modules
365 # to preserve backward-compatible tests that call private methods directly.
367 def _should_skip_result(
368 self, metadata: dict, result_filters: dict, query_context: dict
369 ) -> bool:
370 return should_skip_result(metadata, result_filters, query_context)
372 def _count_business_indicators(self, metadata: dict) -> int:
373 return __import__(
374 f"{__package__}.combining.filters", fromlist=["count_business_indicators"]
375 ).count_business_indicators(metadata)
377 def _boost_score_with_metadata(
378 self, base_score: float, metadata: dict, query_context: dict
379 ) -> float:
380 return boost_score_with_metadata(
381 base_score, metadata, query_context, spacy_analyzer=self.spacy_analyzer
382 )
384 def _apply_content_type_boosting(
385 self, metadata: dict, query_context: dict
386 ) -> float:
387 from .combining import apply_content_type_boosting
389 return apply_content_type_boosting(metadata, query_context)
391 def _apply_section_level_boosting(self, metadata: dict) -> float:
392 from .combining import apply_section_level_boosting
394 return apply_section_level_boosting(metadata)
396 def _apply_content_quality_boosting(self, metadata: dict) -> float:
397 from .combining import apply_content_quality_boosting
399 return apply_content_quality_boosting(metadata)
401 def _apply_conversion_boosting(self, metadata: dict, query_context: dict) -> float:
402 from .combining import apply_conversion_boosting
404 return apply_conversion_boosting(metadata, query_context)
406 def _apply_semantic_boosting(self, metadata: dict, query_context: dict) -> float:
407 from .combining import apply_semantic_boosting
409 return apply_semantic_boosting(metadata, query_context, self.spacy_analyzer)
411 def _apply_fallback_semantic_boosting(
412 self, metadata: dict, query_context: dict
413 ) -> float:
414 from .combining import apply_fallback_semantic_boosting
416 return apply_fallback_semantic_boosting(metadata, query_context)
418 def _apply_diversity_filtering(
419 self, results: list[HybridSearchResult], diversity_factor: float, limit: int
420 ) -> list[HybridSearchResult]:
421 if diversity_factor <= 0.0 or len(results) <= limit:
422 return results[:limit]
424 diverse_results = []
425 used_source_types = set()
426 used_section_types = set()
427 used_sources = set()
429 for result in results:
430 if len(diverse_results) >= limit:
431 break
433 diversity_score = 1.0
434 source_type = result.source_type
435 if source_type in used_source_types:
436 diversity_score *= 1.0 - diversity_factor * 0.3
438 section_type = result.section_type or "unknown"
439 if section_type in used_section_types:
440 diversity_score *= 1.0 - diversity_factor * 0.2
442 source_key = f"{result.source_type}:{result.source_title}"
443 if source_key in used_sources:
444 diversity_score *= 1.0 - diversity_factor * 0.4
446 adjusted_score = result.score * diversity_score
448 if (
449 len(diverse_results) < limit * 0.7
450 or adjusted_score >= result.score * 0.6
451 ):
452 diverse_results.append(result)
453 used_source_types.add(source_type)
454 used_section_types.add(section_type)
455 used_sources.add(source_key)
457 remaining_slots = limit - len(diverse_results)
458 if remaining_slots > 0:
459 remaining_results = [r for r in results if r not in diverse_results]
460 diverse_results.extend(remaining_results[:remaining_slots])
462 return diverse_results[:limit]
464 def _flatten_metadata_components(
465 self, metadata_components: dict[str, Any]
466 ) -> dict[str, Any]:
467 return flatten_metadata_components(metadata_components)