Coverage for src/local_deep_research/advanced_search_system/candidate_exploration/progressive_explorer.py: 99%

1""" 

2Progressive explorer for BrowseComp-style systematic search exploration. 

3""" 

4 

5from dataclasses import dataclass, field 

6from typing import Dict, List, Set, Tuple 

7 

8from loguru import logger 

9 

10from ...utilities.thread_context import preserve_research_context 

11from ...utilities.threading_utils import thread_context 

12from ..parallel_search import run_parallel_searches 

13 

14 

15@dataclass 

16class SearchProgress: 

17 """Track search progress and findings.""" 

18 

19 searched_terms: Set[str] = field(default_factory=set) 

20 found_candidates: Dict[str, float] = field( 

21 default_factory=dict 

22 ) # name -> confidence 

23 verified_facts: Dict[str, str] = field( 

24 default_factory=dict 

25 ) # fact -> source 

26 entity_coverage: Dict[str, Set[str]] = field( 

27 default_factory=dict 

28 ) # entity_type -> searched_entities 

29 search_depth: int = 0 

30 

31 def update_coverage(self, entity_type: str, entity: str): 

32 """Update entity coverage tracking.""" 

33 if entity_type not in self.entity_coverage: 

34 self.entity_coverage[entity_type] = set() 

35 self.entity_coverage[entity_type].add(entity.lower()) 

36 

37 def get_uncovered_entities( 

38 self, entities: Dict[str, List[str]] 

39 ) -> Dict[str, List[str]]: 

40 """Get entities that haven't been searched yet.""" 

41 uncovered = {} 

42 for entity_type, entity_list in entities.items(): 

43 covered = self.entity_coverage.get(entity_type, set()) 

44 uncovered_list = [ 

45 e for e in entity_list if e.lower() not in covered 

46 ] 

47 if uncovered_list: 

48 uncovered[entity_type] = uncovered_list 

49 return uncovered 

50 

51 

52class ProgressiveExplorer: 

53 """ 

54 Explorer that implements progressive search strategies for BrowseComp. 

55 

56 Key features: 

57 1. Tracks search progress to avoid redundancy 

58 2. Progressively combines entities 

59 3. Identifies and pursues promising candidates 

60 4. Maintains simple approach without over-filtering 

61 """ 

62 

63 def __init__(self, search_engine, model): 

64 self.search_engine = search_engine 

65 self.model = model 

66 self.progress = SearchProgress() 

67 self.max_results_per_search = 20 # Keep more results 

68 

69 def explore( 

70 self, 

71 queries: List[str], 

72 constraints: List = None, 

73 max_workers: int = 5, 

74 extracted_entities: Dict[str, List[str]] = None, 

75 ) -> Tuple[List, SearchProgress]: 

76 """ 

77 Execute progressive exploration with entity tracking. 

78 

79 Returns both candidates and search progress for strategy use. 

80 """ 

81 all_results = [] 

82 extracted_entities = extracted_entities or {} 

83 

84 # Execute searches in parallel (like source-based strategy) 

85 search_results = self._parallel_search(queries, max_workers) 

86 

87 # Process results without filtering (trust the LLM later) 

88 for query, results in search_results: 

89 self.progress.searched_terms.add(query.lower()) 

90 

91 # Track which entities were covered in this search 

92 self._update_entity_coverage(query, extracted_entities) 

93 

94 # Extract any specific names/candidates from results 

95 candidates = self._extract_candidates_from_results(results, query) 

96 for candidate_name, confidence in candidates.items(): 

97 if candidate_name in self.progress.found_candidates: 

98 # Update confidence if higher 

99 self.progress.found_candidates[candidate_name] = max( 

100 self.progress.found_candidates[candidate_name], 

101 confidence, 

102 ) 

103 else: 

104 self.progress.found_candidates[candidate_name] = confidence 

105 

106 # Keep all results for final synthesis 

107 all_results.extend(results) 

108 

109 self.progress.search_depth += 1 

110 

111 # Return both results and progress 

112 return all_results, self.progress 

113 

114 def generate_verification_searches( 

115 self, 

116 candidates: Dict[str, float], 

117 constraints: List, 

118 max_searches: int = 5, 

119 ) -> List[str]: 

120 """Generate targeted searches to verify top candidates.""" 

121 if not candidates: 

122 return [] 

123 

124 # Get top candidates by confidence 

125 top_candidates = sorted( 

126 candidates.items(), key=lambda x: x[1], reverse=True 

127 )[:3] 

128 

129 verification_searches = [] 

130 for candidate_name, confidence in top_candidates: 

131 # Generate verification searches for this candidate 

132 for constraint in constraints[:2]: # Verify top constraints 

133 search = f'"{candidate_name}" {constraint.description}' 

134 if search.lower() not in self.progress.searched_terms: 

135 verification_searches.append(search) 

136 

137 return verification_searches[:max_searches] 

138 

139 def _extract_candidates_from_results( 

140 self, results: List[Dict], query: str 

141 ) -> Dict[str, float]: 

