aishastr.live

Agentic Architecture Pattern: Meta Agent

Written by

admin

in


    """
Pattern 281: Meta-Agent Pattern

Purpose: Agent that creates, manages, and coordinates other agents dynamically
"""

from typing import Dict, Any, List, Optional, Set, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict


class AgentType(Enum):
    """Type of managed agent."""
    WORKER = "worker"
    SPECIALIST = "specialist"
    COORDINATOR = "coordinator"
    ANALYZER = "analyzer"
    EXECUTOR = "executor"


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


@dataclass
class ManagedAgent:
    """Represents an agent managed by the meta-agent."""
    agent_id: str
    agent_type: AgentType
    capabilities: Set[str]
    status: AgentStatus = AgentStatus.IDLE
    created_at: datetime = field(default_factory=datetime.now)
    tasks_completed: int = 0
    tasks_failed: int = 0
    last_activity: Optional[datetime] = None
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class TaskAssignment:
    """Task assigned to an agent."""
    task_id: str
    agent_id: str
    task_description: str
    required_capability: str
    assigned_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None
    result: Optional[Any] = None
    success: bool = False


class MetaAgent:
    """
    Implements meta-agent pattern.
    
    Features:
    - Dynamic agent creation and termination
    - Agent capability management
    - Intelligent task routing
    - Load balancing across agents
    - Agent performance monitoring
    - Automatic agent spawning based on demand
    """
    
    def __init__(self):
        self.agents: Dict[str, ManagedAgent] = {}
        self.tasks: Dict[str, TaskAssignment] = {}
        self.capability_registry: Dict[str, Set[str]] = defaultdict(set)
        
        # Statistics
        self.statistics = {
            "total_agents_created": 0,
            "active_agents": 0,
            "total_tasks_assigned": 0,
            "total_tasks_completed": 0,
            "total_tasks_failed": 0
        }
    
    def create_agent(
        self,
        agent_type: AgentType,
        capabilities: Set[str],
        metadata: Dict[str, Any] = None
    ) -> Dict[str, Any]:
        """
        Create a new managed agent.
        
        Args:
            agent_type: Type of agent to create
            capabilities: Set of capabilities the agent has
            metadata: Additional metadata
            
        Returns:
            Agent creation status and ID
        """
        agent_id = f"agent-{len(self.agents) + 1:04d}"
        
        agent = ManagedAgent(
            agent_id=agent_id,
            agent_type=agent_type,
            capabilities=capabilities,
            metadata=metadata or {}
        )
        
        self.agents[agent_id] = agent
        
        # Register capabilities
        for capability in capabilities:
            self.capability_registry[capability].add(agent_id)
        
        self.statistics["total_agents_created"] += 1
        self.statistics["active_agents"] += 1
        
        return {
            "success": True,
            "agent_id": agent_id,
            "agent_type": agent_type.value,
            "capabilities": list(capabilities),
            "message": f"Agent '{agent_id}' created with {len(capabilities)} capabilities"
        }
    
    def terminate_agent(self, agent_id: str) -> Dict[str, Any]:
        """
        Terminate a managed agent.
        
        Args:
            agent_id: ID of agent to terminate
            
        Returns:
            Termination status
        """
        if agent_id not in self.agents:
            return {
                "success": False,
                "message": f"Agent '{agent_id}' not found"
            }
        
        agent = self.agents[agent_id]
        
        # Remove from capability registry
        for capability in agent.capabilities:
            self.capability_registry[capability].discard(agent_id)
        
        agent.status = AgentStatus.TERMINATED
        self.statistics["active_agents"] -= 1
        
        return {
            "success": True,
            "agent_id": agent_id,
            "tasks_completed": agent.tasks_completed,
            "tasks_failed": agent.tasks_failed,
            "message": f"Agent '{agent_id}' terminated"
        }
    
    def assign_task(
        self,
        task_description: str,
        required_capability: str
    ) -> Dict[str, Any]:
        """
        Assign task to most suitable agent.
        
        Args:
            task_description: Description of task
            required_capability: Capability needed for task
            
        Returns:
            Task assignment status
        """
        # Find agents with required capability
        candidate_agent_ids = self.capability_registry.get(required_capability, set())
        
        if not candidate_agent_ids:
            return {
                "success": False,
                "message": f"No agent available with capability '{required_capability}'"
            }
        
        # Filter idle agents
        idle_agents = [
            aid for aid in candidate_agent_ids
            if self.agents[aid].status == AgentStatus.IDLE
        ]
        
        if not idle_agents:
            # All agents busy, pick least loaded
            agent_id = min(
                candidate_agent_ids,
                key=lambda aid: self.agents[aid].tasks_completed - self.agents[aid].tasks_failed
            )
        else:
            # Pick idle agent with best track record
            agent_id = max(
                idle_agents,
                key=lambda aid: self.agents[aid].tasks_completed
            )
        
        # Create task assignment
        task_id = f"task-{len(self.tasks) + 1:04d}"
        assignment = TaskAssignment(
            task_id=task_id,
            agent_id=agent_id,
            task_description=task_description,
            required_capability=required_capability
        )
        
        self.tasks[task_id] = assignment
        
        # Update agent status
        agent = self.agents[agent_id]
        agent.status = AgentStatus.BUSY
        agent.last_activity = datetime.now()
        
        self.statistics["total_tasks_assigned"] += 1
        
        return {
            "success": True,
            "task_id": task_id,
            "agent_id": agent_id,
            "agent_type": agent.agent_type.value,
            "message": f"Task '{task_id}' assigned to '{agent_id}'"
        }
    
    def complete_task(
        self,
        task_id: str,
        result: Any = None,
        success: bool = True
    ) -> Dict[str, Any]:
        """
        Mark task as completed.
        
        Args:
            task_id: Task ID
            result: Task result
            success: Whether task succeeded
            
        Returns:
            Completion status
        """
        if task_id not in self.tasks:
            return {
                "success": False,
                "message": f"Task '{task_id}' not found"
            }
        
        assignment = self.tasks[task_id]
        assignment.completed_at = datetime.now()
        assignment.result = result
        assignment.success = success
        
        # Update agent
        agent = self.agents[assignment.agent_id]
        agent.status = AgentStatus.IDLE
        agent.last_activity = datetime.now()
        
        if success:
            agent.tasks_completed += 1
            self.statistics["total_tasks_completed"] += 1
        else:
            agent.tasks_failed += 1
            self.statistics["total_tasks_failed"] += 1
            agent.status = AgentStatus.FAILED
        
        return {
            "success": True,
            "task_id": task_id,
            "agent_id": assignment.agent_id,
            "task_success": success,
            "duration_seconds": (assignment.completed_at - assignment.assigned_at).total_seconds(),
            "message": f"Task '{task_id}' {'completed' if success else 'failed'}"
        }
    
    def scale_agents(
        self,
        required_capability: str,
        target_count: int,
        agent_type: AgentType = AgentType.WORKER
    ) -> Dict[str, Any]:
        """
        Scale number of agents with specific capability.
        
        Args:
            required_capability: Capability to scale for
            target_count: Target number of agents
            agent_type: Type of agents to create
            
        Returns:
            Scaling operation status
        """
        current_agents = self.capability_registry.get(required_capability, set())
        current_count = len(current_agents)
        
        if current_count == target_count:
            return {
                "success": True,
                "action": "none",
                "current_count": current_count,
                "message": f"Already at target count of {target_count} agents"
            }
        elif current_count  Dict[str, Any]:
        """Get detailed agent status."""
        if agent_id not in self.agents:
            return {
                "success": False,
                "message": f"Agent '{agent_id}' not found"
            }
        
        agent = self.agents[agent_id]
        
        # Find assigned tasks
        assigned_tasks = [
            t.task_id for t in self.tasks.values()
            if t.agent_id == agent_id and t.completed_at is None
        ]
        
        return {
            "success": True,
            "agent_id": agent_id,
            "agent_type": agent.agent_type.value,
            "status": agent.status.value,
            "capabilities": list(agent.capabilities),
            "tasks_completed": agent.tasks_completed,
            "tasks_failed": agent.tasks_failed,
            "success_rate": (agent.tasks_completed / (agent.tasks_completed + agent.tasks_failed)
                           if (agent.tasks_completed + agent.tasks_failed) > 0 else 0.0),
            "assigned_tasks": assigned_tasks,
            "last_activity": agent.last_activity.isoformat() if agent.last_activity else None
        }
    
    def list_agents(
        self,
        agent_type: Optional[AgentType] = None,
        status: Optional[AgentStatus] = None
    ) -> Dict[str, Any]:
        """List agents with optional filters."""
        agents_list = []
        
        for agent_id, agent in self.agents.items():
            if agent_type and agent.agent_type != agent_type:
                continue
            if status and agent.status != status:
                continue
            
            agents_list.append({
                "agent_id": agent_id,
                "agent_type": agent.agent_type.value,
                "status": agent.status.value,
                "capabilities": list(agent.capabilities),
                "tasks_completed": agent.tasks_completed
            })
        
        return {
            "total": len(agents_list),
            "agents": agents_list
        }
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get meta-agent statistics."""
        # Calculate additional metrics
        idle_count = sum(1 for a in self.agents.values() if a.status == AgentStatus.IDLE)
        busy_count = sum(1 for a in self.agents.values() if a.status == AgentStatus.BUSY)
        failed_count = sum(1 for a in self.agents.values() if a.status == AgentStatus.FAILED)
        
        return {
            "total_agents_created": self.statistics["total_agents_created"],
            "active_agents": self.statistics["active_agents"],
            "idle_agents": idle_count,
            "busy_agents": busy_count,
            "failed_agents": failed_count,
            "total_tasks_assigned": self.statistics["total_tasks_assigned"],
            "total_tasks_completed": self.statistics["total_tasks_completed"],
            "total_tasks_failed": self.statistics["total_tasks_failed"],
            "overall_success_rate": (self.statistics["total_tasks_completed"] /
                                    self.statistics["total_tasks_assigned"]
                                    if self.statistics["total_tasks_assigned"] > 0 else 0.0),
            "registered_capabilities": len(self.capability_registry)
        }


class MetaAgentPattern:
    """
    Coordinator for meta-agent pattern.
    
    Manages agent lifecycle and coordination.
    """
    
    def __init__(self):
        self.meta_agent = MetaAgent()
    
    def create_agent(self, agent_type: AgentType, capabilities: Set[str], metadata: Dict[str, Any] = None) -> Dict[str, Any]:
        """Create agent."""
        return self.meta_agent.create_agent(agent_type, capabilities, metadata)
    
    def terminate_agent(self, agent_id: str) -> Dict[str, Any]:
        """Terminate agent."""
        return self.meta_agent.terminate_agent(agent_id)
    
    def assign_task(self, task_description: str, required_capability: str) -> Dict[str, Any]:
        """Assign task."""
        return self.meta_agent.assign_task(task_description, required_capability)
    
    def complete_task(self, task_id: str, result: Any = None, success: bool = True) -> Dict[str, Any]:
        """Complete task."""
        return self.meta_agent.complete_task(task_id, result, success)
    
    def scale_agents(self, required_capability: str, target_count: int, agent_type: AgentType = AgentType.WORKER) -> Dict[str, Any]:
        """Scale agents."""
        return self.meta_agent.scale_agents(required_capability, target_count, agent_type)
    
    def get_agent_status(self, agent_id: str) -> Dict[str, Any]:
        """Get agent status."""
        return self.meta_agent.get_agent_status(agent_id)
    
    def list_agents(self) -> Dict[str, Any]:
        """List agents."""
        return self.meta_agent.list_agents()
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get statistics."""
        return self.meta_agent.get_statistics()


