Coverage for src / local_deep_research / advanced_search_system / knowledge / base_knowledge.py: 47%

1""" 

2Base class for knowledge extraction and generation. 

3""" 

4 

5from loguru import logger 

6from abc import ABC, abstractmethod 

7from typing import List 

8 

9from langchain_core.language_models.chat_models import BaseChatModel 

10 

11 

12class BaseKnowledgeGenerator(ABC): 

13 """Base class for generating knowledge from text.""" 

14 

15 def __init__(self, model: BaseChatModel): 

16 """ 

17 Initialize the knowledge generator. 

18 

19 Args: 

20 model: The language model to use 

21 """ 

22 self.model = model 

23 

24 @abstractmethod 

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

26 """ 

27 Generate knowledge from the given query and context. 

28 

29 Args: 

30 query: The query to generate knowledge for 

31 context: Additional context for knowledge generation 

32 

33 Returns: 

34 """ 

35 pass 

36 

37 @abstractmethod 

38 def generate_knowledge( 

39 self, 

40 query: str, 

41 context: str = "", 

42 current_knowledge: str = "", 

43 questions: List[str] = None, 

44 ) -> str: 

45 """ 

46 Generate knowledge based on query and context. 

47 

48 Args: 

49 query: The query to generate knowledge for 

50 context: Additional context for knowledge generation 

51 current_knowledge: Current accumulated knowledge 

52 questions: List of questions to address 

53 

54 Returns: 

55 str: Generated knowledge 

56 """ 

57 pass 

58 

59 @abstractmethod 

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

61 """ 

62 Generate knowledge for a sub-question. 

63 

64 Args: 

65 sub_query: The sub-question to generate knowledge for 

66 context: Additional context for knowledge generation 

67 

68 Returns: 

69 str: Generated knowledge for the sub-question 

70 """ 

71 pass 

72 

73 @abstractmethod 

74 def compress_knowledge( 

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

76 ) -> str: 

77 """ 

78 Compress and summarize accumulated knowledge. 

79 

80 Args: 

81 current_knowledge: The accumulated knowledge to compress 

82 query: The original research query 

83 section_links: List of source links 

84 **kwargs: Additional arguments 

85 

86 Returns: 

87 str: Compressed knowledge 

88 """ 

89 pass 

90 

91 @abstractmethod 

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

93 """ 

94 Format source links into citations. 

95 

96 Args: 

97 links: List of source links 

98 

99 Returns: 

100 str: Formatted citations 

101 """ 

102 pass 

103 

104 def _validate_knowledge(self, knowledge: str) -> bool: 

105 """ 

106 Validate the knowledge input. 

107 

108 Args: 

109 knowledge: The knowledge to validate 

110 

111 Returns: 

112 bool: True if knowledge is valid, False otherwise 

113 """ 

114 if not knowledge or not isinstance(knowledge, str): 

115 logger.error("Invalid knowledge provided") 

116 return False 

117 return True 

118 

119 def _validate_links(self, links: List[str]) -> bool: 

120 """ 

121 Validate the source links. 

122 

123 Args: 

124 links: List of source links to validate 

125 

126 Returns: 

127 bool: True if links are valid, False otherwise 

128 """ 

129 if not isinstance(links, list): 

130 logger.error("Invalid links format") 

131 return False 

132 if not all(isinstance(link, str) for link in links): 

133 logger.error("Invalid link type in links list") 

134 return False 

135 return True 

136 

137 def _extract_key_points(self, knowledge: str) -> List[str]: 

138 """ 

139 Extract key points from knowledge. 

140 

141 Args: 

142 knowledge: The knowledge to analyze 

143 

144 Returns: 

145 List[str]: List of key points 

146 """ 

147 # This is a placeholder implementation 

148 # Specific implementations should override this method 

149 return knowledge.split("\n")