"""
Pattern 30: Coordination Pattern
Agents coordinate activities, resources, and timing to achieve shared goals.
Manages dependencies, synchronization, and conflict resolution.
"""
from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime, timedelta
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
class TaskStatus(Enum):
"""Status of coordinated task."""
PENDING = "pending"
READY = "ready"
IN_PROGRESS = "in_progress"
BLOCKED = "blocked"
COMPLETED = "completed"
FAILED = "failed"
class ResourceStatus(Enum):
"""Status of shared resource."""
AVAILABLE = "available"
ALLOCATED = "allocated"
IN_USE = "in_use"
MAINTENANCE = "maintenance"
@dataclass
class CoordinatedTask:
"""Task requiring coordination."""
task_id: str
name: str
assigned_to: Optional[str] = None
dependencies: Set[str] = field(default_factory=set)
required_resources: Set[str] = field(default_factory=set)
status: TaskStatus = TaskStatus.PENDING
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
estimated_duration: timedelta = field(default_factory=lambda: timedelta(hours=1))
@dataclass
class SharedResource:
"""Shared resource requiring coordination."""
resource_id: str
name: str
capacity: int
current_users: Set[str] = field(default_factory=set)
status: ResourceStatus = ResourceStatus.AVAILABLE
def is_available(self) -> bool:
"""Check if resource is available."""
return self.status == ResourceStatus.AVAILABLE and len(self.current_users) str:
"""Request to execute a task."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.name}, evaluating if you can execute a task."),
("human", f"Task: {task.name}\n"
f"Dependencies: {len(task.dependencies)} tasks\n"
f"Required resources: {task.required_resources}\n"
f"Estimated duration: {task.estimated_duration}\n\n"
f"Can you execute this task? Respond 'yes' or 'no' with brief reason.")
])
chain = prompt | self.llm
response = chain.invoke({})
return str(response.content)[:150]
def start_task(self, task: CoordinatedTask):
"""Start executing task."""
self.assigned_tasks.add(task.task_id)
task.status = TaskStatus.IN_PROGRESS
task.started_at = datetime.now()
print(f" ▶️ {self.name} started: {task.name}")
def complete_task(self, task: CoordinatedTask):
"""Complete task execution."""
task.status = TaskStatus.COMPLETED
task.completed_at = datetime.now()
self.assigned_tasks.discard(task.task_id)
print(f" ✅ {self.name} completed: {task.name}")
class CoordinationPattern:
"""
Coordination Pattern - Multi-agent activity coordination.
Agents coordinate tasks, resources, and timing through a central
coordinator that manages dependencies and resolves conflicts.
"""
def __init__(self):
self.agents: Dict[str, CoordinatedAgent] = {}
self.tasks: Dict[str, CoordinatedTask] = {}
self.resources: Dict[str, SharedResource] = {}
self.workflow: Optional[StateGraph] = None
self._build_workflow()
def _build_workflow(self):
"""Build LangGraph coordination workflow."""
workflow = StateGraph(CoordinationState)
def check_dependencies(state: CoordinationState) -> CoordinationState:
"""Check task dependencies."""
state["current_action"] = "checking_dependencies"
return state
def allocate_resources(state: CoordinationState) -> CoordinationState:
"""Allocate resources to tasks."""
state["current_action"] = "allocating_resources"
return state
def execute_tasks(state: CoordinationState) -> CoordinationState:
"""Execute ready tasks."""
state["current_action"] = "executing_tasks"
return state
# Add nodes
workflow.add_node("check_deps", check_dependencies)
workflow.add_node("allocate", allocate_resources)
workflow.add_node("execute", execute_tasks)
# Add edges
workflow.set_entry_point("check_deps")
workflow.add_edge("check_deps", "allocate")
workflow.add_edge("allocate", "execute")
workflow.add_edge("execute", END)
self.workflow = workflow.compile()
def register_agent(self, agent: CoordinatedAgent):
"""Register agent for coordination."""
self.agents[agent.name] = agent
def add_task(self, task: CoordinatedTask):
"""Add task to coordination."""
self.tasks[task.task_id] = task
def add_resource(self, resource: SharedResource):
"""Add shared resource."""
self.resources[resource.resource_id] = resource
def coordinate_execution(self):
"""Coordinate task execution."""
print("\n 🔄 Coordinating task execution...")
# Check dependencies and make tasks ready
for task_id, task in self.tasks.items():
if task.status == TaskStatus.PENDING:
deps_met = all(
self.tasks.get(dep_id, CoordinatedTask("", "")).status == TaskStatus.COMPLETED
for dep_id in task.dependencies
)
if deps_met:
task.status = TaskStatus.READY
print(f" ✓ {task.name} is ready (dependencies met)")
# Allocate resources
for task_id, task in self.tasks.items():
if task.status == TaskStatus.READY:
resources_available = all(
self.resources.get(res_id, SharedResource("", "", 0)).is_available()
for res_id in task.required_resources
)
if resources_available:
# Allocate resources
for res_id in task.required_resources:
resource = self.resources[res_id]
resource.status = ResourceStatus.ALLOCATED
# Assign to agent
assigned = self._assign_task(task)
if assigned:
print(f" ✓ {task.name} assigned to {assigned}")
else:
task.status = TaskStatus.BLOCKED
print(f" ⚠️ {task.name} blocked (resources unavailable)")
def _assign_task(self, task: CoordinatedTask) -> Optional[str]:
"""Assign task to available agent."""
for agent_name, agent in self.agents.items():
if len(agent.assigned_tasks) Dict[str, Any]:
"""Execute coordination pattern."""
completed = sum(1 for t in self.tasks.values()
if t.status == TaskStatus.COMPLETED)
in_progress = sum(1 for t in self.tasks.values()
if t.status == TaskStatus.IN_PROGRESS)
blocked = sum(1 for t in self.tasks.values()
if t.status == TaskStatus.BLOCKED)
return {
"pattern": "Coordination",
"scenario": scenario,
"total_agents": len(self.agents),
"total_tasks": len(self.tasks),
"completed_tasks": completed,
"in_progress": in_progress,
"blocked_tasks": blocked,
"total_resources": len(self.resources)
}
# Example usage
if __name__ == "__main__":
print("=== Coordination Pattern Demo ===\n")
# Create coordination pattern
print("1. Creating coordination system...")
pattern = CoordinationPattern()
# Register agents
agents = [
CoordinatedAgent("Alice"),
CoordinatedAgent("Bob"),
CoordinatedAgent("Carol")
]
for agent in agents:
pattern.register_agent(agent)
# Add shared resources
print("\n2. Adding shared resources...")
pattern.add_resource(SharedResource("db1", "Database", capacity=2))
pattern.add_resource(SharedResource("api1", "API Server", capacity=3))
pattern.add_resource(SharedResource("gpu1", "GPU Cluster", capacity=1))
# Add tasks with dependencies
print("\n3. Adding tasks with dependencies...")
task1 = CoordinatedTask("t1", "Data Collection",
required_resources={"db1"})
task2 = CoordinatedTask("t2", "Data Processing",
dependencies={"t1"},
required_resources={"gpu1"})
task3 = CoordinatedTask("t3", "Model Training",
dependencies={"t2"},
required_resources={"gpu1"})
task4 = CoordinatedTask("t4", "API Development",
required_resources={"api1"})
task5 = CoordinatedTask("t5", "Testing",
dependencies={"t3", "t4"},
required_resources={"api1", "db1"})
for task in [task1, task2, task3, task4, task5]:
pattern.add_task(task)
# Round 1: Coordinate execution
print("\n4. Coordination Round 1:")
pattern.coordinate_execution()
# Simulate completion of task1
print("\n5. Simulating task completions...")
pattern.complete_task_coordination("t1")
pattern.complete_task_coordination("t4")
# Round 2: More tasks should be ready
print("\n6. Coordination Round 2:")
pattern.coordinate_execution()
# Complete more tasks
pattern.complete_task_coordination("t2")
# Round 3: Final tasks
print("\n7. Coordination Round 3:")
pattern.coordinate_execution()
# Statistics
print("\n8. Coordination Statistics:")
summary = pattern.run("Software Development Project")
print(f" Total Tasks: {summary['total_tasks']}")
print(f" Completed: {summary['completed_tasks']}")
print(f" In Progress: {summary['in_progress']}")
print(f" Blocked: {summary['blocked_tasks']}")
print(f" Resources: {summary['total_resources']}")
# Task timeline
print("\n9. Task Status:")
for task_id, task in pattern.tasks.items():
status_icon = {"completed": "✅", "in_progress": "▶️",
"blocked": "⚠️", "pending": "⏸️", "ready": "🟢"}.get(task.status.value, "❓")
deps = f" (deps: {len(task.dependencies)})" if task.dependencies else ""
assigned = f" -> {task.assigned_to}" if task.assigned_to else ""
print(f" {status_icon} {task.name}: {task.status.value}{deps}{assigned}")
# Resource utilization
print("\n10. Resource Utilization:")
for res_id, resource in pattern.resources.items():
usage = f"{len(resource.current_users)}/{resource.capacity}"
print(f" {resource.name}: {usage} users, status: {resource.status.value}")
print("\n✓ Coordination pattern enabled multi-agent collaboration!")
Agentic Architecture Pattern: Coordination
Pattern 30: 30 Coordination – Line-by-Line Explanation
Overview
Pattern 30: Coordination Pattern
Agents coordinate activities, resources, and timing to achieve shared goals. Manages dependencies, synchronization, and conflict resolution.
Code Structure Analysis
File Information
- Pattern Number: 30
- Pattern Name: 30 Coordination
- File Path:
patterns\collaboration\30_coordination.py - Total Lines: 357
Detailed Code Explanation
Module Docstring
"""
Pattern 30: Coordination Pattern
Agents coordinate activities, resources, and timing to achieve shared goals.
Manages dependencies, synchronization, and conflict resolution.
"""
Explanation: This module docstring describes the purpose and key features of the 30 Coordination pattern.
Imports Section
from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime, timedelta
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, Set: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime, timedelta: 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
TaskStatus (Line 18)
class TaskStatus(Enum):
PENDING
READY
IN_PROGRESS
BLOCKED
COMPLETED
FAILED
Purpose: Status of coordinated task.
Values:
– PENDING: Initial state, not yet processed
– READY: Represents ready
– IN_PROGRESS: Represents in progress
– BLOCKED: Represents blocked
– COMPLETED: Successfully finished
– FAILED: Execution failed
ResourceStatus (Line 28)
class ResourceStatus(Enum):
AVAILABLE
ALLOCATED
IN_USE
MAINTENANCE
Purpose: Status of shared resource.
Values:
– AVAILABLE: Represents available
– ALLOCATED: Represents allocated
– IN_USE: Represents in use
– MAINTENANCE: Represents maintenance
Classes
TaskStatus (Line 18)
Description: Status of coordinated task.
ResourceStatus (Line 28)
Description: Status of shared resource.
CoordinatedTask (Line 37)
Description: Task requiring coordination.
SharedResource (Line 51)
Description: Shared resource requiring coordination.
Methods:
is_available (Line 59)
def is_available(self):
"""Check if resource is available."""
Purpose: Check if resource is available.
Parameters:
– self: Instance reference
CoordinationState (Line 64)
Description: State for coordination workflow.
CoordinatedAgent (Line 72)
Description: Agent participating in coordination.
Methods:
__init__ (Line 75)
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”)
request_task (Line 81)
def request_task(self, task):
"""Request to execute a task."""
Purpose: Request to execute a task.
Parameters:
– self: Instance reference
– task: Parameter for task
start_task (Line 97)
def start_task(self, task):
"""Start executing task."""
Purpose: Start executing task.
Parameters:
– self: Instance reference
– task: Parameter for task
complete_task (Line 104)
def complete_task(self, task):
"""Complete task execution."""
Purpose: Complete task execution.
Parameters:
– self: Instance reference
– task: Parameter for task
CoordinationPattern (Line 112)
Description: Coordination Pattern – Multi-agent activity coordination.
Agents coordinate tasks, resources, and timing through a central coordinator that manages dependencies and resolves conflicts.
Methods:
__init__ (Line 120)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
_build_workflow (Line 127)
def _build_workflow(self):
"""Build LangGraph coordination workflow."""
Purpose: Build LangGraph coordination workflow.
Parameters:
– self: Instance reference
register_agent (Line 159)
def register_agent(self, agent):
"""Register agent for coordination."""
Purpose: Register agent for coordination.
Parameters:
– self: Instance reference
– agent: Parameter for agent
add_task (Line 163)
def add_task(self, task):
"""Add task to coordination."""
Purpose: Add task to coordination.
Parameters:
– self: Instance reference
– task: Parameter for task
add_resource (Line 167)
def add_resource(self, resource):
"""Add shared resource."""
Purpose: Add shared resource.
Parameters:
– self: Instance reference
– resource: Parameter for resource
coordinate_execution (Line 171)
def coordinate_execution(self):
"""Coordinate task execution."""
Purpose: Coordinate task execution.
Parameters:
– self: Instance reference
_assign_task (Line 209)
def _assign_task(self, task):
"""Assign task to available agent."""
Purpose: Assign task to available agent.
Parameters:
– self: Instance reference
– task: Parameter for task
complete_task_coordination (Line 228)
def complete_task_coordination(self, task_id):
"""Complete task and release resources."""
Purpose: Complete task and release resources.
Parameters:
– self: Instance reference
– task_id: Parameter for task_id
run (Line 247)
def run(self, scenario):
"""Execute coordination pattern."""
Purpose: Execute coordination pattern.
Parameters:
– self: Instance reference
– scenario: Parameter for scenario
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