Coverage for src/local_deep_research/search_system_factory.py: 90%

1""" 

2Factory for creating search strategies. 

3This module provides a centralized way to create search strategies 

4to avoid code duplication. 

5""" 

6 

7from loguru import logger 

8from typing import Optional, Dict, Any, List 

9from langchain_core.language_models import BaseChatModel 

10 

11from .utilities.type_utils import unwrap_setting 

12 

13# Re-export from constants so existing importers don't break 

14from .constants import ( # noqa: F401 

15 AVAILABLE_STRATEGIES, 

16 get_available_strategies, 

17) 

18 

19 

20def _get_setting( 

21 settings_snapshot: Optional[Dict], key: str, default: Any 

22) -> Any: 

23 """Get a setting value from the snapshot, handling nested dict structure.""" 

24 if not settings_snapshot or key not in settings_snapshot: 

25 return default 

26 value = settings_snapshot[key] 

27 return unwrap_setting(value) 

28 

29 

30def create_strategy( 

31 strategy_name: str, 

32 model: BaseChatModel, 

33 search: Any, 

34 all_links_of_system: Optional[List[Dict]] = None, 

35 settings_snapshot: Optional[Dict] = None, 

36 research_context: Optional[Dict] = None, 

37 **kwargs, 

38): 

39 """ 

40 Create a search strategy by name. 

41 

42 Args: 

43 strategy_name: Name of the strategy to create 

44 model: Language model to use 

45 search: Search engine instance 

46 all_links_of_system: List of existing links 

47 settings_snapshot: Settings snapshot 

48 research_context: Research context for special strategies 

49 **kwargs: Additional strategy-specific parameters 

50 

51 Returns: 

52 Strategy instance 

53 """ 

54 if all_links_of_system is None: 

55 all_links_of_system = [] 

56 

57 strategy_name_lower = strategy_name.lower() 

58 

59 # Source-based strategy 

60 if strategy_name_lower in [ 

61 "source-based", 

62 "source_based", 

63 "source_based_search", 

64 ]: 

65 from .advanced_search_system.strategies.source_based_strategy import ( 

66 SourceBasedSearchStrategy, 

67 ) 

68 

69 return SourceBasedSearchStrategy( 

70 model=model, 

71 search=search, 

72 include_text_content=kwargs.get("include_text_content", True), 

73 use_cross_engine_filter=kwargs.get("use_cross_engine_filter", True), 

74 all_links_of_system=all_links_of_system, 

75 use_atomic_facts=kwargs.get("use_atomic_facts", False), 

76 settings_snapshot=settings_snapshot, 

77 search_original_query=kwargs.get("search_original_query", True), 

78 ) 

79 

80 # Focused iteration strategy 

81 if strategy_name_lower in ["focused-iteration", "focused_iteration"]: 

82 from .advanced_search_system.strategies.focused_iteration_strategy import ( 

83 FocusedIterationStrategy, 

84 ) 

85 

86 # Read focused_iteration settings with kwargs override 

87 # adaptive_questions is stored as 0/1 integer, convert to bool 

88 enable_adaptive = bool( 

89 kwargs.get( 

90 "enable_adaptive_questions", 

91 _get_setting( 

92 settings_snapshot, "focused_iteration.adaptive_questions", 0 

93 ), 

94 ) 

95 ) 

96 knowledge_limit = kwargs.get( 

97 "knowledge_summary_limit", 

98 _get_setting( 

99 settings_snapshot, 

100 "focused_iteration.knowledge_summary_limit", 

101 10, 

102 ), 

103 ) 

104 snippet_truncate = kwargs.get( 

105 "knowledge_snippet_truncate", 

106 _get_setting( 

107 settings_snapshot, "focused_iteration.snippet_truncate", 200 

108 ), 

109 ) 

110 question_gen_type = kwargs.get( 

111 "question_generator", 

112 _get_setting( 

113 settings_snapshot, 

114 "focused_iteration.question_generator", 

115 "browsecomp", 

116 ), 

117 ) 

118 prompt_knowledge_truncate = kwargs.get( 

119 "prompt_knowledge_truncate", 

120 _get_setting( 

121 settings_snapshot, 

122 "focused_iteration.prompt_knowledge_truncate", 

123 1500, 

124 ), 

125 ) 

126 previous_searches_limit = kwargs.get( 

127 "previous_searches_limit", 

128 _get_setting( 

129 settings_snapshot, 

130 "focused_iteration.previous_searches_limit", 

131 10, 

132 ), 

133 ) 

134 # Convert 0 to None for "unlimited" 

135 if knowledge_limit == 0: 

136 knowledge_limit = None 

137 if snippet_truncate == 0: 

138 snippet_truncate = None 

139 if prompt_knowledge_truncate == 0: 

140 prompt_knowledge_truncate = None 

141 if previous_searches_limit == 0: 

142 previous_searches_limit = None 

143 