# Example usage
if __name__ == "__main__":
    print("=== Meta-Agent Pattern Demo ===\n")
    
    # Create meta-agent
    pattern = MetaAgentPattern()
    
    print("1. Creating specialized agents...")
    result1 = pattern.create_agent(
        agent_type=AgentType.SPECIALIST,
        capabilities={"data_analysis", "visualization"}
    )
    analyst_id = result1["agent_id"]
    print(f"  {result1['message']}")
    
    result2 = pattern.create_agent(
        agent_type=AgentType.WORKER,
        capabilities={"data_processing", "etl"}
    )
    worker1_id = result2["agent_id"]
    print(f"  {result2['message']}")
    
    result3 = pattern.create_agent(
        agent_type=AgentType.WORKER,
        capabilities={"data_processing"}
    )
    worker2_id = result3["agent_id"]
    print(f"  {result3['message']}")
    
    # List agents
    print("\n2. Listing all agents...")
    agents = pattern.list_agents()
    print(f"  Total agents: {agents['total']}")
    for agent in agents['agents']:
        print(f"    - {agent['agent_id']}: {agent['agent_type']} ({', '.join(agent['capabilities'])})")
    
    # Assign tasks
    print("\n3. Assigning tasks...")
    task1 = pattern.assign_task("Analyze sales data", "data_analysis")
    print(f"  {task1['message']}")
    
    task2 = pattern.assign_task("Process customer data", "data_processing")
    print(f"  {task2['message']}")
    
    task3 = pattern.assign_task("Process transaction data", "data_processing")
    print(f"  {task3['message']}")
    
    # Complete tasks
    print("\n4. Completing tasks...")
    pattern.complete_task(task1["task_id"], result="Analysis complete", success=True)
    pattern.complete_task(task2["task_id"], result="Processing complete", success=True)
    pattern.complete_task(task3["task_id"], success=False)  # Simulate failure
    print("  Tasks completed")
    
    # Check agent status
    print("\n5. Checking agent status...")
    status = pattern.get_agent_status(analyst_id)
    if status['success']:
        print(f"  Agent: {status['agent_id']}")
        print(f"  Status: {status['status']}")
        print(f"  Tasks completed: {status['tasks_completed']}")
        print(f"  Tasks failed: {status['tasks_failed']}")
        print(f"  Success rate: {status['success_rate']:.1%}")
    
    # Scale agents
    print("\n6. Scaling data_processing agents to 5...")
    scale_result = pattern.scale_agents("data_processing", 5)
    print(f"  {scale_result['message']}")
    print(f"  Agents created: {scale_result['agents_created']}")
    
    # Statistics
    print("\n7. Meta-agent statistics:")
    stats = pattern.get_statistics()
    print(f"  Total agents created: {stats['total_agents_created']}")
    print(f"  Active agents: {stats['active_agents']}")
    print(f"  Idle/Busy/Failed: {stats['idle_agents']}/{stats['busy_agents']}/{stats['failed_agents']}")
    print(f"  Total tasks assigned: {stats['total_tasks_assigned']}")
    print(f"  Tasks completed: {stats['total_tasks_completed']}")
    print(f"  Tasks failed: {stats['total_tasks_failed']}")
    print(f"  Success rate: {stats['overall_success_rate']:.1%}")
    
    print("\nāœ“ Comprehensive meta-agent pattern complete!")

