Updated ChainOfThoughtStrategy to use StrategyResult and improved implementation

reasoning/chain_of_thought.py

@@ -5,8 +5,9 @@ from typing import Dict, Any, List, Optional, Tuple
 import json
 from dataclasses import dataclass
 from enum import Enum
+from datetime import datetime
 
-from .base import ReasoningStrategy
+from .base import ReasoningStrategy, StrategyResult
 
 class ThoughtType(Enum):
     """Types of thoughts in the chain."""
@@ -28,6 +29,7 @@ class Thought:
     alternatives: List[str]
     next_steps: List[str]
     metadata: Dict[str, Any]
+    timestamp: str = datetime.now().isoformat()
 
 class ChainOfThoughtStrategy(ReasoningStrategy):
     """
@@ -45,371 +47,394 @@ class ChainOfThoughtStrategy(ReasoningStrategy):
                  parallel_threshold: int = 3,
                  learning_rate: float = 0.1,
                  strategy_weights: Optional[Dict[str, float]] = None):
+        """Initialize Chain of Thought reasoning."""
+        super().__init__()
         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
+            'observation': 0.2,
+            'analysis': 0.3,
+            'hypothesis': 0.2,
+            'verification': 0.15,
+            'conclusion': 0.15
         }
-        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."""
+        # Initialize thought chain
+        self.thoughts: List[Thought] = []
+        
+        # Performance tracking
+        self.performance_metrics = {
+            'avg_confidence': 0.0,
+            'chain_length': 0,
+            'refinement_count': 0,
+            'parallel_paths': 0
+        }
+    
+    async def reason(
+        self,
+        query: str,
+        context: Dict[str, Any]
+    ) -> StrategyResult:
+        """
+        Apply Chain of Thought reasoning to analyze the query.
+        
+        Args:
+            query: The input query to reason about
+            context: Additional context and parameters
+            
+        Returns:
+            StrategyResult containing the reasoning chain and confidence
+        """
         try:
-            # Initialize reasoning chain
-            chain = await self._initialize_chain(query, context)
+            # Reset thought chain
+            self.thoughts = []
+            
+            # Initial observation
+            await self._add_thought(
+                ThoughtType.OBSERVATION,
+                f"Analyzing query: {query}",
+                context
+            )
             
-            # Generate initial thoughts
-            thoughts = await self._generate_thoughts(query, context)
+            # Generate analysis thoughts
+            await self._analyze_query(query, context)
             
-            # Build thought chain
-            chain = await self._build_chain(thoughts, context)
+            # Generate hypotheses
+            hypotheses = await self._generate_hypotheses(context)
+            
+            # Verify hypotheses
+            await self._verify_hypotheses(hypotheses, context)
+            
+            # Draw conclusions
+            conclusion = await self._draw_conclusion(context)
             
             # Reflect and refine
-            if self.enable_reflection:
-                chain = await self._reflect_and_refine(chain, context)
+            if conclusion.confidence < self.min_confidence:
+                await self._reflect_and_refine(context)
+                conclusion = await self._draw_conclusion(context)
             
-            # Extract conclusion
-            conclusion = await self._extract_conclusion(chain, context)
+            # Update performance metrics
+            self._update_metrics()
             
-            # Update thought history
-            self.thought_history.extend(chain)
+            return StrategyResult(
+                strategy_type="chain_of_thought",
+                success=True,
+                answer=conclusion.content,
+                confidence=conclusion.confidence,
+                reasoning_trace=[{
+                    "step": str(t.type.value),
+                    "content": t.content,
+                    "confidence": t.confidence,
+                    "evidence": t.evidence,
+                    "alternatives": t.alternatives,
+                    "next_steps": t.next_steps,
+                    "metadata": t.metadata,
+                    "timestamp": t.timestamp
+                } for t in self.thoughts],
+                metadata={
+                    "num_thoughts": len(self.thoughts),
+                    "thought_types": [t.type.value for t in self.thoughts],
+                    "final_confidence": conclusion.confidence
+                },
+                performance_metrics=self.performance_metrics
+            )
             
-            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
+            logging.error(f"Chain of Thought reasoning error: {str(e)}")
+            return StrategyResult(
+                strategy_type="chain_of_thought",
+                success=False,
+                answer=None,
+                confidence=0.0,
+                reasoning_trace=[{
+                    "step": "error",
+                    "error": str(e),
+                    "timestamp": datetime.now().isoformat()
+                }],
+                metadata={"error": str(e)},
+                performance_metrics=self.performance_metrics
+            )
+    
+    async def _add_thought(
+        self,
+        type: ThoughtType,
+        content: str,
+        context: Dict[str, Any]
+    ) -> Thought:
+        """Add a new thought to the chain."""
+        thought = Thought(
+            type=type,
+            content=content,
+            confidence=self._calculate_confidence(content, context),
+            evidence=self._gather_evidence(content, context),
+            alternatives=self._generate_alternatives(content, context),
+            next_steps=self._determine_next_steps(type, context),
+            metadata=self._extract_metadata(content, context)
+        )
+        self.thoughts.append(thought)
+        return thought
+    
+    async def _analyze_query(
+        self,
+        query: str,
+        context: Dict[str, Any]
+    ) -> None:
+        """Generate analysis thoughts."""
+        # Extract key components
+        components = self._extract_components(query)
         
