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