Coverage for src / local_deep_research / advanced_search_system / questions / standard_question.py: 42%

1""" 

2Standard question generation implementation. 

3""" 

4 

5from datetime import datetime, UTC 

6from typing import List 

7 

8from loguru import logger 

9 

10from .base_question import BaseQuestionGenerator 

11 

12 

13class StandardQuestionGenerator(BaseQuestionGenerator): 

14 """Standard question generator.""" 

15 

16 def generate_questions( 

17 self, 

18 current_knowledge: str, 

19 query: str, 

20 questions_per_iteration: int = 2, 

21 questions_by_iteration: dict = None, 

22 ) -> List[str]: 

23 """Generate follow-up questions based on current knowledge.""" 

24 now = datetime.now(UTC) 

25 current_time = now.strftime("%Y-%m-%d") 

26 questions_by_iteration = questions_by_iteration or {} 

27 

28 logger.info("Generating follow-up questions...") 

29 

30 if questions_by_iteration: 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true

31 prompt = f"""Critically reflect current knowledge (e.g., timeliness), what {questions_per_iteration} high-quality internet search questions remain unanswered to exactly answer the query? 

32 Query: {query} 

33 Today: {current_time} 

34 Past questions: {questions_by_iteration!s} 

35 Knowledge: {current_knowledge} 

36 Include questions that critically reflect current knowledge. 

37 \n\n\nFormat: One question per line, e.g. \n Q: question1 \n Q: question2\n\n""" 

38 else: 

39 prompt = f" You will have follow up questions. First, identify if your knowledge is outdated (high chance). Today: {current_time}. Generate {questions_per_iteration} high-quality internet search questions to exactly answer: {query}\n\n\nFormat: One question per line, e.g. \n Q: question1 \n Q: question2\n\n" 

40 

41 response = self.model.invoke(prompt) 

42 

43 # Handle both string responses and responses with .content attribute 

44 response_text = "" 

45 if hasattr(response, "content"): 45 ↛ 49line 45 didn't jump to line 49 because the condition on line 45 was always true

46 response_text = response.content 

47 else: 

48 # Handle string responses 

49 response_text = str(response) 

50 

51 questions = [ 

52 q.replace("Q:", "").strip() 

53 for q in response_text.split("\n") 

54 if q.strip().startswith("Q:") 

55 ][:questions_per_iteration] 

56 

57 logger.info(f"Generated {len(questions)} follow-up questions") 

58 

59 return questions 

60 

61 def generate_sub_questions( 

62 self, query: str, context: str = "" 

63 ) -> List[str]: 

64 """ 

65 Generate sub-questions from a main query. 

66 

67 Args: 

68 query: The main query to break down 

69 context: Additional context for question generation 

70 

71 Returns: 

72 List[str]: List of generated sub-questions 

73 """ 

74 prompt = f"""You are an expert at breaking down complex questions into simpler sub-questions. 

75 

76Original Question: {query} 

77 

78{context} 

79 

80Break down the original question into 2-5 simpler sub-questions that would help answer the original question when answered in sequence. 

81Follow these guidelines: 

821. Each sub-question should be specific and answerable on its own 

832. Sub-questions should build towards answering the original question 

843. For multi-hop or complex queries, identify the individual facts or entities needed 

854. Ensure the sub-questions can be answered with separate searches 

86 

87Format your response as a numbered list with ONLY the sub-questions, one per line: 

881. First sub-question 

892. Second sub-question 

90... 

91 

92Only provide the numbered sub-questions, nothing else.""" 

93 

94 try: 

95 response = self.model.invoke(prompt) 

96 

97 # Handle both string responses and responses with .content attribute 

98 content = "" 

99 if hasattr(response, "content"): 

100 content = response.content 

101 else: 

102 # Handle string responses 

103 content = str(response) 

104 

105 # Parse sub-questions from the response 

106 sub_questions = [] 

107 for line in content.strip().split("\n"): 

108 line = line.strip() 

109 if line and (line[0].isdigit() or line.startswith("-")): 

110 # Extract sub-question from numbered or bulleted list 

111 parts = ( 

112 line.split(".", 1) 

113 if "." in line 

114 else line.split(" ", 1) 

115 ) 

116 if len(parts) > 1: 

117 sub_question = parts[1].strip() 

118 sub_questions.append(sub_question) 

119 

120 # Limit to at most 5 sub-questions 

121 return sub_questions[:5] 

122 except Exception as e: 

123 logger.exception(f"Error generating sub-questions: {e!s}") 

124 return []