-        Format as:
-        [Analysis]
-        Gaps: ...
-        Assumptions: ...
-        Missing: ...
-        Alternatives: ...
+        # Analyze each component
+        for comp in components:
+            await self._add_thought(
+                ThoughtType.ANALYSIS,
+                f"Analysis of {comp}: {self._analyze_component(comp, context)}",
+                context
+            )
+    
+    async def _generate_hypotheses(
+        self,
+        context: Dict[str, Any]
+    ) -> List[Thought]:
+        """Generate hypothesis thoughts."""
+        hypotheses = []
         
-        [Refinements]
-        Thoughts: ...
-        Reasoning: ...
-        Connections: ...
-        Evidence: ...
-        """
+        # Generate hypotheses based on analysis
+        analysis_thoughts = [t for t in self.thoughts if t.type == ThoughtType.ANALYSIS]
         
-        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)}
+        for analysis in analysis_thoughts:
+            hypothesis = await self._add_thought(
+                ThoughtType.HYPOTHESIS,
+                f"Based on {analysis.content}, hypothesis: {self._generate_hypothesis(analysis, context)}",
+                context
+            )
+            hypotheses.append(hypothesis)
         
-        Provide:
-        1. Main conclusion
-        2. Confidence level
-        3. Supporting evidence
-        4. Alternative conclusions
-        5. Meta-insights gained
-        6. Future considerations
+        return hypotheses
+    
+    async def _verify_hypotheses(
+        self,
+        hypotheses: List[Thought],
+        context: Dict[str, Any]
+    ) -> None:
+        """Verify generated hypotheses."""
+        for hypothesis in hypotheses:
+            await self._add_thought(
+                ThoughtType.VERIFICATION,
+                f"Verifying {hypothesis.content}: {self._verify_hypothesis(hypothesis, context)}",
+                context
+            )
+    
+    async def _draw_conclusion(
+        self,
+        context: Dict[str, Any]
+    ) -> Thought:
+        """Draw conclusion from verified hypotheses."""
+        verified_thoughts = [t for t in self.thoughts if t.type == ThoughtType.VERIFICATION]
         
-        Format as:
-        [Conclusion]
-        Answer: ...
-        Confidence: ...
-        Evidence: ...
-        Alternatives: ...
+        conclusion_content = self._synthesize_conclusion(verified_thoughts, context)
         
-        [Meta]
-        Insights: ...
-        Future: ...
-        """
+        return await self._add_thought(
+            ThoughtType.CONCLUSION,
+            conclusion_content,
+            context
+        )
+    
+    async def _reflect_and_refine(
+        self,
+        context: Dict[str, Any]
+    ) -> None:
+        """Reflect on the reasoning chain and refine if needed."""
+        # Add reflection thought
+        reflection = await self._add_thought(
+            ThoughtType.REFLECTION,
+            self._generate_reflection(self.thoughts, context),
+            context
+        )
         
-        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')
+        # Add refinement if needed
+        if reflection.confidence < self.min_confidence:
+            await self._add_thought(
+                ThoughtType.REFINEMENT,
+                self._generate_refinement(reflection, context),
+                context
+            )
+            
+            self.performance_metrics['refinement_count'] += 1
+    
+    def _calculate_confidence(
+        self,
+        content: str,
+        context: Dict[str, Any]
+    ) -> float:
+        """Calculate confidence score for a thought."""
+        # Base confidence
+        confidence = 0.5
         
-        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}
-                    ))
+        # Adjust based on content length and complexity
+        words = content.split()
+        if len(words) > 50:
+            confidence += 0.1
+        if len(words) > 100:
+            confidence += 0.1
         
-        return observations
-
-    def _parse_thoughts(self, response: str) -> List[Thought]:
-        """Parse generated thoughts."""
-        thoughts = []
-        current = None
+        # Adjust based on evidence
+        evidence = self._gather_evidence(content, context)
+        confidence += min(0.3, len(evidence) * 0.1)
         
-        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
+        return min(1.0, confidence)
+    
+    def _gather_evidence(
+        self,
+        content: str,
+        context: Dict[str, Any]
+    ) -> List[str]:
+        """Gather evidence supporting the thought."""
+        evidence = []
         
