aishastr.live

Agentic Architecture Pattern: Task Agent

Written by

admin

in


    """
Pattern 83: Task Agent Pattern
Purpose: Specialized agent for executing specific tasks
"""

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 TaskStatus(Enum):
    """Task execution status."""
    PENDING = "pending"
    ASSIGNED = "assigned"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class Task:
    """A task to be executed."""
    id: str
    description: str
    priority: int = 1
    status: TaskStatus = TaskStatus.PENDING
    assigned_to: Optional[str] = None
    result: Optional[str] = None
    created_at: datetime = field(default_factory=datetime.now)
    completed_at: Optional[datetime] = None


class TaskAgent:
    """
    Task Agent - Executes specific tasks
    
    Specialized agent responsible for task execution.
    Can handle multiple tasks with priorities.
    """
    
    def __init__(self, name: str, capabilities: List[str], model_name: str = "gpt-4"):
        self.name = name
        self.capabilities = capabilities
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.assigned_tasks: Dict[str, Task] = {}
        self.completed_tasks: List[Task] = []
        
    def can_handle(self, task: Task) -> bool:
        """Check if agent can handle this task."""
        # Simple keyword matching
        task_lower = task.description.lower()
        return any(cap.lower() in task_lower for cap in self.capabilities)
    
    def assign_task(self, task: Task) -> bool:
        """Assign task to this agent."""
        if not self.can_handle(task):
            return False
            
        task.status = TaskStatus.ASSIGNED
        task.assigned_to = self.name
        self.assigned_tasks[task.id] = task
        print(f"   {self.name}: Assigned task '{task.id}' - {task.description}")
        return True
    
    def execute_task(self, task_id: str) -> Dict[str, Any]:
        """Execute assigned task."""
        if task_id not in self.assigned_tasks:
            return {"error": "Task not found"}
        
        task = self.assigned_tasks[task_id]
        task.status = TaskStatus.IN_PROGRESS
        
        # Execute using LLM
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a task agent with capabilities: {', '.join(self.capabilities)}"),
            ("human", "Execute this task: {task_description}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task_description": task.description})
        
        task.status = TaskStatus.COMPLETED
        task.result = str(response.content)
        task.completed_at = datetime.now()
        
        # Move to completed
        del self.assigned_tasks[task_id]
        self.completed_tasks.append(task)
        
        return {
            "task_id": task_id,
            "agent": self.name,
            "status": task.status.value,
            "result": task.result[:100] + "..." if len(task.result) > 100 else task.result
        }
    
    def execute_all(self) -> List[Dict[str, Any]]:
        """Execute all assigned tasks."""
        results = []
        task_ids = list(self.assigned_tasks.keys())
        
        for task_id in task_ids:
            result = self.execute_task(task_id)
            results.append(result)
        
        return results


class TaskAgentPattern:
    """
    Task Agent Pattern - Manages task agents and task distribution
    """
    
    def __init__(self):
        self.agents: Dict[str, TaskAgent] = {}
        self.tasks: Dict[str, Task] = {}
        
    def register_agent(self, agent: TaskAgent):
        """Register a task agent."""
        self.agents[agent.name] = agent
        print(f"✓ Registered agent: {agent.name} (capabilities: {', '.join(agent.capabilities)})")
    
    def create_task(self, task: Task):
        """Create and queue a task."""
        self.tasks[task.id] = task
    
    def distribute_tasks(self) -> Dict[str, Any]:
        """Distribute tasks to capable agents."""
        assigned = 0
        unassigned = []
        
        for task_id, task in list(self.tasks.items()):
            if task.status == TaskStatus.PENDING:
                # Find capable agent
                for agent in self.agents.values():
                    if agent.assign_task(task):
                        del self.tasks[task_id]
                        assigned += 1
                        break
                else:
                    unassigned.append(task_id)
        
        return {
            "assigned": assigned,
            "unassigned": len(unassigned),
            "unassigned_tasks": unassigned
        }
    
    def execute_all_tasks(self) -> Dict[str, Any]:
        """Execute all tasks across all agents."""
        all_results = []
        
        for agent in self.agents.values():
            results = agent.execute_all()
            all_results.extend(results)
        
        return {
            "pattern": "TaskAgent",
            "total_executed": len(all_results),
            "results": all_results
        }
    
    def run(self, tasks: List[Task]) -> Dict[str, Any]:
        """Main execution: create, distribute, and execute tasks."""
        # Create tasks
        for task in tasks:
            self.create_task(task)
        
        # Distribute
        distribution = self.distribute_tasks()
        print(f"\nDistribution: {distribution['assigned']} assigned, {distribution['unassigned']} unassigned")
        
        # Execute
        print("\nExecuting tasks...\n")
        execution = self.execute_all_tasks()
        
        return {
            "pattern": "TaskAgent",
            "distribution": distribution,
            "execution": execution
        }


