Coverage for src / local_deep_research / advanced_search_system / knowledge / standard_knowledge.py: 20%

1""" 

2Standard knowledge generator implementation. 

3""" 

4 

5from loguru import logger 

6from datetime import datetime, UTC 

7from typing import List 

8 

9from .base_knowledge import BaseKnowledgeGenerator 

10 

11 

12class StandardKnowledge(BaseKnowledgeGenerator): 

13 """Standard knowledge generator implementation.""" 

14 

15 def generate_knowledge( 

16 self, 

17 query: str, 

18 context: str = "", 

19 current_knowledge: str = "", 

20 questions: List[str] = None, 

21 ) -> str: 

22 """Generate knowledge based on query and context.""" 

23 now = datetime.now(UTC) 

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

25 

26 logger.info("Generating knowledge...") 

27 

28 if questions: 

29 prompt = f"""Based on the following query and questions, generate comprehensive knowledge: 

30 

31Query: {query} 

32Current Time: {current_time} 

33Context: {context} 

34Current Knowledge: {current_knowledge} 

35Questions: {questions} 

36 

37Generate detailed knowledge that: 

381. Directly answers the query 

392. Addresses each question 

403. Includes relevant facts and details 

414. Is up-to-date with current information 

425. Synthesizes information from multiple sources 

43 

44Format your response as a well-structured paragraph.""" 

45 else: 

46 prompt = f"""Based on the following query, generate comprehensive knowledge: 

47 

48Query: {query} 

49Current Time: {current_time} 

50Context: {context} 

51Current Knowledge: {current_knowledge} 

52 

53Generate detailed knowledge that: 

541. Directly answers the query 

552. Includes relevant facts and details 

563. Is up-to-date with current information 

574. Synthesizes information from multiple sources 

58 

59Format your response as a well-structured paragraph.""" 

60 

61 response = self.model.invoke(prompt) 

62 knowledge = response.content 

63 

64 logger.info("Generated knowledge successfully") 

65 return knowledge 

66 

67 def generate_sub_knowledge(self, sub_query: str, context: str = "") -> str: 

68 """ 

69 Generate knowledge for a sub-question. 

70 

71 Args: 

72 sub_query: The sub-question to generate knowledge for 

73 context: Additional context for knowledge generation 

74 

75 Returns: 

76 str: Generated knowledge for the sub-question 

77 """ 

78 prompt = f"""Generate comprehensive knowledge to answer this sub-question: 

79 

80Sub-question: {sub_query} 

81 

82{context} 

83 

84Generate detailed knowledge that: 

851. Directly answers the sub-question 

862. Includes relevant facts and details 

873. Is up-to-date with current information 

884. Synthesizes information from multiple sources 

89 

90Format your response as a well-structured paragraph.""" 

91 

92 try: 

93 response = self.model.invoke(prompt) 

94 return response.content 

95 except Exception as e: 

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

97 return "" 

98 

99 def generate(self, query: str, context: str) -> str: 

100 """Generate knowledge from the given query and context.""" 

101 return self.generate_knowledge(query, context) 

102 

103 def compress_knowledge( 

104 self, current_knowledge: str, query: str, section_links: list, **kwargs 

105 ) -> str: 

106 """ 

107 Compress and summarize accumulated knowledge. 

108 

109 Args: 

110 current_knowledge: The accumulated knowledge to compress 

111 query: The original research query 

112 section_links: List of source links 

113 **kwargs: Additional arguments 

114 

115 Returns: 

116 str: Compressed knowledge 

117 """ 

118 logger.info( 

119 f"Compressing knowledge for query: {query}. Original length: {len(current_knowledge)}" 

120 ) 

121 

122 prompt = f"""Compress the following accumulated knowledge relevant to the query '{query}'. 

123Retain the key facts, findings, and citations. Remove redundancy. 

124 

125Accumulated Knowledge: 

126{current_knowledge} 

127 

128Compressed Knowledge:""" 

129 

130 try: 

131 response = self.model.invoke(prompt) 

132 compressed_knowledge = response.content 

133 logger.info( 

134 f"Compressed knowledge length: {len(compressed_knowledge)}" 

135 ) 

136 return compressed_knowledge 

137 except Exception as e: 

138 logger.exception(f"Error compressing knowledge: {e!s}") 

139 return current_knowledge # Return original if compression fails 

140 

141 def format_citations(self, links: List[str]) -> str: 

142 """ 

143 Format source links into citations using IEEE style. 

144 

145 Args: 

146 links: List of source links 

147 

148 Returns: 

149 str: Formatted citations in IEEE style 

150 """ 

151 if not links: 

152 return "" 

153 

154 # Format each link as an IEEE citation 

155 citations = [] 

156 for i, link in enumerate(links, 1): 

157 citations.append(f"[{i}] {link}") 

158 

159 return "\n".join(citations)