aishastr.live

Agentic Architecture Pattern: Delegate Agent

Written by

admin

in


    """
Pattern 90: Delegate Agent Pattern

Delegates work to specialized worker agents based on task requirements.
Features task routing, load balancing, and result aggregation.
"""

from typing import Dict, Any, List, Optional, Tuple
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 TaskPriority(Enum):
    """Task priority levels."""
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4


@dataclass
class Task:
    """Represents a task to be delegated."""
    task_id: str
    description: str
    required_skills: List[str]
    priority: TaskPriority = TaskPriority.MEDIUM
    params: Dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class WorkerAgent:
    """Worker agent that can execute tasks."""
    name: str
    skills: List[str]
    capacity: int = 5
    current_load: int = 0
    tasks_completed: int = 0
    
    def can_accept(self) -> bool:
        """Check if worker can accept more tasks."""
        return self.current_load  bool:
        """Check if worker has required skills."""
        return all(skill in self.skills for skill in required_skills)
    
    def match_score(self, required_skills: List[str]) -> float:
        """Calculate match score (0.0 to 1.0)."""
        if not required_skills:
            return 1.0
        matching = sum(1 for skill in required_skills if skill in self.skills)
        return matching / len(required_skills)


@dataclass
class DelegationRecord:
    """Records a delegation event."""
    task: Task
    worker: str
    start_time: datetime = field(default_factory=datetime.now)
    end_time: Optional[datetime] = None
    result: Optional[str] = None
    success: bool = False


class DelegateAgent:
    """Agent that delegates tasks to workers."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.workers: Dict[str, WorkerAgent] = {}
        self.delegation_history: List[DelegationRecord] = []
        self.pending_tasks: List[Task] = []
    
    def register_worker(self, worker: WorkerAgent):
        """Register a worker agent."""
        self.workers[worker.name] = worker
        print(f"   {self.name}: Registered worker '{worker.name}' with skills: {', '.join(worker.skills)}")
    
    def find_best_worker(self, task: Task) -> Optional[WorkerAgent]:
        """Find the best worker for a task."""
        candidates = []
        
        for worker in self.workers.values():
            if worker.can_accept() and worker.has_skills(task.required_skills):
                score = worker.match_score(task.required_skills)
                # Consider load in scoring
                load_factor = 1.0 - (worker.current_load / worker.capacity)
                final_score = score * 0.7 + load_factor * 0.3
                candidates.append((worker, final_score))
        
        if not candidates:
            return None
        
        # Sort by score (higher is better)
        candidates.sort(key=lambda x: x[1], reverse=True)
        return candidates[0][0]
    
    def delegate_task(self, task: Task) -> Optional[str]:
        """Delegate task to best available worker."""
        print(f"\n   {self.name}: Delegating task '{task.task_id}'")
        print(f"   Description: {task.description}")
        print(f"   Required skills: {', '.join(task.required_skills)}")
        print(f"   Priority: {task.priority.name}")
        
        # Find best worker
        worker = self.find_best_worker(task)
        
        if not worker:
            print(f"   {self.name}: No suitable worker found")
            self.pending_tasks.append(task)
            return None
        
        print(f"   {self.name}: Assigned to '{worker.name}' (match: {worker.match_score(task.required_skills):.2f})")
        
        # Update worker load
        worker.current_load += 1
        
        # Execute task via worker
        result = self._execute_task_via_worker(task, worker)
        
        # Update worker
        worker.current_load -= 1
        worker.tasks_completed += 1
        
        # Record delegation
        record = DelegationRecord(
            task=task,
            worker=worker.name,
            end_time=datetime.now(),
            result=result,
            success=True
        )
        self.delegation_history.append(record)
        
        print(f"   {self.name}: Task completed by '{worker.name}'")
        return result
    
    def _execute_task_via_worker(self, task: Task, worker: WorkerAgent) -> str:
        """Execute task through worker agent."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {worker.name}, a worker agent with skills: {', '.join(worker.skills)}. "
                      f"Execute the delegated task."),
            ("human", "Task: {description}\nParameters: {params}\n\nExecute this task.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({
            "description": task.description,
            "params": str(task.params)
        })
        
        return response.content
    
    def process_pending_tasks(self):
        """Try to process pending tasks."""
        if not self.pending_tasks:
            return
        
        print(f"\n   {self.name}: Processing {len(self.pending_tasks)} pending tasks...")
        
        processed = []
        for task in self.pending_tasks:
            result = self.delegate_task(task)
            if result:
                processed.append(task)
        
        # Remove processed tasks
        for task in processed:
            self.pending_tasks.remove(task)
    
    def get_statistics(self) -> Dict[str, Any]:
        """Get delegation statistics."""
        total_delegations = len(self.delegation_history)
        successful = sum(1 for d in self.delegation_history if d.success)
        
        worker_stats = {
            name: {
                "tasks_completed": worker.tasks_completed,
                "current_load": worker.current_load,
                "utilization": f"{worker.current_load / worker.capacity * 100:.1f}%"
            }
            for name, worker in self.workers.items()
        }
        
        return {
            "total_delegations": total_delegations,
            "successful_delegations": successful,
            "pending_tasks": len(self.pending_tasks),
            "worker_stats": worker_stats
        }