Agentic Architecture Pattern: Meta Agent

Pattern 281: 281 Meta Agent – Line-by-Line Explanation

Overview

Pattern 281: Meta-Agent Pattern

Purpose: Agent that creates, manages, and coordinates other agents dynamically

Code Structure Analysis

File Information

  • Pattern Number: 281
  • Pattern Name: 281 Meta Agent
  • File Path: patterns\additional_advanced\281_meta_agent.py
  • Total Lines: 550

Detailed Code Explanation

Module Docstring

"""
Pattern 281: Meta-Agent Pattern

Purpose: Agent that creates, manages, and coordinates other agents dynamically
"""

Explanation: This module docstring describes the purpose and key features of the 281 Meta Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set, Callable
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict  # First 20 imports shown

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

AgentType (Line 14)

class AgentType(Enum):
    WORKER
    SPECIALIST
    COORDINATOR
    ANALYZER
    EXECUTOR

Purpose: Type of managed agent.

Values: – WORKER: Represents worker – SPECIALIST: Represents specialist – COORDINATOR: Represents coordinator – ANALYZER: Represents analyzer – EXECUTOR: Represents executor

AgentStatus (Line 23)

class AgentStatus(Enum):
    IDLE
    BUSY
    FAILED
    TERMINATED

Purpose: Status of managed agent.

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

