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