-        if current:
-            thoughts.append(current)
+        # Extract from context
+        if 'evidence' in context:
+            evidence.extend(context['evidence'])
         
-        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}
+        # Extract from previous thoughts
+        for thought in self.thoughts:
+            if any(term in thought.content.lower() for term in content.lower().split()):
+                evidence.append(f"Supported by previous thought: {thought.content}")
         
-        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 evidence
+    
+    def _generate_alternatives(
+        self,
+        content: str,
+        context: Dict[str, Any]
+    ) -> List[str]:
+        """Generate alternative perspectives."""
+        alternatives = []
         
-        return chain
-
-    def _apply_refinements(self, chain: List[Thought], response: str) -> List[Thought]:
-        """Apply refinements to thought chain."""
-        refined_chain = chain.copy()
+        # Generate opposites
+        words = content.lower().split()
+        opposites = {
+            'increase': 'decrease',
+            'high': 'low',
+            'good': 'bad',
+            'positive': 'negative'
+        }
         
-        # 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)
+        for word in words:
+            if word in opposites:
+                alt = content.replace(word, opposites[word])
+                alternatives.append(f"Alternative: {alt}")
         
-        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": []
-        }
+        return alternatives
+    
+    def _determine_next_steps(
+        self,
+        type: ThoughtType,
+        context: Dict[str, Any]
+    ) -> List[str]:
+        """Determine possible next steps."""
+        steps = []
         
-        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(',')]
+        if type == ThoughtType.OBSERVATION:
+            steps.extend([
+                "Analyze key components",
+                "Identify patterns",
+                "Consider context"
+            ])
+        elif type == ThoughtType.ANALYSIS:
+            steps.extend([
+                "Generate hypotheses",
+                "Look for correlations",
+                "Consider alternatives"
+            ])
+        elif type == ThoughtType.HYPOTHESIS:
+            steps.extend([
+                "Verify hypothesis",
+                "Gather evidence",
+                "Test assumptions"
+            ])
+        elif type == ThoughtType.VERIFICATION:
+            steps.extend([
+                "Draw conclusions",
+                "Consider implications",
+                "Plan actions"
+            ])
         
-        return conclusion
-
-    def _thought_to_dict(self, thought: Thought) -> Dict[str, Any]:
-        """Convert thought to dictionary for serialization."""
+        return steps
+    
+    def _extract_metadata(
+        self,
+        content: str,
+        context: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Extract metadata from thought content."""
         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
+            'length': len(content),
+            'complexity': len(content.split()),
+            'context_keys': list(context.keys()),
+            'timestamp': datetime.now().isoformat()
         }
-
-    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 = []
+    
+    def _extract_components(self, query: str) -> List[str]:
+        """Extract key components from query."""
+        # Simple word-based extraction
+        # Could be enhanced with NLP
+        return [w.strip() for w in query.split() if len(w.strip()) > 3]
+    
+    def _analyze_component(
+        self,
+        component: str,
+        context: Dict[str, Any]
+    ) -> str:
+        """Analyze a single component."""
+        return f"Component {component} analysis based on context"
+    
+    def _generate_hypothesis(
+        self,
+        analysis: Thought,
+        context: Dict[str, Any]
+    ) -> str:
+        """Generate hypothesis from analysis."""
+        return f"Hypothesis generated from {analysis.content}"
+    
+    def _verify_hypothesis(
+        self,
+        hypothesis: Thought,
+        context: Dict[str, Any]
+    ) -> str:
+        """Verify a hypothesis."""
+        return f"Verification of {hypothesis.content}"
+    
+    def _synthesize_conclusion(
+        self,
+        verified_thoughts: List[Thought],
+        context: Dict[str, Any]
+    ) -> str:
+        """Synthesize conclusion from verified thoughts."""
+        return "Conclusion based on verified thoughts: " + \
+               ", ".join(t.content for t in verified_thoughts)
+    
+    def _generate_reflection(
+        self,
+        thoughts: List[Thought],
+        context: Dict[str, Any]
+    ) -> str:
+        """Generate reflection on thought chain."""
+        return f"Reflection on {len(thoughts)} thoughts in chain"
+    
+    def _generate_refinement(
+        self,
+        reflection: Thought,
+        context: Dict[str, Any]
+    ) -> str:
+        """Generate refinement based on reflection."""
+        return f"Refinement based on {reflection.content}"
+    
+    def _update_metrics(self) -> None:
+        """Update performance metrics."""
+        if self.thoughts:
+            self.performance_metrics.update({
+                'avg_confidence': sum(t.confidence for t in self.thoughts) / len(self.thoughts),
+                'chain_length': len(self.thoughts),
+                'parallel_paths': len([t for t in self.thoughts if t.alternatives])
+            })