aishastr.live

Agentic Architecture Pattern: Probabilistic Reasoning

Written by

admin

in


    """
Pattern 50: Probabilistic Reasoning Pattern

Reasons under uncertainty using probability theory.
Features Bayesian inference, belief updates, and probabilistic queries.
"""

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


@dataclass
class ProbabilityDistribution:
    """Discrete probability distribution."""
    variable: str
    probabilities: Dict[Any, float]
    
    def normalize(self):
        """Normalize probabilities to sum to 1."""
        total = sum(self.probabilities.values())
        if total > 0:
            for key in self.probabilities:
                self.probabilities[key] /= total
    
    def get_prob(self, value: Any) -> float:
        """Get probability of a value."""
        return self.probabilities.get(value, 0.0)
    
    def most_likely(self) -> Tuple[Any, float]:
        """Get most likely value."""
        if not self.probabilities:
            return None, 0.0
        max_val = max(self.probabilities.items(), key=lambda x: x[1])
        return max_val


@dataclass
class Evidence:
    """Observed evidence."""
    variable: str
    value: Any
    confidence: float = 1.0


class BayesianNetwork:
    """Simple Bayesian network for probabilistic reasoning."""
    
    def __init__(self):
        self.variables: Dict[str, ProbabilityDistribution] = {}
        self.conditionals: Dict[str, Dict[Tuple, float]] = defaultdict(dict)
    
    def add_prior(self, distribution: ProbabilityDistribution):
        """Add prior probability distribution."""
        distribution.normalize()
        self.variables[distribution.variable] = distribution
    
    def add_conditional(self, variable: str, given_var: str, 
                       given_value: Any, var_value: Any, probability: float):
        """Add conditional probability P(variable=var_value | given_var=given_value)."""
        self.conditionals[variable][(given_var, given_value, var_value)] = probability
    
    def query(self, variable: str, evidence: List[Evidence]) -> ProbabilityDistribution:
        """Query probability of variable given evidence (Bayesian inference)."""
        if variable not in self.variables:
            return ProbabilityDistribution(variable, {})
        
        # Start with prior
        posterior = self.variables[variable].probabilities.copy()
        
        # Update with each piece of evidence using Bayes' rule
        for ev in evidence:
            if ev.variable == variable:
                # Direct evidence
                for value in list(posterior.keys()):
                    if value != ev.value:
                        posterior[value] *= (1 - ev.confidence)
                    else:
                        posterior[value] *= ev.confidence
            else:
                # Conditional evidence
                for value in posterior.keys():
                    # P(variable=value | evidence)
                    conditional_key = (ev.variable, ev.value, value)
                    if conditional_key in self.conditionals[variable]:
                        likelihood = self.conditionals[variable][conditional_key]
                        posterior[value] *= likelihood * ev.confidence
        
        result = ProbabilityDistribution(variable, posterior)
        result.normalize()
        return result


class ProbabilisticAgent:
    """Agent that reasons probabilistically."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.belief_network = BayesianNetwork()
        self.queries_made = 0
    
    def update_beliefs(self, evidence: List[Evidence]):
        """Update beliefs based on evidence."""
        print(f"\n   {self.name}: Updating beliefs with evidence:")
        for ev in evidence:
            print(f"     - {ev.variable} = {ev.value} (confidence: {ev.confidence:.2f})")
    
    def query_belief(self, variable: str, evidence: List[Evidence]) -> ProbabilityDistribution:
        """Query belief about variable."""
        self.queries_made += 1
        print(f"\n   {self.name}: Querying belief about '{variable}'")
        
        posterior = self.belief_network.query(variable, evidence)
        
        print(f"   Posterior distribution:")
        for value, prob in sorted(posterior.probabilities.items(), 
                                 key=lambda x: x[1], reverse=True):
            print(f"     {value}: {prob:.3f}")
        
        return posterior
    
    def make_decision(self, variable: str, evidence: List[Evidence]) -> Tuple[Any, float]:
        """Make decision based on probabilistic reasoning."""
        posterior = self.query_belief(variable, evidence)
        decision, confidence = posterior.most_likely()
        
        print(f"\n   {self.name}: Decision: {decision} (confidence: {confidence:.3f})")
        return decision, confidence


class ProbabilisticReasoningPattern:
    """
    Probabilistic Reasoning Pattern - Reason under uncertainty.
    
    Agents use Bayesian inference to update beliefs and make decisions.
    Handles uncertain evidence and probabilistic queries.
    """
    
    def __init__(self):
        self.agents: Dict[str, ProbabilisticAgent] = {}
    
    def register_agent(self, agent: ProbabilisticAgent):
        """Register probabilistic agent."""
        self.agents[agent.name] = agent
    
    def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Execute probabilistic reasoning."""
        print("=== Probabilistic Reasoning Pattern Demo ===\n")
        
        decisions_made = 0
        
        for scenario in scenarios:
            print(f"\nScenario: {scenario.get('description', 'Unknown')}")
            
            query_var = scenario.get("query")
            evidence_list = [
                Evidence(ev["variable"], ev["value"], ev.get("confidence", 1.0))
                for ev in scenario.get("evidence", [])
            ]
            
            for agent in self.agents.values():
                agent.update_beliefs(evidence_list)
                agent.make_decision(query_var, evidence_list)
                decisions_made += 1
        
        return {
            "pattern": "ProbabilisticReasoning",
            "total_agents": len(self.agents),
            "scenarios": len(scenarios),
            "decisions_made": decisions_made
        }


