"""
Pattern 110: Sensor Fusion Pattern

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 SensorFusionConfig:
    """Configuration for Sensor Fusion Pattern."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class SensorFusionState:
    """State information for Sensor Fusion Pattern."""
    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 SensorFusionAgent:
    """Agent implementing Sensor Fusion Pattern."""
    
    def __init__(self, name: str, config: SensorFusionConfig = None):
        self.name = name
        self.config = config or SensorFusionConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = SensorFusionState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Sensor Fusion Pattern."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Sensor Fusion Pattern. "
                      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 SensorFusionPattern:
    """
    Sensor Fusion Pattern - Enhanced implementation
    
    This pattern provides comprehensive functionality for sensor fusion
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: SensorFusionConfig = None):
        self.config = config or SensorFusionConfig()
        self.agents: Dict[str, SensorFusionAgent] = {}
        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 Sensor Fusion Pattern. 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: SensorFusionAgent):
        """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": "Sensor Fusion Pattern",
                "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": "Sensor Fusion Pattern",
            "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("=== Sensor Fusion Pattern Demo ===\n")
    
    # Create configuration
    config = SensorFusionConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = SensorFusionPattern(config)
    
    # Create and register agents
    agent1 = SensorFusionAgent("Agent1", config)
    agent2 = SensorFusionAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate sensor fusion 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: Sensor Fusion

Pattern 110: 110 Sensor Fusion Pattern – Line-by-Line Explanation

Overview

Pattern 110: Sensor Fusion Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 110
• Pattern Name: 110 Sensor Fusion Pattern
• File Path: patterns\perception_sensing\110_sensor_fusion_pattern.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 110: Sensor Fusion Pattern

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

Explanation: This module docstring describes the purpose and key features of the 110 Sensor Fusion Pattern 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

SensorFusionConfig (Line 18)

Description: Configuration for Sensor Fusion Pattern.

SensorFusionState (Line 27)

Description: State information for Sensor Fusion Pattern.

SensorFusionAgent (Line 36)

Description: Agent implementing Sensor Fusion Pattern.

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 Sensor Fusion Pattern."""

Purpose: Process input according to Sensor Fusion Pattern.

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

SensorFusionPattern (Line 85)

Description: Sensor Fusion Pattern – Enhanced implementation

This pattern provides comprehensive functionality for sensor fusion 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 = SensorFusionConfig(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

    """
Pattern 109: State Estimation Pattern

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 StateEstimationConfig:
    """Configuration for State Estimation Pattern."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class StateEstimationState:
    """State information for State Estimation Pattern."""
    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 StateEstimationAgent:
    """Agent implementing State Estimation Pattern."""
    
    def __init__(self, name: str, config: StateEstimationConfig = None):
        self.name = name
        self.config = config or StateEstimationConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = StateEstimationState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to State Estimation Pattern."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing State Estimation Pattern. "
                      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 StateEstimationPattern:
    """
    State Estimation Pattern - Enhanced implementation
    
    This pattern provides comprehensive functionality for state estimation
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: StateEstimationConfig = None):
        self.config = config or StateEstimationConfig()
        self.agents: Dict[str, StateEstimationAgent] = {}
        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 State Estimation Pattern. 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: StateEstimationAgent):
        """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": "State Estimation Pattern",
                "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": "State Estimation Pattern",
            "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("=== State Estimation Pattern Demo ===\n")
    
    # Create configuration
    config = StateEstimationConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = StateEstimationPattern(config)
    
    # Create and register agents
    agent1 = StateEstimationAgent("Agent1", config)
    agent2 = StateEstimationAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate state estimation 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: State Estimation

Pattern 109: 109 State Estimation Pattern – Line-by-Line Explanation

Overview

Pattern 109: State Estimation Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 109
• Pattern Name: 109 State Estimation Pattern
• File Path: patterns\perception_sensing\109_state_estimation_pattern.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 109: State Estimation Pattern

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

Explanation: This module docstring describes the purpose and key features of the 109 State Estimation Pattern 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

StateEstimationConfig (Line 18)

Description: Configuration for State Estimation Pattern.

StateEstimationState (Line 27)

Description: State information for State Estimation Pattern.

StateEstimationAgent (Line 36)

Description: Agent implementing State Estimation Pattern.

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 State Estimation Pattern."""

Purpose: Process input according to State Estimation Pattern.

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

StateEstimationPattern (Line 85)

Description: State Estimation Pattern – Enhanced implementation

This pattern provides comprehensive functionality for state estimation 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 = StateEstimationConfig(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

    """
Pattern 108: World Model Pattern

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 WorldModelConfig:
    """Configuration for World Model Pattern."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class WorldModelState:
    """State information for World Model Pattern."""
    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 WorldModelAgent:
    """Agent implementing World Model Pattern."""
    
    def __init__(self, name: str, config: WorldModelConfig = None):
        self.name = name
        self.config = config or WorldModelConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = WorldModelState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to World Model Pattern."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing World Model Pattern. "
                      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 WorldModelPattern:
    """
    World Model Pattern - Enhanced implementation
    
    This pattern provides comprehensive functionality for world model
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: WorldModelConfig = None):
        self.config = config or WorldModelConfig()
        self.agents: Dict[str, WorldModelAgent] = {}
        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 World Model Pattern. 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: WorldModelAgent):
        """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": "World Model Pattern",
                "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": "World Model Pattern",
            "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("=== World Model Pattern Demo ===\n")
    
    # Create configuration
    config = WorldModelConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = WorldModelPattern(config)
    
    # Create and register agents
    agent1 = WorldModelAgent("Agent1", config)
    agent2 = WorldModelAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate world model 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: World Model

Pattern 108: 108 World Model Pattern – Line-by-Line Explanation

Overview

Pattern 108: World Model Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 108
• Pattern Name: 108 World Model Pattern
• File Path: patterns\perception_sensing\108_world_model_pattern.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 108: World Model Pattern

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

Explanation: This module docstring describes the purpose and key features of the 108 World Model Pattern 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

WorldModelConfig (Line 18)

Description: Configuration for World Model Pattern.

WorldModelState (Line 27)

Description: State information for World Model Pattern.

WorldModelAgent (Line 36)

Description: Agent implementing World Model Pattern.

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 World Model Pattern."""

Purpose: Process input according to World Model Pattern.

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

WorldModelPattern (Line 85)

Description: World Model Pattern – Enhanced implementation

This pattern provides comprehensive functionality for world model 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 = WorldModelConfig(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

    """
Pattern 107: Environment Model Pattern

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 EnvironmentModelConfig:
    """Configuration for Environment Model Pattern."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class EnvironmentModelState:
    """State information for Environment Model Pattern."""
    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 EnvironmentModelAgent:
    """Agent implementing Environment Model Pattern."""
    
    def __init__(self, name: str, config: EnvironmentModelConfig = None):
        self.name = name
        self.config = config or EnvironmentModelConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = EnvironmentModelState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Environment Model Pattern."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Environment Model Pattern. "
                      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 EnvironmentModelPattern:
    """
    Environment Model Pattern - Enhanced implementation
    
    This pattern provides comprehensive functionality for environment model
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: EnvironmentModelConfig = None):
        self.config = config or EnvironmentModelConfig()
        self.agents: Dict[str, EnvironmentModelAgent] = {}
        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 Environment Model Pattern. 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: EnvironmentModelAgent):
        """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": "Environment Model Pattern",
                "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": "Environment Model Pattern",
            "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("=== Environment Model Pattern Demo ===\n")
    
    # Create configuration
    config = EnvironmentModelConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = EnvironmentModelPattern(config)
    
    # Create and register agents
    agent1 = EnvironmentModelAgent("Agent1", config)
    agent2 = EnvironmentModelAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate environment model 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: Environment Model

Pattern 107: 107 Environment Model Pattern – Line-by-Line Explanation

Overview

Pattern 107: Environment Model Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 107
• Pattern Name: 107 Environment Model Pattern
• File Path: patterns\perception_sensing\107_environment_model_pattern.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 107: Environment Model Pattern

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

Explanation: This module docstring describes the purpose and key features of the 107 Environment Model Pattern 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

EnvironmentModelConfig (Line 18)

Description: Configuration for Environment Model Pattern.

EnvironmentModelState (Line 27)

Description: State information for Environment Model Pattern.

EnvironmentModelAgent (Line 36)

Description: Agent implementing Environment Model Pattern.

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 Environment Model Pattern."""

Purpose: Process input according to Environment Model Pattern.

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

EnvironmentModelPattern (Line 85)

Description: Environment Model Pattern – Enhanced implementation

This pattern provides comprehensive functionality for environment model 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 = EnvironmentModelConfig(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

    """
Pattern 106: Context Aware Pattern

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 ContextAwareConfig:
    """Configuration for Context Aware Pattern."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class ContextAwareState:
    """State information for Context Aware Pattern."""
    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 ContextAwareAgent:
    """Agent implementing Context Aware Pattern."""
    
    def __init__(self, name: str, config: ContextAwareConfig = None):
        self.name = name
        self.config = config or ContextAwareConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = ContextAwareState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Context Aware Pattern."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Context Aware Pattern. "
                      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 ContextAwarePattern:
    """
    Context Aware Pattern - Enhanced implementation
    
    This pattern provides comprehensive functionality for context aware
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: ContextAwareConfig = None):
        self.config = config or ContextAwareConfig()
        self.agents: Dict[str, ContextAwareAgent] = {}
        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 Context Aware Pattern. 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: ContextAwareAgent):
        """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": "Context Aware Pattern",
                "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": "Context Aware Pattern",
            "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("=== Context Aware Pattern Demo ===\n")
    
    # Create configuration
    config = ContextAwareConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = ContextAwarePattern(config)
    
    # Create and register agents
    agent1 = ContextAwareAgent("Agent1", config)
    agent2 = ContextAwareAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate context aware 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: Context Aware

Pattern 106: 106 Context Aware Pattern – Line-by-Line Explanation

Overview

Pattern 106: Context Aware Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 106
• Pattern Name: 106 Context Aware Pattern
• File Path: patterns\perception_sensing\106_context_aware_pattern.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 106: Context Aware Pattern

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

Explanation: This module docstring describes the purpose and key features of the 106 Context Aware Pattern 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

ContextAwareConfig (Line 18)

Description: Configuration for Context Aware Pattern.

ContextAwareState (Line 27)

Description: State information for Context Aware Pattern.

ContextAwareAgent (Line 36)

Description: Agent implementing Context Aware Pattern.

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 Context Aware Pattern."""

Purpose: Process input according to Context Aware Pattern.

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

ContextAwarePattern (Line 85)

Description: Context Aware Pattern – Enhanced implementation

This pattern provides comprehensive functionality for context aware 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 = ContextAwareConfig(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

    """
Pattern 105: Situational Awareness

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 SituationalAwarenessConfig:
    """Configuration for Situational Awareness."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class SituationalAwarenessState:
    """State information for Situational Awareness."""
    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 SituationalAwarenessAgent:
    """Agent implementing Situational Awareness."""
    
    def __init__(self, name: str, config: SituationalAwarenessConfig = None):
        self.name = name
        self.config = config or SituationalAwarenessConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = SituationalAwarenessState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Situational Awareness."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Situational Awareness. "
                      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 SituationalAwarenessPattern:
    """
    Situational Awareness - Enhanced implementation
    
    This pattern provides comprehensive functionality for situational awareness
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: SituationalAwarenessConfig = None):
        self.config = config or SituationalAwarenessConfig()
        self.agents: Dict[str, SituationalAwarenessAgent] = {}
        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 Situational Awareness. 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: SituationalAwarenessAgent):
        """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": "Situational Awareness",
                "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": "Situational Awareness",
            "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("=== Situational Awareness Demo ===\n")
    
    # Create configuration
    config = SituationalAwarenessConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = SituationalAwarenessPattern(config)
    
    # Create and register agents
    agent1 = SituationalAwarenessAgent("Agent1", config)
    agent2 = SituationalAwarenessAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate situational awareness 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: Situational Awareness

Pattern 105: 105 Situational Awareness Pattern – Line-by-Line Explanation

Overview

Pattern 105: Situational Awareness Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 105
• Pattern Name: 105 Situational Awareness Pattern
• File Path: patterns\perception_sensing\105_situational_awareness_pattern.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 105: Situational Awareness Pattern

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

Explanation: This module docstring describes the purpose and key features of the 105 Situational Awareness Pattern 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

SituationalAwarenessConfig (Line 18)

Description: Configuration for Situational Awareness Pattern.

SituationalAwarenessState (Line 27)

Description: State information for Situational Awareness Pattern.

SituationalAwarenessAgent (Line 36)

Description: Agent implementing Situational Awareness Pattern.

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 Situational Awareness Pattern."""

