"""
Pattern 169: Sharding

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


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

Pattern 169: 169 Sharding – Line-by-Line Explanation

Overview

Pattern 169: Sharding

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 169
• Pattern Name: 169 Sharding
• File Path: patterns\scaling\169_sharding.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 169: Sharding

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

Explanation: This module docstring describes the purpose and key features of the 169 Sharding 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

ShardingConfig (Line 18)

Description: Configuration for Sharding.

ShardingState (Line 27)

Description: State information for Sharding.

ShardingAgent (Line 36)

Description: Agent implementing Sharding.

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

Purpose: Process input according to Sharding.

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

ShardingPattern (Line 85)

Description: Sharding – Enhanced implementation

This pattern provides comprehensive functionality for sharding 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 = ShardingConfig(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 168: Load Distribution

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


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

Pattern 168: 168 Load Distribution – Line-by-Line Explanation

Overview

Pattern 168: Load Distribution

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 168
• Pattern Name: 168 Load Distribution
• File Path: patterns\scaling\168_load_distribution.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 168: Load Distribution

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

Explanation: This module docstring describes the purpose and key features of the 168 Load Distribution 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

LoadDistributionConfig (Line 18)

Description: Configuration for Load Distribution.

LoadDistributionState (Line 27)

Description: State information for Load Distribution.

LoadDistributionAgent (Line 36)

Description: Agent implementing Load Distribution.

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 Load Distribution."""

Purpose: Process input according to Load Distribution.

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

LoadDistributionPattern (Line 85)

Description: Load Distribution – Enhanced implementation

