Renamed BayesianReasoning to BayesianStrategy and fixed imports

reasoning/__init__.py

@@ -39,7 +39,7 @@ from .local_llm import LocalLLMStrategy
 
 # Advanced reasoning strategies
 from .multimodal import MultiModalReasoning
-from .bayesian import BayesianReasoning
+from .bayesian import BayesianStrategy
 from .quantum import QuantumReasoning
 from .neurosymbolic import NeurosymbolicReasoning
 from .emergent import EmergentReasoning
@@ -72,7 +72,7 @@ __all__ = [
     
     # Advanced reasoning
     'MultiModalReasoning',
-    'BayesianReasoning',
+    'BayesianStrategy',
     'QuantumReasoning',
     'NeurosymbolicReasoning',
     'EmergentReasoning',

reasoning/bayesian.py

@@ -9,7 +9,7 @@ from datetime import datetime
 import numpy as np
 from collections import defaultdict
 
-from .base import ReasoningStrategy
+from .base import ReasoningStrategy, StrategyResult
 
 @dataclass
 class BayesianHypothesis:
@@ -20,9 +20,8 @@ class BayesianHypothesis:
     posterior: float = 0.0
     evidence: List[Dict[str, Any]] = field(default_factory=list)
 
-class BayesianReasoning(ReasoningStrategy):
-    """
-    Advanced Bayesian reasoning that:
+class BayesianStrategy(ReasoningStrategy):
+    """Advanced Bayesian reasoning that:
     1. Generates hypotheses
     2. Calculates prior probabilities
     3. Updates with evidence
@@ -39,8 +38,15 @@ class BayesianReasoning(ReasoningStrategy):
         self.prior_weight = self.config.get('prior_weight', 0.3)
         self.evidence_threshold = self.config.get('evidence_threshold', 0.1)
         self.min_likelihood = self.config.get('min_likelihood', 0.01)
+        
+        # Initialize hypothesis storage
+        self.hypotheses: List[BayesianHypothesis] = []
     
-    async def reason(self, query: str, context: Dict[str, Any]) -> Dict[str, Any]:
+    async def reason(
+        self,
+        query: str,
+        context: Dict[str, Any]
+    ) -> StrategyResult:
         """
         Apply Bayesian reasoning to analyze probabilities and update beliefs.
         
@@ -49,119 +55,113 @@ class BayesianReasoning(ReasoningStrategy):
             context: Additional context and parameters
             
         Returns:
-            Dict containing reasoning results and confidence scores
+            StrategyResult containing reasoning results and confidence scores
         """
         try:
-            # Generate hypotheses
-            hypotheses = await self._generate_hypotheses(query, context)
+            # Generate initial hypotheses
+            self.hypotheses = await self._generate_hypotheses(query, context)
             
-            # Calculate priors
-            priors = await self._calculate_priors(hypotheses, context)
+            # Calculate prior probabilities
+            priors = await self._calculate_priors(self.hypotheses, context)
             
             # Update with evidence
-            posteriors = await self._update_with_evidence(
-                hypotheses,
-                priors,
-                context
-            )
+            posteriors = await self._update_with_evidence(self.hypotheses, priors, context)
             
             # Generate analysis
             analysis = await self._generate_analysis(posteriors, context)
             
-            return {
-                'answer': self._format_analysis(analysis),
-                'confidence': self._calculate_confidence(posteriors),
-                'hypotheses': hypotheses,
-                'priors': priors,
-                'posteriors': posteriors,
-                'analysis': analysis
-            }
+            # Format results
+            answer = self._format_analysis(analysis)
+            confidence = self._calculate_confidence(posteriors)
+            
+            return StrategyResult(
+                strategy_type="bayesian",
+                success=True,
+                answer=answer,
+                confidence=confidence,
+                reasoning_trace=[{
+                    "step": "bayesian_analysis",
+                    "hypotheses": [h.__dict__ for h in self.hypotheses],
+                    "priors": priors,
+                    "posteriors": posteriors,
+                    "analysis": analysis,
+                    "timestamp": datetime.now().isoformat()
+                }],
+                metadata={
+                    "num_hypotheses": len(self.hypotheses),
+                    "max_posterior": max(posteriors.values()) if posteriors else 0.0,
+                    "config": self.config
+                },
+                performance_metrics={
+                    "prior_weight": self.prior_weight,
+                    "evidence_threshold": self.evidence_threshold,
+                    "min_likelihood": self.min_likelihood
+                }
+            )
             
         except Exception as e:
-            logging.error(f"Bayesian reasoning failed: {str(e)}")
-            return {
-                'error': f"Bayesian reasoning failed: {str(e)}",
-                'confidence': 0.0
-            }
+            logging.error(f"Bayesian reasoning error: {str(e)}")
+            return StrategyResult(
+                strategy_type="bayesian",
+                success=False,
+                answer=None,
+                confidence=0.0,
+                reasoning_trace=[{
+                    "step": "error",
+                    "error": str(e),
+                    "timestamp": datetime.now().isoformat()
+                }],
+                metadata={"error": str(e)},
+                performance_metrics={}
+            )
     
     async def _generate_hypotheses(
         self,
         query: str,
         context: Dict[str, Any]
-    ) -> List[Dict[str, Any]]:
+    ) -> List[BayesianHypothesis]:
         """Generate plausible hypotheses."""
