Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/reasoning/multi_hop.py: 87%

1"""Multi-hop reasoner implementation split into its own module. 

2 

3This module contains the concrete MultiHopReasoner implementation and 

4convenience entrypoint `multi_hop_reason`. 

5""" 

6 

7from __future__ import annotations 

8 

9from datetime import UTC, datetime 

10from typing import TYPE_CHECKING, Any 

11 

12from lexigram.ai.rag.reasoning.base import ( 

13 AbstractReasoner, 

14 ReasoningResult, 

15 ReasoningStep, 

16 ReasoningStrategy, 

17) 

18from lexigram.contracts import ChatMessage 

19 

20if TYPE_CHECKING: 

21 from lexigram.contracts import LLMClientProtocol 

22 from lexigram.contracts.data.vector.protocols import ( 

23 VectorCollectionProtocol, 

24 VectorStoreProtocol, 

25 ) 

26 

27 

28class MultiHopReasoner(AbstractReasoner): 

29 """Multi-hop reasoning over multiple retrieval steps. 

30 

31 This reasoner breaks down complex queries into multiple hops, 

32 where each hop retrieves relevant information and builds upon 

33 previous steps. 

34 """ 

35 

36 def __init__( 

37 self, 

38 llm_client: LLMClientProtocol, 

39 vector_store: VectorCollectionProtocol, 

40 max_hops: int = 3, 

41 top_k_per_hop: int = 3, 

42 confidence_threshold: float = 0.5, 

43 temperature: float = 0.3, 

44 ): 

45 self.llm_client = llm_client 

46 self.vector_store = vector_store 

47 self.max_hops = max_hops 

48 self.top_k_per_hop = top_k_per_hop 

49 self.confidence_threshold = confidence_threshold 

50 self.temperature = temperature 

51 

52 async def reason( 

53 self, 

54 query: str, 

55 initial_context: list[Any] | None = None, 

56 **kwargs, 

57 ) -> ReasoningResult: 

58 steps: list[ReasoningStep] = [] 

59 current_query = query 

60 accumulated_knowledge = [] 

61 

62 if initial_context: 

63 accumulated_knowledge.extend(initial_context) 

64 

65 for hop in range(1, self.max_hops + 1): 

66 retrieved_docs = await self.vector_store.search( # type: ignore[call-arg] 

67 query=current_query, # type: ignore[arg-type] 

68 limit=self.top_k_per_hop, 

69 ) 

70 

71 context_texts: list[str] = [] 

72 for doc in retrieved_docs: 

73 if hasattr(doc, "content"): 

74 text = doc.content 

75 if text is not None: 

76 context_texts.append(text) 

77 elif isinstance(doc, dict) and "content" in doc: 

78 context_texts.append(doc["content"]) 

79 elif isinstance(doc, str): 

80 context_texts.append(doc) 

81 

82 reasoning_prompt = self._build_reasoning_prompt( 

83 original_query=query, 

84 current_query=current_query, 

85 context=context_texts, 

86 previous_steps=steps, 

87 hop_number=hop, 

88 ) 

89 

90 result = await self.llm_client.complete( 

91 messages=[ 

92 ChatMessage( 

93 role="system", 

94 content="You are a helpful assistant that performs step-by-step reasoning to answer complex questions.", 

95 ), 

96 ChatMessage(role="user", content=reasoning_prompt), 

97 ], 

98 temperature=self.temperature, 

99 max_tokens=500, 

100 ) 

101 if result.is_err(): 

102 raise result.unwrap_err() 

103 response = result.unwrap() 

104 

105 step_result = self._parse_reasoning_response(response, hop) 

106 

107 step = ReasoningStep( 

108 step_number=hop, 

109 question=current_query, 

110 context=retrieved_docs, 

111 reasoning=step_result.get("reasoning", ""), 

112 answer=step_result.get("answer", ""), 

113 confidence=step_result.get("confidence", 0.5), 

114 metadata={ 

115 "num_docs": len(retrieved_docs), 

116 "timestamp": datetime.now(UTC).isoformat(), 

117 }, 

118 ) 

119 steps.append(step) 

120 

121 accumulated_knowledge.append(step.answer) 