This pattern provides comprehensive functionality for load distribution 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 = LoadDistributionConfig(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 167: Resource Allocation

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


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

Pattern 167: 167 Resource Allocation – Line-by-Line Explanation

Overview

Pattern 167: Resource Allocation

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 167
• Pattern Name: 167 Resource Allocation
• File Path: patterns\scaling\167_resource_allocation.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 167: Resource Allocation

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

Explanation: This module docstring describes the purpose and key features of the 167 Resource Allocation 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

ResourceAllocationConfig (Line 18)

Description: Configuration for Resource Allocation.

ResourceAllocationState (Line 27)

Description: State information for Resource Allocation.

ResourceAllocationAgent (Line 36)

Description: Agent implementing Resource Allocation.

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 Resource Allocation."""

Purpose: Process input according to Resource Allocation.

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

ResourceAllocationPattern (Line 85)

Description: Resource Allocation – Enhanced implementation

This pattern provides comprehensive functionality for resource allocation 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 = ResourceAllocationConfig(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 166: Agent Lifecycle Management Pattern with LangChain/LangGraph

Purpose: Manage complete lifecycle of agents from creation to termination with LLM insights.
States: Created → Initialized → Active → Suspended → Terminated
"""

from typing import Dict, Any, List, Optional, TypedDict
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import statistics


class LifecycleState(Enum):
    """Agent lifecycle states."""
    CREATED = "created"
    INITIALIZED = "initialized"
    ACTIVE = "active"
    SUSPENDED = "suspended"
    TERMINATED = "terminated"


class TransitionTrigger(Enum):
    """Lifecycle transition triggers."""
    MANUAL = "manual"
    AUTOMATIC = "automatic"
    SCHEDULED = "scheduled"
    ERROR = "error"
    RESOURCE_LIMIT = "resource_limit"


class HealthStatus(Enum):
    """Agent health status."""
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"
    UNKNOWN = "unknown"


class ResourceType(Enum):
    """Resource types for agents."""
    CPU = "cpu"
    MEMORY = "memory"
    NETWORK = "network"
    STORAGE = "storage"


@dataclass
class LifecycleEvent:
    """Event in agent lifecycle."""
    event_id: str
    from_state: LifecycleState
    to_state: LifecycleState
    trigger: TransitionTrigger
    reason: str
    metadata: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ResourceAllocation:
    """Resource allocation for an agent."""
    cpu_cores: float
    memory_mb: int
    network_bandwidth_mbps: int
    storage_gb: int
    allocated_at: datetime = field(default_factory=datetime.now)


@dataclass
class AgentHealth:
    """Agent health metrics."""
    status: HealthStatus
    cpu_usage: float
    memory_usage: float
    error_count: int
    last_active: datetime
    uptime_seconds: float


@dataclass
class ManagedAgent:
    """Agent with comprehensive lifecycle management."""
    id: str
    name: str
    agent_type: str
    state: LifecycleState = LifecycleState.CREATED
    created_at: datetime = field(default_factory=datetime.now)
    events: List[LifecycleEvent] = field(default_factory=list)
    resources: Optional[ResourceAllocation] = None
    health: Optional[AgentHealth] = None
    configuration: Dict[str, Any] = field(default_factory=dict)
    tags: Dict[str, str] = field(default_factory=dict)


@dataclass
class LifecyclePolicy:
    """Policy for lifecycle management."""
    policy_id: str
    auto_suspend_after_idle_seconds: Optional[int] = None
    auto_terminate_after_suspended_seconds: Optional[int] = None
    max_error_count: Optional[int] = None
    health_check_interval_seconds: int = 60


class LifecycleManagementState(TypedDict):
    """State for lifecycle management workflow."""
    agent_id: str
    current_state: str
    target_state: str
    validation_result: bool
    health_check: Dict[str, Any]
    recommendations: List[str]
    transition_approved: bool


class LifecycleManager:
    """
    Advanced lifecycle manager with LangChain/LangGraph integration.
    """
    
    def __init__(self, model_name: str = "gpt-4", temperature: float = 0.3):
        self.llm = ChatOpenAI(model=model_name, temperature=temperature)
        self.agents: Dict[str, ManagedAgent] = {}
        self.agent_llms: Dict[str, ChatOpenAI] = {}
        self.policies: Dict[str, LifecyclePolicy] = {}
        self.workflow = self._build_lifecycle_workflow()
        
        # Statistics
        self.stats = {
            "total_agents_created": 0,
            "total_state_transitions": 0,
            "agents_by_state": {state: 0 for state in LifecycleState},
            "transitions_by_trigger": {trigger: 0 for trigger in TransitionTrigger},
            "avg_agent_lifetime_seconds": 0.0,
            "health_checks_performed": 0
        }
    
    def _build_lifecycle_workflow(self) -> StateGraph:
        """Build LangGraph workflow for lifecycle management."""
        workflow = StateGraph(LifecycleManagementState)
        
        # Add nodes
        workflow.add_node("validate_transition", self._validate_transition_node)
        workflow.add_node("check_health", self._check_health_node)
        workflow.add_node("analyze_impact", self._analyze_impact_node)
        workflow.add_node("approve_transition", self._approve_transition_node)
        workflow.add_node("provide_recommendations", self._provide_recommendations_node)
        
        # Define edges
        workflow.set_entry_point("validate_transition")
        workflow.add_edge("validate_transition", "check_health")
        workflow.add_edge("check_health", "analyze_impact")
        workflow.add_edge("analyze_impact", "approve_transition")
        workflow.add_edge("approve_transition", "provide_recommendations")
        workflow.add_edge("provide_recommendations", END)
        
        return workflow.compile()
    
    def _validate_transition_node(self, state: LifecycleManagementState) -> LifecycleManagementState:
        """Validate if state transition is allowed."""
        agent_id = state["agent_id"]
        current = state["current_state"]
        target = state["target_state"]
        
        # Valid transitions
        valid_transitions = {
            LifecycleState.CREATED.value: [LifecycleState.INITIALIZED.value, LifecycleState.TERMINATED.value],
            LifecycleState.INITIALIZED.value: [LifecycleState.ACTIVE.value, LifecycleState.TERMINATED.value],
            LifecycleState.ACTIVE.value: [LifecycleState.SUSPENDED.value, LifecycleState.TERMINATED.value],
            LifecycleState.SUSPENDED.value: [LifecycleState.ACTIVE.value, LifecycleState.TERMINATED.value],
            LifecycleState.TERMINATED.value: []
        }
        
        is_valid = target in valid_transitions.get(current, [])
        state["validation_result"] = is_valid
        
        return state
    
    def _check_health_node(self, state: LifecycleManagementState) -> LifecycleManagementState:
        """Check agent health."""
        agent_id = state["agent_id"]
        agent = self.agents.get(agent_id)
        
        health_data = {}
        if agent and agent.health:
            health_data = {
                "status": agent.health.status.value,
                "cpu_usage": agent.health.cpu_usage,
                "memory_usage": agent.health.memory_usage,
                "error_count": agent.health.error_count,
                "uptime_seconds": agent.health.uptime_seconds
            }
        else:
            health_data = {
                "status": HealthStatus.UNKNOWN.value,
                "message": "Health data not available"
            }
        
        state["health_check"] = health_data
        self.stats["health_checks_performed"] += 1
        
        return state
    
    def _analyze_impact_node(self, state: LifecycleManagementState) -> LifecycleManagementState:
        """Analyze impact of state transition using LLM."""
        current = state["current_state"]
        target = state["target_state"]
        health = state["health_check"]
        
        # Use LLM to analyze transition impact
        prompt = ChatPromptTemplate.from_template(
            "Analyze the impact of this agent lifecycle transition:\n"
            "Current State: {current_state}\n"
            "Target State: {target_state}\n"
            "Agent Health: {health_data}\n\n"
            "Provide a brief analysis of potential impacts and risks."
        )
        
        chain = prompt | self.llm
        response = chain.invoke({
            "current_state": current,
            "target_state": target,
            "health_data": str(health)
        })
        
        # Store analysis in recommendations for now
        state["recommendations"] = [str(response.content)[:200]]
        
        return state
    
    def _approve_transition_node(self, state: LifecycleManagementState) -> LifecycleManagementState:
        """Approve or reject transition based on validation and health."""
        validation = state["validation_result"]
        health = state["health_check"]
        
        # Approve if validation passes and agent is not critically unhealthy
        approved = validation
        if health.get("status") == HealthStatus.UNHEALTHY.value:
            # Still allow termination even if unhealthy
            if state["target_state"] != LifecycleState.TERMINATED.value:
                approved = False
        
        state["transition_approved"] = approved
        
        return state
    
    def _provide_recommendations_node(self, state: LifecycleManagementState) -> LifecycleManagementState:
        """Provide recommendations for lifecycle management."""
        approved = state["transition_approved"]
        health = state["health_check"]
        
        recommendations = []
        
        if not approved:
            recommendations.append("Transition not approved - check validation rules and agent health")
        
        if health.get("error_count", 0) > 5:
            recommendations.append("High error count detected - consider investigating before proceeding")
        
        if health.get("cpu_usage", 0) > 80:
            recommendations.append("High CPU usage - may need resource scaling")
        
        if health.get("memory_usage", 0) > 80:
            recommendations.append("High memory usage - check for memory leaks")
        
        if recommendations:
            state["recommendations"] = recommendations
        
        return state
    
    def create_agent(
        self,
        agent_id: str,
        name: str,
        agent_type: str = "general",
        configuration: Optional[Dict[str, Any]] = None,
        tags: Optional[Dict[str, str]] = None
    ) -> ManagedAgent:
        """Create a new managed agent."""
        if agent_id in self.agents:
            raise ValueError(f"Agent {agent_id} already exists")
        
        agent = ManagedAgent(
            id=agent_id,
            name=name,
            agent_type=agent_type,
            configuration=configuration or {},
            tags=tags or {}
        )
        
        self.agents[agent_id] = agent
        self.stats["total_agents_created"] += 1
        self.stats["agents_by_state"][LifecycleState.CREATED] += 1
        
        # Record event
        self._record_event(
            agent_id,
            LifecycleState.CREATED,
            LifecycleState.CREATED,
            TransitionTrigger.MANUAL,
            "Agent created"
        )
        
        return agent
    
    def initialize_agent(
        self,
        agent_id: str,
        model: str = "gpt-4",
        resources: Optional[ResourceAllocation] = None
    ) -> bool:
        """Initialize agent with resources."""
        result = self._execute_transition(
            agent_id,
            LifecycleState.INITIALIZED,
            TransitionTrigger.MANUAL,
            "Initialization requested"
        )
        
        if result:
            # Allocate resources
            if resources:
                self.agents[agent_id].resources = resources
            else:
                self.agents[agent_id].resources = ResourceAllocation(
                    cpu_cores=1.0,
                    memory_mb=512,
                    network_bandwidth_mbps=100,
                    storage_gb=10
                )
            
            # Initialize LLM
            self.agent_llms[agent_id] = ChatOpenAI(model=model, temperature=0.7)
            
            # Initialize health
            self.agents[agent_id].health = AgentHealth(
                status=HealthStatus.HEALTHY,
                cpu_usage=0.0,
                memory_usage=0.0,
                error_count=0,
                last_active=datetime.now(),
                uptime_seconds=0.0
            )
        
        return result
    
    def activate_agent(self, agent_id: str) -> bool:
        """Activate agent for work."""
        return self._execute_transition(
            agent_id,
            LifecycleState.ACTIVE,
            TransitionTrigger.MANUAL,
            "Activation requested"
        )
    
    def suspend_agent(
        self,
        agent_id: str,
        reason: str = "Manual suspension",
        trigger: TransitionTrigger = TransitionTrigger.MANUAL
    ) -> bool:
        """Suspend active agent."""
        return self._execute_transition(
            agent_id,
            LifecycleState.SUSPENDED,
            trigger,
            reason
        )
    
    def resume_agent(self, agent_id: str) -> bool:
        """Resume suspended agent."""
        return self._execute_transition(
            agent_id,
            LifecycleState.ACTIVE,
            TransitionTrigger.MANUAL,
            "Resume requested"
        )
    
    def terminate_agent(
        self,
        agent_id: str,
        reason: str = "Normal termination",
        trigger: TransitionTrigger = TransitionTrigger.MANUAL
    ) -> bool:
        """Terminate agent and cleanup resources."""
        result = self._execute_transition(
            agent_id,
            LifecycleState.TERMINATED,
            trigger,
            reason
        )
        
        if result:
            # Cleanup resources
            if agent_id in self.agent_llms:
                del self.agent_llms[agent_id]
            
            # Clear resource allocation
            if self.agents[agent_id].resources:
                self.agents[agent_id].resources = None
        
        return result
    
    def _execute_transition(
        self,
        agent_id: str,
        target_state: LifecycleState,
        trigger: TransitionTrigger,
        reason: str
    ) -> bool:
        """Execute state transition with workflow validation."""
        agent = self.agents.get(agent_id)
        if not agent:
            return False
        
        # Prepare workflow state
        initial_state: LifecycleManagementState = {
            "agent_id": agent_id,
            "current_state": agent.state.value,
            "target_state": target_state.value,
            "validation_result": False,
            "health_check": {},
            "recommendations": [],
            "transition_approved": False
        }
        
        # Run workflow
        result = self.workflow.invoke(initial_state)
        
        # Execute transition if approved
        if result["transition_approved"]:
            self._record_event(
                agent_id,
                agent.state,
                target_state,
                trigger,
                reason
            )
            
            # Update state counts
            self.stats["agents_by_state"][agent.state] -= 1
            agent.state = target_state
            self.stats["agents_by_state"][target_state] += 1
            self.stats["total_state_transitions"] += 1
            self.stats["transitions_by_trigger"][trigger] += 1
            
            return True
        
        return False
    
    def _record_event(
        self,
        agent_id: str,
        from_state: LifecycleState,
        to_state: LifecycleState,
        trigger: TransitionTrigger,
        reason: str,
        metadata: Optional[Dict[str, Any]] = None
    ):
        """Record a lifecycle event."""
        agent = self.agents[agent_id]
        event = LifecycleEvent(
            event_id=f"{agent_id}_event_{len(agent.events)}",
            from_state=from_state,
            to_state=to_state,
            trigger=trigger,
            reason=reason,
            metadata=metadata or {}
        )
        agent.events.append(event)
    
    def update_agent_health(
        self,
        agent_id: str,
        cpu_usage: float,
        memory_usage: float,
        error_count: int = 0
    ):
        """Update agent health metrics."""
        agent = self.agents.get(agent_id)
        if not agent or not agent.health:
            return
        
        agent.health.cpu_usage = cpu_usage
        agent.health.memory_usage = memory_usage
        agent.health.error_count = error_count
        agent.health.last_active = datetime.now()
        
        # Calculate uptime
        agent.health.uptime_seconds = (datetime.now() - agent.created_at).total_seconds()
        
        # Determine health status
        if error_count > 10 or cpu_usage > 95 or memory_usage > 95:
            agent.health.status = HealthStatus.UNHEALTHY
        elif error_count > 5 or cpu_usage > 80 or memory_usage > 80:
            agent.health.status = HealthStatus.DEGRADED
        else:
            agent.health.status = HealthStatus.HEALTHY
    
    def apply_policy(self, agent_id: str, policy: LifecyclePolicy):
        """Apply lifecycle policy to an agent."""
        self.policies[agent_id] = policy
    
    def check_policies(self):
        """Check and enforce lifecycle policies."""
        now = datetime.now()
        
        for agent_id, policy in self.policies.items():
            agent = self.agents.get(agent_id)
            if not agent or not agent.health:
                continue
            
            # Auto-suspend after idle
            if policy.auto_suspend_after_idle_seconds:
                idle_time = (now - agent.health.last_active).total_seconds()
                if (idle_time > policy.auto_suspend_after_idle_seconds and 
                    agent.state == LifecycleState.ACTIVE):
                    self.suspend_agent(
                        agent_id,
                        f"Auto-suspended after {idle_time:.0f}s idle",
                        TransitionTrigger.AUTOMATIC
                    )
            
            # Auto-terminate after suspended
            if policy.auto_terminate_after_suspended_seconds:
                if agent.state == LifecycleState.SUSPENDED:
                    suspended_time = (now - agent.events[-1].timestamp).total_seconds()
                    if suspended_time > policy.auto_terminate_after_suspended_seconds:
                        self.terminate_agent(
                            agent_id,
                            f"Auto-terminated after {suspended_time:.0f}s suspended",
                            TransitionTrigger.AUTOMATIC
                        )
            
            # Terminate on error threshold
            if policy.max_error_count:
                if agent.health.error_count > policy.max_error_count:
                    self.terminate_agent(
                        agent_id,
                        f"Error count exceeded: {agent.health.error_count}",
                        TransitionTrigger.ERROR
                    )
    
    def get_agent_history(self, agent_id: str) -> List[Dict[str, Any]]:
        """Get lifecycle history of agent."""
        agent = self.agents.get(agent_id)
        if not agent:
            return []
        
        return [
            {
                "event_id": e.event_id,
                "from_state": e.from_state.value,
                "to_state": e.to_state.value,
                "trigger": e.trigger.value,
                "reason": e.reason,
                "timestamp": e.timestamp.isoformat()
            }
            for e in agent.events
        ]
    
    def get_agents_by_state(self, state: LifecycleState) -> List[ManagedAgent]:
        """Get all agents in a specific state."""
        return [agent for agent in self.agents.values() if agent.state == state]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get comprehensive lifecycle statistics."""
        # Calculate average lifetime for terminated agents
        terminated_agents = self.get_agents_by_state(LifecycleState.TERMINATED)
        if terminated_agents:
            lifetimes = [
                (agent.events[-1].timestamp - agent.created_at).total_seconds()
                for agent in terminated_agents
                if agent.events
            ]
            self.stats["avg_agent_lifetime_seconds"] = statistics.mean(lifetimes) if lifetimes else 0.0
        
        return {
            "total_agents_created": self.stats["total_agents_created"],
            "total_active_agents": len(self.agents),
            "total_state_transitions": self.stats["total_state_transitions"],
            "agents_by_state": {
                state.value: count
                for state, count in self.stats["agents_by_state"].items()
            },
            "transitions_by_trigger": {
                trigger.value: count
                for trigger, count in self.stats["transitions_by_trigger"].items()
            },
            "avg_agent_lifetime_seconds": round(self.stats["avg_agent_lifetime_seconds"], 2),
            "health_checks_performed": self.stats["health_checks_performed"],
            "policies_applied": len(self.policies)
        }


class AgentLifecycleManagementPattern:
    """
    Comprehensive agent lifecycle management pattern coordinator with LangChain/LangGraph.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.manager = LifecycleManager(model_name=model_name)
    
    def create(
        self,
        agent_id: str,
        name: str,
        agent_type: str = "general"
    ) -> ManagedAgent:
        """Create a new agent."""
        return self.manager.create_agent(agent_id, name, agent_type)
    
    def initialize(self, agent_id: str, model: str = "gpt-4") -> bool:
        """Initialize agent."""
        return self.manager.initialize_agent(agent_id, model)
    
    def activate(self, agent_id: str) -> bool:
        """Activate agent."""
        return self.manager.activate_agent(agent_id)
    
    def suspend(self, agent_id: str, reason: str = "Manual suspension") -> bool:
        """Suspend agent."""
        return self.manager.suspend_agent(agent_id, reason)
    
    def resume(self, agent_id: str) -> bool:
        """Resume agent."""
        return self.manager.resume_agent(agent_id)
    
    def terminate(self, agent_id: str, reason: str = "Normal termination") -> bool:
        """Terminate agent."""
        return self.manager.terminate_agent(agent_id, reason)
    
    def update_health(
        self,
        agent_id: str,
        cpu_usage: float,
        memory_usage: float,
        error_count: int = 0
    ):
        """Update agent health."""
        self.manager.update_agent_health(agent_id, cpu_usage, memory_usage, error_count)
    
    def get_history(self, agent_id: str) -> List[Dict[str, Any]]:
        """Get agent history."""
        return self.manager.get_agent_history(agent_id)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get statistics."""
        return self.manager.get_statistics()



if __name__ == "__main__":
    print("=== Agent Lifecycle Management Pattern with LangChain/LangGraph Demo ===\n")
    
    # Note: This demo shows the structure. In production, you'd need:
    # - export OPENAI_API_KEY="your-key-here"
    # - Actual LLM calls would be made
    
    print("Scenario 1: Complete Lifecycle")
    print("-" * 50)
    manager = AgentLifecycleManagementPattern(model_name="gpt-4")
    
    # Create agents
    agent1 = manager.create("agent-001", "DataProcessor", "data_processing")
    print(f"✓ Created {agent1.name} ({agent1.id})")
    
    agent2 = manager.create("agent-002", "APIWorker", "api_integration")
    print(f"✓ Created {agent2.name} ({agent2.id})")
    
    agent3 = manager.create("agent-003", "MonitorAgent", "monitoring")
    print(f"✓ Created {agent3.name} ({agent3.id})")
    
    # Initialize agents
    print("\nInitializing agents...")
    manager.initialize("agent-001", "gpt-4")
    manager.initialize("agent-002", "gpt-3.5-turbo")
    manager.initialize("agent-003", "gpt-4")
    print("✓ All agents initialized")
    
    # Activate agents
    print("\nActivating agents...")
    manager.activate("agent-001")
    manager.activate("agent-002")
    manager.activate("agent-003")
    print("✓ All agents activated")
    
    print("\nScenario 2: Health Monitoring")
    print("-" * 50)
    
    # Simulate health updates
    health_updates = [
        ("agent-001", 45.5, 60.2, 0),
        ("agent-002", 78.3, 85.1, 3),
        ("agent-003", 92.5, 88.0, 8),
    ]
    
    print("Updating agent health metrics:")
    for agent_id, cpu, mem, errors in health_updates:
        manager.update_health(agent_id, cpu, mem, errors)
        agent = manager.manager.agents[agent_id]
        print(f"  {agent_id}: CPU={cpu}%, MEM={mem}%, Errors={errors}")
        if agent.health:
            print(f"    Health Status: {agent.health.status.value}")
    
    print("\nScenario 3: Suspension and Resumption")
    print("-" * 50)
    
    # Suspend unhealthy agent
    print("Suspending agent with high error count...")
    success = manager.suspend("agent-003", "High error rate detected")
    if success:
        print("✓ agent-003 suspended successfully")
    
    # Resume agent after fix
    print("\nResuming agent after fixes...")
    success = manager.resume("agent-003")
    if success:
        print("✓ agent-003 resumed successfully")
    
    print("\nScenario 4: Lifecycle Policies")
    print("-" * 50)
    
    # Apply auto-management policy
    policy = LifecyclePolicy(
        policy_id="aggressive_cleanup",
        auto_suspend_after_idle_seconds=300,
        auto_terminate_after_suspended_seconds=600,
        max_error_count=10
    )
    
    manager.manager.apply_policy("agent-002", policy)
    print(f"✓ Applied policy '{policy.policy_id}' to agent-002")
    print(f"  - Auto-suspend after {policy.auto_suspend_after_idle_seconds}s idle")
    print(f"  - Auto-terminate after {policy.auto_terminate_after_suspended_seconds}s suspended")
    print(f"  - Max errors: {policy.max_error_count}")
    
    # Check policies (would normally run in background)
    print("\nChecking policy enforcement...")
    manager.manager.check_policies()
    print("✓ Policy check complete")
    
    print("\nScenario 5: Lifecycle History")
    print("-" * 50)
    
    # Get detailed history for agent
    history = manager.get_history("agent-001")
    print(f"Lifecycle history for agent-001:")
    for event in history:
        print(f"  {event['from_state']} → {event['to_state']}")
        print(f"    Trigger: {event['trigger']}")
        print(f"    Reason: {event['reason']}")
        print(f"    Time: {event['timestamp']}")
    
    print("\nScenario 6: Agent Termination")
    print("-" * 50)
    
    # Terminate agents
    terminations = [
        ("agent-001", "Task completed successfully"),
        ("agent-002", "Resource reallocation"),
    ]
    
    print("Terminating agents:")
    for agent_id, reason in terminations:
        success = manager.terminate(agent_id, reason)
        if success:
            print(f"✓ {agent_id} terminated: {reason}")
    
    print("\nScenario 7: State Distribution")
    print("-" * 50)
    
    # Get agents by state
    active_agents = manager.manager.get_agents_by_state(LifecycleState.ACTIVE)
    terminated_agents = manager.manager.get_agents_by_state(LifecycleState.TERMINATED)
    
    print(f"Active agents: {len(active_agents)}")
    for agent in active_agents:
        print(f"  - {agent.id} ({agent.name})")
    
    print(f"\nTerminated agents: {len(terminated_agents)}")
    for agent in terminated_agents:
        print(f"  - {agent.id} ({agent.name})")
    
    print("\nScenario 8: Statistics")
    print("-" * 50)
    
    stats = manager.get_statistics()
    print(f"Total Agents Created: {stats['total_agents_created']}")
    print(f"Total Active Agents: {stats['total_active_agents']}")
    print(f"Total State Transitions: {stats['total_state_transitions']}")
    print(f"Health Checks Performed: {stats['health_checks_performed']}")
    print(f"Policies Applied: {stats['policies_applied']}")
    
    print("\nAgents by State:")
    for state, count in stats['agents_by_state'].items():
        if count > 0:
            print(f"  {state}: {count}")
    
    print("\nTransitions by Trigger:")
    for trigger, count in stats['transitions_by_trigger'].items():
        if count > 0:
            print(f"  {trigger}: {count}")
    
    if stats['avg_agent_lifetime_seconds'] > 0:
        print(f"\nAverage Agent Lifetime: {stats['avg_agent_lifetime_seconds']:.2f}s")
    
    print("\n" + "=" * 50)
    print("Agent lifecycle management pattern demonstration complete!")
    print("=" * 50)

    

Agentic Architecture Pattern: Agent Lifecycle Management

Pattern 166: 166 Agent Lifecycle Management – Line-by-Line Explanation

Overview

Pattern 166: Agent Lifecycle Management Pattern with LangChain/LangGraph

Purpose: Manage complete lifecycle of agents from creation to termination with LLM insights. States: Created → Initialized → Active → Suspended → Terminated

Code Structure Analysis

File Information

• Pattern Number: 166
• Pattern Name: 166 Agent Lifecycle Management
• File Path: patterns\scaling\166_agent_lifecycle_management.py
• Total Lines: 816

Detailed Code Explanation

Module Docstring

"""
Pattern 166: Agent Lifecycle Management Pattern with LangChain/LangGraph

Purpose: Manage complete lifecycle of agents from creation to termination with LLM insights.
States: Created → Initialized → Active → Suspended → Terminated
"""

Explanation: This module docstring describes the purpose and key features of the 166 Agent Lifecycle Management pattern.

Imports Section

from typing import Dict, Any, List, Optional, TypedDict
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
import statistics  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, TypedDict: Type hints for better code documentation – from dataclasses import dataclass, field: Data class decorators for clean data structures – from datetime import datetime, timedelta: 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

Enumerations

LifecycleState (Line 18)

class LifecycleState(Enum):
    CREATED
    INITIALIZED
    ACTIVE
    SUSPENDED
    TERMINATED

Purpose: Agent lifecycle states.

Values: – CREATED: Represents created – INITIALIZED: Represents initialized – ACTIVE: Currently operational – SUSPENDED: Represents suspended – TERMINATED: Represents terminated

TransitionTrigger (Line 27)

class TransitionTrigger(Enum):
    MANUAL
    AUTOMATIC
    SCHEDULED
    ERROR
    RESOURCE_LIMIT

Purpose: Lifecycle transition triggers.

Values: – MANUAL: Represents manual – AUTOMATIC: Represents automatic – SCHEDULED: Represents scheduled – ERROR: Error severity level – RESOURCE_LIMIT: Represents resource limit

HealthStatus (Line 36)

class HealthStatus(Enum):
    HEALTHY
    DEGRADED
    UNHEALTHY
    UNKNOWN

Purpose: Agent health status.

Values: – HEALTHY: Represents healthy – DEGRADED: Represents degraded – UNHEALTHY: Represents unhealthy – UNKNOWN: Represents unknown

ResourceType (Line 44)

class ResourceType(Enum):
    CPU
    MEMORY
    NETWORK
    STORAGE

Purpose: Resource types for agents.

Values: – CPU: Represents cpu – MEMORY: Represents memory – NETWORK: Represents network – STORAGE: Represents storage

Classes

LifecycleState (Line 18)

Description: Agent lifecycle states.

TransitionTrigger (Line 27)

Description: Lifecycle transition triggers.

HealthStatus (Line 36)

Description: Agent health status.

ResourceType (Line 44)

Description: Resource types for agents.

LifecycleEvent (Line 53)

Description: Event in agent lifecycle.

ResourceAllocation (Line 65)

Description: Resource allocation for an agent.

AgentHealth (Line 75)

Description: Agent health metrics.

ManagedAgent (Line 86)

Description: Agent with comprehensive lifecycle management.

LifecyclePolicy (Line 101)

Description: Policy for lifecycle management.

LifecycleManagementState (Line 110)

Description: State for lifecycle management workflow.

LifecycleManager (Line 121)

Description: Advanced lifecycle manager with LangChain/LangGraph integration.

Methods:

__init__ (Line 126)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – model_name: LLM model to use (e.g., “gpt-4”) – temperature: Randomness in LLM responses (0.0-1.0)

_build_lifecycle_workflow (Line 143)

def _build_lifecycle_workflow(self):
    """Build LangGraph workflow for lifecycle management."""

Purpose: Build LangGraph workflow for lifecycle management.

Parameters: – self: Instance reference

_validate_transition_node (Line 164)

def _validate_transition_node(self, state):
    """Validate if state transition is allowed."""

Purpose: Validate if state transition is allowed.

Parameters: – self: Instance reference – state: Current workflow state

_check_health_node (Line 184)

def _check_health_node(self, state):
    """Check agent health."""

Purpose: Check agent health.

Parameters: – self: Instance reference – state: Current workflow state

_analyze_impact_node (Line 209)

def _analyze_impact_node(self, state):
    """Analyze impact of state transition using LLM."""

Purpose: Analyze impact of state transition using LLM.

Parameters: – self: Instance reference – state: Current workflow state

_approve_transition_node (Line 236)

def _approve_transition_node(self, state):
    """Approve or reject transition based on validation and health."""

Purpose: Approve or reject transition based on validation and health.

Parameters: – self: Instance reference – state: Current workflow state

_provide_recommendations_node (Line 252)

def _provide_recommendations_node(self, state):
    """Provide recommendations for lifecycle management."""

Purpose: Provide recommendations for lifecycle management.

Parameters: – self: Instance reference – state: Current workflow state

create_agent (Line 276)

def create_agent(self, agent_id, name, agent_type, configuration, tags):
    """Create a new managed agent."""

Purpose: Create a new managed agent.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – name: Identifier or label – agent_type: Parameter for agent_type – configuration: Parameter for configuration – tags: Parameter for tags

initialize_agent (Line 311)

def initialize_agent(self, agent_id, model, resources):
    """Initialize agent with resources."""

Purpose: Initialize agent with resources.

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

activate_agent (Line 352)

def activate_agent(self, agent_id):
    """Activate agent for work."""

Purpose: Activate agent for work.

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

suspend_agent (Line 361)

def suspend_agent(self, agent_id, reason, trigger):
    """Suspend active agent."""

Purpose: Suspend active agent.

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

resume_agent (Line 375)

def resume_agent(self, agent_id):
    """Resume suspended agent."""

Purpose: Resume suspended agent.

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

terminate_agent (Line 384)

def terminate_agent(self, agent_id, reason, trigger):
    """Terminate agent and cleanup resources."""

Purpose: Terminate agent and cleanup resources.

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

_execute_transition (Line 409)

def _execute_transition(self, agent_id, target_state, trigger, reason):
    """Execute state transition with workflow validation."""

Purpose: Execute state transition with workflow validation.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – target_state: Parameter for target_state – trigger: Parameter for trigger – reason: Parameter for reason

_record_event (Line 456)

def _record_event(self, agent_id, from_state, to_state, trigger, reason, metadata):
    """Record a lifecycle event."""

Purpose: Record a lifecycle event.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – from_state: Parameter for from_state – to_state: Parameter for to_state – trigger: Parameter for trigger – reason: Parameter for reason – metadata: Parameter for metadata

…and 6 more methods

AgentLifecycleManagementPattern (Line 601)

Description: Comprehensive agent lifecycle management pattern coordinator with LangChain/LangGraph.

Methods:

__init__ (Line 606)

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

def create(self, agent_id, name, agent_type):
    """Create a new agent."""

Purpose: Create a new agent.

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

initialize (Line 618)

def initialize(self, agent_id, model):
    """Initialize agent."""

Purpose: Initialize agent.

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

activate (Line 622)

def activate(self, agent_id):
    """Activate agent."""

Purpose: Activate agent.

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

suspend (Line 626)

def suspend(self, agent_id, reason):
    """Suspend agent."""

Purpose: Suspend agent.

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

resume (Line 630)

def resume(self, agent_id):
    """Resume agent."""

Purpose: Resume agent.

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

terminate (Line 634)

def terminate(self, agent_id, reason):
    """Terminate agent."""

Purpose: Terminate agent.

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

update_health (Line 638)

def update_health(self, agent_id, cpu_usage, memory_usage, error_count):
    """Update agent health."""

Purpose: Update agent health.

Parameters: – self: Instance reference – agent_id: Parameter for agent_id – cpu_usage: Parameter for cpu_usage – memory_usage: Parameter for memory_usage – error_count: Parameter for error_count

get_history (Line 648)

def get_history(self, agent_id):
    """Get agent history."""

Purpose: Get agent history.

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

get_statistics (Line 652)

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

Purpose: Get 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 = LifecycleState(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 165: Agent Pool Pattern

Purpose: Maintain a pool of reusable agents to avoid creation/destruction overhead.
Similar to connection pooling in databases.
"""

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


class PooledAgentStatus(Enum):
    """Status of pooled agent."""
    AVAILABLE = "available"
    IN_USE = "in_use"
    RETIRED = "retired"


@dataclass
class PooledAgent:
    """Agent in the pool."""
    id: str
    role: str
    status: PooledAgentStatus = PooledAgentStatus.AVAILABLE
    created_at: datetime = field(default_factory=datetime.now)
    last_used: datetime = field(default_factory=datetime.now)
    tasks_completed: int = 0


class AgentPoolPattern:
    """
    Agent Pool Pattern - Reusable agent instances.
    
    Features:
    - Pre-warmed agent pool
    - Checkout/checkin mechanism
    - Pool size management (min/max)
    - Idle agent cleanup
    - Usage statistics
    """
    
    def __init__(self, min_size: int = 2, max_size: int = 10, model: str = "gpt-4"):
        self.min_size = min_size
        self.max_size = max_size
        self.model = model
        self.pool: Dict[str, PooledAgent] = {}
        self.agent_llms: Dict[str, ChatOpenAI] = {}
        self.checkouts = 0
        self.checkins = 0
        
        # Initialize pool with minimum size
        for i in range(min_size):
            self._create_agent(f"pooled-agent-{i}")
    
    def _create_agent(self, agent_id: str):
        """Internal: Create a new pooled agent."""
        agent = PooledAgent(id=agent_id, role="general")
        self.pool[agent_id] = agent
        self.agent_llms[agent_id] = ChatOpenAI(model=self.model, temperature=0.7)
    
    def checkout(self) -> Optional[str]:
        """Checkout an available agent from pool."""
        # Find available agent
        for agent_id, agent in self.pool.items():
            if agent.status == PooledAgentStatus.AVAILABLE:
                agent.status = PooledAgentStatus.IN_USE
                agent.last_used = datetime.now()
                self.checkouts += 1
                return agent_id
        
        # Try to create new agent if under max size
        if len(self.pool)  bool:
        """Return agent to pool."""
        agent = self.pool.get(agent_id)
        if agent and agent.status == PooledAgentStatus.IN_USE:
            agent.status = PooledAgentStatus.AVAILABLE
            self.checkins += 1
            return True
        return False
    
    def execute_task(self, agent_id: str, task: str) -> str:
        """Execute task with checked-out agent."""
        agent = self.pool.get(agent_id)
        llm = self.agent_llms.get(agent_id)
        
        if not agent or not llm or agent.status != PooledAgentStatus.IN_USE:
            return "Agent not available"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {agent.id} from the agent pool"),
            ("human", "{task}")
        ])
        
        chain = prompt | llm
        response = chain.invoke({"task": task})
        agent.tasks_completed += 1
        
        return str(response.content)
    
    def cleanup_idle(self, max_idle_seconds: float = 600.0):
        """Remove agents that have been idle too long."""
        current_time = datetime.now()
        to_remove = []
        
        for agent_id, agent in self.pool.items():
            if agent.status == PooledAgentStatus.AVAILABLE:
                idle_time = (current_time - agent.last_used).total_seconds()
                if idle_time > max_idle_seconds and len(self.pool) > self.min_size:
                    to_remove.append(agent_id)
        
        for agent_id in to_remove:
            del self.pool[agent_id]
            del self.agent_llms[agent_id]
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get pool statistics."""
        available = sum(1 for a in self.pool.values() if a.status == PooledAgentStatus.AVAILABLE)
        in_use = sum(1 for a in self.pool.values() if a.status == PooledAgentStatus.IN_USE)
        
        return {
            "pool_size": len(self.pool),
            "available": available,
            "in_use": in_use,
            "total_checkouts": self.checkouts,
            "total_checkins": self.checkins,
            "utilization": in_use / len(self.pool) if self.pool else 0
        }


if __name__ == "__main__":
    print("=== Agent Pool Pattern Demo ===\n")
    
    pool = AgentPoolPattern(min_size=2, max_size=5)
    
    print(f"Pool initialized with {len(pool.pool)} agents\n")
    
    # Checkout agents and use them
    print("Checking out agents...")
    agent1 = pool.checkout()
    agent2 = pool.checkout()
    
    if agent1:
        print(f"Checked out: {agent1}")
        result = pool.execute_task(agent1, "Process data")
        print(f"Result: {result[:50]}...")
        pool.checkin(agent1)
        print(f"Checked in: {agent1}")
    
    if agent2:
        print(f"Checked out: {agent2}")
        result = pool.execute_task(agent2, "Generate report")
        print(f"Result: {result[:50]}...")
        pool.checkin(agent2)
        print(f"Checked in: {agent2}")
    
    # Statistics
    print("\n=== Statistics ===")
    stats = pool.get_statistics()
    for key, value in stats.items():
        print(f"{key}: {value}")

    

Agentic Architecture Pattern: Agent Pool

Pattern 165: 165 Agent Pool – Line-by-Line Explanation

Overview

Pattern 165: Agent Pool Pattern

Purpose: Maintain a pool of reusable agents to avoid creation/destruction overhead. Similar to connection pooling in databases.

Code Structure Analysis

File Information

• Pattern Number: 165
• Pattern Name: 165 Agent Pool
• File Path: patterns\scaling\165_agent_pool.py
• Total Lines: 172

Detailed Code Explanation

Module Docstring

"""
Pattern 165: Agent Pool Pattern

Purpose: Maintain a pool of reusable agents to avoid creation/destruction overhead.
Similar to connection pooling in databases.
"""

Explanation: This module docstring describes the purpose and key features of the 165 Agent Pool 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_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: 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

PooledAgentStatus (Line 16)

class PooledAgentStatus(Enum):
    AVAILABLE
    IN_USE
    RETIRED

Purpose: Status of pooled agent.

Values: – AVAILABLE: Represents available – IN_USE: Represents in use – RETIRED: Represents retired

Classes

PooledAgentStatus (Line 16)

Description: Status of pooled agent.

PooledAgent (Line 24)

Description: Agent in the pool.

AgentPoolPattern (Line 34)

Description: Agent Pool Pattern – Reusable agent instances.

Features: – Pre-warmed agent pool – Checkout/checkin mechanism – Pool size management (min/max) – Idle agent cleanup – Usage statistics

Methods:

__init__ (Line 46)

def __init__(self, min_size, max_size, model):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – min_size: Parameter for min_size – max_size: Parameter for max_size – model: Parameter for model

_create_agent (Line 59)

def _create_agent(self, agent_id):
    """Internal: Create a new pooled agent."""

Purpose: Internal: Create a new pooled agent.

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

checkout (Line 65)

def checkout(self):
    """Checkout an available agent from pool."""

Purpose: Checkout an available agent from pool.

Parameters: – self: Instance reference

checkin (Line 83)

def checkin(self, agent_id):
    """Return agent to pool."""

Purpose: Return agent to pool.

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

execute_task (Line 92)

def execute_task(self, agent_id, task):
    """Execute task with checked-out agent."""

Purpose: Execute task with checked-out agent.

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

cleanup_idle (Line 111)

def cleanup_idle(self, max_idle_seconds):
    """Remove agents that have been idle too long."""

Purpose: Remove agents that have been idle too long.

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

get_statistics (Line 126)

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

Purpose: Get pool 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 = PooledAgentStatus(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 164: Dynamic Agent Creation Pattern

Purpose: Create and destroy agents on-the-fly based on workload requirements.
"""

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