-        hypotheses = []
-        
         # Extract key terms for hypothesis generation
-        terms = set(query.lower().split())
+        terms = self._extract_factors(query, set())
         
-        # Generate hypotheses based on context and terms
-        if 'options' in context:
-            # Use provided options as hypotheses
-            for option in context['options']:
-                hypotheses.append({
-                    'name': option,
-                    'description': f"Hypothesis based on option: {option}",
-                    'factors': self._extract_factors(option, terms)
-                })
-        else:
-            # Generate default hypotheses
-            hypotheses.extend([
-                {
-                    'name': 'primary',
-                    'description': "Primary hypothesis based on direct interpretation",
-                    'factors': self._extract_factors(query, terms)
-                },
-                {
-                    'name': 'alternative',
-                    'description': "Alternative hypothesis considering other factors",
-                    'factors': self._generate_alternative_factors(terms)
-                }
-            ])
+        # Generate alternative hypotheses
+        alternatives = self._generate_alternative_factors(terms)
+        
+        # Create hypothesis objects
+        hypotheses = []
+        for name, prior in alternatives.items():
+            hypotheses.append(BayesianHypothesis(
+                name=name,
+                prior=prior,
+                likelihood=1.0  # Initial likelihood
+            ))
         
         return hypotheses
     
     async def _calculate_priors(
         self,
-        hypotheses: List[Dict[str, Any]],
+        hypotheses: List[BayesianHypothesis],
         context: Dict[str, Any]
     ) -> Dict[str, float]:
         """Calculate prior probabilities."""
         priors = {}
+        total_prior = sum(h.prior for h in hypotheses)
         
-        # Get historical data if available
-        history = context.get('history', {})
-        total_cases = sum(history.values()) if history else len(hypotheses)
-        
-        for hypothesis in hypotheses:
-            name = hypothesis['name']
-            
-            # Calculate prior from history or use uniform prior
-            if name in history:
-                priors[name] = history[name] / total_cases
-            else:
-                priors[name] = 1.0 / len(hypotheses)
-            
-            # Adjust prior based on factors
-            factor_weight = len(hypothesis['factors']) / 10  # Normalize factor count
-            priors[name] = (
-                priors[name] * (1 - self.prior_weight) +
-                factor_weight * self.prior_weight
-            )
-        
-        # Normalize priors
-        total_prior = sum(priors.values())
         if total_prior > 0:
-            priors = {
-                name: prob / total_prior
-                for name, prob in priors.items()
-            }
+            # Normalize priors
+            for h in hypotheses:
+                priors[h.name] = h.prior / total_prior
+        else:
+            # Equal priors if no information
+            prior = 1.0 / len(hypotheses)
+            for h in hypotheses:
+                priors[h.name] = prior
         
         return priors
     
     async def _update_with_evidence(
         self,
-        hypotheses: List[Dict[str, Any]],
+        hypotheses: List[BayesianHypothesis],
         priors: Dict[str, float],
         context: Dict[str, Any]
     ) -> Dict[str, float]:
@@ -170,51 +170,41 @@ class BayesianReasoning(ReasoningStrategy):
         
         # Get evidence from context
         evidence = context.get('evidence', [])
-        if not evidence:
-            return posteriors
         
         for e in evidence:
-            # Calculate likelihood for each hypothesis
+            # Calculate likelihoods
             likelihoods = {}
-            for hypothesis in hypotheses:
-                name = hypothesis['name']
-                likelihood = self._calculate_likelihood(hypothesis, e)
-                likelihoods[name] = max(likelihood, self.min_likelihood)
+            total_likelihood = 0.0
             
-            # Update posteriors using Bayes' rule
-            total_probability = sum(
-                likelihoods[name] * posteriors[name]
-                for name in posteriors
-            )
+            for h in hypotheses:
+                likelihood = await self._calculate_likelihood(h, e)
+                likelihoods[h.name] = max(likelihood, self.min_likelihood)
+                total_likelihood += likelihood * posteriors[h.name]
             
-            if total_probability > 0:
-                posteriors = {
-                    name: (likelihoods[name] * posteriors[name]) / total_probability
-                    for name in posteriors
-                }
+            # Update posteriors using Bayes' rule
+            if total_likelihood > 0:
+                for h in hypotheses:
+                    posteriors[h.name] = (
+                        likelihoods[h.name] * posteriors[h.name] / total_likelihood
+                    )
         
         return posteriors
     
