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 

11from ...utilities.llm_utils import invoke_llm_sync 

12 

13 

14class AtomicFactQuestionGenerator(BaseQuestionGenerator): 

15 """ 

16 Generates questions by decomposing complex queries into atomic facts. 

17 

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

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

20 about connections. 

21 """ 

22 

23 def generate_questions( 

24 self, 

25 current_knowledge: str, 

26 query: str, 

27 questions_per_iteration: int = 5, 

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

29 ) -> List[str]: 

30 """ 

31 Generate atomic fact questions from a complex query. 

32 

33 Args: 

34 current_knowledge: The accumulated knowledge so far 

35 query: The original research query 

36 questions_per_iteration: Number of questions to generate 

37 questions_by_iteration: Questions generated in previous iterations 

38 

39 Returns: 

40 List of atomic fact questions 

41 """ 

42 questions_by_iteration = questions_by_iteration or {} 

43 

44 # On first iteration, decompose the query 

45 if not questions_by_iteration: 

46 return self._decompose_to_atomic_facts(query) 

47 

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

49 return self._generate_gap_filling_questions( 

50 query, 

51 current_knowledge, 

52 questions_by_iteration, 

53 questions_per_iteration, 

54 ) 

55 

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

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

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

59 

60Query: {query} 

61 

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

631. Be about ONE thing only 

642. Be searchable on its own 

653. Not depend on other facts 

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

67 

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

69- The geographical/geological aspect 

70- The naming aspect 

71- The historical events 

72- The statistical comparisons 

73 

74Return ONLY the questions, one per line. 

75Example format: 

76What locations were formed by glaciers? 

77What geographic features are named after body parts? 

78Where did falls occur between specific dates? 

79""" 

80 

81 response = invoke_llm_sync(self.model, prompt) 

82 

83 response_text = get_llm_response_text(response) 

84 

85 # Parse questions 

86 questions = [] 

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

88 line = line.strip() 

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

90 # Clean up any numbering or bullets 

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

92 if line.startswith(prefix): 

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

94 questions.append(line) 

95 

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

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

98 

99 def _generate_gap_filling_questions( 

100 self, 

101 original_query: str, 

102 current_knowledge: str, 

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

104 questions_per_iteration: int, 

105 ) -> List[str]: 

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

107 

108 # Check if we have enough information to start reasoning 

109 if len(questions_by_iteration) >= 3: 

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

111 

112Original Query: {original_query} 

113 

114Current Knowledge: 

115{current_knowledge} 

116 

117Previous Questions: 

118{self._format_previous_questions(questions_by_iteration)} 

119 

120Generate {questions_per_iteration} questions that: 

1211. Connect different facts you've found 

1222. Fill specific gaps in knowledge 

1233. Search for locations that match multiple criteria 

1244. Verify specific details 

125 

126Return ONLY the questions, one per line. 

127""" 

128 else: 

129 # Still gathering basic facts 

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

131 

132Original Query: {original_query} 

133 

134Previous Questions: 

135{self._format_previous_questions(questions_by_iteration)} 

136 

137Current Knowledge: 

138{current_knowledge} 

139 

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

141Focus on facts not yet explored. 

142 

143Return ONLY the questions, one per line. 

144""" 

145 

146 response = invoke_llm_sync(self.model, prompt) 

147 

148 response_text = get_llm_response_text(response) 

149 

150 # Parse questions 

151 questions = [] 

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

153 line = line.strip() 

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

155 # Clean up any numbering or bullets 

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

157 if line.startswith(prefix): 

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

159 questions.append(line) 

160 

161 return questions[:questions_per_iteration]