"""
Pattern 124: Leader-Follower Pattern
Purpose: Dynamic leadership with follower coordination and leader election
"""
from typing import Dict, Any, List, Optional, Set
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 AgentRole(Enum):
"""Agent roles in leader-follower pattern."""
LEADER = "leader"
FOLLOWER = "follower"
CANDIDATE = "candidate"
@dataclass
class Task:
"""Task to be executed."""
task_id: str
description: str
priority: int = 1
assigned_to: Optional[str] = None
status: str = "pending" # pending, in_progress, completed
result: Optional[str] = None
@dataclass
class Vote:
"""Vote in leader election."""
voter_id: str
candidate_id: str
term: int
timestamp: datetime = field(default_factory=datetime.now)
class LeaderFollowerAgent:
"""Agent that can be leader or follower."""
def __init__(self, agent_id: str, model_name: str = "gpt-4"):
self.agent_id = agent_id
self.role = AgentRole.FOLLOWER
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
self.current_leader: Optional[str] = None
self.term = 0
self.votes_received: List[Vote] = []
self.tasks_completed: List[Task] = []
self.leadership_score = 0.0
def become_leader(self, term: int):
"""Become the leader."""
self.role = AgentRole.LEADER
self.term = term
self.leadership_score = 1.0
print(f" 🔷 {self.agent_id} became LEADER for term {term}")
def become_follower(self, leader_id: str):
"""Become a follower."""
self.role = AgentRole.FOLLOWER
self.current_leader = leader_id
print(f" ↳ {self.agent_id} following {leader_id}")
def execute_task(self, task: Task) -> str:
"""Execute an assigned task."""
prompt = ChatPromptTemplate.from_messages([
("system", f"You are {self.agent_id}, a {self.role.value} agent. Execute tasks efficiently."),
("human", "Task: {description}")
])
chain = prompt | self.llm
response = chain.invoke({"description": task.description})
task.status = "completed"
task.assigned_to = self.agent_id
task.result = str(response.content)
self.tasks_completed.append(task)
return task.result
def delegate_task(self, task: Task, follower_id: str) -> str:
"""Leader delegates task to follower."""
if self.role != AgentRole.LEADER:
raise ValueError("Only leader can delegate tasks")
task.assigned_to = follower_id
task.status = "delegated"
return f"Task {task.task_id} delegated to {follower_id}"
class LeaderFollowerPattern:
"""
Leader-Follower Pattern - Dynamic leadership with follower coordination
Features:
- Leader election
- Task delegation from leader to followers
- Failover and re-election
"""
def __init__(self):
self.agents: Dict[str, LeaderFollowerAgent] = {}
self.current_leader: Optional[str] = None
self.term = 0
self.task_queue: List[Task] = []
def register_agent(self, agent: LeaderFollowerAgent):
"""Register an agent."""
self.agents[agent.agent_id] = agent
def elect_leader(self) -> str:
"""Elect a new leader based on leadership scores."""
if not self.agents:
raise ValueError("No agents registered")
self.term += 1
# Simple election: agent with highest leadership score or first agent
candidates = list(self.agents.values())
leader = max(candidates, key=lambda a: (a.leadership_score, a.agent_id))
# Update roles
leader.become_leader(self.term)
self.current_leader = leader.agent_id
for agent in self.agents.values():
if agent.agent_id != leader.agent_id:
agent.become_follower(leader.agent_id)
return leader.agent_id
def submit_task(self, description: str, priority: int = 1) -> Task:
"""Submit a task to be executed."""
task = Task(
task_id=f"task_{len(self.task_queue) + 1}",
description=description,
priority=priority
)
self.task_queue.append(task)
return task
def process_tasks(self) -> List[Dict[str, Any]]:
"""Leader processes task queue by delegating to followers."""
if not self.current_leader:
self.elect_leader()
leader = self.agents[self.current_leader]
followers = [a for a in self.agents.values() if a.role == AgentRole.FOLLOWER]
if not followers:
# Leader executes all tasks if no followers
results = []
for task in self.task_queue:
result = leader.execute_task(task)
results.append({
"task_id": task.task_id,
"executed_by": leader.agent_id,
"status": task.status
})
self.task_queue.clear()
return results
# Delegate tasks to followers in round-robin fashion
results = []
for i, task in enumerate(self.task_queue):
follower = followers[i % len(followers)]
leader.delegate_task(task, follower.agent_id)
result = follower.execute_task(task)
results.append({
"task_id": task.task_id,
"delegated_by": leader.agent_id,
"executed_by": follower.agent_id,
"status": task.status
})
self.task_queue.clear()
return results
def get_statistics(self) -> Dict[str, Any]:
"""Get pattern statistics."""
leader_stats = None
if self.current_leader:
leader = self.agents[self.current_leader]
leader_stats = {
"id": leader.agent_id,
"term": leader.term,
"tasks_completed": len(leader.tasks_completed)
}
follower_stats = []
for agent in self.agents.values():
if agent.role == AgentRole.FOLLOWER:
follower_stats.append({
"id": agent.agent_id,
"tasks_completed": len(agent.tasks_completed)
})
return {
"current_term": self.term,
"leader": leader_stats,
"followers": follower_stats,
"total_agents": len(self.agents),
"pending_tasks": len(self.task_queue)
}
# Example usage
if __name__ == "__main__":
print("=== Leader-Follower Pattern Demo ===\n")
# Create pattern
pattern = LeaderFollowerPattern()
# Create agents
agent1 = LeaderFollowerAgent("Agent1")
agent2 = LeaderFollowerAgent("Agent2")
agent3 = LeaderFollowerAgent("Agent3")
# Set initial leadership scores
agent1.leadership_score = 0.9
agent2.leadership_score = 0.7
agent3.leadership_score = 0.8
pattern.register_agent(agent1)
pattern.register_agent(agent2)
pattern.register_agent(agent3)
# Elect initial leader
print("Initial Election:")
leader_id = pattern.elect_leader()
print(f"\nCurrent leader: {leader_id}\n")
# Submit tasks
print("Submitting tasks...")
pattern.submit_task("Analyze market trends", priority=3)
pattern.submit_task("Generate report", priority=2)
pattern.submit_task("Review data quality", priority=1)
# Process tasks
print("\nProcessing tasks...")
results = pattern.process_tasks()
print("\nTask Results:")
for result in results:
print(f" Task {result['task_id']}: {result['status']}")
print(f" Executed by: {result['executed_by']}")
# Show statistics
print("\n=== Statistics ===")
stats = pattern.get_statistics()
print(f"Current Term: {stats['current_term']}")
print(f"\nLeader: {stats['leader']['id']}")
print(f" Tasks Completed: {stats['leader']['tasks_completed']}")
print(f"\nFollowers: {len(stats['followers'])}")
for follower in stats['followers']:
print(f" {follower['id']}: {follower['tasks_completed']} tasks completed")
Agentic Architecture Pattern: Leader Follower
Pattern 124: 124 Leader Follower Pattern – Line-by-Line Explanation
Overview
Pattern 124: Leader-Follower Pattern
Dynamic leader election with followers Implements distributed agent coordination using LangChain/LangGraph.
Code Structure Analysis
File Information
- Pattern Number: 124
- Pattern Name: 124 Leader Follower Pattern
- File Path:
patterns\distributed_agent\124_leader_follower_pattern.py - Total Lines: 178
Detailed Code Explanation
Module Docstring
"""
Pattern 124: Leader-Follower Pattern
Dynamic leader election with followers
Implements distributed agent coordination using LangChain/LangGraph.
"""
Explanation: This module docstring describes the purpose and key features of the 124 Leader Follower 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
import random # 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
AgentStatus (Line 19)
class AgentStatus(Enum):
ACTIVE
IDLE
BUSY
FAILED
Purpose: Status of distributed agent.
Values:
– ACTIVE: Currently operational
– IDLE: Not currently active
– BUSY: Represents busy
– FAILED: Execution failed
Classes
AgentStatus (Line 19)
Description: Status of distributed agent.
DistributedAgent (Line 28)
Description: Represents a distributed agent node.
Task (Line 38)
Description: Task to be executed.
LeaderFollowerPattern (Line 47)
Description: Leader-Follower Pattern – Dynamic leader election with followers
Coordinates multiple distributed agents for scalable and fault-tolerant execution of tasks across a distributed system.
Methods:
__init__ (Line 55)
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”)
register_agent (Line 60)
def register_agent(self, agent_id, capabilities):
"""Register a new distributed agent."""
Purpose: Register a new distributed agent.
Parameters:
– self: Instance reference
– agent_id: Parameter for agent_id
– capabilities: Parameter for capabilities
select_agent (Line 70)
def select_agent(self, task):
"""Select best agent for task."""
Purpose: Select best agent for task.
Parameters:
– self: Instance reference
– task: Parameter for task
assign_task (Line 84)
def assign_task(self, task):
"""Assign task to an agent."""
Purpose: Assign task to an agent.
Parameters:
– self: Instance reference
– task: Parameter for task
execute_task (Line 97)
def execute_task(self, task):
"""Execute task on assigned agent."""
Purpose: Execute task on assigned agent.
Parameters:
– self: Instance reference
– task: Parameter for task
run (Line 129)
def run(self, task_description):
"""Main execution method."""
Purpose: Main execution method.
Parameters:
– self: Instance reference
– task_description: Parameter for task_description
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 = AgentStatus(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