Purpose: Process input according to Situational Awareness Pattern.

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

SituationalAwarenessPattern (Line 85)

Description: Situational Awareness Pattern – Enhanced implementation

This pattern provides comprehensive functionality for situational awareness 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 = SituationalAwarenessConfig(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

    """
Pattern 104: Perception Action Loop

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 PerceptionActionLoopConfig:
    """Configuration for Perception Action Loop."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class PerceptionActionLoopState:
    """State information for Perception Action Loop."""
    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 PerceptionActionLoopAgent:
    """Agent implementing Perception Action Loop."""
    
    def __init__(self, name: str, config: PerceptionActionLoopConfig = None):
        self.name = name
        self.config = config or PerceptionActionLoopConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = PerceptionActionLoopState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Perception Action Loop."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Perception Action Loop. "
                      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 PerceptionActionLoopPattern:
    """
    Perception Action Loop - Enhanced implementation
    
    This pattern provides comprehensive functionality for perception action loop
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: PerceptionActionLoopConfig = None):
        self.config = config or PerceptionActionLoopConfig()
        self.agents: Dict[str, PerceptionActionLoopAgent] = {}
        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 Perception Action Loop. 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: PerceptionActionLoopAgent):
        """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": "Perception Action Loop",
                "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": "Perception Action Loop",
            "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("=== Perception Action Loop Demo ===\n")
    
    # Create configuration
    config = PerceptionActionLoopConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = PerceptionActionLoopPattern(config)
    
    # Create and register agents
    agent1 = PerceptionActionLoopAgent("Agent1", config)
    agent2 = PerceptionActionLoopAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate perception action loop 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: Perception Action Loop

Pattern 104: 104 Perception Action Loop Pattern – Line-by-Line Explanation

Overview

Pattern 104: Perception-Action Loop Pattern

Continuous cycle of perceiving environment, processing information, deciding actions, executing, and repeating. Core pattern for embodied agents.

Code Structure Analysis

File Information

• Pattern Number: 104
• Pattern Name: 104 Perception Action Loop Pattern
• File Path: patterns\perception_sensing\104_perception_action_loop_pattern.py
• Total Lines: 286

Detailed Code Explanation

Module Docstring

"""
Pattern 104: Perception-Action Loop Pattern

Continuous cycle of perceiving environment, processing information,
deciding actions, executing, and repeating. Core pattern for embodied agents.
"""

Explanation: This module docstring describes the purpose and key features of the 104 Perception Action Loop Pattern 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

Enumerations

LoopPhase (Line 18)

class LoopPhase(Enum):
    PERCEIVE
    PROCESS
    DECIDE
    ACT

Purpose: Phases in perception-action loop.

Values: – PERCEIVE: Represents perceive – PROCESS: Represents process – DECIDE: Represents decide – ACT: Represents act

Classes

LoopPhase (Line 18)

Description: Phases in perception-action loop.

Percept (Line 27)

Description: Perceived information from environment.

Decision (Line 35)

Description: Decision made by agent.

ActionResult (Line 43)

Description: Result of action execution.

PerceptionActionAgent (Line 50)

Description: Agent that operates in perception-action loop.

Methods:

__init__ (Line 53)

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

def perceive(self, environment):
    """Perceive environment through sensors."""

Purpose: Perceive environment through sensors.

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

process (Line 83)

def process(self, percept):
    """Process perceived information."""

Purpose: Process perceived information.

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

decide (Line 98)

def decide(self, processed_info, context):
    """Decide what action to take."""

Purpose: Decide what action to take.

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

act (Line 129)

def act(self, decision, environment):
    """Execute decided action in environment."""

Purpose: Execute decided action in environment.

Parameters: – self: Instance reference – decision: Parameter for decision – environment: Parameter for environment

run_loop (Line 158)

def run_loop(self, environment, context):
    """Execute one complete perception-action loop cycle."""

Purpose: Execute one complete perception-action loop cycle.

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

PerceptionActionLoopPattern (Line 194)

Description: Perception-Action Loop Pattern – Continuous sense-think-act cycle.

Features: – Continuous operation – Environment sensing – Real-time processing – Action execution – Feedback integration

Methods:

__init__ (Line 206)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

create_agent (Line 209)

def create_agent(self, name):
    """Create perception-action agent."""

Purpose: Create perception-action agent.

Parameters: – self: Instance reference – name: Identifier or label

run (Line 215)

def run(self, agent_name, environment_sequence, cycles):
    """Run perception-action loops."""

Purpose: Run perception-action loops.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – environment_sequence: Parameter for environment_sequence – cycles: Parameter for cycles

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

    """
Pattern 103: Observer

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 ObserverConfig:
    """Configuration for Observer."""
    model_name: str = "gpt-4"
    temperature: float = 0.7
    max_iterations: int = 10
    enable_logging: bool = True


@dataclass
class ObserverState:
    """State information for Observer."""
    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 ObserverAgent:
    """Agent implementing Observer."""
    
    def __init__(self, name: str, config: ObserverConfig = None):
        self.name = name
        self.config = config or ObserverConfig()
        self.llm = ChatOpenAI(
            model=self.config.model_name,
            temperature=self.config.temperature
        )
        self.state = ObserverState()
        self.execution_history: List[Dict[str, Any]] = []
    
    def process(self, input_data: Any) -> Dict[str, Any]:
        """Process input according to Observer."""
        self.state.current_step += 1
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, implementing Observer. "
                      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 ObserverPattern:
    """
    Observer - Enhanced implementation
    
    This pattern provides comprehensive functionality for observer
    with full LangChain/LangGraph integration.
    
    Features:
    - Multi-agent coordination
    - State management
    - Execution tracking
    - LangGraph workflow integration
    """
    
    def __init__(self, config: ObserverConfig = None):
        self.config = config or ObserverConfig()
        self.agents: Dict[str, ObserverAgent] = {}
        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 Observer. 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: ObserverAgent):
        """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": "Observer",
                "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": "Observer",
            "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("=== Observer Demo ===\n")
    
    # Create configuration
    config = ObserverConfig(
        model_name="gpt-4",
        temperature=0.7,
        enable_logging=True
    )
    
    # Create pattern
    pattern = ObserverPattern(config)
    
    # Create and register agents
    agent1 = ObserverAgent("Agent1", config)
    agent2 = ObserverAgent("Agent2", config)
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    
    print("\nExecuting pattern workflow...")
    
    # Execute with workflow
    result1 = pattern.run("Demonstrate observer 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: Observer

Pattern 103: 103 Observer Pattern – Line-by-Line Explanation

Overview

Pattern 103: Observer Pattern Subscribe to and react to state changes in observed subjects.

Code Structure Analysis

File Information

• Pattern Number: 103
• Pattern Name: 103 Observer Pattern
• File Path: patterns\perception_sensing\103_observer_pattern.py
• Total Lines: 181

Detailed Code Explanation

Module Docstring

"""
Pattern 103: Observer Pattern
Subscribe to and react to state changes in observed subjects.
"""

Explanation: This module docstring describes the purpose and key features of the 103 Observer Pattern pattern.

Imports Section

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

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Protocol: 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 langchain_openai import ChatOpenAI: LangChain framework integration – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration

Classes

StateChange (Line 14)

Description: Represents a state change event.

Observer (Line 23)

Description: Observer interface.

Methods:

update (Line 25)

def update(self, change):
    """Called when observed subject changes."""

Purpose: Called when observed subject changes.

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

Subject (Line 30)

Description: Subject that can be observed.

Methods:

__init__ (Line 33)

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

Purpose: No method docstring.

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

attach (Line 38)

def attach(self, observer):
    """Attach an observer."""

Purpose: Attach an observer.

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

detach (Line 43)

def detach(self, observer):
    """Detach an observer."""

Purpose: Detach an observer.

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

set_state (Line 48)

def set_state(self, property_name, value):
    """Set state and notify observers."""

Purpose: Set state and notify observers.

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

get_state (Line 61)

def get_state(self, property_name):
    """Get state property."""

Purpose: Get state property.

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

_notify (Line 65)

def _notify(self, change):
    """Notify all observers of state change."""

Purpose: Notify all observers of state change.

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

ObserverAgent (Line 71)

Description: Agent that observes subjects and reacts to changes.

Methods:

__init__ (Line 74)

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 (Line 79)

def update(self, change):
    """React to state change."""

Purpose: React to state change.

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

_generate_reaction (Line 87)

def _generate_reaction(self, change):
    """Generate intelligent reaction to state change."""

Purpose: Generate intelligent reaction to state change.

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

get_observation_count (Line 104)

def get_observation_count(self):
    """Get number of observations made."""

Purpose: Get number of observations made.

Parameters: – self: Instance reference

ObserverPattern (Line 109)

Description: Observer Pattern – Subscribe and react to state changes.

Methods:

__init__ (Line 112)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

create_subject (Line 116)

def create_subject(self, subject_id):
    """Create an observable subject."""

Purpose: Create an observable subject.

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

create_observer (Line 122)

def create_observer(self, name):
    """Create an observer agent."""

Purpose: Create an observer agent.

Parameters: – self: Instance reference – name: Identifier or label

run (Line 128)

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

Purpose: Execute observer pattern 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 = StateChange(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

    """
Pattern 102: Monitor Agent Pattern

Agent that continuously watches specific conditions and triggers actions based on rules.
Implements reactive monitoring with LangChain-based rule evaluation.
"""

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


class MonitorStatus(Enum):
    """Status of monitor."""
    HEALTHY = "healthy"
    WARNING = "warning"
    CRITICAL = "critical"
    UNKNOWN = "unknown"


@dataclass
class MonitorRule:
    """Rule for monitoring."""
    rule_id: str
    name: str
    condition: Callable[[Any], bool]
    action: Callable[[Any], None]
    priority: int = 1  # 1=low, 5=critical
    enabled: bool = True
    trigger_count: int = 0
    last_triggered: Optional[datetime] = None


@dataclass
class MonitorEvent:
    """Event detected by monitor."""
    event_id: str
    monitor_id: str
    rule_id: str
    status: MonitorStatus
    data: Any
    message: str
    timestamp: datetime = field(default_factory=datetime.now)


class MonitorAgent:
    """
    Agent that monitors a specific target (system, service, metric) and takes action
    when conditions are met.
    """
    
    def __init__(
        self,
        monitor_id: str,
        name: str,
        check_function: Callable[[], Any],
        check_interval: int = 5,
        model_name: str = "gpt-4"
    ):
        self.monitor_id = monitor_id
        self.name = name
        self.check_function = check_function
        self.check_interval = check_interval
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        
        self.rules: Dict[str, MonitorRule] = {}
        self.events: List[MonitorEvent] = []
        self.status = MonitorStatus.UNKNOWN
        self.is_monitoring = False
        self._monitor_thread: Optional[threading.Thread] = None
        
    def add_rule(self, rule: MonitorRule):
        """Add a monitoring rule."""
        self.rules[rule.rule_id] = rule
    
    def remove_rule(self, rule_id: str):
        """Remove a monitoring rule."""
        if rule_id in self.rules:
            del self.rules[rule_id]
    
    def check(self) -> Dict[str, Any]:
        """Perform a single check."""
        data = self.check_function()
        
        # Evaluate all rules
        triggered_rules = []
        for rule in self.rules.values():
            if rule.enabled and rule.condition(data):
                rule.trigger_count += 1
                rule.last_triggered = datetime.now()
                triggered_rules.append(rule)
                
                # Execute action
                try:
                    rule.action(data)
                except Exception as e:
                    print(f"Error executing action for rule {rule.rule_id}: {e}")
        
        # Determine status using LLM
        if triggered_rules:
            status = self._determine_status(data, triggered_rules)
            message = f"{len(triggered_rules)} rule(s) triggered"
            
            # Create event
            for rule in triggered_rules:
                event = MonitorEvent(
                    event_id=f"evt-{len(self.events)+1}",
                    monitor_id=self.monitor_id,
                    rule_id=rule.rule_id,
                    status=status,
                    data=data,
                    message=f"{rule.name}: {message}"
                )
                self.events.append(event)
        else:
            status = MonitorStatus.HEALTHY
            message = "All checks passed"
        
        self.status = status
        
        return {
            "monitor_id": self.monitor_id,
            "status": status.value,
            "data": data,
            "triggered_rules": len(triggered_rules),
            "message": message
        }
    
    def _determine_status(self, data: Any, triggered_rules: List[MonitorRule]) -> MonitorStatus:
        """Determine overall status based on triggered rules."""
        max_priority = max(rule.priority for rule in triggered_rules)
        
        if max_priority >= 4:
            return MonitorStatus.CRITICAL
        elif max_priority >= 2:
            return MonitorStatus.WARNING
        else:
            return MonitorStatus.HEALTHY
    
    def start_monitoring(self):
        """Start continuous monitoring in background thread."""
        if self.is_monitoring:
            return
        
        self.is_monitoring = True
        self._monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
        self._monitor_thread.start()
    
    def stop_monitoring(self):
        """Stop continuous monitoring."""
        self.is_monitoring = False
        if self._monitor_thread:
            self._monitor_thread.join(timeout=self.check_interval + 1)
    
    def _monitor_loop(self):
        """Background monitoring loop."""
        while self.is_monitoring:
            self.check()
            time.sleep(self.check_interval)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get monitoring statistics."""
        return {
            "monitor_id": self.monitor_id,
            "name": self.name,
            "status": self.status.value,
            "total_events": len(self.events),
            "critical_events": len([e for e in self.events if e.status == MonitorStatus.CRITICAL]),
            "warning_events": len([e for e in self.events if e.status == MonitorStatus.WARNING]),
            "rules": len(self.rules),
            "active_rules": len([r for r in self.rules.values() if r.enabled])
        }


class MonitorAgentPattern:
    """
    Coordinates multiple monitor agents for comprehensive system monitoring.
    Provides centralized alerting and dashboard capabilities.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.monitors: Dict[str, MonitorAgent] = {}
        
    def register_monitor(self, monitor: MonitorAgent):
        """Register a monitor agent."""
        self.monitors[monitor.monitor_id] = monitor
    
    def check_all(self) -> Dict[str, Dict[str, Any]]:
        """Check all monitors."""
        results = {}
        for monitor_id, monitor in self.monitors.items():
            results[monitor_id] = monitor.check()
        return results
    
    def start_all(self):
        """Start all monitors."""
        for monitor in self.monitors.values():
            monitor.start_monitoring()
    
    def stop_all(self):
        """Stop all monitors."""
        for monitor in self.monitors.values():
            monitor.stop_monitoring()
    
    def get_system_status(self) -> str:
        """Get overall system status using LLM analysis."""
        stats = {mid: m.get_statistics() for mid, m in self.monitors.items()}
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a system monitoring AI. Analyze monitor data and provide status."),
            ("human", """Monitor Statistics:
{stats}

Provide:
1. Overall system health (HEALTHY/WARNING/CRITICAL)
2. Summary of issues
3. Recommended actions""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "stats": "\n".join([f"{k}: {v}" for k, v in stats.items()])
        })
        
        return str(response.content)
    
    def get_recent_events(self, limit: int = 10) -> List[MonitorEvent]:
        """Get recent events from all monitors."""
        all_events = []
        for monitor in self.monitors.values():
            all_events.extend(monitor.events)
        
        return sorted(all_events, key=lambda e: e.timestamp, reverse=True)[:limit]


def main():
    """Example usage of Monitor Agent Pattern."""
    print("=" * 80)
    print("Monitor Agent Pattern - System Monitoring Demo")
    print("=" * 80)
    
    # Create monitoring pattern
    pattern = MonitorAgentPattern()
    
    # Simulate monitored systems
    import random
    
    cpu_usage = [20.0]
    memory_usage = [50.0]
    
    def check_cpu():
        cpu_usage[0] += random.uniform(-5, 10)
        cpu_usage[0] = max(0, min(100, cpu_usage[0]))
        return {"cpu": cpu_usage[0], "cores": 4}
    
    def check_memory():
        memory_usage[0] += random.uniform(-3, 8)
        memory_usage[0] = max(0, min(100, memory_usage[0]))
        return {"memory": memory_usage[0], "total_gb": 16}
    
    def check_disk():
        return {"used_gb": random.randint(100, 200), "total_gb": 500}
    
    # Create CPU monitor
    cpu_monitor = MonitorAgent(
        monitor_id="cpu-monitor",
        name="CPU Monitor",
        check_function=check_cpu,
        check_interval=2
    )
    
    cpu_monitor.add_rule(MonitorRule(
        rule_id="cpu-high",
        name="High CPU Usage",
        condition=lambda data: data["cpu"] > 70,
        action=lambda data: print(f"⚠️  CPU at {data['cpu']:.1f}%"),
        priority=3
    ))
    
    cpu_monitor.add_rule(MonitorRule(
        rule_id="cpu-critical",
        name="Critical CPU Usage",
        condition=lambda data: data["cpu"] > 90,
        action=lambda data: print(f"🚨 CRITICAL: CPU at {data['cpu']:.1f}%"),
        priority=5
    ))
    
    # Create Memory monitor
    memory_monitor = MonitorAgent(
        monitor_id="mem-monitor",
        name="Memory Monitor",
        check_function=check_memory,
        check_interval=2
    )
    
    memory_monitor.add_rule(MonitorRule(
        rule_id="mem-high",
        name="High Memory Usage",
        condition=lambda data: data["memory"] > 75,
        action=lambda data: print(f"⚠️  Memory at {data['memory']:.1f}%"),
        priority=3
    ))
    
    # Create Disk monitor
    disk_monitor = MonitorAgent(
        monitor_id="disk-monitor",
        name="Disk Monitor",
        check_function=check_disk,
        check_interval=3
    )
    
    disk_monitor.add_rule(MonitorRule(
        rule_id="disk-full",
        name="Disk Almost Full",
        condition=lambda data: (data["used_gb"] / data["total_gb"]) > 0.8,
        action=lambda data: print(f"⚠️  Disk at {(data['used_gb']/data['total_gb'])*100:.1f}%"),
        priority=4
    ))
    
    # Register monitors
    pattern.register_monitor(cpu_monitor)
    pattern.register_monitor(memory_monitor)
    pattern.register_monitor(disk_monitor)
    
    # Perform manual checks
    print("\n📊 Manual Check Results:")
    results = pattern.check_all()
    for monitor_id, result in results.items():
        print(f"\n{monitor_id}:")
        print(f"  Status: {result['status']}")
        print(f"  Data: {result['data']}")
        print(f"  Message: {result['message']}")
    
    # Start continuous monitoring
    print("\n" + "=" * 80)
    print("🔄 Starting continuous monitoring for 10 seconds...")
    print("=" * 80)
    pattern.start_all()
    time.sleep(10)
    pattern.stop_all()
    
    # Show statistics
    print("\n" + "=" * 80)
    print("📈 Monitoring Statistics")
    print("=" * 80)
    for monitor_id, monitor in pattern.monitors.items():
        stats = monitor.get_statistics()
        print(f"\n{monitor_id}:")
        for key, value in stats.items():
            print(f"  {key}: {value}")
    
    # Show recent events
    print("\n" + "=" * 80)
    print("📋 Recent Events")
    print("=" * 80)
    events = pattern.get_recent_events(limit=10)
    for event in events:
        print(f"\n[{event.status.value.upper()}] {event.monitor_id}")
        print(f"  Rule: {event.rule_id}")
        print(f"  Message: {event.message}")
        print(f"  Time: {event.timestamp.strftime('%H:%M:%S')}")
    
    # Get system status
    print("\n" + "=" * 80)
    print("🖥️  System Status Analysis")
    print("=" * 80)
    status = pattern.get_system_status()
    print(status)


if __name__ == "__main__":
    main()

    

Agentic Architecture Pattern: Monitor Agent

Pattern 102: 102 Monitor Agent Pattern – Line-by-Line Explanation

Overview

Pattern 102: Monitor Agent Pattern Continuous monitoring and alerting based on thresholds.

Code Structure Analysis

File Information

• Pattern Number: 102
• Pattern Name: 102 Monitor Agent Pattern
• File Path: patterns\perception_sensing\102_monitor_agent_pattern.py
• Total Lines: 224

Detailed Code Explanation

Module Docstring

"""
Pattern 102: Monitor Agent Pattern
Continuous monitoring and alerting based on thresholds.
"""

Explanation: This module docstring describes the purpose and key features of the 102 Monitor Agent Pattern 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_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate  # 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_openai import ChatOpenAI: LangChain framework integration – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration

Enumerations

AlertLevel (Line 14)

class AlertLevel(Enum):
    INFO
    WARNING
    CRITICAL

Purpose: Alert severity levels.

Values: – INFO: Informational severity level – WARNING: Warning severity level – CRITICAL: Represents critical

Classes

AlertLevel (Line 14)

Description: Alert severity levels.

MonitoringMetric (Line 22)

Description: A metric to monitor.

Methods:

get_alert_level (Line 31)

def get_alert_level(self):
    """Get alert level based on thresholds."""

Purpose: Get alert level based on thresholds.

Parameters: – self: Instance reference

Alert (Line 41)

Description: An alert generated by monitoring.

MonitorAgent (Line 51)

Description: Agent that monitors metrics and generates alerts.

Methods:

__init__ (Line 54)

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_metric (Line 65)

def register_metric(self, metric):
    """Register a metric to monitor."""

Purpose: Register a metric to monitor.

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

update_metric (Line 69)

def update_metric(self, name, value):
    """Update a metric value and check for alerts."""

Purpose: Update a metric value and check for alerts.

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

_generate_alert (Line 86)

def _generate_alert(self, metric, level):
    """Generate an alert with LLM-enhanced message."""

Purpose: Generate an alert with LLM-enhanced message.

Parameters: – self: Instance reference – metric: Parameter for metric – level: Parameter for level

register_handler (Line 112)

def register_handler(self, level, handler):
    """Register a handler for specific alert level."""

Purpose: Register a handler for specific alert level.

Parameters: – self: Instance reference – level: Parameter for level – handler: Parameter for handler

_handle_alert (Line 116)

def _handle_alert(self, alert):
    """Execute handlers for an alert."""

Purpose: Execute handlers for an alert.

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

get_status (Line 121)

def get_status(self):
    """Get monitoring status."""

Purpose: Get monitoring status.

Parameters: – self: Instance reference

MonitorAgentPattern (Line 134)

Description: Monitor Agent Pattern – Continuous monitoring and alerting.

Methods:

__init__ (Line 137)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

create_agent (Line 140)

def create_agent(self, name):
    """Create a monitoring agent."""

Purpose: Create a monitoring agent.

Parameters: – self: Instance reference – name: Identifier or label

run (Line 146)

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

Purpose: Execute monitoring 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 = AlertLevel(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

    """
Pattern 101: Sensor Agent Pattern

Specialized agent that monitors and collects data from various sources (APIs, files, sensors).
Uses LangChain for processing sensor data and LangGraph for sensor orchestration.
"""

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


class SensorType(Enum):
    """Types of sensors."""
    TEMPERATURE = "temperature"
    HUMIDITY = "humidity"
    PRESSURE = "pressure"
    MOTION = "motion"
    LIGHT = "light"
    API = "api"
    FILE = "file"
    DATABASE = "database"


@dataclass
class SensorReading:
    """Represents a sensor reading."""
    sensor_id: str
    sensor_type: SensorType
    value: Any
    unit: str
    timestamp: datetime = field(default_factory=datetime.now)
    metadata: Dict[str, Any] = field(default_factory=dict)
    

@dataclass
class SensorAlert:
    """Alert triggered by sensor threshold violation."""
    sensor_id: str
    reading: SensorReading
    threshold_type: str  # "above", "below", "equal"
    threshold_value: float
    severity: str  # "low", "medium", "high", "critical"
    message: str
    timestamp: datetime = field(default_factory=datetime.now)


class SensorAgent:
    """Agent that monitors a specific sensor and generates alerts."""
    
    def __init__(
        self,
        sensor_id: str,
        sensor_type: SensorType,
        read_function: Callable[[], Any],
        unit: str = "",
        model_name: str = "gpt-4"
    ):
        self.sensor_id = sensor_id
        self.sensor_type = sensor_type
        self.read_function = read_function
        self.unit = unit
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        
        self.readings: List[SensorReading] = []
        self.alerts: List[SensorAlert] = []
        self.thresholds: Dict[str, float] = {}
        
    def read_sensor(self) -> SensorReading:
        """Read from the sensor and create a reading."""
        value = self.read_function()
        reading = SensorReading(
            sensor_id=self.sensor_id,
            sensor_type=self.sensor_type,
            value=value,
            unit=self.unit
        )
        self.readings.append(reading)
        self._check_thresholds(reading)
        return reading
    
    def set_threshold(self, threshold_type: str, value: float):
        """Set a threshold for alerts."""
        self.thresholds[threshold_type] = value
    
    def _check_thresholds(self, reading: SensorReading):
        """Check if reading violates any thresholds."""
        value = float(reading.value)
        
        if "above" in self.thresholds and value > self.thresholds["above"]:
            self._create_alert(reading, "above", self.thresholds["above"])
        
        if "below" in self.thresholds and value  1 else "Threshold violation detected"
        
        alert = SensorAlert(
            sensor_id=self.sensor_id,
            reading=reading,
            threshold_type=threshold_type,
            threshold_value=threshold_value,
            severity=severity,
            message=message
        )
        self.alerts.append(alert)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get statistics about sensor readings."""
        if not self.readings:
            return {"count": 0}
        
        values = [float(r.value) for r in self.readings]
        return {
            "sensor_id": self.sensor_id,
            "sensor_type": self.sensor_type.value,
            "count": len(self.readings),
            "min": min(values),
            "max": max(values),
            "avg": sum(values) / len(values),
            "latest": values[-1],
            "alerts": len(self.alerts)
        }


class SensorAgentPattern:
    """
    Coordinates multiple sensor agents for comprehensive monitoring.
    Implements sensor fusion and multi-sensor analysis.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.sensors: Dict[str, SensorAgent] = {}
        self.monitoring_active = False
        
    def register_sensor(self, sensor: SensorAgent):
        """Register a sensor agent."""
        self.sensors[sensor.sensor_id] = sensor
        
    def read_all_sensors(self) -> Dict[str, SensorReading]:
        """Read from all registered sensors."""
        readings = {}
        for sensor_id, sensor in self.sensors.items():
            readings[sensor_id] = sensor.read_sensor()
        return readings
    
    def get_all_alerts(self) -> List[SensorAlert]:
        """Get all alerts from all sensors."""
        alerts = []
        for sensor in self.sensors.values():
            alerts.extend(sensor.alerts)
        return sorted(alerts, key=lambda a: a.timestamp, reverse=True)
    
    def analyze_environment(self) -> str:
        """Analyze overall environment based on all sensors."""
        stats = {sid: s.get_statistics() for sid, s in self.sensors.items()}
        alerts = self.get_all_alerts()
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are an environmental monitoring AI. Analyze sensor data and provide insights."),
            ("human", """Sensor Statistics:
{stats}

Recent Alerts:
{alerts}

Provide:
1. Overall environmental status
2. Any concerning patterns
3. Recommendations for action""")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "stats": "\n".join([f"{k}: {v}" for k, v in stats.items()]),
            "alerts": "\n".join([f"- {a.severity.upper()}: {a.message}" for a in alerts[:5]])
        })
        
        return str(response.content)
    
    def monitor_continuous(self, duration_seconds: int, interval_seconds: int = 1):
        """Continuously monitor sensors for specified duration."""
        self.monitoring_active = True
        start_time = time.time()
        
        print(f"Starting continuous monitoring for {duration_seconds} seconds...")
        
        while self.monitoring_active and (time.time() - start_time)  (time.time() - interval_seconds):
                print(f"⚠️  ALERT: {new_alerts[0].message}")
            
            time.sleep(interval_seconds)
        
        print("Monitoring stopped.")
    
    def stop_monitoring(self):
        """Stop continuous monitoring."""
        self.monitoring_active = False


def main():
    """Example usage of Sensor Agent Pattern."""
    print("=" * 80)
    print("Sensor Agent Pattern - Environmental Monitoring Demo")
    print("=" * 80)
    
    # Create sensor pattern
    pattern = SensorAgentPattern()
    
    # Simulate sensor functions
    import random
    
    temp_base = 22.0
    humid_base = 60.0
    
    def read_temperature():
        return round(temp_base + random.uniform(-2, 5), 1)
    
    def read_humidity():
        return round(humid_base + random.uniform(-10, 15), 1)
    
    def read_light():
        return random.randint(100, 1000)
    
    # Create sensor agents
    temp_sensor = SensorAgent(
        sensor_id="temp-001",
        sensor_type=SensorType.TEMPERATURE,
        read_function=read_temperature,
        unit="°C"
    )
    temp_sensor.set_threshold("above", 25.0)
    temp_sensor.set_threshold("below", 18.0)
    
    humid_sensor = SensorAgent(
        sensor_id="humid-001",
        sensor_type=SensorType.HUMIDITY,
        read_function=read_humidity,
        unit="%"
    )
    humid_sensor.set_threshold("above", 70.0)
    humid_sensor.set_threshold("below", 40.0)
    
    light_sensor = SensorAgent(
        sensor_id="light-001",
        sensor_type=SensorType.LIGHT,
        read_function=read_light,
        unit="lux"
    )
    light_sensor.set_threshold("below", 200)
    
    # Register sensors
    pattern.register_sensor(temp_sensor)
    pattern.register_sensor(humid_sensor)
    pattern.register_sensor(light_sensor)
    
    # Read sensors multiple times
    print("\n📊 Reading sensors (10 samples)...")
    for i in range(10):
        readings = pattern.read_all_sensors()
        print(f"\nSample {i+1}:")
        for sensor_id, reading in readings.items():
            print(f"  {sensor_id}: {reading.value}{reading.unit}")
        time.sleep(0.1)
    
    # Show statistics
    print("\n" + "=" * 80)
    print("📈 Sensor Statistics")
    print("=" * 80)
    for sensor_id, sensor in pattern.sensors.items():
        stats = sensor.get_statistics()
        print(f"\n{sensor_id}:")
        for key, value in stats.items():
            print(f"  {key}: {value}")
    
    # Show alerts
    print("\n" + "=" * 80)
    print("⚠️  Alerts")
    print("=" * 80)
    alerts = pattern.get_all_alerts()
    if alerts:
        for alert in alerts[:10]:
            print(f"\n[{alert.severity.upper()}] {alert.sensor_id}")
            print(f"  Value: {alert.reading.value}{alert.reading.unit}")
            print(f"  Threshold: {alert.threshold_type} {alert.threshold_value}")
            print(f"  Message: {alert.message}")
    else:
        print("No alerts generated.")
    
    # Environment analysis
    print("\n" + "=" * 80)
    print("🌍 Environment Analysis")
    print("=" * 80)
    analysis = pattern.analyze_environment()
    print(analysis)


if __name__ == "__main__":
    main()

    

Agentic Architecture Pattern: Sensor Agent

Pattern 101: 101 Sensor Agent Pattern – Line-by-Line Explanation

Overview

Pattern 101: Sensor Agent Pattern

Agents that perceive and collect data from environment through sensors. Processes raw sensor data into meaningful information.

Code Structure Analysis

File Information

• Pattern Number: 101
• Pattern Name: 101 Sensor Agent Pattern
• File Path: patterns\perception_sensing\101_sensor_agent_pattern.py
• Total Lines: 213

Detailed Code Explanation

Module Docstring

"""
Pattern 101: Sensor Agent Pattern

Agents that perceive and collect data from environment through sensors.
Processes raw sensor data into meaningful information.
"""

Explanation: This module docstring describes the purpose and key features of the 101 Sensor Agent Pattern pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import random
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: 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

SensorType (Line 17)

class SensorType(Enum):
    TEMPERATURE
    HUMIDITY
    PRESSURE
    LIGHT
    MOTION
    PROXIMITY
    TEXT
    IMAGE

Purpose: Types of sensors.

Values: – TEMPERATURE: Represents temperature – HUMIDITY: Represents humidity – PRESSURE: Represents pressure – LIGHT: Represents light – MOTION: Represents motion – PROXIMITY: Represents proximity – TEXT: Represents text – IMAGE: Represents image

Classes

SensorType (Line 17)

Description: Types of sensors.

SensorReading (Line 30)

Description: Single sensor reading.

SensorData (Line 41)

Description: Processed sensor data.

Sensor (Line 48)

Description: Individual sensor component.

Methods:

__init__ (Line 51)

def __init__(self, sensor_id, sensor_type, unit):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – sensor_id: Parameter for sensor_id – sensor_type: Parameter for sensor_type – unit: Parameter for unit

read (Line 57)

def read(self):
    """Simulate sensor reading."""

Purpose: Simulate sensor reading.

Parameters: – self: Instance reference

SensorAgent (Line 81)

Description: Agent that manages and processes sensors.

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_sensor (Line 89)

def add_sensor(self, sensor):
    """Add a sensor to the agent."""

Purpose: Add a sensor to the agent.

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

read_all_sensors (Line 94)

def read_all_sensors(self):
    """Read all sensors."""

Purpose: Read all sensors.

Parameters: – self: Instance reference

analyze_readings (Line 98)

def analyze_readings(self, readings):
    """Analyze sensor readings using LLM."""

Purpose: Analyze sensor readings using LLM.

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

detect_anomalies (Line 114)

def detect_anomalies(self, readings):
    """Detect anomalies in sensor data."""

Purpose: Detect anomalies in sensor data.

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

SensorAgentPattern (Line 129)

Description: Sensor Agent Pattern – Multi-sensor perception system.

Methods:

__init__ (Line 134)

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

create_agent (Line 138)

def create_agent(self, name):
    """Create a sensor agent."""

Purpose: Create a sensor agent.

Parameters: – self: Instance reference – name: Identifier or label

run (Line 144)

def run(self, agent_name):
    """Execute sensor reading and analysis."""

Purpose: Execute sensor reading and analysis.

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

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