Coverage for src/local_deep_research/advanced_search_system/knowledge/followup_context_manager.py: 97%
109 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2Follow-up Context Manager
4Manages and processes past research context for follow-up questions.
5This is a standalone class with no abstract base, to avoid implementing
6many unused abstract methods.
7"""
9from typing import Dict, List, Any, Optional
10from loguru import logger
12from langchain_core.language_models.chat_models import BaseChatModel
14from ...utilities.json_utils import get_llm_response_text
17class FollowUpContextHandler:
18 """
19 Manages past research context for follow-up research.
21 This class handles:
22 1. Loading and structuring past research data
23 2. Summarizing findings for follow-up context
24 3. Extracting relevant information for new searches
25 4. Building comprehensive context for strategies
26 """
28 def __init__(
29 self, model: BaseChatModel, settings_snapshot: Optional[Dict] = None
30 ):
31 """
32 Initialize the context manager.
34 Args:
35 model: Language model for processing context
36 settings_snapshot: Optional settings snapshot
37 """
38 self.model = model
39 self.settings_snapshot = settings_snapshot or {}
40 self.past_research_cache = {}
42 def build_context(
43 self, research_data: Dict[str, Any], follow_up_query: str
44 ) -> Dict[str, Any]:
45 """
46 Build comprehensive context from past research.
48 Args:
49 research_data: Past research data including findings, sources, etc.
50 follow_up_query: The follow-up question being asked
52 Returns:
53 Structured context dictionary for follow-up research
54 """
55 logger.info(f"Building context for follow-up: {follow_up_query}")
57 # Extract all components
58 return {
59 "parent_research_id": research_data.get("research_id", ""),
60 "original_query": research_data.get("query", ""),
61 "follow_up_query": follow_up_query,
62 "past_findings": self._extract_findings(research_data),
63 "past_sources": self._extract_sources(research_data),
64 "key_entities": self._extract_entities(research_data),
65 "summary": self._create_summary(research_data, follow_up_query),
66 "report_content": research_data.get("report_content", ""),
67 "formatted_findings": research_data.get("formatted_findings", ""),
68 "all_links_of_system": research_data.get("all_links_of_system", []),
69 "metadata": self._extract_metadata(research_data),
70 }
72 def _extract_findings(self, research_data: Dict) -> str:
73 """
74 Extract and format findings from past research.
76 Args:
77 research_data: Past research data
79 Returns:
80 Formatted findings string
81 """
82 findings_parts = []
84 # Check various possible locations for findings
85 if formatted := research_data.get("formatted_findings"):
86 findings_parts.append(formatted)
88 if report := research_data.get("report_content"):
89 # Take first part of report if no formatted findings
90 if not findings_parts:
91 findings_parts.append(report[:2000])
93 if not findings_parts:
94 # Multi-turn chat supplies its condensed prior findings under the
95 # "past_findings" key (it has no formatted_findings/report_content
96 # of its own). Honor it before declaring nothing available —
97 # otherwise the chat-built summary is silently dropped and the
98 # follow-up prompt sees "No previous findings available".
99 if past_findings := research_data.get("past_findings"):
100 return past_findings
101 return "No previous findings available"
103 return "\n\n".join(findings_parts)
105 def _extract_sources(self, research_data: Dict) -> List[Dict]:
106 """
107 Extract and structure sources from past research.
109 Args:
110 research_data: Past research data
112 Returns:
113 List of source dictionaries
114 """
115 sources = []
116 seen_urls = set()
118 # Check all possible source fields
119 for field in ["resources", "all_links_of_system", "past_links"]:
120 if field_sources := research_data.get(field, []):
121 for source in field_sources:
122 url = source.get("url", "")
123 # Avoid duplicates by URL
124 if url and url not in seen_urls:
125 sources.append(source)
126 seen_urls.add(url)
127 elif not url:
128 # Include sources without URLs (shouldn't happen but be safe)
129 sources.append(source)
131 return sources
133 def _extract_entities(self, research_data: Dict) -> List[str]:
134 """
135 Extract key entities from past research.
137 Args:
138 research_data: Past research data
140 Returns:
141 List of key entities
142 """
143 findings = self._extract_findings(research_data)
145 if not findings or not self.model:
146 return []
148 prompt = f"""
149Extract key entities (names, places, organizations, concepts) from these research findings:
151{findings[:2000]}
153Return up to 10 most important entities, one per line.
154"""
156 try:
157 response = self.model.invoke(prompt)
158 entities = [
159 line.strip()
160 for line in get_llm_response_text(response).split("\n")
161 if line.strip()
162 ]
163 return entities[:10]
164 except Exception:
165 logger.warning("Failed to extract entities")
166 return []
168 def _create_summary(self, research_data: Dict, follow_up_query: str) -> str:
169 """
170 Create a targeted summary of past research relevant to the follow-up question.
171 This is used internally for building context.
173 Args:
174 research_data: Past research data
175 follow_up_query: The follow-up question
177 Returns:
178 Targeted summary for context building
179 """
180 findings = self._extract_findings(research_data)
181 original_query = research_data.get("query", "")
183 # For internal context, create a brief targeted summary
184 return self._generate_summary(
185 findings=findings,
186 query=follow_up_query,
187 original_query=original_query,
188 max_sentences=5,
189 purpose="context",
190 )
192 def _extract_metadata(self, research_data: Dict) -> Dict:
193 """
194 Extract metadata from past research.
196 Args:
197 research_data: Past research data
199 Returns:
200 Metadata dictionary
201 """
202 return {
203 "strategy": research_data.get("strategy", ""),
204 "mode": research_data.get("mode", ""),
205 "created_at": research_data.get("created_at", ""),
206 "research_meta": research_data.get("research_meta", {}),
207 }
209 def summarize_for_followup(
210 self, findings: str, query: str, max_length: int = 1000
211 ) -> str:
212 """
213 Create a concise summary of findings for external use (e.g., in prompts).
214 This creates a length-constrained summary suitable for inclusion in LLM prompts.
216 Args:
217 findings: Past research findings
218 query: Follow-up query
219 max_length: Maximum length of summary in characters
221 Returns:
222 Concise summary constrained to max_length
223 """
224 # Use the shared summary generation with specific parameters for external use
225 return self._generate_summary(
226 findings=findings,
227 query=query,
228 original_query=None,
229 max_sentences=max_length
230 // 100, # Approximate sentences based on length
231 purpose="prompt",
232 max_length=max_length,
233 )
235 def _generate_summary(
236 self,
237 findings: str,
238 query: str,
239 original_query: Optional[str] = None,
240 max_sentences: int = 5,
241 purpose: str = "context",
242 max_length: Optional[int] = None,
243 ) -> str:
244 """
245 Shared summary generation logic.
247 Args:
248 findings: Research findings to summarize
249 query: Follow-up query
250 original_query: Original research query (optional)
251 max_sentences: Maximum number of sentences
252 purpose: Purpose of summary ("context" or "prompt")
253 max_length: Maximum character length (optional)
255 Returns:
256 Generated summary
257 """
258 if not findings:
259 return ""
261 # If findings are already short enough, return as-is
262 if max_length and len(findings) <= max_length:
263 return findings
265 if not self.model:
266 # Fallback without model
267 if max_length:
268 return findings[:max_length] + "..."
269 return findings[:500] + "..."
271 # Build prompt based on purpose
272 if purpose == "context" and original_query:
273 prompt = f"""
274Create a brief summary of previous research findings that are relevant to this follow-up question:
276Original research question: "{original_query}"
277Follow-up question: "{query}"
279Previous findings:
280{findings[:3000]}
282Provide a {max_sentences}-sentence summary focusing on aspects relevant to the follow-up question.
283"""
284 else:
285 prompt = f"""
286Summarize these research findings in relation to the follow-up question:
288Follow-up question: "{query}"
290Findings:
291{findings[:4000]}
293Create a summary of {max_sentences} sentences that captures the most relevant information.
294"""
296 try:
297 response = self.model.invoke(prompt)
298 summary = get_llm_response_text(response)
300 # Apply length constraint if specified
301 if max_length and len(summary) > max_length: 301 ↛ 302line 301 didn't jump to line 302 because the condition on line 301 was never true
302 summary = summary[:max_length] + "..."
304 return summary
305 except Exception:
306 logger.warning("Summary generation failed")
307 # Fallback to truncation
308 if max_length: 308 ↛ 310line 308 didn't jump to line 310 because the condition on line 308 was always true
309 return findings[:max_length] + "..."
310 return findings[:500] + "..."
312 def identify_gaps(
313 self, research_data: Dict, follow_up_query: str
314 ) -> List[str]:
315 """
316 Identify information gaps that the follow-up should address.
318 Args:
319 research_data: Past research data
320 follow_up_query: Follow-up question
322 Returns:
323 List of identified gaps
324 """
325 findings = self._extract_findings(research_data)
327 if not findings or not self.model:
328 return []
330 prompt = f"""
331Based on the previous research and the follow-up question, identify information gaps:
333Previous research findings:
334{findings[:2000]}
336Follow-up question: "{follow_up_query}"
338What specific information is missing or needs clarification? List up to 5 gaps, one per line.
339"""
341 try:
342 response = self.model.invoke(prompt)
343 gaps = [
344 line.strip()
345 for line in get_llm_response_text(response).split("\n")
346 if line.strip()
347 ]
348 return gaps[:5]
349 except Exception:
350 logger.warning("Failed to identify gaps")
351 return []
353 def format_for_settings_snapshot(
354 self, context: Dict[str, Any]
355 ) -> Dict[str, Any]:
356 """
357 Format context for inclusion in settings snapshot.
358 Only includes essential metadata, not actual content.
360 Args:
361 context: Full context dictionary
363 Returns:
364 Minimal metadata for settings snapshot
365 """
366 # Only include minimal metadata in settings snapshot
367 # Settings snapshot should be for settings, not data
368 return {
369 "followup_metadata": {
370 "parent_research_id": context.get("parent_research_id"),
371 "is_followup": True,
372 "has_context": bool(context.get("past_findings")),
373 }
374 }
376 def get_relevant_context_for_llm(
377 self, context: Dict[str, Any], max_tokens: int = 2000
378 ) -> str:
379 """
380 Get a concise version of context for LLM prompts.
382 Args:
383 context: Full context dictionary
384 max_tokens: Approximate maximum tokens
386 Returns:
387 Concise context string
388 """
389 parts = []
391 # Add original and follow-up queries
392 parts.append(f"Original research: {context.get('original_query', '')}")
393 parts.append(
394 f"Follow-up question: {context.get('follow_up_query', '')}"
395 )
397 # Add summary
398 if summary := context.get("summary"):
399 parts.append(f"\nPrevious findings summary:\n{summary}")
401 # Add key entities
402 if entities := context.get("key_entities"):
403 parts.append(f"\nKey entities: {', '.join(entities[:5])}")
405 # Add source count
406 if sources := context.get("past_sources"):
407 parts.append(f"\nAvailable sources: {len(sources)}")
409 result = "\n".join(parts)
411 # Truncate if needed (rough approximation: 4 chars per token)
412 max_chars = max_tokens * 4
413 if len(result) > max_chars:
414 result = result[:max_chars] + "..."
416 return result