142 """Extract potential answer candidates from search results.""" 

143 candidates = {} 

144 

145 # Simple extraction based on titles and snippets 

146 for result in results[:10]: # Focus on top results 

147 title = result.get("title", "") 

148 snippet = result.get("snippet", "") 

149 

150 # Look for proper nouns and specific names 

151 # This is simplified - in practice, might use NER or more sophisticated extraction 

152 combined_text = f"{title} {snippet}" 

153 

154 # Extract quoted terms as potential candidates 

155 import re 

156 

157 quoted_terms = re.findall(r'"([^"]+)"', combined_text) 

158 for term in quoted_terms: 

159 if ( 

160 len(term) > 2 and len(term) < 50 

161 ): # Reasonable length for an answer 

162 candidates[term] = 0.3 # Base confidence from appearance 

163 

164 # Boost confidence if appears in title 

165 if title: 

166 # Titles often contain the actual answer 

167 title_words = title.split() 

168 for i in range(len(title_words)): 

169 for j in range(i + 1, min(i + 4, len(title_words) + 1)): 

170 phrase = " ".join(title_words[i:j]) 

171 if ( 

172 len(phrase) > 3 and phrase[0].isupper() 

173 ): # Likely proper noun 

174 candidates[phrase] = candidates.get(phrase, 0) + 0.2 

175 

176 return candidates 

177 

178 def _update_entity_coverage( 

179 self, query: str, entities: Dict[str, List[str]] 

180 ): 

181 """Track which entities have been covered in searches.""" 

182 query_lower = query.lower() 

183 

184 for entity_type, entity_list in entities.items(): 

185 for entity in entity_list: 

186 if entity.lower() in query_lower: 

187 self.progress.update_coverage(entity_type, entity) 

188 

189 def suggest_next_searches( 

190 self, entities: Dict[str, List[str]], max_suggestions: int = 5 

191 ) -> List[str]: 

192 """Suggest next searches based on coverage and findings.""" 

193 suggestions = [] 

194 

195 # 1. Check uncovered entities 

196 uncovered = self.progress.get_uncovered_entities(entities) 

197 

198 # 2. If we have candidates, verify them with uncovered constraints 

199 if self.progress.found_candidates: 

200 top_candidate = max( 

201 self.progress.found_candidates.items(), key=lambda x: x[1] 

202 )[0] 

203 

204 # Combine candidate with uncovered entities 

205 for entity_list in uncovered.values(): 

206 for entity in entity_list[:2]: 

207 search = f'"{top_candidate}" {entity}' 

208 if search.lower() not in self.progress.searched_terms: 

209 suggestions.append(search) 

210 

211 # 3. Otherwise, create new combinations of uncovered entities 

212 else: 

213 # Focus on systematic coverage 

214 if uncovered.get("temporal"): 

215 # Year-by-year with key term 

216 key_term = ( 

217 entities.get("names", [""])[0] 

218 or entities.get("descriptors", [""])[0] 

219 ) 

220 for year in uncovered["temporal"][:3]: 

221 search = f"{key_term} {year}".strip() 

222 if search.lower() not in self.progress.searched_terms: 222 ↛ 220line 222 didn't jump to line 220 because the condition on line 222 was always true

223 suggestions.append(search) 

224 

225 if uncovered.get("names") and uncovered.get("descriptors"): 

226 # Combine names with descriptors 

227 for name in uncovered["names"][:2]: 

228 for desc in uncovered["descriptors"][:2]: 

229 search = f"{name} {desc}" 

230 if search.lower() not in self.progress.searched_terms: 230 ↛ 228line 230 didn't jump to line 228 because the condition on line 230 was always true

231 suggestions.append(search) 

232 

233 return suggestions[:max_suggestions] 

234 

235 def _parallel_search( 

236 self, queries: List[str], max_workers: int 

237 ) -> List[Tuple[str, List[Dict]]]: 

238 """Execute searches in parallel and return results.""" 

239 results = [] 

240 

241 def search_query(query): 

242 try: 

243 search_results = self.search_engine.run(query) 

244 return search_results or [] 

245 except Exception: 

246 logger.exception(f"Error searching '{query}'") 

247 return [] 

248 

249 # Create context-preserving wrapper for the search function 

250 context_aware_search = preserve_research_context(search_query) 

251 

252 # Run searches in parallel. thread_context (the factory, not a call) 

253 # is invoked once per query so each worker thread gets its OWN fresh 

254 # Flask app context, required so current_app / g stay accessible 

255 # inside self.search_engine.run, matching SourceBasedSearchStrategy 

256 # and FocusedIterationStrategy. Without it, a search that touches 

257 # current_app raises "Working outside of application context" in the 

258 # worker, which the per-query handler above swallows into an empty 

259 # result. 

260 completed = run_parallel_searches( 

261 queries, 

262 context_aware_search, 

263 max_workers=max_workers, 

264 context_factory=thread_context, 

265 ) 

266 results.extend(completed) 

267 

268 return results