@dataclass
class AgentTemplate:
    """Template for creating new agents."""
    role: str
    capabilities: List[str]
    model: str = "gpt-4"
    
@dataclass
class DynamicAgent:
    """Dynamically created agent."""
    id: str
    template: AgentTemplate
    created_at: datetime = field(default_factory=datetime.now)
    tasks_completed: int = 0
    lifetime_seconds: float = 0.0


class DynamicAgentCreationPattern:
    """
    Dynamic Agent Creation - Create agents on-demand.
    
    Features:
    - Template-based agent creation
    - Automatic cleanup of idle agents
    - Capability-based agent selection
    - Lifecycle tracking
    """
    
    def __init__(self):
        self.templates: Dict[str, AgentTemplate] = {}
        self.active_agents: Dict[str, DynamicAgent] = {}
        self.agent_llms: Dict[str, ChatOpenAI] = {}
        self.creation_count = 0
        self.destruction_count = 0
        
    def register_template(self, name: str, template: AgentTemplate):
        """Register an agent template."""
        self.templates[name] = template
    
    def create_agent(self, template_name: str) -> Optional[str]:
        """Create new agent from template."""
        template = self.templates.get(template_name)
        if not template:
            return None
        
        agent_id = f"{template.role}-{uuid.uuid4().hex[:8]}"
        agent = DynamicAgent(id=agent_id, template=template)
        
        self.active_agents[agent_id] = agent
        self.agent_llms[agent_id] = ChatOpenAI(model=template.model, temperature=0.7)
        self.creation_count += 1
        
        print(f"✨ Created agent: {agent_id} ({template.role})")
        return agent_id
    
    def destroy_agent(self, agent_id: str) -> bool:
        """Destroy an agent."""
        if agent_id in self.active_agents:
            agent = self.active_agents[agent_id]
            lifetime = (datetime.now() - agent.created_at).total_seconds()
            agent.lifetime_seconds = lifetime
            
            del self.active_agents[agent_id]
            del self.agent_llms[agent_id]
            self.destruction_count += 1
            
            print(f"💥 Destroyed agent: {agent_id} (lived {lifetime:.1f}s, completed {agent.tasks_completed} tasks)")
            return True
        return False
    
    def find_agent_for_task(self, required_capability: str) -> Optional[str]:
        """Find or create agent with required capability."""
        # Try to find existing agent
        for agent_id, agent in self.active_agents.items():
            if required_capability in agent.template.capabilities:
                return agent_id
        
        # Create new agent if needed
        for template_name, template in self.templates.items():
            if required_capability in template.capabilities:
                return self.create_agent(template_name)
        
        return None
    
    def process_task(self, agent_id: str, task: str) -> str:
        """Process task with agent."""
        agent = self.active_agents.get(agent_id)
        llm = self.agent_llms.get(agent_id)
        
        if not agent or not llm:
            return "Agent not found"
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {agent.id} with role {agent.template.role}"),
            ("human", "{task}")
        ])
        
        chain = prompt | llm
        response = chain.invoke({"task": task})
        agent.tasks_completed += 1
        
        return str(response.content)
    
    def cleanup_idle_agents(self, max_lifetime: float = 300.0):
        """Clean up agents that have been idle too long."""
        current_time = datetime.now()
        to_destroy = []
        
        for agent_id, agent in self.active_agents.items():
            lifetime = (current_time - agent.created_at).total_seconds()
            if lifetime > max_lifetime and agent.tasks_completed == 0:
                to_destroy.append(agent_id)
        
        for agent_id in to_destroy:
            self.destroy_agent(agent_id)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get creation statistics."""
        return {
            "active_agents": len(self.active_agents),
            "total_created": self.creation_count,
            "total_destroyed": self.destruction_count,
            "templates": len(self.templates)
        }


if __name__ == "__main__":
    print("=== Dynamic Agent Creation Pattern Demo ===\n")
    
    pattern = DynamicAgentCreationPattern()
    
    # Register templates
    pattern.register_template("analyst", AgentTemplate(
        role="analyst",
        capabilities=["analyze", "report"],
        model="gpt-4"
    ))
    pattern.register_template("writer", AgentTemplate(
        role="writer",
        capabilities=["write", "edit"],
        model="gpt-3.5-turbo"
    ))
    
    print("Registered 2 templates\n")
    
    # Process tasks - create agents dynamically
    tasks = [
        ("analyze", "Analyze this dataset"),
        ("write", "Write a summary"),
        ("analyze", "Create report")
    ]
    
    for capability, task in tasks:
        agent_id = pattern.find_agent_for_task(capability)
        if agent_id:
            result = pattern.process_task(agent_id, task)
            print(f"Task '{task}' processed by {agent_id}")
    
    # Show statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    for key, value in stats.items():
        print(f"{key}: {value}")

    

Agentic Architecture Pattern: Dynamic Agent Creation

Pattern 164: 164 Dynamic Agent Creation – Line-by-Line Explanation

Overview

Pattern 164: Dynamic Agent Creation Pattern

Purpose: Create and destroy agents on-the-fly based on workload requirements.

Code Structure Analysis

File Information

• Pattern Number: 164
• Pattern Name: 164 Dynamic Agent Creation
• File Path: patterns\scaling\164_dynamic_agent_creation.py
• Total Lines: 178

Detailed Code Explanation

Module Docstring

"""
Pattern 164: Dynamic Agent Creation Pattern

Purpose: Create and destroy agents on-the-fly based on workload requirements.
"""

Explanation: This module docstring describes the purpose and key features of the 164 Dynamic Agent Creation pattern.

Imports Section

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

AgentTemplate (Line 16)

Description: Template for creating new agents.

DynamicAgent (Line 23)

Description: Dynamically created agent.

DynamicAgentCreationPattern (Line 32)

Description: Dynamic Agent Creation – Create agents on-demand.

Features: – Template-based agent creation – Automatic cleanup of idle agents – Capability-based agent selection – Lifecycle tracking

Methods:

__init__ (Line 43)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_template (Line 50)

def register_template(self, name, template):
    """Register an agent template."""

Purpose: Register an agent template.

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

create_agent (Line 54)

def create_agent(self, template_name):
    """Create new agent from template."""

Purpose: Create new agent from template.

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

destroy_agent (Line 70)

def destroy_agent(self, agent_id):
    """Destroy an agent."""

Purpose: Destroy an agent.

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

find_agent_for_task (Line 85)

def find_agent_for_task(self, required_capability):
    """Find or create agent with required capability."""

Purpose: Find or create agent with required capability.

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

process_task (Line 99)

def process_task(self, agent_id, task):
    """Process task with agent."""

Purpose: Process task with agent.

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

cleanup_idle_agents (Line 118)

def cleanup_idle_agents(self, max_lifetime):
    """Clean up agents that have been idle too long."""

Purpose: Clean up agents that have been idle too long.

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

get_statistics (Line 131)

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

Purpose: Get creation 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 = AgentTemplate(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 163: Elastic Agent Pattern

Purpose: Dynamically scale resources both vertically and horizontally based on demand.
Combines both scaling strategies for optimal resource utilization.
"""

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


class ScalingDirection(Enum):
    """Direction of scaling."""
    SCALE_UP_VERTICAL = "scale_up_vertical"
    SCALE_DOWN_VERTICAL = "scale_down_vertical"
    SCALE_OUT_HORIZONTAL = "scale_out_horizontal"
    SCALE_IN_HORIZONTAL = "scale_in_horizontal"
    NO_CHANGE = "no_change"


@dataclass
class ResourceMetrics:
    """Current resource utilization metrics."""
    cpu_usage: float  # 0.0 to 1.0
    memory_usage: float  # 0.0 to 1.0
    throughput: float  # tasks per second
    response_time: float  # seconds
    queue_length: int
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ElasticPolicy:
    """Elasticity policy configuration."""
    cpu_scale_up_threshold: float = 0.8
    cpu_scale_down_threshold: float = 0.3
    queue_scale_out_threshold: int = 5
    queue_scale_in_threshold: int = 0
    cooldown_period: float = 60.0  # seconds
    min_instances: int = 1
    max_instances: int = 10


