Coverage for src/local_deep_research/advanced_search_system/findings/repository.py: 94%
192 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Findings repository for managing research findings.
3"""
5from loguru import logger
6from typing import Dict, List, Union
8from langchain_core.documents import Document
9from langchain_core.language_models import BaseLanguageModel
11from ...utilities.search_utilities import format_findings
12from .base_findings import BaseFindingsRepository
13from ...utilities.llm_utils import invoke_llm_sync
16def format_links(links: List[Dict]) -> str:
17 """Format a list of links into a readable string.
19 Args:
20 links: List of dictionaries containing 'title' and 'url' keys
22 Returns:
23 str: Formatted string of links
24 """
25 return "\n".join(
26 f"{i + 1}. {link['title']}\n URL: {link['url']}"
27 for i, link in enumerate(links)
28 )
31class FindingsRepository(BaseFindingsRepository):
32 """Repository for managing research findings."""
34 def __init__(self, model: BaseLanguageModel):
35 """Initialize the repository.
37 Args:
38 model: The LLM model to use for synthesis
39 """
40 super().__init__(model)
41 self.findings: Dict[str, List[Dict]] = {}
42 self.documents: List[Document] = []
43 self.questions_by_iteration: Dict[int, List[str]] = {}
45 def add_finding(self, query: str, finding: Dict | str) -> None:
46 """Add a finding for a query."""
47 self.findings.setdefault(query, [])
49 # Convert to dictionary if it's a string
50 if isinstance(finding, str):
51 finding_dict = {
52 "phase": "Synthesis",
53 "content": finding,
54 "question": query,
55 "search_results": [],
56 "documents": [],
57 }
58 self.findings[query].append(finding_dict)
59 else:
60 # It's already a dictionary
61 self.findings[query].append(finding)
63 # Store raw synthesized content if it's the final synthesis
64 # Only check for phase if it's a dictionary
65 if (
66 isinstance(finding, dict)
67 and finding.get("phase") == "Final synthesis"
68 ):
69 self.findings[query + "_synthesis"] = [
70 {
71 "phase": "Synthesis",
72 "content": finding.get("content", ""),
73 "question": query,
74 "search_results": [],
75 "documents": [],
76 }
77 ]
79 logger.info(
80 f"Added finding for query: {query}. Total findings: {len(self.findings[query])}"
81 )
83 def get_findings(self, query: str) -> List[Dict]:
84 """Get findings for a query.
86 Args:
87 query: The query to get findings for
89 Returns:
90 List of findings for the query
91 """
92 return self.findings.get(query, [])
94 def clear_findings(self, query: str) -> None:
95 """Clear findings for a query.
97 Args:
98 query: The query to clear findings for
99 """
100 if query in self.findings:
101 del self.findings[query]
102 logger.info(f"Cleared findings for query: {query}")
104 def add_documents(self, documents: List[Document]) -> None:
105 """Add documents to the repository.
107 Args:
108 documents: List of documents to add
109 """
110 self.documents.extend(documents)
111 logger.info(f"Added {len(documents)} documents to repository")
113 def set_questions_by_iteration(
114 self, questions_by_iteration: Dict[int, List[str]]
115 ) -> None:
116 """Set the questions by iteration.
118 Args:
119 questions_by_iteration: Dictionary mapping iteration numbers to lists of questions
120 """
121 self.questions_by_iteration = questions_by_iteration.copy()
122 logger.info(
123 f"Set questions for {len(questions_by_iteration)} iterations"
124 )
126 def format_findings_to_text(
127 self, findings_list: List[Dict], synthesized_content: str
128 ) -> str:
129 """Format findings into a detailed text output using the utility function.
131 Args:
132 findings_list: List of finding dictionaries from the strategy execution.
133 synthesized_content: The final synthesized content generated by the LLM.
135 Returns:
136 str: Formatted text output.
137 """
138 logger.info(
139 f"Formatting final report. Number of detailed findings: {len(findings_list)}. Synthesized content length: {len(synthesized_content)}. Number of question iterations: {len(self.questions_by_iteration)}"
140 )
141 # Log details about the inputs
142 logger.debug(
143 f"Detailed findings list structure (first item type if exists): {type(findings_list[0]) if findings_list else 'Empty'}"
144 )
145 logger.debug(
146 f"Questions by iteration keys: {list(self.questions_by_iteration.keys())}"
147 )
148 if findings_list:
149 logger.debug(
150 f"First finding item keys: {list(findings_list[0].keys())}"
151 )
153 try:
154 # Pass the detailed findings list, the synthesized content (as current_knowledge), and the stored questions
155 formatted_report = format_findings(
156 findings_list,
157 synthesized_content, # This goes to the 'current_knowledge' param in format_findings
158 self.questions_by_iteration,
159 )
160 logger.info("Successfully formatted final report.")
161 return formatted_report
162 except Exception:
163 logger.exception("Error occurred during final report formatting")
164 # Preserve the answer, including an "Error:" prefix when synthesis
165 # also failed. A wrapper hides that prefix from the worker's safe
166 # recovery path and can turn raw exception details into a report.
167 return synthesized_content
169 def synthesize_findings(
170 self,
171 query: str,
172 sub_queries: List[str],
173 findings: List[Union[Dict, str]],
174 accumulated_knowledge: str = None,
175 old_formatting: bool = False,
176 ) -> str:
177 """
178 Synthesize accumulated knowledge into a final answer.
180 Args:
181 query: The original query
182 sub_queries: List of sub-queries (for context)
183 findings: List of findings strings or dictionaries from previous steps
184 accumulated_knowledge: Optional pre-existing knowledge to incorporate
185 old_formatting: Whether to use the old formatting approach
187 Returns:
188 str: Synthesized final answer content.
189 """
190 logger.info(f"synthesize_findings called with query: '{query}'")
191 logger.info(
192 f"sub_queries type: {type(sub_queries)}, length: {len(sub_queries)}"
193 )
194 logger.info(f"findings type: {type(findings)}, length: {len(findings)}")
196 # Use provided accumulated_knowledge or join findings if it's None
197 if accumulated_knowledge is None:
198 # Convert findings to text if they are dictionaries
199 finding_texts = []
200 for item in findings:
201 if isinstance(item, dict) and "content" in item:
202 finding_texts.append(item["content"])
203 elif isinstance(item, str):
204 finding_texts.append(item)
205 accumulated_knowledge = "\n\n".join(finding_texts)
207 if findings:
208 logger.info(f"first finding type: {type(findings[0])}")
209 if isinstance(findings[0], dict):
210 logger.info(
211 f"first finding keys: {list(findings[0].keys()) if hasattr(findings[0], 'keys') else 'No keys'}"
212 )
213 if "content" in findings[0]:
214 logger.info(
215 f"first finding content type: {type(findings[0]['content'])}"
216 )
217 elif isinstance(findings[0], str): 217 ↛ 223line 217 didn't jump to line 223 because the condition on line 217 was always true
218 logger.info(f"first finding string length: {len(findings[0])}")
219 logger.info(
220 f"first finding string preview: {findings[0][:100]}..."
221 )
223 if old_formatting:
224 # Convert findings list if it contains strings instead of dictionaries
225 findings_list = []
226 for i, item in enumerate(findings):
227 if isinstance(item, str):
228 findings_list.append(
229 {"phase": f"Finding {i + 1}", "content": item}
230 )
231 elif isinstance(item, dict): 231 ↛ 226line 231 didn't jump to line 226 because the condition on line 231 was always true
232 findings_list.append(item)
234 return format_findings(
235 findings_list=findings_list,
236 synthesized_content=accumulated_knowledge,
237 questions_by_iteration=self.questions_by_iteration,
238 )
239 try:
240 # Extract finding content texts for the prompt
241 finding_texts = []
242 for item in findings:
243 if isinstance(item, dict) and "content" in item:
244 finding_texts.append(item["content"])
245 elif isinstance(item, str):
246 finding_texts.append(item)
248 # Use finding_texts for the prompt
249 current_knowledge = (
250 "\n\n".join(finding_texts) if finding_texts else ""
251 )
253 # Check if knowledge exceeds a reasonable token limit (rough estimate based on characters)
254 # 1 token ≈ 4 characters in English
255 estimated_tokens = len(current_knowledge) / 4
256 max_safe_tokens = (
257 12000 # Adjust based on your model's context window
258 )
260 if estimated_tokens > max_safe_tokens:
261 logger.warning(
262 f"Knowledge size may exceed model's capacity: ~{int(estimated_tokens)} tokens"
263 )
264 # Truncate if needed (keeping the beginning and end which are often most important)
265 # This is a simple approach - a more sophisticated chunking might be better
266 if len(current_knowledge) > 24000: # ~6000 tokens 266 ↛ 278line 266 didn't jump to line 278 because the condition on line 266 was always true
267 first_part = current_knowledge[
268 :12000
269 ] # ~3000 tokens from start
270 last_part = current_knowledge[
271 -12000:
272 ] # ~3000 tokens from end
273 current_knowledge = f"{first_part}\n\n[...content truncated due to length...]\n\n{last_part}"
274 logger.info(
275 "Knowledge truncated to fit within token limits"
276 )
278 prompt = f"""Use IEEE style citations [1], [2], etc. Never make up your own citations. Synthesize the following accumulated knowledge into a comprehensive answer for the original query.
279Format the response with clear sections, citations, and a concise summary.
281Original Query: {query}
283Accumulated Knowledge:
284{current_knowledge}
286Sub-questions asked (for context):
287{chr(10).join(f"- {sq}" for sq in sub_queries)}
289Generate a well-structured, concise answer that:
2901. Starts with a clear explanation of the most important points
2912. Organizes information into logical sections with headers if needed
2923. Maintains logical flow and prioritizes important information over minor details
2934. Avoids repetition and unnecessary detail
295Use IEEE style citations [1], [2], etc. Never make up your own citations.
296"""
298 logger.info(
299 f"Synthesizing final answer. Query: '{query}'. Knowledge length: {len(current_knowledge)}. Prompt length: {len(prompt)}"
300 )
301 # Log first 500 chars of prompt for debugging context length issues
302 logger.debug(
303 f"Synthesis prompt (first 500 chars): {prompt[:500]}..."
304 )
306 try:
307 # Add timeout handling
308 import platform
309 import signal
310 import threading
311 from contextlib import contextmanager
313 # Check if we're on Windows
314 if platform.system() == "Windows":
316 def timeout_handler(timeout_seconds, callback, args):
317 def handler():
318 callback(*args)
320 timer = threading.Timer(timeout_seconds, handler)
321 timer.daemon = True
322 return timer
324 def invoke_with_timeout(
325 timeout_seconds, func, *args, **kwargs
326 ):
327 """
328 Function for implementing timeouts on Windows
329 """
330 import contextvars as _cv
332 result = None
333 exception = None
334 completed = False
335 # Capture the calling thread's context (request_user,
336 # search_context) so func sees the same user identity
337 # inside the worker thread instead of an empty context.
338 _ctx = _cv.copy_context()
340 def target():
341 nonlocal result, exception, completed
342 try:
343 result = _ctx.run(func, *args, **kwargs)
344 completed = True
345 except Exception as e:
346 exception = e
348 thread = threading.Thread(target=target)
349 thread.daemon = True
351 try:
352 thread.start()
353 thread.join(timeout_seconds)
354 if not completed and thread.is_alive():
355 raise TimeoutError(
356 f"Operation timed out after {timeout_seconds} seconds"
357 )
358 if exception: 358 ↛ 359line 358 didn't jump to line 359 because the condition on line 358 was never true
359 raise exception
360 return result
361 finally:
362 # Nothing to clean up
363 pass
365 # Use Windows-compatible timeout
366 try:
367 logger.info(
368 "Using Windows-compatible timeout for LLM invocation"
369 )
370 response = invoke_with_timeout(
371 120, invoke_llm_sync, self.model, prompt
372 )
374 # <think> tags are stripped centrally in ProcessingLLMWrapper;
375 # keep only a shape-guard for non-message returns.
376 if hasattr(response, "content"):
377 synthesized_content = response.content
378 else:
379 synthesized_content = str(response)
381 logger.info(
382 f"Successfully synthesized final answer for query: '{query}'"
383 )
384 # Return only the synthesized content from the LLM
385 return synthesized_content
386 except TimeoutError as timeout_error:
387 logger.exception(
388 f"LLM invocation timed out during synthesis for query '{query}': {timeout_error}"
389 )
390 # Return more specific error about timeout
391 return "Error: Final answer synthesis failed due to LLM timeout. Please check your LLM service or try with a smaller query scope."
393 else:
394 # Unix-compatible timeout using SIGALRM
395 @contextmanager
396 def timeout(seconds, message="Operation timed out"):
397 def signal_handler(signum, frame):
398 raise TimeoutError(message) # noqa: TRY301 — signal handler, not caught by enclosing try
400 signal.signal(signal.SIGALRM, signal_handler)
401 signal.alarm(seconds)
402 try:
403 yield
404 finally:
405 signal.alarm(0)
407 # Try with a timeout (adjust seconds as needed)
408 try:
409 with timeout(
410 120, "LLM invocation timed out after 120 seconds"
411 ):
412 response = invoke_llm_sync(self.model, prompt)
414 # <think> tags are stripped centrally in ProcessingLLMWrapper;
415 # keep only a shape-guard for non-message returns.
416 if hasattr(response, "content"):
417 synthesized_content = response.content
418 else:
419 synthesized_content = str(response)
421 logger.info(
422 f"Successfully synthesized final answer for query: '{query}'"
423 )
424 # Return only the synthesized content from the LLM
425 return synthesized_content
426 except TimeoutError as timeout_error:
427 logger.exception(
428 f"LLM invocation timed out during synthesis for query '{query}': {timeout_error}"
429 )
430 # Return more specific error about timeout
431 return "Error: Final answer synthesis failed due to LLM timeout. Please check your LLM service or try with a smaller query scope."
433 except Exception as invoke_error:
434 logger.exception(
435 f"LLM invocation failed during synthesis for query '{query}': {invoke_error}"
436 )
438 # Attempt to determine the type of error
439 error_message = str(invoke_error).lower()
440 error_type = "unknown"
442 if "timeout" in error_message or "timed out" in error_message:
443 error_type = "timeout"
444 elif (
445 "too many tokens" in error_message
446 or "context length" in error_message
447 or "token limit" in error_message
448 ):
449 error_type = "token_limit"
450 elif (
451 "rate limit" in error_message
452 or "rate_limit" in error_message
453 ):
454 error_type = "rate_limit"
455 elif (
456 "connection" in error_message or "network" in error_message
457 ):
458 error_type = "connection"
459 elif (
460 "api key" in error_message
461 or "authentication" in error_message
462 ):
463 error_type = "authentication"
465 # Return more detailed error message based on type
466 if error_type == "timeout":
467 return "Error: Failed to synthesize final answer due to LLM timeout. Please check your connection or try again later."
468 if error_type == "token_limit":
469 return "Error: Failed to synthesize final answer due to token limit exceeded. Try reducing the scope of your query."
470 if error_type == "rate_limit":
471 return "Error: Failed to synthesize final answer due to LLM rate limit. Please try again in a few minutes."
472 if error_type == "connection":
473 return "Error: Failed to synthesize final answer due to connection issues. Please check your internet connection and LLM service status."
474 if error_type == "authentication":
475 return "Error: Failed to synthesize final answer due to authentication issues. Please check your API keys."
476 # Generic error with details
477 return f"Error: Failed to synthesize final answer. LLM error: {invoke_error!s}"
479 except Exception as e:
480 # Catch potential errors during prompt construction or logging itself
481 logger.exception(
482 f"Error preparing or executing synthesis for query '{query}'"
483 )
484 # Return a specific error message for synthesis failure
485 return f"Error: Failed to synthesize final answer from knowledge. Details: {e!s}"