# Example usage
if __name__ == "__main__":
    print("=== Probabilistic Reasoning Pattern Demo ===\n")
    
    # Create pattern
    pattern = ProbabilisticReasoningPattern()
    
    # Create agent
    print("1. Creating probabilistic agent...")
    agent = ProbabilisticAgent("DiagnosticAgent")
    
    # Define probability model
    print("\n2. Building Bayesian network...")
    
    # Prior: Disease prevalence
    agent.belief_network.add_prior(ProbabilityDistribution(
        variable="disease",
        probabilities={"flu": 0.1, "cold": 0.3, "covid": 0.05, "healthy": 0.55}
    ))
    
    # Conditionals: P(symptom | disease)
    # P(fever | disease)
    agent.belief_network.add_conditional("disease", "symptom", "fever", "flu", 0.8)
    agent.belief_network.add_conditional("disease", "symptom", "fever", "cold", 0.3)
    agent.belief_network.add_conditional("disease", "symptom", "fever", "covid", 0.9)
    agent.belief_network.add_conditional("disease", "symptom", "fever", "healthy", 0.1)
    
    # P(cough | disease)
    agent.belief_network.add_conditional("disease", "symptom", "cough", "flu", 0.6)
    agent.belief_network.add_conditional("disease", "symptom", "cough", "cold", 0.8)
    agent.belief_network.add_conditional("disease", "symptom", "cough", "covid", 0.7)
    agent.belief_network.add_conditional("disease", "symptom", "cough", "healthy", 0.1)
    
    # P(fatigue | disease)
    agent.belief_network.add_conditional("disease", "symptom", "fatigue", "flu", 0.9)
    agent.belief_network.add_conditional("disease", "symptom", "fatigue", "cold", 0.4)
    agent.belief_network.add_conditional("disease", "symptom", "fatigue", "covid", 0.8)
    agent.belief_network.add_conditional("disease", "symptom", "fatigue", "healthy", 0.2)
    
    pattern.register_agent(agent)
    
    # Test scenarios
    print("\n3. Running diagnostic scenarios...")
    
    scenarios = [
        {
            "description": "Patient with fever only",
            "query": "disease",
            "evidence": [
                {"variable": "symptom", "value": "fever", "confidence": 0.9}
            ]
        },
        {
            "description": "Patient with fever and cough",
            "query": "disease",
            "evidence": [
                {"variable": "symptom", "value": "fever", "confidence": 0.9},
                {"variable": "symptom", "value": "cough", "confidence": 0.8}
            ]
        },
        {
            "description": "Patient with all symptoms",
            "query": "disease",
            "evidence": [
                {"variable": "symptom", "value": "fever", "confidence": 0.9},
                {"variable": "symptom", "value": "cough", "confidence": 0.8},
                {"variable": "symptom", "value": "fatigue", "confidence": 0.85}
            ]
        }
    ]
    
    summary = pattern.run(scenarios)
    
    print(f"\n4. Summary:")
    print(f"   Scenarios: {summary['scenarios']}")
    print(f"   Decisions Made: {summary['decisions_made']}")
    print(f"   Queries: {agent.queries_made}")
    
    print("\nāœ“ Probabilistic reasoning handled uncertainty!")

Agentic Architecture Pattern: Probabilistic Reasoning

Pattern 50: 50 Probabilistic Reasoning – Line-by-Line Explanation

Overview

Pattern 50: Probabilistic Reasoning Pattern

Reasons under uncertainty using probability theory. Features Bayesian inference, belief updates, and probabilistic queries.

