Coverage for src/local_deep_research/advanced_search_system/filters/cross_engine_filter.py: 98%
94 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Cross-engine search result filter implementation.
3"""
5from typing import Dict, List
7from loguru import logger
9from ...settings.exceptions import SettingsValueError
10from ...utilities.json_utils import extract_json, get_llm_response_text
11from .base_filter import BaseFilter
12from ...utilities.llm_utils import invoke_llm_sync
15class CrossEngineFilter(BaseFilter):
16 """Filter that ranks and filters results from multiple search engines."""
18 def __init__(
19 self,
20 model,
21 max_results=None,
22 default_reorder=True,
23 default_reindex=True,
24 settings_snapshot=None,
25 ):
26 """
27 Initialize the cross-engine filter.
29 Args:
30 model: Language model to use for relevance assessment
31 max_results: Maximum number of results to keep after filtering
32 default_reorder: Default setting for reordering results by relevance
33 default_reindex: Default setting for reindexing results after filtering
34 settings_snapshot: Settings snapshot for thread context
35 """
36 super().__init__(model)
37 # Import from thread_settings to avoid database dependencies
38 from ...config.thread_settings import (
39 get_setting_from_snapshot,
40 NoSettingsContextError,
41 )
43 # Get max_results from database settings if not provided
44 if max_results is None:
45 try:
46 max_results = get_setting_from_snapshot(
47 "search.cross_engine_max_results",
48 default=100,
49 settings_snapshot=settings_snapshot,
50 )
51 if max_results is not None:
52 max_results = int(max_results)
53 else:
54 max_results = 100
55 except SettingsValueError:
56 raise
57 except (NoSettingsContextError, TypeError, ValueError):
58 max_results = 100
59 self.max_results = max_results
61 # Max number of result previews shown to the LLM for relevance ranking.
62 # Higher values let the LLM evaluate more candidates but increase prompt
63 # size and latency.
64 try:
65 self.max_context_items = int(
66 get_setting_from_snapshot(
67 "search.cross_engine_max_context_items",
68 default=30,
69 settings_snapshot=settings_snapshot,
70 )
71 )
72 except SettingsValueError:
73 raise
74 except (NoSettingsContextError, TypeError, ValueError):
75 self.max_context_items = 30
77 self.default_reorder = default_reorder
78 self.default_reindex = default_reindex
80 def _prepare_and_return(self, results, *, reindex, start_index):
81 """Optionally reindex results and return them."""
82 if reindex:
83 for i, result in enumerate(results):
84 result["index"] = str(i + start_index + 1)
85 return results
87 def _valid_unique_indices(self, ranked_indices, upper_bound):
88 """Yield valid indices once, preserving first-seen order."""
89 seen = set()
90 for idx in ranked_indices:
91 if not isinstance(idx, int) or isinstance(idx, bool):
92 logger.warning(
93 f"Skipping non-integer ranked index from cross-engine filter: {idx!r}"
94 )
95 continue
96 if idx in seen:
97 continue
98 if 0 <= idx < upper_bound:
99 seen.add(idx)
100 yield idx
102 def filter_results(
103 self,
104 results: List[Dict],
105 query: str,
106 reorder=None,
107 reindex=None,
108 start_index=0,
109 **kwargs,
110 ) -> List[Dict]:
111 """
112 Filter and rank search results from multiple engines by relevance.
114 Args:
115 results: Combined list of search results from all engines
116 query: The original search query
117 reorder: Whether to reorder results by relevance (default: use instance default)
118 reindex: Whether to update result indices after filtering (default: use instance default)
119 start_index: Starting index for the results (used for continuous indexing)
120 **kwargs: Additional parameters
122 Returns:
123 Filtered list of search results
124 """
125 # Use instance defaults if not specified
126 if reorder is None:
127 reorder = self.default_reorder
128 if reindex is None:
129 reindex = self.default_reindex
131 if not self.model or len(results) <= 10: # Don't filter if few results
132 return self._prepare_and_return(
133 results[: min(self.max_results, len(results))],
134 reindex=reindex,
135 start_index=start_index,
136 )
138 max_context_items = min(self.max_context_items, len(results))
139 context_results = results[:max_context_items]
141 # Create context for LLM
142 preview_context = []
143 for i, result in enumerate(context_results):
144 title = result.get("title", "Untitled").strip()
145 snippet = result.get("snippet", "").strip()
146 engine = result.get("engine", "Unknown engine")
148 # Clean up snippet if too long
149 if len(snippet) > 200:
150 snippet = snippet[:200] + "..."
152 preview_context.append(
153 f"[{i}] Engine: {engine} | Title: {title}\nSnippet: {snippet}"
154 )
156 context = "\n\n".join(preview_context)
158 prompt = f"""You are a search result filter. Your task is to rank search results from multiple engines by relevance to a query.
160Query: "{query}"
162Search Results:
163{context}
165Return the search results as a JSON array of indices, ranked from most to least relevant to the query.
166Only include indices of results that are actually relevant to the query.
167For example: [3, 0, 7, 1]
169If no results seem relevant to the query, return an empty array: []"""
171 try:
172 # Get LLM's evaluation
173 response = invoke_llm_sync(self.model, prompt)
174 response_text = get_llm_response_text(response)
175 ranked_indices = extract_json(response_text, expected_type=list)
177 if ranked_indices is not None:
178 # If not reordering, just filter based on the indices
179 if not reorder:
180 # Just keep the results that were deemed relevant
181 filtered_results = []
182 for idx in sorted(
183 self._valid_unique_indices(
184 ranked_indices, len(context_results)
185 )
186 ): # Sort to maintain original order
187 filtered_results.append(context_results[idx])
189 # Limit results if needed
190 final_results = filtered_results[
191 : min(self.max_results, len(filtered_results))
192 ]
194 if not final_results and results:
195 logger.info(
196 "Cross-engine filtering removed all "
197 "results, returning top 10 originals"
198 )
199 return self._prepare_and_return(
200 context_results[: min(10, len(context_results))],
201 reindex=reindex,
202 start_index=start_index,
203 )
205 logger.info(
206 f"Cross-engine filtering kept {len(final_results)} out of {len(results)} results without reordering"
207 )
208 return self._prepare_and_return(
209 final_results,
210 reindex=reindex,
211 start_index=start_index,
212 )
214 # Create ranked results list (reordering)
215 ranked_results = []
216 for idx in self._valid_unique_indices(
217 ranked_indices, len(context_results)
218 ):
219 ranked_results.append(context_results[idx])
221 # If filtering removed everything, return top results
222 if not ranked_results and results:
223 logger.info(
224 "Cross-engine filtering removed all results, returning top 10 originals instead"
225 )
226 return self._prepare_and_return(
227 context_results[: min(10, len(context_results))],
228 reindex=reindex,
229 start_index=start_index,
230 )
232 # Limit results if needed
233 max_filtered = min(self.max_results, len(ranked_results))
234 final_results = ranked_results[:max_filtered]
236 logger.info(
237 f"Cross-engine filtering kept {len(final_results)} out of {len(results)} results with reordering={reorder}, reindex={reindex}"
238 )
239 return self._prepare_and_return(
240 final_results,
241 reindex=reindex,
242 start_index=start_index,
243 )
244 logger.info(
245 "Could not find JSON array in response, returning original results"
246 )
247 return self._prepare_and_return(
248 context_results[: min(self.max_results, len(context_results))],
249 reindex=reindex,
250 start_index=start_index,
251 )
253 except Exception:
254 logger.exception("Cross-engine filtering error")
255 return self._prepare_and_return(
256 context_results[: min(self.max_results, len(context_results))],
257 reindex=reindex,
258 start_index=start_index,
259 )