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

1""" 

2Standard question generation implementation. 

3""" 

4 

5from datetime import datetime, UTC 

6from typing import List, Optional 

7 

8from loguru import logger 

9 

10from ...utilities.json_utils import get_llm_response_text 

11from .base_question import BaseQuestionGenerator 

12from ...utilities.llm_utils import invoke_llm_sync 

13 

14 

15class StandardQuestionGenerator(BaseQuestionGenerator): 

16 """Standard question generator.""" 

17 

18 def generate_questions( 

19 self, 

20 current_knowledge: str, 

21 query: str, 

22 questions_per_iteration: int = 2, 

23 questions_by_iteration: Optional[dict] = None, 

24 ) -> List[str]: 

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

26 now = datetime.now(UTC) 

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

28 questions_by_iteration = questions_by_iteration or {} 

29 

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

31 

32 if questions_by_iteration: 

33 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? 

34 Query: {query} 

35 Today: {current_time} 

36 Past questions: {questions_by_iteration!s} 

37 Knowledge: {current_knowledge} 

38 Include questions that critically reflect current knowledge. 

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

40 else: 

41 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" 

42 

43 response = invoke_llm_sync(self.model, prompt) 

44 

45 response_text = get_llm_response_text(response) 

46 

47 questions = [ 

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

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

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

51 ][:questions_per_iteration] 

52 

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

54 

55 return questions 

56 

57 def generate_sub_questions( 

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

59 ) -> List[str]: 

60 """ 

61 Generate sub-questions from a main query. 

62 

63 Args: 

64 query: The main query to break down 

65 context: Additional context for question generation 

66 

67 Returns: 

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

69 """ 

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

71 

72Original Question: {query} 

73 

74{context} 

75 

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

77Follow these guidelines: 

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

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

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

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

82 

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

841. First sub-question 

852. Second sub-question 

86... 

87 

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

89 

90 try: 

91 response = invoke_llm_sync(self.model, prompt) 

92 

93 content = get_llm_response_text(response) 

94 

95 # Parse sub-questions from the response 

96 sub_questions = [] 

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

98 line = line.strip() 

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

100 # Extract sub-question from numbered or bulleted list 

101 parts = ( 

102 line.split(".", 1) 

103 if "." in line 

104 else line.split(" ", 1) 

105 ) 

106 if len(parts) > 1: 106 ↛ 97line 106 didn't jump to line 97 because the condition on line 106 was always true

107 sub_question = parts[1].strip() 

108 sub_questions.append(sub_question) 

109 

110 # Limit to at most 5 sub-questions 

111 return sub_questions[:5] 

112 except Exception: 

113 logger.exception("Error generating sub-questions") 

114 return []