122 

123 if step_result.get("is_final", False): 

124 break 

125 

126 if step.confidence < self.confidence_threshold: 

127 break 

128 

129 next_query = step_result.get("next_question") 

130 if not next_query or next_query == current_query: 

131 break 

132 

133 current_query = next_query 

134 

135 final_answer = await self._generate_final_answer(query, steps) 

136 

137 overall_confidence = ( 

138 sum(step.confidence for step in steps) / len(steps) if steps else 0.0 

139 ) 

140 

141 return ReasoningResult( 

142 query=query, 

143 final_answer=final_answer, 

144 steps=steps, 

145 strategy=ReasoningStrategy.MULTI_HOP, 

146 total_hops=len(steps), 

147 overall_confidence=overall_confidence, 

148 metadata={ 

149 "max_hops": self.max_hops, 

150 "top_k_per_hop": self.top_k_per_hop, 

151 "timestamp": datetime.now(UTC).isoformat(), 

152 }, 

153 ) 

154 

155 def _build_reasoning_prompt( 

156 self, 

157 original_query: str, 

158 current_query: str, 

159 context: list[str], 

160 previous_steps: list[ReasoningStep], 

161 hop_number: int, 

162 ) -> str: 

163 prompt = f"Original Question: {original_query}\n\n" 

164 

165 if previous_steps: 

166 prompt += "Previous Reasoning Steps:\n" 

167 for step in previous_steps: 

168 prompt += f"Step {step.step_number}: {step.question}\n" 

169 prompt += f"Answer: {step.answer}\n\n" 

170 

171 prompt += f"Current Question (Step {hop_number}): {current_query}\n\n" 

172 

173 prompt += "Retrieved Context:\n" 

174 for i, ctx in enumerate(context, 1): 

175 prompt += f"[{i}] {ctx}\n\n" 

176 

177 prompt += """Please analyze this step and provide: 

1781. Your reasoning about what information is relevant 

1792. An answer to the current question based on the context 

1803. A confidence score (0.0 to 1.0) 

1814. Whether this is the final answer or if more steps are needed 

1825. If not final, what the next question should be 

183 

184Format your response as: 

185REASONING: <your reasoning> 

186ANSWER: <answer to current question> 

187CONFIDENCE: <0.0-1.0> 

188IS_FINAL: <yes/no> 

189NEXT_QUESTION: <next question if not final> 

190""" 

191 return prompt 

192 

193 def _parse_reasoning_response(self, response: Any, hop: int) -> dict: 

194 """Extract text from a provider response and parse it using the 

195 shared parsing helper in `reasoning.parsers`. 

196 """ 

197 if hasattr(response, "content"): 

198 text = response.content 

199 elif hasattr(response, "choices") and response.choices: 

200 text = response.choices[0].message.content 

201 elif isinstance(response, dict) and "content" in response: 

202 text = response["content"] 

203 else: 

204 text = str(response) 

205 

206 # Import locally to avoid circular import during package initialization 

207 from lexigram.ai.rag.reasoning.parsers import parse_reasoning_response_text 

208 

209 return parse_reasoning_response_text(text) 

210 

211 async def _generate_final_answer( 

212 self, 

213 query: str, 

214 steps: list[ReasoningStep], 

215 ) -> str: 

216 if not steps: 

217 return "Unable to answer the question with available information." 

218 

219 if steps[-1].answer: 

220 return steps[-1].answer 

221 

222 return "Unable to generate final answer." 

223 

224 

225async def multi_hop_reason( 

226 query: str, 

227 llm_client: LLMClientProtocol, 

228 vector_store: VectorStoreProtocol, 

229 strategy: ReasoningStrategy = ReasoningStrategy.MULTI_HOP, 

230 **kwargs, 

231) -> ReasoningResult: 

232 from lexigram.ai.rag.reasoning.strategy_registry import ( 

233 ReasoningStrategyRegistry, 

234 ) 

235 

236 registry = ReasoningStrategyRegistry.with_defaults() 

237 return await registry.reason( 

238 strategy, 

239 llm_client, 

240 vector_store, # type: ignore[arg-type] 

241 query, 

242 kwargs, 

243 )