Tag: llm

Agentic Architecture Pattern: Environment Model


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

November 4, 2025

Agentic Architecture Pattern: Context Aware


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

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

November 4, 2025

Agentic Architecture Pattern: Situational Awareness


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

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

November 4, 2025

Agentic Architecture Pattern: Perception Action Loop


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

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

November 4, 2025

Agentic Architecture Pattern: Observer


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

November 4, 2025

Agentic Architecture Pattern: Monitor Agent


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

November 4, 2025

Agentic Architecture Pattern: Sensor Agent


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

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

November 4, 2025

Agentic Architecture Pattern: Service Oriented


    """
Pattern 100: Service-Oriented Pattern

Agents expose functionality as services with well-defined interfaces.
Services can be discovered, composed, and orchestrated.
"""

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


class ServiceStatus(Enum):
    """Service status."""
    AVAILABLE = "available"
    BUSY = "busy"
    UNAVAILABLE = "unavailable"
    MAINTENANCE = "maintenance"


@dataclass
class ServiceContract:
    """Service contract defining interface."""
    service_name: str
    version: str
    description: str
    input_schema: Dict[str, str]
    output_schema: Dict[str, str]
    sla: Dict[str, Any] = field(default_factory=dict)  # Service Level Agreement


@dataclass
class ServiceCall:
    """Track service invocations."""
    service_name: str
    caller: str
    input_data: Any
    output_data: Any
    timestamp: datetime = field(default_factory=datetime.now)
    execution_time: float = 0.0
    success: bool = True


class Service:
    """Service with defined contract."""
    
    def __init__(self, contract: ServiceContract, implementation: Callable):
        self.contract = contract
        self.implementation = implementation
        self.status = ServiceStatus.AVAILABLE
        self.call_count = 0
        self.total_execution_time = 0.0
    
    def invoke(self, input_data: Dict[str, Any], caller: str = "unknown") -> Dict[str, Any]:
        """Invoke the service."""
        if self.status != ServiceStatus.AVAILABLE:
            return {
                "success": False,
                "error": f"Service {self.contract.service_name} is {self.status.value}"
            }
        
        try:
            start = datetime.now()
            result = self.implementation(input_data)
            execution_time = (datetime.now() - start).total_seconds()
            
            self.call_count += 1
            self.total_execution_time += execution_time
            
            return {
                "success": True,
                "service": self.contract.service_name,
                "version": self.contract.version,
                "result": result,
                "execution_time": execution_time
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    def get_metrics(self) -> Dict[str, Any]:
        """Get service metrics."""
        avg_time = self.total_execution_time / self.call_count if self.call_count > 0 else 0
        return {
            "service": self.contract.service_name,
            "status": self.status.value,
            "call_count": self.call_count,
            "average_execution_time": f"{avg_time:.3f}s"
        }


class ServiceRegistry:
    """Registry for service discovery."""
    
    def __init__(self):
        self.services: Dict[str, Service] = {}
    
    def register(self, service: Service):
        """Register a service."""
        self.services[service.contract.service_name] = service
        print(f"Service registered: {service.contract.service_name} v{service.contract.version}")
    
    def unregister(self, service_name: str):
        """Unregister a service."""
        if service_name in self.services:
            del self.services[service_name]
            print(f"Service unregistered: {service_name}")
    
    def discover(self, query: str = "") -> List[str]:
        """Discover available services."""
        if not query:
            return list(self.services.keys())
        return [
            name for name in self.services.keys()
            if query.lower() in name.lower()
        ]
    
    def get_service(self, service_name: str) -> Optional[Service]:
        """Get a service by name."""
        return self.services.get(service_name)


class ServiceOrientedPattern:
    """
    Service-Oriented Pattern - SOA for agents.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.registry = ServiceRegistry()
        self.call_history: List[ServiceCall] = []
    
    def create_llm_service(self, service_name: str, prompt_template: str) -> Service:
        """Create an LLM-backed service."""
        
        def implementation(input_data: Dict[str, Any]) -> str:
            prompt = ChatPromptTemplate.from_messages([
                ("system", prompt_template),
                ("human", "{input}")
            ])
            chain = prompt | self.llm
            response = chain.invoke({"input": str(input_data)})
            return str(response.content)
        
        contract = ServiceContract(
            service_name=service_name,
            version="1.0",
            description=f"LLM service: {service_name}",
            input_schema={"input": "any"},
            output_schema={"output": "string"}
        )
        
        return Service(contract, implementation)
    
    def call_service(self, service_name: str, input_data: Dict[str, Any], 
                    caller: str = "client") -> Dict[str, Any]:
        """Call a service."""
        service = self.registry.get_service(service_name)
        
        if not service:
            return {
                "success": False,
                "error": f"Service {service_name} not found"
            }
        
        result = service.invoke(input_data, caller)
        
        # Track call
        call = ServiceCall(
            service_name=service_name,
            caller=caller,
            input_data=input_data,
            output_data=result.get("result"),
            execution_time=result.get("execution_time", 0.0),
            success=result.get("success", False)
        )
        self.call_history.append(call)
        
        return result
    
    def compose_services(self, pipeline: List[tuple], input_data: Dict[str, Any]) -> Dict[str, Any]:
        """Compose multiple services in a pipeline."""
        results = []
        current_input = input_data
        
        for service_name, output_key in pipeline:
            result = self.call_service(service_name, current_input)
            
            if not result.get("success"):
                return {
                    "success": False,
                    "error": f"Service {service_name} failed: {result.get('error')}"
                }
            
            results.append({
                "service": service_name,
                "output": result.get("result")
            })
            
            # Pass output as input to next service
            current_input = {output_key: result.get("result")}
        
        return {
            "success": True,
            "pipeline": [s for s, _ in pipeline],
            "results": results
        }
    
    def run(self, service_name: str, input_data: Dict[str, Any]) -> Dict[str, Any]:
        """Execute service call."""
        result = self.call_service(service_name, input_data)
        
        return {
            "service": service_name,
            "input": input_data,
            "output": result,
            "available_services": self.registry.discover()
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get system statistics."""
        return {
            "total_services": len(self.registry.services),
            "total_calls": len(self.call_history),
            "service_metrics": [
                service.get_metrics()
                for service in self.registry.services.values()
            ]
        }


# Example usage
if __name__ == "__main__":
    # Create pattern
    pattern = ServiceOrientedPattern()
    
    # Create and register services
    text_analyzer = pattern.create_llm_service(
        "text_analyzer",
        "Analyze the text and provide insights about sentiment, tone, and key topics."
    )
    pattern.registry.register(text_analyzer)
    
    summarizer = pattern.create_llm_service(
        "summarizer",
        "Summarize the following text in 2-3 sentences."
    )
    pattern.registry.register(summarizer)
    
    translator = pattern.create_llm_service(
        "translator",
        "Translate the text to Spanish."
    )
    pattern.registry.register(translator)
    
    print("=== Service-Oriented Pattern Demo ===\n")
    
    # Service discovery
    print("Available services:", pattern.registry.discover())
    print()
    
    # Call individual service
    print("1. Text Analysis Service:")
    result1 = pattern.run(
        "text_analyzer",
        {"text": "This product is amazing! I love using it every day."}
    )
    print(f"   Success: {result1['output']['success']}")
    print(f"   Result: {result1['output']['result'][:150]}...\n")
    
    # Call another service
    print("2. Summarization Service:")
    result2 = pattern.run(
        "summarizer",
        {"text": "Artificial intelligence is transforming industries worldwide. From healthcare to finance, AI systems are helping humans make better decisions. Machine learning algorithms can process vast amounts of data quickly."}
    )
    print(f"   Success: {result2['output']['success']}")
    print(f"   Result: {result2['output']['result']}\n")
    
    # Service composition
    print("3. Service Composition (Analyze → Summarize):")
    composition_result = pattern.compose_services(
        pipeline=[
            ("text_analyzer", "analysis"),
            ("summarizer", "summary")
        ],
        input_data={"text": "The future of AI is bright with many opportunities."}
    )
    print(f"   Success: {composition_result['success']}")
    print(f"   Pipeline: {' → '.join(composition_result['pipeline'])}")
    for step in composition_result['results']:
        print(f"   {step['service']}: {step['output'][:100]}...")
    
    # Show statistics
    print("\n\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Total services: {stats['total_services']}")
    print(f"Total calls: {stats['total_calls']}")
    print("\nService Metrics:")
    for metrics in stats['service_metrics']:
        print(f"  {metrics['service']}: {metrics['call_count']} calls, avg time: {metrics['average_execution_time']}")

Agentic Architecture Pattern: Service Oriented

Pattern 100: 100 Service Oriented – Line-by-Line Explanation

Overview

Pattern 100: Service-Oriented Pattern

Agents expose functionality as services with well-defined interfaces. Services can be discovered, composed, and orchestrated.

Code Structure Analysis

File Information

Pattern Number: 100
Pattern Name: 100 Service Oriented
File Path: patterns\tool_resource\100_service_oriented.py
Total Lines: 307

Detailed Code Explanation

Module Docstring

"""
Pattern 100: Service-Oriented Pattern

Agents expose functionality as services with well-defined interfaces.
Services can be discovered, composed, and orchestrated.
"""

Explanation: This module docstring describes the purpose and key features of the 100 Service Oriented pattern.

Imports Section

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

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Callable: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime: Date and time utilities – from enum import Enum: Enumeration support for fixed sets of values – from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Enumerations

ServiceStatus (Line 16)

class ServiceStatus(Enum):
    AVAILABLE
    BUSY
    UNAVAILABLE
    MAINTENANCE

Purpose: Service status.

Values: – AVAILABLE: Represents available – BUSY: Represents busy – UNAVAILABLE: Represents unavailable – MAINTENANCE: Represents maintenance

Classes

ServiceStatus (Line 16)

Description: Service status.

ServiceContract (Line 25)

Description: Service contract defining interface.

ServiceCall (Line 36)

Description: Track service invocations.

Service (Line 47)

Description: Service with defined contract.

Methods:

`init` (Line 50)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – contract: Parameter for contract – implementation: Parameter for implementation

`invoke` (Line 57)

def invoke(self, input_data, caller):
    """Invoke the service."""

Purpose: Invoke the service.

Parameters: – self: Instance reference – input_data: Parameter for input_data – caller: Parameter for caller

`get_metrics` (Line 86)

def get_metrics(self):
    """Get service metrics."""

Purpose: Get service metrics.

Parameters: – self: Instance reference

ServiceRegistry (Line 97)

Description: Registry for service discovery.

Methods:

`init` (Line 100)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`register` (Line 103)

def register(self, service):
    """Register a service."""

Purpose: Register a service.

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

`unregister` (Line 108)

def unregister(self, service_name):
    """Unregister a service."""

Purpose: Unregister a service.

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

`discover` (Line 114)

def discover(self, query):
    """Discover available services."""

Purpose: Discover available services.

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

`get_service` (Line 123)

def get_service(self, service_name):
    """Get a service by name."""

Purpose: Get a service by name.

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

ServiceOrientedPattern (Line 128)

Description: Service-Oriented Pattern – SOA for agents.

Methods:

`init` (Line 133)

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_llm_service` (Line 138)

def create_llm_service(self, service_name, prompt_template):
    """Create an LLM-backed service."""

Purpose: Create an LLM-backed service.

Parameters: – self: Instance reference – service_name: Parameter for service_name – prompt_template: Parameter for prompt_template

`call_service` (Line 160)

def call_service(self, service_name, input_data, caller):
    """Call a service."""

Purpose: Call a service.

Parameters: – self: Instance reference – service_name: Parameter for service_name – input_data: Parameter for input_data – caller: Parameter for caller

`compose_services` (Line 186)

def compose_services(self, pipeline, input_data):
    """Compose multiple services in a pipeline."""

Purpose: Compose multiple services in a pipeline.

Parameters: – self: Instance reference – pipeline: Parameter for pipeline – input_data: Parameter for input_data

`run` (Line 214)

def run(self, service_name, input_data):
    """Execute service call."""

Purpose: Execute service call.

Parameters: – self: Instance reference – service_name: Parameter for service_name – input_data: Parameter for input_data

`get_statistics` (Line 225)

def get_statistics(self):
    """Get system statistics."""

Purpose: Get system statistics.

Parameters: – self: Instance reference

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

November 4, 2025

Agentic Architecture Pattern: Capability Based


    """
Pattern 99: Capability-Based Pattern

Agents declare and discover capabilities dynamically. Tasks are matched to agents
based on their declared capabilities, enabling flexible task distribution.
"""

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


class CapabilityLevel(Enum):
    """Proficiency level for capabilities."""
    BASIC = 1
    INTERMEDIATE = 2
    ADVANCED = 3
    EXPERT = 4


@dataclass
class Capability:
    """Agent capability definition."""
    name: str
    description: str
    level: CapabilityLevel
    parameters: Dict[str, Any] = field(default_factory=dict)
    
    def __hash__(self):
        return hash(self.name)
    
    def __eq__(self, other):
        return self.name == other.name


@dataclass
class Task:
    """Task requiring specific capabilities."""
    id: str
    description: str
    required_capabilities: Set[str]
    priority: int = 1
    assigned_to: Optional[str] = None


class CapableAgent:
    """Agent with declared capabilities."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.capabilities: Dict[str, Capability] = {}
        self.tasks_completed: int = 0
    
    def declare_capability(self, capability: Capability):
        """Declare a capability."""
        self.capabilities[capability.name] = capability
        print(f"{self.name} declared capability: {capability.name} ({capability.level.name})")
    
    def has_capability(self, capability_name: str) -> bool:
        """Check if agent has a capability."""
        return capability_name in self.capabilities
    
    def has_all_capabilities(self, capability_names: Set[str]) -> bool:
        """Check if agent has all required capabilities."""
        return all(self.has_capability(cap) for cap in capability_names)
    
    def get_capability_level(self, capability_name: str) -> Optional[CapabilityLevel]:
        """Get capability level."""
        if capability_name in self.capabilities:
            return self.capabilities[capability_name].level
        return None
    
    def execute_task(self, task: Task) -> str:
        """Execute a task using capabilities."""
        if not self.has_all_capabilities(task.required_capabilities):
            return f"Cannot execute - missing capabilities: {task.required_capabilities - set(self.capabilities.keys())}"
        
        # Use LLM to execute
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} with capabilities: {', '.join(self.capabilities.keys())}"),
            ("human", "Execute this task: {task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task.description})
        self.tasks_completed += 1
        
        return str(response.content)


class CapabilityBasedPattern:
    """
    Capability-Based Pattern - Match tasks to agents by capabilities.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.agents: Dict[str, CapableAgent] = {}
        self.tasks: List[Task] = []
        self.assignments: List[Dict[str, Any]] = []
    
    def register_agent(self, agent: CapableAgent):
        """Register an agent."""
        self.agents[agent.name] = agent
        print(f"Agent registered: {agent.name}")
    
    def find_capable_agents(self, required_capabilities: Set[str]) -> List[CapableAgent]:
        """Find all agents with required capabilities."""
        return [
            agent for agent in self.agents.values()
            if agent.has_all_capabilities(required_capabilities)
        ]
    
    def find_best_agent(self, task: Task) -> Optional[CapableAgent]:
        """Find best agent for task based on capability levels."""
        capable_agents = self.find_capable_agents(task.required_capabilities)
        
        if not capable_agents:
            return None
        
        # Score agents by capability levels
        def score_agent(agent: CapableAgent) -> float:
            return sum(
                agent.get_capability_level(cap).value 
                for cap in task.required_capabilities
            )
        
        return max(capable_agents, key=score_agent)
    
    def assign_task(self, task: Task) -> Dict[str, Any]:
        """Assign task to best agent."""
        agent = self.find_best_agent(task)
        
        if not agent:
            return {
                "success": False,
                "task_id": task.id,
                "message": f"No agent found with capabilities: {task.required_capabilities}"
            }
        
        task.assigned_to = agent.name
        self.tasks.append(task)
        
        # Execute task
        result = agent.execute_task(task)
        
        assignment = {
            "task_id": task.id,
            "agent": agent.name,
            "capabilities_used": list(task.required_capabilities),
            "result": result,
            "timestamp": datetime.now()
        }
        self.assignments.append(assignment)
        
        return {
            "success": True,
            "task_id": task.id,
            "assigned_to": agent.name,
            "result": result
        }
    
    def get_capability_matrix(self) -> Dict[str, List[str]]:
        """Get matrix of agents and their capabilities."""
        return {
            agent_name: list(agent.capabilities.keys())
            for agent_name, agent in self.agents.items()
        }
    
    def run(self, task_description: str, 
            required_capabilities: Set[str]) -> Dict[str, Any]:
        """Execute capability-based task assignment."""
        task = Task(
            id=f"task_{len(self.tasks) + 1}",
            description=task_description,
            required_capabilities=required_capabilities
        )
        
        result = self.assign_task(task)
        
        return {
            "task": task_description,
            "required_capabilities": list(required_capabilities),
            "assignment": result,
            "capability_matrix": self.get_capability_matrix()
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pattern statistics."""
        return {
            "total_agents": len(self.agents),
            "total_tasks": len(self.tasks),
            "successful_assignments": sum(1 for a in self.assignments if a),
            "agent_performance": {
                name: agent.tasks_completed
                for name, agent in self.agents.items()
            }
        }


# Example usage
if __name__ == "__main__":
    # Create pattern
    pattern = CapabilityBasedPattern()
    
    # Create agents with capabilities
    agent1 = CapableAgent("DataScientist")
    agent1.declare_capability(Capability("data_analysis", "Analyze datasets", CapabilityLevel.EXPERT))
    agent1.declare_capability(Capability("visualization", "Create visualizations", CapabilityLevel.ADVANCED))
    agent1.declare_capability(Capability("python", "Python programming", CapabilityLevel.EXPERT))
    
    agent2 = CapableAgent("WebDeveloper")
    agent2.declare_capability(Capability("web_development", "Build web applications", CapabilityLevel.EXPERT))
    agent2.declare_capability(Capability("javascript", "JavaScript programming", CapabilityLevel.ADVANCED))
    agent2.declare_capability(Capability("api_design", "Design REST APIs", CapabilityLevel.INTERMEDIATE))
    
    agent3 = CapableAgent("DevOps")
    agent3.declare_capability(Capability("deployment", "Deploy applications", CapabilityLevel.EXPERT))
    agent3.declare_capability(Capability("monitoring", "Monitor systems", CapabilityLevel.ADVANCED))
    agent3.declare_capability(Capability("python", "Python programming", CapabilityLevel.INTERMEDIATE))
    
    # Register agents
    pattern.register_agent(agent1)
    pattern.register_agent(agent2)
    pattern.register_agent(agent3)
    
    print("\n=== Capability-Based Pattern Demo ===\n")
    
    # Task 1: Data analysis
    print("1. Task: Analyze customer data")
    result1 = pattern.run(
        "Analyze customer purchase patterns and create visualizations",
        {"data_analysis", "visualization"}
    )
    print(f"   Assigned to: {result1['assignment']['assigned_to']}")
    print(f"   Result: {result1['assignment']['result'][:100]}...\n")
    
    # Task 2: Web development
    print("2. Task: Build dashboard API")
    result2 = pattern.run(
        "Create REST API for dashboard application",
        {"web_development", "api_design"}
    )
    print(f"   Assigned to: {result2['assignment']['assigned_to']}")
    print(f"   Result: {result2['assignment']['result'][:100]}...\n")
    
    # Task 3: Multi-capability
    print("3. Task: Deploy monitoring solution")
    result3 = pattern.run(
        "Deploy and monitor the application",
        {"deployment", "monitoring"}
    )
    print(f"   Assigned to: {result3['assignment']['assigned_to']}")
    print(f"   Result: {result3['assignment']['result'][:100]}...\n")
    
    # Show capability matrix
    print("\n=== Capability Matrix ===")
    for agent_name, capabilities in pattern.get_capability_matrix().items():
        print(f"{agent_name}: {', '.join(capabilities)}")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    print(f"Total agents: {stats['total_agents']}")
    print(f"Total tasks: {stats['total_tasks']}")
    print(f"Agent performance: {stats['agent_performance']}")

Agentic Architecture Pattern: Capability Based

Pattern 99: 99 Capability Based – Line-by-Line Explanation

Overview

Pattern 99: Capability-Based Pattern

Agents declare and discover capabilities dynamically. Tasks are matched to agents based on their declared capabilities, enabling flexible task distribution.

Code Structure Analysis

File Information

Pattern Number: 99
Pattern Name: 99 Capability Based
File Path: patterns\tool_resource\99_capability_based.py
Total Lines: 271

Detailed Code Explanation

Module Docstring

"""
Pattern 99: Capability-Based Pattern

Agents declare and discover capabilities dynamically. Tasks are matched to agents
based on their declared capabilities, enabling flexible task distribution.
"""

Explanation: This module docstring describes the purpose and key features of the 99 Capability Based pattern.

Imports Section

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

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Set: 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

CapabilityLevel (Line 16)

class CapabilityLevel(Enum):
    BASIC
    INTERMEDIATE
    ADVANCED
    EXPERT

Purpose: Proficiency level for capabilities.

Values: – BASIC: Represents basic – INTERMEDIATE: Represents intermediate – ADVANCED: Represents advanced – EXPERT: Represents expert

Classes

CapabilityLevel (Line 16)

Description: Proficiency level for capabilities.

Capability (Line 25)

Description: Agent capability definition.

Methods:

`hash` (Line 32)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

`eq` (Line 35)

def __eq__(self, other):
    """No method docstring."""

Purpose: No method docstring.

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

Task (Line 40)

Description: Task requiring specific capabilities.

CapableAgent (Line 49)

Description: Agent with declared capabilities.

Methods:

`init` (Line 52)

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

`declare_capability` (Line 58)

def declare_capability(self, capability):
    """Declare a capability."""

Purpose: Declare a capability.

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

`has_capability` (Line 63)

def has_capability(self, capability_name):
    """Check if agent has a capability."""

Purpose: Check if agent has a capability.

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

`has_all_capabilities` (Line 67)

def has_all_capabilities(self, capability_names):
    """Check if agent has all required capabilities."""

Purpose: Check if agent has all required capabilities.

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

`get_capability_level` (Line 71)

def get_capability_level(self, capability_name):
    """Get capability level."""

Purpose: Get capability level.

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

`execute_task` (Line 77)

def execute_task(self, task):
    """Execute a task using capabilities."""

Purpose: Execute a task using capabilities.

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

CapabilityBasedPattern (Line 95)

Description: Capability-Based Pattern – Match tasks to agents by capabilities.

Methods:

`init` (Line 100)

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

`register_agent` (Line 106)

def register_agent(self, agent):
    """Register an agent."""

Purpose: Register an agent.

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

`find_capable_agents` (Line 111)

def find_capable_agents(self, required_capabilities):
    """Find all agents with required capabilities."""

Purpose: Find all agents with required capabilities.

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

`find_best_agent` (Line 118)

def find_best_agent(self, task):
    """Find best agent for task based on capability levels."""

Purpose: Find best agent for task based on capability levels.

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

`assign_task` (Line 134)

def assign_task(self, task):
    """Assign task to best agent."""

Purpose: Assign task to best agent.

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

`get_capability_matrix` (Line 167)

def get_capability_matrix(self):
    """Get matrix of agents and their capabilities."""

Purpose: Get matrix of agents and their capabilities.

Parameters: – self: Instance reference

`run` (Line 174)

def run(self, task_description, required_capabilities):
    """Execute capability-based task assignment."""

Purpose: Execute capability-based task assignment.

Parameters: – self: Instance reference – task_description: Parameter for task_description – required_capabilities: Parameter for required_capabilities

`get_statistics` (Line 192)

def get_statistics(self):
    """Get pattern statistics."""

Purpose: Get pattern statistics.

Parameters: – self: Instance reference

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

November 4, 2025

Agentic Architecture Pattern: Resource Manager


    """
Pattern 98: Resource Manager Pattern

Centralized management of computational resources (CPU, memory, API quotas, tokens)
across agents with allocation, tracking, and optimization.
"""

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


class ResourceType(Enum):
    """Types of resources to manage."""
    CPU = "cpu"
    MEMORY = "memory"
    API_CALLS = "api_calls"
    TOKENS = "tokens"
    STORAGE = "storage"


@dataclass
class Resource:
    """Resource definition."""
    type: ResourceType
    total_capacity: float
    available: float
    unit: str = ""
    
    def allocate(self, amount: float) -> bool:
        """Allocate resource."""
        if self.available >= amount:
            self.available -= amount
            return True
        return False
    
    def release(self, amount: float):
        """Release resource."""
        self.available = min(self.available + amount, self.total_capacity)
    
    def utilization(self) -> float:
        """Get utilization percentage."""
        return ((self.total_capacity - self.available) / self.total_capacity) * 100


@dataclass
class ResourceAllocation:
    """Track resource allocation."""
    agent_id: str
    resource_type: ResourceType
    amount: float
    timestamp: datetime = field(default_factory=datetime.now)
    released: bool = False


class ResourceManagerPattern:
    """
    Resource Manager Pattern - Centralized resource management.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.5)
        self.resources: Dict[ResourceType, Resource] = {}
        self.allocations: List[ResourceAllocation] = []
        self.agents: Dict[str, Dict[ResourceType, float]] = {}
    
    def add_resource(self, resource_type: ResourceType, capacity: float, unit: str = ""):
        """Add a resource pool."""
        self.resources[resource_type] = Resource(
            type=resource_type,
            total_capacity=capacity,
            available=capacity,
            unit=unit
        )
        print(f"Resource added: {resource_type.value} ({capacity} {unit})")
    
    def request_resources(self, agent_id: str, 
                         requests: Dict[ResourceType, float]) -> Dict[str, Any]:
        """Request resources for an agent."""
        # Check if all resources are available
        can_allocate = all(
            self.resources.get(r_type) and 
            self.resources[r_type].available >= amount
            for r_type, amount in requests.items()
        )
        
        if not can_allocate:
            return {
                "success": False,
                "message": "Insufficient resources",
                "requested": requests,
                "available": {
                    r_type: self.resources[r_type].available 
                    for r_type in requests.keys()
                    if r_type in self.resources
                }
            }
        
        # Allocate all resources
        allocated = {}
        for r_type, amount in requests.items():
            if self.resources[r_type].allocate(amount):
                allocation = ResourceAllocation(
                    agent_id=agent_id,
                    resource_type=r_type,
                    amount=amount
                )
                self.allocations.append(allocation)
                allocated[r_type] = amount
                
                # Track per agent
                if agent_id not in self.agents:
                    self.agents[agent_id] = {}
                self.agents[agent_id][r_type] = self.agents[agent_id].get(r_type, 0) + amount
        
        return {
            "success": True,
            "message": "Resources allocated",
            "agent_id": agent_id,
            "allocated": allocated
        }
    
    def release_resources(self, agent_id: str, 
                         releases: Optional[Dict[ResourceType, float]] = None):
        """Release resources from an agent."""
        if releases is None:
            # Release all resources for agent
            releases = self.agents.get(agent_id, {})
        
        for r_type, amount in releases.items():
            if r_type in self.resources:
                self.resources[r_type].release(amount)
                
                # Update agent tracking
                if agent_id in self.agents and r_type in self.agents[agent_id]:
                    self.agents[agent_id][r_type] -= amount
                    if self.agents[agent_id][r_type]  Dict[str, Any]:
        """Get status of all resources."""
        return {
            r_type.value: {
                "total": res.total_capacity,
                "available": res.available,
                "used": res.total_capacity - res.available,
                "utilization": f"{res.utilization():.1f}%",
                "unit": res.unit
            }
            for r_type, res in self.resources.items()
        }
    
    def get_agent_allocations(self, agent_id: str) -> Dict[str, Any]:
        """Get allocations for specific agent."""
        return {
            "agent_id": agent_id,
            "allocations": self.agents.get(agent_id, {}),
            "total_allocations": sum(self.agents.get(agent_id, {}).values())
        }
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute resource management scenario."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are a resource manager. Analyze the resource usage scenario."),
            ("human", "{scenario}\n\nCurrent status: {status}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "scenario": scenario,
            "status": str(self.get_resource_status())
        })
        
        return {
            "scenario": scenario,
            "analysis": response.content,
            "resource_status": self.get_resource_status()
        }


# Example usage
if __name__ == "__main__":
    # Create resource manager
    manager = ResourceManagerPattern()
    
    # Add resources
    manager.add_resource(ResourceType.CPU, 100.0, "cores")
    manager.add_resource(ResourceType.MEMORY, 16384, "MB")
    manager.add_resource(ResourceType.API_CALLS, 1000, "calls/hour")
    manager.add_resource(ResourceType.TOKENS, 1000000, "tokens")
    
    print("\n=== Resource Manager Pattern Demo ===\n")
    
    # Initial status
    print("Initial Resource Status:")
    for r_type, info in manager.get_resource_status().items():
        print(f"  {r_type}: {info['available']}/{info['total']} {info['unit']} available ({info['utilization']} used)")
    
    # Agent 1 requests resources
    print("\n\n1. Agent 'DataProcessor' requesting resources:")
    result1 = manager.request_resources("DataProcessor", {
        ResourceType.CPU: 20.0,
        ResourceType.MEMORY: 4096,
        ResourceType.TOKENS: 100000
    })
    print(f"   Success: {result1['success']}")
    print(f"   Allocated: {result1['allocated']}")
    
    # Agent 2 requests resources
    print("\n2. Agent 'ModelTrainer' requesting resources:")
    result2 = manager.request_resources("ModelTrainer", {
        ResourceType.CPU: 50.0,
        ResourceType.MEMORY: 8192,
        ResourceType.API_CALLS: 500
    })
    print(f"   Success: {result2['success']}")
    print(f"   Allocated: {result2['allocated']}")
    
    # Show updated status
    print("\n\nUpdated Resource Status:")
    for r_type, info in manager.get_resource_status().items():
        print(f"  {r_type}: {info['available']}/{info['total']} {info['unit']} available ({info['utilization']} used)")
    
    # Show agent allocations
    print("\n\nAgent Allocations:")
    for agent_id in ["DataProcessor", "ModelTrainer"]:
        allocs = manager.get_agent_allocations(agent_id)
        print(f"  {agent_id}: {allocs['allocations']}")
    
    # Release resources
    print("\n\n3. Releasing resources from 'DataProcessor':")
    release_result = manager.release_resources("DataProcessor")
    print(f"   Success: {release_result['success']}")
    print(f"   Released: {release_result['released']}")
    
    # Final status
    print("\n\nFinal Resource Status:")
    for r_type, info in manager.get_resource_status().items():
        print(f"  {r_type}: {info['available']}/{info['total']} {info['unit']} available ({info['utilization']} used)")

Agentic Architecture Pattern: Resource Manager

Pattern 98: 98 Resource Manager – Line-by-Line Explanation

Overview

Pattern 98: Resource Manager Pattern

Centralized management of computational resources (CPU, memory, API quotas, tokens) across agents with allocation, tracking, and optimization.

Code Structure Analysis

File Information

Pattern Number: 98
Pattern Name: 98 Resource Manager
File Path: patterns\tool_resource\98_resource_manager.py
Total Lines: 257

Detailed Code Explanation

Module Docstring

"""
Pattern 98: Resource Manager Pattern

Centralized management of computational resources (CPU, memory, API quotas, tokens)
across agents with allocation, tracking, and optimization.
"""

Explanation: This module docstring describes the purpose and key features of the 98 Resource Manager 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  # First 20 imports shown

Enumerations

ResourceType (Line 16)

class ResourceType(Enum):
    CPU
    MEMORY
    API_CALLS
    TOKENS
    STORAGE

Purpose: Types of resources to manage.

Values: – CPU: Represents cpu – MEMORY: Represents memory – API_CALLS: Represents api calls – TOKENS: Represents tokens – STORAGE: Represents storage

Classes

ResourceType (Line 16)

Description: Types of resources to manage.

Resource (Line 26)

Description: Resource definition.

Methods:

`allocate` (Line 33)

def allocate(self, amount):
    """Allocate resource."""

Purpose: Allocate resource.

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

`release` (Line 40)

def release(self, amount):
    """Release resource."""

Purpose: Release resource.

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

`utilization` (Line 44)

def utilization(self):
    """Get utilization percentage."""

Purpose: Get utilization percentage.

Parameters: – self: Instance reference

ResourceAllocation (Line 50)

Description: Track resource allocation.

ResourceManagerPattern (Line 59)

Description: Resource Manager Pattern – Centralized resource management.

Methods:

`init` (Line 64)

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

`add_resource` (Line 70)

def add_resource(self, resource_type, capacity, unit):
    """Add a resource pool."""

Purpose: Add a resource pool.

Parameters: – self: Instance reference – resource_type: Parameter for resource_type – capacity: Parameter for capacity – unit: Parameter for unit

`request_resources` (Line 80)

def request_resources(self, agent_id, requests):
    """Request resources for an agent."""

Purpose: Request resources for an agent.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – requests: Parameter for requests

`release_resources` (Line 126)

def release_resources(self, agent_id, releases):
    """Release resources from an agent."""

Purpose: Release resources from an agent.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – releases: Parameter for releases

`get_resource_status` (Line 157)

def get_resource_status(self):
    """Get status of all resources."""

Purpose: Get status of all resources.

Parameters: – self: Instance reference

`get_agent_allocations` (Line 170)

def get_agent_allocations(self, agent_id):
    """Get allocations for specific agent."""

Purpose: Get allocations for specific agent.

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

`run` (Line 178)

def run(self, scenario):
    """Execute resource management scenario."""

Purpose: Execute resource management 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 = ResourceType(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

November 4, 2025

Tag: llm

Agentic Architecture Pattern: Environment Model

Agentic Architecture Pattern: Environment Model

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

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Classes

EnvironmentModelConfig (Line 18)

EnvironmentModelState (Line 27)

EnvironmentModelAgent (Line 36)

__init__ (Line 39)

process (Line 49)

get_statistics (Line 75)

EnvironmentModelPattern (Line 85)

__init__ (Line 99)

_build_workflow (Line 105)

register_agent (Line 150)

execute (Line 156)

run (Line 188)

get_statistics (Line 192)

WorkflowState (Line 108)

Design Patterns Used

Usage Example

Key Concepts

Extension Points

Agentic Architecture Pattern: Context Aware

Agentic Architecture Pattern: Context Aware

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

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Classes

ContextAwareConfig (Line 18)

ContextAwareState (Line 27)

ContextAwareAgent (Line 36)

__init__ (Line 39)

process (Line 49)

get_statistics (Line 75)

ContextAwarePattern (Line 85)

__init__ (Line 99)

_build_workflow (Line 105)

register_agent (Line 150)

execute (Line 156)

run (Line 188)

get_statistics (Line 192)

WorkflowState (Line 108)

Design Patterns Used

Usage Example

Key Concepts

Extension Points

Agentic Architecture Pattern: Situational Awareness

Agentic Architecture Pattern: Situational Awareness

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

Overview

Code Structure Analysis

File Information

Detailed Code Explanation

Module Docstring

Imports Section

Classes

SituationalAwarenessConfig (Line 18)

SituationalAwarenessState (Line 27)

SituationalAwarenessAgent (Line 36)

__init__ (Line 39)

process (Line 49)

get_statistics (Line 75)

SituationalAwarenessPattern (Line 85)

__init__ (Line 99)

_build_workflow (Line 105)

register_agent (Line 150)

execute (Line 156)

run (Line 188)

get_statistics (Line 192)

`init` (Line 39)

`process` (Line 49)

`get_statistics` (Line 75)

`init` (Line 99)

`_build_workflow` (Line 105)

`register_agent` (Line 150)

`execute` (Line 156)

`run` (Line 188)

`get_statistics` (Line 192)

`init` (Line 39)

`process` (Line 49)

`get_statistics` (Line 75)

`init` (Line 99)

`_build_workflow` (Line 105)

`register_agent` (Line 150)

`execute` (Line 156)

`run` (Line 188)

`get_statistics` (Line 192)

`init` (Line 39)

`process` (Line 49)

`get_statistics` (Line 75)

`init` (Line 99)

`_build_workflow` (Line 105)

`register_agent` (Line 150)

`execute` (Line 156)

`run` (Line 188)

`get_statistics` (Line 192)

`init` (Line 53)

`perceive` (Line 62)

`process` (Line 83)

`decide` (Line 98)

`act` (Line 129)

`run_loop` (Line 158)

`init` (Line 206)

`create_agent` (Line 209)

`run` (Line 215)

`update` (Line 25)

`init` (Line 33)

`attach` (Line 38)

`detach` (Line 43)

`set_state` (Line 48)

`get_state` (Line 61)

`_notify` (Line 65)

`init` (Line 74)

`update` (Line 79)

`_generate_reaction` (Line 87)

`get_observation_count` (Line 104)

`init` (Line 112)

`create_subject` (Line 116)

`create_observer` (Line 122)

`run` (Line 128)