Code Structure Analysis

File Information

  • Pattern Number: 50
  • Pattern Name: 50 Probabilistic Reasoning
  • File Path: patterns\decision_making\50_probabilistic_reasoning.py
  • Total Lines: 256

Detailed Code Explanation

Module Docstring

"""
Pattern 50: Probabilistic Reasoning Pattern

Reasons under uncertainty using probability theory.
Features Bayesian inference, belief updates, and probabilistic queries.
"""

Explanation: This module docstring describes the purpose and key features of the 50 Probabilistic Reasoning pattern.

Imports Section

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from collections import defaultdict
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Tuple: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

ProbabilityDistribution (Line 16)

Description: Discrete probability distribution.

Methods:

normalize (Line 21)
def normalize(self):
    """Normalize probabilities to sum to 1."""

Purpose: Normalize probabilities to sum to 1.

Parameters: – self: Instance reference

get_prob (Line 28)
def get_prob(self, value):
    """Get probability of a value."""

Purpose: Get probability of a value.

Parameters: – self: Instance reference – value: Parameter for value

most_likely (Line 32)
def most_likely(self):
    """Get most likely value."""

Purpose: Get most likely value.

Parameters: – self: Instance reference

Evidence (Line 41)

Description: Observed evidence.

BayesianNetwork (Line 48)

Description: Simple Bayesian network for probabilistic reasoning.

Methods:

__init__ (Line 51)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

add_prior (Line 55)
def add_prior(self, distribution):
    """Add prior probability distribution."""

Purpose: Add prior probability distribution.

Parameters: – self: Instance reference – distribution: Parameter for distribution

add_conditional (Line 60)
def add_conditional(self, variable, given_var, given_value, var_value, probability):
    """Add conditional probability P(variable=var_value | given_var=given_value)."""

Purpose: Add conditional probability P(variable=var_value | given_var=given_value).

Parameters: – self: Instance reference – variable: Parameter for variable – given_var: Parameter for given_var – given_value: Parameter for given_value – var_value: Parameter for var_value – probability: Parameter for probability

query (Line 65)
def query(self, variable, evidence):
    """Query probability of variable given evidence (Bayesian inference)."""

Purpose: Query probability of variable given evidence (Bayesian inference).

Parameters: – self: Instance reference – variable: Parameter for variable – evidence: Parameter for evidence

ProbabilisticAgent (Line 96)

Description: Agent that reasons probabilistically.

Methods:

__init__ (Line 99)
def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

update_beliefs (Line 105)
def update_beliefs(self, evidence):
    """Update beliefs based on evidence."""

Purpose: Update beliefs based on evidence.

Parameters: – self: Instance reference – evidence: Parameter for evidence

query_belief (Line 111)
def query_belief(self, variable, evidence):
    """Query belief about variable."""

Purpose: Query belief about variable.

Parameters: – self: Instance reference – variable: Parameter for variable – evidence: Parameter for evidence

make_decision (Line 125)
def make_decision(self, variable, evidence):
    """Make decision based on probabilistic reasoning."""

Purpose: Make decision based on probabilistic reasoning.

Parameters: – self: Instance reference – variable: Parameter for variable – evidence: Parameter for evidence

ProbabilisticReasoningPattern (Line 134)

Description: Probabilistic Reasoning Pattern – Reason under uncertainty.

Agents use Bayesian inference to update beliefs and make decisions. Handles uncertain evidence and probabilistic queries.

Methods:

__init__ (Line 142)
def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 145)
def register_agent(self, agent):
    """Register probabilistic agent."""

Purpose: Register probabilistic agent.

Parameters: – self: Instance reference – agent: Parameter for agent

run (Line 149)
def run(self, scenarios):
    """Execute probabilistic reasoning."""

Purpose: Execute probabilistic reasoning.

Parameters: – self: Instance reference – scenarios: Parameter for scenarios

Design Patterns Used

  • Strategy Pattern: Different execution strategies for various scenarios
  • State Pattern: Managing different states in the workflow
  • Chain of Responsibility: Sequential processing through workflow nodes
  • Observer Pattern: Tracking and monitoring pattern execution
  • Factory Pattern: Creating instances with proper configuration

Usage Example

# Initialize the pattern
pattern = ProbabilityDistribution(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

  • LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
  • Workflow Orchestration: LangGraph manages execution flow through nodes
  • State Management: TypedDict defines workflow state structure
  • Error Handling: Comprehensive error handling and fallback mechanisms
  • Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts