Coverage for src/local_deep_research/advanced_search_system/filters/followup_relevance_filter.py: 98%
63 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Follow-up Relevance Filter
4Filters and ranks past research sources based on their relevance
5to follow-up questions.
6"""
8from typing import Dict, List
9from loguru import logger
11from .base_filter import BaseFilter
12from ...utilities.json_utils import extract_json, get_llm_response_text
13from ...utilities.llm_utils import invoke_llm_sync
16class FollowUpRelevanceFilter(BaseFilter):
17 """
18 Filters past research sources by relevance to follow-up questions.
20 This filter analyzes sources from previous research and determines
21 which ones are most relevant to the new follow-up question.
22 """
24 def filter_results(
25 self, results: List[Dict], query: str, max_results: int = 10, **kwargs
26 ) -> List[Dict]:
27 """
28 Filter search results by relevance to the follow-up query.
30 Args:
31 results: List of source dictionaries from past research
32 query: The follow-up query
33 max_results: Maximum number of results to return (default: 10)
34 **kwargs: Additional parameters:
35 - past_findings: Summary of past findings for context
36 - original_query: The original research query
38 Returns:
39 Filtered list of relevant sources
40 """
41 if not results:
42 return []
44 past_findings = kwargs.get("past_findings", "")
45 original_query = kwargs.get("original_query", "")
47 # Use LLM to select relevant sources
48 relevant_indices = self._select_relevant_sources(
49 results, query, past_findings, max_results, original_query
50 )
52 # Return selected sources. Indices are already validated and
53 # deduplicated by _select_relevant_sources, so each is in range.
54 filtered = [results[i] for i in relevant_indices]
56 logger.info(
57 f"Filtered {len(results)} sources to {len(filtered)} relevant ones "
58 f"for follow-up query. Kept indices: {relevant_indices}"
59 )
61 return filtered
63 def _valid_unique_indices(self, indices, upper_bound):
64 """Yield in-range source indices once each, preserving first-seen order.
66 Accepts ints and floats (floats are truncated toward zero, e.g.
67 ``1.9`` becomes ``1``; some models emit ``0.0``) and
68 rejects booleans, non-numeric values, negative indices, and indices at
69 or beyond ``upper_bound``. Deduplication prevents the same source from
70 being selected more than once.
71 """
72 seen = set()
73 for raw in indices:
74 if isinstance(raw, bool):
75 continue
76 if isinstance(raw, int):
77 idx = raw
78 elif isinstance(raw, float): 78 ↛ 81line 78 didn't jump to line 81 because the condition on line 78 was always true
79 idx = int(raw)
80 else:
81 continue
82 if idx in seen:
83 continue
84 if 0 <= idx < upper_bound:
85 seen.add(idx)
86 yield idx
88 def _select_relevant_sources(
89 self,
90 sources: List[Dict],
91 query: str,
92 context: str,
93 max_results: int,
94 original_query: str = "",
95 ) -> List[int]:
96 """
97 Select relevant sources using LLM.
99 Args:
100 sources: List of source dictionaries
101 query: The follow-up query
102 context: Past findings context
103 max_results: Maximum number of sources to select
104 original_query: The original research query
106 Returns:
107 List of indices of relevant sources
108 """
109 if not self.model:
110 # If no model available, return first max_results
111 return list(range(min(max_results, len(sources))))
113 # Build source list for LLM
114 source_list = []
115 for i, source in enumerate(sources):
116 title = source.get("title") or "Unknown"
117 url = source.get("url") or ""
118 snippet = (
119 source.get("snippet") or source.get("content_preview") or ""
120 )[:150]
121 source_list.append(
122 f"{i}. {title}\n URL: {url}\n Content: {snippet}"
123 )
125 sources_text = "\n\n".join(source_list)
127 # Include context if available for better selection
128 context_section = ""
129 if context or original_query:
130 parts = []
131 if original_query:
132 parts.append(f"Original research question: {original_query}")
133 if context:
134 parts.append(f"Previous research findings:\n{context}")
136 context_section = f"""
137Previous Research Context:
138{chr(10).join(parts)}
140---
141"""
143 # This text is an LLM source-selection prompt; it is never executed as SQL.
144 prompt = f"""
145Select the most relevant sources for answering this follow-up question based on the previous research context.
146{context_section}
147Follow-up question: "{query}"
149Available sources from previous research:
150{sources_text}
152Instructions:
153- Select sources that are most relevant to the follow-up question given the context
154- Consider which sources directly address the question or provide essential information
155- Think about what the user is asking for in relation to the previous findings
156- Return ONLY a JSON array of source numbers (e.g., [0, 2, 5, 7])
157- Do not include any explanation or other text
159Return the indices of relevant sources as a JSON array:""" # nosec B608
161 try:
162 response = invoke_llm_sync(self.model, prompt)
163 content = get_llm_response_text(response)
165 # Parse JSON response
166 indices = extract_json(content, expected_type=list)
168 if indices is not None:
169 # Validate, bound-check, and deduplicate the parsed indices
170 indices = list(
171 self._valid_unique_indices(indices, len(sources))
172 )
173 else:
174 logger.debug("Failed to parse JSON, attempting regex fallback")
175 # Fallback to regex extraction
176 import re
178 numbers = re.findall(r"\d+", content)
179 indices = list(
180 self._valid_unique_indices(
181 (int(n) for n in numbers), len(sources)
182 )
183 )
185 return indices
186 except Exception as e:
187 logger.debug(f"LLM source selection failed: {e}")
188 # Fallback to first max_results sources
189 return list(range(min(max_results, len(sources))))