Coverage for src / local_deep_research / followup_research / service.py: 95%
52 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:55 +0000
1"""
2Service layer for follow-up research functionality.
4This service handles the business logic for follow-up research,
5including loading parent research context and orchestrating the search.
6"""
8from typing import Dict, Any
9from loguru import logger
11from ..database.models import ResearchHistory
12from ..database.session_context import get_user_db_session
13from ..web.services.research_sources_service import ResearchSourcesService
14from .models import FollowUpRequest
17class FollowUpResearchService:
18 """Service for handling follow-up research operations."""
20 def __init__(self, username: str | None = None):
21 """
22 Initialize the follow-up research service.
24 Args:
25 username: Username for database access
26 """
27 self.username = username
29 def load_parent_research(self, parent_research_id: str) -> Dict[str, Any]:
30 """
31 Load parent research data from the database.
33 Args:
34 parent_research_id: ID of the parent research
36 Returns:
37 Dictionary containing parent research data including:
38 - report_content: The generated report
39 - resources: List of research resources/links
40 - query: Original research query
41 - strategy: Strategy used
42 """
43 try:
44 with get_user_db_session(self.username) as session:
45 # Load research history
46 research = (
47 session.query(ResearchHistory)
48 .filter_by(id=parent_research_id)
49 .first()
50 )
52 if not research:
53 logger.warning(
54 f"Parent research not found: {parent_research_id}"
55 )
56 return {}
58 logger.info(
59 f"Found research: {research.id}, has meta: {research.research_meta is not None}"
60 )
62 # Use the ResearchSourcesService to get sources properly from database
63 sources_service = ResearchSourcesService()
64 resource_list = sources_service.get_research_sources(
65 parent_research_id, username=self.username
66 )
68 logger.info(
69 f"Found {len(resource_list)} sources from ResearchResource table"
70 )
72 # If no sources in database, try to get from research_meta as fallback
73 if not resource_list and research.research_meta:
74 logger.info(
75 "No sources in database, checking research_meta"
76 )
77 logger.info(
78 f"Research meta keys: {list(research.research_meta.keys()) if isinstance(research.research_meta, dict) else 'Not a dict'}"
79 )
81 # Try different possible locations for sources in research_meta
82 meta_sources = (
83 research.research_meta.get("all_links_of_system", [])
84 or research.research_meta.get("sources", [])
85 or research.research_meta.get("links", [])
86 or []
87 )
89 if meta_sources: 89 ↛ 119line 89 didn't jump to line 119 because the condition on line 89 was always true
90 logger.info(
91 f"Found {len(meta_sources)} sources in research_meta, saving to database"
92 )
93 # Source migration is non-fatal: the followup can
94 # still proceed using meta_sources directly even if
95 # saving to the ResearchResource table fails (e.g.
96 # encrypted DB password expired mid-session).
97 try:
98 saved = sources_service.save_research_sources(
99 parent_research_id,
100 meta_sources,
101 username=self.username,
102 )
103 logger.info(f"Saved {saved} sources to database")
105 # Now retrieve them properly formatted
106 resource_list = (
107 sources_service.get_research_sources(
108 parent_research_id,
109 username=self.username,
110 )
111 )
112 except Exception:
113 logger.exception(
114 f"Failed to save/retrieve sources for parent research "
115 f"{parent_research_id} (continuing with raw meta_sources)"
116 )
118 # Convert to dictionary format
119 parent_data = {
120 "research_id": research.id,
121 "query": research.query,
122 "report_content": research.report_content,
123 "formatted_findings": research.research_meta.get(
124 "formatted_findings", ""
125 )
126 if research.research_meta
127 else "",
128 "strategy": research.research_meta.get("strategy_name", "")
129 if research.research_meta
130 else "",
131 "resources": resource_list,
132 "all_links_of_system": resource_list,
133 }
135 logger.info(
136 f"Loaded parent research {parent_research_id} with "
137 f"{len(resource_list)} sources"
138 )
140 return parent_data
142 except Exception:
143 logger.exception("Error loading parent research")
144 return {}
146 def prepare_research_context(
147 self, parent_research_id: str
148 ) -> Dict[str, Any]:
149 """
150 Prepare the research context for the contextual follow-up strategy.
152 Args:
153 parent_research_id: ID of the parent research
155 Returns:
156 Research context dictionary for the strategy
157 """
158 parent_data = self.load_parent_research(parent_research_id)
160 if not parent_data:
161 logger.warning("No parent data found, returning empty context")
162 return {}
164 # Format context for the strategy
165 return {
166 "parent_research_id": parent_research_id,
167 "past_links": parent_data.get("all_links_of_system", []),
168 "past_findings": parent_data.get("formatted_findings", ""),
169 "report_content": parent_data.get("report_content", ""),
170 "resources": parent_data.get("resources", []),
171 "all_links_of_system": parent_data.get("all_links_of_system", []),
172 "original_query": parent_data.get("query", ""),
173 }
175 def perform_followup(self, request: FollowUpRequest) -> Dict[str, Any]:
176 """
177 Perform a follow-up research based on parent research.
179 This method prepares the context and parameters for the research system
180 to use the contextual follow-up strategy.
182 Args:
183 request: FollowUpRequest with question and parent research ID
185 Returns:
186 Dictionary with research parameters for the research system
187 """
188 # Prepare the research context from parent
189 research_context = self.prepare_research_context(
190 request.parent_research_id
191 )
193 if not research_context:
194 logger.warning(
195 f"Parent research not found: {request.parent_research_id}, using empty context"
196 )
197 # Use empty context to allow follow-up without parent
198 research_context = {
199 "parent_research_id": request.parent_research_id,
200 "past_links": [],
201 "past_findings": "",
202 "report_content": "",
203 "resources": [],
204 "all_links_of_system": [],
205 "original_query": "",
206 }
208 # Prepare parameters for the research system
209 research_params = {
210 "query": request.question,
211 "strategy": "contextual-followup",
212 "delegate_strategy": request.strategy,
213 "max_iterations": request.max_iterations,
214 "questions_per_iteration": request.questions_per_iteration,
215 "research_context": research_context,
216 "parent_research_id": request.parent_research_id,
217 }
219 logger.info(
220 f"Prepared follow-up research for question: '{request.question}' "
221 f"based on parent: {request.parent_research_id}"
222 )
224 return research_params