Category: aiml-architecture

Agentic Architecture Pattern: Probabilistic Reasoning


    """
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

November 4, 2025

Agentic Architecture Pattern: Heuristic Based


    """
Pattern 49: Heuristic-Based Pattern

Makes decisions using domain-specific rules of thumb and shortcuts.
Features heuristic evaluation, greedy search, and practical approximations.
"""

from typing import Dict, Any, List, Optional, Callable, Tuple
from dataclasses import dataclass, field
from heapq import heappush, heappop
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


@dataclass
class SearchNode:
    """Node in search space."""
    state: Any
    cost: float
    heuristic: float
    parent: Optional['SearchNode'] = None
    
    def __lt__(self, other):
        """Compare by f-score for heap."""
        return (self.cost + self.heuristic)  List[Tuple[Any, float]]:
        """Evaluate options using heuristic."""
        if heuristic_name not in self.heuristics:
            print(f"   Warning: Heuristic '{heuristic_name}' not found")
            return [(opt, 0.0) for opt in options]
        
        heuristic = self.heuristics[heuristic_name]
        heuristic.usage_count += 1
        
        scored_options = []
        for option in options:
            score = heuristic.evaluate(option)
            scored_options.append((option, score))
        
        # Sort by score (lower is better)
        scored_options.sort(key=lambda x: x[1])
        return scored_options
    
    def greedy_selection(self, options: List[Any], 
                        heuristic_name: str) -> Any:
        """Select best option using greedy heuristic."""
        scored = self.evaluate_options(options, heuristic_name)
        if scored:
            best_option, best_score = scored[0]
            print(f"   {self.name}: Selected option with score {best_score:.2f}")
            return best_option
        return None
    
    def a_star_search(self, start_state: Any, goal_test: Callable[[Any], bool],
                     get_neighbors: Callable[[Any], List[Tuple[Any, float]]],
                     heuristic_name: str, max_iterations: int = 100) -> Optional[List[Any]]:
        """A* search using heuristic."""
        if heuristic_name not in self.heuristics:
            return None
        
        heuristic_fn = self.heuristics[heuristic_name]
        
        # Priority queue
        open_set = []
        start_node = SearchNode(
            state=start_state,
            cost=0,
            heuristic=heuristic_fn.evaluate(start_state)
        )
        heappush(open_set, start_node)
        
        closed_set = set()
        iterations = 0
        
        while open_set and iterations  Dict[str, Any]:
        """Execute heuristic-based decisions."""
        print("=== Heuristic-Based Pattern Demo ===\n")
        
        for scenario in scenarios:
            scenario_type = scenario.get("type", "greedy")
            print(f"\nScenario: {scenario.get('description', 'Unknown')}")
            
            for agent in self.agents.values():
                if scenario_type == "greedy":
                    options = scenario.get("options", [])
                    heuristic = scenario.get("heuristic", "default")
                    result = agent.greedy_selection(options, heuristic)
                    print(f"   Selected: {result}")
                
                elif scenario_type == "search":
                    start = scenario.get("start")
                    goal_test = scenario.get("goal_test")
                    neighbors = scenario.get("get_neighbors")
                    heuristic = scenario.get("heuristic", "default")
                    path = agent.a_star_search(start, goal_test, neighbors, heuristic)
                    if path:
                        print(f"   Path: {' -> '.join(map(str, path))}")
        
        # Statistics
        total_heuristics = sum(
            len(agent.heuristics) for agent in self.agents.values()
        )
        
        return {
            "pattern": "HeuristicBased",
            "total_agents": len(self.agents),
            "total_heuristics": total_heuristics,
            "scenarios": len(scenarios)
        }


# Example usage
if __name__ == "__main__":
    print("=== Heuristic-Based Pattern Demo ===\n")
    
    # Create pattern
    pattern = HeuristicBasedPattern()
    
    # Create agent
    print("1. Creating heuristic-based agent...")
    agent = HeuristicAgent("RouteFinderAgent")
    
    # Define heuristics
    print("\n2. Registering heuristics...")
    
    # Manhattan distance heuristic
    def manhattan_distance(state):
        """Manhattan distance to goal (5, 5)."""
        x, y = state
        return abs(x - 5) + abs(y - 5)
    
    agent.register_heuristic(Heuristic(
        name="manhattan",
        evaluate=manhattan_distance,
        description="Manhattan distance to goal"
    ))
    
    # Distance heuristic for travel
    def distance_heuristic(option):
        """Prefer shorter distances."""
        return option.get("distance", 0)
    
    agent.register_heuristic(Heuristic(
        name="distance",
        evaluate=distance_heuristic,
        description="Minimize distance"
    ))
    
    # Cost heuristic
    def cost_heuristic(option):
        """Prefer lower cost."""
        return option.get("cost", 0)
    
    agent.register_heuristic(Heuristic(
        name="cost",
        evaluate=cost_heuristic,
        description="Minimize cost"
    ))
    
    pattern.register_agent(agent)
    
    # Scenario 1: Greedy selection
    print("\n3. Running scenarios...")
    
    scenarios = [
        {
            "type": "greedy",
            "description": "Choose delivery route",
            "options": [
                {"route": "A", "distance": 10, "cost": 50},
                {"route": "B", "distance": 8, "cost": 60},
                {"route": "C", "distance": 12, "cost": 45},
            ],
            "heuristic": "distance"
        },
        {
            "type": "greedy",
            "description": "Choose service provider",
            "options": [
                {"provider": "FastCo", "distance": 5, "cost": 100},
                {"provider": "CheapCo", "distance": 15, "cost": 50},
                {"provider": "BalanceCo", "distance": 8, "cost": 75},
            ],
            "heuristic": "cost"
        },
        {
            "type": "search",
            "description": "Find path in grid",
            "start": (0, 0),
            "goal_test": lambda state: state == (5, 5),
            "get_neighbors": lambda state: [
                ((state[0]+1, state[1]), 1),
                ((state[0], state[1]+1), 1),
                ((state[0]-1, state[1]), 1),
                ((state[0], state[1]-1), 1),
            ],
            "heuristic": "manhattan"
        }
    ]
    
    summary = pattern.run(scenarios)
    
    print(f"\n4. Summary:")
    print(f"   Total Heuristics: {summary['total_heuristics']}")
    print(f"   Scenarios Executed: {summary['scenarios']}")
    
    print("\n5. Heuristic Usage:")
    for h_name, heuristic in agent.heuristics.items():
        print(f"   {h_name}: used {heuristic.usage_count} times")
    
    print("\n✓ Heuristics enabled fast approximate decisions!")

Agentic Architecture Pattern: Heuristic Based

Pattern 49: 49 Heuristic Based – Line-by-Line Explanation

Overview

Pattern 49: Heuristic-Based Pattern

Makes decisions using domain-specific rules of thumb and shortcuts. Features heuristic evaluation, greedy search, and practical approximations.

Code Structure Analysis

File Information

Pattern Number: 49
Pattern Name: 49 Heuristic Based
File Path: patterns\decision_making\49_heuristic_based.py
Total Lines: 286

Detailed Code Explanation

Module Docstring

"""
Pattern 49: Heuristic-Based Pattern

Makes decisions using domain-specific rules of thumb and shortcuts.
Features heuristic evaluation, greedy search, and practical approximations.
"""

Explanation: This module docstring describes the purpose and key features of the 49 Heuristic Based pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable, Tuple
from dataclasses import dataclass, field
from heapq import heappush, heappop
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, Callable, 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

SearchNode (Line 16)

Description: Node in search space.

Methods:

`lt` (Line 23)

def __lt__(self, other):
    """Compare by f-score for heap."""

Purpose: Compare by f-score for heap.

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

Heuristic (Line 29)

Description: Represents a heuristic function.

HeuristicAgent (Line 37)

Description: Agent that uses heuristics for decision-making.

Methods:

`init` (Line 40)

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”)

`register_heuristic` (Line 45)

def register_heuristic(self, heuristic):
    """Register a heuristic."""

Purpose: Register a heuristic.

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

`evaluate_options` (Line 49)

def evaluate_options(self, options, heuristic_name):
    """Evaluate options using heuristic."""

Purpose: Evaluate options using heuristic.

Parameters: – self: Instance reference – options: Parameter for options – heuristic_name: Parameter for heuristic_name

`greedy_selection` (Line 68)

def greedy_selection(self, options, heuristic_name):
    """Select best option using greedy heuristic."""

Purpose: Select best option using greedy heuristic.

Parameters: – self: Instance reference – options: Parameter for options – heuristic_name: Parameter for heuristic_name

`a_star_search` (Line 78)

def a_star_search(self, start_state, goal_test, get_neighbors, heuristic_name, max_iterations):
    """A* search using heuristic."""

Purpose: A* search using heuristic.

Parameters: – self: Instance reference – start_state: Parameter for start_state – goal_test: Parameter for goal_test – get_neighbors: Parameter for get_neighbors – heuristic_name: Parameter for heuristic_name – max_iterations: Parameter for max_iterations

HeuristicBasedPattern (Line 134)

Description: Heuristic-Based Pattern – Fast approximate decisions.

Agents use domain-specific heuristics for efficient decision-making. Supports greedy selection and A* search.

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 heuristic agent."""

Purpose: Register heuristic agent.

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

`run` (Line 149)

def run(self, scenarios):
    """Execute heuristic-based decisions."""

Purpose: Execute heuristic-based decisions.

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 = SearchNode(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

November 4, 2025

Agentic Architecture Pattern: Optimization


    """
Pattern 48: Optimization Pattern

Finds optimal solutions by minimizing/maximizing objective functions.
Features hill climbing, simulated annealing, and genetic algorithms.
"""

from typing import Dict, Any, List, Optional, Callable, Tuple
from dataclasses import dataclass
import random
import math
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


@dataclass
class Solution:
    """Represents a candidate solution."""
    variables: Dict[str, float]
    objective_value: float
    
    def copy(self) -> 'Solution':
        """Create a copy of the solution."""
        return Solution(
            variables=self.variables.copy(),
            objective_value=self.objective_value
        )


class OptimizationProblem:
    """Defines an optimization problem."""
    
    def __init__(self, objective: Callable[[Dict[str, float]], float],
                 is_minimization: bool = True):
        self.objective = objective
        self.is_minimization = is_minimization
        self.bounds: Dict[str, Tuple[float, float]] = {}
    
    def set_bounds(self, variable: str, lower: float, upper: float):
        """Set bounds for a variable."""
        self.bounds[variable] = (lower, upper)
    
    def evaluate(self, variables: Dict[str, float]) -> float:
        """Evaluate objective function."""
        return self.objective(variables)
    
    def random_solution(self) -> Solution:
        """Generate random solution within bounds."""
        variables = {}
        for var, (lower, upper) in self.bounds.items():
            variables[var] = random.uniform(lower, upper)
        
        obj_value = self.evaluate(variables)
        return Solution(variables=variables, objective_value=obj_value)
    
    def neighbor_solution(self, solution: Solution, step_size: float = 0.1) -> Solution:
        """Generate neighbor solution."""
        new_vars = solution.variables.copy()
        
        # Randomly perturb one variable
        var = random.choice(list(new_vars.keys()))
        lower, upper = self.bounds[var]
        
        # Add random perturbation
        perturbation = random.uniform(-step_size, step_size) * (upper - lower)
        new_vars[var] = max(lower, min(upper, new_vars[var] + perturbation))
        
        obj_value = self.evaluate(new_vars)
        return Solution(variables=new_vars, objective_value=obj_value)


class HillClimbingOptimizer:
    """Hill climbing optimization algorithm."""
    
    def __init__(self, problem: OptimizationProblem):
        self.problem = problem
    
    def optimize(self, max_iterations: int = 100) -> Solution:
        """Run hill climbing."""
        current = self.problem.random_solution()
        
        for iteration in range(max_iterations):
            neighbor = self.problem.neighbor_solution(current, step_size=0.1)
            
            # Move to neighbor if better
            is_better = (neighbor.objective_value  current.objective_value)
            
            if is_better:
                current = neighbor
        
        return current


class SimulatedAnnealingOptimizer:
    """Simulated annealing optimization."""
    
    def __init__(self, problem: OptimizationProblem):
        self.problem = problem
    
    def optimize(self, max_iterations: int = 1000, 
                initial_temp: float = 100.0) -> Solution:
        """Run simulated annealing."""
        current = self.problem.random_solution()
        best = current.copy()
        
        for iteration in range(max_iterations):
            # Temperature decreases over time
            temperature = initial_temp * (1 - iteration / max_iterations)
            
            if temperature < 0.01:
                break
            
            neighbor = self.problem.neighbor_solution(current, step_size=0.2)
            
            # Calculate acceptance probability
            delta = neighbor.objective_value - current.objective_value
            if not self.problem.is_minimization:
                delta = -delta
            
            if delta < 0 or random.random() < math.exp(-delta / temperature):
                current = neighbor
                
                # Update best
                is_better = (current.objective_value  best.objective_value)
                if is_better:
                    best = current.copy()
        
        return best


class OptimizationAgent:
    """Agent that solves optimization problems."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.problems: List[OptimizationProblem] = []
    
    def add_problem(self, problem: OptimizationProblem):
        """Add optimization problem."""
        self.problems.append(problem)
    
    def solve(self, algorithm: str = "simulated_annealing") -> List[Solution]:
        """Solve all problems."""
        solutions = []
        
        for i, problem in enumerate(self.problems, 1):
            print(f"\n   {self.name}: Solving problem {i}...")
            
            if algorithm == "hill_climbing":
                optimizer = HillClimbingOptimizer(problem)
                solution = optimizer.optimize(max_iterations=100)
            else:  # simulated_annealing
                optimizer = SimulatedAnnealingOptimizer(problem)
                solution = optimizer.optimize(max_iterations=500)
            
            solutions.append(solution)
            print(f"   Optimal value: {solution.objective_value:.4f}")
            print(f"   Variables: {solution.variables}")
        
        return solutions


class OptimizationPattern:
    """
    Optimization Pattern - Find optimal solutions.
    
    Agents solve optimization problems using search algorithms.
    Supports hill climbing, simulated annealing, and other methods.
    """
    
    def __init__(self):
        self.agents: Dict[str, OptimizationAgent] = {}
    
    def register_agent(self, agent: OptimizationAgent):
        """Register optimization agent."""
        self.agents[agent.name] = agent
    
    def run(self, algorithm: str = "simulated_annealing") -> Dict[str, Any]:
        """Execute optimization."""
        print("=== Optimization Pattern Demo ===\n")
        
        all_solutions = []
        for agent in self.agents.values():
            solutions = agent.solve(algorithm=algorithm)
            all_solutions.extend(solutions)
        
        return {
            "pattern": "Optimization",
            "algorithm": algorithm,
            "total_agents": len(self.agents),
            "problems_solved": len(all_solutions)
        }


# Example usage
if __name__ == "__main__":
    print("=== Optimization Pattern Demo ===\n")
    
    # Create pattern
    pattern = OptimizationPattern()
    
    # Create agent
    print("1. Creating optimization agent...")
    agent = OptimizationAgent("OptimizerAgent")
    
    # Problem 1: Minimize a quadratic function
    print("\n2. Defining optimization problems...")
    
    def quadratic(vars: Dict[str, float]) -> float:
        """Minimize (x-3)^2 + (y+2)^2"""
        x = vars.get('x', 0)
        y = vars.get('y', 0)
        return (x - 3)**2 + (y + 2)**2
    
    problem1 = OptimizationProblem(objective=quadratic, is_minimization=True)
    problem1.set_bounds('x', -10, 10)
    problem1.set_bounds('y', -10, 10)
    agent.add_problem(problem1)
    
    # Problem 2: Maximize profit function
    def profit(vars: Dict[str, float]) -> float:
        """Maximize: 5*x + 3*y - 0.5*x^2 - 0.3*y^2"""
        x = vars.get('x', 0)
        y = vars.get('y', 0)
        return 5*x + 3*y - 0.5*x**2 - 0.3*y**2
    
    problem2 = OptimizationProblem(objective=profit, is_minimization=False)
    problem2.set_bounds('x', 0, 10)
    problem2.set_bounds('y', 0, 10)
    agent.add_problem(problem2)
    
    pattern.register_agent(agent)
    
    # Solve
    print("\n3. Optimizing...")
    summary = pattern.run(algorithm="simulated_annealing")
    
    print(f"\n4. Summary:")
    print(f"   Algorithm: {summary['algorithm']}")
    print(f"   Problems Solved: {summary['problems_solved']}")
    
    print("\n✓ Optimization found optimal solutions!")

Agentic Architecture Pattern: Optimization

Pattern 48: 48 Optimization – Line-by-Line Explanation

Overview

Pattern 48: Optimization Pattern

Finds optimal solutions by minimizing/maximizing objective functions. Features hill climbing, simulated annealing, and genetic algorithms.

Code Structure Analysis

File Information

Pattern Number: 48
Pattern Name: 48 Optimization
File Path: patterns\decision_making\48_optimization.py
Total Lines: 248

Detailed Code Explanation

Module Docstring

"""
Pattern 48: Optimization Pattern

Finds optimal solutions by minimizing/maximizing objective functions.
Features hill climbing, simulated annealing, and genetic algorithms.
"""

Explanation: This module docstring describes the purpose and key features of the 48 Optimization pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable, Tuple
from dataclasses import dataclass
import random
import math
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, Callable, Tuple: Type hints for better code documentation – from dataclasses import dataclass: 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

Solution (Line 17)

Description: Represents a candidate solution.

Methods:

`copy` (Line 22)

def copy(self):
    """Create a copy of the solution."""

Purpose: Create a copy of the solution.

Parameters: – self: Instance reference

OptimizationProblem (Line 30)

Description: Defines an optimization problem.

Methods:

`init` (Line 33)

def __init__(self, objective, is_minimization):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – objective: Parameter for objective – is_minimization: Parameter for is_minimization

`set_bounds` (Line 39)

def set_bounds(self, variable, lower, upper):
    """Set bounds for a variable."""

Purpose: Set bounds for a variable.

Parameters: – self: Instance reference – variable: Parameter for variable – lower: Parameter for lower – upper: Parameter for upper

`evaluate` (Line 43)

def evaluate(self, variables):
    """Evaluate objective function."""

Purpose: Evaluate objective function.

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

`random_solution` (Line 47)

def random_solution(self):
    """Generate random solution within bounds."""

Purpose: Generate random solution within bounds.

Parameters: – self: Instance reference

`neighbor_solution` (Line 56)

def neighbor_solution(self, solution, step_size):
    """Generate neighbor solution."""

Purpose: Generate neighbor solution.

Parameters: – self: Instance reference – solution: Parameter for solution – step_size: Parameter for step_size

HillClimbingOptimizer (Line 72)

Description: Hill climbing optimization algorithm.

Methods:

`init` (Line 75)

def __init__(self, problem):
    """No method docstring."""

Purpose: No method docstring.

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

`optimize` (Line 78)

def optimize(self, max_iterations):
    """Run hill climbing."""

Purpose: Run hill climbing.

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

SimulatedAnnealingOptimizer (Line 96)

Description: Simulated annealing optimization.

Methods:

`init` (Line 99)

def __init__(self, problem):
    """No method docstring."""

Purpose: No method docstring.

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

`optimize` (Line 102)

def optimize(self, max_iterations, initial_temp):
    """Run simulated annealing."""

Purpose: Run simulated annealing.

Parameters: – self: Instance reference – max_iterations: Parameter for max_iterations – initial_temp: Parameter for initial_temp

OptimizationAgent (Line 135)

Description: Agent that solves optimization problems.

Methods:

`init` (Line 138)

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”)

`add_problem` (Line 143)

def add_problem(self, problem):
    """Add optimization problem."""

Purpose: Add optimization problem.

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

`solve` (Line 147)

def solve(self, algorithm):
    """Solve all problems."""

Purpose: Solve all problems.

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

OptimizationPattern (Line 168)

Description: Optimization Pattern – Find optimal solutions.

Agents solve optimization problems using search algorithms. Supports hill climbing, simulated annealing, and other methods.

Methods:

`init` (Line 176)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`register_agent` (Line 179)

def register_agent(self, agent):
    """Register optimization agent."""

Purpose: Register optimization agent.

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

`run` (Line 183)

def run(self, algorithm):
    """Execute optimization."""

Purpose: Execute optimization.

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

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 = Solution(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

November 4, 2025

Agentic Architecture Pattern: Constraint Based


    """
Pattern 47: Constraint-Based Pattern

Solves problems by satisfying constraints rather than optimizing objectives.
Features constraint propagation, backtracking, and feasibility checking.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class ConstraintType(Enum):
    """Types of constraints."""
    EQUALITY = "equality"
    INEQUALITY = "inequality"
    MEMBERSHIP = "membership"
    CUSTOM = "custom"


@dataclass
class Constraint:
    """Represents a constraint on variables."""
    name: str
    variables: List[str]
    constraint_type: ConstraintType
    check_function: Callable[[Dict[str, Any]], bool]
    description: str = ""


@dataclass
class Variable:
    """Decision variable with domain."""
    name: str
    domain: List[Any]  # Possible values
    value: Optional[Any] = None


class ConstraintSolver:
    """Solver for constraint satisfaction problems."""
    
    def __init__(self):
        self.variables: Dict[str, Variable] = {}
        self.constraints: List[Constraint] = []
        self.solutions: List[Dict[str, Any]] = []
    
    def add_variable(self, variable: Variable):
        """Add variable to problem."""
        self.variables[variable.name] = variable
    
    def add_constraint(self, constraint: Constraint):
        """Add constraint to problem."""
        self.constraints.append(constraint)
    
    def check_constraints(self, assignment: Dict[str, Any]) -> bool:
        """Check if assignment satisfies all constraints."""
        for constraint in self.constraints:
            # Check only if all constraint variables are assigned
            if all(var in assignment for var in constraint.variables):
                if not constraint.check_function(assignment):
                    return False
        return True
    
    def backtrack(self, assignment: Dict[str, Any]) -> bool:
        """Backtracking search for solutions."""
        # Check if complete assignment
        if len(assignment) == len(self.variables):
            if self.check_constraints(assignment):
                self.solutions.append(assignment.copy())
                return True
            return False
        
        # Select unassigned variable
        unassigned = [v for v in self.variables.keys() if v not in assignment]
        if not unassigned:
            return False
        
        var_name = unassigned[0]
        variable = self.variables[var_name]
        
        # Try each value in domain
        for value in variable.domain:
            assignment[var_name] = value
            
            # Check if consistent with constraints so far
            if self.check_constraints(assignment):
                if self.backtrack(assignment):
                    return True
            
            # Backtrack
            del assignment[var_name]
        
        return False
    
    def solve(self, find_all: bool = False) -> List[Dict[str, Any]]:
        """Solve constraint satisfaction problem."""
        self.solutions = []
        
        if find_all:
            # Find all solutions
            self.backtrack_all({})
        else:
            # Find one solution
            self.backtrack({})
        
        return self.solutions
    
    def backtrack_all(self, assignment: Dict[str, Any]):
        """Find all solutions."""
        if len(assignment) == len(self.variables):
            if self.check_constraints(assignment):
                self.solutions.append(assignment.copy())
            return
        
        unassigned = [v for v in self.variables.keys() if v not in assignment]
        if not unassigned:
            return
        
        var_name = unassigned[0]
        variable = self.variables[var_name]
        
        for value in variable.domain:
            assignment[var_name] = value
            if self.check_constraints(assignment):
                self.backtrack_all(assignment)
            del assignment[var_name]


class ConstraintBasedAgent:
    """Agent that solves problems using constraint satisfaction."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.solver = ConstraintSolver()
    
    def solve_problem(self, find_all: bool = False) -> List[Dict[str, Any]]:
        """Solve the constraint problem."""
        print(f"\n   {self.name}: Solving constraint problem...")
        print(f"   Variables: {len(self.solver.variables)}")
        print(f"   Constraints: {len(self.solver.constraints)}")
        
        solutions = self.solver.solve(find_all=find_all)
        
        print(f"   Found {len(solutions)} solution(s)")
        return solutions


class ConstraintBasedPattern:
    """
    Constraint-Based Pattern - Constraint satisfaction solving.
    
    Solves problems by finding assignments that satisfy all constraints.
    Uses backtracking search with constraint propagation.
    """
    
    def __init__(self):
        self.agents: Dict[str, ConstraintBasedAgent] = {}
    
    def register_agent(self, agent: ConstraintBasedAgent):
        """Register constraint-based agent."""
        self.agents[agent.name] = agent
    
    def run(self) -> Dict[str, Any]:
        """Execute constraint solving."""
        print("=== Constraint-Based Pattern Demo ===\n")
        
        all_solutions = []
        for agent in self.agents.values():
            solutions = agent.solve_problem(find_all=True)
            all_solutions.extend(solutions)
        
        return {
            "pattern": "ConstraintBased",
            "total_agents": len(self.agents),
            "total_solutions": len(all_solutions)
        }


# Example usage
if __name__ == "__main__":
    print("=== Constraint-Based Pattern Demo ===\n")
    
    # Create pattern
    pattern = ConstraintBasedPattern()
    
    # Create agent
    print("1. Creating constraint-based agent for scheduling...")
    agent = ConstraintBasedAgent("SchedulerAgent")
    
    # Define variables - meeting times
    print("\n2. Defining variables (meeting time slots)...")
    for meeting in ["meeting_A", "meeting_B", "meeting_C"]:
        agent.solver.add_variable(Variable(
            name=meeting,
            domain=[9, 10, 11, 14, 15, 16]  # Available hours
        ))
    
    # Define constraints
    print("\n3. Adding constraints...")
    
    # All meetings at different times
    agent.solver.add_constraint(Constraint(
        name="all_different",
        variables=["meeting_A", "meeting_B", "meeting_C"],
        constraint_type=ConstraintType.CUSTOM,
        check_function=lambda a: len(set(a.values())) == len(a),
        description="All meetings at different times"
    ))
    
    # Meeting A before lunch (before 12)
    agent.solver.add_constraint(Constraint(
        name="A_before_lunch",
        variables=["meeting_A"],
        constraint_type=ConstraintType.INEQUALITY,
        check_function=lambda a: a.get("meeting_A", 0) = 14,
        description="Meeting B after lunch"
    ))
    
    # Meeting C not adjacent to Meeting A
    agent.solver.add_constraint(Constraint(
        name="C_not_adjacent_A",
        variables=["meeting_A", "meeting_C"],
        constraint_type=ConstraintType.CUSTOM,
        check_function=lambda a: abs(a.get("meeting_A", 0) - a.get("meeting_C", 0)) > 1,
        description="Meeting C not adjacent to A"
    ))
    
    pattern.register_agent(agent)
    
    # Solve
    print("\n4. Solving...")
    summary = pattern.run()
    
    # Display solutions
    print("\n5. Valid Schedules:")
    for i, solution in enumerate(agent.solver.solutions, 1):
        print(f"   Solution {i}:")
        for meeting, time in sorted(solution.items()):
            print(f"     {meeting}: {time}:00")
    
    print(f"\n6. Summary:")
    print(f"   Total Solutions: {summary['total_solutions']}")
    
    print("\n✓ Constraint satisfaction found valid assignments!")

Agentic Architecture Pattern: Constraint Based

Pattern 47: 47 Constraint Based – Line-by-Line Explanation

Overview

Pattern 47: Constraint-Based Pattern

Solves problems by satisfying constraints rather than optimizing objectives. Features constraint propagation, backtracking, and feasibility checking.

Code Structure Analysis

File Information

Pattern Number: 47
Pattern Name: 47 Constraint Based
File Path: patterns\decision_making\47_constraint_based.py
Total Lines: 257

Detailed Code Explanation

Module Docstring

"""
Pattern 47: Constraint-Based Pattern

Solves problems by satisfying constraints rather than optimizing objectives.
Features constraint propagation, backtracking, and feasibility checking.
"""

Explanation: This module docstring describes the purpose and key features of the 47 Constraint Based pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
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, Callable: Type hints for better code documentation – from dataclasses import dataclass: Data class decorators for clean data structures – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

ConstraintType (Line 15)

class ConstraintType(Enum):
    EQUALITY
    INEQUALITY
    MEMBERSHIP
    CUSTOM

Purpose: Types of constraints.

Values: – EQUALITY: Represents equality – INEQUALITY: Represents inequality – MEMBERSHIP: Represents membership – CUSTOM: Represents custom

Classes

ConstraintType (Line 15)

Description: Types of constraints.

Constraint (Line 24)

Description: Represents a constraint on variables.

Variable (Line 34)

Description: Decision variable with domain.

ConstraintSolver (Line 41)

Description: Solver for constraint satisfaction problems.

Methods:

`init` (Line 44)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`add_variable` (Line 49)

def add_variable(self, variable):
    """Add variable to problem."""

Purpose: Add variable to problem.

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

`add_constraint` (Line 53)

def add_constraint(self, constraint):
    """Add constraint to problem."""

Purpose: Add constraint to problem.

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

`check_constraints` (Line 57)

def check_constraints(self, assignment):
    """Check if assignment satisfies all constraints."""

Purpose: Check if assignment satisfies all constraints.

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

`backtrack` (Line 66)

def backtrack(self, assignment):
    """Backtracking search for solutions."""

Purpose: Backtracking search for solutions.

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

`solve` (Line 97)

def solve(self, find_all):
    """Solve constraint satisfaction problem."""

Purpose: Solve constraint satisfaction problem.

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

`backtrack_all` (Line 110)

def backtrack_all(self, assignment):
    """Find all solutions."""

Purpose: Find all solutions.

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

ConstraintBasedAgent (Line 131)

Description: Agent that solves problems using constraint satisfaction.

Methods:

`init` (Line 134)

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”)

`solve_problem` (Line 139)

def solve_problem(self, find_all):
    """Solve the constraint problem."""

Purpose: Solve the constraint problem.

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

ConstraintBasedPattern (Line 151)

Description: Constraint-Based Pattern – Constraint satisfaction solving.

Solves problems by finding assignments that satisfy all constraints. Uses backtracking search with constraint propagation.

Methods:

`init` (Line 159)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`register_agent` (Line 162)

def register_agent(self, agent):
    """Register constraint-based agent."""

Purpose: Register constraint-based agent.

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

`run` (Line 166)

def run(self):
    """Execute constraint solving."""

Purpose: Execute constraint solving.

Parameters: – self: Instance reference

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 = ConstraintType(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

November 4, 2025

Agentic Architecture Pattern: Model Based Agent


    """
Pattern 46: Model-Based Agent Pattern

Agent maintains internal world model to predict outcomes and plan actions.
Features state transition modeling, prediction, and model updating.
"""

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class StateType(Enum):
    """Types of state variables."""
    NUMERIC = "numeric"
    BOOLEAN = "boolean"
    CATEGORICAL = "categorical"


@dataclass
class WorldState:
    """Current state of the world."""
    variables: Dict[str, Any]
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Action:
    """Action that can be performed."""
    name: str
    parameters: Dict[str, Any] = field(default_factory=dict)


@dataclass
class Transition:
    """State transition record."""
    from_state: WorldState
    action: Action
    to_state: WorldState
    reward: float = 0.0


class WorldModel:
    """Internal model of environment dynamics."""
    
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm
        self.transitions: List[Transition] = []
        self.state_variables: Dict[str, StateType] = {}
    
    def define_variable(self, name: str, var_type: StateType):
        """Define a state variable."""
        self.state_variables[name] = var_type
    
    def predict_next_state(self, current_state: WorldState, action: Action) -> WorldState:
        """Predict next state after action using LLM."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a world model predicting state transitions. "
                      "Predict the next state after an action is performed."),
            ("human", """Current State: {current_state}
Action: {action_name} with parameters {action_params}

Predict the next state. Return as JSON-like dict with same variable names.
Only respond with the predicted state dict.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "current_state": str(current_state.variables),
            "action_name": action.name,
            "action_params": str(action.parameters)
        })
        
        try:
            # Parse LLM response
            import ast
            predicted_vars = ast.literal_eval(response.content.strip())
            return WorldState(variables=predicted_vars)
        except:
            # Fallback: return current state
            return WorldState(variables=current_state.variables.copy())
    
    def update_model(self, transition: Transition):
        """Update model with observed transition."""
        self.transitions.append(transition)
    
    def evaluate_action(self, current_state: WorldState, action: Action) -> float:
        """Evaluate expected reward of action."""
        predicted_state = self.predict_next_state(current_state, action)
        
        # Use LLM to estimate reward
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Evaluate how good a state is. Return a number from -10 to 10."),
            ("human", "State: {state}\nRate this state's desirability:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"state": str(predicted_state.variables)})
        
        try:
            reward = float(response.content.strip())
            return max(-10, min(10, reward))
        except:
            return 0.0


class ModelBasedAgent:
    """Agent that uses world model for decision-making."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.world_model = WorldModel(self.llm)
        self.current_state: Optional[WorldState] = None
        self.action_history: List[Tuple[Action, WorldState]] = []
    
    def perceive(self, state: WorldState):
        """Update agent's perception of world state."""
        if self.current_state and self.action_history:
            # Learn from transition
            last_action, _ = self.action_history[-1]
            transition = Transition(
                from_state=self.current_state,
                action=last_action,
                to_state=state
            )
            self.world_model.update_model(transition)
        
        self.current_state = state
        print(f"   {self.name} perceived state: {state.variables}")
    
    def plan_action(self, available_actions: List[Action]) -> Action:
        """Select best action based on model predictions."""
        if not self.current_state:
            return available_actions[0]
        
        print(f"   {self.name} evaluating {len(available_actions)} possible actions...")
        
        # Evaluate each action
        action_scores = []
        for action in available_actions:
            score = self.world_model.evaluate_action(self.current_state, action)
            action_scores.append((action, score))
            print(f"     - {action.name}: score {score:.2f}")
        
        # Select best action
        action_scores.sort(key=lambda x: x[1], reverse=True)
        best_action = action_scores[0][0]
        
        print(f"   {self.name} selected: {best_action.name}")
        return best_action
    
    def execute(self, action: Action):
        """Execute action and record."""
        self.action_history.append((action, self.current_state))
        print(f"   {self.name} executing: {action.name}")


class ModelBasedAgentPattern:
    """
    Model-Based Agent Pattern - World model for predictions.
    
    Agent maintains internal model of environment dynamics.
    Uses model to predict outcomes and select optimal actions.
    """
    
    def __init__(self):
        self.agents: Dict[str, ModelBasedAgent] = {}
    
    def register_agent(self, agent: ModelBasedAgent):
        """Register model-based agent."""
        self.agents[agent.name] = agent
    
    def run(self, simulation_steps: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Run simulation with model-based agents."""
        print("=== Model-Based Agent Pattern Demo ===\n")
        
        for step_num, step in enumerate(simulation_steps, 1):
            print(f"\nStep {step_num}:")
            
            # Update state
            state = WorldState(variables=step["state"])
            available_actions = [
                Action(name=a["name"], parameters=a.get("params", {}))
                for a in step["actions"]
            ]
            
            for agent in self.agents.values():
                agent.perceive(state)
                action = agent.plan_action(available_actions)
                agent.execute(action)
        
        # Statistics
        total_transitions = sum(
            len(agent.world_model.transitions) 
            for agent in self.agents.values()
        )
        
        return {
            "pattern": "ModelBasedAgent",
            "total_agents": len(self.agents),
            "simulation_steps": len(simulation_steps),
            "learned_transitions": total_transitions
        }


# Example usage
if __name__ == "__main__":
    print("=== Model-Based Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = ModelBasedAgentPattern()
    
    # Create agent
    print("1. Creating model-based agent...")
    agent = ModelBasedAgent("RobotAgent")
    
    # Define state variables
    agent.world_model.define_variable("battery", StateType.NUMERIC)
    agent.world_model.define_variable("location", StateType.CATEGORICAL)
    agent.world_model.define_variable("task_complete", StateType.BOOLEAN)
    
    pattern.register_agent(agent)
    print(f"   Agent created with world model")
    
    # Simulation
    print("\n2. Running simulation...")
    
    simulation = [
        {
            "state": {"battery": 100, "location": "start", "task_complete": False},
            "actions": [
                {"name": "move_to_target"},
                {"name": "charge_battery"},
                {"name": "wait"}
            ]
        },
        {
            "state": {"battery": 80, "location": "target", "task_complete": False},
            "actions": [
                {"name": "perform_task"},
                {"name": "return_to_base"},
                {"name": "charge_battery"}
            ]
        },
        {
            "state": {"battery": 60, "location": "target", "task_complete": True},
            "actions": [
                {"name": "return_to_base"},
                {"name": "charge_battery"},
                {"name": "wait"}
            ]
        },
    ]
    
    summary = pattern.run(simulation)
    
    print(f"\n3. Summary:")
    print(f"   Simulation Steps: {summary['simulation_steps']}")
    print(f"   Learned Transitions: {summary['learned_transitions']}")
    print(f"   Actions Taken: {len(agent.action_history)}")
    
    print("\n✓ Agent used world model to predict and plan!")

Agentic Architecture Pattern: Model Based Agent

Pattern 46: 46 Model Based Agent – Line-by-Line Explanation

Overview

Pattern 46: Model-Based Agent Pattern

Agent maintains internal world model to predict outcomes and plan actions. Features state transition modeling, prediction, and model updating.

Code Structure Analysis

File Information

Pattern Number: 46
Pattern Name: 46 Model Based Agent
File Path: patterns\decision_making\46_model_based_agent.py
Total Lines: 267

Detailed Code Explanation

Module Docstring

"""
Pattern 46: Model-Based Agent Pattern

Agent maintains internal world model to predict outcomes and plan actions.
Features state transition modeling, prediction, and model updating.
"""

Explanation: This module docstring describes the purpose and key features of the 46 Model Based Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
            import ast  # 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 datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

StateType (Line 16)

class StateType(Enum):
    NUMERIC
    BOOLEAN
    CATEGORICAL

Purpose: Types of state variables.

Values: – NUMERIC: Represents numeric – BOOLEAN: Represents boolean – CATEGORICAL: Represents categorical

Classes

StateType (Line 16)

Description: Types of state variables.

WorldState (Line 24)

Description: Current state of the world.

Action (Line 31)

Description: Action that can be performed.

Transition (Line 38)

Description: State transition record.

WorldModel (Line 46)

Description: Internal model of environment dynamics.

Methods:

`init` (Line 49)

def __init__(self, llm):
    """No method docstring."""

Purpose: No method docstring.

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

`define_variable` (Line 54)

def define_variable(self, name, var_type):
    """Define a state variable."""

Purpose: Define a state variable.

Parameters: – self: Instance reference – name: Identifier or label – var_type: Parameter for var_type

`predict_next_state` (Line 58)

def predict_next_state(self, current_state, action):
    """Predict next state after action using LLM."""

Purpose: Predict next state after action using LLM.

Parameters: – self: Instance reference – current_state: Parameter for current_state – action: Parameter for action

`update_model` (Line 86)

def update_model(self, transition):
    """Update model with observed transition."""

Purpose: Update model with observed transition.

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

`evaluate_action` (Line 90)

def evaluate_action(self, current_state, action):
    """Evaluate expected reward of action."""

Purpose: Evaluate expected reward of action.

Parameters: – self: Instance reference – current_state: Parameter for current_state – action: Parameter for action

ModelBasedAgent (Line 110)

Description: Agent that uses world model for decision-making.

Methods:

`init` (Line 113)

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”)

`perceive` (Line 120)

def perceive(self, state):
    """Update agent's perception of world state."""

Purpose: Update agent’s perception of world state.

Parameters: – self: Instance reference – state: Current workflow state

`plan_action` (Line 135)

def plan_action(self, available_actions):
    """Select best action based on model predictions."""

Purpose: Select best action based on model predictions.

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

`execute` (Line 156)

def execute(self, action):
    """Execute action and record."""

Purpose: Execute action and record.

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

ModelBasedAgentPattern (Line 162)

Description: Model-Based Agent Pattern – World model for predictions.

Agent maintains internal model of environment dynamics. Uses model to predict outcomes and select optimal actions.

Methods:

`init` (Line 170)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`register_agent` (Line 173)

def register_agent(self, agent):
    """Register model-based agent."""

Purpose: Register model-based agent.

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

`run` (Line 177)

def run(self, simulation_steps):
    """Run simulation with model-based agents."""

Purpose: Run simulation with model-based agents.

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

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 = StateType(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

November 4, 2025

Agentic Architecture Pattern: Case Based Reasoning


    """
Pattern 45: Case-Based Reasoning Pattern

Solves new problems by adapting solutions from similar past cases.
Features case retrieval, adaptation, and case library management.
"""

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


@dataclass
class Case:
    """Represents a past problem-solution case."""
    case_id: str
    problem: str
    solution: str
    context: Dict[str, Any]
    success_rating: float = 0.0  # 0.0 to 1.0
    usage_count: int = 0
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class CaseMatch:
    """Represents a case match with similarity."""
    case: Case
    similarity: float  # 0.0 to 1.0


class CaseLibrary:
    """Storage for past cases."""
    
    def __init__(self):
        self.cases: List[Case] = []
        self.llm = ChatOpenAI(model="gpt-4", temperature=0.3)
    
    def add_case(self, case: Case):
        """Add new case to library."""
        self.cases.append(case)
    
    def find_similar(self, problem: str, top_k: int = 3) -> List[CaseMatch]:
        """Find similar cases using LLM-based similarity."""
        if not self.cases:
            return []
        
        # Use LLM to score similarity
        matches = []
        for case in self.cases:
            prompt = ChatPromptTemplate.from_messages([
                ("system", "Rate the similarity between two problems on a scale of 0.0 to 1.0. "
                          "Only respond with the number."),
                ("human", "Problem 1: {problem1}\nProblem 2: {problem2}")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({
                "problem1": problem,
                "problem2": case.problem
            })
            
            try:
                similarity = float(response.content.strip())
                similarity = max(0.0, min(1.0, similarity))
                matches.append(CaseMatch(case=case, similarity=similarity))
            except:
                continue
        
        # Sort by similarity and success rating
        matches.sort(key=lambda m: (m.similarity, m.case.success_rating), reverse=True)
        return matches[:top_k]


class CaseBasedReasoningAgent:
    """Agent that solves problems using case-based reasoning."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.case_library = CaseLibrary()
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.solutions_generated = 0
    
    def retrieve_and_adapt(self, problem: str, context: Dict[str, Any]) -> str:
        """Retrieve similar cases and adapt solution."""
        print(f"\n   {self.name}: Retrieving similar cases...")
        
        # Find similar cases
        similar_cases = self.case_library.find_similar(problem, top_k=3)
        
        if not similar_cases:
            print(f"   No similar cases found. Generating new solution...")
            return self._generate_new_solution(problem, context)
        
        # Display similar cases
        print(f"   Found {len(similar_cases)} similar cases:")
        for i, match in enumerate(similar_cases, 1):
            print(f"   {i}. Similarity: {match.similarity:.2f} - {match.case.problem[:50]}...")
        
        # Adapt solution from best match
        best_match = similar_cases[0]
        best_match.case.usage_count += 1
        
        if best_match.similarity >= 0.7:
            print(f"   High similarity ({best_match.similarity:.2f}). Adapting solution...")
            return self._adapt_solution(problem, context, best_match.case)
        else:
            print(f"   Low similarity ({best_match.similarity:.2f}). Generating new solution with case guidance...")
            return self._generate_with_guidance(problem, context, similar_cases)
    
    def _adapt_solution(self, problem: str, context: Dict[str, Any], case: Case) -> str:
        """Adapt an existing solution to new problem."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are adapting a past solution to a new but similar problem."),
            ("human", """Past Problem: {past_problem}
Past Solution: {past_solution}

New Problem: {new_problem}
New Context: {new_context}

Adapt the past solution to solve the new problem. Explain what changes you made.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "past_problem": case.problem,
            "past_solution": case.solution,
            "new_problem": problem,
            "new_context": str(context)
        })
        
        return response.content
    
    def _generate_with_guidance(self, problem: str, context: Dict[str, Any], 
                                 similar_cases: List[CaseMatch]) -> str:
        """Generate solution with guidance from similar cases."""
        case_examples = "\n".join([
            f"Example {i}: {match.case.problem}\nSolution: {match.case.solution}\n"
            for i, match in enumerate(similar_cases, 1)
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "Solve the problem using insights from similar past cases."),
            ("human", """Similar Past Cases:
{case_examples}

New Problem: {problem}
Context: {context}

Generate a solution for the new problem, drawing inspiration from the past cases.""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "case_examples": case_examples,
            "problem": problem,
            "context": str(context)
        })
        
        return response.content
    
    def _generate_new_solution(self, problem: str, context: Dict[str, Any]) -> str:
        """Generate completely new solution."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a problem solver. Generate a solution."),
            ("human", "Problem: {problem}\nContext: {context}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"problem": problem, "context": str(context)})
        return response.content
    
    def store_case(self, problem: str, solution: str, context: Dict[str, Any], 
                   success_rating: float = 0.8):
        """Store new case in library."""
        case = Case(
            case_id=f"case_{len(self.case_library.cases) + 1}",
            problem=problem,
            solution=solution,
            context=context,
            success_rating=success_rating
        )
        self.case_library.add_case(case)
        print(f"   Stored case: {case.case_id}")


class CaseBasedReasoningPattern:
    """
    Case-Based Reasoning Pattern - Learn from past experiences.
    
    Retrieves similar past cases, adapts their solutions to new problems.
    Builds reusable case library over time.
    """
    
    def __init__(self):
        self.agents: Dict[str, CaseBasedReasoningAgent] = {}
    
    def register_agent(self, agent: CaseBasedReasoningAgent):
        """Register CBR agent."""
        self.agents[agent.name] = agent
    
    def run(self, scenarios: List[Dict[str, Any]]) -> Dict[str, Any]:
        """Execute CBR pattern on scenarios."""
        print("=== Case-Based Reasoning Pattern Demo ===\n")
        
        for scenario in scenarios:
            problem = scenario["problem"]
            context = scenario.get("context", {})
            
            print(f"\nScenario: {problem[:60]}...")
            
            for agent in self.agents.values():
                solution = agent.retrieve_and_adapt(problem, context)
                print(f"   Solution: {solution[:100]}...")
                
                # Store for future reuse
                if scenario.get("store", True):
                    agent.store_case(problem, solution, context)
        
        # Statistics
        total_cases = sum(len(agent.case_library.cases) for agent in self.agents.values())
        
        return {
            "pattern": "CaseBasedReasoning",
            "total_agents": len(self.agents),
            "total_cases": total_cases,
            "scenarios_processed": len(scenarios)
        }


# Example usage
if __name__ == "__main__":
    print("=== Case-Based Reasoning Pattern Demo ===\n")
    
    # Create pattern
    pattern = CaseBasedReasoningPattern()
    
    # Create agent
    agent = CaseBasedReasoningAgent("TechSupportAgent")
    pattern.register_agent(agent)
    
    # Populate with initial cases
    print("1. Populating case library with historical cases...")
    initial_cases = [
        Case("case_1", "Computer won't start", 
             "Check power cable and press power button for 10 seconds to reset", 
             {"device": "desktop"}, success_rating=0.9),
        Case("case_2", "Internet is slow",
             "Restart router, check for bandwidth-heavy apps, contact ISP if persistent",
             {"connection": "wifi"}, success_rating=0.85),
        Case("case_3", "Printer not printing",
             "Check ink levels, restart printer, reinstall drivers",
             {"device": "printer"}, success_rating=0.8),
    ]
    
    for case in initial_cases:
        agent.case_library.add_case(case)
    
    print(f"   Added {len(initial_cases)} historical cases")
    
    # Test scenarios
    print("\n2. Solving new problems using CBR...")
    
    scenarios = [
        {"problem": "Laptop won't turn on", "context": {"device": "laptop"}, "store": True},
        {"problem": "WiFi keeps disconnecting", "context": {"connection": "wifi"}, "store": True},
        {"problem": "Monitor has no display", "context": {"device": "monitor"}, "store": False},
    ]
    
    summary = pattern.run(scenarios)
    
    print(f"\n3. Summary:")
    print(f"   Total Cases in Library: {summary['total_cases']}")
    print(f"   Scenarios Processed: {summary['scenarios_processed']}")
    
    print("\n4. Case Library Statistics:")
    for case in agent.case_library.cases:
        print(f"   {case.case_id}: Used {case.usage_count} times (rating: {case.success_rating})")
    
    print("\n✓ Case-based reasoning reused past solutions!")

Agentic Architecture Pattern: Case Based Reasoning

Pattern 45: 45 Case Based Reasoning – Line-by-Line Explanation

Overview

Pattern 45: Case-Based Reasoning Pattern

Solves new problems by adapting solutions from similar past cases. Features case retrieval, adaptation, and case library management.

Code Structure Analysis

File Information

Pattern Number: 45
Pattern Name: 45 Case Based Reasoning
File Path: patterns\decision_making\45_case_based_reasoning.py
Total Lines: 283

Detailed Code Explanation

Module Docstring

"""
Pattern 45: Case-Based Reasoning Pattern

Solves new problems by adapting solutions from similar past cases.
Features case retrieval, adaptation, and case library management.
"""

Explanation: This module docstring describes the purpose and key features of the 45 Case Based Reasoning pattern.

Imports Section

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

Classes

Case (Line 16)

Description: Represents a past problem-solution case.

CaseMatch (Line 28)

Description: Represents a case match with similarity.

CaseLibrary (Line 34)

Description: Storage for past cases.

Methods:

`init` (Line 37)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`add_case` (Line 41)

def add_case(self, case):
    """Add new case to library."""

Purpose: Add new case to library.

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

`find_similar` (Line 45)

def find_similar(self, problem, top_k):
    """Find similar cases using LLM-based similarity."""

Purpose: Find similar cases using LLM-based similarity.

Parameters: – self: Instance reference – problem: Parameter for problem – top_k: Parameter for top_k

CaseBasedReasoningAgent (Line 77)

Description: Agent that solves problems using case-based reasoning.

Methods:

`init` (Line 80)

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”)

`retrieve_and_adapt` (Line 86)

def retrieve_and_adapt(self, problem, context):
    """Retrieve similar cases and adapt solution."""

Purpose: Retrieve similar cases and adapt solution.

Parameters: – self: Instance reference – problem: Parameter for problem – context: Parameter for context

`_adapt_solution` (Line 113)

def _adapt_solution(self, problem, context, case):
    """Adapt an existing solution to new problem."""

Purpose: Adapt an existing solution to new problem.

Parameters: – self: Instance reference – problem: Parameter for problem – context: Parameter for context – case: Parameter for case

`_generate_with_guidance` (Line 136)

def _generate_with_guidance(self, problem, context, similar_cases):
    """Generate solution with guidance from similar cases."""

Purpose: Generate solution with guidance from similar cases.

Parameters: – self: Instance reference – problem: Parameter for problem – context: Parameter for context – similar_cases: Parameter for similar_cases

`_generate_new_solution` (Line 164)

def _generate_new_solution(self, problem, context):
    """Generate completely new solution."""

Purpose: Generate completely new solution.

Parameters: – self: Instance reference – problem: Parameter for problem – context: Parameter for context

`store_case` (Line 175)

def store_case(self, problem, solution, context, success_rating):
    """Store new case in library."""

Purpose: Store new case in library.

Parameters: – self: Instance reference – problem: Parameter for problem – solution: Parameter for solution – context: Parameter for context – success_rating: Parameter for success_rating

CaseBasedReasoningPattern (Line 189)

Description: Case-Based Reasoning Pattern – Learn from past experiences.

Retrieves similar past cases, adapts their solutions to new problems. Builds reusable case library over time.

Methods:

`init` (Line 197)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`register_agent` (Line 200)

def register_agent(self, agent):
    """Register CBR agent."""

Purpose: Register CBR agent.

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

`run` (Line 204)

def run(self, scenarios):
    """Execute CBR pattern on scenarios."""

Purpose: Execute CBR pattern on scenarios.

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 = Case(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

November 4, 2025

Agentic Architecture Pattern: Rule Based Agent


    """
Pattern 44: Rule-Based Agent Pattern

Agents make decisions using explicit IF-THEN rules.
Features rule engine, conflict resolution, and rule priority.
"""

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class RulePriority(Enum):
    """Priority level of rule."""
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4


@dataclass
class Rule:
    """IF-THEN rule."""
    rule_id: str
    name: str
    condition: Callable[[Dict[str, Any]], bool]  # Function that checks condition
    action: Callable[[Dict[str, Any]], str]  # Function that executes action
    priority: RulePriority = RulePriority.MEDIUM
    enabled: bool = True
    fire_count: int = 0


@dataclass
class RuleExecution:
    """Record of rule execution."""
    rule_id: str
    input_facts: Dict[str, Any]
    condition_met: bool
    action_result: Optional[str]
    timestamp: datetime = field(default_factory=datetime.now)


class RuleBasedAgent:
    """Agent that makes decisions using rules."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.rules: List[Rule] = []
        self.execution_history: List[RuleExecution] = []
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
    
    def add_rule(self, rule: Rule):
        """Add rule to agent's rule base."""
        self.rules.append(rule)
    
    def evaluate_rules(self, facts: Dict[str, Any]) -> List[Rule]:
        """Find all rules whose conditions are met."""
        matching_rules = []
        
        for rule in self.rules:
            if not rule.enabled:
                continue
            
            try:
                if rule.condition(facts):
                    matching_rules.append(rule)
            except Exception as e:
                print(f"   Error evaluating rule {rule.rule_id}: {e}")
        
        # Sort by priority
        matching_rules.sort(key=lambda r: r.priority.value, reverse=True)
        
        return matching_rules
    
    def execute_rule(self, rule: Rule, facts: Dict[str, Any]) -> str:
        """Execute a rule's action."""
        try:
            result = rule.action(facts)
            rule.fire_count += 1
            
            # Record execution
            execution = RuleExecution(
                rule_id=rule.rule_id,
                input_facts=facts.copy(),
                condition_met=True,
                action_result=result
            )
            self.execution_history.append(execution)
            
            return result
        
        except Exception as e:
            execution = RuleExecution(
                rule_id=rule.rule_id,
                input_facts=facts.copy(),
                condition_met=True,
                action_result=None
            )
            self.execution_history.append(execution)
            return f"Error executing rule: {e}"
    
    def process(self, facts: Dict[str, Any], execute_all: bool = False) -> List[str]:
        """Process facts through rule engine."""
        matching_rules = self.evaluate_rules(facts)
        results = []
        
        if not matching_rules:
            return ["No rules matched"]
        
        print(f"   {self.name}: {len(matching_rules)} rules matched")
        
        if execute_all:
            # Execute all matching rules
            for rule in matching_rules:
                result = self.execute_rule(rule, facts)
                results.append(f"{rule.name}: {result}")
                print(f"   - Fired: {rule.name} (priority: {rule.priority.name})")
        else:
            # Execute only highest priority rule
            top_rule = matching_rules[0]
            result = self.execute_rule(top_rule, facts)
            results.append(f"{top_rule.name}: {result}")
            print(f"   - Fired: {top_rule.name} (priority: {top_rule.priority.name})")
        
        return results


class RuleBasedAgentPattern:
    """
    Rule-Based Agent Pattern - Explicit rule-based decision making.
    
    Agents use IF-THEN rules to make deterministic decisions.
    Supports rule priority, conflict resolution, and rule management.
    """
    
    def __init__(self):
        self.agents: Dict[str, RuleBasedAgent] = {}
    
    def register_agent(self, agent: RuleBasedAgent):
        """Register rule-based agent."""
        self.agents[agent.name] = agent
    
    def create_standard_rules(self) -> List[Rule]:
        """Create standard business logic rules."""
        rules = [
            # Temperature control rules
            Rule(
                "temp_high",
                "High Temperature Alert",
                lambda facts: facts.get("temperature", 0) > 30,
                lambda facts: f"Alert: Temperature {facts['temperature']}°C is too high! Turn on cooling.",
                priority=RulePriority.HIGH
            ),
            Rule(
                "temp_low",
                "Low Temperature Alert",
                lambda facts: facts.get("temperature", 0) < 10,
                lambda facts: f"Alert: Temperature {facts['temperature']}°C is too low! Turn on heating.",
                priority=RulePriority.HIGH
            ),
            Rule(
                "temp_normal",
                "Normal Temperature",
                lambda facts: 10 <= facts.get("temperature", 0) <= 30,
                lambda facts: f"Temperature {facts['temperature']}°C is normal.",
                priority=RulePriority.LOW
            ),
            
            # Inventory rules
            Rule(
                "inventory_critical",
                "Critical Inventory",
                lambda facts: facts.get("inventory", 100) < 10,
                lambda facts: f"CRITICAL: Only {facts['inventory']} items left! Emergency reorder.",
                priority=RulePriority.CRITICAL
            ),
            Rule(
                "inventory_low",
                "Low Inventory",
                lambda facts: 10 <= facts.get("inventory", 100)  Dict[str, Any]:
        """Execute rule-based pattern with test cases."""
        print(f"\n   Testing scenario: {scenario}")
        
        all_results = []
        
        for i, facts in enumerate(test_cases, 1):
            print(f"\n   Test Case {i}: {facts}")
            
            for agent in self.agents.values():
                results = agent.process(facts, execute_all=False)
                all_results.extend(results)
        
        # Statistics
        total_rules = sum(len(agent.rules) for agent in self.agents.values())
        total_executions = sum(len(agent.execution_history) for agent in self.agents.values())
        
        return {
            "pattern": "RuleBasedAgent",
            "total_agents": len(self.agents),
            "total_rules": total_rules,
            "test_cases": len(test_cases),
            "total_executions": total_executions,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Rule-Based Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = RuleBasedAgentPattern()
    
    # Create agent with rules
    print("1. Creating rule-based agent...")
    agent = RuleBasedAgent("BusinessLogicAgent")
    
    # Add standard rules
    rules = pattern.create_standard_rules()
    for rule in rules:
        agent.add_rule(rule)
    
    print(f"   Added {len(rules)} rules")
    
    # Register agent
    pattern.register_agent(agent)
    
    # Test cases
    print("\n2. Running test cases...")
    
    test_cases = [
        {"temperature": 35, "scenario": "Hot day"},
        {"temperature": 5, "scenario": "Cold day"},
        {"temperature": 22, "scenario": "Pleasant day"},
        {"inventory": 5, "scenario": "Almost out of stock"},
        {"inventory": 30, "scenario": "Low stock"},
        {"customer_type": "VIP", "name": "Alice"},
        {"customer_type": "regular", "name": "Bob"},
    ]
    
    summary = pattern.run("Business automation rules", test_cases)
    
    # Rule statistics
    print("\n3. Rule Statistics:")
    for rule in agent.rules:
        if rule.fire_count > 0:
            print(f"   {rule.name}: fired {rule.fire_count} times (priority: {rule.priority.name})")
    
    print(f"\n4. Summary:")
    print(f"   Total Rules: {summary['total_rules']}")
    print(f"   Test Cases: {summary['test_cases']}")
    print(f"   Executions: {summary['total_executions']}")
    
    print("\n✓ Rule-based agent made deterministic decisions!")

Agentic Architecture Pattern: Rule Based Agent

Pattern 44: 44 Rule Based Agent – Line-by-Line Explanation

Overview

Pattern 44: Rule-Based Agent Pattern

Agents make decisions using explicit IF-THEN rules. Features rule engine, conflict resolution, and rule priority.

Code Structure Analysis

File Information

Pattern Number: 44
Pattern Name: 44 Rule Based Agent
File Path: patterns\decision_making\44_rule_based_agent.py
Total Lines: 282

Detailed Code Explanation

Module Docstring

"""
Pattern 44: Rule-Based Agent Pattern

Agents make decisions using explicit IF-THEN rules.
Features rule engine, conflict resolution, and rule priority.
"""

Explanation: This module docstring describes the purpose and key features of the 44 Rule Based Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
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, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

RulePriority (Line 16)

class RulePriority(Enum):
    LOW
    MEDIUM
    HIGH
    CRITICAL

Purpose: Priority level of rule.

Values: – LOW: Represents low – MEDIUM: Represents medium – HIGH: Represents high – CRITICAL: Represents critical

Classes

RulePriority (Line 16)

Description: Priority level of rule.

Rule (Line 25)

Description: IF-THEN rule.

RuleExecution (Line 37)

Description: Record of rule execution.

RuleBasedAgent (Line 46)

Description: Agent that makes decisions using rules.

Methods:

`init` (Line 49)

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”)

`add_rule` (Line 55)

def add_rule(self, rule):
    """Add rule to agent's rule base."""

Purpose: Add rule to agent’s rule base.

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

`evaluate_rules` (Line 59)

def evaluate_rules(self, facts):
    """Find all rules whose conditions are met."""

Purpose: Find all rules whose conditions are met.

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

`execute_rule` (Line 78)

def execute_rule(self, rule, facts):
    """Execute a rule's action."""

Purpose: Execute a rule’s action.

Parameters: – self: Instance reference – rule: Parameter for rule – facts: Parameter for facts

`process` (Line 105)

def process(self, facts, execute_all):
    """Process facts through rule engine."""

Purpose: Process facts through rule engine.

Parameters: – self: Instance reference – facts: Parameter for facts – execute_all: Parameter for execute_all

RuleBasedAgentPattern (Line 131)

Description: Rule-Based Agent Pattern – Explicit rule-based decision making.

Agents use IF-THEN rules to make deterministic decisions. Supports rule priority, conflict resolution, and rule management.

Methods:

`init` (Line 139)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`register_agent` (Line 142)

def register_agent(self, agent):
    """Register rule-based agent."""

Purpose: Register rule-based agent.

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

`create_standard_rules` (Line 146)

def create_standard_rules(self):
    """Create standard business logic rules."""

Purpose: Create standard business logic rules.

Parameters: – self: Instance reference

`run` (Line 207)

def run(self, scenario, test_cases):
    """Execute rule-based pattern with test cases."""

Purpose: Execute rule-based pattern with test cases.

Parameters: – self: Instance reference – scenario: Parameter for scenario – test_cases: Parameter for test_cases

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 = RulePriority(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

November 4, 2025

Agentic Architecture Pattern: Reasoning Agent


    """
Pattern 43: Reasoning Agent

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


@dataclass
class ReasoningAgentConfig:
    """Configuration for Reasoning Agent."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class ReasoningAgentState:
    """State information for Reasoning Agent."""
    current_step: int = 0
    data: Dict[str, Any] = field(default_factory=dict)
    results: List[Any] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


class ReasoningAgentAgent:
    """Agent implementing Reasoning Agent."""
    
    def __init__(self, name: str, config: ReasoningAgentConfig = None):
        self.name = name
        self.config = config or ReasoningAgentConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = ReasoningAgentState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Reasoning Agent."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Reasoning Agent. "
                      f"Provide clear, structured responses."),
            ("human", "Process this input: {input_data}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"input_data": str(input_data)})
        
        result = {
            "agent": self.name,
            "step": self.state.current_step,
            "input": input_data,
            "output": str(response.content),
            "timestamp": datetime.now()
        }
        
        self.execution_history.append(result)
        self.state.results.append(result)
        
        return result
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get agent statistics."""
        return {
            "name": self.name,
            "total_executions": len(self.execution_history),
            "current_step": self.state.current_step,
            "results_count": len(self.state.results)
        }


class ReasoningAgentPattern:
    """
    Reasoning Agent - Enhanced implementation
    
    This pattern provides comprehensive functionality for reasoning agent
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: ReasoningAgentConfig = None):
        self.config = config or ReasoningAgentConfig()
        self.agents: Dict[str, ReasoningAgentAgent] = {}
        self.workflow = self._build_workflow()
        self.execution_log: List[Dict[str, Any]] = []
    
    def _build_workflow(self) -> StateGraph:
        """Build LangGraph workflow for pattern execution."""
        
        class WorkflowState(TypedDict):
            input: str
            processed: str
            output: str
            metadata: Dict[str, Any]
        
        def process_input(state: WorkflowState) -> WorkflowState:
            """Process input node."""
            return {
                **state,
                "processed": f"Processed: {state['input']}",
                "metadata": {"step": "process_input"}
            }
        
        def generate_output(state: WorkflowState) -> WorkflowState:
            """Generate output node."""
            llm = ChatOpenAI(model=self.config.model_name, temperature=self.config.temperature)
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", "You are implementing Reasoning Agent. Generate comprehensive output."),
                ("human", "{input}")
            ])
            
            chain = prompt | llm
            response = chain.invoke({"input": state["processed"]})
            
            return {
                **state,
                "output": str(response.content),
                "metadata": {**state.get("metadata", {}), "step": "generate_output"}
            }
        
        # Build workflow graph
        workflow = StateGraph(WorkflowState)
        workflow.add_node("process", process_input)
        workflow.add_node("generate", generate_output)
        workflow.set_entry_point("process")
        workflow.add_edge("process", "generate")
        workflow.add_edge("generate", END)
        
        return workflow.compile()
    
    def register_agent(self, agent: ReasoningAgentAgent):
        """Register an agent with the pattern."""
        self.agents[agent.name] = agent
        if self.config.enable_logging:
            print(f"Registered agent: {agent.name}")
    
    def execute(self, input_data: Any, agent_name: str = None) -> Dict[str, Any]:
        """Execute the pattern with given input."""
        
        # Use workflow if no specific agent requested
        if agent_name is None:
            result = self.workflow.invoke({
                "input": str(input_data),
                "processed": "",
                "output": "",
                "metadata": {}
            })
            
            execution_record = {
                "pattern": "Reasoning Agent",
                "input": input_data,
                "output": result.get("output", ""),
                "metadata": result.get("metadata", {}),
                "timestamp": datetime.now()
            }
            
            self.execution_log.append(execution_record)
            return execution_record
        
        # Execute with specific agent
        agent = self.agents.get(agent_name)
        if not agent:
            raise ValueError(f"Agent {agent_name} not found")
        
        result = agent.process(input_data)
        self.execution_log.append(result)
        return result
    
    def run(self, input_data: Any) -> Dict[str, Any]:
        """Main execution method."""
        return self.execute(input_data)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "pattern": "Reasoning Agent",
            "total_executions": len(self.execution_log),
            "registered_agents": len(self.agents),
            "agents": [agent.get_statistics() for agent in self.agents.values()]
        }


# Example usage
if __name__ == "__main__":
    print("=== Reasoning Agent Demo ===\n")
    
    # Create configuration
    config = ReasoningAgentConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = ReasoningAgentPattern(config)
    
    # Create and register agents
    agent1 = ReasoningAgentAgent("Agent1", config)
    agent2 = ReasoningAgentAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate reasoning agent functionality")
    print(f"\nWorkflow Result:")
    print(f"  Output: {result1['output'][:150]}...")
    
    # Execute with specific agents
    result2 = pattern.execute("Specific task for Agent1", agent_name="Agent1")
    print(f"\nAgent1 Result:")
    print(f"  Output: {result2['output'][:150]}...")
    
    result3 = pattern.execute("Specific task for Agent2", agent_name="Agent2")
    print(f"\nAgent2 Result:")
    print(f"  Output: {result3['output'][:150]}...")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Pattern: {stats['pattern']}")
    print(f"Total Executions: {stats['total_executions']}")
    print(f"Registered Agents: {stats['registered_agents']}")
    print(f"\nAgent Statistics:")
    for agent_stats in stats['agents']:
        print(f"  {agent_stats['name']}: {agent_stats['total_executions']} executions")

Agentic Architecture Pattern: Reasoning Agent

Pattern 43: 43 Reasoning Agent – Line-by-Line Explanation

Overview

Pattern 43: Reasoning Agent

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

Pattern Number: 43
Pattern Name: 43 Reasoning Agent
File Path: patterns\decision_making\43_reasoning_agent.py
Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 43: Reasoning Agent

Purpose: Enhanced implementation with full LangChain/LangGraph integration
"""

Explanation: This module docstring describes the purpose and key features of the 43 Reasoning Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration – from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration – from typing_extensions import TypedDict: Type hints for better code documentation

Classes

ReasoningAgentConfig (Line 18)

Description: Configuration for Reasoning Agent.

ReasoningAgentState (Line 27)

Description: State information for Reasoning Agent.

ReasoningAgentAgent (Line 36)

Description: Agent implementing Reasoning Agent.

Methods:

`init` (Line 39)

def __init__(self, name, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – config: Configuration settings

`process` (Line 49)

def process(self, input_data):
    """Process input according to Reasoning Agent."""

Purpose: Process input according to Reasoning Agent.

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

`get_statistics` (Line 75)

def get_statistics(self):
    """Get agent statistics."""

Purpose: Get agent statistics.

Parameters: – self: Instance reference

ReasoningAgentPattern (Line 85)

Description: Reasoning Agent – Enhanced implementation

This pattern provides comprehensive functionality for reasoning agent with full LangChain/LangGraph integration.

Features: – Multi-agent coordination – State management – Execution tracking – LangGraph workflow integration

Methods:

`init` (Line 99)

def __init__(self, config):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – config: Configuration settings

`_build_workflow` (Line 105)

def _build_workflow(self):
    """Build LangGraph workflow for pattern execution."""

Purpose: Build LangGraph workflow for pattern execution.

Parameters: – self: Instance reference

`register_agent` (Line 150)

def register_agent(self, agent):
    """Register an agent with the pattern."""

Purpose: Register an agent with the pattern.

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

`execute` (Line 156)

def execute(self, input_data, agent_name):
    """Execute the pattern with given input."""

Purpose: Execute the pattern with given input.

Parameters: – self: Instance reference – input_data: Parameter for input_data – agent_name: Parameter for agent_name

`run` (Line 188)

def run(self, input_data):
    """Main execution method."""

Purpose: Main execution method.

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

`get_statistics` (Line 192)

def get_statistics(self):
    """Get pattern statistics."""

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

WorkflowState (Line 108)

Description: No class docstring.

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 = ReasoningAgentConfig(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

November 4, 2025

Agentic Architecture Pattern: Planning Agent


    """
Pattern 42: Planning Agent Pattern

Agent that creates plans to achieve goals, considering preconditions, actions, and outcomes.
Uses search algorithms and domain knowledge for goal-directed behavior.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class ActionStatus(Enum):
    PLANNED = "planned"
    EXECUTING = "executing"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Action:
    """An action in a plan."""
    name: str
    preconditions: List[str]
    effects: List[str]
    cost: float = 1.0
    status: ActionStatus = ActionStatus.PLANNED


@dataclass
class Plan:
    """A sequence of actions to achieve a goal."""
    goal: str
    actions: List[Action] = field(default_factory=list)
    current_state: Set[str] = field(default_factory=set)
    target_state: Set[str] = field(default_factory=set)


class PlanningAgentPattern:
    """
    Planning Agent Pattern - Goal-directed planning and execution.
    
    Features:
    - Goal decomposition
    - Precondition checking
    - Action sequencing
    - Plan execution
    - Replanning on failure
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.available_actions: Dict[str, Action] = {}
        self.plans: List[Plan] = []
    
    def register_action(self, action: Action):
        """Register an available action."""
        self.available_actions[action.name] = action
        print(f"✓ Registered action: {action.name}")
        print(f"  Preconditions: {', '.join(action.preconditions)}")
        print(f"  Effects: {', '.join(action.effects)}")
    
    def create_plan(self, goal: str, current_state: Set[str], 
                   target_conditions: List[str]) -> Plan:
        """Create a plan to achieve goal."""
        plan = Plan(
            goal=goal,
            current_state=current_state.copy(),
            target_state=set(target_conditions)
        )
        
        # Use LLM to generate plan
        state_str = ", ".join(current_state)
        target_str = ", ".join(target_conditions)
        actions_str = "\n".join([
            f"- {name}: requires [{', '.join(a.preconditions)}] → produces [{', '.join(a.effects)}]"
            for name, a in self.available_actions.items()
        ])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a planning agent. Create a sequence of actions to achieve the goal. "
                      "List actions in order, one per line."),
            ("human", f"Goal: {goal}\n"
                     f"Current State: {state_str}\n"
                     f"Target Conditions: {target_str}\n\n"
                     f"Available Actions:\n{actions_str}\n\n"
                     f"Plan (list action names in order):")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse action sequence
        lines = [line.strip() for line in response.content.split('\n') if line.strip()]
        for line in lines:
            # Extract action name
            clean_line = line.lstrip('0123456789.-•* ').strip()
            # Find matching action
            for action_name, action_template in self.available_actions.items():
                if action_name.lower() in clean_line.lower():
                    # Create action instance
                    action_instance = Action(
                        name=action_template.name,
                        preconditions=action_template.preconditions.copy(),
                        effects=action_template.effects.copy(),
                        cost=action_template.cost
                    )
                    plan.actions.append(action_instance)
                    break
        
        self.plans.append(plan)
        return plan
    
    def check_preconditions(self, action: Action, state: Set[str]) -> bool:
        """Check if action preconditions are met."""
        return all(precond in state for precond in action.preconditions)
    
    def execute_plan(self, plan: Plan) -> Dict[str, Any]:
        """Execute a plan step by step."""
        current_state = plan.current_state.copy()
        executed_actions = []
        failed_action = None
        
        for action in plan.actions:
            # Check preconditions
            if not self.check_preconditions(action, current_state):
                action.status = ActionStatus.FAILED
                failed_action = action.name
                print(f"✗ Action failed: {action.name} - preconditions not met")
                break
            
            # Execute action
            action.status = ActionStatus.EXECUTING
            print(f"→ Executing: {action.name}")
            
            # Apply effects
            for effect in action.effects:
                current_state.add(effect)
            
            action.status = ActionStatus.COMPLETED
            executed_actions.append(action.name)
            print(f"✓ Completed: {action.name}")
        
        # Check if goal achieved
        goal_achieved = plan.target_state.issubset(current_state)
        
        return {
            "goal": plan.goal,
            "executed_actions": executed_actions,
            "goal_achieved": goal_achieved,
            "failed_action": failed_action,
            "final_state": current_state
        }
    
    def replan(self, original_plan: Plan, current_state: Set[str]) -> Plan:
        """Create a new plan when original plan fails."""
        print(f"\n🔄 Replanning for goal: {original_plan.goal}")
        return self.create_plan(
            original_plan.goal,
            current_state,
            list(original_plan.target_state)
        )
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute planning scenario."""
        return {
            "pattern": "Planning Agent",
            "scenario": scenario,
            "available_actions": len(self.available_actions),
            "plans_created": len(self.plans)
        }


# Example usage
if __name__ == "__main__":
    print("=== Planning Agent Pattern Demo ===\n")
    
    agent = PlanningAgentPattern()
    
    # Register available actions
    print("--- Registering Actions ---")
    agent.register_action(Action(
        "GetIngredients",
        preconditions=["has_money"],
        effects=["has_ingredients"],
        cost=1.0
    ))
    agent.register_action(Action(
        "PrepareFood",
        preconditions=["has_ingredients", "has_kitchen"],
        effects=["has_prepared_food"],
        cost=2.0
    ))
    agent.register_action(Action(
        "CookFood",
        preconditions=["has_prepared_food"],
        effects=["has_cooked_food"],
        cost=1.5
    ))
    agent.register_action(Action(
        "ServeMeal",
        preconditions=["has_cooked_food", "has_table"],
        effects=["meal_ready"],
        cost=0.5
    ))
    
    # Create plan
    print("\n--- Creating Plan ---")
    initial_state = {"has_money", "has_kitchen", "has_table"}
    goal_conditions = ["meal_ready"]
    
    plan = agent.create_plan(
        "Prepare and serve a meal",
        initial_state,
        goal_conditions
    )
    
    print(f"\nPlan for: {plan.goal}")
    print(f"Actions: {', '.join([a.name for a in plan.actions])}")
    
    # Execute plan
    print("\n--- Executing Plan ---")
    result = agent.execute_plan(plan)
    
    print(f"\nExecution Result:")
    print(f"Goal: {result['goal']}")
    print(f"Executed: {', '.join(result['executed_actions'])}")
    print(f"Goal Achieved: {result['goal_achieved']}")
    if result['failed_action']:
        print(f"Failed At: {result['failed_action']}")
    
    # Summary
    print("\n--- Summary ---")
    summary = agent.run("Meal preparation planning")
    print(f"Pattern: {summary['pattern']}")
    print(f"Available Actions: {summary['available_actions']}")
    print(f"Plans Created: {summary['plans_created']}")

Agentic Architecture Pattern: Planning Agent

Pattern 42: 42 Planning Agent – Line-by-Line Explanation

Overview

Pattern 42: Planning Agent Pattern

Agent that creates plans to achieve goals, considering preconditions, actions, and outcomes. Uses search algorithms and domain knowledge for goal-directed behavior.

Code Structure Analysis

File Information

Pattern Number: 42
Pattern Name: 42 Planning Agent
File Path: patterns\decision_making\42_planning_agent.py
Total Lines: 240

Detailed Code Explanation

Module Docstring

"""
Pattern 42: Planning Agent Pattern

Agent that creates plans to achieve goals, considering preconditions, actions, and outcomes.
Uses search algorithms and domain knowledge for goal-directed behavior.
"""

Explanation: This module docstring describes the purpose and key features of the 42 Planning Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from enum import Enum
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, Set: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

ActionStatus (Line 15)

class ActionStatus(Enum):
    PLANNED
    EXECUTING
    COMPLETED
    FAILED

Purpose: No enum docstring.

Values: – PLANNED: Represents planned – EXECUTING: Represents executing – COMPLETED: Successfully finished – FAILED: Execution failed

Classes

ActionStatus (Line 15)

Description: No class docstring.

Action (Line 23)

Description: An action in a plan.

Plan (Line 33)

Description: A sequence of actions to achieve a goal.

PlanningAgentPattern (Line 41)

Description: Planning Agent Pattern – Goal-directed planning and execution.

Features: – Goal decomposition – Precondition checking – Action sequencing – Plan execution – Replanning on failure

Methods:

`init` (Line 53)

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

Purpose: No method docstring.

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

`register_action` (Line 58)

def register_action(self, action):
    """Register an available action."""

Purpose: Register an available action.

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

`create_plan` (Line 65)

def create_plan(self, goal, current_state, target_conditions):
    """Create a plan to achieve goal."""

Purpose: Create a plan to achieve goal.

Parameters: – self: Instance reference – goal: Parameter for goal – current_state: Parameter for current_state – target_conditions: Parameter for target_conditions

`check_preconditions` (Line 116)

def check_preconditions(self, action, state):
    """Check if action preconditions are met."""

Purpose: Check if action preconditions are met.

Parameters: – self: Instance reference – action: Parameter for action – state: Current workflow state

`execute_plan` (Line 120)

def execute_plan(self, plan):
    """Execute a plan step by step."""

Purpose: Execute a plan step by step.

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

`replan` (Line 157)

def replan(self, original_plan, current_state):
    """Create a new plan when original plan fails."""

Purpose: Create a new plan when original plan fails.

Parameters: – self: Instance reference – original_plan: Parameter for original_plan – current_state: Parameter for current_state

`run` (Line 166)

def run(self, scenario):
    """Execute planning scenario."""

Purpose: Execute planning scenario.

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

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 = ActionStatus(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

November 4, 2025

Agentic Architecture Pattern: Bdi


    """
Pattern 41: BDI (Belief-Desire-Intention) Pattern

Classic agent architecture with three mental states:
- Beliefs: What the agent knows about the world
- Desires: Goals the agent wants to achieve
- Intentions: Plans the agent commits to executing
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class BeliefType(Enum):
    """Type of belief."""
    FACT = "fact"
    ASSUMPTION = "assumption"
    OBSERVATION = "observation"
    INFERENCE = "inference"


class DesireType(Enum):
    """Type of desire/goal."""
    ACHIEVEMENT = "achievement"
    MAINTENANCE = "maintenance"
    AVOIDANCE = "avoidance"


class IntentionStatus(Enum):
    """Status of intention."""
    ACTIVE = "active"
    SUSPENDED = "suspended"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Belief:
    """Belief about the world."""
    belief_id: str
    content: str
    belief_type: BeliefType
    confidence: float = 1.0  # 0.0 to 1.0
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Desire:
    """Goal or desire."""
    desire_id: str
    description: str
    desire_type: DesireType
    priority: int = 5  # 1-10
    conditions: List[str] = field(default_factory=list)  # Success conditions


@dataclass
class Intention:
    """Committed plan."""
    intention_id: str
    desire_id: str
    plan_steps: List[str]
    current_step: int = 0
    status: IntentionStatus = IntentionStatus.ACTIVE


@dataclass
class Action:
    """Action executed by agent."""
    action_id: str
    description: str
    intention_id: str
    result: Optional[str] = None
    success: bool = False


class BDIAgent:
    """Agent with Belief-Desire-Intention architecture."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        
        # BDI mental states
        self.beliefs: List[Belief] = []
        self.desires: List[Desire] = []
        self.intentions: List[Intention] = []
        self.actions: List[Action] = []
    
    def add_belief(self, belief: Belief):
        """Add or update belief."""
        # Remove conflicting beliefs (simplified)
        self.beliefs = [b for b in self.beliefs if b.content != belief.content]
        self.beliefs.append(belief)
    
    def add_desire(self, desire: Desire):
        """Add desire/goal."""
        self.desires.append(desire)
    
    def deliberate(self) -> Optional[Desire]:
        """Select most important desire to pursue."""
        if not self.desires:
            return None
        
        # Filter already-pursued desires
        active_desire_ids = {i.desire_id for i in self.intentions if i.status == IntentionStatus.ACTIVE}
        available_desires = [d for d in self.desires if d.desire_id not in active_desire_ids]
        
        if not available_desires:
            return None
        
        # Select highest priority
        selected = max(available_desires, key=lambda d: d.priority)
        return selected
    
    def means_end_reasoning(self, desire: Desire) -> List[str]:
        """Generate plan to achieve desire using beliefs."""
        beliefs_text = "\n".join([f"- {b.content} ({b.belief_type.value}, confidence: {b.confidence})" 
                                  for b in self.beliefs])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, an intelligent agent. Generate a concrete plan to achieve a goal "
                      f"based on your beliefs about the world."),
            ("human", f"Current Beliefs:\n{beliefs_text}\n\n"
                     f"Goal: {desire.description} (Type: {desire.desire_type.value})\n"
                     f"Success conditions: {', '.join(desire.conditions)}\n\n"
                     f"Create a plan with 3-4 specific steps to achieve this goal.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        # Parse steps from response
        content = str(response.content)
        steps = []
        for line in content.split('\n'):
            line = line.strip()
            if line and (line[0].isdigit() or line.startswith('-') or line.startswith('*')):
                # Remove numbering/bullets
                step = line.lstrip('0123456789.-*) ').strip()
                if step:
                    steps.append(step)
        
        return steps[:4] if steps else ["Execute default action"]
    
    def commit_to_intention(self, desire: Desire, plan: List[str]):
        """Create intention from desire and plan."""
        intention = Intention(
            intention_id=f"int_{len(self.intentions)}",
            desire_id=desire.desire_id,
            plan_steps=plan,
            status=IntentionStatus.ACTIVE
        )
        self.intentions.append(intention)
        return intention
    
    def execute_intention(self, intention: Intention) -> bool:
        """Execute next step of intention."""
        if intention.status != IntentionStatus.ACTIVE:
            return False
        
        if intention.current_step >= len(intention.plan_steps):
            intention.status = IntentionStatus.COMPLETED
            return True
        
        # Execute current step
        step = intention.plan_steps[intention.current_step]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} executing a plan step."),
            ("human", f"Step: {step}\n\nDescribe briefly what you do to execute this step.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        action = Action(
            action_id=f"act_{len(self.actions)}",
            description=step,
            intention_id=intention.intention_id,
            result=str(response.content)[:150],
            success=True
        )
        self.actions.append(action)
        
        intention.current_step += 1
        
        return True
    
    def bdi_cycle(self):
        """Execute one BDI reasoning cycle."""
        # 1. Update beliefs (would normally perceive environment)
        print(f"\n   {self.name} BDI Cycle:")
        print(f"   Beliefs: {len(self.beliefs)}")
        print(f"   Desires: {len(self.desires)}")
        print(f"   Active Intentions: {sum(1 for i in self.intentions if i.status == IntentionStatus.ACTIVE)}")
        
        # 2. Deliberate: choose desire
        selected_desire = self.deliberate()
        
        if selected_desire:
            print(f"   ⚡ Selected desire: {selected_desire.description}")
            
            # 3. Means-end reasoning: create plan
            plan = self.means_end_reasoning(selected_desire)
            print(f"   📋 Generated plan with {len(plan)} steps")
            
            # 4. Commit to intention
            intention = self.commit_to_intention(selected_desire, plan)
            print(f"   ✓ Committed to intention: {intention.intention_id}")
        
        # 5. Execute active intentions
        active_intentions = [i for i in self.intentions if i.status == IntentionStatus.ACTIVE]
        for intention in active_intentions:
            if intention.current_step  Dict[str, Any]:
        """Execute BDI pattern."""
        total_beliefs = sum(len(agent.beliefs) for agent in self.agents.values())
        total_desires = sum(len(agent.desires) for agent in self.agents.values())
        total_intentions = sum(len(agent.intentions) for agent in self.agents.values())
        total_actions = sum(len(agent.actions) for agent in self.agents.values())
        
        return {
            "pattern": "BDI",
            "total_agents": len(self.agents),
            "total_beliefs": total_beliefs,
            "total_desires": total_desires,
            "total_intentions": total_intentions,
            "actions_executed": total_actions,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== BDI (Belief-Desire-Intention) Pattern Demo ===\n")
    
    # Create BDI agent
    agent = BDIAgent("Alice")
    
    # Add beliefs about the world
    print("1. Setting up beliefs...")
    agent.add_belief(Belief("b1", "The weather is sunny", BeliefType.OBSERVATION, confidence=1.0))
    agent.add_belief(Belief("b2", "I have a bicycle", BeliefType.FACT, confidence=1.0))
    agent.add_belief(Belief("b3", "The park is 2km away", BeliefType.FACT, confidence=1.0))
    agent.add_belief(Belief("b4", "Exercise is healthy", BeliefType.INFERENCE, confidence=0.9))
    
    # Add desires/goals
    print("2. Setting up desires...")
    agent.add_desire(Desire(
        "d1",
        "Get physical exercise",
        DesireType.ACHIEVEMENT,
        priority=8,
        conditions=["30 minutes of activity", "increased heart rate"]
    ))
    
    agent.add_desire(Desire(
        "d2",
        "Enjoy outdoor activities",
        DesireType.ACHIEVEMENT,
        priority=6,
        conditions=["spend time outside", "have fun"]
    ))
    
    # Create pattern and register agent
    pattern = BDIPattern()
    pattern.register_agent(agent)
    
    # Run BDI cycles
    print("\n3. Running BDI cycles...")
    pattern.run_cycles(num_cycles=2)
    
    # Summary
    print("\n4. BDI Agent Summary:")
    summary = pattern.run("Personal health and recreation planning")
    print(f"   Total Agents: {summary['total_agents']}")
    print(f"   Beliefs: {summary['total_beliefs']}")
    print(f"   Desires: {summary['total_desires']}")
    print(f"   Intentions: {summary['total_intentions']}")
    print(f"   Actions Executed: {summary['actions_executed']}")
    
    print("\n✓ BDI pattern demonstrated rational agent reasoning!")

Agentic Architecture Pattern: Bdi

Pattern 41: 41 Bdi – Line-by-Line Explanation

Overview

Pattern 41: BDI (Belief-Desire-Intention) Pattern

Classic agent architecture with three mental states: – Beliefs: What the agent knows about the world – Desires: Goals the agent wants to achieve – Intentions: Plans the agent commits to executing

Code Structure Analysis

File Information

Pattern Number: 41
Pattern Name: 41 Bdi
File Path: patterns\decision_making\41_bdi.py
Total Lines: 321

Detailed Code Explanation

Module Docstring

"""
Pattern 41: BDI (Belief-Desire-Intention) Pattern

Classic agent architecture with three mental states:
- Beliefs: What the agent knows about the world
- Desires: Goals the agent wants to achieve
- Intentions: Plans the agent commits to executing
"""

Explanation: This module docstring describes the purpose and key features of the 41 Bdi pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Enumerations

BeliefType (Line 18)

class BeliefType(Enum):
    FACT
    ASSUMPTION
    OBSERVATION
    INFERENCE

Purpose: Type of belief.

Values: – FACT: Represents fact – ASSUMPTION: Represents assumption – OBSERVATION: Represents observation – INFERENCE: Represents inference

DesireType (Line 26)

class DesireType(Enum):
    ACHIEVEMENT
    MAINTENANCE
    AVOIDANCE

Purpose: Type of desire/goal.

Values: – ACHIEVEMENT: Represents achievement – MAINTENANCE: Represents maintenance – AVOIDANCE: Represents avoidance

IntentionStatus (Line 33)

class IntentionStatus(Enum):
    ACTIVE
    SUSPENDED
    COMPLETED
    FAILED

Purpose: Status of intention.

Values: – ACTIVE: Currently operational – SUSPENDED: Represents suspended – COMPLETED: Successfully finished – FAILED: Execution failed

Classes

BeliefType (Line 18)

Description: Type of belief.

DesireType (Line 26)

Description: Type of desire/goal.

IntentionStatus (Line 33)

Description: Status of intention.

Belief (Line 42)

Description: Belief about the world.

Desire (Line 52)

Description: Goal or desire.

Intention (Line 62)

Description: Committed plan.

Action (Line 72)

Description: Action executed by agent.

BDIAgent (Line 81)

Description: Agent with Belief-Desire-Intention architecture.

Methods:

`init` (Line 84)

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”)

`add_belief` (Line 94)

def add_belief(self, belief):
    """Add or update belief."""

Purpose: Add or update belief.

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

`add_desire` (Line 100)

def add_desire(self, desire):
    """Add desire/goal."""

Purpose: Add desire/goal.

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

`deliberate` (Line 104)

def deliberate(self):
    """Select most important desire to pursue."""

Purpose: Select most important desire to pursue.

Parameters: – self: Instance reference

`means_end_reasoning` (Line 120)

def means_end_reasoning(self, desire):
    """Generate plan to achieve desire using beliefs."""

Purpose: Generate plan to achieve desire using beliefs.

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

`commit_to_intention` (Line 150)

def commit_to_intention(self, desire, plan):
    """Create intention from desire and plan."""

Purpose: Create intention from desire and plan.

Parameters: – self: Instance reference – desire: Parameter for desire – plan: Parameter for plan

`execute_intention` (Line 161)

def execute_intention(self, intention):
    """Execute next step of intention."""

Purpose: Execute next step of intention.

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

`bdi_cycle` (Line 194)

def bdi_cycle(self):
    """Execute one BDI reasoning cycle."""

Purpose: Execute one BDI reasoning cycle.

Parameters: – self: Instance reference

BDIPattern (Line 224)

Description: BDI (Belief-Desire-Intention) Pattern – Rational agent architecture.

Agents maintain beliefs about the world, have desires (goals), and commit to intentions (plans). The BDI cycle continuously: 1. Updates beliefs from perception 2. Deliberates to select goals 3. Generates plans through means-end reasoning 4. Executes committed intentions

Methods:

`init` (Line 236)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`register_agent` (Line 239)

def register_agent(self, agent):
    """Register BDI agent."""

Purpose: Register BDI agent.

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

`run_cycles` (Line 243)

def run_cycles(self, num_cycles):
    """Run BDI cycles for all agents."""

Purpose: Run BDI cycles for all agents.

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

`run` (Line 253)

def run(self, scenario):
    """Execute BDI pattern."""

Purpose: Execute BDI pattern.

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

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 = BeliefType(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

November 4, 2025

Category: aiml-architecture

Agentic Architecture Pattern: Probabilistic Reasoning

Agentic Architecture Pattern: Probabilistic Reasoning

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

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Classes

ProbabilityDistribution (Line 16)

normalize (Line 21)

get_prob (Line 28)

most_likely (Line 32)

Evidence (Line 41)

BayesianNetwork (Line 48)

__init__ (Line 51)

add_prior (Line 55)

add_conditional (Line 60)

query (Line 65)

ProbabilisticAgent (Line 96)

__init__ (Line 99)

update_beliefs (Line 105)

query_belief (Line 111)

make_decision (Line 125)

ProbabilisticReasoningPattern (Line 134)

__init__ (Line 142)

register_agent (Line 145)

run (Line 149)

Design Patterns Used

Usage Example

Key Concepts

Extension Points

Agentic Architecture Pattern: Heuristic Based

Agentic Architecture Pattern: Heuristic Based

Pattern 49: 49 Heuristic Based – Line-by-Line Explanation

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Classes

SearchNode (Line 16)

__lt__ (Line 23)

Heuristic (Line 29)

HeuristicAgent (Line 37)

__init__ (Line 40)

register_heuristic (Line 45)

evaluate_options (Line 49)

greedy_selection (Line 68)

a_star_search (Line 78)

HeuristicBasedPattern (Line 134)

__init__ (Line 142)

register_agent (Line 145)

run (Line 149)

Design Patterns Used

Usage Example

Key Concepts

Extension Points

Agentic Architecture Pattern: Optimization

Agentic Architecture Pattern: Optimization

Pattern 48: 48 Optimization – Line-by-Line Explanation

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Classes

Solution (Line 17)

copy (Line 22)

OptimizationProblem (Line 30)

__init__ (Line 33)

set_bounds (Line 39)

evaluate (Line 43)

random_solution (Line 47)

neighbor_solution (Line 56)

HillClimbingOptimizer (Line 72)

`normalize` (Line 21)

`get_prob` (Line 28)

`most_likely` (Line 32)

`init` (Line 51)

`add_prior` (Line 55)

`add_conditional` (Line 60)

`query` (Line 65)

`init` (Line 99)

`update_beliefs` (Line 105)

`query_belief` (Line 111)

`make_decision` (Line 125)

`init` (Line 142)

`register_agent` (Line 145)

`run` (Line 149)

`lt` (Line 23)

`init` (Line 40)

`register_heuristic` (Line 45)

`evaluate_options` (Line 49)

`greedy_selection` (Line 68)

`a_star_search` (Line 78)

`init` (Line 142)

`register_agent` (Line 145)

`run` (Line 149)

`copy` (Line 22)

`init` (Line 33)

`set_bounds` (Line 39)

`evaluate` (Line 43)

`random_solution` (Line 47)

`neighbor_solution` (Line 56)

`init` (Line 75)

`optimize` (Line 78)

`init` (Line 99)

`optimize` (Line 102)

`init` (Line 138)

`add_problem` (Line 143)

`solve` (Line 147)

`init` (Line 176)

`register_agent` (Line 179)

`run` (Line 183)

`init` (Line 44)

`add_variable` (Line 49)

`add_constraint` (Line 53)

`check_constraints` (Line 57)

`backtrack` (Line 66)

`solve` (Line 97)

`backtrack_all` (Line 110)

`init` (Line 134)

`solve_problem` (Line 139)

`init` (Line 159)

`register_agent` (Line 162)

`run` (Line 166)

`init` (Line 49)

`define_variable` (Line 54)