Coverage for src/local_deep_research/advanced_search_system/candidate_exploration/adaptive_explorer.py: 100%
131 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Adaptive candidate explorer implementation.
4This explorer adapts its search strategy based on the success of different
5approaches and the quality of candidates found.
6"""
8import time
9from collections import defaultdict
10from typing import List, Optional
12from loguru import logger
14from ..candidates.base_candidate import Candidate
15from ..constraints.base_constraint import Constraint
16from ...utilities.json_utils import get_llm_response_text
17from .base_explorer import (
18 BaseCandidateExplorer,
19 ExplorationResult,
20 ExplorationStrategy,
21)
22from ...utilities.llm_utils import invoke_llm_sync
25class AdaptiveExplorer(BaseCandidateExplorer):
26 """
27 Adaptive candidate explorer that learns from search results.
29 This explorer:
30 1. Tries different search strategies
31 2. Tracks which strategies work best
32 3. Adapts future searches based on success rates
33 4. Focuses effort on the most productive approaches
34 """
36 def __init__(
37 self,
38 *args,
39 initial_strategies: List[str] = None,
40 adaptation_threshold: int = 5, # Adapt after this many searches
41 **kwargs,
42 ):
43 """
44 Initialize adaptive explorer.
46 Args:
47 initial_strategies: Starting search strategies to try
48 adaptation_threshold: Number of searches before adapting
49 """
50 super().__init__(*args, **kwargs)
52 self.initial_strategies = initial_strategies or [
53 "direct_search",
54 "synonym_expansion",
55 "category_exploration",
56 "related_terms",
57 ]
59 self.adaptation_threshold = adaptation_threshold
61 # Track strategy performance
62 self.strategy_stats = defaultdict(
63 lambda: {"attempts": 0, "candidates_found": 0, "quality_sum": 0.0}
64 )
65 self.current_strategy = self.initial_strategies[0]
67 def explore(
68 self,
69 initial_query: str,
70 constraints: Optional[List[Constraint]] = None,
71 entity_type: Optional[str] = None,
72 ) -> ExplorationResult:
73 """Explore candidates using adaptive strategy."""
74 start_time = time.time()
75 logger.info(f"Starting adaptive exploration for: {initial_query}")
77 all_candidates = []
78 exploration_paths = []
79 total_searched = 0
81 # Track current strategy performance
82 search_count = 0
84 while self._should_continue_exploration(
85 start_time, len(all_candidates)
86 ):
87 # Choose strategy based on current performance
88 strategy = self._choose_strategy(search_count)
90 # Generate query using chosen strategy
91 query = self._generate_query_with_strategy(
92 initial_query, strategy, all_candidates, constraints
93 )
95 if not query or query.lower() in self.explored_queries:
96 # Try next strategy or stop
97 if not self._try_next_strategy():
98 break
99 continue
101 # Execute search
102 logger.info(
103 f"Using strategy '{strategy}' for query: {query[:50]}..."
104 )
105 results = self._execute_search(query)
106 candidates = self._extract_candidates_from_results(
107 results, entity_type
108 )
110 # Track strategy performance
111 self._update_strategy_stats(strategy, candidates)
113 # Add results
114 all_candidates.extend(candidates)
115 total_searched += 1
116 search_count += 1
118 exploration_paths.append(
119 f"{strategy}: {query} -> {len(candidates)} candidates"
120 )
122 # Adapt strategy if threshold reached
123 if search_count >= self.adaptation_threshold:
124 self._adapt_strategy()
125 search_count = 0
127 # Process final results
128 unique_candidates = self._deduplicate_candidates(all_candidates)
129 ranked_candidates = self._rank_candidates_by_relevance(
130 unique_candidates, initial_query
131 )
132 final_candidates = ranked_candidates[: self.max_candidates]
134 elapsed_time = time.time() - start_time
135 logger.info(
136 f"Adaptive exploration completed: {len(final_candidates)} candidates in {elapsed_time:.1f}s"
137 )
139 return ExplorationResult(
140 candidates=final_candidates,
141 total_searched=total_searched,
142 unique_candidates=len(unique_candidates),
143 exploration_paths=exploration_paths,
144 metadata={
145 "strategy": "adaptive",
146 "strategy_stats": dict(self.strategy_stats),
147 "final_strategy": self.current_strategy,
148 "entity_type": entity_type,
149 },
150 elapsed_time=elapsed_time,
151 strategy_used=ExplorationStrategy.ADAPTIVE,
152 )
154 def generate_exploration_queries(
155 self,
156 base_query: str,
157 found_candidates: List[Candidate],
158 constraints: Optional[List[Constraint]] = None,
159 ) -> List[str]:
160 """Generate queries using adaptive approach."""
161 queries = []
163 # Generate queries using best performing strategies
164 top_strategies = self._get_top_strategies(3)
166 for strategy in top_strategies:
167 query = self._generate_query_with_strategy(
168 base_query, strategy, found_candidates, constraints
169 )
170 if query:
171 queries.append(query)
173 return queries
175 def _choose_strategy(self, search_count: int) -> str:
176 """Choose the best strategy based on current performance."""
177 if search_count < self.adaptation_threshold:
178 # Use current strategy during initial phase
179 return self.current_strategy
181 # Choose best performing strategy
182 best_strategies = self._get_top_strategies(1)
183 return best_strategies[0] if best_strategies else self.current_strategy
185 def _get_top_strategies(self, n: int) -> List[str]:
186 """Get top N performing strategies."""
187 if not self.strategy_stats:
188 return self.initial_strategies[:n]
190 # Sort by candidates found per attempt
191 sorted_strategies = sorted(
192 self.strategy_stats.items(),
193 key=lambda x: x[1]["candidates_found"] / max(x[1]["attempts"], 1),
194 reverse=True,
195 )
197 return [strategy for strategy, _ in sorted_strategies[:n]]
199 def _generate_query_with_strategy(
200 self,
201 base_query: str,
202 strategy: str,
203 found_candidates: List[Candidate],
204 constraints: Optional[List[Constraint]] = None,
205 ) -> Optional[str]:
206 """Generate a query using specific strategy."""
207 try:
208 if strategy == "direct_search":
209 return self._direct_search_query(base_query)
210 if strategy == "synonym_expansion":
211 return self._synonym_expansion_query(base_query)
212 if strategy == "category_exploration":
213 return self._category_exploration_query(
214 base_query, found_candidates
215 )
216 if strategy == "related_terms":
217 return self._related_terms_query(base_query, found_candidates)
218 if strategy == "constraint_focused" and constraints:
219 return self._constraint_focused_query(base_query, constraints)
220 return self._direct_search_query(base_query)
222 except Exception:
223 logger.exception(f"Error generating query with strategy {strategy}")
224 return None
226 def _direct_search_query(self, base_query: str) -> str:
227 """Generate direct search variation."""
228 variations = [
229 f'"{base_query}" examples',
230 f"{base_query} list",
231 f"{base_query} instances",
232 f"types of {base_query}",
233 ]
235 # Choose variation not yet explored
236 for variation in variations:
237 if variation.lower() not in self.explored_queries:
238 return variation
240 return base_query
242 def _synonym_expansion_query(self, base_query: str) -> Optional[str]:
243 """Generate query with synonym expansion."""
244 prompt = f"""
245Generate a search query that means the same as "{base_query}" but uses different words.
246Focus on synonyms and alternative terminology.
248Query:
249"""
251 try:
252 response = get_llm_response_text(
253 invoke_llm_sync(self.model, prompt)
254 ).strip()
255 return response if response != base_query else None
256 except Exception as e:
257 logger.debug(
258 f"Error generating synonym query for '{base_query}': {e}"
259 )
260 return None
262 def _category_exploration_query(
263 self, base_query: str, found_candidates: List[Candidate]
264 ) -> Optional[str]:
265 """Generate query exploring categories of found candidates."""
266 if not found_candidates:
267 return f"categories of {base_query}"
269 sample_names = [c.name for c in found_candidates[:3]]
270 return f"similar to {', '.join(sample_names)}"
272 def _related_terms_query(
273 self, base_query: str, found_candidates: List[Candidate]
274 ) -> Optional[str]:
275 """Generate query using related terms."""
276 prompt = f"""
277Given the search topic "{base_query}", suggest a related search term that would find similar but different examples.
279Related search term:
280"""
282 try:
283 response = get_llm_response_text(
284 invoke_llm_sync(self.model, prompt)
285 ).strip()
286 return response if response != base_query else None
287 except Exception as e:
288 logger.debug(
289 f"Error generating related terms query for '{base_query}': {e}"
290 )
291 return None
293 def _constraint_focused_query(
294 self, base_query: str, constraints: List[Constraint]
295 ) -> Optional[str]:
296 """Generate query focused on a specific constraint."""
297 if not constraints:
298 return None
300 # Pick least explored constraint
301 constraint = constraints[0] # Simple selection
302 return f"{base_query} {constraint.value}"
304 def _update_strategy_stats(
305 self, strategy: str, candidates: List[Candidate]
306 ):
307 """Update performance statistics for a strategy."""
308 self.strategy_stats[strategy]["attempts"] += 1
309 self.strategy_stats[strategy]["candidates_found"] += len(candidates)
311 # Simple quality assessment (could be more sophisticated)
312 quality = len(candidates) * 0.1 # Basic quality based on quantity
313 self.strategy_stats[strategy]["quality_sum"] += quality
315 def _adapt_strategy(self):
316 """Adapt current strategy based on performance."""
317 best_strategies = self._get_top_strategies(1)
318 if best_strategies and best_strategies[0] != self.current_strategy:
319 old_strategy = self.current_strategy
320 self.current_strategy = best_strategies[0]
321 logger.info(
322 f"Adapted strategy from '{old_strategy}' to '{self.current_strategy}'"
323 )
325 def _try_next_strategy(self) -> bool:
326 """Try the next available strategy."""
327 current_index = (
328 self.initial_strategies.index(self.current_strategy)
329 if self.current_strategy in self.initial_strategies
330 else 0
331 )
332 next_index = (current_index + 1) % len(self.initial_strategies)
334 if next_index == 0: # We've tried all strategies
335 return False
337 self.current_strategy = self.initial_strategies[next_index]
338 return True