Coverage for src/local_deep_research/advanced_search_system/questions/atomic_fact_question.py: 100%

1""" 

2Atomic fact question generator for complex queries. 

3Decomposes complex queries into atomic, independently searchable facts. 

4""" 

5 

6from loguru import logger 

7from typing import Dict, List, Optional 

8 

9from ...utilities.json_utils import get_llm_response_text 

10from .base_question import BaseQuestionGenerator 

11 

12 

13class AtomicFactQuestionGenerator(BaseQuestionGenerator): 

14 """ 

15 Generates questions by decomposing complex queries into atomic facts. 

16 

17 This approach prevents the system from searching for documents that match 

18 ALL criteria at once, instead finding facts independently and then reasoning 

19 about connections. 

20 """ 

21 

22 def generate_questions( 

23 self, 

24 current_knowledge: str, 

25 query: str, 

26 questions_per_iteration: int = 5, 

27 questions_by_iteration: Optional[Dict[int, List[str]]] = None, 

28 ) -> List[str]: 

29 """ 

30 Generate atomic fact questions from a complex query. 

31 

32 Args: 

33 current_knowledge: The accumulated knowledge so far 

34 query: The original research query 

35 questions_per_iteration: Number of questions to generate 

36 questions_by_iteration: Questions generated in previous iterations 

37 

38 Returns: 

39 List of atomic fact questions 

40 """ 

41 questions_by_iteration = questions_by_iteration or {} 

42 

43 # On first iteration, decompose the query 

44 if not questions_by_iteration: 

45 return self._decompose_to_atomic_facts(query) 

46 

47 # On subsequent iterations, fill knowledge gaps or explore connections 

48 return self._generate_gap_filling_questions( 

49 query, 

50 current_knowledge, 

51 questions_by_iteration, 

52 questions_per_iteration, 

53 ) 

54 

55 def _decompose_to_atomic_facts(self, query: str) -> List[str]: 

56 """Decompose complex query into atomic, searchable facts.""" 

57 prompt = f"""Decompose this complex query into simple, atomic facts that can be searched independently. 

58 

59Query: {query} 

60 

61Break this down into individual facts that can be searched separately. Each fact should: 

621. Be about ONE thing only 

632. Be searchable on its own 

643. Not depend on other facts 

654. Use general terms (e.g., "body parts" not specific ones) 

66 

67For example, if the query is about a location with multiple criteria, create separate questions for: 

68- The geographical/geological aspect 

69- The naming aspect 

70- The historical events 

71- The statistical comparisons 

72 

73Return ONLY the questions, one per line. 

74Example format: 

75What locations were formed by glaciers? 

76What geographic features are named after body parts? 

77Where did falls occur between specific dates? 

78""" 

79 

80 response = self.model.invoke(prompt) 

81 

82 response_text = get_llm_response_text(response) 

83 

84 # Parse questions 

85 questions = [] 

86 for line in response_text.strip().split("\n"): 

87 line = line.strip() 

88 if line and not line.startswith("#") and len(line) > 10: 

89 # Clean up any numbering or bullets 

90 for prefix in ["1.", "2.", "3.", "4.", "5.", "-", "*", "•"]: 

91 if line.startswith(prefix): 

92 line = line[len(prefix) :].strip() 

93 questions.append(line) 

94 

95 logger.info(f"Decomposed query into {len(questions)} atomic facts") 

96 return questions[:5] # Limit to 5 atomic facts 

97 

98 def _generate_gap_filling_questions( 

99 self, 

100 original_query: str, 

101 current_knowledge: str, 

102 questions_by_iteration: Dict[int, List[str]], 

103 questions_per_iteration: int, 

104 ) -> List[str]: 

105 """Generate questions to fill knowledge gaps or make connections.""" 

106 

107 # Check if we have enough information to start reasoning 

108 if len(questions_by_iteration) >= 3: 

109 prompt = f"""Based on the accumulated knowledge, generate questions that help connect the facts or fill remaining gaps. 

110 

111Original Query: {original_query} 

112 

113Current Knowledge: 

114{current_knowledge} 

115 

116Previous Questions: 

117{self._format_previous_questions(questions_by_iteration)} 

118 

119Generate {questions_per_iteration} questions that: 

1201. Connect different facts you've found 

1212. Fill specific gaps in knowledge 

1223. Search for locations that match multiple criteria 

1234. Verify specific details 

124 

125Return ONLY the questions, one per line. 

126""" 

127 else: 

128 # Still gathering basic facts 

129 prompt = f"""Continue gathering atomic facts for this query. 

130 

131Original Query: {original_query} 

132 

133Previous Questions: 

134{self._format_previous_questions(questions_by_iteration)} 

135 

136Current Knowledge: 

137{current_knowledge} 

138 

139Generate {questions_per_iteration} more atomic fact questions that help build a complete picture. 

140Focus on facts not yet explored. 

141 

142Return ONLY the questions, one per line. 

143""" 

144 

145 response = self.model.invoke(prompt) 

146 

147 response_text = get_llm_response_text(response) 

148 

149 # Parse questions 

150 questions = [] 

151 for line in response_text.strip().split("\n"): 

152 line = line.strip() 

153 if line and not line.startswith("#") and len(line) > 10: 

154 # Clean up any numbering or bullets 

155 for prefix in ["1.", "2.", "3.", "4.", "5.", "-", "*", "•"]: 

156 if line.startswith(prefix): 

157 line = line[len(prefix) :].strip() 

158 questions.append(line) 

159 

160 return questions[:questions_per_iteration]