class DelegateAgentPattern:
    """
    Delegate Agent Pattern - Task delegation and routing.
    
    Delegate agents route tasks to specialized workers based on
    skills, availability, and load balancing.
    """
    
    def __init__(self):
        self.delegates: Dict[str, DelegateAgent] = {}
    
    def register_delegate(self, delegate: DelegateAgent):
        """Register delegate agent."""
        self.delegates[delegate.name] = delegate
    
    def run(self, tasks: List[Task]) -> Dict[str, Any]:
        """Execute delegation scenarios."""
        print("=== Delegate Agent Pattern Demo ===\n")
        
        for delegate in self.delegates.values():
            for task in tasks:
                delegate.delegate_task(task)
            
            # Try pending tasks
            delegate.process_pending_tasks()
        
        total_delegations = sum(
            len(d.delegation_history)
            for d in self.delegates.values()
        )
        
        return {
            "pattern": "DelegateAgent",
            "total_delegates": len(self.delegates),
            "total_delegations": total_delegations,
            "tasks_processed": len(tasks)
        }


# Example usage
if __name__ == "__main__":
    print("=== Delegate Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = DelegateAgentPattern()
    
    # Create delegate agent
    print("1. Creating delegate agent...")
    delegate = DelegateAgent("TaskCoordinator")
    
    # Register workers
    print("\n2. Registering worker agents...")
    
    delegate.register_worker(WorkerAgent(
        name="DataWorker",
        skills=["data_processing", "analysis", "sql"],
        capacity=3
    ))
    
    delegate.register_worker(WorkerAgent(
        name="WebWorker",
        skills=["web_scraping", "api_calls", "http"],
        capacity=5
    ))
    
    delegate.register_worker(WorkerAgent(
        name="MLWorker",
        skills=["machine_learning", "training", "prediction"],
        capacity=2
    ))
    
    delegate.register_worker(WorkerAgent(
        name="GeneralWorker",
        skills=["data_processing", "web_scraping", "text_processing"],
        capacity=4
    ))
    
    pattern.register_delegate(delegate)
    
    # Create tasks
    print("\n3. Creating tasks to delegate...")
    
    tasks = [
        Task(
            task_id="T001",
            description="Analyze sales data and generate report",
            required_skills=["data_processing", "analysis"],
            priority=TaskPriority.HIGH
        ),
        Task(
            task_id="T002",
            description="Scrape product information from website",
            required_skills=["web_scraping"],
            priority=TaskPriority.MEDIUM
        ),
        Task(
            task_id="T003",
            description="Train predictive model on customer data",
            required_skills=["machine_learning", "training"],
            priority=TaskPriority.HIGH
        ),
        Task(
            task_id="T004",
            description="Extract data from SQL database",
            required_skills=["sql", "data_processing"],
            priority=TaskPriority.LOW
        ),
        Task(
            task_id="T005",
            description="Process text documents",
            required_skills=["text_processing"],
            priority=TaskPriority.MEDIUM
        )
    ]
    
    # Execute
    print("\n4. Delegating tasks...")
    summary = pattern.run(tasks)
    
    print(f"\n5. Summary:")
    print(f"   Total Delegations: {summary['total_delegations']}")
    print(f"   Tasks Processed: {summary['tasks_processed']}")
    
    print("\n6. Delegation Statistics:")
    stats = delegate.get_statistics()
    print(f"   Successful Delegations: {stats['successful_delegations']}")
    print(f"   Pending Tasks: {stats['pending_tasks']}")
    
    print("\n7. Worker Performance:")
    for worker_name, worker_stat in stats['worker_stats'].items():
        print(f"   {worker_name}:")
        print(f"     Tasks Completed: {worker_stat['tasks_completed']}")
        print(f"     Current Load: {worker_stat['current_load']}")
        print(f"     Utilization: {worker_stat['utilization']}")
    
    print("\nāœ“ Delegate agent successfully routed tasks to workers!")

Agentic Architecture Pattern: Delegate Agent

Pattern 90: 90 Delegate Agent – Line-by-Line Explanation

Overview

Pattern 90: Delegate Agent Pattern

Delegates work to specialized worker agents based on task requirements. Features task routing, load balancing, and result aggregation.

Code Structure Analysis

File Information

  • Pattern Number: 90
  • Pattern Name: 90 Delegate Agent
  • File Path: patterns\specialized_agent\90_delegate_agent.py
  • Total Lines: 336

Detailed Code Explanation

Module Docstring

"""
Pattern 90: Delegate Agent Pattern

Delegates work to specialized worker agents based on task requirements.
Features task routing, load balancing, and result aggregation.
"""

Explanation: This module docstring describes the purpose and key features of the 90 Delegate Agent pattern.

Imports Section

from typing import Dict, Any, List, Optional, Tuple
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, Tuple: 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

TaskPriority (Line 16)

class TaskPriority(Enum):
    LOW
    MEDIUM
    HIGH
    CRITICAL

Purpose: Task priority levels.

Values: – LOW: Represents low – MEDIUM: Represents medium – HIGH: Represents high – CRITICAL: Represents critical

Classes

TaskPriority (Line 16)

Description: Task priority levels.

Task (Line 25)

Description: Represents a task to be delegated.

WorkerAgent (Line 36)

Description: Worker agent that can execute tasks.

Methods:

can_accept (Line 44)
def can_accept(self):
    """Check if worker can accept more tasks."""

Purpose: Check if worker can accept more tasks.

Parameters: – self: Instance reference

has_skills (Line 48)
def has_skills(self, required_skills):
    """Check if worker has required skills."""

Purpose: Check if worker has required skills.

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

match_score (Line 52)
def match_score(self, required_skills):
    """Calculate match score (0.0 to 1.0)."""

Purpose: Calculate match score (0.0 to 1.0).

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

DelegationRecord (Line 61)

Description: Records a delegation event.

DelegateAgent (Line 71)

Description: Agent that delegates tasks to workers.

Methods:

__init__ (Line 74)
def __init__(self, name, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – model_name: LLM model to use (e.g., “gpt-4”)

register_worker (Line 81)
def register_worker(self, worker):
    """Register a worker agent."""

Purpose: Register a worker agent.

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

find_best_worker (Line 86)
def find_best_worker(self, task):
    """Find the best worker for a task."""

Purpose: Find the best worker for a task.

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

delegate_task (Line 105)
def delegate_task(self, task):
    """Delegate task to best available worker."""

Purpose: Delegate task to best available worker.

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

_execute_task_via_worker (Line 145)
def _execute_task_via_worker(self, task, worker):
    """Execute task through worker agent."""

Purpose: Execute task through worker agent.

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

process_pending_tasks (Line 161)
def process_pending_tasks(self):
    """Try to process pending tasks."""

Purpose: Try to process pending tasks.

Parameters: – self: Instance reference

get_statistics (Line 178)
def get_statistics(self):
    """Get delegation statistics."""

Purpose: Get delegation statistics.

Parameters: – self: Instance reference

DelegateAgentPattern (Line 200)

Description: Delegate Agent Pattern – Task delegation and routing.

Delegate agents route tasks to specialized workers based on skills, availability, and load balancing.

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_delegate (Line 211)
def register_delegate(self, delegate):
    """Register delegate agent."""

Purpose: Register delegate agent.

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

run (Line 215)
def run(self, tasks):
    """Execute delegation scenarios."""

Purpose: Execute delegation scenarios.

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

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