aishastr.live

Agentic Architecture Pattern: Rule Based Agent

Written by

admin

in


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

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

Comments

Leave a Reply

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

More posts