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