"""
Pattern 126: Fault-Tolerant Pattern

Continues operation despite agent failures
Implements distributed agent coordination using LangChain/LangGraph.
"""

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


class AgentStatus(Enum):
    """Status of distributed agent."""
    ACTIVE = "active"
    IDLE = "idle"
    BUSY = "busy"
    FAILED = "failed"


@dataclass
class DistributedAgent:
    """Represents a distributed agent node."""
    agent_id: str
    status: AgentStatus = AgentStatus.IDLE
    load: float = 0.0  # 0.0 to 1.0
    capabilities: List[str] = field(default_factory=list)
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class Task:
    """Task to be executed."""
    task_id: str
    description: str
    assigned_to: Optional[str] = None
    status: str = "pending"
    result: Optional[str] = None


class FaultTolerantPattern:
    """
    Fault-Tolerant Pattern - Continues operation despite agent failures
    
    Coordinates multiple distributed agents for scalable and fault-tolerant
    execution of tasks across a distributed system.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.3)
        self.agents: Dict[str, DistributedAgent] = {}
        self.tasks: List[Task] = []
    
    def register_agent(self, agent_id: str, capabilities: List[str] = None):
        """Register a new distributed agent."""
        agent = DistributedAgent(
            agent_id=agent_id,
            capabilities=capabilities or ["general"],
            status=AgentStatus.IDLE
        )
        self.agents[agent_id] = agent
        return agent
    
    def select_agent(self, task: Task) -> Optional[str]:
        """Select best agent for task."""
        available = [
            (aid, agent) for aid, agent in self.agents.items()
            if agent.status in [AgentStatus.IDLE, AgentStatus.ACTIVE] and agent.load  bool:
        """Assign task to an agent."""
        agent_id = self.select_agent(task)
        if not agent_id:
            return False
        
        task.assigned_to = agent_id
        task.status = "assigned"
        self.agents[agent_id].load += 0.2
        self.agents[agent_id].status = AgentStatus.BUSY
        
        return True
    
    def execute_task(self, task: Task) -> Dict[str, Any]:
        """Execute task on assigned agent."""
        if not task.assigned_to:
            if not self.assign_task(task):
                return {"status": "failed", "reason": "no_agent_available"}
        
        agent = self.agents[task.assigned_to]
        
        # Simulate task execution with LLM
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are agent {agent.agent_id} in a Fault-Tolerant system."),
            ("human", "{task_description}")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({"task_description": task.description})
        
        task.result = str(response.content)
        task.status = "completed"
        
        # Update agent status
        agent.load = max(0.0, agent.load - 0.2)
        if agent.load  Dict[str, Any]:
        """Main execution method."""
        task = Task(
            task_id=f"task_{len(self.tasks) + 1}",
            description=task_description
        )
        self.tasks.append(task)
        
        result = self.execute_task(task)
        
        return {
            "pattern": "Fault-Tolerant",
            "task_id": task.task_id,
            "assigned_to": task.assigned_to,
            "status": result["status"],
            "result": result.get("result", ""),
            "total_agents": len(self.agents),
            "active_agents": sum(1 for a in self.agents.values() if a.status == AgentStatus.ACTIVE or a.status == AgentStatus.BUSY)
        }


# Example usage
if __name__ == "__main__":
    print("=== Fault-Tolerant Pattern Demo ===\n")
    
    pattern = FaultTolerantPattern()
    
    # Register agents
    for i in range(3):
        pattern.register_agent(f"agent_{i+1}", ["compute", "process"])
    
    print(f"Registered {len(pattern.agents)} agents")
    
    # Execute tasks
    tasks = [
        "Process data batch 1",
        "Analyze results",
        "Generate report"
    ]
    
    for task_desc in tasks:
        result = pattern.run(task_desc)
        print(f"\nTask: {task_desc}")
        print(f"  Assigned to: {result['assigned_to']}")
        print(f"  Status: {result['status']}")
        print(f"  Result: {result['result'][:100]}...")
    
    print(f"\nTotal tasks executed: {len(pattern.tasks)}")
    print(f"Active agents: {result['active_agents']}/{result['total_agents']}")

    

Agentic Architecture Pattern: Fault Tolerant

Pattern 126: 126 Fault Tolerant Pattern – Line-by-Line Explanation

Overview

Pattern 126: Fault-Tolerant Pattern

Continues operation despite agent failures Implements distributed agent coordination using LangChain/LangGraph.

Code Structure Analysis

File Information

• Pattern Number: 126
• Pattern Name: 126 Fault Tolerant Pattern
• File Path: patterns\distributed_agent\126_fault_tolerant_pattern.py
• Total Lines: 178

Detailed Code Explanation

Module Docstring

"""
Pattern 126: Fault-Tolerant Pattern

Continues operation despite agent failures
Implements distributed agent coordination using LangChain/LangGraph.
"""

Explanation: This module docstring describes the purpose and key features of the 126 Fault Tolerant 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.

FaultTolerantPattern (Line 47)

Description: Fault-Tolerant Pattern – Continues operation despite agent failures

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