Classes

AgentType (Line 14)

Description: Type of managed agent.

AgentStatus (Line 23)

Description: Status of managed agent.

ManagedAgent (Line 32)

Description: Represents an agent managed by the meta-agent.

TaskAssignment (Line 46)

Description: Task assigned to an agent.

MetaAgent (Line 58)

Description: Implements meta-agent pattern.

Features: – Dynamic agent creation and termination – Agent capability management – Intelligent task routing – Load balancing across agents – Agent performance monitoring – Automatic agent spawning based on demand

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

create_agent (Line 85)
def create_agent(self, agent_type, capabilities, metadata):
    """Create a new managed agent.

Args:
    agent_type: Type of agent to create
    capabilities: Set of capabilities the agent has
    metadata: Additional metadata

Returns:
    Agent creation status and ID"""

Purpose: Create a new managed agent.

Args: agent_type: Type of agent to create capabilities: Set of capabilities the agent has metadata: Additional metadata

Returns: Agent creation status and ID

Parameters: – self: Instance reference – agent_type: Parameter for agent_type – capabilities: Parameter for capabilities – metadata: Parameter for metadata

terminate_agent (Line 128)
def terminate_agent(self, agent_id):
    """Terminate a managed agent.

Args:
    agent_id: ID of agent to terminate

Returns:
    Termination status"""

