Coverage for src/local_deep_research/api/research_functions.py: 93%
234 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2API module for Local Deep Research.
3Provides programmatic access to search and research capabilities.
4"""
6from datetime import datetime, UTC
7from typing import Any, Callable
9from loguru import logger
10from local_deep_research.settings.logger import log_settings
12from ..config.llm_config import get_llm
13from ..config.search_config import get_search
14from ..config.thread_settings import get_setting_from_snapshot
15from ..report_generator import IntegratedReportGenerator
16from ..search_system import AdvancedSearchSystem
17from ..utilities.db_utils import no_db_settings
18from ..utilities.thread_context import clear_search_context, set_search_context
19from ..utilities.type_utils import overlay_runtime_settings
20from .settings_utils import create_settings_snapshot
21from ..utilities.llm_utils import invoke_llm_sync
22from ..security.log_sanitizer import redact_and_bound_for_log
25def _close_system(system):
26 """Close an AdvancedSearchSystem and its associated resources."""
27 from ..utilities.resource_utils import safe_close
29 safe_close(system, "search system")
30 if hasattr(system, "search"): 30 ↛ 32line 30 didn't jump to line 32 because the condition on line 30 was always true
31 safe_close(system.search, "search engine")
32 if hasattr(system, "model"): 32 ↛ exitline 32 didn't return from function '_close_system' because the condition on line 32 was always true
33 safe_close(system.model, "system LLM")
36def _init_search_system(
37 model_name: str | None = None,
38 temperature: float = 0.7,
39 provider: str | None = None,
40 openai_endpoint_url: str | None = None,
41 progress_callback: Callable[[str, int, dict], None] | None = None,
42 search_tool: str | None = None,
43 search_strategy: str | None = None,
44 iterations: int | None = None,
45 questions_per_iteration: int | None = None,
46 retrievers: dict[str, Any] | None = None,
47 llms: dict[str, Any] | None = None,
48 username: str | None = None,
49 research_id: str | None = None,
50 research_context: dict[str, Any] | None = None,
51 programmatic_mode: bool = True,
52 search_original_query: bool = True,
53 settings_snapshot: dict[str, Any] | None = None,
54 **kwargs: Any,
55) -> AdvancedSearchSystem:
56 """
57 Initializes the advanced search system with specified parameters. This function sets up
58 and returns an instance of the AdvancedSearchSystem using the provided configuration
59 options such as model name, temperature for randomness in responses, provider service
60 details, endpoint URL, and an optional search tool.
62 Args:
63 model_name: Name of the model to use (if None, uses database setting)
64 temperature: LLM temperature for generation
65 provider: Provider to use (if None, uses database setting)
66 openai_endpoint_url: Custom endpoint URL to use (if None, uses database
67 setting)
68 progress_callback: Optional callback function to receive progress updates
69 search_tool: Search engine to use (searxng, wikipedia, arxiv, etc.). If None, uses snapshot/default
70 search_strategy: Search strategy to use. If None, uses snapshot or source-based
71 iterations: Number of research cycles to perform. If None, uses snapshot/default
72 questions_per_iteration: Number of questions to generate per cycle. If None, uses snapshot/default
73 retrievers: Optional dictionary of {name: retriever} pairs to use as search engines
74 llms: Optional dictionary of {name: llm} pairs to use as language models
75 programmatic_mode: If True, disables database operations and metrics tracking
76 search_original_query: Whether to include the original query in the first iteration of search
78 Returns:
79 AdvancedSearchSystem: An instance of the configured AdvancedSearchSystem.
81 """
82 # Register retrievers if provided
83 if retrievers:
84 from ..web_search_engines.retriever_registry import retriever_registry
86 retriever_registry.register_multiple(retrievers, username=username)
87 logger.info(
88 f"Registered {len(retrievers)} retrievers: {list(retrievers.keys())}"
89 )
91 # Register LLMs if provided
92 if llms:
93 from ..llm import register_llm
95 for name, llm_instance in llms.items():
96 register_llm(name, llm_instance, username=username)
97 logger.info(f"Registered {len(llms)} LLMs: {list(llms.keys())}")
99 # Use settings_snapshot from parameter, or fall back to kwargs
100 if settings_snapshot is None:
101 settings_snapshot = kwargs.get("settings_snapshot")
103 if settings_snapshot is not None:
104 settings_snapshot = dict(settings_snapshot)
105 overlay_runtime_settings(
106 settings_snapshot,
107 search_tool=search_tool,
108 model_provider=provider,
109 model=model_name,
110 custom_endpoint=openai_endpoint_url,
111 iterations=iterations,
112 questions_per_iteration=questions_per_iteration,
113 search_strategy=search_strategy,
114 )
116 if search_strategy is None and settings_snapshot:
117 search_strategy = get_setting_from_snapshot(
118 "search.search_strategy",
119 default=None,
120 settings_snapshot=settings_snapshot,
121 )
123 # Get language model with custom temperature
124 llm = get_llm(
125 temperature=temperature,
126 openai_endpoint_url=openai_endpoint_url,
127 model_name=model_name,
128 provider=provider,
129 research_id=research_id,
130 research_context=research_context,
131 settings_snapshot=settings_snapshot,
132 username=username,
133 )
135 # Set the search engine if specified or get from settings
136 search_engine = None
138 try:
139 # If no search_tool provided, get from settings_snapshot
140 if not search_tool and settings_snapshot:
141 search_tool = get_setting_from_snapshot(
142 "search.tool", default=None, settings_snapshot=settings_snapshot
143 )
145 if search_tool:
146 search_engine = get_search(
147 search_tool,
148 llm_instance=llm,
149 username=username,
150 settings_snapshot=settings_snapshot,
151 programmatic_mode=programmatic_mode,
152 )
153 if search_engine is None: 153 ↛ 154line 153 didn't jump to line 154 because the condition on line 153 was never true
154 logger.warning(
155 "Could not create search engine '{}', using default.",
156 redact_and_bound_for_log(str(search_tool), 200),
157 )
159 # Create search system with custom parameters
160 logger.info(
161 "Search strategy: {}",
162 redact_and_bound_for_log(str(search_strategy), 200),
163 )
164 system = AdvancedSearchSystem(
165 llm=llm,
166 search=search_engine,
167 strategy_name=search_strategy or "source-based",
168 username=username,
169 research_id=research_id,
170 research_context=research_context,
171 settings_snapshot=settings_snapshot,
172 programmatic_mode=programmatic_mode,
173 search_original_query=search_original_query,
174 max_iterations=iterations,
175 questions_per_iteration=questions_per_iteration,
176 )
177 except Exception:
178 from ..utilities.resource_utils import safe_close
180 safe_close(llm, "init LLM")
181 raise
183 # Set progress callback if provided
184 if progress_callback:
185 system.set_progress_callback(progress_callback)
187 return system
190@no_db_settings
191def quick_summary(
192 query: str,
193 research_id: str | None = None,
194 retrievers: dict[str, Any] | None = None,
195 llms: dict[str, Any] | None = None,
196 username: str | None = None,
197 provider: str | None = None,
198 api_key: str | None = None,
199 temperature: float | None = None,
200 max_search_results: int | None = None,
201 settings: dict[str, Any] | None = None,
202 settings_override: dict[str, Any] | None = None,
203 search_original_query: bool = True,
204 **kwargs: Any,
205) -> dict[str, Any]:
206 """
207 Generate a quick research summary for a given query.
209 Args:
210 query: The research query to analyze
211 research_id: Optional research ID (int or UUID string) for tracking metrics
212 retrievers: Optional dictionary of {name: retriever} pairs to use as search engines
213 llms: Optional dictionary of {name: llm} pairs to use as language models
214 provider: LLM provider to use (e.g., 'openai', 'anthropic'). For programmatic API only.
215 api_key: API key for the provider. For programmatic API only.
216 temperature: LLM temperature (0.0-1.0). For programmatic API only.
217 max_search_results: Maximum number of search results to return. For programmatic API only.
218 settings: Base settings dict to use instead of defaults. For programmatic API only.
219 settings_override: Dictionary of settings to override (e.g., {"llm.max_tokens": 4000}). For programmatic API only.
220 search_original_query: Whether to include the original query in the first iteration of search.
221 Set to False for news searches to avoid sending long subscription prompts to search engines.
222 **kwargs: Additional configuration for the search system. Will be forwarded to
223 `_init_search_system()`.
225 Returns:
226 Dictionary containing the research results with keys:
227 - 'summary': The generated summary text
228 - 'findings': List of detailed findings from each search
229 - 'iterations': Number of iterations performed
230 - 'questions': Questions generated during research
232 Examples:
233 # Simple usage with defaults
234 result = quick_summary("What is quantum computing?")
236 # With custom provider
237 result = quick_summary(
238 "What is quantum computing?",
239 provider="anthropic",
240 api_key="your-api-key-here"
241 )
243 # With advanced settings
244 result = quick_summary(
245 "What is quantum computing?",
246 temperature=0.2,
247 settings_override={"search.engines.arxiv.enabled": True}
248 )
249 """
250 # (#6938: the /api/v1 routes pass the body's query unbounded.)
251 logger.info(
252 "Generating quick summary for query: {}",
253 redact_and_bound_for_log(str(query), 200),
254 )
256 if "settings_snapshot" not in kwargs:
257 snapshot_kwargs = {}
258 if provider is not None:
259 snapshot_kwargs["provider"] = provider
260 if api_key is not None:
261 snapshot_kwargs["api_key"] = api_key
262 if temperature is not None:
263 snapshot_kwargs["temperature"] = temperature
264 if max_search_results is not None:
265 snapshot_kwargs["max_search_results"] = max_search_results
267 if (
268 not snapshot_kwargs
269 and settings is None
270 and settings_override is None
271 ):
272 logger.warning(
273 "No settings_snapshot or explicit config provided to quick_summary(). "
274 "Using defaults and environment variables. For explicit control, "
275 "pass settings_snapshot=create_settings_snapshot(...)."
276 )
278 kwargs["settings_snapshot"] = create_settings_snapshot(
279 base_settings=settings,
280 overrides=settings_override,
281 **snapshot_kwargs,
282 )
283 log_settings(
284 kwargs["settings_snapshot"],
285 "Created settings snapshot for programmatic API",
286 )
287 else:
288 log_settings(
289 kwargs["settings_snapshot"],
290 "Using provided settings snapshot for programmatic API",
291 )
293 # Generate a research_id if none provided
294 if research_id is None:
295 import uuid
297 research_id = str(uuid.uuid4())
298 logger.debug(f"Generated research_id: {research_id}")
300 # Register retrievers if provided
301 if retrievers:
302 from ..web_search_engines.retriever_registry import retriever_registry
304 retriever_registry.register_multiple(retrievers, username=username)
305 logger.info(
306 f"Registered {len(retrievers)} retrievers: {list(retrievers.keys())}"
307 )
309 # Register LLMs if provided
310 if llms:
311 from ..llm import register_llm
313 for name, llm_instance in llms.items():
314 register_llm(name, llm_instance, username=username)
315 logger.info(f"Registered {len(llms)} LLMs: {list(llms.keys())}")
317 search_context = {
318 "research_id": research_id, # Pass UUID or integer directly
319 "research_query": query,
320 "research_mode": kwargs.get("research_mode", "quick"),
321 "research_phase": "init",
322 "search_iteration": 0,
323 "search_engine_selected": kwargs.get("search_tool"),
324 "username": username, # Include username for metrics tracking
325 "user_password": kwargs.get(
326 "user_password"
327 ), # Include password for metrics tracking
328 # Thread-safe settings snapshot propagated to background search
329 # threads (engine config, per-user resolution, egress scope).
330 "settings_snapshot": kwargs.get("settings_snapshot") or {},
331 }
332 set_search_context(search_context)
334 system = None
335 try:
336 # Remove research_mode from kwargs before passing to _init_search_system
337 init_kwargs = {k: v for k, v in kwargs.items() if k != "research_mode"}
338 # Make sure username is passed to the system
339 init_kwargs["username"] = username
340 init_kwargs["research_id"] = research_id
341 init_kwargs["research_context"] = search_context
342 init_kwargs["search_original_query"] = search_original_query
343 system = _init_search_system(llms=llms, **init_kwargs)
345 # Perform the search and analysis
346 results = system.analyze_topic(query)
348 # Extract the summary from the current knowledge
349 if results and "current_knowledge" in results:
350 summary = results["current_knowledge"]
351 else:
352 summary = "Unable to generate summary for the query."
354 # Prepare the return value (guard against None results)
355 if results is None: 355 ↛ 356line 355 didn't jump to line 356 because the condition on line 355 was never true
356 results = {}
357 return {
358 "research_id": research_id,
359 "summary": summary,
360 "findings": results.get("findings", []),
361 "iterations": results.get("iterations", 0),
362 "questions": results.get("questions", {}),
363 "formatted_findings": results.get("formatted_findings", ""),
364 # Render-only by design: report.uncited_sources_mode shapes the
365 # rendered ## Sources block, not this array, which keeps the
366 # full accumulated set for API consumers.
367 "sources": results.get("all_links_of_system", []),
368 }
369 finally:
370 if system is not None:
371 _close_system(system)
372 clear_search_context()
375@no_db_settings
376def generate_report(
377 query: str,
378 output_file: str | None = None,
379 progress_callback: Callable | None = None,
380 searches_per_section: int = 2,
381 retrievers: dict[str, Any] | None = None,
382 llms: dict[str, Any] | None = None,
383 username: str | None = None,
384 provider: str | None = None,
385 api_key: str | None = None,
386 temperature: float | None = None,
387 max_search_results: int | None = None,
388 settings: dict[str, Any] | None = None,
389 settings_override: dict[str, Any] | None = None,
390 **kwargs: Any,
391) -> dict[str, Any]:
392 """
393 Generate a comprehensive, structured research report for a given query.
395 Args:
396 query: The research query to analyze
397 output_file: Optional path to save report markdown file
398 progress_callback: Optional callback function to receive progress updates
399 searches_per_section: The number of searches to perform for each
400 section in the report.
401 retrievers: Optional dictionary of {name: retriever} pairs to use as search engines
402 llms: Optional dictionary of {name: llm} pairs to use as language models
403 provider: LLM provider to use (e.g., 'openai', 'anthropic'). For programmatic API only.
404 api_key: API key for the provider. For programmatic API only.
405 temperature: LLM temperature (0.0-1.0). For programmatic API only.
406 max_search_results: Maximum number of search results to return. For programmatic API only.
407 settings: Base settings dict to use instead of defaults. For programmatic API only.
408 settings_override: Dictionary of settings to override. For programmatic API only.
409 **kwargs: Additional configuration for the search system.
411 Returns:
412 Dictionary containing the research report with keys:
413 - 'content': The full report content in markdown format
414 - 'metadata': Report metadata including generated timestamp and query
415 - 'file_path': Path to saved file (if output_file was provided)
417 Examples:
418 # Simple usage with settings snapshot
419 from local_deep_research.api.settings_utils import create_settings_snapshot
420 settings = create_settings_snapshot({"programmatic_mode": True})
421 result = generate_report("AI research", settings_snapshot=settings)
423 # Save to file
424 result = generate_report(
425 "AI research",
426 output_file="report.md",
427 settings_snapshot=settings
428 )
429 """
430 logger.info(
431 "Generating comprehensive research report for query: {}",
432 redact_and_bound_for_log(str(query), 200),
433 )
435 if "settings_snapshot" not in kwargs:
436 snapshot_kwargs = {}
437 if provider is not None:
438 snapshot_kwargs["provider"] = provider
439 if api_key is not None: 439 ↛ 440line 439 didn't jump to line 440 because the condition on line 439 was never true
440 snapshot_kwargs["api_key"] = api_key
441 if temperature is not None: 441 ↛ 442line 441 didn't jump to line 442 because the condition on line 441 was never true
442 snapshot_kwargs["temperature"] = temperature
443 if max_search_results is not None: 443 ↛ 444line 443 didn't jump to line 444 because the condition on line 443 was never true
444 snapshot_kwargs["max_search_results"] = max_search_results
446 if (
447 not snapshot_kwargs
448 and settings is None
449 and settings_override is None
450 ):
451 logger.warning(
452 "No settings_snapshot or explicit config provided to generate_report(). "
453 "Using defaults and environment variables. For explicit control, "
454 "pass settings_snapshot=create_settings_snapshot(...)."
455 )
457 kwargs["settings_snapshot"] = create_settings_snapshot(
458 base_settings=settings,
459 overrides=settings_override,
460 **snapshot_kwargs,
461 )
462 log_settings(
463 kwargs["settings_snapshot"],
464 "Created settings snapshot for programmatic API",
465 )
466 else:
467 log_settings(
468 kwargs["settings_snapshot"],
469 "Using provided settings snapshot for programmatic API",
470 )
472 # Register retrievers if provided
473 if retrievers:
474 from ..web_search_engines.retriever_registry import retriever_registry
476 retriever_registry.register_multiple(retrievers, username=username)
477 logger.info(
478 f"Registered {len(retrievers)} retrievers: {list(retrievers.keys())}"
479 )
481 # Register LLMs if provided
482 if llms:
483 from ..llm import register_llm
485 for name, llm_instance in llms.items():
486 register_llm(name, llm_instance, username=username)
487 logger.info(f"Registered {len(llms)} LLMs: {list(llms.keys())}")
489 import uuid
491 search_context = {
492 "research_id": str(uuid.uuid4()),
493 "research_query": query,
494 "research_mode": "report",
495 "research_phase": "init",
496 "search_iteration": 0,
497 "search_engine_selected": kwargs.get("search_tool"),
498 "username": username,
499 "user_password": kwargs.get("user_password"),
500 "settings_snapshot": kwargs.get("settings_snapshot") or {},
501 }
502 set_search_context(search_context)
504 system = None
505 try:
506 system = _init_search_system(
507 retrievers=retrievers, llms=llms, username=username, **kwargs
508 )
509 # Set progress callback if provided
510 if progress_callback:
511 system.set_progress_callback(progress_callback)
513 # Perform the initial research
514 initial_findings = system.analyze_topic(query)
516 # Generate the structured report
517 report_generator = IntegratedReportGenerator(
518 search_system=system,
519 llm=system.model,
520 searches_per_section=searches_per_section,
521 settings_snapshot=kwargs.get("settings_snapshot"),
522 )
523 report = report_generator.generate_report(initial_findings, query)
525 # Save report to file if path is provided
526 if output_file and report and "content" in report:
527 from ..security.file_write_verifier import write_file_verified
529 write_file_verified(
530 output_file,
531 report["content"],
532 "api.allow_file_output",
533 context="API research report",
534 settings_snapshot=kwargs.get("settings_snapshot"),
535 )
536 logger.info(f"Report saved to {output_file}")
537 report["file_path"] = output_file
538 return report
539 finally:
540 if system is not None: 540 ↛ 542line 540 didn't jump to line 542 because the condition on line 540 was always true
541 _close_system(system)
542 clear_search_context()
545@no_db_settings
546def detailed_research(
547 query: str,
548 research_id: str | None = None,
549 retrievers: dict[str, Any] | None = None,
550 llms: dict[str, Any] | None = None,
551 username: str | None = None,
552 **kwargs: Any,
553) -> dict[str, Any]:
554 """
555 Perform detailed research with comprehensive analysis.
557 Similar to generate_report but returns structured data instead of markdown.
559 Args:
560 query: The research query to analyze
561 research_id: Optional research ID (int or UUID string) for tracking metrics
562 retrievers: Optional dictionary of {name: retriever} pairs to use as search engines
563 llms: Optional dictionary of {name: llm} pairs to use as language models
564 username: Optional username for per-user cache isolation
565 **kwargs: Configuration for the search system. Pass settings_snapshot
566 (via create_settings_snapshot()) to configure provider, temperature, etc.
568 Returns:
569 Dictionary containing detailed research results
570 """
571 logger.info("Performing detailed research for query: {}", query)
573 if "settings_snapshot" not in kwargs:
574 logger.warning(
575 "No settings_snapshot provided to detailed_research(). "
576 "Using defaults and environment variables. For explicit control, "
577 "pass settings_snapshot=create_settings_snapshot(provider=..., "
578 "overrides={'search.tool': ...})."
579 )
580 kwargs["settings_snapshot"] = create_settings_snapshot()
582 # Generate a research_id if none provided
583 if research_id is None:
584 import uuid
586 research_id = str(uuid.uuid4())
587 logger.debug(f"Generated research_id: {research_id}")
589 # Register retrievers if provided
590 if retrievers:
591 from ..web_search_engines.retriever_registry import retriever_registry
593 retriever_registry.register_multiple(retrievers, username=username)
594 logger.info(
595 f"Registered {len(retrievers)} retrievers: {list(retrievers.keys())}"
596 )
598 # Register LLMs if provided
599 if llms:
600 from ..llm import register_llm
602 for name, llm_instance in llms.items():
603 register_llm(name, llm_instance, username=username)
604 logger.info(f"Registered {len(llms)} LLMs: {list(llms.keys())}")
606 search_context = {
607 "research_id": research_id,
608 "research_query": query,
609 "research_mode": "detailed",
610 "research_phase": "init",
611 "search_iteration": 0,
612 "search_engine_selected": kwargs.get("search_tool"),
613 "username": username,
614 "user_password": kwargs.get("user_password"),
615 "settings_snapshot": kwargs.get("settings_snapshot") or {},
616 }
617 set_search_context(search_context)
619 system = None
620 try:
621 # Initialize system
622 system = _init_search_system(
623 retrievers=retrievers, llms=llms, username=username, **kwargs
624 )
626 # Perform detailed research
627 results = system.analyze_topic(query)
629 # Return comprehensive results (guard against None results)
630 if results is None: 630 ↛ 631line 630 didn't jump to line 631 because the condition on line 630 was never true
631 results = {}
632 return {
633 "query": query,
634 "research_id": research_id,
635 "summary": results.get("current_knowledge", ""),
636 "findings": results.get("findings", []),
637 "iterations": results.get("iterations", 0),
638 "questions": results.get("questions", {}),
639 "formatted_findings": results.get("formatted_findings", ""),
640 # Render-only by design (see above): the uncited-sources mode
641 # shapes the rendered ## Sources block; this array keeps the
642 # full accumulated set.
643 "sources": results.get("all_links_of_system", []),
644 "metadata": {
645 "timestamp": datetime.now(UTC).isoformat(),
646 "search_tool": kwargs.get("search_tool", "searxng"),
647 "iterations_requested": kwargs.get("iterations", 1),
648 "strategy": kwargs.get("search_strategy", "source_based"),
649 },
650 }
651 finally:
652 if system is not None: 652 ↛ 654line 652 didn't jump to line 654 because the condition on line 652 was always true
653 _close_system(system)
654 clear_search_context()
657@no_db_settings
658def analyze_documents(
659 query: str,
660 collection_name: str,
661 max_results: int = 10,
662 temperature: float = 0.7,
663 force_reindex: bool = False,
664 output_file: str | None = None,
665 *,
666 username: str | None = None,
667 settings_snapshot: dict[str, Any] | None = None,
668 programmatic_mode: bool = True,
669) -> dict[str, Any]:
670 """
671 Search and analyze documents in a specific local collection.
673 Args:
674 query: The search query
675 collection_name: Name of the local document collection to search
676 max_results: Maximum number of results to return
677 temperature: LLM temperature for summary generation
678 force_reindex: Whether to force reindexing the collection
679 output_file: Optional path to save analysis results to a file
680 username: Optional username for thread context. REST callers pass the
681 authenticated user; programmatic SDK callers can omit.
682 settings_snapshot: Settings snapshot for the user's stored configuration
683 (LLM provider/model, embedding model, etc.). REST callers pass the
684 snapshot from the user's encrypted DB; SDK callers can omit to use
685 JSON defaults + LDR_* env vars.
686 programmatic_mode: If True (default for SDK callers), disables DB-backed
687 metrics. REST callers pass False so per-user rate-limit estimates
688 persist across requests.
690 Returns:
691 Dictionary containing:
692 - 'summary': Summary of the findings
693 - 'documents': List of matching documents with content and metadata
694 """
695 if settings_snapshot is None:
696 settings_snapshot = create_settings_snapshot()
698 logger.info(
699 "Analyzing documents in collection '{}' for query: {}",
700 redact_and_bound_for_log(str(collection_name), 200),
701 redact_and_bound_for_log(str(query), 200),
702 )
704 llm = None
705 search = None
706 try:
707 # Get language model with custom temperature. Pass username so a
708 # per-user registered LLM (and per-user egress classification)
709 # resolves for this caller.
710 llm = get_llm(
711 temperature=temperature,
712 settings_snapshot=settings_snapshot,
713 username=username,
714 )
716 # Get search engine for the specified collection
717 search = get_search(
718 collection_name,
719 llm_instance=llm,
720 username=username,
721 settings_snapshot=settings_snapshot,
722 programmatic_mode=programmatic_mode,
723 )
725 if not search:
726 from ..utilities.resource_utils import safe_close
728 safe_close(llm, "LLM")
729 llm = None
730 return {
731 "summary": f"Error: Collection '{collection_name}' not found or not properly configured.",
732 "documents": [],
733 }
735 # Set max results
736 search.max_results = max_results
737 # Perform the search
738 results = search.run(query)
740 if not results:
741 return {
742 "summary": f"No documents found in collection '{collection_name}' for query: '{query}'",
743 "documents": [],
744 }
746 # Get LLM to generate a summary of the results
748 docs_text = "\n\n".join(
749 [
750 f"Document {i + 1}:"
751 f" {doc.get('content', doc.get('snippet', ''))[:1000]}"
752 for i, doc in enumerate(results[:5])
753 ]
754 ) # Limit to first 5 docs and 1000 chars each
756 summary_prompt = f"""Analyze these document excerpts related to the query: "{query}"
758 {docs_text}
760 Provide a concise summary of the key information found in these documents related to the query.
761 """
763 import time
765 llm_start_time = time.time()
766 logger.info(
767 f"Starting LLM summary generation (prompt length: {len(summary_prompt)} chars)..."
768 )
770 summary_response = invoke_llm_sync(llm, summary_prompt)
772 llm_elapsed = time.time() - llm_start_time
773 logger.info(f"LLM summary generation completed in {llm_elapsed:.2f}s")
775 if hasattr(summary_response, "content"): 775 ↛ 778line 775 didn't jump to line 778 because the condition on line 775 was always true
776 summary = summary_response.content
777 else:
778 summary = str(summary_response)
780 # Create result dictionary
781 analysis_result = {
782 "summary": summary,
783 "documents": results,
784 "collection": collection_name,
785 "document_count": len(results),
786 }
788 # Save to file if requested
789 if output_file:
790 from ..security.file_write_verifier import write_file_verified
792 content = f"# Document Analysis: {query}\n\n"
793 content += f"## Summary\n\n{summary}\n\n"
794 content += f"## Documents Found: {len(results)}\n\n"
796 for i, doc in enumerate(results):
797 content += (
798 f"### Document {i + 1}: {doc.get('title', 'Untitled')}\n\n"
799 )
800 content += f"**Source:** {doc.get('link', 'Unknown')}\n\n"
801 content += f"**Content:**\n\n{doc.get('content', doc.get('snippet', 'No content available'))[:1000]}...\n\n"
802 content += "---\n\n"
804 write_file_verified(
805 output_file,
806 content,
807 "api.allow_file_output",
808 context="API document analysis",
809 settings_snapshot=settings_snapshot,
810 )
812 analysis_result["file_path"] = output_file
813 logger.info(f"Analysis saved to {output_file}")
815 return analysis_result
816 finally:
817 from ..utilities.resource_utils import safe_close
819 safe_close(search, "search engine")
820 safe_close(llm, "LLM")