if __name__ == "__main__":
    print("=== Task Agent Pattern Demo ===\n")
    
    pattern = TaskAgentPattern()
    
    # Register specialized agents
    pattern.register_agent(TaskAgent("DataProcessor", ["data", "process", "analyze"]))
    pattern.register_agent(TaskAgent("ReportGenerator", ["report", "generate", "document"]))
    pattern.register_agent(TaskAgent("Notifier", ["notify", "alert", "send"]))
    
    # Create tasks
    tasks = [
        Task("t1", "Process customer data and extract insights", priority=1),
        Task("t2", "Generate monthly sales report", priority=2),
        Task("t3", "Send notification to stakeholders", priority=1),
        Task("t4", "Analyze market trends data", priority=2),
    ]
    
    print("\nTasks created:")
    for task in tasks:
        print(f"  {task.id}: {task.description}")
    
    # Run pattern
    summary = pattern.run(tasks)
    
    print("\n" + "="*70)
    print(f"Completed {summary['execution']['total_executed']} tasks!")
    
    print("\nTask Results:")
    for result in summary['execution']['results'][:2]:
        print(f"  {result['task_id']} ({result['agent']}): {result['result']}")

Agentic Architecture Pattern: Task Agent

Pattern 83: 83 Task Agent – Line-by-Line Explanation

Overview

Pattern 83: Task Agent Pattern Purpose: Specialized agent for executing specific tasks

Code Structure Analysis

File Information

  • Pattern Number: 83
  • Pattern Name: 83 Task Agent
  • File Path: patterns\specialized_agent\83_task_agent.py
  • Total Lines: 218

Detailed Code Explanation

Module Docstring

"""
Pattern 83: Task Agent Pattern
Purpose: Specialized agent for executing specific tasks
"""

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

TaskStatus (Line 14)

class TaskStatus(Enum):
    PENDING
    ASSIGNED
    IN_PROGRESS
    COMPLETED
    FAILED

Purpose: Task execution status.

Values: – PENDING: Initial state, not yet processed – ASSIGNED: Represents assigned – IN_PROGRESS: Represents in progress – COMPLETED: Successfully finished – FAILED: Execution failed

Classes

TaskStatus (Line 14)

Description: Task execution status.

Task (Line 24)

Description: A task to be executed.

TaskAgent (Line 36)

Description: Task Agent – Executes specific tasks

Specialized agent responsible for task execution. Can handle multiple tasks with priorities.

Methods:

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

Purpose: No method docstring.

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

can_handle (Line 51)
def can_handle(self, task):
    """Check if agent can handle this task."""

Purpose: Check if agent can handle this task.

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

assign_task (Line 57)
def assign_task(self, task):
    """Assign task to this agent."""

Purpose: Assign task to this agent.

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

execute_task (Line 68)
def execute_task(self, task_id):
    """Execute assigned task."""

Purpose: Execute assigned task.

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

execute_all (Line 100)
def execute_all(self):
    """Execute all assigned tasks."""

Purpose: Execute all assigned tasks.

Parameters: – self: Instance reference

TaskAgentPattern (Line 112)

Description: Task Agent Pattern – Manages task agents and task distribution

Methods:

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

Purpose: No method docstring.

Parameters: – self: Instance reference

register_agent (Line 121)
def register_agent(self, agent):
    """Register a task agent."""

Purpose: Register a task agent.

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

create_task (Line 126)
def create_task(self, task):
    """Create and queue a task."""

Purpose: Create and queue a task.

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

distribute_tasks (Line 130)
def distribute_tasks(self):
    """Distribute tasks to capable agents."""

Purpose: Distribute tasks to capable agents.

Parameters: – self: Instance reference

execute_all_tasks (Line 152)
def execute_all_tasks(self):
    """Execute all tasks across all agents."""

Purpose: Execute all tasks across all agents.

Parameters: – self: Instance reference

run (Line 166)
def run(self, tasks):
    """Main execution: create, distribute, and execute tasks."""

Purpose: Main execution: create, distribute, and execute tasks.

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