Coverage for src/local_deep_research/advanced_search_system/candidate_exploration/adaptive_explorer.py: 100%
130 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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)
24class AdaptiveExplorer(BaseCandidateExplorer):
25 """
26 Adaptive candidate explorer that learns from search results.
28 This explorer:
29 1. Tries different search strategies
30 2. Tracks which strategies work best
31 3. Adapts future searches based on success rates
32 4. Focuses effort on the most productive approaches
33 """
35 def __init__(
36 self,
37 *args,
38 initial_strategies: List[str] = None,
39 adaptation_threshold: int = 5, # Adapt after this many searches
40 **kwargs,
41 ):
42 """
43 Initialize adaptive explorer.
45 Args:
46 initial_strategies: Starting search strategies to try
47 adaptation_threshold: Number of searches before adapting
48 """
49 super().__init__(*args, **kwargs)
51 self.initial_strategies = initial_strategies or [
52 "direct_search",
53 "synonym_expansion",
54 "category_exploration",
55 "related_terms",
56 ]
58 self.adaptation_threshold = adaptation_threshold
60 # Track strategy performance
61 self.strategy_stats = defaultdict(
62 lambda: {"attempts": 0, "candidates_found": 0, "quality_sum": 0.0}
63 )
64 self.current_strategy = self.initial_strategies[0]
66 def explore(
67 self,
68 initial_query: str,
69 constraints: Optional[List[Constraint]] = None,
70 entity_type: Optional[str] = None,
71 ) -> ExplorationResult:
72 """Explore candidates using adaptive strategy."""
73 start_time = time.time()
74 logger.info(f"Starting adaptive exploration for: {initial_query}")
76 all_candidates = []
77 exploration_paths = []
78 total_searched = 0
80 # Track current strategy performance
81 search_count = 0
83 while self._should_continue_exploration(
84 start_time, len(all_candidates)
85 ):
86 # Choose strategy based on current performance
87 strategy = self._choose_strategy(search_count)
89 # Generate query using chosen strategy
90 query = self._generate_query_with_strategy(
91 initial_query, strategy, all_candidates, constraints
92 )
94 if not query or query.lower() in self.explored_queries:
95 # Try next strategy or stop
96 if not self._try_next_strategy():
97 break
98 continue
100 # Execute search
101 logger.info(
102 f"Using strategy '{strategy}' for query: {query[:50]}..."
103 )
104 results = self._execute_search(query)
105 candidates = self._extract_candidates_from_results(
106 results, entity_type
107 )
109 # Track strategy performance
110 self._update_strategy_stats(strategy, candidates)
112 # Add results
113 all_candidates.extend(candidates)
114 total_searched += 1
115 search_count += 1
117 exploration_paths.append(
118 f"{strategy}: {query} -> {len(candidates)} candidates"
119 )
121 # Adapt strategy if threshold reached
122 if search_count >= self.adaptation_threshold:
123 self._adapt_strategy()
124 search_count = 0
126 # Process final results
127 unique_candidates = self._deduplicate_candidates(all_candidates)
128 ranked_candidates = self._rank_candidates_by_relevance(
129 unique_candidates, initial_query
130 )
131 final_candidates = ranked_candidates[: self.max_candidates]
133 elapsed_time = time.time() - start_time
134 logger.info(
135 f"Adaptive exploration completed: {len(final_candidates)} candidates in {elapsed_time:.1f}s"
136 )
138 return ExplorationResult(
139 candidates=final_candidates,
140 total_searched=total_searched,
141 unique_candidates=len(unique_candidates),
142 exploration_paths=exploration_paths,
143 metadata={
144 "strategy": "adaptive",
145 "strategy_stats": dict(self.strategy_stats),
146 "final_strategy": self.current_strategy,
147 "entity_type": entity_type,
148 },
149 elapsed_time=elapsed_time,
150 strategy_used=ExplorationStrategy.ADAPTIVE,
151 )
153 def generate_exploration_queries(
154 self,
155 base_query: str,
156 found_candidates: List[Candidate],
157 constraints: Optional[List[Constraint]] = None,
158 ) -> List[str]:
159 """Generate queries using adaptive approach."""
160 queries = []
162 # Generate queries using best performing strategies
163 top_strategies = self._get_top_strategies(3)
165 for strategy in top_strategies:
166 query = self._generate_query_with_strategy(
167 base_query, strategy, found_candidates, constraints
168 )
169 if query:
170 queries.append(query)
172 return queries
174 def _choose_strategy(self, search_count: int) -> str:
175 """Choose the best strategy based on current performance."""
176 if search_count < self.adaptation_threshold:
177 # Use current strategy during initial phase
178 return self.current_strategy
180 # Choose best performing strategy
181 best_strategies = self._get_top_strategies(1)
182 return best_strategies[0] if best_strategies else self.current_strategy
184 def _get_top_strategies(self, n: int) -> List[str]:
185 """Get top N performing strategies."""
186 if not self.strategy_stats:
187 return self.initial_strategies[:n]
189 # Sort by candidates found per attempt
190 sorted_strategies = sorted(
191 self.strategy_stats.items(),
192 key=lambda x: x[1]["candidates_found"] / max(x[1]["attempts"], 1),
193 reverse=True,
194 )
196 return [strategy for strategy, _ in sorted_strategies[:n]]
198 def _generate_query_with_strategy(
199 self,
200 base_query: str,
201 strategy: str,
202 found_candidates: List[Candidate],
203 constraints: Optional[List[Constraint]] = None,
204 ) -> Optional[str]:
205 """Generate a query using specific strategy."""
206 try:
207 if strategy == "direct_search":
208 return self._direct_search_query(base_query)
209 if strategy == "synonym_expansion":
210 return self._synonym_expansion_query(base_query)
211 if strategy == "category_exploration":
212 return self._category_exploration_query(
213 base_query, found_candidates
214 )
215 if strategy == "related_terms":
216 return self._related_terms_query(base_query, found_candidates)
217 if strategy == "constraint_focused" and constraints:
218 return self._constraint_focused_query(base_query, constraints)
219 return self._direct_search_query(base_query)
221 except Exception:
222 logger.exception(f"Error generating query with strategy {strategy}")
223 return None
225 def _direct_search_query(self, base_query: str) -> str:
226 """Generate direct search variation."""
227 variations = [
228 f'"{base_query}" examples',
229 f"{base_query} list",
230 f"{base_query} instances",
231 f"types of {base_query}",
232 ]
234 # Choose variation not yet explored
235 for variation in variations:
236 if variation.lower() not in self.explored_queries:
237 return variation
239 return base_query
241 def _synonym_expansion_query(self, base_query: str) -> Optional[str]:
242 """Generate query with synonym expansion."""
243 prompt = f"""
244Generate a search query that means the same as "{base_query}" but uses different words.
245Focus on synonyms and alternative terminology.
247Query:
248"""
250 try:
251 response = get_llm_response_text(self.model.invoke(prompt)).strip()
252 return response if response != base_query else None
253 except Exception as e:
254 logger.debug(
255 f"Error generating synonym query for '{base_query}': {e}"
256 )
257 return None
259 def _category_exploration_query(
260 self, base_query: str, found_candidates: List[Candidate]
261 ) -> Optional[str]:
262 """Generate query exploring categories of found candidates."""
263 if not found_candidates:
264 return f"categories of {base_query}"
266 sample_names = [c.name for c in found_candidates[:3]]
267 return f"similar to {', '.join(sample_names)}"
269 def _related_terms_query(
270 self, base_query: str, found_candidates: List[Candidate]
271 ) -> Optional[str]:
272 """Generate query using related terms."""
273 prompt = f"""
274Given the search topic "{base_query}", suggest a related search term that would find similar but different examples.
276Related search term:
277"""
279 try:
280 response = get_llm_response_text(self.model.invoke(prompt)).strip()
281 return response if response != base_query else None
282 except Exception as e:
283 logger.debug(
284 f"Error generating related terms query for '{base_query}': {e}"
285 )
286 return None
288 def _constraint_focused_query(
289 self, base_query: str, constraints: List[Constraint]
290 ) -> Optional[str]:
291 """Generate query focused on a specific constraint."""
292 if not constraints:
293 return None
295 # Pick least explored constraint
296 constraint = constraints[0] # Simple selection
297 return f"{base_query} {constraint.value}"
299 def _update_strategy_stats(
300 self, strategy: str, candidates: List[Candidate]
301 ):
302 """Update performance statistics for a strategy."""
303 self.strategy_stats[strategy]["attempts"] += 1
304 self.strategy_stats[strategy]["candidates_found"] += len(candidates)
306 # Simple quality assessment (could be more sophisticated)
307 quality = len(candidates) * 0.1 # Basic quality based on quantity
308 self.strategy_stats[strategy]["quality_sum"] += quality
310 def _adapt_strategy(self):
311 """Adapt current strategy based on performance."""
312 best_strategies = self._get_top_strategies(1)
313 if best_strategies and best_strategies[0] != self.current_strategy:
314 old_strategy = self.current_strategy
315 self.current_strategy = best_strategies[0]
316 logger.info(
317 f"Adapted strategy from '{old_strategy}' to '{self.current_strategy}'"
318 )
320 def _try_next_strategy(self) -> bool:
321 """Try the next available strategy."""
322 current_index = (
323 self.initial_strategies.index(self.current_strategy)
324 if self.current_strategy in self.initial_strategies
325 else 0
326 )
327 next_index = (current_index + 1) % len(self.initial_strategies)
329 if next_index == 0: # We've tried all strategies
330 return False
332 self.current_strategy = self.initial_strategies[next_index]
333 return True