Purpose: Terminate a managed agent.

Args: agent_id: ID of agent to terminate

Returns: Termination status

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

assign_task (Line 161)
def assign_task(self, task_description, required_capability):
    """Assign task to most suitable agent.

Args:
    task_description: Description of task
    required_capability: Capability needed for task

Returns:
    Task assignment status"""

Purpose: Assign task to most suitable agent.

Args: task_description: Description of task required_capability: Capability needed for task

Returns: Task assignment status

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

complete_task (Line 230)
def complete_task(self, task_id, result, success):
    """Mark task as completed.

Args:
    task_id: Task ID
    result: Task result
    success: Whether task succeeded

Returns:
    Completion status"""

Purpose: Mark task as completed.

Args: task_id: Task ID result: Task result success: Whether task succeeded

Returns: Completion status

Parameters: – self: Instance reference – task_id: Parameter for task_id – result: Parameter for result – success: Parameter for success

scale_agents (Line 280)
def scale_agents(self, required_capability, target_count, agent_type):
    """Scale number of agents with specific capability.

Args:
    required_capability: Capability to scale for
    target_count: Target number of agents
    agent_type: Type of agents to create

Returns:
    Scaling operation status"""

Purpose: Scale number of agents with specific capability.

Args: required_capability: Capability to scale for target_count: Target number of agents agent_type: Type of agents to create

Returns: Scaling operation status

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

get_agent_status (Line 345)
def get_agent_status(self, agent_id):
    """Get detailed agent status."""

Purpose: Get detailed agent status.

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

list_agents (Line 375)
def list_agents(self, agent_type, status):
    """List agents with optional filters."""

Purpose: List agents with optional filters.

Parameters: – self: Instance reference – agent_type: Parameter for agent_type – status: Parameter for status

get_statistics (Line 402)
def get_statistics(self):
    """Get meta-agent statistics."""

Purpose: Get meta-agent statistics.

Parameters: – self: Instance reference

MetaAgentPattern (Line 425)

Description: Coordinator for meta-agent pattern.

Manages agent lifecycle and coordination.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

create_agent (Line 435)
def create_agent(self, agent_type, capabilities, metadata):
    """Create agent."""

Purpose: Create agent.

Parameters: – self: Instance reference – agent_type: Parameter for agent_type – capabilities: Parameter for capabilities – metadata: Parameter for metadata

terminate_agent (Line 439)
def terminate_agent(self, agent_id):
    """Terminate agent."""

Purpose: Terminate agent.

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

assign_task (Line 443)
def assign_task(self, task_description, required_capability):
    """Assign task."""

Purpose: Assign task.

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

complete_task (Line 447)
def complete_task(self, task_id, result, success):
    """Complete task."""

Purpose: Complete task.

Parameters: – self: Instance reference – task_id: Parameter for task_id – result: Parameter for result – success: Parameter for success

scale_agents (Line 451)
def scale_agents(self, required_capability, target_count, agent_type):
    """Scale agents."""

Purpose: Scale agents.

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

get_agent_status (Line 455)
def get_agent_status(self, agent_id):
    """Get agent status."""

Purpose: Get agent status.

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

list_agents (Line 459)
def list_agents(self):
    """List agents."""

Purpose: List agents.

Parameters: – self: Instance reference

get_statistics (Line 463)
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 = AgentType(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