Coverage for src/local_deep_research/advanced_search_system/candidate_exploration/base_explorer.py: 98%
132 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Base candidate explorer for inheritance-based exploration system.
4This module provides the base interface and common functionality for
5candidate exploration implementations.
6"""
8from abc import ABC, abstractmethod
9from dataclasses import dataclass
10from enum import Enum
11from typing import Any, Optional
13from langchain_core.language_models import BaseChatModel
14from loguru import logger
16from ..candidates.base_candidate import Candidate
17from ..constraints.base_constraint import Constraint
18from ...utilities.json_utils import get_llm_response_text
19from ...utilities.llm_utils import invoke_llm_sync
22class ExplorationStrategy(Enum):
23 """Different exploration strategies."""
25 BREADTH_FIRST = "breadth_first" # Explore widely first
26 DEPTH_FIRST = "depth_first" # Deep dive into promising areas
27 CONSTRAINT_GUIDED = "constraint_guided" # Let constraints guide exploration
28 DIVERSITY_FOCUSED = "diversity_focused" # Maximize candidate diversity
29 ADAPTIVE = "adaptive" # Adapt based on findings
32@dataclass
33class ExplorationResult:
34 """Result of candidate exploration."""
36 candidates: list[Candidate]
37 total_searched: int
38 unique_candidates: int
39 exploration_paths: list[str]
40 metadata: dict[str, Any]
41 elapsed_time: float
42 strategy_used: ExplorationStrategy
45class BaseCandidateExplorer(ABC):
46 """
47 Base class for candidate exploration implementations.
49 This provides the common interface and shared functionality that
50 all candidate explorers should implement.
51 """
53 def __init__(
54 self,
55 model: BaseChatModel,
56 search_engine,
57 max_candidates: int = 50,
58 max_search_time: float = 60.0,
59 **kwargs,
60 ):
61 """
62 Initialize the base candidate explorer.
64 Args:
65 model: Language model for analysis
66 search_engine: Search engine for finding candidates
67 max_candidates: Maximum number of candidates to find
68 max_search_time: Maximum time to spend searching
69 **kwargs: Additional parameters for specific implementations
70 """
71 self.model = model
72 self.search_engine = search_engine
73 self.max_candidates = max_candidates
74 self.max_search_time = max_search_time
76 # Tracking
77 self.explored_queries: set[str] = set()
78 self.found_candidates: dict[str, Candidate] = {}
80 @abstractmethod
81 def explore(
82 self,
83 initial_query: str,
84 constraints: Optional[list[Constraint]] = None,
85 entity_type: Optional[str] = None,
86 ) -> ExplorationResult:
87 """
88 Explore and discover candidates.
90 Args:
91 initial_query: Starting query for exploration
92 constraints: Optional constraints to guide exploration
93 entity_type: Optional entity type to focus on
95 Returns:
96 ExplorationResult: Complete exploration results
97 """
98 pass
100 @abstractmethod
101 def generate_exploration_queries(
102 self,
103 base_query: str,
104 found_candidates: list[Candidate],
105 constraints: Optional[list[Constraint]] = None,
106 ) -> list[str]:
107 """
108 Generate new queries for continued exploration.
110 Args:
111 base_query: Original base query
112 found_candidates: Candidates found so far
113 constraints: Optional constraints to consider
115 Returns:
116 List[str]: New queries to explore
117 """
118 pass
120 def _execute_search(self, query: str) -> dict[str, Any]:
121 """Execute a search query."""
122 try:
123 # Mark query as explored
124 self.explored_queries.add(query.lower())
126 # Execute search
127 results = self.search_engine.run(query)
129 # Handle different result formats
130 if isinstance(results, list):
131 # If results is a list, wrap it in the expected format
132 formatted_results = {"results": results, "query": query}
133 logger.info(
134 f"Search '{query[:50]}...' returned {len(results)} results"
135 )
136 return formatted_results
137 if isinstance(results, dict):
138 # If results is already a dict, use it as is
139 result_count = len(results.get("results", []))
140 logger.info(
141 f"Search '{query[:50]}...' returned {result_count} results"
142 )
143 return results
144 # Unknown format, return empty
145 logger.warning(f"Unknown search result format: {type(results)}")
146 return {"results": [], "query": query}
148 except Exception:
149 logger.exception(f"Error executing search '{query}'")
150 return {"results": []}
152 def _extract_candidates_from_results(
153 self,
154 results: dict[str, Any],
155 original_query: str = None,
156 entity_type: Optional[str] = None,
157 ) -> list[Candidate]:
158 """Generate answer candidates directly from search results using LLM."""
159 candidates = []
161 # Collect all search result content
162 all_content = []
163 for result in results.get("results", []):
164 title = result.get("title", "")
165 snippet = result.get("snippet", "")
166 if title or snippet: 166 ↛ 163line 166 didn't jump to line 163 because the condition on line 166 was always true
167 all_content.append(f"Title: {title}\nContent: {snippet}")
169 if not all_content or not original_query:
170 return candidates
172 # Generate answer candidates using LLM
173 answer_candidates = self._generate_answer_candidates(
174 original_query,
175 "\n\n".join(all_content[:10]), # Limit to first 10 results
176 )
178 for answer in answer_candidates:
179 if answer and answer not in self.found_candidates:
180 candidate = Candidate(
181 name=answer,
182 metadata={
183 "source": "llm_answer_generation",
184 "query": results.get("query", ""),
185 "original_query": original_query,
186 "result_count": len(results.get("results", [])),
187 },
188 )
189 candidates.append(candidate)
190 self.found_candidates[answer] = candidate
192 return candidates
194 def _generate_answer_candidates(
195 self, question: str, search_content: str
196 ) -> list[str]:
197 """Generate multiple answer candidates from search results."""
198 prompt = f"""
199Question: {question}
201Based on these search results, provide 3-5 possible answers:
203{search_content}
205Give me multiple possible answers, one per line:
206"""
208 try:
209 response = invoke_llm_sync(self.model, prompt)
210 content = get_llm_response_text(response)
212 # Parse multiple answers
213 answers = []
214 for line in content.split("\n"):
215 line = line.strip()
216 if line: 216 ↛ 214line 216 didn't jump to line 214 because the condition on line 216 was always true
217 # Clean up common prefixes and formatting
218 line = line.lstrip("•-*1234567890.").strip()
219 if line and len(line) > 2: # Skip very short answers
220 answers.append(line)
222 return answers[:5] # Limit to 5 candidates max
224 except Exception:
225 logger.exception("Error generating answer candidates")
226 return []
228 def _extract_entity_names(
229 self, text: str, entity_type: Optional[str] = None
230 ) -> list[str]:
231 """Extract entity names from text using LLM."""
232 if not text.strip():
233 return []
235 prompt = f"""
236Extract specific entity names from this text.
237{"Focus on: " + entity_type if entity_type else "Extract any named entities."}
239Text: {text[:500]}
241Return only the names, one per line. Be selective - only include clear, specific names.
242Do not include:
243- Generic terms or categories
244- Adjectives or descriptions
245- Common words
247Names:
248"""
250 try:
251 response = get_llm_response_text(
252 invoke_llm_sync(self.model, prompt)
253 )
255 # Parse response into names
256 names = []
257 for line in response.split("\n"):
258 name = line.strip()
259 if (
260 name
261 and len(name) > 2
262 and not name.lower().startswith(("the ", "a ", "an "))
263 ):
264 names.append(name)
266 return names[:5] # Limit to top 5 per text
268 except Exception:
269 logger.exception("Error extracting entity names")
270 return []
272 def _should_continue_exploration(
273 self, start_time: float, candidates_found: int
274 ) -> bool:
275 """Determine if exploration should continue."""
276 import time
278 elapsed = time.time() - start_time
280 # Stop if time limit reached
281 if elapsed > self.max_search_time:
282 logger.info(f"Time limit reached ({elapsed:.1f}s)")
283 return False
285 # Stop if candidate limit reached
286 if candidates_found >= self.max_candidates:
287 logger.info(f"Candidate limit reached ({candidates_found})")
288 return False
290 return True
292 def _deduplicate_candidates(
293 self, candidates: list[Candidate]
294 ) -> list[Candidate]:
295 """Remove duplicate candidates based on name similarity."""
296 unique_candidates = []
297 seen_names = set()
299 for candidate in candidates:
300 # Simple deduplication by exact name match
301 name_key = candidate.name.lower().strip()
302 if name_key not in seen_names:
303 seen_names.add(name_key)
304 unique_candidates.append(candidate)
306 return unique_candidates
308 def _rank_candidates_by_relevance(
309 self, candidates: list[Candidate], query: str
310 ) -> list[Candidate]:
311 """Rank candidates by relevance to original query."""
312 if not candidates:
313 return candidates
315 # Simple relevance scoring based on metadata
316 for candidate in candidates:
317 score = 0.0
319 # Score based on source query similarity
320 if "query" in candidate.metadata:
321 # Simple word overlap scoring
322 query_words = set(query.lower().split())
323 candidate_query_words = set(
324 candidate.metadata["query"].lower().split()
325 )
326 overlap = len(query_words.intersection(candidate_query_words))
327 score += overlap * 0.1
329 # Score based on result title relevance
330 if "result_title" in candidate.metadata:
331 title_words = set(
332 candidate.metadata["result_title"].lower().split()
333 )
334 overlap = len(query_words.intersection(title_words))
335 score += overlap * 0.2
337 candidate.relevance_score = score
339 # Sort by relevance
340 return sorted(
341 candidates,
342 key=lambda c: getattr(c, "relevance_score", 0.0),
343 reverse=True,
344 )