Coverage for src/local_deep_research/advanced_search_system/filters/followup_relevance_filter.py: 98%

1""" 

2Follow-up Relevance Filter 

3 

4Filters and ranks past research sources based on their relevance 

5to follow-up questions. 

6""" 

7 

8from typing import Dict, List 

9from loguru import logger 

10 

11from .base_filter import BaseFilter 

12from ...utilities.json_utils import extract_json, get_llm_response_text 

13 

14 

15class FollowUpRelevanceFilter(BaseFilter): 

16 """ 

17 Filters past research sources by relevance to follow-up questions. 

18 

19 This filter analyzes sources from previous research and determines 

20 which ones are most relevant to the new follow-up question. 

21 """ 

22 

23 def filter_results( 

24 self, results: List[Dict], query: str, max_results: int = 10, **kwargs 

25 ) -> List[Dict]: 

26 """ 

27 Filter search results by relevance to the follow-up query. 

28 

29 Args: 

30 results: List of source dictionaries from past research 

31 query: The follow-up query 

32 max_results: Maximum number of results to return (default: 10) 

33 **kwargs: Additional parameters: 

34 - past_findings: Summary of past findings for context 

35 - original_query: The original research query 

36 

37 Returns: 

38 Filtered list of relevant sources 

39 """ 

40 if not results: 

41 return [] 

42 

43 past_findings = kwargs.get("past_findings", "") 

44 original_query = kwargs.get("original_query", "") 

45 

46 # Use LLM to select relevant sources 

47 relevant_indices = self._select_relevant_sources( 

48 results, query, past_findings, max_results, original_query 

49 ) 

50 

51 # Return selected sources. Indices are already validated and 

52 # deduplicated by _select_relevant_sources, so each is in range. 

53 filtered = [results[i] for i in relevant_indices] 

54 

55 logger.info( 

56 f"Filtered {len(results)} sources to {len(filtered)} relevant ones " 

57 f"for follow-up query. Kept indices: {relevant_indices}" 

58 ) 

59 

60 return filtered 

61 

62 def _valid_unique_indices(self, indices, upper_bound): 

63 """Yield in-range source indices once each, preserving first-seen order. 

64 

65 Accepts ints and floats (floats are truncated toward zero, e.g. 

66 ``1.9`` becomes ``1``; some models emit ``0.0``) and 

67 rejects booleans, non-numeric values, negative indices, and indices at 

68 or beyond ``upper_bound``. Deduplication prevents the same source from 

69 being selected more than once. 

70 """ 

71 seen = set() 

72 for raw in indices: 

73 if isinstance(raw, bool): 

74 continue 

75 if isinstance(raw, int): 

76 idx = raw 

77 elif isinstance(raw, float): 77 ↛ 80line 77 didn't jump to line 80 because the condition on line 77 was always true

78 idx = int(raw) 

79 else: 

80 continue 

81 if idx in seen: 

82 continue 

83 if 0 <= idx < upper_bound: 

84 seen.add(idx) 

85 yield idx 

86 

87 def _select_relevant_sources( 

88 self, 

89 sources: List[Dict], 

90 query: str, 

91 context: str, 

92 max_results: int, 

93 original_query: str = "", 

94 ) -> List[int]: 

95 """ 

96 Select relevant sources using LLM. 

97 

98 Args: 

99 sources: List of source dictionaries 

100 query: The follow-up query 

101 context: Past findings context 

102 max_results: Maximum number of sources to select 

103 original_query: The original research query 

104 

105 Returns: 

106 List of indices of relevant sources 

107 """ 

108 if not self.model: 

109 # If no model available, return first max_results 

110 return list(range(min(max_results, len(sources)))) 

111 

112 # Build source list for LLM 

113 source_list = [] 

114 for i, source in enumerate(sources): 

115 title = source.get("title") or "Unknown" 

116 url = source.get("url") or "" 

117 snippet = ( 

118 source.get("snippet") or source.get("content_preview") or "" 

119 )[:150] 

120 source_list.append( 

121 f"{i}. {title}\n URL: {url}\n Content: {snippet}" 

122 ) 

123 

124 sources_text = "\n\n".join(source_list) 

125 

126 # Include context if available for better selection 

127 context_section = "" 

128 if context or original_query: 

129 parts = [] 

130 if original_query: 

131 parts.append(f"Original research question: {original_query}") 

132 if context: 

133 parts.append(f"Previous research findings:\n{context}") 

134 

135 context_section = f""" 

136Previous Research Context: 

137{chr(10).join(parts)} 

138 

139--- 

140""" 

141 

142 prompt = f""" 

143Select the most relevant sources for answering this follow-up question based on the previous research context. 

144{context_section} 

145Follow-up question: "{query}" 

146 

147Available sources from previous research: 

148{sources_text} 

149 

150Instructions: 

151- Select sources that are most relevant to the follow-up question given the context 

152- Consider which sources directly address the question or provide essential information 

153- Think about what the user is asking for in relation to the previous findings 

154- Return ONLY a JSON array of source numbers (e.g., [0, 2, 5, 7]) 

155- Do not include any explanation or other text 

156 

157Return the indices of relevant sources as a JSON array:""" 

158 

159 try: 

160 response = self.model.invoke(prompt) 

161 content = get_llm_response_text(response) 

162 

163 # Parse JSON response 

164 indices = extract_json(content, expected_type=list) 

165 

166 if indices is not None: 

167 # Validate, bound-check, and deduplicate the parsed indices 

168 indices = list( 

169 self._valid_unique_indices(indices, len(sources)) 

170 ) 

171 else: 

172 logger.debug("Failed to parse JSON, attempting regex fallback") 

173 # Fallback to regex extraction 

174 import re 

175 

176 numbers = re.findall(r"\d+", content) 

177 indices = list( 

178 self._valid_unique_indices( 

179 (int(n) for n in numbers), len(sources) 

180 ) 

181 ) 

182 

183 return indices 

184 except Exception as e: 

185 logger.debug(f"LLM source selection failed: {e}") 

186 # Fallback to first max_results sources 

187 return list(range(min(max_results, len(sources))))