« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1from __future__ import annotations
3from datetime import UTC, datetime
4from typing import Any
6from lexigram.ai.rag.reasoning.base import (
7 AbstractReasoner,
8 ReasoningResult,
9 ReasoningStep,
10 ReasoningStrategy,
11)
12from lexigram.contracts import (
13 ChatMessage,
14 LLMClientProtocol,
15)
18class ChainOfThoughtReasoner(AbstractReasoner):
19 """Chain-of-thought reasoning for complex queries.
21 This reasoner generates explicit reasoning steps before arriving
22 at an answer, improving reasoning quality for complex questions.
23 """
25 def __init__(
26 self,
27 llm_client: LLMClientProtocol,
28 max_thoughts: int = 5,
29 temperature: float = 0.3,
30 ):
31 """Initialize chain-of-thought reasoner.
33 Args:
34 llm_client: LLM client for generation.
35 max_thoughts: Maximum number of thoughts to generate.
36 temperature: Temperature for LLM generation.
37 """
38 self.llm_client = llm_client
39 self.max_thoughts = max_thoughts
40 self.temperature = temperature
42 async def reason(
43 self,
44 query: str,
45 initial_context: list[Any] | None = None,
46 **kwargs,
47 ) -> ReasoningResult:
48 """Perform chain-of-thought reasoning without retrieval."""
49 context_str = ""
50 if initial_context:
51 context_texts = []
52 for doc in initial_context:
53 if hasattr(doc, "content"):
54 context_texts.append(doc.content)
55 elif isinstance(doc, dict) and "content" in doc:
56 context_texts.append(doc["content"])
57 elif isinstance(doc, str):
58 context_texts.append(doc)
60 context_str = "\n\n".join(
61 f"[{i + 1}] {ctx}" for i, ctx in enumerate(context_texts)
62 )
64 return await self.reason_with_context(query, context_str)
66 async def reason_with_context(
67 self,
68 query: str,
69 context: str = "",
70 ) -> ReasoningResult:
71 """Perform chain-of-thought reasoning with given context."""
72 prompt = f"Question: {query}\n\n"
74 if context:
75 prompt += f"Context:\n{context}\n\n"
77 prompt += """Please think through this step-by-step and show your reasoning:
79Let's approach this systematically:
80"""
82 result = await self.llm_client.complete(
83 messages=[
84 ChatMessage(
85 role="system",
86 content="You are a helpful assistant that thinks step-by-step and shows explicit reasoning before answering questions.",
87 ),
88 ChatMessage(role="user", content=prompt),
89 ],
90 temperature=self.temperature,
91 max_tokens=800,
92 )
93 if result.is_err():
94 raise result.unwrap_err()
95 response = result.unwrap()
97 # Extract text from response
98 if hasattr(response, "content"):
99 text = response.content
100 elif hasattr(response, "choices") and response.choices:
101 text = response.choices[0].message.content
102 elif isinstance(response, dict) and "content" in response:
103 text = response["content"]
104 else:
105 text = str(response)
107 # Parse thoughts and final answer
108 steps = self._parse_chain_of_thought(text)
110 # Extract final answer (usually last step or after "Therefore"/"In conclusion")
111 final_answer = self._extract_final_answer(text, steps)
113 return ReasoningResult(
114 query=query,
115 final_answer=final_answer,
116 steps=steps,
117 strategy=ReasoningStrategy.CHAIN_OF_THOUGHT,
118 total_hops=len(steps),
119 overall_confidence=0.8, # CoT generally has good confidence
120 metadata={
121 "max_thoughts": self.max_thoughts,
122 "has_context": bool(context),
123 "timestamp": datetime.now(UTC).isoformat(),
124 },
125 )
127 def _parse_chain_of_thought(self, text: str) -> list[ReasoningStep]:
128 """Parse chain of thought into reasoning steps."""
129 steps: list[ReasoningStep] = []
130 lines = text.strip().split("\n")
132 step_markers = ["Step", "Thought", "1.", "2.", "3.", "4.", "5.", "-", "•"]
133 current_step = None
134 step_num = 0
136 for raw_line in lines:
137 line = raw_line.strip()
138 if not line:
139 continue
141 # Check if this starts a new step
142 is_step = any(line.startswith(marker) for marker in step_markers)
144 if is_step:
145 if current_step:
146 steps.append(current_step)
148 step_num += 1
149 # Remove step marker
150 for marker in step_markers:
151 if line.startswith(marker):
152 line = line[len(marker) :].strip()
153 if line.startswith((".", ":")):
154 line = line[1:].strip()
155 break
157 current_step = ReasoningStep(
158 step_number=step_num,
159 question="",
160 reasoning=line,
161 answer="",
162 confidence=0.8,
163 )
164 elif current_step:
165 # Continue current step
166 current_step.reasoning += " " + line
168 # Add last step
169 if current_step:
170 steps.append(current_step)
172 return steps
174 def _extract_final_answer(self, text: str, steps: list[ReasoningStep]) -> str:
175 """Extract final answer from chain of thought."""
176 # Look for conclusion markers
177 conclusion_markers = [
178 "Therefore,",
179 "In conclusion,",
180 "The answer is",
181 "So,",
182 "Thus,",
183 "Hence,",
184 "Finally,",
185 ]
187 lines = text.strip().split("\n")
188 for i, line in enumerate(lines):
189 for marker in conclusion_markers:
190 if marker.lower() in line.lower():
191 # Return rest of this line and subsequent lines
192 answer_lines = [line]
193 for j in range(i + 1, min(i + 3, len(lines))):
194 if lines[j].strip():
195 answer_lines.append(lines[j])
196 return " ".join(answer_lines)
198 # If no conclusion marker, try to get last substantial sentence
199 if steps and steps[-1].reasoning:
200 return steps[-1].reasoning
202 # Fallback: return last non-empty line
203 for line in reversed(lines):
204 if line.strip() and len(line) > 20:
205 return line.strip()
207 return "Unable to extract final answer."