144 strategy = FocusedIterationStrategy( 

145 model=model, 

146 search=search, 

147 all_links_of_system=all_links_of_system, 

148 max_iterations=kwargs.get("max_iterations", 8), 

149 questions_per_iteration=kwargs.get("questions_per_iteration", 5), 

150 settings_snapshot=settings_snapshot, 

151 # Options read from settings (with kwargs override) 

152 enable_adaptive_questions=enable_adaptive, 

153 enable_early_termination=kwargs.get( 

154 "enable_early_termination", False 

155 ), 

156 knowledge_summary_limit=knowledge_limit, 

157 knowledge_snippet_truncate=snippet_truncate, 

158 prompt_knowledge_truncate=prompt_knowledge_truncate, 

159 previous_searches_limit=previous_searches_limit, 

160 ) 

161 

162 # Override question generator if flexible is selected 

163 if question_gen_type == "flexible": 

164 from .advanced_search_system.questions.flexible_browsecomp_question import ( 

165 FlexibleBrowseCompQuestionGenerator, 

166 ) 

167 

168 # Pass truncation settings to flexible generator 

169 strategy.question_generator = FlexibleBrowseCompQuestionGenerator( 

170 model, 

171 knowledge_truncate_length=prompt_knowledge_truncate, 

172 previous_searches_limit=previous_searches_limit, 

173 ) 

174 

175 return strategy 

176 

177 # Focused iteration strategy with standard citation handler 

178 if strategy_name_lower in [ 

179 "focused-iteration-standard", 

180 "focused_iteration_standard", 

181 ]: 

182 from .advanced_search_system.strategies.focused_iteration_strategy import ( 

183 FocusedIterationStrategy, 

184 ) 

185 from .citation_handler import CitationHandler 

186 

187 # Use standard citation handler (same question generator as regular focused-iteration) 

188 standard_citation_handler = CitationHandler( 

189 model, handler_type="standard", settings_snapshot=settings_snapshot 

190 ) 

191 

192 # Read focused_iteration settings with kwargs override 

193 # adaptive_questions is stored as 0/1 integer, convert to bool 

194 enable_adaptive = bool( 

195 kwargs.get( 

196 "enable_adaptive_questions", 

197 _get_setting( 

198 settings_snapshot, "focused_iteration.adaptive_questions", 0 

199 ), 

200 ) 

201 ) 

202 knowledge_limit = kwargs.get( 

203 "knowledge_summary_limit", 

204 _get_setting( 

205 settings_snapshot, 

206 "focused_iteration.knowledge_summary_limit", 

207 10, 

208 ), 

209 ) 

210 snippet_truncate = kwargs.get( 

211 "knowledge_snippet_truncate", 

212 _get_setting( 

213 settings_snapshot, "focused_iteration.snippet_truncate", 200 

214 ), 

215 ) 

216 question_gen_type = kwargs.get( 

217 "question_generator", 

218 _get_setting( 

219 settings_snapshot, 

220 "focused_iteration.question_generator", 

221 "browsecomp", 

222 ), 

223 ) 

224 prompt_knowledge_truncate = kwargs.get( 

225 "prompt_knowledge_truncate", 

226 _get_setting( 

227 settings_snapshot, 

228 "focused_iteration.prompt_knowledge_truncate", 

229 1500, 

230 ), 

231 ) 

232 previous_searches_limit = kwargs.get( 

233 "previous_searches_limit", 

234 _get_setting( 

235 settings_snapshot, 

236 "focused_iteration.previous_searches_limit", 

237 10, 

238 ), 

239 ) 

240 # Convert 0 to None for "unlimited" 

241 if knowledge_limit == 0: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true

242 knowledge_limit = None 

243 if snippet_truncate == 0: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true

244 snippet_truncate = None 

245 if prompt_knowledge_truncate == 0: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true

246 prompt_knowledge_truncate = None 

247 if previous_searches_limit == 0: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true

248 previous_searches_limit = None 

249 

250 strategy = FocusedIterationStrategy( 

251 model=model, 

252 search=search, 

253 citation_handler=standard_citation_handler, 

254 all_links_of_system=all_links_of_system, 

255 max_iterations=kwargs.get("max_iterations", 8), 

256 questions_per_iteration=kwargs.get("questions_per_iteration", 5), 

257 use_browsecomp_optimization=True, # Keep BrowseComp features 

258 settings_snapshot=settings_snapshot, 

259 # Options read from settings (with kwargs override) 

260 enable_adaptive_questions=enable_adaptive, 

261 enable_early_termination=kwargs.get( 

262 "enable_early_termination", False 

263 ), 

264 knowledge_summary_limit=knowledge_limit, 

265 knowledge_snippet_truncate=snippet_truncate, 

266 prompt_knowledge_truncate=prompt_knowledge_truncate, 

267 previous_searches_limit=previous_searches_limit, 

268 ) 

269 

270 # Override question generator if flexible is selected 

271 if question_gen_type == "flexible": 271 ↛ 272line 271 didn't jump to line 272 because the condition on line 271 was never true