-    def _calculate_likelihood(
+    async def _calculate_likelihood(
         self,
-        hypothesis: Dict[str, Any],
+        hypothesis: BayesianHypothesis,
         evidence: Dict[str, Any]
     ) -> float:
         """Calculate likelihood of evidence given hypothesis."""
-        # Extract evidence factors
-        evidence_factors = set(
-            str(v).lower()
-            for v in evidence.values()
-            if isinstance(v, (str, int, float))
-        )
-        
-        # Compare with hypothesis factors
-        common_factors = evidence_factors.intersection(hypothesis['factors'])
+        # Simple likelihood calculation
+        # Could be enhanced with more sophisticated methods
+        base_likelihood = 0.5
         
-        if not evidence_factors:
-            return 0.5  # Neutral likelihood if no factors
+        # Adjust based on evidence strength
+        strength = evidence.get('strength', 0.0)
+        likelihood = base_likelihood * (1 + strength)
         
-        return len(common_factors) / len(evidence_factors)
+        return min(1.0, max(self.min_likelihood, likelihood))
     
     async def _generate_analysis(
         self,
@@ -222,104 +212,77 @@ class BayesianReasoning(ReasoningStrategy):
         context: Dict[str, Any]
     ) -> Dict[str, Any]:
         """Generate probabilistic analysis."""
-        # Sort hypotheses by posterior probability
-        ranked_hypotheses = sorted(
-            posteriors.items(),
-            key=lambda x: x[1],
-            reverse=True
-        )
+        analysis = {
+            'top_hypothesis': max(posteriors.items(), key=lambda x: x[1]),
+            'confidence': self._calculate_confidence(posteriors),
+            'distribution': posteriors,
+            'summary': []
+        }
         
-        # Calculate statistics
-        mean = np.mean(list(posteriors.values()))
-        std = np.std(list(posteriors.values()))
-        entropy = -sum(
-            p * np.log2(p) if p > 0 else 0
-            for p in posteriors.values()
-        )
+        # Generate summary points
+        for name, prob in sorted(posteriors.items(), key=lambda x: x[1], reverse=True):
+            analysis['summary'].append({
+                'hypothesis': name,
+                'probability': prob,
+                'strength': 'strong' if prob > 0.7 else 'moderate' if prob > 0.3 else 'weak'
+            })
         
-        return {
-            'top_hypothesis': ranked_hypotheses[0][0],
-            'probability': ranked_hypotheses[0][1],
-            'alternatives': [
-                {'name': name, 'probability': prob}
-                for name, prob in ranked_hypotheses[1:]
-            ],
-            'statistics': {
-                'mean': mean,
-                'std': std,
-                'entropy': entropy
-            }
-        }
+        return analysis
     
     def _format_analysis(self, analysis: Dict[str, Any]) -> str:
         """Format analysis into readable text."""
-        sections = []
+        top_hyp, top_prob = analysis['top_hypothesis']
         
-        # Top hypothesis
-        sections.append(
-            f"Most likely hypothesis: {analysis['top_hypothesis']} "
-            f"(probability: {analysis['probability']:.2%})"
-        )
+        text = [
+            f"Based on Bayesian analysis:",
+            f"- Most likely hypothesis: {top_hyp} (probability: {top_prob:.2f})",
+            "\nProbability distribution:"
+        ]
         
-        # Alternative hypotheses
-        if analysis['alternatives']:
-            sections.append("\nAlternative hypotheses:")
-            for alt in analysis['alternatives']:
-                sections.append(
-                    f"- {alt['name']}: {alt['probability']:.2%}"
-                )
-        
-        # Statistics
-        stats = analysis['statistics']
-        sections.append("\nDistribution statistics:")
-        sections.append(f"- Mean probability: {stats['mean']:.2%}")
-        sections.append(f"- Standard deviation: {stats['std']:.2%}")
-        sections.append(f"- Entropy: {stats['entropy']:.2f} bits")
+        for item in analysis['summary']:
+            text.append(
+                f"- {item['hypothesis']}: {item['probability']:.2f} "
+                f"({item['strength']} evidence)"
+            )
         
-        return "\n".join(sections)
+        return "\n".join(text)
     
     def _calculate_confidence(self, posteriors: Dict[str, float]) -> float:
         """Calculate overall confidence score."""
         if not posteriors:
             return 0.0
         
-        # Base confidence
-        confidence = 0.5
-        
-        # Adjust based on probability distribution
-        probs = list(posteriors.values())
-        
-        # Strong leading hypothesis increases confidence
-        max_prob = max(probs)
-        if max_prob > 0.8:
-            confidence += 0.3
-        elif max_prob > 0.6:
-            confidence += 0.2
-        elif max_prob > 0.4:
-            confidence += 0.1
-        
-        # Low entropy (clear distinction) increases confidence
-        entropy = -sum(p * np.log2(p) if p > 0 else 0 for p in probs)
-        max_entropy = -np.log2(1/len(probs))  # Maximum possible entropy
+        # Get top two probabilities
+        probs = sorted(posteriors.values(), reverse=True)
+        top_prob = probs[0]
         
-        if entropy < 0.3 * max_entropy:
-            confidence += 0.2
-        elif entropy < 0.6 * max_entropy:
-            confidence += 0.1
+        if len(probs) > 1:
+            # Consider the gap between top hypotheses
+            second_prob = probs[1]
+            margin = top_prob - second_prob
+            
+            # Confidence increases with both probability and margin
+            confidence = (top_prob + margin) / 2
+        else:
+            confidence = top_prob
         
-        return min(confidence, 1.0)
+        return min(1.0, max(0.0, confidence))
     
     def _extract_factors(self, text: str, terms: Set[str]) -> Set[str]:
         """Extract relevant factors from text."""
-        return set(word.lower() for word in text.split() if word.lower() in terms)
+        # Simple word-based extraction
+        # Could be enhanced with NLP techniques
+        words = text.lower().split()
+        return set(words).union(terms)
     
-    def _generate_alternative_factors(self, terms: Set[str]) -> Set[str]:
+    def _generate_alternative_factors(self, terms: Set[str]) -> Dict[str, float]:
         """Generate factors for alternative hypothesis."""
-        # Simple approach: use terms not in primary hypothesis
-        return set(
-            word for word in terms
-            if not any(
-                similar in word or word in similar
-                for similar in terms
-            )
-        )
+        # Simple alternative generation
+        # Could be enhanced with domain knowledge
+        alternatives = {
+            'primary': 0.6,
+            'alternative': 0.3,
+            'null': 0.1
+        }
+        
+        return alternatives