"""Chain of Thought reasoning implementation with advanced features.""" import logging from typing import Dict, Any, List, Optional, Tuple import json from dataclasses import dataclass from enum import Enum from .base import ReasoningStrategy class ThoughtType(Enum): """Types of thoughts in the chain.""" OBSERVATION = "observation" ANALYSIS = "analysis" HYPOTHESIS = "hypothesis" VERIFICATION = "verification" CONCLUSION = "conclusion" REFLECTION = "reflection" REFINEMENT = "refinement" @dataclass class Thought: """Represents a single thought in the chain.""" type: ThoughtType content: str confidence: float evidence: List[str] alternatives: List[str] next_steps: List[str] metadata: Dict[str, Any] class ChainOfThoughtStrategy(ReasoningStrategy): """ Advanced Chain of Thought reasoning implementation with: - Hierarchical thought chains - Confidence scoring - Alternative path exploration - Self-reflection and refinement - Evidence tracking - Meta-learning capabilities """ def __init__(self, min_confidence: float = 0.7, parallel_threshold: int = 3, learning_rate: float = 0.1, strategy_weights: Optional[Dict[str, float]] = None): self.min_confidence = min_confidence self.parallel_threshold = parallel_threshold self.learning_rate = learning_rate self.strategy_weights = strategy_weights or { "LOCAL_LLM": 0.8, "CHAIN_OF_THOUGHT": 0.6, "TREE_OF_THOUGHTS": 0.5, "META_LEARNING": 0.4 } self.thought_history: List[Thought] = [] async def reason(self, query: str, context: Dict[str, Any]) -> Dict[str, Any]: """Main reasoning method implementing chain of thought.""" try: # Initialize reasoning chain chain = await self._initialize_chain(query, context) # Generate initial thoughts thoughts = await self._generate_thoughts(query, context) # Build thought chain chain = await self._build_chain(thoughts, context) # Reflect and refine if self.enable_reflection: chain = await self._reflect_and_refine(chain, context) # Extract conclusion conclusion = await self._extract_conclusion(chain, context) # Update thought history self.thought_history.extend(chain) return { "success": True, "answer": conclusion["answer"], "confidence": conclusion["confidence"], "reasoning_chain": [self._thought_to_dict(t) for t in chain], "alternatives": conclusion["alternatives"], "evidence": conclusion["evidence"], "meta_insights": conclusion["meta_insights"] } except Exception as e: logging.error(f"Error in chain of thought reasoning: {str(e)}") return {"success": False, "error": str(e)} async def _initialize_chain(self, query: str, context: Dict[str, Any]) -> List[Thought]: """Initialize the thought chain with observations.""" prompt = f""" Initialize chain of thought for query: Query: {query} Context: {json.dumps(context)} Provide initial observations: 1. Key elements in query 2. Relevant context factors 3. Initial hypotheses 4. Potential approaches Format as: [O1] Element: ... | Relevance: ... | Confidence: ... [O2] Context: ... | Impact: ... | Confidence: ... [O3] Hypothesis: ... | Support: ... | Confidence: ... [O4] Approach: ... | Rationale: ... | Confidence: ... """ response = await context["groq_api"].predict(prompt) return self._parse_observations(response["answer"]) async def _generate_thoughts(self, query: str, context: Dict[str, Any]) -> List[Thought]: """Generate candidate thoughts for the chain.""" prompt = f""" Generate thoughts for query analysis: Query: {query} Context: {json.dumps(context)} For each thought provide: 1. [Type]: {" | ".join([t.value for t in ThoughtType])} 2. [Content]: Main thought 3. [Evidence]: Supporting evidence 4. [Alternatives]: Alternative perspectives 5. [Next]: Potential next steps 6. [Confidence]: 0-1 score Format as: [T1] Type: ... Content: ... Evidence: ... Alternatives: ... Next: ... Confidence: ... """ response = await context["groq_api"].predict(prompt) return self._parse_thoughts(response["answer"]) async def _build_chain(self, thoughts: List[Thought], context: Dict[str, Any]) -> List[Thought]: """Build coherent chain from candidate thoughts.""" prompt = f""" Build coherent thought chain: Thoughts: {json.dumps([self._thought_to_dict(t) for t in thoughts])} Context: {json.dumps(context)} For each step specify: 1. Selected thought 2. Reasoning for selection 3. Connection to previous 4. Expected impact Format as: [S1] Thought: ... Reason: ... Connection: ... Impact: ... """ response = await context["groq_api"].predict(prompt) return self._parse_chain(response["answer"], thoughts) async def _reflect_and_refine(self, chain: List[Thought], context: Dict[str, Any]) -> List[Thought]: """Reflect on and refine the thought chain.""" prompt = f""" Reflect on thought chain: Chain: {json.dumps([self._thought_to_dict(t) for t in chain])} Context: {json.dumps(context)} Analyze for: 1. Logical gaps 2. Weak assumptions 3. Missing evidence 4. Alternative perspectives Suggest refinements: 1. Additional thoughts 2. Modified reasoning 3. New connections 4. Evidence needs Format as: [Analysis] Gaps: ... Assumptions: ... Missing: ... Alternatives: ... [Refinements] Thoughts: ... Reasoning: ... Connections: ... Evidence: ... """ response = await context["groq_api"].predict(prompt) return self._apply_refinements(chain, response["answer"]) async def _extract_conclusion(self, chain: List[Thought], context: Dict[str, Any]) -> Dict[str, Any]: """Extract final conclusion from thought chain.""" prompt = f""" Extract conclusion from thought chain: Chain: {json.dumps([self._thought_to_dict(t) for t in chain])} Context: {json.dumps(context)} Provide: 1. Main conclusion 2. Confidence level 3. Supporting evidence 4. Alternative conclusions 5. Meta-insights gained 6. Future considerations Format as: [Conclusion] Answer: ... Confidence: ... Evidence: ... Alternatives: ... [Meta] Insights: ... Future: ... """ response = await context["groq_api"].predict(prompt) return self._parse_conclusion(response["answer"]) def _parse_observations(self, response: str) -> List[Thought]: """Parse initial observations into thoughts.""" observations = [] lines = response.split('\n') for line in lines: if line.startswith('[O'): parts = line.split('|') if len(parts) >= 3: main_part = parts[0].split(']')[1].strip() key, content = main_part.split(':', 1) evidence = [p.strip() for p in parts[1].split(':')[1].strip().split(',')] try: confidence = float(parts[2].split(':')[1].strip()) except: confidence = 0.5 observations.append(Thought( type=ThoughtType.OBSERVATION, content=content.strip(), confidence=confidence, evidence=evidence, alternatives=[], next_steps=[], metadata={"key": key} )) return observations def _parse_thoughts(self, response: str) -> List[Thought]: """Parse generated thoughts.""" thoughts = [] current = None for line in response.split('\n'): line = line.strip() if not line: continue if line.startswith('[T'): if current: thoughts.append(current) current = None elif line.startswith('Type:'): type_str = line[5:].strip() try: thought_type = ThoughtType(type_str.lower()) current = Thought( type=thought_type, content="", confidence=0.0, evidence=[], alternatives=[], next_steps=[], metadata={} ) except ValueError: logging.warning(f"Invalid thought type: {type_str}") elif current: if line.startswith('Content:'): current.content = line[8:].strip() elif line.startswith('Evidence:'): current.evidence = [e.strip() for e in line[9:].split(',')] elif line.startswith('Alternatives:'): current.alternatives = [a.strip() for a in line[13:].split(',')] elif line.startswith('Next:'): current.next_steps = [n.strip() for n in line[5:].split(',')] elif line.startswith('Confidence:'): try: current.confidence = float(line[11:].strip()) except: current.confidence = 0.5 if current: thoughts.append(current) return thoughts def _parse_chain(self, response: str, thoughts: List[Thought]) -> List[Thought]: """Parse and order thoughts into a chain.""" chain = [] thought_map = {self._thought_to_dict(t)["content"]: t for t in thoughts} for line in response.split('\n'): if line.startswith('Thought:'): content = line[8:].strip() if content in thought_map: chain.append(thought_map[content]) return chain def _apply_refinements(self, chain: List[Thought], response: str) -> List[Thought]: """Apply refinements to thought chain.""" refined_chain = chain.copy() # Parse refinements sections = response.split('[') for section in sections: if section.startswith('Refinements]'): lines = section.split('\n')[1:] for line in lines: if line.startswith('Thoughts:'): new_thoughts = self._parse_refinement_thoughts(line[9:]) refined_chain.extend(new_thoughts) return refined_chain def _parse_refinement_thoughts(self, refinements: str) -> List[Thought]: """Parse refinement thoughts.""" thoughts = [] for refinement in refinements.split(';'): if refinement.strip(): thoughts.append(Thought( type=ThoughtType.REFINEMENT, content=refinement.strip(), confidence=0.8, # Refinements typically have high confidence evidence=[], alternatives=[], next_steps=[], metadata={"refined": True} )) return thoughts def _parse_conclusion(self, response: str) -> Dict[str, Any]: """Parse final conclusion.""" conclusion = { "answer": "", "confidence": 0.0, "evidence": [], "alternatives": [], "meta_insights": [], "future_considerations": [] } sections = response.split('[') for section in sections: if section.startswith('Conclusion]'): lines = section.split('\n')[1:] for line in lines: if line.startswith('Answer:'): conclusion["answer"] = line[7:].strip() elif line.startswith('Confidence:'): try: conclusion["confidence"] = float(line[11:].strip()) except: conclusion["confidence"] = 0.5 elif line.startswith('Evidence:'): conclusion["evidence"] = [e.strip() for e in line[9:].split(',')] elif line.startswith('Alternatives:'): conclusion["alternatives"] = [a.strip() for a in line[13:].split(',')] elif section.startswith('Meta]'): lines = section.split('\n')[1:] for line in lines: if line.startswith('Insights:'): conclusion["meta_insights"] = [i.strip() for i in line[9:].split(',')] elif line.startswith('Future:'): conclusion["future_considerations"] = [f.strip() for f in line[7:].split(',')] return conclusion def _thought_to_dict(self, thought: Thought) -> Dict[str, Any]: """Convert thought to dictionary for serialization.""" return { "type": thought.type.value, "content": thought.content, "confidence": thought.confidence, "evidence": thought.evidence, "alternatives": thought.alternatives, "next_steps": thought.next_steps, "metadata": thought.metadata } def get_thought_history(self) -> List[Dict[str, Any]]: """Get the history of all thoughts processed.""" return [self._thought_to_dict(t) for t in self.thought_history] def clear_history(self) -> None: """Clear thought history.""" self.thought_history = []