@dataclass
class ScalingDecision:
    """Record of scaling decision."""
    direction: ScalingDirection
    reason: str
    metrics: ResourceMetrics
    timestamp: datetime = field(default_factory=datetime.now)


class ElasticAgent:
    """Agent with elastic scaling capabilities."""
    
    def __init__(self, agent_id: str, tier: str = "standard"):
        self.agent_id = agent_id
        self.tier = tier
        self.llm = ChatOpenAI(model="gpt-4" if tier == "premium" else "gpt-3.5-turbo", 
                             temperature=0.7)
        self.tasks_processed = 0
        self.is_active = True
    
    def process_task(self, task: str) -> str:
        """Process a task."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are elastic agent {self.agent_id} (tier: {self.tier})"),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        self.tasks_processed += 1
        
        return str(response.content)


class ElasticAgentPattern:
    """
    Elastic Agent Pattern - Dynamic bidirectional scaling.
    
    Features:
    - Vertical scaling: Upgrade/downgrade agent tiers
    - Horizontal scaling: Add/remove agent instances
    - Policy-based decisions
    - Cooldown periods to prevent thrashing
    - Metrics tracking
    """
    
    def __init__(self, policy: Optional[ElasticPolicy] = None):
        self.policy = policy or ElasticPolicy()
        self.agents: List[ElasticAgent] = []
        self.task_queue: List[str] = []
        self.metrics_history: List[ResourceMetrics] = []
        self.scaling_decisions: List[ScalingDecision] = []
        self.last_scaling_time = datetime.now()
        
        # Initialize with minimum instances
        for i in range(self.policy.min_instances):
            self.agents.append(ElasticAgent(f"agent-{i}"))
    
    def collect_metrics(self) -> ResourceMetrics:
        """Collect current resource metrics."""
        active_agents = [a for a in self.agents if a.is_active]
        
        # Simulate metrics (in real system, would measure actual resources)
        metrics = ResourceMetrics(
            cpu_usage=len(self.task_queue) / (len(active_agents) * 10 + 1),
            memory_usage=len(active_agents) / self.policy.max_instances,
            throughput=sum(a.tasks_processed for a in active_agents) / max(1, len(self.metrics_history)),
            response_time=1.0 + (len(self.task_queue) * 0.1),
            queue_length=len(self.task_queue)
        )
        
        self.metrics_history.append(metrics)
        return metrics
    
    def make_scaling_decision(self, metrics: ResourceMetrics) -> ScalingDecision:
        """Decide on scaling action based on metrics."""
        # Check cooldown period
        time_since_last = (datetime.now() - self.last_scaling_time).total_seconds()
        if time_since_last  self.policy.queue_scale_out_threshold:
            if len(self.agents)  self.policy.cpu_scale_up_threshold:
            standard_agents = [a for a in self.agents if a.tier == "standard"]
            if standard_agents:
                return ScalingDecision(
                    ScalingDirection.SCALE_UP_VERTICAL,
                    f"CPU usage {metrics.cpu_usage:.2f} exceeds threshold",
                    metrics
                )
        
        # Check for horizontal scale in (remove instances)
        if metrics.queue_length  self.policy.min_instances:
                return ScalingDecision(
                    ScalingDirection.SCALE_IN_HORIZONTAL,
                    f"Queue empty and excess capacity",
                    metrics
                )
        
        # Check for vertical scale down (downgrade tiers)
        if metrics.cpu_usage  self.policy.min_instances:
                removed = self.agents.pop()
                removed.is_active = False
                print(f"➖ Scaled in: Removed {removed.agent_id} (Total: {len(self.agents)})")
        
        elif decision.direction == ScalingDirection.SCALE_UP_VERTICAL:
            # Upgrade an agent to premium tier
            for agent in self.agents:
                if agent.tier == "standard":
                    agent.tier = "premium"
                    agent.llm = ChatOpenAI(model="gpt-4", temperature=0.7)
                    print(f"⬆️ Scaled up: Upgraded {agent.agent_id} to premium")
                    break
        
        elif decision.direction == ScalingDirection.SCALE_DOWN_VERTICAL:
            # Downgrade an agent to standard tier
            for agent in self.agents:
                if agent.tier == "premium":
                    agent.tier = "standard"
                    agent.llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)
                    print(f"⬇️ Scaled down: Downgraded {agent.agent_id} to standard")
                    break
    
    def add_tasks(self, tasks: List[str]):
        """Add tasks to queue."""
        self.task_queue.extend(tasks)
    
    def process_tasks(self):
        """Process tasks with current agent pool."""
        while self.task_queue and any(a.is_active for a in self.agents):
            # Get available agent
            for agent in self.agents:
                if agent.is_active and self.task_queue:
                    task = self.task_queue.pop(0)
                    agent.process_task(task)
            
            # Check if we should scale
            metrics = self.collect_metrics()
            decision = self.make_scaling_decision(metrics)
            self.execute_scaling(decision)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get elasticity statistics."""
        premium_count = sum(1 for a in self.agents if a.tier == "premium")
        standard_count = len(self.agents) - premium_count
        
        return {
            "total_agents": len(self.agents),
            "premium_agents": premium_count,
            "standard_agents": standard_count,
            "tasks_in_queue": len(self.task_queue),
            "total_scaling_actions": len(self.scaling_decisions),
            "metrics_collected": len(self.metrics_history)
        }


# Example usage
if __name__ == "__main__":
    print("=== Elastic Agent Pattern Demo ===\n")
    
    # Create elastic pattern with custom policy
    policy = ElasticPolicy(
        cpu_scale_up_threshold=0.7,
        queue_scale_out_threshold=3,
        min_instances=2,
        max_instances=6,
        cooldown_period=5.0  # Short for demo
    )
    
    pattern = ElasticAgentPattern(policy)
    
    print(f"Initial agents: {len(pattern.agents)}\n")
    
    # Simulate varying workload
    print("Phase 1: Light load (2 tasks)")
    pattern.add_tasks(["Task A", "Task B"])
    pattern.process_tasks()
    
    print(f"\nPhase 2: Heavy load (10 tasks)")
    pattern.add_tasks([f"Task-{i}" for i in range(10)])
    time.sleep(6)  # Wait for cooldown
    pattern.process_tasks()
    
    print(f"\nPhase 3: Processing...")
    time.sleep(6)
    pattern.process_tasks()
    
    # Statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    for key, value in stats.items():
        print(f"{key}: {value}")
    
    # Scaling history
    print(f"\n=== Scaling History ===")
    for decision in pattern.scaling_decisions:
        print(f"{decision.direction.value}: {decision.reason}")

    

Agentic Architecture Pattern: Elastic Agent

Pattern 163: 163 Elastic Agent – Line-by-Line Explanation

Overview

Pattern 163: Elastic Agent Pattern

Purpose: Dynamically scale resources both vertically and horizontally based on demand. Combines both scaling strategies for optimal resource utilization.

Code Structure Analysis

File Information

• Pattern Number: 163
• Pattern Name: 163 Elastic Agent
• File Path: patterns\scaling\163_elastic_agent.py
• Total Lines: 283

Detailed Code Explanation

Module Docstring

"""
Pattern 163: Elastic Agent Pattern

Purpose: Dynamically scale resources both vertically and horizontally based on demand.
Combines both scaling strategies for optimal resource utilization.
"""

Explanation: This module docstring describes the purpose and key features of the 163 Elastic Agent pattern.

Imports Section

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

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

Enumerations

ScalingDirection (Line 17)

class ScalingDirection(Enum):
    SCALE_UP_VERTICAL
    SCALE_DOWN_VERTICAL
    SCALE_OUT_HORIZONTAL
    SCALE_IN_HORIZONTAL
    NO_CHANGE

Purpose: Direction of scaling.

Values: – SCALE_UP_VERTICAL: Represents scale up vertical – SCALE_DOWN_VERTICAL: Represents scale down vertical – SCALE_OUT_HORIZONTAL: Represents scale out horizontal – SCALE_IN_HORIZONTAL: Represents scale in horizontal – NO_CHANGE: Represents no change

Classes

ScalingDirection (Line 17)

Description: Direction of scaling.

ResourceMetrics (Line 27)

Description: Current resource utilization metrics.

ElasticPolicy (Line 38)

Description: Elasticity policy configuration.

ScalingDecision (Line 50)

Description: Record of scaling decision.

ElasticAgent (Line 58)

Description: Agent with elastic scaling capabilities.

Methods:

__init__ (Line 61)

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

Purpose: No method docstring.

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

process_task (Line 69)

def process_task(self, task):
    """Process a task."""

Purpose: Process a task.

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

ElasticAgentPattern (Line 83)

Description: Elastic Agent Pattern – Dynamic bidirectional scaling.

Features: – Vertical scaling: Upgrade/downgrade agent tiers – Horizontal scaling: Add/remove agent instances – Policy-based decisions – Cooldown periods to prevent thrashing – Metrics tracking

Methods:

__init__ (Line 95)

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

Purpose: No method docstring.

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

collect_metrics (Line 107)

def collect_metrics(self):
    """Collect current resource metrics."""

Purpose: Collect current resource metrics.

Parameters: – self: Instance reference

make_scaling_decision (Line 123)

def make_scaling_decision(self, metrics):
    """Decide on scaling action based on metrics."""

Purpose: Decide on scaling action based on metrics.

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

execute_scaling (Line 170)

def execute_scaling(self, decision):
    """Execute the scaling decision."""

Purpose: Execute the scaling decision.

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

add_tasks (Line 209)

def add_tasks(self, tasks):
    """Add tasks to queue."""

Purpose: Add tasks to queue.

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

process_tasks (Line 213)

def process_tasks(self):
    """Process tasks with current agent pool."""

Purpose: Process tasks with current agent pool.

Parameters: – self: Instance reference

get_statistics (Line 227)

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

