agentic-system / reasoning /bayesian.py
Cascade Bot
Renamed BayesianReasoning to BayesianStrategy and fixed imports
bb6f0d3
"""Advanced Bayesian reasoning for probabilistic analysis."""
import logging
from typing import Dict, Any, List, Optional, Set, Union, Type, Tuple
import json
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import numpy as np
from collections import defaultdict
from .base import ReasoningStrategy, StrategyResult
@dataclass
class BayesianHypothesis:
"""Bayesian hypothesis with probabilities."""
name: str
prior: float
likelihood: float
posterior: float = 0.0
evidence: List[Dict[str, Any]] = field(default_factory=list)
class BayesianStrategy(ReasoningStrategy):
"""Advanced Bayesian reasoning that:
1. Generates hypotheses
2. Calculates prior probabilities
3. Updates with evidence
4. Computes posteriors
5. Provides probabilistic analysis
"""
def __init__(self, config: Optional[Dict[str, Any]] = None):
"""Initialize Bayesian reasoning."""
super().__init__()
self.config = config or {}
# Configure Bayesian parameters
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]
) -> StrategyResult:
"""
Apply Bayesian reasoning to analyze probabilities and update beliefs.
Args:
query: The input query to reason about
context: Additional context and parameters
Returns:
StrategyResult containing reasoning results and confidence scores
"""
try:
# Generate initial hypotheses
self.hypotheses = await self._generate_hypotheses(query, context)
# Calculate prior probabilities
priors = await self._calculate_priors(self.hypotheses, context)
# Update with evidence
posteriors = await self._update_with_evidence(self.hypotheses, priors, context)
# Generate analysis
analysis = await self._generate_analysis(posteriors, context)
# 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 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[BayesianHypothesis]:
"""Generate plausible hypotheses."""
# Extract key terms for hypothesis generation
terms = self._extract_factors(query, set())
# 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[BayesianHypothesis],
context: Dict[str, Any]
) -> Dict[str, float]:
"""Calculate prior probabilities."""
priors = {}
total_prior = sum(h.prior for h in hypotheses)
if total_prior > 0:
# 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[BayesianHypothesis],
priors: Dict[str, float],
context: Dict[str, Any]
) -> Dict[str, float]:
"""Update probabilities with evidence."""
posteriors = priors.copy()
# Get evidence from context
evidence = context.get('evidence', [])
for e in evidence:
# Calculate likelihoods
likelihoods = {}
total_likelihood = 0.0
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]
# 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
async def _calculate_likelihood(
self,
hypothesis: BayesianHypothesis,
evidence: Dict[str, Any]
) -> float:
"""Calculate likelihood of evidence given hypothesis."""
# Simple likelihood calculation
# Could be enhanced with more sophisticated methods
base_likelihood = 0.5
# Adjust based on evidence strength
strength = evidence.get('strength', 0.0)
likelihood = base_likelihood * (1 + strength)
return min(1.0, max(self.min_likelihood, likelihood))
async def _generate_analysis(
self,
posteriors: Dict[str, float],
context: Dict[str, Any]
) -> Dict[str, Any]:
"""Generate probabilistic analysis."""
analysis = {
'top_hypothesis': max(posteriors.items(), key=lambda x: x[1]),
'confidence': self._calculate_confidence(posteriors),
'distribution': posteriors,
'summary': []
}
# 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 analysis
def _format_analysis(self, analysis: Dict[str, Any]) -> str:
"""Format analysis into readable text."""
top_hyp, top_prob = analysis['top_hypothesis']
text = [
f"Based on Bayesian analysis:",
f"- Most likely hypothesis: {top_hyp} (probability: {top_prob:.2f})",
"\nProbability distribution:"
]
for item in analysis['summary']:
text.append(
f"- {item['hypothesis']}: {item['probability']:.2f} "
f"({item['strength']} evidence)"
)
return "\n".join(text)
def _calculate_confidence(self, posteriors: Dict[str, float]) -> float:
"""Calculate overall confidence score."""
if not posteriors:
return 0.0
# Get top two probabilities
probs = sorted(posteriors.values(), reverse=True)
top_prob = probs[0]
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(1.0, max(0.0, confidence))
def _extract_factors(self, text: str, terms: Set[str]) -> Set[str]:
"""Extract relevant factors from text."""
# 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]) -> Dict[str, float]:
"""Generate factors for alternative hypothesis."""
# Simple alternative generation
# Could be enhanced with domain knowledge
alternatives = {
'primary': 0.6,
'alternative': 0.3,
'null': 0.1
}
return alternatives