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

1""" 

2Standard knowledge generator implementation. 

3""" 

4 

5from loguru import logger 

6from datetime import datetime, UTC 

7from typing import List 

8 

9from ...utilities.json_utils import get_llm_response_text 

10from .base_knowledge import BaseKnowledgeGenerator 

11 

12 

13class StandardKnowledge(BaseKnowledgeGenerator): 

14 """Standard knowledge generator implementation.""" 

15 

16 def generate_knowledge( 

17 self, 

18 query: str, 

19 context: str = "", 

20 current_knowledge: str = "", 

21 questions: List[str] = None, 

22 ) -> str: 

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

24 now = datetime.now(UTC) 

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

26 

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

28 

29 if questions: 

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

31 

32Query: {query} 

33Current Time: {current_time} 

34Context: {context} 

35Current Knowledge: {current_knowledge} 

36Questions: {questions} 

37 

38Generate detailed knowledge that: 

391. Directly answers the query 

402. Addresses each question 

413. Includes relevant facts and details 

424. Is up-to-date with current information 

435. Synthesizes information from multiple sources 

44 

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

46 else: 

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

48 

49Query: {query} 

50Current Time: {current_time} 

51Context: {context} 

52Current Knowledge: {current_knowledge} 

53 

54Generate detailed knowledge that: 

551. Directly answers the query 

562. Includes relevant facts and details 

573. Is up-to-date with current information 

584. Synthesizes information from multiple sources 

59 

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

61 

62 knowledge = get_llm_response_text(self.model.invoke(prompt)) 

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 return get_llm_response_text(self.model.invoke(prompt)) 

94 except Exception: 

95 logger.exception("Error generating sub-knowledge") 

96 return "" 

97 

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

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

100 return self.generate_knowledge(query, context) 

101 

102 def compress_knowledge( 

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

104 ) -> str: 

105 """ 

106 Compress and summarize accumulated knowledge. 

107 

108 Args: 

109 current_knowledge: The accumulated knowledge to compress 

110 query: The original research query 

111 section_links: List of source links 

112 **kwargs: Additional arguments 

113 

114 Returns: 

115 str: Compressed knowledge 

116 """ 

117 logger.info( 

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

119 ) 

120 

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

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

123 

124Accumulated Knowledge: 

125{current_knowledge} 

126 

127Compressed Knowledge:""" 

128 

129 try: 

130 compressed_knowledge = get_llm_response_text( 

131 self.model.invoke(prompt) 

132 ) 

133 logger.info( 

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

135 ) 

136 return compressed_knowledge 

137 except Exception: 

138 logger.exception("Error compressing knowledge") 

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)