272 from .advanced_search_system.questions.flexible_browsecomp_question import ( 

273 FlexibleBrowseCompQuestionGenerator, 

274 ) 

275 

276 # Pass truncation settings to flexible generator 

277 strategy.question_generator = FlexibleBrowseCompQuestionGenerator( 

278 model, 

279 knowledge_truncate_length=prompt_knowledge_truncate, 

280 previous_searches_limit=previous_searches_limit, 

281 ) 

282 

283 return strategy 

284 

285 # News aggregation strategy (used internally by the news subsystem) 

286 if strategy_name_lower in [ 

287 "news", 

288 "news_aggregation", 

289 "news-aggregation", 

290 ]: 

291 from .advanced_search_system.strategies.news_strategy import ( 

292 NewsAggregationStrategy, 

293 ) 

294 

295 return NewsAggregationStrategy( 

296 model=model, 

297 search=search, 

298 all_links_of_system=all_links_of_system, 

299 ) 

300 

301 # Topic organization strategy 

302 if strategy_name_lower in [ 

303 "topic-organization", 

304 "topic_organization", 

305 "topic", 

306 ]: 

307 from .advanced_search_system.strategies.topic_organization_strategy import ( 

308 TopicOrganizationStrategy, 

309 ) 

310 

311 return TopicOrganizationStrategy( 

312 model=model, 

313 search=search, 

314 all_links_of_system=all_links_of_system, 

315 settings_snapshot=settings_snapshot, 

316 min_sources_per_topic=1, # Allow single-source topics 

317 use_cross_engine_filter=kwargs.get("use_cross_engine_filter", True), 

318 filter_reorder=kwargs.get("filter_reorder", True), 

319 filter_reindex=kwargs.get("filter_reindex", True), 

320 cross_engine_max_results=kwargs.get( # type: ignore[arg-type] 

321 "cross_engine_max_results", None 

322 ), 

323 search_original_query=kwargs.get("search_original_query", True), 

324 max_topics=kwargs.get("max_topics", 5), 

325 similarity_threshold=kwargs.get("similarity_threshold", 0.3), 

326 use_focused_iteration=kwargs.get("use_focused_iteration", False), 

327 enable_refinement=kwargs.get( 

328 "enable_refinement", False 

329 ), # Disable refinement iterations for now 

330 max_refinement_iterations=kwargs.get( 

331 "max_refinement_iterations", 

332 1, # Set to 1 iteration for faster results 

333 ), 

334 generate_text=kwargs.get("generate_text", True), 

335 ) 

336 

337 # LangGraph agent strategy (parallel subagent research). 

338 # ``mcp`` / ``agentic`` were removed (#4548); they remain here as 

339 # deprecated aliases so existing saved settings, queued runs, and API 

340 # callers route to the closest successor (langgraph-agent) instead of 

341 # the source-based fallback below. 

342 if strategy_name_lower in [ 

343 "langgraph-agent", 

344 "langgraph_agent", 

345 "mcp", 

346 "agentic", 

347 ]: 

348 if strategy_name_lower in ("mcp", "agentic"): 

349 logger.warning( 

350 f"Strategy {strategy_name!r} was removed (#4548); " 

351 "using 'langgraph-agent' instead." 

352 ) 

353 from .advanced_search_system.strategies.langgraph_agent_strategy import ( 

354 LangGraphAgentStrategy, 

355 ) 

356 

357 return LangGraphAgentStrategy( 

358 model=model, 

359 search=search, 

360 max_iterations=kwargs.get( 

361 "max_iterations", 

362 _get_setting( 

363 settings_snapshot, "langgraph_agent.max_iterations", 50 

364 ), 

365 ), 

366 max_sub_iterations=kwargs.get( 

367 "max_sub_iterations", 

368 _get_setting( 

369 settings_snapshot, "langgraph_agent.max_sub_iterations", 8 

370 ), 

371 ), 

372 include_sub_research=kwargs.get( 

373 "include_sub_research", 

374 _get_setting( 

375 settings_snapshot, 

376 "langgraph_agent.include_sub_research", 

377 True, 

378 ), 

379 ), 

380 programmatic_mode=kwargs.get("programmatic_mode", False), 

381 all_links_of_system=all_links_of_system, 

382 settings_snapshot=settings_snapshot, 

383 ) 

384 

385 # Default to source-based if unknown 

386 logger.warning( 

387 f"Unknown strategy: {strategy_name}, defaulting to source-based" 

388 ) 

389 from .advanced_search_system.strategies.source_based_strategy import ( 

390 SourceBasedSearchStrategy, 

391 ) 

392 

393 return SourceBasedSearchStrategy( 

394 model=model, 

395 search=search, 

396 include_text_content=True, 

397 use_cross_engine_filter=True, 

398 all_links_of_system=all_links_of_system, 

399 use_atomic_facts=False, 

400 settings_snapshot=settings_snapshot, 

401 search_original_query=kwargs.get("search_original_query", True), 

402 )