Purpose: Get elasticity 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 = ScalingDirection(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 162: Horizontal Scaling Pattern

Purpose: Scale by adding more agent instances to distribute workload.
Load balancing across multiple identical agents.
"""

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


class AgentStatus(Enum):
    """Status of agent instance."""
    IDLE = "idle"
    BUSY = "busy"
    FAILED = "failed"


@dataclass
class AgentInstance:
    """Individual agent instance in the pool."""
    id: str
    name: str
    status: AgentStatus = AgentStatus.IDLE
    tasks_completed: int = 0
    current_load: float = 0.0
    created_at: datetime = field(default_factory=datetime.now)


@dataclass
class Task:
    """Task to be processed."""
    id: str
    description: str
    priority: int = 1
    assigned_to: Optional[str] = None
    status: str = "pending"
    result: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.now)


class HorizontalScalingAgent:
    """Single agent instance."""
    
    def __init__(self, instance: AgentInstance, model_name: str = "gpt-4"):
        self.instance = instance
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
    
    def process(self, task: Task) -> str:
        """Process a task."""
        self.instance.status = AgentStatus.BUSY
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.instance.name} (ID: {self.instance.id})"),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task.description})
        
        self.instance.tasks_completed += 1
        self.instance.status = AgentStatus.IDLE
        
        return str(response.content)


class LoadBalancer:
    """Load balancing strategies."""
    
    @staticmethod
    def round_robin(instances: List[AgentInstance], last_index: int) -> int:
        """Round robin: cycle through instances."""
        available = [i for i, inst in enumerate(instances) 
                    if inst.status == AgentStatus.IDLE]
        if not available:
            return -1
        
        # Find next available after last_index
        for i in available:
            if i > last_index:
                return i
        return available[0]
    
    @staticmethod
    def least_loaded(instances: List[AgentInstance]) -> int:
        """Least loaded: choose instance with lowest load."""
        available = [(i, inst) for i, inst in enumerate(instances) 
                    if inst.status == AgentStatus.IDLE]
        if not available:
            return -1
        
        return min(available, key=lambda x: x[1].current_load)[0]
    
    @staticmethod
    def least_tasks(instances: List[AgentInstance]) -> int:
        """Least tasks: choose instance with fewest completed tasks."""
        available = [(i, inst) for i, inst in enumerate(instances) 
                    if inst.status == AgentStatus.IDLE]
        if not available:
            return -1
        
        return min(available, key=lambda x: x[1].tasks_completed)[0]


class HorizontalScalingPattern:
    """
    Horizontal Scaling Pattern - Scale by adding more agent instances.
    
    Features:
    - Dynamic instance creation/removal
    - Load balancing strategies
    - Task queue management
    - Auto-scaling based on queue size
    """
    
    def __init__(self, min_instances: int = 1, max_instances: int = 10,
                 model_name: str = "gpt-4"):
        self.min_instances = min_instances
        self.max_instances = max_instances
        self.model_name = model_name
        self.instances: List[AgentInstance] = []
        self.agents: Dict[str, HorizontalScalingAgent] = {}
        self.task_queue: List[Task] = []
        self.completed_tasks: List[Task] = []
        self.last_robin_index = -1
        
        # Initialize minimum instances
        for _ in range(min_instances):
            self.add_instance()
    
    def add_instance(self) -> Optional[str]:
        """Add a new agent instance."""
        if len(self.instances) >= self.max_instances:
            return None
        
        instance_id = str(uuid.uuid4())[:8]
        instance = AgentInstance(
            id=instance_id,
            name=f"Agent-{len(self.instances) + 1}"
        )
        self.instances.append(instance)
        self.agents[instance_id] = HorizontalScalingAgent(instance, self.model_name)
        
        return instance_id
    
    def remove_instance(self) -> bool:
        """Remove an idle instance."""
        if len(self.instances)  idle_count * 2 and len(self.instances)  2 and queue_size == 0 and len(self.instances) > self.min_instances:
            if self.remove_instance():
                print(f"📉 Scaled down: Now {len(self.instances)} instances")
    
    def process_queue(self, strategy: str = "round_robin"):
        """Process tasks from queue using load balancing strategy."""
        while self.task_queue:
            # Select instance based on strategy
            if strategy == "round_robin":
                idx = LoadBalancer.round_robin(self.instances, self.last_robin_index)
                self.last_robin_index = idx
            elif strategy == "least_loaded":
                idx = LoadBalancer.least_loaded(self.instances)
            elif strategy == "least_tasks":
                idx = LoadBalancer.least_tasks(self.instances)
            else:
                idx = 0
            
            if idx == -1:
                break  # No available instances
            
            # Assign and process task
            task = self.task_queue.pop(0)
            instance = self.instances[idx]
            agent = self.agents[instance.id]
            
            task.assigned_to = instance.id
            task.status = "processing"
            
            result = agent.process(task)
            
            task.result = result
            task.status = "completed"
            self.completed_tasks.append(task)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get scaling statistics."""
        return {
            "total_instances": len(self.instances),
            "idle_instances": sum(1 for i in self.instances if i.status == AgentStatus.IDLE),
            "busy_instances": sum(1 for i in self.instances if i.status == AgentStatus.BUSY),
            "queued_tasks": len(self.task_queue),
            "completed_tasks": len(self.completed_tasks),
            "total_tasks_processed": sum(i.tasks_completed for i in self.instances)
        }


# Example usage
if __name__ == "__main__":
    print("=== Horizontal Scaling Pattern Demo ===\n")
    
    # Create pattern with 1-5 instances
    pattern = HorizontalScalingPattern(min_instances=1, max_instances=5)
    
    print(f"Initial instances: {len(pattern.instances)}\n")
    
    # Add multiple tasks
    tasks = [
        Task(id=f"task-{i}", description=f"Analyze document {i}", priority=1)
        for i in range(1, 11)
    ]
    
    print("Adding 10 tasks to queue...")
    for task in tasks:
        pattern.add_task(task)
    
    print(f"Queue size: {len(pattern.task_queue)}")
    print(f"Instances after auto-scale: {len(pattern.instances)}\n")
    
    # Process with round-robin
    print("Processing with round-robin strategy...")
    pattern.process_queue(strategy="round_robin")
    
    # Statistics
    print("\n=== Statistics ===")
    stats = pattern.get_statistics()
    for key, value in stats.items():
        print(f"{key}: {value}")
    
    # Show instance details
    print("\n=== Instance Details ===")
    for instance in pattern.instances:
        print(f"{instance.name} (ID: {instance.id})")
        print(f"  Status: {instance.status.value}")
        print(f"  Tasks completed: {instance.tasks_completed}")
    
    # Show completed tasks
    print(f"\n=== Sample Completed Tasks ===")
    for task in pattern.completed_tasks[:3]:
        print(f"Task {task.id}: Assigned to {task.assigned_to}")
        print(f"  Result preview: {task.result[:100]}..." if task.result else "  No result")

    

Agentic Architecture Pattern: Horizontal Scaling

Pattern 162: 162 Horizontal Scaling – Line-by-Line Explanation

Overview

Pattern 162: Horizontal Scaling Pattern

Purpose: Scale by adding more agent instances to distribute workload. Load balancing across multiple identical agents.

Code Structure Analysis

File Information

• Pattern Number: 162
• Pattern Name: 162 Horizontal Scaling
• File Path: patterns\scaling\162_horizontal_scaling.py
• Total Lines: 274

Detailed Code Explanation

Module Docstring

"""
Pattern 162: Horizontal Scaling Pattern

Purpose: Scale by adding more agent instances to distribute workload.
Load balancing across multiple identical agents.
"""

Explanation: This module docstring describes the purpose and key features of the 162 Horizontal Scaling pattern.

Imports Section

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

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

Enumerations

AgentStatus (Line 17)

class AgentStatus(Enum):
    IDLE
    BUSY
    FAILED

Purpose: Status of agent instance.

Values: – IDLE: Not currently active – BUSY: Represents busy – FAILED: Execution failed

Classes

AgentStatus (Line 17)

Description: Status of agent instance.

AgentInstance (Line 25)

Description: Individual agent instance in the pool.

Task (Line 36)

Description: Task to be processed.

HorizontalScalingAgent (Line 47)

Description: Single agent instance.

Methods:

__init__ (Line 50)

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

Purpose: No method docstring.

Parameters: – self: Instance reference – instance: Parameter for instance – model_name: LLM model to use (e.g., “gpt-4”)

process (Line 54)

def process(self, task):
    """Process a task."""

Purpose: Process a task.

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

LoadBalancer (Line 72)

Description: Load balancing strategies.

Methods:

round_robin (Line 76)

def round_robin(instances, last_index):
    """Round robin: cycle through instances."""

Purpose: Round robin: cycle through instances.

Parameters: – instances: Parameter for instances – last_index: Parameter for last_index

least_loaded (Line 90)

def least_loaded(instances):
    """Least loaded: choose instance with lowest load."""

Purpose: Least loaded: choose instance with lowest load.

Parameters: – instances: Parameter for instances

least_tasks (Line 100)

def least_tasks(instances):
    """Least tasks: choose instance with fewest completed tasks."""

Purpose: Least tasks: choose instance with fewest completed tasks.

Parameters: – instances: Parameter for instances

HorizontalScalingPattern (Line 110)

Description: Horizontal Scaling Pattern – Scale by adding more agent instances.

Features: – Dynamic instance creation/removal – Load balancing strategies – Task queue management – Auto-scaling based on queue size

Methods:

__init__ (Line 121)

def __init__(self, min_instances, max_instances, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – min_instances: Parameter for min_instances – max_instances: Parameter for max_instances – model_name: LLM model to use (e.g., “gpt-4”)

add_instance (Line 136)

def add_instance(self):
    """Add a new agent instance."""

Purpose: Add a new agent instance.

Parameters: – self: Instance reference

remove_instance (Line 151)

def remove_instance(self):
    """Remove an idle instance."""

Purpose: Remove an idle instance.

Parameters: – self: Instance reference

add_task (Line 166)

def add_task(self, task):
    """Add task to queue."""

Purpose: Add task to queue.

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

auto_scale (Line 171)

def auto_scale(self):
    """Auto-scale based on queue size."""

Purpose: Auto-scale based on queue size.

Parameters: – self: Instance reference

process_queue (Line 187)

def process_queue(self, strategy):
    """Process tasks from queue using load balancing strategy."""

Purpose: Process tasks from queue using load balancing strategy.

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

get_statistics (Line 218)

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

Purpose: Get scaling 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 = AgentStatus(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

    """
Pattern 161: Vertical Scaling Pattern

Purpose: Scale agent capabilities by upgrading to more powerful models/resources
within a single instance (CPU, memory, model size).
"""

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 ResourceTier(Enum):
    """Resource tiers for vertical scaling."""
    BASIC = "gpt-3.5-turbo"
    STANDARD = "gpt-4"
    PREMIUM = "gpt-4-turbo-preview"
    ENTERPRISE = "gpt-4-32k"


@dataclass
class ScalingMetrics:
    """Metrics for scaling decisions."""
    load: float  # 0.0 to 1.0
    response_time: float  # seconds
    success_rate: float  # 0.0 to 1.0
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class ScalingEvent:
    """Record of scaling event."""
    from_tier: ResourceTier
    to_tier: ResourceTier
    reason: str
    metrics: ScalingMetrics
    timestamp: datetime = field(default_factory=datetime.now)


class VerticalScalingAgent:
    """Agent with vertical scaling capabilities."""
    
    def __init__(self, name: str, initial_tier: ResourceTier = ResourceTier.BASIC):
        self.name = name
        self.current_tier = initial_tier
        self.llm = ChatOpenAI(model=initial_tier.value, temperature=0.7)
        self.metrics_history: List[ScalingMetrics] = []
        self.scaling_events: List[ScalingEvent] = []
        
    def scale_up(self) -> bool:
        """Scale up to next tier."""
        tiers = [ResourceTier.BASIC, ResourceTier.STANDARD, 
                ResourceTier.PREMIUM, ResourceTier.ENTERPRISE]
        current_idx = tiers.index(self.current_tier)
        
        if current_idx  bool:
        """Scale down to previous tier."""
        tiers = [ResourceTier.BASIC, ResourceTier.STANDARD, 
                ResourceTier.PREMIUM, ResourceTier.ENTERPRISE]
        current_idx = tiers.index(self.current_tier)
        
        if current_idx > 0:
            old_tier = self.current_tier
            self.current_tier = tiers[current_idx - 1]
            self.llm = ChatOpenAI(model=self.current_tier.value, temperature=0.7)
            
            event = ScalingEvent(
                from_tier=old_tier,
                to_tier=self.current_tier,
                reason="Manual scale down",
                metrics=self.metrics_history[-1] if self.metrics_history else ScalingMetrics(0, 0, 0)
            )
            self.scaling_events.append(event)
            return True
        return False
    
    def process(self, task: str) -> Dict[str, Any]:
        """Process task with current tier."""
        start_time = datetime.now()
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name} running on {self.current_tier.value}"),
            ("human", "{task}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task": task})
        
        end_time = datetime.now()
        response_time = (end_time - start_time).total_seconds()
        
        # Record metrics
        metrics = ScalingMetrics(
            load=0.5,  # Simulated
            response_time=response_time,
            success_rate=1.0
        )
        self.metrics_history.append(metrics)
        
        return {
            "agent": self.name,
            "tier": self.current_tier.name,
            "result": response.content,
            "response_time": response_time
        }


class VerticalScalingPattern:
    """
    Vertical Scaling Pattern - Scale by upgrading resources/capabilities.
    
    Auto-scales based on load, response time, and success rate thresholds.
    """
    
    def __init__(self):
        self.agents: Dict[str, VerticalScalingAgent] = {}
        self.scale_up_threshold = 0.8  # Load threshold to scale up
        self.scale_down_threshold = 0.3  # Load threshold to scale down
        self.response_time_threshold = 5.0  # seconds
        
    def register_agent(self, agent: VerticalScalingAgent):
        """Register an agent for scaling management."""
        self.agents[agent.name] = agent
    
    def check_and_scale(self, agent_name: str) -> Optional[str]:
        """Check metrics and auto-scale if needed."""
        agent = self.agents.get(agent_name)
        if not agent or not agent.metrics_history:
            return None
        
        recent_metrics = agent.metrics_history[-5:]  # Last 5 metrics
        avg_load = sum(m.load for m in recent_metrics) / len(recent_metrics)
        avg_response_time = sum(m.response_time for m in recent_metrics) / len(recent_metrics)
        avg_success_rate = sum(m.success_rate for m in recent_metrics) / len(recent_metrics)
        
        # Scale up if overloaded or slow
        if avg_load > self.scale_up_threshold or avg_response_time > self.response_time_threshold:
            if agent.scale_up():
                return f"Scaled up {agent_name} to {agent.current_tier.name}"
        
        # Scale down if underutilized
        elif avg_load  0.95:
            if agent.scale_down():
                return f"Scaled down {agent_name} to {agent.current_tier.name}"
        
        return None
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get scaling statistics."""
        stats = {}
        for name, agent in self.agents.items():
            stats[name] = {
                "current_tier": agent.current_tier.name,
                "scaling_events": len(agent.scaling_events),
                "metrics_collected": len(agent.metrics_history),
                "recent_response_time": agent.metrics_history[-1].response_time if agent.metrics_history else 0
            }
        return stats


# Example usage
if __name__ == "__main__":
    print("=== Vertical Scaling Pattern Demo ===\n")
    
    # Create pattern
    pattern = VerticalScalingPattern()
    
    # Create agent starting at BASIC tier
    agent = VerticalScalingAgent("TaskAgent", ResourceTier.BASIC)
    pattern.register_agent(agent)
    
    print(f"Initial tier: {agent.current_tier.name}\n")
    
    # Process tasks
    tasks = [
        "Summarize this text",
        "Analyze complex data patterns",
        "Generate creative content"
    ]
    
    for i, task in enumerate(tasks, 1):
        print(f"Task {i}: {task}")
        result = agent.process(task)
        print(f"  Tier: {result['tier']}")
        print(f"  Response time: {result['response_time']:.2f}s")
        
        # Simulate increasing load
        if agent.metrics_history:
            agent.metrics_history[-1].load = 0.5 + (i * 0.2)
        
        # Check for auto-scaling
        scale_action = pattern.check_and_scale("TaskAgent")
        if scale_action:
            print(f"  ⚡ {scale_action}")
        print()
    
    # Show statistics
    print("\n=== Scaling Statistics ===")
    stats = pattern.get_statistics()
    for agent_name, agent_stats in stats.items():
        print(f"\n{agent_name}:")
        for key, value in agent_stats.items():
            print(f"  {key}: {value}")
    
    # Show scaling events
    print(f"\n=== Scaling Events ===")
    for event in agent.scaling_events:
        print(f"{event.from_tier.name} → {event.to_tier.name}: {event.reason}")

    

Agentic Architecture Pattern: Vertical Scaling

Pattern 161: 161 Vertical Scaling – Line-by-Line Explanation

Overview

Pattern 161: Vertical Scaling Pattern

Purpose: Scale agent capabilities by upgrading to more powerful models/resources within a single instance (CPU, memory, model size).

Code Structure Analysis

File Information

• Pattern Number: 161
• Pattern Name: 161 Vertical Scaling
• File Path: patterns\scaling\161_vertical_scaling.py
• Total Lines: 227

Detailed Code Explanation

Module Docstring

"""
Pattern 161: Vertical Scaling Pattern

Purpose: Scale agent capabilities by upgrading to more powerful models/resources
within a single instance (CPU, memory, model size).
"""

Explanation: This module docstring describes the purpose and key features of the 161 Vertical Scaling 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

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

Enumerations

ResourceTier (Line 16)

class ResourceTier(Enum):
    BASIC
    STANDARD
    PREMIUM
    ENTERPRISE

Purpose: Resource tiers for vertical scaling.

Values: – BASIC: Represents basic – STANDARD: Represents standard – PREMIUM: Represents premium – ENTERPRISE: Represents enterprise

Classes

ResourceTier (Line 16)

Description: Resource tiers for vertical scaling.

ScalingMetrics (Line 25)

Description: Metrics for scaling decisions.

ScalingEvent (Line 34)

Description: Record of scaling event.

VerticalScalingAgent (Line 43)

Description: Agent with vertical scaling capabilities.

Methods:

__init__ (Line 46)

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

Purpose: No method docstring.

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

scale_up (Line 53)

def scale_up(self):
    """Scale up to next tier."""

Purpose: Scale up to next tier.

Parameters: – self: Instance reference

scale_down (Line 74)

def scale_down(self):
    """Scale down to previous tier."""

Purpose: Scale down to previous tier.

Parameters: – self: Instance reference

process (Line 95)

def process(self, task):
    """Process task with current tier."""

Purpose: Process task with current tier.

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

VerticalScalingPattern (Line 126)

Description: Vertical Scaling Pattern – Scale by upgrading resources/capabilities.

Auto-scales based on load, response time, and success rate thresholds.

Methods:

__init__ (Line 133)

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 139)

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

Purpose: Register an agent for scaling management.

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

check_and_scale (Line 143)

def check_and_scale(self, agent_name):
    """Check metrics and auto-scale if needed."""

Purpose: Check metrics and auto-scale if needed.

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

get_statistics (Line 166)

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

Purpose: Get scaling 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 = ResourceTier(name="example", model_name="gpt-4")

# Configure and execute
result = pattern.execute(data={"key": "value"})

# Check results
print(result)

Key Concepts

• LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
• Workflow Orchestration: LangGraph manages execution flow through nodes
• State Management: TypedDict defines workflow state structure
• Error Handling: Comprehensive error handling and fallback mechanisms
• Extensibility: Designed for easy customization and extension

Extension Points

This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities

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


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

Pattern 160: 160 Logic Based Pattern – Line-by-Line Explanation

Overview

Pattern 160: Logic Based Pattern

Purpose: Enhanced implementation with full LangChain/LangGraph integration

Code Structure Analysis

File Information

• Pattern Number: 160
• Pattern Name: 160 Logic Based Pattern
• File Path: patterns\hybrid_ai\160_logic_based_pattern.py
• Total Lines: 247

Detailed Code Explanation

Module Docstring

"""
Pattern 160: Logic Based Pattern

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

Explanation: This module docstring describes the purpose and key features of the 160 Logic Based 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

LogicBasedConfig (Line 18)

Description: Configuration for Logic Based Pattern.

LogicBasedState (Line 27)

Description: State information for Logic Based Pattern.

LogicBasedAgent (Line 36)

Description: Agent implementing Logic Based 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 Logic Based Pattern."""

Purpose: Process input according to Logic Based 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

LogicBasedPattern (Line 85)

Description: Logic Based Pattern – Enhanced implementation

This pattern provides comprehensive functionality for logic based 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 = LogicBasedConfig(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