Coverage for src/local_deep_research/search_system_factory.py: 90%
75 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Factory for creating search strategies.
3This module provides a centralized way to create search strategies
4to avoid code duplication.
5"""
7from loguru import logger
8from typing import Optional, Dict, Any, List
9from langchain_core.language_models import BaseChatModel
11from .utilities.type_utils import unwrap_setting
13# Re-export from constants so existing importers don't break
14from .constants import ( # noqa: F401
15 AVAILABLE_STRATEGIES,
16 get_available_strategies,
17)
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)
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.
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
51 Returns:
52 Strategy instance
53 """
54 if all_links_of_system is None:
55 all_links_of_system = []
57 strategy_name_lower = strategy_name.lower()
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 )
69 return SourceBasedSearchStrategy(
70 model=model,
71 search=search,
72 include_text_content=kwargs.get("include_text_content"),
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 max_iterations=kwargs.get("max_iterations"),
79 questions_per_iteration=kwargs.get("questions_per_iteration"),
80 )
82 # Focused iteration strategy
83 if strategy_name_lower in ["focused-iteration", "focused_iteration"]:
84 from .advanced_search_system.strategies.focused_iteration_strategy import (
85 FocusedIterationStrategy,
86 )
88 # Read focused_iteration settings with kwargs override
89 # adaptive_questions is stored as 0/1 integer, convert to bool
90 enable_adaptive = bool(
91 kwargs.get(
92 "enable_adaptive_questions",
93 _get_setting(
94 settings_snapshot, "focused_iteration.adaptive_questions", 0
95 ),
96 )
97 )
98 knowledge_limit = kwargs.get(
99 "knowledge_summary_limit",
100 _get_setting(
101 settings_snapshot,
102 "focused_iteration.knowledge_summary_limit",
103 10,
104 ),
105 )
106 snippet_truncate = kwargs.get(
107 "knowledge_snippet_truncate",
108 _get_setting(
109 settings_snapshot, "focused_iteration.snippet_truncate", 200
110 ),
111 )
112 question_gen_type = kwargs.get(
113 "question_generator",
114 _get_setting(
115 settings_snapshot,
116 "focused_iteration.question_generator",
117 "browsecomp",
118 ),
119 )
120 prompt_knowledge_truncate = kwargs.get(
121 "prompt_knowledge_truncate",
122 _get_setting(
123 settings_snapshot,
124 "focused_iteration.prompt_knowledge_truncate",
125 1500,
126 ),
127 )
128 previous_searches_limit = kwargs.get(
129 "previous_searches_limit",
130 _get_setting(
131 settings_snapshot,
132 "focused_iteration.previous_searches_limit",
133 10,
134 ),
135 )
136 # Convert 0 to None for "unlimited"
137 if knowledge_limit == 0:
138 knowledge_limit = None
139 if snippet_truncate == 0:
140 snippet_truncate = None
141 if prompt_knowledge_truncate == 0:
142 prompt_knowledge_truncate = None
143 if previous_searches_limit == 0:
144 previous_searches_limit = None
146 strategy = FocusedIterationStrategy(
147 model=model,
148 search=search,
149 all_links_of_system=all_links_of_system,
150 max_iterations=kwargs.get("max_iterations", 8),
151 questions_per_iteration=kwargs.get("questions_per_iteration", 5),
152 settings_snapshot=settings_snapshot,
153 # Options read from settings (with kwargs override)
154 enable_adaptive_questions=enable_adaptive,
155 enable_early_termination=kwargs.get(
156 "enable_early_termination", False
157 ),
158 knowledge_summary_limit=knowledge_limit,
159 knowledge_snippet_truncate=snippet_truncate,
160 prompt_knowledge_truncate=prompt_knowledge_truncate,
161 previous_searches_limit=previous_searches_limit,
162 )
164 # Override question generator if flexible is selected
165 if question_gen_type == "flexible":
166 from .advanced_search_system.questions.flexible_browsecomp_question import (
167 FlexibleBrowseCompQuestionGenerator,
168 )
170 # Pass truncation settings to flexible generator
171 strategy.question_generator = FlexibleBrowseCompQuestionGenerator(
172 model,
173 knowledge_truncate_length=prompt_knowledge_truncate,
174 previous_searches_limit=previous_searches_limit,
175 )
177 return strategy
179 # Focused iteration strategy with standard citation handler
180 if strategy_name_lower in [
181 "focused-iteration-standard",
182 "focused_iteration_standard",
183 ]:
184 from .advanced_search_system.strategies.focused_iteration_strategy import (
185 FocusedIterationStrategy,
186 )
187 from .citation_handler import CitationHandler
189 # Use standard citation handler (same question generator as regular focused-iteration)
190 standard_citation_handler = CitationHandler(
191 model, handler_type="standard", settings_snapshot=settings_snapshot
192 )
194 # Read focused_iteration settings with kwargs override
195 # adaptive_questions is stored as 0/1 integer, convert to bool
196 enable_adaptive = bool(
197 kwargs.get(
198 "enable_adaptive_questions",
199 _get_setting(
200 settings_snapshot, "focused_iteration.adaptive_questions", 0
201 ),
202 )
203 )
204 knowledge_limit = kwargs.get(
205 "knowledge_summary_limit",
206 _get_setting(
207 settings_snapshot,
208 "focused_iteration.knowledge_summary_limit",
209 10,
210 ),
211 )
212 snippet_truncate = kwargs.get(
213 "knowledge_snippet_truncate",
214 _get_setting(
215 settings_snapshot, "focused_iteration.snippet_truncate", 200
216 ),
217 )
218 question_gen_type = kwargs.get(
219 "question_generator",
220 _get_setting(
221 settings_snapshot,
222 "focused_iteration.question_generator",
223 "browsecomp",
224 ),
225 )
226 prompt_knowledge_truncate = kwargs.get(
227 "prompt_knowledge_truncate",
228 _get_setting(
229 settings_snapshot,
230 "focused_iteration.prompt_knowledge_truncate",
231 1500,
232 ),
233 )
234 previous_searches_limit = kwargs.get(
235 "previous_searches_limit",
236 _get_setting(
237 settings_snapshot,
238 "focused_iteration.previous_searches_limit",
239 10,
240 ),
241 )
242 # Convert 0 to None for "unlimited"
243 if knowledge_limit == 0: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 knowledge_limit = None
245 if snippet_truncate == 0: 245 ↛ 246line 245 didn't jump to line 246 because the condition on line 245 was never true
246 snippet_truncate = None
247 if prompt_knowledge_truncate == 0: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
248 prompt_knowledge_truncate = None
249 if previous_searches_limit == 0: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 previous_searches_limit = None
252 strategy = FocusedIterationStrategy(
253 model=model,
254 search=search,
255 citation_handler=standard_citation_handler,
256 all_links_of_system=all_links_of_system,
257 max_iterations=kwargs.get("max_iterations", 8),
258 questions_per_iteration=kwargs.get("questions_per_iteration", 5),
259 use_browsecomp_optimization=True, # Keep BrowseComp features
260 settings_snapshot=settings_snapshot,
261 # Options read from settings (with kwargs override)
262 enable_adaptive_questions=enable_adaptive,
263 enable_early_termination=kwargs.get(
264 "enable_early_termination", False
265 ),
266 knowledge_summary_limit=knowledge_limit,
267 knowledge_snippet_truncate=snippet_truncate,
268 prompt_knowledge_truncate=prompt_knowledge_truncate,
269 previous_searches_limit=previous_searches_limit,
270 )
272 # Override question generator if flexible is selected
273 if question_gen_type == "flexible": 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true
274 from .advanced_search_system.questions.flexible_browsecomp_question import (
275 FlexibleBrowseCompQuestionGenerator,
276 )
278 # Pass truncation settings to flexible generator
279 strategy.question_generator = FlexibleBrowseCompQuestionGenerator(
280 model,
281 knowledge_truncate_length=prompt_knowledge_truncate,
282 previous_searches_limit=previous_searches_limit,
283 )
285 return strategy
287 # News aggregation strategy (used internally by the news subsystem)
288 if strategy_name_lower in [
289 "news",
290 "news_aggregation",
291 "news-aggregation",
292 ]:
293 from .advanced_search_system.strategies.news_strategy import (
294 NewsAggregationStrategy,
295 )
297 return NewsAggregationStrategy(
298 model=model,
299 search=search,
300 all_links_of_system=all_links_of_system,
301 )
303 # Topic organization strategy
304 if strategy_name_lower in [
305 "topic-organization",
306 "topic_organization",
307 "topic",
308 ]:
309 from .advanced_search_system.strategies.topic_organization_strategy import (
310 TopicOrganizationStrategy,
311 )
313 return TopicOrganizationStrategy(
314 model=model,
315 search=search,
316 all_links_of_system=all_links_of_system,
317 settings_snapshot=settings_snapshot,
318 min_sources_per_topic=1, # Allow single-source topics
319 use_cross_engine_filter=kwargs.get("use_cross_engine_filter", True),
320 filter_reorder=kwargs.get("filter_reorder", True),
321 filter_reindex=kwargs.get("filter_reindex", True),
322 cross_engine_max_results=kwargs.get( # type: ignore[arg-type]
323 "cross_engine_max_results", None
324 ),
325 search_original_query=kwargs.get("search_original_query", True),
326 max_topics=kwargs.get("max_topics", 5),
327 similarity_threshold=kwargs.get("similarity_threshold", 0.3),
328 use_focused_iteration=kwargs.get("use_focused_iteration", False),
329 enable_refinement=kwargs.get(
330 "enable_refinement", False
331 ), # Disable refinement iterations for now
332 max_refinement_iterations=kwargs.get(
333 "max_refinement_iterations",
334 1, # Set to 1 iteration for faster results
335 ),
336 generate_text=kwargs.get("generate_text", True),
337 )
339 # LangGraph agent strategy (parallel subagent research).
340 # ``mcp`` / ``agentic`` were removed (#4548); they remain here as
341 # deprecated aliases so existing saved settings, queued runs, and API
342 # callers route to the closest successor (langgraph-agent) instead of
343 # the source-based fallback below.
344 if strategy_name_lower in [
345 "langgraph-agent",
346 "langgraph_agent",
347 "mcp",
348 "agentic",
349 ]:
350 if strategy_name_lower in ("mcp", "agentic"):
351 logger.warning(
352 f"Strategy {strategy_name!r} was removed (#4548); "
353 "using 'langgraph-agent' instead."
354 )
355 from .advanced_search_system.strategies.langgraph_agent_strategy import (
356 LangGraphAgentStrategy,
357 )
359 return LangGraphAgentStrategy(
360 model=model,
361 search=search,
362 max_iterations=kwargs.get(
363 "max_iterations",
364 _get_setting(
365 settings_snapshot, "langgraph_agent.max_iterations", 50
366 ),
367 ),
368 max_sub_iterations=kwargs.get(
369 "max_sub_iterations",
370 _get_setting(
371 settings_snapshot, "langgraph_agent.max_sub_iterations", 8
372 ),
373 ),
374 include_sub_research=kwargs.get(
375 "include_sub_research",
376 _get_setting(
377 settings_snapshot,
378 "langgraph_agent.include_sub_research",
379 True,
380 ),
381 ),
382 programmatic_mode=kwargs.get("programmatic_mode", False),
383 all_links_of_system=all_links_of_system,
384 settings_snapshot=settings_snapshot,
385 )
387 # Default to source-based if unknown
388 logger.warning(
389 f"Unknown strategy: {strategy_name}, defaulting to source-based"
390 )
391 from .advanced_search_system.strategies.source_based_strategy import (
392 SourceBasedSearchStrategy,
393 )
395 return SourceBasedSearchStrategy(
396 model=model,
397 search=search,
398 include_text_content=kwargs.get("include_text_content", True),
399 use_cross_engine_filter=True,
400 all_links_of_system=all_links_of_system,
401 use_atomic_facts=False,
402 settings_snapshot=settings_snapshot,
403 search_original_query=kwargs.get("search_original_query", True),
404 )