aishastr.live

Agentic Architecture Pattern: Agent Factory

Written by

admin

in


    """
Pattern 282: Agent Factory Pattern

Purpose: Creational pattern for dynamic agent instantiation with templates
"""

from typing import Dict, Any, List, Optional, Type, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from abc import ABC, abstractmethod
import copy


class AgentRole(Enum):
    """Agent roles."""
    WORKER = "worker"
    ANALYZER = "analyzer"
    COORDINATOR = "coordinator"
    SPECIALIST = "specialist"
    MANAGER = "manager"


@dataclass
class AgentTemplate:
    """Template for agent creation."""
    template_id: str
    role: AgentRole
    capabilities: List[str]
    default_config: Dict[str, Any] = field(default_factory=dict)
    initialization_params: Dict[str, Any] = field(default_factory=dict)
    description: str = ""


@dataclass
class AgentInstance:
    """Created agent instance."""
    instance_id: str
    template_id: str
    role: AgentRole
    capabilities: List[str]
    config: Dict[str, Any]
    created_at: datetime = field(default_factory=datetime.now)
    status: str = "active"
    tasks_completed: int = 0
    metadata: Dict[str, Any] = field(default_factory=dict)


class AgentFactory:
    """
    Implements agent factory pattern.
    
    Features:
    - Template-based agent creation
    - Dynamic agent instantiation
    - Configuration management
    - Agent lifecycle tracking
    - Template inheritance
    - Prototype cloning
    """
    
    def __init__(self):
        self.templates: Dict[str, AgentTemplate] = {}
        self.instances: Dict[str, AgentInstance] = {}
        self.creation_count_by_template: Dict[str, int] = {}
        
        # Statistics
        self.statistics = {
            "total_templates": 0,
            "total_instances_created": 0,
            "active_instances": 0,
            "by_role": {role.value: 0 for role in AgentRole}
        }
    
    def register_template(
        self,
        template_id: str,
        role: AgentRole,
        capabilities: List[str],
        default_config: Dict[str, Any] = None,
        description: str = ""
    ) -> Dict[str, Any]:
        """
        Register an agent template.
        
        Args:
            template_id: Unique template identifier
            role: Agent role
            capabilities: List of capabilities
            default_config: Default configuration
            description: Template description
            
        Returns:
            Registration status
        """
        if template_id in self.templates:
            return {
                "success": False,
                "message": f"Template '{template_id}' already exists"
            }
        
        template = AgentTemplate(
            template_id=template_id,
            role=role,
            capabilities=capabilities,
            default_config=default_config or {},
            description=description
        )
        
        self.templates[template_id] = template
        self.creation_count_by_template[template_id] = 0
        self.statistics["total_templates"] += 1
        
        return {
            "success": True,
            "template_id": template_id,
            "role": role.value,
            "capabilities": capabilities,
            "message": f"Template '{template_id}' registered"
        }
    
    def create_agent(
        self,
        template_id: str,
        custom_config: Dict[str, Any] = None,
        metadata: Dict[str, Any] = None
    ) -> Dict[str, Any]:
        """
        Create agent from template.
        
        Args:
            template_id: Template to use
            custom_config: Custom configuration overrides
            metadata: Additional metadata
            
        Returns:
            Created agent information
        """
        if template_id not in self.templates:
            return {
                "success": False,
                "message": f"Template '{template_id}' not found"
            }
        
        template = self.templates[template_id]
        
        # Generate instance ID
        self.creation_count_by_template[template_id] += 1
        instance_id = f"{template_id}-{self.creation_count_by_template[template_id]:04d}"
        
        # Merge configurations
        config = copy.deepcopy(template.default_config)
        if custom_config:
            config.update(custom_config)
        
        # Create instance
        instance = AgentInstance(
            instance_id=instance_id,
            template_id=template_id,
            role=template.role,
            capabilities=template.capabilities.copy(),
            config=config,
            metadata=metadata or {}
        )
        
        self.instances[instance_id] = instance
        self.statistics["total_instances_created"] += 1
        self.statistics["active_instances"] += 1
        self.statistics["by_role"][template.role.value] += 1
        
        return {
            "success": True,
            "instance_id": instance_id,
            "template_id": template_id,
            "role": template.role.value,
            "capabilities": template.capabilities,
            "config": config,
            "message": f"Agent '{instance_id}' created from template '{template_id}'"
        }
    
    def create_batch(
        self,
        template_id: str,
        count: int,
        custom_config: Dict[str, Any] = None
    ) -> Dict[str, Any]:
        """
        Create multiple agents from template.
        
        Args:
            template_id: Template to use
            count: Number of agents to create
            custom_config: Custom configuration
            
        Returns:
            Batch creation results
        """
        if template_id not in self.templates:
            return {
                "success": False,
                "message": f"Template '{template_id}' not found"
            }
        
        created_ids = []
        for _ in range(count):
            result = self.create_agent(template_id, custom_config)
            if result["success"]:
                created_ids.append(result["instance_id"])
        
        return {
            "success": True,
            "template_id": template_id,
            "requested_count": count,
            "created_count": len(created_ids),
            "instance_ids": created_ids,
            "message": f"Created {len(created_ids)} agents from template '{template_id}'"
        }
    
    def clone_agent(
        self,
        source_instance_id: str,
        custom_config: Dict[str, Any] = None
    ) -> Dict[str, Any]:
        """
        Clone existing agent (prototype pattern).
        
        Args:
            source_instance_id: Agent to clone
            custom_config: Configuration overrides
            
        Returns:
            Cloned agent information
        """
        if source_instance_id not in self.instances:
            return {
                "success": False,
                "message": f"Instance '{source_instance_id}' not found"
            }
        
        source = self.instances[source_instance_id]
        
        # Create from same template with source's config
        merged_config = copy.deepcopy(source.config)
        if custom_config:
            merged_config.update(custom_config)
        
        return self.create_agent(
            template_id=source.template_id,
            custom_config=merged_config,
            metadata={"cloned_from": source_instance_id}
        )
    
    def destroy_agent(self, instance_id: str) -> Dict[str, Any]:
        """
        Destroy agent instance.
        
        Args:
            instance_id: Instance to destroy
            
        Returns:
            Destruction status
        """
        if instance_id not in self.instances:
            return {
                "success": False,
                "message": f"Instance '{instance_id}' not found"
            }
        
        instance = self.instances[instance_id]
        instance.status = "destroyed"
        
        self.statistics["active_instances"] -= 1
        self.statistics["by_role"][instance.role.value] -= 1
        
        return {
            "success": True,
            "instance_id": instance_id,
            "tasks_completed": instance.tasks_completed,
            "lifetime_seconds": (datetime.now() - instance.created_at).total_seconds(),
            "message": f"Agent '{instance_id}' destroyed"
        }
    
    def get_template_info(self, template_id: str) -> Dict[str, Any]:
        """Get template information."""
        if template_id not in self.templates:
            return {
                "success": False,
                "message": f"Template '{template_id}' not found"
            }
        
        template = self.templates[template_id]
        instances_created = self.creation_count_by_template[template_id]
        
        # Count active instances
        active_instances = [
            iid for iid, inst in self.instances.items()
            if inst.template_id == template_id and inst.status == "active"
        ]
        
        return {
            "success": True,
            "template_id": template_id,
            "role": template.role.value,
            "capabilities": template.capabilities,
            "description": template.description,
            "default_config": template.default_config,
            "instances_created": instances_created,
            "active_instances": len(active_instances),
            "instance_ids": active_instances
        }
    
    def get_instance_info(self, instance_id: str) -> Dict[str, Any]:
        """Get instance information."""
        if instance_id not in self.instances:
            return {
                "success": False,
                "message": f"Instance '{instance_id}' not found"
            }
        
        instance = self.instances[instance_id]
        
        return {
            "success": True,
            "instance_id": instance_id,
            "template_id": instance.template_id,
            "role": instance.role.value,
            "capabilities": instance.capabilities,
            "status": instance.status,
            "tasks_completed": instance.tasks_completed,
            "created_at": instance.created_at.isoformat(),
            "lifetime_seconds": (datetime.now() - instance.created_at).total_seconds(),
            "config": instance.config,
            "metadata": instance.metadata
        }
    
    def list_templates(self) -> Dict[str, Any]:
        """List all templates."""
        templates_list = [
            {
                "template_id": tid,
                "role": t.role.value,
                "capabilities": t.capabilities,
                "instances_created": self.creation_count_by_template[tid]
            }
            for tid, t in self.templates.items()
        ]
        
        return {
            "total": len(templates_list),
            "templates": templates_list
        }
    
    def list_instances(
        self,
        template_id: Optional[str] = None,
        role: Optional[AgentRole] = None,
        status: Optional[str] = None
    ) -> Dict[str, Any]:
        """List instances with optional filters."""
        instances_list = []
        
        for iid, inst in self.instances.items():
            if template_id and inst.template_id != template_id:
                continue
            if role and inst.role != role:
                continue
            if status and inst.status != status:
                continue
            
            instances_list.append({
                "instance_id": iid,
                "template_id": inst.template_id,
                "role": inst.role.value,
                "status": inst.status,
                "tasks_completed": inst.tasks_completed
            })
        
        return {
            "total": len(instances_list),
            "instances": instances_list
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get factory statistics."""
        return {
            "total_templates": self.statistics["total_templates"],
            "total_instances_created": self.statistics["total_instances_created"],
            "active_instances": self.statistics["active_instances"],
            "instances_by_role": self.statistics["by_role"],
            "average_instances_per_template": (
                self.statistics["total_instances_created"] / self.statistics["total_templates"]
                if self.statistics["total_templates"] > 0 else 0
            )
        }


class AgentFactoryPattern:
    """
    Coordinator for agent factory pattern.
    
    Manages agent creation and templates.
    """
    
    def __init__(self):
        self.factory = AgentFactory()
    
    def register_template(self, template_id: str, role: AgentRole, capabilities: List[str], **kwargs) -> Dict[str, Any]:
        """Register template."""
        return self.factory.register_template(template_id, role, capabilities, **kwargs)
    
    def create_agent(self, template_id: str, custom_config: Dict[str, Any] = None, metadata: Dict[str, Any] = None) -> Dict[str, Any]:
        """Create agent."""
        return self.factory.create_agent(template_id, custom_config, metadata)
    
    def create_batch(self, template_id: str, count: int, custom_config: Dict[str, Any] = None) -> Dict[str, Any]:
        """Create batch."""
        return self.factory.create_batch(template_id, count, custom_config)
    
    def clone_agent(self, source_instance_id: str, custom_config: Dict[str, Any] = None) -> Dict[str, Any]:
        """Clone agent."""
        return self.factory.clone_agent(source_instance_id, custom_config)
    
    def destroy_agent(self, instance_id: str) -> Dict[str, Any]:
        """Destroy agent."""
        return self.factory.destroy_agent(instance_id)
    
    def get_template_info(self, template_id: str) -> Dict[str, Any]:
        """Get template info."""
        return self.factory.get_template_info(template_id)
    
    def list_templates(self) -> Dict[str, Any]:
        """List templates."""
        return self.factory.list_templates()
    
    def list_instances(self) -> Dict[str, Any]:
        """List instances."""
        return self.factory.list_instances()
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get statistics."""
        return self.factory.get_statistics()


# Example usage
if __name__ == "__main__":
    print("=== Agent Factory Pattern Demo ===\n")
    
    # Create factory
    pattern = AgentFactoryPattern()
    
    print("1. Registering agent templates...")
    pattern.register_template(
        "data-worker",
        AgentRole.WORKER,
        ["data_processing", "etl"],
        default_config={"batch_size": 100, "timeout": 30},
        description="Worker for data processing tasks"
    )
    
    pattern.register_template(
        "ml-specialist",
        AgentRole.SPECIALIST,
        ["machine_learning", "model_training"],
        default_config={"max_epochs": 100, "learning_rate": 0.001},
        description="Specialist for ML tasks"
    )
    
    pattern.register_template(
        "coordinator",
        AgentRole.COORDINATOR,
        ["task_coordination", "workflow_management"],
        description="Coordinates multiple agents"
    )
    
    templates = pattern.list_templates()
    print(f"  Registered {templates['total']} templates")
    for tmpl in templates['templates']:
        print(f"    - {tmpl['template_id']}: {tmpl['role']}")
    
    print("\n2. Creating agents from templates...")
    agent1 = pattern.create_agent("data-worker")
    print(f"  {agent1['message']}")
    
    agent2 = pattern.create_agent(
        "ml-specialist",
        custom_config={"learning_rate": 0.01}
    )
    print(f"  {agent2['message']}")
    
    print("\n3. Creating batch of workers...")
    batch = pattern.create_batch("data-worker", 3)
    print(f"  {batch['message']}")
    print(f"  IDs: {', '.join(batch['instance_ids'])}")
    
    print("\n4. Cloning an agent...")
    clone = pattern.clone_agent(
        agent1["instance_id"],
        custom_config={"batch_size": 200}
    )
    print(f"  {clone['message']}")
    
    print("\n5. Getting template info...")
    info = pattern.get_template_info("data-worker")
    if info['success']:
        print(f"  Template: {info['template_id']}")
        print(f"  Role: {info['role']}")
        print(f"  Instances created: {info['instances_created']}")
        print(f"  Active instances: {info['active_instances']}")
    
    print("\n6. Listing all instances...")
    instances = pattern.list_instances()
    print(f"  Total instances: {instances['total']}")
    for inst in instances['instances'][:5]:  # Show first 5
        print(f"    - {inst['instance_id']}: {inst['role']} ({inst['status']})")
    
    print("\n7. Factory statistics:")
    stats = pattern.get_statistics()
    print(f"  Total templates: {stats['total_templates']}")
    print(f"  Total instances created: {stats['total_instances_created']}")
    print(f"  Active instances: {stats['active_instances']}")
    print(f"  Avg instances per template: {stats['average_instances_per_template']:.1f}")
    print("  By role:")
    for role, count in stats['instances_by_role'].items():
        if count > 0:
            print(f"    - {role}: {count}")
    
    print("\nāœ“ Comprehensive agent factory pattern complete!")

Agentic Architecture Pattern: Agent Factory

Pattern 282: 282 Agent Factory – Line-by-Line Explanation

Overview

Pattern 282: Agent Factory Pattern

Purpose: Creational pattern for dynamic agent instantiation with templates

Code Structure Analysis

File Information

  • Pattern Number: 282
  • Pattern Name: 282 Agent Factory
  • File Path: patterns\additional_advanced\282_agent_factory.py
  • Total Lines: 528

Detailed Code Explanation

Module Docstring

"""
Pattern 282: Agent Factory Pattern

Purpose: Creational pattern for dynamic agent instantiation with templates
"""

Explanation: This module docstring describes the purpose and key features of the 282 Agent Factory pattern.

Imports Section

from typing import Dict, Any, List, Optional, Type, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from abc import ABC, abstractmethod
import copy  # First 20 imports shown

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

Enumerations

AgentRole (Line 15)

class AgentRole(Enum):
    WORKER
    ANALYZER
    COORDINATOR
    SPECIALIST
    MANAGER

Purpose: Agent roles.

Values: – WORKER: Represents worker – ANALYZER: Represents analyzer – COORDINATOR: Represents coordinator – SPECIALIST: Represents specialist – MANAGER: Represents manager

Classes

AgentRole (Line 15)

Description: Agent roles.

AgentTemplate (Line 25)

Description: Template for agent creation.

AgentInstance (Line 36)

Description: Created agent instance.

AgentFactory (Line 49)

Description: Implements agent factory pattern.

Features: – Template-based agent creation – Dynamic agent instantiation – Configuration management – Agent lifecycle tracking – Template inheritance – Prototype cloning

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_template (Line 75)
def register_template(self, template_id, role, capabilities, default_config, description):
    """Register an agent template.

Args:
    template_id: Unique template identifier
    role: Agent role
    capabilities: List of capabilities
    default_config: Default configuration
    description: Template description

Returns:
    Registration status"""

Purpose: Register an agent template.

Args: template_id: Unique template identifier role: Agent role capabilities: List of capabilities default_config: Default configuration description: Template description

Returns: Registration status

Parameters: – self: Instance reference – template_id: Parameter for template_id – role: Parameter for role – capabilities: Parameter for capabilities – default_config: Parameter for default_config – description: Parameter for description

create_agent (Line 122)
def create_agent(self, template_id, custom_config, metadata):
    """Create agent from template.

Args:
    template_id: Template to use
    custom_config: Custom configuration overrides
    metadata: Additional metadata

Returns:
    Created agent information"""

Purpose: Create agent from template.

Args: template_id: Template to use custom_config: Custom configuration overrides metadata: Additional metadata

Returns: Created agent information

Parameters: – self: Instance reference – template_id: Parameter for template_id – custom_config: Parameter for custom_config – metadata: Parameter for metadata

create_batch (Line 181)
def create_batch(self, template_id, count, custom_config):
    """Create multiple agents from template.

Args:
    template_id: Template to use
    count: Number of agents to create
    custom_config: Custom configuration

Returns:
    Batch creation results"""

Purpose: Create multiple agents from template.

Args: template_id: Template to use count: Number of agents to create custom_config: Custom configuration

Returns: Batch creation results

Parameters: – self: Instance reference – template_id: Parameter for template_id – count: Parameter for count – custom_config: Parameter for custom_config

clone_agent (Line 219)
def clone_agent(self, source_instance_id, custom_config):
    """Clone existing agent (prototype pattern).

Args:
    source_instance_id: Agent to clone
    custom_config: Configuration overrides

Returns:
    Cloned agent information"""

Purpose: Clone existing agent (prototype pattern).

Args: source_instance_id: Agent to clone custom_config: Configuration overrides

Returns: Cloned agent information

Parameters: – self: Instance reference – source_instance_id: Parameter for source_instance_id – custom_config: Parameter for custom_config

destroy_agent (Line 253)
def destroy_agent(self, instance_id):
    """Destroy agent instance.

Args:
    instance_id: Instance to destroy

Returns:
    Destruction status"""

Purpose: Destroy agent instance.

Args: instance_id: Instance to destroy

Returns: Destruction status

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

get_template_info (Line 283)
def get_template_info(self, template_id):
    """Get template information."""

Purpose: Get template information.

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

get_instance_info (Line 312)
def get_instance_info(self, instance_id):
    """Get instance information."""

Purpose: Get instance information.

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

list_templates (Line 336)
def list_templates(self):
    """List all templates."""

Purpose: List all templates.

Parameters: – self: Instance reference

list_instances (Line 353)
def list_instances(self, template_id, role, status):
    """List instances with optional filters."""

Purpose: List instances with optional filters.

Parameters: – self: Instance reference – template_id: Parameter for template_id – role: Parameter for role – status: Parameter for status

get_statistics (Line 383)
def get_statistics(self):
    """Get factory statistics."""

Purpose: Get factory statistics.

Parameters: – self: Instance reference

AgentFactoryPattern (Line 397)

Description: Coordinator for agent factory pattern.

Manages agent creation and templates.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_template (Line 407)
def register_template(self, template_id, role, capabilities):
    """Register template."""

Purpose: Register template.

Parameters: – self: Instance reference – template_id: Parameter for template_id – role: Parameter for role – capabilities: Parameter for capabilities

create_agent (Line 411)
def create_agent(self, template_id, custom_config, metadata):
    """Create agent."""

Purpose: Create agent.

Parameters: – self: Instance reference – template_id: Parameter for template_id – custom_config: Parameter for custom_config – metadata: Parameter for metadata

create_batch (Line 415)
def create_batch(self, template_id, count, custom_config):
    """Create batch."""

Purpose: Create batch.

Parameters: – self: Instance reference – template_id: Parameter for template_id – count: Parameter for count – custom_config: Parameter for custom_config

clone_agent (Line 419)
def clone_agent(self, source_instance_id, custom_config):
    """Clone agent."""

Purpose: Clone agent.

Parameters: – self: Instance reference – source_instance_id: Parameter for source_instance_id – custom_config: Parameter for custom_config

destroy_agent (Line 423)
def destroy_agent(self, instance_id):
    """Destroy agent."""

Purpose: Destroy agent.

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

get_template_info (Line 427)
def get_template_info(self, template_id):
    """Get template info."""

Purpose: Get template info.

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

list_templates (Line 431)
def list_templates(self):
    """List templates."""

Purpose: List templates.

Parameters: – self: Instance reference

list_instances (Line 435)
def list_instances(self):
    """List instances."""

Purpose: List instances.

Parameters: – self: Instance reference

get_statistics (Line 439)
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 = AgentRole(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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts