Coverage for src/local_deep_research/advanced_search_system/candidate_exploration/parallel_explorer.py: 99%
92 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Parallel candidate explorer implementation.
4This explorer runs multiple search queries in parallel to quickly discover
5a wide range of candidates.
6"""
8import concurrent.futures
9import time
10from typing import List, Optional
12from loguru import logger
14from ...database.thread_local_session import thread_cleanup
15from ...utilities.json_utils import get_llm_response_text
16from ..candidates.base_candidate import Candidate
17from ..constraints.base_constraint import Constraint
18from .base_explorer import (
19 BaseCandidateExplorer,
20 ExplorationResult,
21 ExplorationStrategy,
22)
23from ...utilities.llm_utils import invoke_llm_sync
26class ParallelExplorer(BaseCandidateExplorer):
27 """
28 Parallel candidate explorer that runs multiple searches concurrently.
30 This explorer:
31 1. Generates multiple search queries from the initial query
32 2. Runs searches in parallel for speed
33 3. Collects and deduplicates candidates
34 4. Focuses on breadth-first exploration
35 """
37 def __init__(
38 self,
39 *args,
40 max_workers: int = 5,
41 queries_per_round: int = 8,
42 max_rounds: int = 3,
43 **kwargs,
44 ):
45 """
46 Initialize parallel explorer.
48 Args:
49 max_workers: Maximum number of parallel search threads
50 queries_per_round: Number of queries to generate per round
51 max_rounds: Maximum exploration rounds
52 """
53 super().__init__(*args, **kwargs)
54 self.max_workers = max_workers
55 self.queries_per_round = queries_per_round
56 self.max_rounds = max_rounds
58 def explore(
59 self,
60 initial_query: str,
61 constraints: Optional[List[Constraint]] = None,
62 entity_type: Optional[str] = None,
63 ) -> ExplorationResult:
64 """Explore candidates using parallel search strategy."""
65 start_time = time.time()
66 logger.info(f"Starting parallel exploration for: {initial_query}")
68 all_candidates = []
69 exploration_paths = []
70 total_searched = 0
72 # Initial search
73 current_queries = [initial_query]
75 with concurrent.futures.ThreadPoolExecutor(
76 max_workers=self.max_workers
77 ) as executor:
78 for round_num in range(self.max_rounds):
79 if not self._should_continue_exploration(
80 start_time, len(all_candidates)
81 ):
82 break
84 logger.info(
85 f"Exploration round {round_num + 1}: {len(current_queries)} queries"
86 )
88 # Submit all queries for parallel execution
89 future_to_query = {
90 executor.submit(
91 thread_cleanup(self._execute_search), query
92 ): query
93 for query in current_queries
94 }
96 round_candidates = []
98 # Collect results as they complete
99 for future in concurrent.futures.as_completed(future_to_query):
100 query = future_to_query[future]
101 total_searched += 1
103 try:
104 results = future.result()
105 candidates = self._extract_candidates_from_results(
106 results, entity_type
107 )
108 round_candidates.extend(candidates)
109 exploration_paths.append(
110 f"Round {round_num + 1}: {query} -> {len(candidates)} candidates"
111 )
113 except Exception:
114 logger.exception(f"Error processing query '{query}'")
116 # Add new candidates
117 all_candidates.extend(round_candidates)
119 # Generate queries for next round
120 if round_num < self.max_rounds - 1:
121 current_queries = self.generate_exploration_queries(
122 initial_query, all_candidates, constraints
123 )[: self.queries_per_round]
125 if not current_queries:
126 logger.info("No more queries to explore")
127 break
129 # Deduplicate and rank
130 unique_candidates = self._deduplicate_candidates(all_candidates)
131 ranked_candidates = self._rank_candidates_by_relevance(
132 unique_candidates, initial_query
133 )
135 # Limit to max candidates
136 final_candidates = ranked_candidates[: self.max_candidates]
138 elapsed_time = time.time() - start_time
139 logger.info(
140 f"Parallel exploration completed: {len(final_candidates)} unique candidates in {elapsed_time:.1f}s"
141 )
143 return ExplorationResult(
144 candidates=final_candidates,
145 total_searched=total_searched,
146 unique_candidates=len(unique_candidates),
147 exploration_paths=exploration_paths,
148 metadata={
149 "strategy": "parallel",
150 "rounds": min(round_num + 1, self.max_rounds),
151 "max_workers": self.max_workers,
152 "entity_type": entity_type,
153 },
154 elapsed_time=elapsed_time,
155 strategy_used=ExplorationStrategy.BREADTH_FIRST,
156 )
158 def generate_exploration_queries(
159 self,
160 base_query: str,
161 found_candidates: List[Candidate],
162 constraints: Optional[List[Constraint]] = None,
163 ) -> List[str]:
164 """Generate queries for parallel exploration."""
165 queries = []
167 # Query variations based on base query
168 base_variations = self._generate_query_variations(base_query)
169 queries.extend(base_variations)
171 # Queries based on found candidates
172 if found_candidates:
173 candidate_queries = self._generate_candidate_based_queries(
174 found_candidates, base_query
175 )
176 queries.extend(candidate_queries)
178 # Constraint-based queries
179 if constraints:
180 constraint_queries = self._generate_constraint_queries(
181 constraints, base_query
182 )
183 queries.extend(constraint_queries)
185 # Remove already explored queries
186 new_queries = [
187 q for q in queries if q.lower() not in self.explored_queries
188 ]
190 return new_queries[: self.queries_per_round]
192 def _generate_query_variations(self, base_query: str) -> List[str]:
193 """Generate variations of the base query."""
194 try:
195 prompt = f"""
196Generate 4 search query variations for: "{base_query}"
198Each variation should:
1991. Use different keywords but same intent
2002. Be specific and searchable
2013. Focus on finding concrete examples or instances
203Format as numbered list:
2041. [query]
2052. [query]
2063. [query]
2074. [query]
208"""
210 response = get_llm_response_text(
211 invoke_llm_sync(self.model, prompt)
212 ).strip()
214 # Parse numbered list
215 queries = []
216 for line in response.split("\n"):
217 line = line.strip()
218 if line and any(line.startswith(f"{i}.") for i in range(1, 10)):
219 # Remove number prefix
220 query = line.split(".", 1)[1].strip()
221 if query: 221 ↛ 216line 221 didn't jump to line 216 because the condition on line 221 was always true
222 queries.append(query)
224 return queries[:4]
226 except Exception:
227 logger.exception("Error generating query variations")
228 return []
230 def _generate_candidate_based_queries(
231 self, candidates: List[Candidate], base_query: str
232 ) -> List[str]:
233 """Generate queries based on found candidates."""
234 queries = []
236 # Sample a few candidates to avoid too many queries
237 sample_candidates = candidates[:3]
239 for candidate in sample_candidates:
240 # Query for similar entities
241 queries.append(f'similar to "{candidate.name}"')
242 queries.append(f'like "{candidate.name}" examples')
244 return queries
246 def _generate_constraint_queries(
247 self, constraints: List[Constraint], base_query: str
248 ) -> List[str]:
249 """Generate queries focusing on specific constraints."""
250 queries = []
252 # Sample constraints to avoid too many queries
253 for constraint in constraints[:2]:
254 queries.append(f"{constraint.value} examples")
255 queries.append(f'"{constraint.value}" instances')
257 return queries