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