Coverage for src/local_deep_research/api/research_functions.py: 93%
228 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +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 .settings_utils import create_settings_snapshot
22def _close_system(system):
23 """Close an AdvancedSearchSystem and its associated resources."""
24 from ..utilities.resource_utils import safe_close
26 safe_close(system, "search system")
27 if hasattr(system, "search"): 27 ↛ 29line 27 didn't jump to line 29 because the condition on line 27 was always true
28 safe_close(system.search, "search engine")
29 if hasattr(system, "model"): 29 ↛ exitline 29 didn't return from function '_close_system' because the condition on line 29 was always true
30 safe_close(system.model, "system LLM")
33def _init_search_system(
34 model_name: str | None = None,
35 temperature: float = 0.7,
36 provider: str | None = None,
37 openai_endpoint_url: str | None = None,
38 progress_callback: Callable[[str, int, dict], None] | None = None,
39 search_tool: str | None = None,
40 search_strategy: str = "source_based",
41 iterations: int = 1,
42 questions_per_iteration: int = 1,
43 retrievers: dict[str, Any] | None = None,
44 llms: dict[str, Any] | None = None,
45 username: str | None = None,
46 research_id: str | None = None,
47 research_context: dict[str, Any] | None = None,
48 programmatic_mode: bool = True,
49 search_original_query: bool = True,
50 settings_snapshot: dict[str, Any] | None = None,
51 **kwargs: Any,
52) -> AdvancedSearchSystem:
53 """
54 Initializes the advanced search system with specified parameters. This function sets up
55 and returns an instance of the AdvancedSearchSystem using the provided configuration
56 options such as model name, temperature for randomness in responses, provider service
57 details, endpoint URL, and an optional search tool.
59 Args:
60 model_name: Name of the model to use (if None, uses database setting)
61 temperature: LLM temperature for generation
62 provider: Provider to use (if None, uses database setting)
63 openai_endpoint_url: Custom endpoint URL to use (if None, uses database
64 setting)
65 progress_callback: Optional callback function to receive progress updates
66 search_tool: Search engine to use (searxng, wikipedia, arxiv, etc.). If None, uses default
67 search_strategy: Search strategy to use (modular, source_based, etc.). If None, uses default
68 iterations: Number of research cycles to perform
69 questions_per_iteration: Number of questions to generate per cycle
70 search_strategy: The name of the search strategy to use.
71 retrievers: Optional dictionary of {name: retriever} pairs to use as search engines
72 llms: Optional dictionary of {name: llm} pairs to use as language models
73 programmatic_mode: If True, disables database operations and metrics tracking
74 search_original_query: Whether to include the original query in the first iteration of search
76 Returns:
77 AdvancedSearchSystem: An instance of the configured AdvancedSearchSystem.
79 """
80 # Register retrievers if provided
81 if retrievers:
82 from ..web_search_engines.retriever_registry import retriever_registry
84 retriever_registry.register_multiple(retrievers)
85 logger.info(
86 f"Registered {len(retrievers)} retrievers: {list(retrievers.keys())}"
87 )
89 # Register LLMs if provided
90 if llms:
91 from ..llm import register_llm
93 for name, llm_instance in llms.items():
94 register_llm(name, llm_instance)
95 logger.info(f"Registered {len(llms)} LLMs: {list(llms.keys())}")
97 # Use settings_snapshot from parameter, or fall back to kwargs
98 if settings_snapshot is None:
99 settings_snapshot = kwargs.get("settings_snapshot")
101 # Get language model with custom temperature
102 llm = get_llm(
103 temperature=temperature,
104 openai_endpoint_url=openai_endpoint_url,
105 model_name=model_name,
106 provider=provider,
107 research_id=research_id,
108 research_context=research_context,
109 settings_snapshot=settings_snapshot,
110 )
112 # Set the search engine if specified or get from settings
113 search_engine = None
115 try:
116 # If no search_tool provided, get from settings_snapshot
117 if not search_tool and settings_snapshot:
118 search_tool = get_setting_from_snapshot(
119 "search.tool", settings_snapshot=settings_snapshot
120 )
122 if search_tool:
123 search_engine = get_search(
124 search_tool,
125 llm_instance=llm,
126 username=username,
127 settings_snapshot=settings_snapshot,
128 programmatic_mode=programmatic_mode,
129 )
130 if search_engine is None: 130 ↛ 131line 130 didn't jump to line 131 because the condition on line 130 was never true
131 logger.warning(
132 f"Could not create search engine '{search_tool}', using default."
133 )
135 # Create search system with custom parameters
136 logger.info("Search strategy: {}", search_strategy)
137 system = AdvancedSearchSystem(
138 llm=llm,
139 search=search_engine,
140 strategy_name=search_strategy,
141 username=username,
142 research_id=research_id,
143 research_context=research_context,
144 settings_snapshot=settings_snapshot,
145 programmatic_mode=programmatic_mode,
146 search_original_query=search_original_query,
147 )
148 except Exception:
149 from ..utilities.resource_utils import safe_close
151 safe_close(llm, "init LLM")
152 raise
154 # Override default settings with user-provided values
155 system.max_iterations = iterations
156 system.questions_per_iteration = questions_per_iteration
158 # Set progress callback if provided
159 if progress_callback:
160 system.set_progress_callback(progress_callback)
162 return system
165@no_db_settings
166def quick_summary(
167 query: str,
168 research_id: str | None = None,
169 retrievers: dict[str, Any] | None = None,
170 llms: dict[str, Any] | None = None,
171 username: str | None = None,
172 provider: str | None = None,
173 api_key: str | None = None,
174 temperature: float | None = None,
175 max_search_results: int | None = None,
176 settings: dict[str, Any] | None = None,
177 settings_override: dict[str, Any] | None = None,
178 search_original_query: bool = True,
179 **kwargs: Any,
180) -> dict[str, Any]:
181 """
182 Generate a quick research summary for a given query.
184 Args:
185 query: The research query to analyze
186 research_id: Optional research ID (int or UUID string) for tracking metrics
187 retrievers: Optional dictionary of {name: retriever} pairs to use as search engines
188 llms: Optional dictionary of {name: llm} pairs to use as language models
189 provider: LLM provider to use (e.g., 'openai', 'anthropic'). For programmatic API only.
190 api_key: API key for the provider. For programmatic API only.
191 temperature: LLM temperature (0.0-1.0). For programmatic API only.
192 max_search_results: Maximum number of search results to return. For programmatic API only.
193 settings: Base settings dict to use instead of defaults. For programmatic API only.
194 settings_override: Dictionary of settings to override (e.g., {"llm.max_tokens": 4000}). For programmatic API only.
195 search_original_query: Whether to include the original query in the first iteration of search.
196 Set to False for news searches to avoid sending long subscription prompts to search engines.
197 **kwargs: Additional configuration for the search system. Will be forwarded to
198 `_init_search_system()`.
200 Returns:
201 Dictionary containing the research results with keys:
202 - 'summary': The generated summary text
203 - 'findings': List of detailed findings from each search
204 - 'iterations': Number of iterations performed
205 - 'questions': Questions generated during research
207 Examples:
208 # Simple usage with defaults
209 result = quick_summary("What is quantum computing?")
211 # With custom provider
212 result = quick_summary(
213 "What is quantum computing?",
214 provider="anthropic",
215 api_key="sk-ant-..."
216 )
218 # With advanced settings
219 result = quick_summary(
220 "What is quantum computing?",
221 temperature=0.2,
222 settings_override={"search.engines.arxiv.enabled": True}
223 )
224 """
225 logger.info("Generating quick summary for query: {}", query)
227 if "settings_snapshot" not in kwargs:
228 snapshot_kwargs = {}
229 if provider is not None:
230 snapshot_kwargs["provider"] = provider
231 if api_key is not None:
232 snapshot_kwargs["api_key"] = api_key
233 if temperature is not None:
234 snapshot_kwargs["temperature"] = temperature
235 if max_search_results is not None:
236 snapshot_kwargs["max_search_results"] = max_search_results
238 if (
239 not snapshot_kwargs
240 and settings is None
241 and settings_override is None
242 ):
243 logger.warning(
244 "No settings_snapshot or explicit config provided to quick_summary(). "
245 "Using defaults and environment variables. For explicit control, "
246 "pass settings_snapshot=create_settings_snapshot(...)."
247 )
249 kwargs["settings_snapshot"] = create_settings_snapshot(
250 base_settings=settings,
251 overrides=settings_override,
252 **snapshot_kwargs,
253 )
254 log_settings(
255 kwargs["settings_snapshot"],
256 "Created settings snapshot for programmatic API",
257 )
258 else:
259 log_settings(
260 kwargs["settings_snapshot"],
261 "Using provided settings snapshot for programmatic API",
262 )
264 # Generate a research_id if none provided
265 if research_id is None:
266 import uuid
268 research_id = str(uuid.uuid4())
269 logger.debug(f"Generated research_id: {research_id}")
271 # Register retrievers if provided
272 if retrievers:
273 from ..web_search_engines.retriever_registry import retriever_registry
275 retriever_registry.register_multiple(retrievers)
276 logger.info(
277 f"Registered {len(retrievers)} retrievers: {list(retrievers.keys())}"
278 )
280 # Register LLMs if provided
281 if llms:
282 from ..llm import register_llm
284 for name, llm_instance in llms.items():
285 register_llm(name, llm_instance)
286 logger.info(f"Registered {len(llms)} LLMs: {list(llms.keys())}")
288 search_context = {
289 "research_id": research_id, # Pass UUID or integer directly
290 "research_query": query,
291 "research_mode": kwargs.get("research_mode", "quick"),
292 "research_phase": "init",
293 "search_iteration": 0,
294 "search_engine_selected": kwargs.get("search_tool"),
295 "username": username, # Include username for metrics tracking
296 "user_password": kwargs.get(
297 "user_password"
298 ), # Include password for metrics tracking
299 # Thread-safe settings snapshot propagated to background search
300 # threads (engine config, per-user resolution, egress scope).
301 "settings_snapshot": kwargs.get("settings_snapshot") or {},
302 }
303 set_search_context(search_context)
305 system = None
306 try:
307 # Remove research_mode from kwargs before passing to _init_search_system
308 init_kwargs = {k: v for k, v in kwargs.items() if k != "research_mode"}
309 # Make sure username is passed to the system
310 init_kwargs["username"] = username
311 init_kwargs["research_id"] = research_id
312 init_kwargs["research_context"] = search_context
313 init_kwargs["search_original_query"] = search_original_query
314 system = _init_search_system(llms=llms, **init_kwargs)
316 # Perform the search and analysis
317 results = system.analyze_topic(query)
319 # Extract the summary from the current knowledge
320 if results and "current_knowledge" in results:
321 summary = results["current_knowledge"]
322 else:
323 summary = "Unable to generate summary for the query."
325 # Prepare the return value (guard against None results)
326 if results is None: 326 ↛ 327line 326 didn't jump to line 327 because the condition on line 326 was never true
327 results = {}
328 return {
329 "research_id": research_id,
330 "summary": summary,
331 "findings": results.get("findings", []),
332 "iterations": results.get("iterations", 0),
333 "questions": results.get("questions", {}),
334 "formatted_findings": results.get("formatted_findings", ""),
335 "sources": results.get("all_links_of_system", []),
336 }
337 finally:
338 if system is not None:
339 _close_system(system)
340 clear_search_context()
343@no_db_settings
344def generate_report(
345 query: str,
346 output_file: str | None = None,
347 progress_callback: Callable | None = None,
348 searches_per_section: int = 2,
349 retrievers: dict[str, Any] | None = None,
350 llms: dict[str, Any] | None = None,
351 username: str | None = None,
352 provider: str | None = None,
353 api_key: str | None = None,
354 temperature: float | None = None,
355 max_search_results: int | None = None,
356 settings: dict[str, Any] | None = None,
357 settings_override: dict[str, Any] | None = None,
358 **kwargs: Any,
359) -> dict[str, Any]:
360 """
361 Generate a comprehensive, structured research report for a given query.
363 Args:
364 query: The research query to analyze
365 output_file: Optional path to save report markdown file
366 progress_callback: Optional callback function to receive progress updates
367 searches_per_section: The number of searches to perform for each
368 section in the report.
369 retrievers: Optional dictionary of {name: retriever} pairs to use as search engines
370 llms: Optional dictionary of {name: llm} pairs to use as language models
371 provider: LLM provider to use (e.g., 'openai', 'anthropic'). For programmatic API only.
372 api_key: API key for the provider. For programmatic API only.
373 temperature: LLM temperature (0.0-1.0). For programmatic API only.
374 max_search_results: Maximum number of search results to return. For programmatic API only.
375 settings: Base settings dict to use instead of defaults. For programmatic API only.
376 settings_override: Dictionary of settings to override. For programmatic API only.
377 **kwargs: Additional configuration for the search system.
379 Returns:
380 Dictionary containing the research report with keys:
381 - 'content': The full report content in markdown format
382 - 'metadata': Report metadata including generated timestamp and query
383 - 'file_path': Path to saved file (if output_file was provided)
385 Examples:
386 # Simple usage with settings snapshot
387 from local_deep_research.api.settings_utils import create_settings_snapshot
388 settings = create_settings_snapshot({"programmatic_mode": True})
389 result = generate_report("AI research", settings_snapshot=settings)
391 # Save to file
392 result = generate_report(
393 "AI research",
394 output_file="report.md",
395 settings_snapshot=settings
396 )
397 """
398 logger.info("Generating comprehensive research report for query: {}", query)
400 if "settings_snapshot" not in kwargs:
401 snapshot_kwargs = {}
402 if provider is not None:
403 snapshot_kwargs["provider"] = provider
404 if api_key is not None: 404 ↛ 405line 404 didn't jump to line 405 because the condition on line 404 was never true
405 snapshot_kwargs["api_key"] = api_key
406 if temperature is not None: 406 ↛ 407line 406 didn't jump to line 407 because the condition on line 406 was never true
407 snapshot_kwargs["temperature"] = temperature
408 if max_search_results is not None: 408 ↛ 409line 408 didn't jump to line 409 because the condition on line 408 was never true
409 snapshot_kwargs["max_search_results"] = max_search_results
411 if (
412 not snapshot_kwargs
413 and settings is None
414 and settings_override is None
415 ):
416 logger.warning(
417 "No settings_snapshot or explicit config provided to generate_report(). "
418 "Using defaults and environment variables. For explicit control, "
419 "pass settings_snapshot=create_settings_snapshot(...)."
420 )
422 kwargs["settings_snapshot"] = create_settings_snapshot(
423 base_settings=settings,
424 overrides=settings_override,
425 **snapshot_kwargs,
426 )
427 log_settings(
428 kwargs["settings_snapshot"],
429 "Created settings snapshot for programmatic API",
430 )
431 else:
432 log_settings(
433 kwargs["settings_snapshot"],
434 "Using provided settings snapshot for programmatic API",
435 )
437 # Register retrievers if provided
438 if retrievers:
439 from ..web_search_engines.retriever_registry import retriever_registry
441 retriever_registry.register_multiple(retrievers)
442 logger.info(
443 f"Registered {len(retrievers)} retrievers: {list(retrievers.keys())}"
444 )
446 # Register LLMs if provided
447 if llms:
448 from ..llm import register_llm
450 for name, llm_instance in llms.items():
451 register_llm(name, llm_instance)
452 logger.info(f"Registered {len(llms)} LLMs: {list(llms.keys())}")
454 import uuid
456 search_context = {
457 "research_id": str(uuid.uuid4()),
458 "research_query": query,
459 "research_mode": "report",
460 "research_phase": "init",
461 "search_iteration": 0,
462 "search_engine_selected": kwargs.get("search_tool"),
463 "username": username,
464 "user_password": kwargs.get("user_password"),
465 "settings_snapshot": kwargs.get("settings_snapshot") or {},
466 }
467 set_search_context(search_context)
469 system = None
470 try:
471 system = _init_search_system(
472 retrievers=retrievers, llms=llms, username=username, **kwargs
473 )
474 # Set progress callback if provided
475 if progress_callback:
476 system.set_progress_callback(progress_callback)
478 # Perform the initial research
479 initial_findings = system.analyze_topic(query)
481 # Generate the structured report
482 report_generator = IntegratedReportGenerator(
483 search_system=system,
484 llm=system.model,
485 searches_per_section=searches_per_section,
486 settings_snapshot=kwargs.get("settings_snapshot"),
487 )
488 report = report_generator.generate_report(initial_findings, query)
490 # Save report to file if path is provided
491 if output_file and report and "content" in report:
492 from ..security.file_write_verifier import write_file_verified
494 write_file_verified(
495 output_file,
496 report["content"],
497 "api.allow_file_output",
498 context="API research report",
499 settings_snapshot=kwargs.get("settings_snapshot"),
500 )
501 logger.info(f"Report saved to {output_file}")
502 report["file_path"] = output_file
503 return report
504 finally:
505 if system is not None: 505 ↛ 507line 505 didn't jump to line 507 because the condition on line 505 was always true
506 _close_system(system)
507 clear_search_context()
510@no_db_settings
511def detailed_research(
512 query: str,
513 research_id: str | None = None,
514 retrievers: dict[str, Any] | None = None,
515 llms: dict[str, Any] | None = None,
516 username: str | None = None,
517 **kwargs: Any,
518) -> dict[str, Any]:
519 """
520 Perform detailed research with comprehensive analysis.
522 Similar to generate_report but returns structured data instead of markdown.
524 Args:
525 query: The research query to analyze
526 research_id: Optional research ID (int or UUID string) for tracking metrics
527 retrievers: Optional dictionary of {name: retriever} pairs to use as search engines
528 llms: Optional dictionary of {name: llm} pairs to use as language models
529 username: Optional username for per-user cache isolation
530 **kwargs: Configuration for the search system. Pass settings_snapshot
531 (via create_settings_snapshot()) to configure provider, temperature, etc.
533 Returns:
534 Dictionary containing detailed research results
535 """
536 logger.info("Performing detailed research for query: {}", query)
538 if "settings_snapshot" not in kwargs:
539 logger.warning(
540 "No settings_snapshot provided to detailed_research(). "
541 "Using defaults and environment variables. For explicit control, "
542 "pass settings_snapshot=create_settings_snapshot(provider=..., "
543 "overrides={'search.tool': ...})."
544 )
545 kwargs["settings_snapshot"] = create_settings_snapshot()
547 # Generate a research_id if none provided
548 if research_id is None:
549 import uuid
551 research_id = str(uuid.uuid4())
552 logger.debug(f"Generated research_id: {research_id}")
554 # Register retrievers if provided
555 if retrievers:
556 from ..web_search_engines.retriever_registry import retriever_registry
558 retriever_registry.register_multiple(retrievers)
559 logger.info(
560 f"Registered {len(retrievers)} retrievers: {list(retrievers.keys())}"
561 )
563 # Register LLMs if provided
564 if llms:
565 from ..llm import register_llm
567 for name, llm_instance in llms.items():
568 register_llm(name, llm_instance)
569 logger.info(f"Registered {len(llms)} LLMs: {list(llms.keys())}")
571 search_context = {
572 "research_id": research_id,
573 "research_query": query,
574 "research_mode": "detailed",
575 "research_phase": "init",
576 "search_iteration": 0,
577 "search_engine_selected": kwargs.get("search_tool"),
578 "username": username,
579 "user_password": kwargs.get("user_password"),
580 "settings_snapshot": kwargs.get("settings_snapshot") or {},
581 }
582 set_search_context(search_context)
584 system = None
585 try:
586 # Initialize system
587 system = _init_search_system(
588 retrievers=retrievers, llms=llms, username=username, **kwargs
589 )
591 # Perform detailed research
592 results = system.analyze_topic(query)
594 # Return comprehensive results (guard against None results)
595 if results is None: 595 ↛ 596line 595 didn't jump to line 596 because the condition on line 595 was never true
596 results = {}
597 return {
598 "query": query,
599 "research_id": research_id,
600 "summary": results.get("current_knowledge", ""),
601 "findings": results.get("findings", []),
602 "iterations": results.get("iterations", 0),
603 "questions": results.get("questions", {}),
604 "formatted_findings": results.get("formatted_findings", ""),
605 "sources": results.get("all_links_of_system", []),
606 "metadata": {
607 "timestamp": datetime.now(UTC).isoformat(),
608 "search_tool": kwargs.get("search_tool", "searxng"),
609 "iterations_requested": kwargs.get("iterations", 1),
610 "strategy": kwargs.get("search_strategy", "source_based"),
611 },
612 }
613 finally:
614 if system is not None: 614 ↛ 616line 614 didn't jump to line 616 because the condition on line 614 was always true
615 _close_system(system)
616 clear_search_context()
619@no_db_settings
620def analyze_documents(
621 query: str,
622 collection_name: str,
623 max_results: int = 10,
624 temperature: float = 0.7,
625 force_reindex: bool = False,
626 output_file: str | None = None,
627 *,
628 username: str | None = None,
629 settings_snapshot: dict[str, Any] | None = None,
630 programmatic_mode: bool = True,
631) -> dict[str, Any]:
632 """
633 Search and analyze documents in a specific local collection.
635 Args:
636 query: The search query
637 collection_name: Name of the local document collection to search
638 max_results: Maximum number of results to return
639 temperature: LLM temperature for summary generation
640 force_reindex: Whether to force reindexing the collection
641 output_file: Optional path to save analysis results to a file
642 username: Optional username for thread context. REST callers pass the
643 authenticated user; programmatic SDK callers can omit.
644 settings_snapshot: Settings snapshot for the user's stored configuration
645 (LLM provider/model, embedding model, etc.). REST callers pass the
646 snapshot from the user's encrypted DB; SDK callers can omit to use
647 JSON defaults + LDR_* env vars.
648 programmatic_mode: If True (default for SDK callers), disables DB-backed
649 metrics. REST callers pass False so per-user rate-limit estimates
650 persist across requests.
652 Returns:
653 Dictionary containing:
654 - 'summary': Summary of the findings
655 - 'documents': List of matching documents with content and metadata
656 """
657 if settings_snapshot is None:
658 settings_snapshot = create_settings_snapshot()
660 logger.info(
661 f"Analyzing documents in collection '{collection_name}' for query: {query}"
662 )
664 llm = None
665 search = None
666 try:
667 # Get language model with custom temperature
668 llm = get_llm(
669 temperature=temperature, settings_snapshot=settings_snapshot
670 )
672 # Get search engine for the specified collection
673 search = get_search(
674 collection_name,
675 llm_instance=llm,
676 username=username,
677 settings_snapshot=settings_snapshot,
678 programmatic_mode=programmatic_mode,
679 )
681 if not search:
682 from ..utilities.resource_utils import safe_close
684 safe_close(llm, "LLM")
685 llm = None
686 return {
687 "summary": f"Error: Collection '{collection_name}' not found or not properly configured.",
688 "documents": [],
689 }
691 # Set max results
692 search.max_results = max_results
693 # Perform the search
694 results = search.run(query)
696 if not results:
697 return {
698 "summary": f"No documents found in collection '{collection_name}' for query: '{query}'",
699 "documents": [],
700 }
702 # Get LLM to generate a summary of the results
704 docs_text = "\n\n".join(
705 [
706 f"Document {i + 1}:"
707 f" {doc.get('content', doc.get('snippet', ''))[:1000]}"
708 for i, doc in enumerate(results[:5])
709 ]
710 ) # Limit to first 5 docs and 1000 chars each
712 summary_prompt = f"""Analyze these document excerpts related to the query: "{query}"
714 {docs_text}
716 Provide a concise summary of the key information found in these documents related to the query.
717 """
719 import time
721 llm_start_time = time.time()
722 logger.info(
723 f"Starting LLM summary generation (prompt length: {len(summary_prompt)} chars)..."
724 )
726 summary_response = llm.invoke(summary_prompt)
728 llm_elapsed = time.time() - llm_start_time
729 logger.info(f"LLM summary generation completed in {llm_elapsed:.2f}s")
731 if hasattr(summary_response, "content"): 731 ↛ 734line 731 didn't jump to line 734 because the condition on line 731 was always true
732 summary = summary_response.content
733 else:
734 summary = str(summary_response)
736 # Create result dictionary
737 analysis_result = {
738 "summary": summary,
739 "documents": results,
740 "collection": collection_name,
741 "document_count": len(results),
742 }
744 # Save to file if requested
745 if output_file:
746 from ..security.file_write_verifier import write_file_verified
748 content = f"# Document Analysis: {query}\n\n"
749 content += f"## Summary\n\n{summary}\n\n"
750 content += f"## Documents Found: {len(results)}\n\n"
752 for i, doc in enumerate(results):
753 content += (
754 f"### Document {i + 1}: {doc.get('title', 'Untitled')}\n\n"
755 )
756 content += f"**Source:** {doc.get('link', 'Unknown')}\n\n"
757 content += f"**Content:**\n\n{doc.get('content', doc.get('snippet', 'No content available'))[:1000]}...\n\n"
758 content += "---\n\n"
760 write_file_verified(
761 output_file,
762 content,
763 "api.allow_file_output",
764 context="API document analysis",
765 settings_snapshot=settings_snapshot,
766 )
768 analysis_result["file_path"] = output_file
769 logger.info(f"Analysis saved to {output_file}")
771 return analysis_result
772 finally:
773 from ..utilities.resource_utils import safe_close
775 safe_close(search, "search engine")
776 safe_close(llm, "LLM")