Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/synthesis/synthesizers/abstractive.py: 16%

1"""Abstractive synthesizer implementation. 

2 

3This module implements an LLM-based abstractive synthesizer that generates 

4new responses from context chunks. 

5""" 

6 

7from __future__ import annotations 

8 

9from lexigram.ai.rag.synthesis.synthesizers.base import AbstractSynthesizer 

10from lexigram.ai.rag.synthesis.types import ( 

11 ContextChunk, 

12 SynthesisResult, 

13 SynthesisStrategy, 

14) 

15from lexigram.contracts import ( 

16 LLMClientProtocol, 

17) 

18from lexigram.contracts.ai.llm import ChatMessage, Role 

19 

20 

21class AbstractiveSynthesizer(AbstractSynthesizer): 

22 """LLM-based abstractive synthesizer. 

23 

24 This synthesizer uses an LLM to generate a new response based on the 

25 retrieved context chunks and query. 

26 

27 Attributes: 

28 llm_client: LLM client for generation 

29 max_context_chunks: Maximum number of chunks to include 

30 include_citations: Whether to ask for citations 

31 temperature: LLM temperature (0-1) 

32 """ 

33 

34 def __init__( 

35 self, 

36 llm_client: LLMClientProtocol, 

37 max_context_chunks: int = 5, 

38 include_citations: bool = True, 

39 temperature: float = 0.3, 

40 ): 

41 """Initialize the abstractive synthesizer. 

42 

43 Args: 

44 llm_client: LLM client implementing LLMClientProtocol protocol 

45 max_context_chunks: Maximum context chunks to use 

46 include_citations: Whether to include citations 

47 temperature: LLM temperature for generation 

48 """ 

49 self.llm_client = llm_client 

50 self.max_context_chunks = max_context_chunks 

51 self.include_citations = include_citations 

52 self.temperature = temperature 

53 

54 def _build_prompt( 

55 self, 

56 query: str, 

57 context_chunks: list[ContextChunk], 

58 ) -> str: 

59 """Build the synthesis prompt. 

60 

61 Args: 

62 query: The user query 

63 context_chunks: Context chunks to use 

64 

65 Returns: 

66 Formatted prompt string 

67 """ 

68 # Limit chunks 

69 chunks_to_use = context_chunks[: self.max_context_chunks] 

70 

71 # Build context section 

72 context_parts = [] 

73 for i, chunk in enumerate(chunks_to_use, 1): 

74 context_parts.append(f"[{i}] {chunk.text}") 

75 if chunk.source: 

76 context_parts.append(f" Source: {chunk.source}") 

77 

78 context_text = "\n\n".join(context_parts) 

79 

80 # Build prompt 

81 prompt = f"""You are a helpful assistant that answers questions based on provided context. 

82 

83Context: 

84{context_text} 

85 

86Question: {query} 

87 

88Instructions: 

89- Answer the question using ONLY information from the provided context 

90- Be concise and accurate 

91- If the context doesn't contain enough information, say so 

92- Do not make up or infer information not present in the context""" 

93 

94 if self.include_citations: 

95 prompt += "\n- Cite sources using [1], [2], etc. when referring to specific information" 

96 

97 prompt += "\n\nAnswer:" 

98 

99 return prompt 

100 

101 async def _synthesize_internal( 

102 self, 

103 query: str, 

104 context_chunks: list[ContextChunk], 

105 **kwargs, 

106 ) -> SynthesisResult: 

107 """Synthesize response using LLM. 

108 

109 Args: 

110 query: The user query 

111 context_chunks: Retrieved context chunks 

112 **kwargs: Additional parameters (e.g., temperature) 

113 

114 Returns: 

115 SynthesisResult with LLM-generated response 

116 

117 Raises: 

118 ValueError: If query is empty or no context chunks provided 

119 """ 

120 if not query: 

121 msg = "Query cannot be empty" 

122 raise ValueError(msg) 

123 if not context_chunks: 

124 msg = "No context chunks provided" 

125 raise ValueError(msg) 

126 

127 # Build prompt 

128 prompt = self._build_prompt(query, context_chunks) 

129 

130 # Get temperature from kwargs or use default 

131 temperature = kwargs.get("temperature", self.temperature) 

132 

133 # Generate response using LLM 

134 messages = [ChatMessage(role=Role.USER, content=prompt)] 

135 

136 try: 

137 result = await self.llm_client.complete( 

138 messages, 

139 temperature=temperature, 

140 max_tokens=kwargs.get("max_tokens", 500), 

141 ) 

142 if result.is_err(): 

143 raise result.unwrap_err() 

144 response = result.unwrap() 

145 except Exception as e: 

146 msg = f"LLM generation failed: {e}" 

147 raise RuntimeError(msg) from e 

148 

149 response_text_str = response.content 

150 

151 # Extract citations if included 

152 citations = [] 

153 if self.include_citations: 

154 # Simple citation extraction (can be improved) 

155 import re 

156 

157 citation_pattern = r"\[(\d+)\]" 

158 cited_numbers = set(re.findall(citation_pattern, response_text_str)) 

159 

160 for num_str in cited_numbers: 

161 num = int(num_str) 

162 if 1 <= num <= len(context_chunks): 

163 chunk = context_chunks[num - 1] 

164 citations.append( 

165 { 

166 "number": num, 

167 "source": chunk.source, 

168 "score": chunk.score, 

169 }, 

170 ) 

171 

172 # Determine which chunks were actually used 

173 chunks_used = context_chunks[: self.max_context_chunks] 

174 

175 return SynthesisResult( 

176 query=query, 

177 response=response_text_str.strip(), 

178 strategy=SynthesisStrategy.ABSTRACTIVE, 

179 context_chunks=chunks_used, 

180 citations=citations, 

181 metadata={ 

182 "num_chunks_provided": len(chunks_used), 

183 "temperature": temperature, 

184 "include_citations": self.include_citations, 

185 "prompt_length": len(prompt), 

186 }, 

187 )