"""
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