"""
Pattern 141: Supervisor Pattern
Supervisor agent that coordinates worker agents, monitors their performance,
and handles failures. Implements hierarchical task delegation with LangGraph.
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any, List, Optional
from enum import Enum
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict
class WorkerStatus(Enum):
"""Status of worker agent."""
IDLE = "idle"
BUSY = "busy"
FAILED = "failed"
OFFLINE = "offline"
@dataclass
class Task:
"""Represents a task to be executed."""
task_id: str
description: str
requirements: List[str]
priority: int = 1
assigned_to: Optional[str] = None
status: str = "pending" # pending, assigned, in_progress, completed, failed
result: Optional[Any] = None
created_at: datetime = field(default_factory=datetime.now)
completed_at: Optional[datetime] = None
@dataclass
class WorkerAgent:
"""Worker agent that executes tasks."""
worker_id: str
name: str
capabilities: List[str]
status: WorkerStatus = WorkerStatus.IDLE
current_task: Optional[Task] = None
completed_tasks: int = 0
failed_tasks: int = 0
class SupervisorAgent:
"""
Supervisor that manages worker agents and delegates tasks.
"""
def __init__(self, name: str = "Supervisor", model_name: str = "gpt-4"):
self.name = name
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
self.workers: Dict[str, WorkerAgent] = {}
self.task_queue: List[Task] = []
self.completed_tasks: List[Task] = []
def register_worker(self, worker: WorkerAgent):
"""Register a worker agent."""
self.workers[worker.worker_id] = worker
def submit_task(self, task: Task):
"""Submit a new task to the supervisor."""
self.task_queue.append(task)
def assign_task(self, task: Task) -> Optional[WorkerAgent]:
"""Assign a task to the most suitable available worker."""
# Find available workers with required capabilities
suitable_workers = [
w for w in self.workers.values()
if w.status == WorkerStatus.IDLE and
any(cap in w.capabilities for cap in task.requirements)
]
if not suitable_workers:
return None
# Use LLM to select best worker
worker_info = "\n".join([
f"- {w.worker_id}: {w.name} | Capabilities: {', '.join(w.capabilities)} | "
f"Completed: {w.completed_tasks} | Failed: {w.failed_tasks}"
for w in suitable_workers
])
prompt = ChatPromptTemplate.from_messages([
("system", "You are a task assignment AI. Select the best worker for the task."),
("human", """Task: {task_description}
Requirements: {requirements}
Priority: {priority}
Available Workers:
{workers}
Select the worker_id of the best worker for this task. Reply with only the worker_id.""")
])
chain = prompt | self.llm
response = chain.invoke({
"task_description": task.description,
"requirements": ", ".join(task.requirements),
"priority": task.priority,
"workers": worker_info
})
selected_id = str(response.content).strip()
# Find the worker
selected_worker = next(
(w for w in suitable_workers if w.worker_id == selected_id),
suitable_workers[0] # Fallback to first suitable worker
)
# Assign task
task.assigned_to = selected_worker.worker_id
task.status = "assigned"
selected_worker.status = WorkerStatus.BUSY
selected_worker.current_task = task
return selected_worker
def execute_task(self, worker: WorkerAgent, task: Task) -> Dict[str, Any]:
"""Execute a task using the assigned worker."""
task.status = "in_progress"
# Simulate task execution with LLM
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {worker.name}, a worker with capabilities: {', '.join(worker.capabilities)}"),
("human", "Execute this task: {task_description}\n\nProvide the result:")
])
chain = prompt | self.llm
try:
response = chain.invoke({"task_description": task.description})
result = str(response.content)
task.status = "completed"
task.result = result
task.completed_at = datetime.now()
worker.completed_tasks += 1
return {"success": True, "result": result}
except Exception as e:
task.status = "failed"
task.result = f"Error: {str(e)}"
worker.failed_tasks += 1
worker.status = WorkerStatus.FAILED
return {"success": False, "error": str(e)}
finally:
worker.current_task = None
if worker.status != WorkerStatus.FAILED:
worker.status = WorkerStatus.IDLE
def process_queue(self) -> List[Dict[str, Any]]:
"""Process all tasks in the queue."""
results = []
while self.task_queue:
task = self.task_queue.pop(0)
# Try to assign task
worker = self.assign_task(task)
if worker:
# Execute task
result = self.execute_task(worker, task)
results.append({
"task_id": task.task_id,
"worker_id": worker.worker_id,
"status": task.status,
"result": result
})
self.completed_tasks.append(task)
else:
# No suitable worker available, put back in queue
self.task_queue.append(task)
results.append({
"task_id": task.task_id,
"status": "waiting",
"reason": "No suitable worker available"
})
break # Stop processing for now
return results
def get_worker_statistics(self) -> Dict[str, Any]:
"""Get statistics about workers."""
return {
"total_workers": len(self.workers),
"idle_workers": len([w for w in self.workers.values() if w.status == WorkerStatus.IDLE]),
"busy_workers": len([w for w in self.workers.values() if w.status == WorkerStatus.BUSY]),
"failed_workers": len([w for w in self.workers.values() if w.status == WorkerStatus.FAILED]),
"workers": [
{
"worker_id": w.worker_id,
"name": w.name,
"status": w.status.value,
"completed": w.completed_tasks,
"failed": w.failed_tasks
}
for w in self.workers.values()
]
}
def get_task_statistics(self) -> Dict[str, Any]:
"""Get statistics about tasks."""
return {
"queued": len(self.task_queue),
"completed": len(self.completed_tasks),
"success_rate": (
len([t for t in self.completed_tasks if t.status == "completed"]) /
len(self.completed_tasks) * 100
if self.completed_tasks else 0
)
}
class SupervisorPattern:
"""
LangGraph-based supervisor pattern with state management.
"""
def __init__(self, model_name: str = "gpt-4"):
self.supervisor = SupervisorAgent(model_name=model_name)
self.workflow = self._create_workflow()
def _create_workflow(self) -> StateGraph:
"""Create LangGraph workflow for supervision."""
class WorkflowState(TypedDict):
tasks: List[Dict[str, Any]]
results: List[Dict[str, Any]]
current_task: Optional[Dict[str, Any]]
workflow = StateGraph(WorkflowState)
def receive_tasks(state: WorkflowState) -> WorkflowState:
"""Receive and queue tasks."""
return state
def assign_task(state: WorkflowState) -> WorkflowState:
"""Assign next task to worker."""
if state["tasks"]:
state["current_task"] = state["tasks"][0]
return state
def execute_task(state: WorkflowState) -> WorkflowState:
"""Execute the current task."""
if state["current_task"]:
# Process through supervisor
results = self.supervisor.process_queue()
state["results"].extend(results)
state["tasks"] = state["tasks"][1:] # Remove completed task
return state
def check_queue(state: WorkflowState) -> str:
"""Check if more tasks to process."""
return "assign" if state["tasks"] else "end"
workflow.add_node("receive", receive_tasks)
workflow.add_node("assign", assign_task)
workflow.add_node("execute", execute_task)
workflow.set_entry_point("receive")
workflow.add_edge("receive", "assign")
workflow.add_edge("assign", "execute")
workflow.add_conditional_edges("execute", check_queue, {
"assign": "assign",
"end": END
})
return workflow.compile()
def run(self, tasks: List[Task]) -> Dict[str, Any]:
"""Run the supervisor pattern."""
# Submit tasks to supervisor
for task in tasks:
self.supervisor.submit_task(task)
# Execute workflow
initial_state = {
"tasks": [{"task_id": t.task_id, "description": t.description} for t in tasks],
"results": [],
"current_task": None
}
final_state = self.workflow.invoke(initial_state)
return {
"supervisor": self.supervisor.name,
"tasks_processed": len(tasks),
"results": final_state["results"],
"worker_stats": self.supervisor.get_worker_statistics(),
"task_stats": self.supervisor.get_task_statistics()
}
def main():
"""Example usage of Supervisor Pattern."""
print("=" * 80)
print("Supervisor Pattern - Task Delegation Demo")
print("=" * 80)
# Create supervisor pattern
pattern = SupervisorPattern()
supervisor = pattern.supervisor
# Create worker agents
workers = [
WorkerAgent(
worker_id="worker-1",
name="Data Analyst",
capabilities=["data_analysis", "statistics", "visualization"]
),
WorkerAgent(
worker_id="worker-2",
name="Software Developer",
capabilities=["coding", "debugging", "testing"]
),
WorkerAgent(
worker_id="worker-3",
name="Content Writer",
capabilities=["writing", "editing", "research"]
),
WorkerAgent(
worker_id="worker-4",
name="Full Stack Engineer",
capabilities=["coding", "data_analysis", "deployment"]
)
]
# Register workers
for worker in workers:
supervisor.register_worker(worker)
# Create tasks
tasks = [
Task(
task_id="task-1",
description="Analyze customer data and create visualization",
requirements=["data_analysis", "visualization"],
priority=3
),
Task(
task_id="task-2",
description="Write documentation for API endpoints",
requirements=["writing", "research"],
priority=2
),
Task(
task_id="task-3",
description="Debug authentication module",
requirements=["coding", "debugging"],
priority=5
),
Task(
task_id="task-4",
description="Create statistical report on sales trends",
requirements=["statistics", "data_analysis"],
priority=4
),
Task(
task_id="task-5",
description="Implement new feature with tests",
requirements=["coding", "testing"],
priority=3
)
]
# Run supervisor pattern
print("\nš Submitting tasks...")
for task in tasks:
print(f" - {task.task_id}: {task.description} (Priority: {task.priority})")
result = pattern.run(tasks)
# Display results
print("\n" + "=" * 80)
print("ā
Execution Results")
print("=" * 80)
for task_result in result["results"]:
print(f"\nTask: {task_result['task_id']}")
print(f" Status: {task_result['status']}")
if "worker_id" in task_result:
print(f" Worker: {task_result['worker_id']}")
if "result" in task_result:
print(f" Result: {str(task_result['result'])[:100]}...")
# Display statistics
print("\n" + "=" * 80)
print("š Worker Statistics")
print("=" * 80)
worker_stats = result["worker_stats"]
print(f"Total Workers: {worker_stats['total_workers']}")
print(f"Idle: {worker_stats['idle_workers']}")
print(f"Busy: {worker_stats['busy_workers']}")
print(f"Failed: {worker_stats['failed_workers']}")
print("\nWorker Details:")
for worker in worker_stats["workers"]:
print(f" {worker['name']} ({worker['worker_id']})")
print(f" Status: {worker['status']}")
print(f" Completed: {worker['completed']} | Failed: {worker['failed']}")
print("\n" + "=" * 80)
print("š Task Statistics")
print("=" * 80)
task_stats = result["task_stats"]
print(f"Queued: {task_stats['queued']}")
print(f"Completed: {task_stats['completed']}")
print(f"Success Rate: {task_stats['success_rate']:.1f}%")
if __name__ == "__main__":
main()
Agentic Architecture Pattern: Supervisor
Pattern 141: 141 Supervisor Pattern – Line-by-Line Explanation
Overview
Pattern 141: Supervisor Pattern
A supervisor agent oversees and coordinates multiple worker agents, deciding which agent should handle each task and managing the workflow.
Code Structure Analysis
File Information
- Pattern Number: 141
- Pattern Name: 141 Supervisor Pattern
- File Path:
patterns\workflow\141_supervisor_pattern.py - Total Lines: 227
Detailed Code Explanation
Module Docstring
"""
Pattern 141: Supervisor Pattern
A supervisor agent oversees and coordinates multiple worker agents,
deciding which agent should handle each task and managing the workflow.
"""
Explanation: This module docstring describes the purpose and key features of the 141 Supervisor Pattern 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_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict # 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_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
– from langgraph.graph import StateGraph, END: LangGraph for workflow orchestration
– from typing_extensions import TypedDict: Type hints for better code documentation
Enumerations
WorkerStatus (Line 18)
class WorkerStatus(Enum):
IDLE
BUSY
OFFLINE
Purpose: Worker status.
Values:
– IDLE: Not currently active
– BUSY: Represents busy
– OFFLINE: Represents offline
Classes
WorkerStatus (Line 18)
Description: Worker status.
WorkerAgent (Line 26)
Description: Worker agent managed by supervisor.
Task (Line 36)
Description: Task to be executed.
SupervisorState (Line 47)
Description: State for supervisor workflow.
SupervisorPattern (Line 55)
Description: Supervisor Pattern – Coordinates multiple worker agents.
Methods:
__init__ (Line 60)
def __init__(self, model_name):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– model_name: LLM model to use (e.g., “gpt-4”)
add_worker (Line 66)
def add_worker(self, worker):
"""Add a worker to the team."""
Purpose: Add a worker to the team.
Parameters:
– self: Instance reference
– worker: Parameter for worker
_select_worker (Line 70)
def _select_worker(self, state):
"""Supervisor selects appropriate worker."""
Purpose: Supervisor selects appropriate worker.
Parameters:
– self: Instance reference
– state: Current workflow state
_execute_task (Line 95)
def _execute_task(self, state):
"""Worker executes the task."""
Purpose: Worker executes the task.
Parameters:
– self: Instance reference
– state: Current workflow state
run (Line 122)
def run(self, task_description):
"""Execute task through supervisor workflow."""
Purpose: Execute task through supervisor workflow.
Parameters:
– self: Instance reference
– task_description: Parameter for task_description
get_statistics (Line 167)
def get_statistics(self):
"""Get supervisor statistics."""
Purpose: Get supervisor 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 = WorkerStatus(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
Leave a Reply