"""
Pattern 33: Matrix Organization Pattern

Agents report to both functional managers and project managers simultaneously.
Dual reporting structure balances functional expertise with project needs.
"""

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 ManagerType(Enum):
    FUNCTIONAL = "functional"
    PROJECT = "project"


@dataclass
class MatrixAgent:
    """Agent with dual reporting structure."""
    name: str
    functional_area: str  # e.g., Engineering, Design, Marketing
    functional_manager: Optional[str] = None
    projects: List[str] = field(default_factory=list)
    project_managers: List[str] = field(default_factory=list)
    tasks: List[Dict[str, Any]] = field(default_factory=list)


@dataclass
class Manager:
    """Manager in matrix organization."""
    name: str
    manager_type: ManagerType
    area: str  # Functional area or Project name
    direct_reports: Set[str] = field(default_factory=set)


@dataclass
class ConflictResolution:
    """Conflict between functional and project priorities."""
    agent_name: str
    functional_priority: str
    project_priority: str
    resolution: Optional[str] = None
    timestamp: datetime = field(default_factory=datetime.now)


class MatrixOrganizationPattern:
    """
    Matrix Organization Pattern - Dual reporting structure.
    
    Features:
    - Functional managers (expertise areas)
    - Project managers (initiatives)
    - Agents report to both
    - Priority conflict resolution
    - Resource balancing
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.agents: Dict[str, MatrixAgent] = {}
        self.functional_managers: Dict[str, Manager] = {}
        self.project_managers: Dict[str, Manager] = {}
        self.conflicts: List[ConflictResolution] = []
    
    def add_functional_manager(self, name: str, area: str):
        """Add a functional manager."""
        manager = Manager(name, ManagerType.FUNCTIONAL, area)
        self.functional_managers[name] = manager
        print(f"✓ Added functional manager: {name} ({area})")
    
    def add_project_manager(self, name: str, project: str):
        """Add a project manager."""
        manager = Manager(name, ManagerType.PROJECT, project)
        self.project_managers[name] = manager
        print(f"✓ Added project manager: {name} (Project: {project})")
    
    def add_agent(self, agent: MatrixAgent):
        """Add an agent with functional assignment."""
        self.agents[agent.name] = agent
        
        # Assign to functional manager
        for fm_name, fm in self.functional_managers.items():
            if fm.area == agent.functional_area:
                agent.functional_manager = fm_name
                fm.direct_reports.add(agent.name)
                break
        
        print(f"✓ Added agent: {agent.name} ({agent.functional_area}) → "
              f"Functional Manager: {agent.functional_manager}")
    
    def assign_to_project(self, agent_name: str, project_name: str, 
                          project_manager_name: str):
        """Assign agent to a project with project manager."""
        if agent_name in self.agents and project_manager_name in self.project_managers:
            agent = self.agents[agent_name]
            pm = self.project_managers[project_manager_name]
            
            agent.projects.append(project_name)
            agent.project_managers.append(project_manager_name)
            pm.direct_reports.add(agent_name)
            
            print(f"✓ {agent_name} assigned to project '{project_name}' → PM: {project_manager_name}")
    
    def assign_task(self, agent_name: str, task: str, source_manager: str, 
                    manager_type: ManagerType) -> Dict[str, Any]:
        """Assign task from either functional or project manager."""
        if agent_name not in self.agents:
            return {"success": False, "error": "Agent not found"}
        
        agent = self.agents[agent_name]
        task_info = {
            "task": task,
            "source_manager": source_manager,
            "manager_type": manager_type.value,
            "assigned_at": datetime.now()
        }
        agent.tasks.append(task_info)
        
        return {"success": True, "agent": agent_name, "task": task}
    
    def resolve_priority_conflict(self, agent_name: str, functional_task: str, 
                                   project_task: str) -> ConflictResolution:
        """Resolve conflict between functional and project priorities."""
        if agent_name not in self.agents:
            return None
        
        agent = self.agents[agent_name]
        conflict = ConflictResolution(
            agent_name=agent_name,
            functional_priority=functional_task,
            project_priority=project_task
        )
        
        # Use LLM to resolve conflict
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are resolving a priority conflict for {agent_name}, "
                      f"who works in {agent.functional_area}. "
                      f"They report to both a functional manager and project managers. "
                      f"Provide a balanced resolution that considers both priorities."),
            ("human", f"Functional Manager Priority: {functional_task}\n"
                     f"Project Manager Priority: {project_task}\n\n"
                     f"Resolution:")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        conflict.resolution = response.content.strip()
        
        self.conflicts.append(conflict)
        return conflict
    
    def get_agent_workload(self, agent_name: str) -> Dict[str, Any]:
        """Get agent's workload across all dimensions."""
        if agent_name not in self.agents:
            return {}
        
        agent = self.agents[agent_name]
        return {
            "agent": agent_name,
            "functional_area": agent.functional_area,
            "functional_manager": agent.functional_manager,
            "active_projects": len(agent.projects),
            "project_managers": agent.project_managers,
            "total_tasks": len(agent.tasks),
            "tasks_by_manager_type": {
                "functional": sum(1 for t in agent.tasks if t["manager_type"] == "functional"),
                "project": sum(1 for t in agent.tasks if t["manager_type"] == "project")
            }
        }
    
    def get_manager_team(self, manager_name: str) -> Dict[str, Any]:
        """Get team reporting to a manager."""
        if manager_name in self.functional_managers:
            manager = self.functional_managers[manager_name]
        elif manager_name in self.project_managers:
            manager = self.project_managers[manager_name]
        else:
            return {}
        
        return {
            "manager": manager_name,
            "type": manager.manager_type.value,
            "area": manager.area,
            "direct_reports": list(manager.direct_reports)
        }
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute matrix organization scenario."""
        return {
            "pattern": "Matrix Organization",
            "scenario": scenario,
            "agents": len(self.agents),
            "functional_managers": len(self.functional_managers),
            "project_managers": len(self.project_managers),
            "conflicts_resolved": len(self.conflicts)
        }


# Example usage
if __name__ == "__main__":
    print("=== Matrix Organization Pattern Demo ===\n")
    
    org = MatrixOrganizationPattern()
    
    # Add functional managers
    print("--- Functional Managers ---")
    org.add_functional_manager("Sarah", "Engineering")
    org.add_functional_manager("Mike", "Design")
    org.add_functional_manager("Lisa", "Data Science")
    
    # Add project managers
    print("\n--- Project Managers ---")
    org.add_project_manager("Tom", "Mobile App Launch")
    org.add_project_manager("Jane", "AI Integration")
    
    # Add agents
    print("\n--- Agents (Dual Reporting) ---")
    agents = [
        MatrixAgent("Alice", "Engineering"),
        MatrixAgent("Bob", "Design"),
        MatrixAgent("Carol", "Data Science"),
        MatrixAgent("David", "Engineering")
    ]
    
    for agent in agents:
        org.add_agent(agent)
    
    # Assign agents to projects
    print("\n--- Project Assignments ---")
    org.assign_to_project("Alice", "Mobile App Launch", "Tom")
    org.assign_to_project("Bob", "Mobile App Launch", "Tom")
    org.assign_to_project("Carol", "AI Integration", "Jane")
    org.assign_to_project("David", "AI Integration", "Jane")
    
    # Assign tasks from both types of managers
    print("\n--- Task Assignments ---")
    org.assign_task("Alice", "Code review for team", "Sarah", ManagerType.FUNCTIONAL)
    org.assign_task("Alice", "Build login feature", "Tom", ManagerType.PROJECT)
    org.assign_task("Carol", "Train ML model", "Jane", ManagerType.PROJECT)
    org.assign_task("Carol", "Mentor junior analysts", "Lisa", ManagerType.FUNCTIONAL)
    
    # Resolve priority conflict
    print("\n--- Priority Conflict Resolution ---")
    conflict = org.resolve_priority_conflict(
        "Alice",
        "Lead technical training session (Functional Manager)",
        "Complete critical app release (Project Manager)"
    )
    print(f"Agent: {conflict.agent_name}")
    print(f"Functional: {conflict.functional_priority}")
    print(f"Project: {conflict.project_priority}")
    print(f"Resolution: {conflict.resolution}")
    
    # Show agent workload
    print("\n--- Agent Workload ---")
    for agent_name in ["Alice", "Carol"]:
        workload = org.get_agent_workload(agent_name)
        print(f"\n{agent_name}:")
        print(f"  Functional Area: {workload['functional_area']}")
        print(f"  Functional Manager: {workload['functional_manager']}")
        print(f"  Active Projects: {workload['active_projects']}")
        print(f"  Project Managers: {', '.join(workload['project_managers'])}")
        print(f"  Total Tasks: {workload['total_tasks']}")
        print(f"  Tasks by Type: Functional={workload['tasks_by_manager_type']['functional']}, "
              f"Project={workload['tasks_by_manager_type']['project']}")
    
    # Show manager teams
    print("\n--- Manager Teams ---")
    for manager_name in ["Sarah", "Tom"]:
        team = org.get_manager_team(manager_name)
        print(f"\n{manager_name} ({team['type']} - {team['area']}):")
        print(f"  Reports: {', '.join(team['direct_reports'])}")
    
    print("\n--- Summary ---")
    result = org.run("Dual reporting structure demo")
    print(f"Pattern: {result['pattern']}")
    print(f"Total Agents: {result['agents']}")
    print(f"Functional Managers: {result['functional_managers']}")
    print(f"Project Managers: {result['project_managers']}")
    print(f"Conflicts Resolved: {result['conflicts_resolved']}")

    

Agentic Architecture Pattern: Matrix Organization

Pattern 33: 33 Matrix Organization – Line-by-Line Explanation

Overview

Pattern 33: Matrix Organization Pattern

Agents report to both functional managers and project managers simultaneously. Dual reporting structure balances functional expertise with project needs.

Code Structure Analysis

File Information

• Pattern Number: 33
• Pattern Name: 33 Matrix Organization
• File Path: patterns\organizational\33_matrix_organization.py
• Total Lines: 286

Detailed Code Explanation

Module Docstring

"""
Pattern 33: Matrix Organization Pattern

Agents report to both functional managers and project managers simultaneously.
Dual reporting structure balances functional expertise with project needs.
"""

Explanation: This module docstring describes the purpose and key features of the 33 Matrix Organization pattern.

Imports Section

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  # 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: 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

Enumerations

ManagerType (Line 16)

class ManagerType(Enum):
    FUNCTIONAL
    PROJECT

Purpose: No enum docstring.

Values: – FUNCTIONAL: Represents functional – PROJECT: Represents project

Classes

ManagerType (Line 16)

Description: No class docstring.

MatrixAgent (Line 22)

Description: Agent with dual reporting structure.

Manager (Line 33)

Description: Manager in matrix organization.

ConflictResolution (Line 42)

Description: Conflict between functional and project priorities.

MatrixOrganizationPattern (Line 51)

Description: Matrix Organization Pattern – Dual reporting structure.

Features: – Functional managers (expertise areas) – Project managers (initiatives) – Agents report to both – Priority conflict resolution – Resource balancing

Methods:

__init__ (Line 63)

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_functional_manager (Line 70)

def add_functional_manager(self, name, area):
    """Add a functional manager."""

Purpose: Add a functional manager.

Parameters: – self: Instance reference – name: Identifier or label – area: Parameter for area

add_project_manager (Line 76)

def add_project_manager(self, name, project):
    """Add a project manager."""

Purpose: Add a project manager.

Parameters: – self: Instance reference – name: Identifier or label – project: Parameter for project

add_agent (Line 82)

def add_agent(self, agent):
    """Add an agent with functional assignment."""

Purpose: Add an agent with functional assignment.

Parameters: – self: Instance reference – agent: Parameter for agent

assign_to_project (Line 96)

def assign_to_project(self, agent_name, project_name, project_manager_name):
    """Assign agent to a project with project manager."""

Purpose: Assign agent to a project with project manager.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – project_name: Parameter for project_name – project_manager_name: Parameter for project_manager_name

assign_task (Line 109)

def assign_task(self, agent_name, task, source_manager, manager_type):
    """Assign task from either functional or project manager."""

Purpose: Assign task from either functional or project manager.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – task: Parameter for task – source_manager: Parameter for source_manager – manager_type: Parameter for manager_type

resolve_priority_conflict (Line 126)

def resolve_priority_conflict(self, agent_name, functional_task, project_task):
    """Resolve conflict between functional and project priorities."""

Purpose: Resolve conflict between functional and project priorities.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name – functional_task: Parameter for functional_task – project_task: Parameter for project_task

get_agent_workload (Line 157)

def get_agent_workload(self, agent_name):
    """Get agent's workload across all dimensions."""

Purpose: Get agent’s workload across all dimensions.

Parameters: – self: Instance reference – agent_name: Parameter for agent_name

get_manager_team (Line 176)

def get_manager_team(self, manager_name):
    """Get team reporting to a manager."""

Purpose: Get team reporting to a manager.

Parameters: – self: Instance reference – manager_name: Parameter for manager_name

run (Line 192)

def run(self, scenario):
    """Execute matrix organization scenario."""

Purpose: Execute matrix organization scenario.

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 = ManagerType(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

    """
Pattern 32: Flat Organization Pattern

A decentralized organizational structure where agents operate as peers with equal authority.
No hierarchical reporting - all agents collaborate directly with autonomy and shared responsibility.
"""

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing_extensions import TypedDict


@dataclass
class PeerAgent:
    """Agent in flat organization with equal standing."""
    name: str
    role: str
    expertise: List[str]
    active_tasks: List[str] = field(default_factory=list)
    completed_tasks: List[str] = field(default_factory=list)
    collaborations: Set[str] = field(default_factory=set)
    
    def __hash__(self):
        return hash(self.name)


@dataclass
class CollaborationRequest:
    """Request for peer collaboration."""
    requester: str
    task: str
    required_expertise: List[str]
    timestamp: datetime = field(default_factory=datetime.now)
    responders: List[str] = field(default_factory=list)


class FlatOrganizationPattern:
    """
    Flat Organization Pattern - Peer-based structure with no hierarchy.
    
    Features:
    - Equal authority among all agents
    - Direct peer-to-peer collaboration
    - Self-organization and task claiming
    - Shared decision-making
    - Transparent communication
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.peers: Dict[str, PeerAgent] = {}
        self.collaboration_requests: List[CollaborationRequest] = []
        self.shared_decisions: List[Dict[str, Any]] = []
        
    def add_peer(self, peer: PeerAgent):
        """Add a peer agent to the organization."""
        self.peers[peer.name] = peer
        print(f"✓ Added peer: {peer.name} ({peer.role}) - Expertise: {', '.join(peer.expertise)}")
    
    def request_collaboration(self, requester_name: str, task: str, 
                             required_expertise: List[str]) -> CollaborationRequest:
        """Request collaboration from peers."""
        request = CollaborationRequest(
            requester=requester_name,
            task=task,
            required_expertise=required_expertise
        )
        self.collaboration_requests.append(request)
        
        # Find peers with matching expertise
        for peer_name, peer in self.peers.items():
            if peer_name != requester_name:
                if any(exp in peer.expertise for exp in required_expertise):
                    request.responders.append(peer_name)
                    peer.collaborations.add(requester_name)
                    self.peers[requester_name].collaborations.add(peer_name)
        
        return request
    
    def assign_task(self, task: str) -> Optional[str]:
        """Peer self-selects task based on expertise and availability."""
        # Find available peers
        available_peers = [p for p in self.peers.values() if len(p.active_tasks)  Dict[str, Any]:
        """All peers contribute to decision-making equally."""
        votes = {}
        
        for peer_name, peer in self.peers.items():
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {peer_name}, a {peer.role} in a flat organization. "
                          f"Your expertise: {', '.join(peer.expertise)}. "
                          f"Vote on the decision and provide reasoning."),
                ("human", f"Decision: {decision_topic}\n\nYour vote (Yes/No) and reasoning:")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({})
            content = response.content.strip()
            
            # Parse vote
            vote = "Yes" if "yes" in content.lower()[:50] else "No"
            votes[peer_name] = {"vote": vote, "reasoning": content}
        
        # Count votes
        yes_votes = sum(1 for v in votes.values() if v["vote"] == "Yes")
        no_votes = len(votes) - yes_votes
        
        decision = {
            "topic": decision_topic,
            "votes": votes,
            "yes_votes": yes_votes,
            "no_votes": no_votes,
            "result": "Approved" if yes_votes > no_votes else "Rejected",
            "timestamp": datetime.now()
        }
        
        self.shared_decisions.append(decision)
        return decision
    
    def complete_task(self, peer_name: str, task: str):
        """Mark task as completed for a peer."""
        if peer_name in self.peers:
            peer = self.peers[peer_name]
            if task in peer.active_tasks:
                peer.active_tasks.remove(task)
                peer.completed_tasks.append(task)
    
    def get_collaboration_network(self) -> Dict[str, List[str]]:
        """Get peer-to-peer collaboration network."""
        return {
            peer_name: list(peer.collaborations)
            for peer_name, peer in self.peers.items()
        }
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute flat organization scenario."""
        return {
            "pattern": "Flat Organization",
            "scenario": scenario,
            "total_peers": len(self.peers),
            "collaboration_requests": len(self.collaboration_requests),
            "shared_decisions": len(self.shared_decisions),
            "collaboration_network": self.get_collaboration_network()
        }


# Example usage
if __name__ == "__main__":
    print("=== Flat Organization Pattern Demo ===\n")
    
    org = FlatOrganizationPattern()
    
    # Add peer agents
    peers = [
        PeerAgent("Alice", "Developer", ["Python", "FastAPI", "Testing"]),
        PeerAgent("Bob", "Designer", ["UI/UX", "Figma", "User Research"]),
        PeerAgent("Carol", "Data Scientist", ["Python", "ML", "Data Analysis"]),
        PeerAgent("David", "DevOps", ["Docker", "Kubernetes", "CI/CD"]),
        PeerAgent("Eve", "Product Manager", ["Strategy", "User Research", "Analytics"])
    ]
    
    for peer in peers:
        org.add_peer(peer)
    
    print("\n--- Collaboration Requests ---")
    # Request collaborations
    req1 = org.request_collaboration(
        "Alice", 
        "Build ML inference API",
        ["Python", "ML"]
    )
    print(f"✓ {req1.requester} requested help with: {req1.task}")
    print(f"  Responders: {', '.join(req1.responders)}")
    
    req2 = org.request_collaboration(
        "Bob",
        "User research for new feature",
        ["User Research", "Analytics"]
    )
    print(f"✓ {req2.requester} requested help with: {req2.task}")
    print(f"  Responders: {', '.join(req2.responders)}")
    
    print("\n--- Self-Selected Task Assignment ---")
    # Peers self-select tasks
    tasks = [
        "Implement user authentication",
        "Design dashboard mockups",
        "Deploy to production"
    ]
    
    for task in tasks:
        assignee = org.assign_task(task)
        if assignee:
            print(f"✓ {assignee} self-selected: {task}")
    
    print("\n--- Shared Decision Making ---")
    # Make shared decisions
    decision = org.make_shared_decision(
        "Should we adopt microservices architecture?"
    )
    print(f"Decision: {decision['topic']}")
    print(f"Result: {decision['result']} ({decision['yes_votes']} Yes, {decision['no_votes']} No)")
    
    for peer_name, vote_info in decision['votes'].items():
        print(f"  {peer_name}: {vote_info['vote']}")
        print(f"    → {vote_info['reasoning'][:100]}...")
    
    print("\n--- Organization Statistics ---")
    result = org.run("Team collaboration demo")
    print(f"Total Peers: {result['total_peers']}")
    print(f"Collaboration Requests: {result['collaboration_requests']}")
    print(f"Shared Decisions: {result['shared_decisions']}")
    print("\nCollaboration Network:")
    for peer, collaborators in result['collaboration_network'].items():
        if collaborators:
            print(f"  {peer} ↔ {', '.join(collaborators)}")

    

Agentic Architecture Pattern: Flat Organization

Pattern 32: 32 Flat Organization – Line-by-Line Explanation

Overview

Pattern 32: Flat Organization Pattern

A decentralized organizational structure where agents operate as peers with equal authority. No hierarchical reporting – all agents collaborate directly with autonomy and shared responsibility.

Code Structure Analysis

File Information

• Pattern Number: 32
• Pattern Name: 32 Flat Organization
• File Path: patterns\organizational\32_flat_organization.py
• Total Lines: 249

Detailed Code Explanation

Module Docstring

"""
Pattern 32: Flat Organization Pattern

A decentralized organizational structure where agents operate as peers with equal authority.
No hierarchical reporting - all agents collaborate directly with autonomy and shared responsibility.
"""

Explanation: This module docstring describes the purpose and key features of the 32 Flat Organization pattern.

Imports Section

from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
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: Date and time utilities – 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

Classes

PeerAgent (Line 18)

Description: Agent in flat organization with equal standing.

Methods:

__hash__ (Line 27)

def __hash__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

CollaborationRequest (Line 32)

Description: Request for peer collaboration.

FlatOrganizationPattern (Line 41)

Description: Flat Organization Pattern – Peer-based structure with no hierarchy.

Features: – Equal authority among all agents – Direct peer-to-peer collaboration – Self-organization and task claiming – Shared decision-making – Transparent communication

Methods:

__init__ (Line 53)

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_peer (Line 59)

def add_peer(self, peer):
    """Add a peer agent to the organization."""

Purpose: Add a peer agent to the organization.

Parameters: – self: Instance reference – peer: Parameter for peer

request_collaboration (Line 64)

def request_collaboration(self, requester_name, task, required_expertise):
    """Request collaboration from peers."""

Purpose: Request collaboration from peers.

Parameters: – self: Instance reference – requester_name: Parameter for requester_name – task: Parameter for task – required_expertise: Parameter for required_expertise

assign_task (Line 84)

def assign_task(self, task):
    """Peer self-selects task based on expertise and availability."""

Purpose: Peer self-selects task based on expertise and availability.

Parameters: – self: Instance reference – task: Parameter for task

make_shared_decision (Line 116)

def make_shared_decision(self, decision_topic):
    """All peers contribute to decision-making equally."""

Purpose: All peers contribute to decision-making equally.

Parameters: – self: Instance reference – decision_topic: Parameter for decision_topic

complete_task (Line 152)

def complete_task(self, peer_name, task):
    """Mark task as completed for a peer."""

Purpose: Mark task as completed for a peer.

Parameters: – self: Instance reference – peer_name: Parameter for peer_name – task: Parameter for task

get_collaboration_network (Line 160)

def get_collaboration_network(self):
    """Get peer-to-peer collaboration network."""

Purpose: Get peer-to-peer collaboration network.

Parameters: – self: Instance reference

run (Line 167)

def run(self, scenario):
    """Execute flat organization scenario."""

Purpose: Execute flat organization scenario.

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 = PeerAgent(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

    """
Pattern 31: Hierarchical Agent Pattern

Agents organized in management hierarchy with supervisors and subordinates.
Tasks flow down hierarchy, results flow up. Supports delegation and escalation.
"""

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


class HierarchyLevel(Enum):
    """Level in organizational hierarchy."""
    EXECUTIVE = "executive"
    MANAGER = "manager"
    SUPERVISOR = "supervisor"
    WORKER = "worker"


@dataclass
class HierarchicalTask:
    """Task in hierarchical system."""
    task_id: str
    description: str
    level: HierarchyLevel
    assigned_to: Optional[str] = None
    delegated_from: Optional[str] = None
    status: str = "pending"
    result: Optional[str] = None


@dataclass
class HierarchicalAgent:
    """Agent in hierarchy."""
    name: str
    level: HierarchyLevel
    role: str
    supervisor: Optional[str] = None
    subordinates: List[str] = field(default_factory=list)
    span_of_control: int = 5  # Max direct reports


class HierarchicalAgentPattern:
    """
    Hierarchical Agent Pattern - Management hierarchy and delegation.
    
    Agents are organized in tree structure with clear reporting lines.
    Executives delegate to managers, managers to supervisors, supervisors to workers.
    Results and escalations flow up the chain of command.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.agents: Dict[str, HierarchicalAgent] = {}
        self.tasks: List[HierarchicalTask] = []
        self.organization_chart: Dict[str, List[str]] = {}
    
    def add_agent(self, agent: HierarchicalAgent):
        """Add agent to hierarchy."""
        self.agents[agent.name] = agent
        
        # Update organization chart
        if agent.supervisor:
            if agent.supervisor not in self.organization_chart:
                self.organization_chart[agent.supervisor] = []
            self.organization_chart[agent.supervisor].append(agent.name)
    
    def delegate_task(self, task: HierarchicalTask, from_agent: str) -> bool:
        """Delegate task down the hierarchy."""
        if from_agent not in self.agents:
            return False
        
        supervisor = self.agents[from_agent]
        
        # Find appropriate subordinate
        if from_agent in self.organization_chart:
            subordinates = self.organization_chart[from_agent]
            if subordinates:
                # Delegate to first available subordinate
                subordinate_name = subordinates[0]
                task.assigned_to = subordinate_name
                task.delegated_from = from_agent
                task.status = "delegated"
                
                print(f"   - {from_agent} delegated to {subordinate_name}: {task.description}")
                return True
        
        # If no subordinates, execute task self
        task.assigned_to = from_agent
        task.status = "assigned"
        return True
    
    def execute_task(self, task: HierarchicalTask) -> str:
        """Execute task using LLM."""
        if not task.assigned_to or task.assigned_to not in self.agents:
            return "Error: No agent assigned"
        
        agent = self.agents[task.assigned_to]
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {agent.name}, a {agent.level.value} level {agent.role} in the organization. "
                      f"Supervisor: {agent.supervisor or 'None (top level)'}"),
            ("human", f"Task: {task.description}\n"
                     f"Delegated from: {task.delegated_from or 'Initial assignment'}\n\n"
                     f"Provide a brief execution result (2-3 sentences).")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        result = str(response.content)[:200]
        task.result = result
        task.status = "completed"
        
        return result
    
    def escalate_issue(self, issue: str, from_agent: str) -> Optional[str]:
        """Escalate issue up the hierarchy."""
        if from_agent not in self.agents:
            return None
        
        agent = self.agents[from_agent]
        
        if agent.supervisor:
            print(f"   ⬆️ {from_agent} escalated to {agent.supervisor}: {issue[:50]}...")
            
            # Supervisor addresses issue
            supervisor = self.agents[agent.supervisor]
            
            prompt = ChatPromptTemplate.from_messages([
                ("system", f"You are {supervisor.name}, a {supervisor.level.value} handling an escalated issue."),
                ("human", f"Issue from {from_agent}: {issue}\n\nProvide guidance (brief).")
            ])
            
            chain = prompt | self.llm
            response = chain.invoke({})
            
            return str(response.content)[:150]
        
        return "No supervisor to escalate to"
    
    def process_hierarchical_work(self, initial_task: HierarchicalTask) -> Dict[str, Any]:
        """Process work through hierarchy."""
        print(f"\n   Processing: {initial_task.description}")
        
        # Add task
        self.tasks.append(initial_task)
        
        # Delegate down hierarchy
        current_agent = initial_task.assigned_to
        if current_agent:
            delegated = self.delegate_task(initial_task, current_agent)
            
            if delegated and initial_task.assigned_to:
                # Execute at appropriate level
                result = self.execute_task(initial_task)
                print(f"   ✓ Completed by {initial_task.assigned_to}")
                
                return {
                    "success": True,
                    "task_id": initial_task.task_id,
                    "assigned_to": initial_task.assigned_to,
                    "result": result
                }
        
        return {"success": False, "error": "Task delegation failed"}
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute hierarchical agent pattern."""
        completed_tasks = sum(1 for t in self.tasks if t.status == "completed")
        
        # Calculate hierarchy depth
        max_depth = 0
        for agent in self.agents.values():
            depth = 0
            current = agent
            while current.supervisor:
                depth += 1
                if current.supervisor in self.agents:
                    current = self.agents[current.supervisor]
                else:
                    break
            max_depth = max(max_depth, depth)
        
        return {
            "pattern": "HierarchicalAgent",
            "total_agents": len(self.agents),
            "hierarchy_depth": max_depth + 1,
            "total_tasks": len(self.tasks),
            "completed_tasks": completed_tasks,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Hierarchical Agent Pattern Demo ===\n")
    
    # Create pattern
    pattern = HierarchicalAgentPattern()
    
    # Build organizational hierarchy
    print("1. Building organizational hierarchy...")
    
    # Executive level
    ceo = HierarchicalAgent("Alice", HierarchyLevel.EXECUTIVE, "CEO")
    pattern.add_agent(ceo)
    
    # Manager level
    cto = HierarchicalAgent("Bob", HierarchyLevel.MANAGER, "CTO", supervisor="Alice")
    cmo = HierarchicalAgent("Carol", HierarchyLevel.MANAGER, "CMO", supervisor="Alice")
    pattern.add_agent(cto)
    pattern.add_agent(cmo)
    
    # Supervisor level
    eng_lead = HierarchicalAgent("Dave", HierarchyLevel.SUPERVISOR, "Engineering Lead", supervisor="Bob")
    marketing_lead = HierarchicalAgent("Eve", HierarchyLevel.SUPERVISOR, "Marketing Lead", supervisor="Carol")
    pattern.add_agent(eng_lead)
    pattern.add_agent(marketing_lead)
    
    # Worker level
    dev1 = HierarchicalAgent("Frank", HierarchyLevel.WORKER, "Developer", supervisor="Dave")
    dev2 = HierarchicalAgent("Grace", HierarchyLevel.WORKER, "Developer", supervisor="Dave")
    marketer = HierarchicalAgent("Henry", HierarchyLevel.WORKER, "Marketing Specialist", supervisor="Eve")
    pattern.add_agent(dev1)
    pattern.add_agent(dev2)
    pattern.add_agent(marketer)
    
    # Create and process tasks
    print("\n2. Processing tasks through hierarchy...")
    
    task1 = HierarchicalTask(
        task_id="t1",
        description="Develop new product feature",
        level=HierarchyLevel.EXECUTIVE,
        assigned_to="Alice"
    )
    
    task2 = HierarchicalTask(
        task_id="t2",
        description="Launch marketing campaign",
        level=HierarchyLevel.MANAGER,
        assigned_to="Carol"
    )
    
    result1 = pattern.process_hierarchical_work(task1)
    result2 = pattern.process_hierarchical_work(task2)
    
    # Escalation example
    print("\n3. Escalation example...")
    guidance = pattern.escalate_issue(
        "Technical blocker on API integration",
        "Frank"
    )
    if guidance:
        print(f"   Guidance: {guidance[:80]}...")
    
    # Summary
    print("\n4. Organizational Summary:")
    summary = pattern.run("Corporate project management")
    print(f"   Total Agents: {summary['total_agents']}")
    print(f"   Hierarchy Depth: {summary['hierarchy_depth']} levels")
    print(f"   Tasks Processed: {summary['total_tasks']}")
    print(f"   Completed: {summary['completed_tasks']}")
    
    print("\n✓ Hierarchical pattern organized agents in management structure!")

    

Agentic Architecture Pattern: Hierarchical Agent

Pattern 31: 31 Hierarchical Agent – Line-by-Line Explanation

Overview

Pattern 31: Hierarchical Agent Pattern

Agents organized in management hierarchy with supervisors and subordinates. Tasks flow down hierarchy, results flow up. Supports delegation and escalation.

Code Structure Analysis

File Information

• Pattern Number: 31
• Pattern Name: 31 Hierarchical Agent
• File Path: patterns\organizational\31_hierarchical_agent.py
• Total Lines: 271

Detailed Code Explanation

Module Docstring

"""
Pattern 31: Hierarchical Agent Pattern

Agents organized in management hierarchy with supervisors and subordinates.
Tasks flow down hierarchy, results flow up. Supports delegation and escalation.
"""

Explanation: This module docstring describes the purpose and key features of the 31 Hierarchical Agent 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  # 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

Enumerations

HierarchyLevel (Line 16)

class HierarchyLevel(Enum):
    EXECUTIVE
    MANAGER
    SUPERVISOR
    WORKER

Purpose: Level in organizational hierarchy.

Values: – EXECUTIVE: Represents executive – MANAGER: Represents manager – SUPERVISOR: Represents supervisor – WORKER: Represents worker

Classes

HierarchyLevel (Line 16)

Description: Level in organizational hierarchy.

HierarchicalTask (Line 25)

Description: Task in hierarchical system.

HierarchicalAgent (Line 37)

Description: Agent in hierarchy.

HierarchicalAgentPattern (Line 47)

Description: Hierarchical Agent Pattern – Management hierarchy and delegation.

Agents are organized in tree structure with clear reporting lines. Executives delegate to managers, managers to supervisors, supervisors to workers. Results and escalations flow up the chain of command.

Methods:

__init__ (Line 56)

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_agent (Line 62)

def add_agent(self, agent):
    """Add agent to hierarchy."""

Purpose: Add agent to hierarchy.

Parameters: – self: Instance reference – agent: Parameter for agent

delegate_task (Line 72)

def delegate_task(self, task, from_agent):
    """Delegate task down the hierarchy."""

Purpose: Delegate task down the hierarchy.

Parameters: – self: Instance reference – task: Parameter for task – from_agent: Parameter for from_agent

execute_task (Line 97)

def execute_task(self, task):
    """Execute task using LLM."""

Purpose: Execute task using LLM.

Parameters: – self: Instance reference – task: Parameter for task

escalate_issue (Line 121)

def escalate_issue(self, issue, from_agent):
    """Escalate issue up the hierarchy."""

Purpose: Escalate issue up the hierarchy.

Parameters: – self: Instance reference – issue: Parameter for issue – from_agent: Parameter for from_agent

process_hierarchical_work (Line 146)

def process_hierarchical_work(self, initial_task):
    """Process work through hierarchy."""

Purpose: Process work through hierarchy.

Parameters: – self: Instance reference – initial_task: Parameter for initial_task

run (Line 172)

def run(self, scenario):
    """Execute hierarchical agent pattern."""

Purpose: Execute hierarchical agent 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 = HierarchyLevel(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

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

    """
Pattern 29: Consensus Pattern

Agents reach agreement through iterative consensus-building mechanisms.
Supports multiple consensus algorithms: unanimous, majority, quorum-based.
"""

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 ConsensusType(Enum):
    """Type of consensus mechanism."""
    UNANIMOUS = "unanimous"  # All must agree
    MAJORITY = "majority"  # >50% must agree
    SUPERMAJORITY = "supermajority"  # 2/3 must agree
    QUORUM = "quorum"  # Minimum number must participate


class ProposalStatus(Enum):
    """Status of consensus proposal."""
    PROPOSED = "proposed"
    DISCUSSING = "discussing"
    VOTING = "voting"
    ACCEPTED = "accepted"
    REJECTED = "rejected"
    NO_CONSENSUS = "no_consensus"


@dataclass
class Proposal:
    """Proposal for consensus."""
    proposal_id: str
    proposer: str
    content: str
    status: ProposalStatus = ProposalStatus.PROPOSED
    votes_for: Set[str] = field(default_factory=set)
    votes_against: Set[str] = field(default_factory=set)
    abstentions: Set[str] = field(default_factory=set)
    discussion_rounds: int = 0
    created_at: datetime = field(default_factory=datetime.now)


@dataclass
class ConsensusRound:
    """Round of consensus building."""
    round_id: int
    proposal_id: str
    participants: List[str]
    votes_cast: int
    consensus_reached: bool
    timestamp: datetime = field(default_factory=datetime.now)


class ConsensusAgent:
    """Agent participating in consensus."""
    
    def __init__(self, name: str, model_name: str = "gpt-4"):
        self.name = name
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.proposals_voted: Set[str] = set()
    
    def propose(self, proposal_id: str, content: str) -> Proposal:
        """Make a proposal."""
        proposal = Proposal(
            proposal_id=proposal_id,
            proposer=self.name,
            content=content
        )
        print(f"   📋 {self.name} proposed: {content}")
        return proposal
    
    def evaluate_proposal(self, proposal: Proposal, context: str = "") -> str:
        """Evaluate a proposal and provide stance."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, evaluating a proposal for consensus. "
                      f"Consider the proposal carefully and provide your stance."),
            ("human", f"Proposal: {proposal.content}\n"
                     f"Proposer: {proposal.proposer}\n"
                     f"Context: {context}\n\n"
                     f"Do you support this proposal? Answer with 'support', 'oppose', or 'abstain' "
                     f"and explain why in 1-2 sentences.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        return str(response.content)[:200]
    
    def vote(self, proposal: Proposal, decision: str):
        """Cast vote on proposal."""
        self.proposals_voted.add(proposal.proposal_id)
        
        if "support" in decision.lower():
            proposal.votes_for.add(self.name)
            print(f"   ✅ {self.name} voted FOR")
        elif "oppose" in decision.lower():
            proposal.votes_against.add(self.name)
            print(f"   ❌ {self.name} voted AGAINST")
        else:
            proposal.abstentions.add(self.name)
            print(f"   ⚪ {self.name} ABSTAINED")


class ConsensusPattern:
    """
    Consensus Pattern - Agreement through iterative consensus.
    
    Agents work together to reach consensus on proposals through
    discussion, voting, and iterative refinement.
    """
    
    def __init__(self, consensus_type: ConsensusType = ConsensusType.MAJORITY,
                 quorum: Optional[int] = None):
        self.consensus_type = consensus_type
        self.quorum = quorum
        self.agents: Dict[str, ConsensusAgent] = {}
        self.proposals: Dict[str, Proposal] = {}
        self.consensus_rounds: List[ConsensusRound] = []
    
    def register_agent(self, agent: ConsensusAgent):
        """Register agent in consensus group."""
        self.agents[agent.name] = agent
    
    def submit_proposal(self, proposal: Proposal):
        """Submit proposal for consensus."""
        self.proposals[proposal.proposal_id] = proposal
        proposal.status = ProposalStatus.DISCUSSING
    
    def conduct_consensus(self, proposal_id: str, max_rounds: int = 3) -> bool:
        """Conduct consensus building rounds."""
        if proposal_id not in self.proposals:
            return False
        
        proposal = self.proposals[proposal_id]
        proposal.status = ProposalStatus.VOTING
        
        print(f"\n   🤝 Starting consensus on: {proposal.content}")
        print(f"   Consensus type: {self.consensus_type.value}")
        
        for round_num in range(max_rounds):
            print(f"\n   === Round {round_num + 1} ===")
            proposal.discussion_rounds += 1
            
            # Each agent evaluates and votes
            for agent_name, agent in self.agents.items():
                evaluation = agent.evaluate_proposal(proposal)
                agent.vote(proposal, evaluation)
            
            # Check consensus
            consensus_reached = self._check_consensus(proposal)
            
            # Record round
            consensus_round = ConsensusRound(
                round_id=round_num,
                proposal_id=proposal_id,
                participants=list(self.agents.keys()),
                votes_cast=len(proposal.votes_for) + len(proposal.votes_against),
                consensus_reached=consensus_reached
            )
            self.consensus_rounds.append(consensus_round)
            
            if consensus_reached:
                proposal.status = ProposalStatus.ACCEPTED
                print(f"\n   ✅ Consensus REACHED after {round_num + 1} rounds!")
                return True
        
        proposal.status = ProposalStatus.NO_CONSENSUS
        print(f"\n   ❌ No consensus after {max_rounds} rounds")
        return False
    
    def _check_consensus(self, proposal: Proposal) -> bool:
        """Check if consensus is reached."""
        total_agents = len(self.agents)
        votes_for = len(proposal.votes_for)
        votes_against = len(proposal.votes_against)
        votes_cast = votes_for + votes_against
        
        print(f"      Votes: {votes_for} for, {votes_against} against, "
              f"{len(proposal.abstentions)} abstain")
        
        # Check quorum if required
        if self.quorum and votes_cast  total_agents / 2
        
        elif self.consensus_type == ConsensusType.SUPERMAJORITY:
            return votes_for >= (total_agents * 2 / 3)
        
        elif self.consensus_type == ConsensusType.QUORUM:
            return votes_cast >= (self.quorum or total_agents)
        
        return False
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute consensus pattern."""
        total_proposals = len(self.proposals)
        accepted = sum(1 for p in self.proposals.values() 
                      if p.status == ProposalStatus.ACCEPTED)
        
        return {
            "pattern": "Consensus",
            "scenario": scenario,
            "consensus_type": self.consensus_type.value,
            "total_agents": len(self.agents),
            "total_proposals": total_proposals,
            "accepted_proposals": accepted,
            "consensus_rounds": len(self.consensus_rounds)
        }


# Example usage
if __name__ == "__main__":
    print("=== Consensus Pattern Demo ===\n")
    
    # Create consensus group (majority rule)
    print("1. Creating consensus group (majority rule)...")
    pattern = ConsensusPattern(consensus_type=ConsensusType.MAJORITY)
    
    # Register agents
    agents = [
        ConsensusAgent("Alice"),
        ConsensusAgent("Bob"),
        ConsensusAgent("Carol"),
        ConsensusAgent("Dave"),
        ConsensusAgent("Eve")
    ]
    
    for agent in agents:
        pattern.register_agent(agent)
    
    print(f"   Registered {len(agents)} agents")
    
    # Proposal 1: Should be accepted
    print("\n2. Proposal 1: Adopt new AI safety guidelines")
    proposal1 = agents[0].propose("prop1", "Adopt comprehensive AI safety guidelines for all projects")
    pattern.submit_proposal(proposal1)
    result1 = pattern.conduct_consensus("prop1", max_rounds=1)
    
    # Proposal 2: Different proposal
    print("\n3. Proposal 2: Increase project budget by 50%")
    proposal2 = agents[1].propose("prop2", "Increase project budget by 50% to accelerate development")
    pattern.submit_proposal(proposal2)
    result2 = pattern.conduct_consensus("prop2", max_rounds=1)
    
    # Proposal 3: Another proposal
    print("\n4. Proposal 3: Switch to open-source model")
    proposal3 = agents[2].propose("prop3", "Switch primary AI model from proprietary to open-source")
    pattern.submit_proposal(proposal3)
    result3 = pattern.conduct_consensus("prop3", max_rounds=1)
    
    # Try unanimous consensus
    print("\n\n5. Testing UNANIMOUS consensus (harder to achieve)...")
    unanimous_pattern = ConsensusPattern(consensus_type=ConsensusType.UNANIMOUS)
    
    for agent in agents[:3]:  # Smaller group for unanimous
        unanimous_pattern.register_agent(agent)
    
    proposal4 = agents[0].propose("prop4", "Schedule team meeting for next Monday at 2 PM")
    unanimous_pattern.submit_proposal(proposal4)
    result4 = unanimous_pattern.conduct_consensus("prop4", max_rounds=1)
    
    # Statistics
    print("\n\n6. Consensus Statistics:")
    summary = pattern.run("Team decision making")
    print(f"   Total Proposals: {summary['total_proposals']}")
    print(f"   Accepted: {summary['accepted_proposals']}")
    print(f"   Rejection rate: {((summary['total_proposals'] - summary['accepted_proposals']) / max(1, summary['total_proposals']) * 100):.1f}%")
    print(f"   Consensus rounds: {summary['consensus_rounds']}")
    
    # Vote distribution
    print("\n7. Vote Distribution:")
    for prop_id, proposal in pattern.proposals.items():
        print(f"   {prop_id}: {len(proposal.votes_for)} for, "
              f"{len(proposal.votes_against)} against, "
              f"{len(proposal.abstentions)} abstain -> {proposal.status.value}")
    
    print("\n✓ Consensus pattern enabled collaborative decision making!")

    

Agentic Architecture Pattern: Consensus

Pattern 29: 29 Consensus – Line-by-Line Explanation

Overview

Pattern 29: Consensus Pattern

Agents reach agreement through iterative consensus-building mechanisms. Supports multiple consensus algorithms: unanimous, majority, quorum-based.

Code Structure Analysis

File Information

• Pattern Number: 29
• Pattern Name: 29 Consensus
• File Path: patterns\collaboration\29_consensus.py
• Total Lines: 289

Detailed Code Explanation

Module Docstring

"""
Pattern 29: Consensus Pattern

Agents reach agreement through iterative consensus-building mechanisms.
Supports multiple consensus algorithms: unanimous, majority, quorum-based.
"""

Explanation: This module docstring describes the purpose and key features of the 29 Consensus pattern.

Imports Section

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  # 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: 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

Enumerations

ConsensusType (Line 16)

class ConsensusType(Enum):
    UNANIMOUS
    MAJORITY
    SUPERMAJORITY
    QUORUM

Purpose: Type of consensus mechanism.

Values: – UNANIMOUS: Represents unanimous – MAJORITY: Represents majority – SUPERMAJORITY: Represents supermajority – QUORUM: Represents quorum

ProposalStatus (Line 24)

class ProposalStatus(Enum):
    PROPOSED
    DISCUSSING
    VOTING
    ACCEPTED
    REJECTED
    NO_CONSENSUS

Purpose: Status of consensus proposal.

Values: – PROPOSED: Represents proposed – DISCUSSING: Represents discussing – VOTING: Represents voting – ACCEPTED: Represents accepted – REJECTED: Represents rejected – NO_CONSENSUS: Represents no consensus

Classes

ConsensusType (Line 16)

Description: Type of consensus mechanism.

ProposalStatus (Line 24)

Description: Status of consensus proposal.

Proposal (Line 35)

Description: Proposal for consensus.

ConsensusRound (Line 49)

Description: Round of consensus building.

ConsensusAgent (Line 59)

Description: Agent participating in consensus.

Methods:

__init__ (Line 62)

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”)

propose (Line 67)

def propose(self, proposal_id, content):
    """Make a proposal."""

Purpose: Make a proposal.

Parameters: – self: Instance reference – proposal_id: Parameter for proposal_id – content: Parameter for content

evaluate_proposal (Line 77)

def evaluate_proposal(self, proposal, context):
    """Evaluate a proposal and provide stance."""

Purpose: Evaluate a proposal and provide stance.

Parameters: – self: Instance reference – proposal: Parameter for proposal – context: Parameter for context

vote (Line 94)

def vote(self, proposal, decision):
    """Cast vote on proposal."""

Purpose: Cast vote on proposal.

Parameters: – self: Instance reference – proposal: Parameter for proposal – decision: Parameter for decision

ConsensusPattern (Line 109)

Description: Consensus Pattern – Agreement through iterative consensus.

Agents work together to reach consensus on proposals through discussion, voting, and iterative refinement.

Methods:

__init__ (Line 117)

def __init__(self, consensus_type, quorum):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – consensus_type: Parameter for consensus_type – quorum: Parameter for quorum

register_agent (Line 125)

def register_agent(self, agent):
    """Register agent in consensus group."""

Purpose: Register agent in consensus group.

Parameters: – self: Instance reference – agent: Parameter for agent

submit_proposal (Line 129)

def submit_proposal(self, proposal):
    """Submit proposal for consensus."""

Purpose: Submit proposal for consensus.

Parameters: – self: Instance reference – proposal: Parameter for proposal

conduct_consensus (Line 134)

def conduct_consensus(self, proposal_id, max_rounds):
    """Conduct consensus building rounds."""

Purpose: Conduct consensus building rounds.

Parameters: – self: Instance reference – proposal_id: Parameter for proposal_id – max_rounds: Parameter for max_rounds

_check_consensus (Line 176)

def _check_consensus(self, proposal):
    """Check if consensus is reached."""

Purpose: Check if consensus is reached.

Parameters: – self: Instance reference – proposal: Parameter for proposal

run (Line 205)

def run(self, scenario):
    """Execute consensus pattern."""

Purpose: Execute consensus 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 = ConsensusType(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

    """
Pattern 28: Swarm Intelligence Pattern

Emergent collective behavior from simple agent interactions.
Inspired by natural swarms (bees, ants, birds) - local interactions lead to global solutions.
"""

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
import random
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


@dataclass
class SwarmAgent:
    """Individual agent in the swarm."""
    agent_id: str
    position: Tuple[float, float]  # 2D position
    velocity: Tuple[float, float]  # Current velocity
    best_position: Tuple[float, float]  # Personal best
    best_score: float = float('-inf')
    neighbors: List[str] = field(default_factory=list)


@dataclass
class SwarmProblem:
    """Problem to be solved by swarm."""
    problem_id: str
    description: str
    search_space: Tuple[Tuple[float, float], Tuple[float, float]]  # ((x_min, x_max), (y_min, y_max))
    goal: str


@dataclass
class SwarmSolution:
    """Solution found by swarm."""
    problem_id: str
    best_position: Tuple[float, float]
    best_score: float
    iterations: int
    convergence_history: List[float]


class SwarmIntelligencePattern:
    """
    Swarm Intelligence Pattern - Emergent collective optimization.
    
    Multiple simple agents follow local rules and interact with neighbors.
    Global optimal solutions emerge from collective behavior.
    Inspired by Particle Swarm Optimization (PSO).
    """
    
    def __init__(self, num_agents: int = 20, inertia: float = 0.7, 
                 cognitive: float = 1.5, social: float = 1.5, model_name: str = "gpt-4"):
        self.num_agents = num_agents
        self.inertia = inertia  # Weight for current velocity
        self.cognitive = cognitive  # Weight for personal best
        self.social = social  # Weight for global best
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        
        self.swarm: List[SwarmAgent] = []
        self.global_best_position: Optional[Tuple[float, float]] = None
        self.global_best_score: float = float('-inf')
    
    def initialize_swarm(self, search_space: Tuple[Tuple[float, float], Tuple[float, float]]):
        """Initialize swarm agents in search space."""
        self.swarm = []
        (x_min, x_max), (y_min, y_max) = search_space
        
        for i in range(self.num_agents):
            position = (
                random.uniform(x_min, x_max),
                random.uniform(y_min, y_max)
            )
            velocity = (
                random.uniform(-1, 1),
                random.uniform(-1, 1)
            )
            
            agent = SwarmAgent(
                agent_id=f"agent_{i}",
                position=position,
                velocity=velocity,
                best_position=position
            )
            self.swarm.append(agent)
    
    def evaluate_position(self, position: Tuple[float, float], problem: SwarmProblem) -> float:
        """
        Evaluate fitness of position for problem.
        Uses LLM to score position quality.
        """
        x, y = position
        
        # For demo: simple optimization function (can be replaced with LLM evaluation)
        # Maximize: -(x^2 + y^2) (peak at origin)
        score = -(x**2 + y**2)
        
        return score
    
    def update_velocity(self, agent: SwarmAgent) -> Tuple[float, float]:
        """Update agent velocity based on personal and global best."""
        if not self.global_best_position:
            return agent.velocity
        
        vx, vy = agent.velocity
        x, y = agent.position
        pbx, pby = agent.best_position
        gbx, gby = self.global_best_position
        
        # PSO velocity update formula
        r1, r2 = random.random(), random.random()
        
        new_vx = (self.inertia * vx + 
                  self.cognitive * r1 * (pbx - x) + 
                  self.social * r2 * (gbx - x))
        
        new_vy = (self.inertia * vy + 
                  self.cognitive * r1 * (pby - y) + 
                  self.social * r2 * (gby - y))
        
        # Limit velocity
        max_velocity = 2.0
        new_vx = max(-max_velocity, min(max_velocity, new_vx))
        new_vy = max(-max_velocity, min(max_velocity, new_vy))
        
        return (new_vx, new_vy)
    
    def update_position(self, agent: SwarmAgent, search_space: Tuple[Tuple[float, float], Tuple[float, float]]):
        """Update agent position based on velocity."""
        (x_min, x_max), (y_min, y_max) = search_space
        
        x, y = agent.position
        vx, vy = agent.velocity
        
        new_x = x + vx
        new_y = y + vy
        
        # Keep in bounds
        new_x = max(x_min, min(x_max, new_x))
        new_y = max(y_min, min(y_max, new_y))
        
        agent.position = (new_x, new_y)
    
    def solve(self, problem: SwarmProblem, max_iterations: int = 30) -> SwarmSolution:
        """Solve problem using swarm intelligence."""
        print(f"\n   Problem: {problem.description}")
        print(f"   Swarm size: {self.num_agents} agents")
        print(f"   Search space: {problem.search_space}")
        
        # Initialize swarm
        self.initialize_swarm(problem.search_space)
        convergence_history = []
        
        # Optimize over iterations
        for iteration in range(max_iterations):
            # Evaluate all agents
            for agent in self.swarm:
                score = self.evaluate_position(agent.position, problem)
                
                # Update personal best
                if score > agent.best_score:
                    agent.best_score = score
                    agent.best_position = agent.position
                
                # Update global best
                if score > self.global_best_score:
                    self.global_best_score = score
                    self.global_best_position = agent.position
            
            # Update velocities and positions
            for agent in self.swarm:
                agent.velocity = self.update_velocity(agent)
                self.update_position(agent, problem.search_space)
            
            convergence_history.append(self.global_best_score)
            
            if iteration % 10 == 0:
                print(f"   Iteration {iteration}: Best score = {self.global_best_score:.4f}")
        
        print(f"   ✓ Final best position: {self.global_best_position}")
        print(f"   ✓ Final best score: {self.global_best_score:.4f}")
        
        solution = SwarmSolution(
            problem_id=problem.problem_id,
            best_position=self.global_best_position or (0, 0),
            best_score=self.global_best_score,
            iterations=max_iterations,
            convergence_history=convergence_history
        )
        
        return solution
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute swarm intelligence pattern."""
        return {
            "pattern": "SwarmIntelligence",
            "swarm_size": self.num_agents,
            "inertia": self.inertia,
            "cognitive_weight": self.cognitive,
            "social_weight": self.social,
            "global_best_score": self.global_best_score,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Swarm Intelligence Pattern Demo ===\n")
    
    # Create swarm pattern
    print("1. Creating swarm...")
    pattern = SwarmIntelligencePattern(
        num_agents=25,
        inertia=0.7,
        cognitive=1.5,
        social=1.5
    )
    
    # Define optimization problem
    print("\n2. Defining optimization problem...")
    problem = SwarmProblem(
        problem_id="optimize1",
        description="Find maximum point in 2D space (minimize distance from origin)",
        search_space=((-10, 10), (-10, 10)),
        goal="Find position closest to origin (0, 0)"
    )
    
    # Solve using swarm
    print("\n3. Solving with swarm intelligence...")
    solution = pattern.solve(problem, max_iterations=30)
    
    # Summary
    print("\n4. Solution Summary:")
    print(f"   Best Position: ({solution.best_position[0]:.4f}, {solution.best_position[1]:.4f})")
    print(f"   Best Score: {solution.best_score:.4f}")
    print(f"   Iterations: {solution.iterations}")
    print(f"   Improvement: {solution.convergence_history[-1] - solution.convergence_history[0]:.4f}")
    
    summary = pattern.run("2D optimization via swarm")
    print(f"\n5. Pattern Configuration:")
    print(f"   Swarm Size: {summary['swarm_size']}")
    print(f"   Inertia: {summary['inertia']}")
    print(f"   Cognitive: {summary['cognitive_weight']}")
    print(f"   Social: {summary['social_weight']}")
    
    print("\n✓ Swarm intelligence found optimal solution through emergent behavior!")

    

Agentic Architecture Pattern: Swarm Intelligence

Pattern 28: 28 Swarm Intelligence – Line-by-Line Explanation

Overview

Pattern 28: Swarm Intelligence Pattern

Emergent collective behavior from simple agent interactions. Inspired by natural swarms (bees, ants, birds) – local interactions lead to global solutions.

Code Structure Analysis

File Information

• Pattern Number: 28
• Pattern Name: 28 Swarm Intelligence
• File Path: patterns\collaboration\28_swarm_intelligence.py
• Total Lines: 250

Detailed Code Explanation

Module Docstring

"""
Pattern 28: Swarm Intelligence Pattern

Emergent collective behavior from simple agent interactions.
Inspired by natural swarms (bees, ants, birds) - local interactions lead to global solutions.
"""

Explanation: This module docstring describes the purpose and key features of the 28 Swarm Intelligence pattern.

Imports Section

from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
import random
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # First 20 imports shown

Explanation: The module imports necessary dependencies: – from typing import Dict, Any, List, Optional, Tuple: 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 langchain_core.prompts import ChatPromptTemplate: LangChain framework integration – from langchain_openai import ChatOpenAI: LangChain framework integration

Classes

SwarmAgent (Line 17)

Description: Individual agent in the swarm.

SwarmProblem (Line 28)

Description: Problem to be solved by swarm.

SwarmSolution (Line 37)

Description: Solution found by swarm.

SwarmIntelligencePattern (Line 46)

Description: Swarm Intelligence Pattern – Emergent collective optimization.

Multiple simple agents follow local rules and interact with neighbors. Global optimal solutions emerge from collective behavior. Inspired by Particle Swarm Optimization (PSO).

Methods:

__init__ (Line 55)

def __init__(self, num_agents, inertia, cognitive, social, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – num_agents: Parameter for num_agents – inertia: Parameter for inertia – cognitive: Parameter for cognitive – social: Parameter for social – model_name: LLM model to use (e.g., “gpt-4”)

initialize_swarm (Line 67)

def initialize_swarm(self, search_space):
    """Initialize swarm agents in search space."""

Purpose: Initialize swarm agents in search space.

Parameters: – self: Instance reference – search_space: Parameter for search_space

evaluate_position (Line 90)

def evaluate_position(self, position, problem):
    """Evaluate fitness of position for problem.
Uses LLM to score position quality."""

Purpose: Evaluate fitness of position for problem. Uses LLM to score position quality.

Parameters: – self: Instance reference – position: Parameter for position – problem: Parameter for problem

update_velocity (Line 103)

def update_velocity(self, agent):
    """Update agent velocity based on personal and global best."""

Purpose: Update agent velocity based on personal and global best.

Parameters: – self: Instance reference – agent: Parameter for agent

update_position (Line 131)

def update_position(self, agent, search_space):
    """Update agent position based on velocity."""

Purpose: Update agent position based on velocity.

Parameters: – self: Instance reference – agent: Parameter for agent – search_space: Parameter for search_space

solve (Line 147)

def solve(self, problem, max_iterations):
    """Solve problem using swarm intelligence."""

Purpose: Solve problem using swarm intelligence.

Parameters: – self: Instance reference – problem: Parameter for problem – max_iterations: Parameter for max_iterations

run (Line 196)

def run(self, scenario):
    """Execute swarm intelligence pattern."""

Purpose: Execute swarm intelligence 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 = SwarmAgent(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

    """
Pattern 27: Team-Based Pattern

Agents organized into formal teams with defined roles and responsibilities.
Teams have leaders, members, and coordinate to achieve team goals.
"""

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


class TeamRole(Enum):
    """Role within a team."""
    LEADER = "leader"
    MEMBER = "member"
    SPECIALIST = "specialist"
    COORDINATOR = "coordinator"


@dataclass
class TeamMember:
    """Member of a team."""
    name: str
    role: TeamRole
    skills: List[str]
    availability: float = 1.0  # 0.0 to 1.0
    tasks_assigned: List[str] = field(default_factory=list)


@dataclass
class TeamTask:
    """Task assigned to team."""
    task_id: str
    description: str
    required_skills: List[str]
    assigned_to: Optional[str] = None
    status: str = "pending"  # pending, in_progress, completed
    priority: int = 5  # 1-10


@dataclass
class Team:
    """Team of agents."""
    team_id: str
    name: str
    purpose: str
    members: List[TeamMember] = field(default_factory=list)
    tasks: List[TeamTask] = field(default_factory=list)
    achievements: List[str] = field(default_factory=list)


class TeamAgent:
    """Agent that works as part of a team."""
    
    def __init__(self, name: str, role: TeamRole, skills: List[str], model_name: str = "gpt-4"):
        self.member = TeamMember(name=name, role=role, skills=skills)
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
    
    def work_on_task(self, task: TeamTask, team: Team) -> str:
        """Work on assigned task."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.member.name}, a {self.member.role.value} with skills: {', '.join(self.member.skills)}. "
                      f"You're part of team '{team.name}' working on: {team.purpose}"),
            ("human", f"Task: {task.description}\n"
                     f"Required skills: {', '.join(task.required_skills)}\n"
                     f"Priority: {task.priority}/10\n\n"
                     f"Describe your approach to completing this task (2-3 sentences).")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        return str(response.content)[:200]
    
    def collaborate_with(self, other_member: TeamMember, context: str) -> str:
        """Collaborate with another team member."""
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.member.name} collaborating with {other_member.name}."),
            ("human", f"Context: {context}\n\nHow will you work together? (brief response)")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        return str(response.content)[:150]


class TeamBasedPattern:
    """
    Team-Based Pattern - Structured team organization and coordination.
    
    Agents are organized into teams with formal roles and responsibilities.
    Teams have leaders, coordinate work, and achieve collective goals.
    """
    
    def __init__(self):
        self.teams: Dict[str, Team] = {}
        self.agents: Dict[str, TeamAgent] = {}
    
    def create_team(self, team_id: str, name: str, purpose: str) -> Team:
        """Create a new team."""
        team = Team(team_id=team_id, name=name, purpose=purpose)
        self.teams[team_id] = team
        return team
    
    def add_member_to_team(self, team_id: str, agent: TeamAgent):
        """Add agent to team."""
        if team_id in self.teams:
            self.teams[team_id].members.append(agent.member)
            self.agents[agent.member.name] = agent
    
    def assign_task(self, team_id: str, task: TeamTask) -> bool:
        """Assign task to appropriate team member."""
        if team_id not in self.teams:
            return False
        
        team = self.teams[team_id]
        team.tasks.append(task)
        
        # Find best member for task based on skills
        best_member = None
        best_match_score = 0
        
        for member in team.members:
            # Count matching skills
            matches = sum(1 for skill in task.required_skills if skill in member.skills)
            if matches > best_match_score and member.availability > 0.3:
                best_match_score = matches
                best_member = member
        
        if best_member:
            task.assigned_to = best_member.name
            task.status = "in_progress"
            best_member.tasks_assigned.append(task.task_id)
            return True
        
        return False
    
    def execute_team_work(self, team_id: str) -> Dict[str, Any]:
        """Execute work for team."""
        if team_id not in self.teams:
            return {"success": False, "error": "Team not found"}
        
        team = self.teams[team_id]
        results = []
        
        print(f"\n   Team '{team.name}' working on: {team.purpose}")
        
        # Execute tasks
        for task in team.tasks:
            if task.assigned_to and task.assigned_to in self.agents:
                agent = self.agents[task.assigned_to]
                print(f"   - {task.assigned_to} working on: {task.description}")
                
                work_result = agent.work_on_task(task, team)
                results.append({
                    "task": task.task_id,
                    "member": task.assigned_to,
                    "result": work_result
                })
                
                task.status = "completed"
                team.achievements.append(f"Completed: {task.description}")
        
        # Team collaboration
        if len(team.members) >= 2:
            member1 = team.members[0]
            member2 = team.members[1]
            if member1.name in self.agents and member2.name in self.agents:
                agent1 = self.agents[member1.name]
                collab = agent1.collaborate_with(member2, f"Team goal: {team.purpose}")
                print(f"   - Collaboration: {collab[:80]}...")
        
        return {
            "success": True,
            "team": team.name,
            "tasks_completed": sum(1 for t in team.tasks if t.status == "completed"),
            "achievements": len(team.achievements),
            "results": results
        }
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute team-based pattern."""
        total_tasks = sum(len(team.tasks) for team in self.teams.values())
        completed_tasks = sum(
            sum(1 for t in team.tasks if t.status == "completed")
            for team in self.teams.values()
        )
        
        return {
            "pattern": "TeamBased",
            "total_teams": len(self.teams),
            "total_agents": len(self.agents),
            "total_tasks": total_tasks,
            "completed_tasks": completed_tasks,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Team-Based Pattern Demo ===\n")
    
    # Create pattern
    pattern = TeamBasedPattern()
    
    # Create teams
    print("1. Creating teams...")
    dev_team = pattern.create_team("dev1", "Development Team", "Build mobile app")
    design_team = pattern.create_team("design1", "Design Team", "Create UI/UX designs")
    
    # Add members to development team
    print("\n2. Adding team members...")
    dev_lead = TeamAgent("Alice", TeamRole.LEADER, ["python", "architecture", "testing"])
    backend_dev = TeamAgent("Bob", TeamRole.MEMBER, ["python", "databases", "APIs"])
    frontend_dev = TeamAgent("Carol", TeamRole.MEMBER, ["javascript", "react", "CSS"])
    
    pattern.add_member_to_team("dev1", dev_lead)
    pattern.add_member_to_team("dev1", backend_dev)
    pattern.add_member_to_team("dev1", frontend_dev)
    
    # Add members to design team
    design_lead = TeamAgent("Dave", TeamRole.LEADER, ["UI_design", "prototyping"])
    ux_specialist = TeamAgent("Eve", TeamRole.SPECIALIST, ["UX_research", "user_testing"])
    
    pattern.add_member_to_team("design1", design_lead)
    pattern.add_member_to_team("design1", ux_specialist)
    
    # Assign tasks
    print("\n3. Assigning tasks...")
    task1 = TeamTask("t1", "Implement authentication system", ["python", "security"], priority=9)
    task2 = TeamTask("t2", "Build REST API", ["python", "APIs"], priority=8)
    task3 = TeamTask("t3", "Create responsive UI", ["javascript", "CSS"], priority=7)
    task4 = TeamTask("t4", "Design app mockups", ["UI_design", "prototyping"], priority=9)
    task5 = TeamTask("t5", "Conduct user research", ["UX_research"], priority=6)
    
    pattern.assign_task("dev1", task1)
    pattern.assign_task("dev1", task2)
    pattern.assign_task("dev1", task3)
    pattern.assign_task("design1", task4)
    pattern.assign_task("design1", task5)
    
    # Execute team work
    print("\n4. Executing team work...")
    dev_result = pattern.execute_team_work("dev1")
    design_result = pattern.execute_team_work("design1")
    
    # Summary
    print("\n5. Team Performance Summary:")
    summary = pattern.run("Multi-team software development")
    print(f"   Total Teams: {summary['total_teams']}")
    print(f"   Total Agents: {summary['total_agents']}")
    print(f"   Tasks Completed: {summary['completed_tasks']}/{summary['total_tasks']}")
    
    print("\n✓ Team-based pattern organized agents into effective teams!")

    

Agentic Architecture Pattern: Team Based

Pattern 27: 27 Team Based – Line-by-Line Explanation

Overview

Pattern 27: Team-Based Pattern

Agents organized into formal teams with defined roles and responsibilities. Teams have leaders, members, and coordinate to achieve team goals.

Code Structure Analysis

File Information

• Pattern Number: 27
• Pattern Name: 27 Team Based
• File Path: patterns\collaboration\27_team_based.py
• Total Lines: 258

Detailed Code Explanation

Module Docstring

"""
Pattern 27: Team-Based Pattern

Agents organized into formal teams with defined roles and responsibilities.
Teams have leaders, members, and coordinate to achieve team goals.
"""

Explanation: This module docstring describes the purpose and key features of the 27 Team Based 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  # 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

Enumerations

TeamRole (Line 16)

class TeamRole(Enum):
    LEADER
    MEMBER
    SPECIALIST
    COORDINATOR

Purpose: Role within a team.

Values: – LEADER: Represents leader – MEMBER: Represents member – SPECIALIST: Represents specialist – COORDINATOR: Represents coordinator

Classes

TeamRole (Line 16)

Description: Role within a team.

TeamMember (Line 25)

Description: Member of a team.

TeamTask (Line 35)

Description: Task assigned to team.

Team (Line 46)

Description: Team of agents.

TeamAgent (Line 56)

Description: Agent that works as part of a team.

Methods:

__init__ (Line 59)

def __init__(self, name, role, skills, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – role: Parameter for role – skills: Parameter for skills – model_name: LLM model to use (e.g., “gpt-4”)

work_on_task (Line 63)

def work_on_task(self, task, team):
    """Work on assigned task."""

Purpose: Work on assigned task.

Parameters: – self: Instance reference – task: Parameter for task – team: Parameter for team

collaborate_with (Line 78)

def collaborate_with(self, other_member, context):
    """Collaborate with another team member."""

Purpose: Collaborate with another team member.

Parameters: – self: Instance reference – other_member: Parameter for other_member – context: Parameter for context

TeamBasedPattern (Line 90)

Description: Team-Based Pattern – Structured team organization and coordination.

Agents are organized into teams with formal roles and responsibilities. Teams have leaders, coordinate work, and achieve collective goals.

Methods:

__init__ (Line 98)

def __init__(self):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference

create_team (Line 102)

def create_team(self, team_id, name, purpose):
    """Create a new team."""

Purpose: Create a new team.

Parameters: – self: Instance reference – team_id: Parameter for team_id – name: Identifier or label – purpose: Parameter for purpose

add_member_to_team (Line 108)

def add_member_to_team(self, team_id, agent):
    """Add agent to team."""

Purpose: Add agent to team.

Parameters: – self: Instance reference – team_id: Parameter for team_id – agent: Parameter for agent

assign_task (Line 114)

def assign_task(self, team_id, task):
    """Assign task to appropriate team member."""

Purpose: Assign task to appropriate team member.

Parameters: – self: Instance reference – team_id: Parameter for team_id – task: Parameter for task

execute_team_work (Line 141)

def execute_team_work(self, team_id):
    """Execute work for team."""

Purpose: Execute work for team.

Parameters: – self: Instance reference – team_id: Parameter for team_id

run (Line 184)

def run(self, scenario):
    """Execute team-based pattern."""

Purpose: Execute team-based 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 = TeamRole(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

    """
Pattern 26: Coalition Formation Pattern

Agents form temporary alliances (coalitions) to achieve mutual goals.
Features coalition evaluation, membership negotiation, and benefit distribution.
"""

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 CoalitionStatus(Enum):
    """Status of coalition."""
    FORMING = "forming"
    ACTIVE = "active"
    DISSOLVED = "dissolved"
    COMPLETED = "completed"


@dataclass
class CoalitionGoal:
    """Goal that coalition works toward."""
    goal_id: str
    description: str
    required_capabilities: List[str]
    reward: float
    deadline: Optional[datetime] = None


@dataclass
class Coalition:
    """Group of agents working together."""
    coalition_id: str
    name: str
    goal: CoalitionGoal
    members: Set[str] = field(default_factory=set)
    leader: Optional[str] = None
    status: CoalitionStatus = CoalitionStatus.FORMING
    total_reward: float = 0
    formed_at: datetime = field(default_factory=datetime.now)


@dataclass
class CoalitionAgent:
    """Agent that can form coalitions."""
    name: str
    capabilities: List[str]
    reputation: float = 5.0  # 0-10
    current_coalitions: Set[str] = field(default_factory=set)
    rewards_earned: float = 0


class CoalitionFormationPattern:
    """
    Coalition Formation Pattern - Dynamic alliance formation.
    
    Agents evaluate potential coalitions, negotiate membership,
    and work together to achieve shared goals with reward distribution.
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.agents: Dict[str, CoalitionAgent] = {}
        self.coalitions: Dict[str, Coalition] = {}
        self.goals: List[CoalitionGoal] = []
    
    def register_agent(self, agent: CoalitionAgent):
        """Register agent for coalition formation."""
        self.agents[agent.name] = agent
    
    def add_goal(self, goal: CoalitionGoal):
        """Add goal that requires coalition."""
        self.goals.append(goal)
    
    def evaluate_coalition_value(self, agent: CoalitionAgent, 
                                 coalition_goal: CoalitionGoal,
                                 potential_members: List[str]) -> float:
        """Evaluate value of joining coalition for agent."""
        # Check capability match
        agent_capabilities_match = sum(
            1 for cap in coalition_goal.required_capabilities 
            if cap in agent.capabilities
        )
        
        # Check if coalition can meet requirements
        all_capabilities = set(agent.capabilities)
        for member_name in potential_members:
            if member_name in self.agents:
                all_capabilities.update(self.agents[member_name].capabilities)
        
        can_achieve = all(
            cap in all_capabilities 
            for cap in coalition_goal.required_capabilities
        )
        
        if not can_achieve:
            return 0.0
        
        # Estimate reward share (simple: divide by members)
        estimated_share = coalition_goal.reward / (len(potential_members) + 1)
        
        # Adjust by reputation
        value = estimated_share * (agent.reputation / 10.0)
        
        return value
    
    def form_coalition(self, goal: CoalitionGoal) -> Optional[Coalition]:
        """Attempt to form coalition for goal."""
        print(f"\n   Forming coalition for: {goal.description}")
        print(f"   Required: {', '.join(goal.required_capabilities)}")
        
        # Find capable agents
        interested_agents = []
        for agent in self.agents.values():
            # Check if agent has any required capability
            has_relevant_skill = any(
                cap in agent.capabilities 
                for cap in goal.required_capabilities
            )
            if has_relevant_skill:
                interested_agents.append(agent.name)
        
        if not interested_agents:
            print(f"   ✗ No agents with required capabilities")
            return None
        
        # Evaluate coalition value for each agent
        coalition_members = set()
        for agent_name in interested_agents:
            agent = self.agents[agent_name]
            value = self.evaluate_coalition_value(
                agent, goal, list(coalition_members)
            )
            
            if value > 0:
                coalition_members.add(agent_name)
                print(f"   + {agent_name} joins (value: {value:.1f})")
        
        # Check if coalition can achieve goal
        all_capabilities = set()
        for member in coalition_members:
            all_capabilities.update(self.agents[member].capabilities)
        
        can_achieve = all(
            cap in all_capabilities 
            for cap in goal.required_capabilities
        )
        
        if not can_achieve:
            print(f"   ✗ Coalition cannot meet all requirements")
            return None
        
        # Create coalition
        leader = max(coalition_members, 
                    key=lambda m: self.agents[m].reputation)
        
        coalition = Coalition(
            coalition_id=f"coal_{len(self.coalitions)}",
            name=f"Coalition for {goal.goal_id}",
            goal=goal,
            members=coalition_members,
            leader=leader,
            status=CoalitionStatus.ACTIVE
        )
        
        # Update agent memberships
        for member in coalition_members:
            self.agents[member].current_coalitions.add(coalition.coalition_id)
        
        self.coalitions[coalition.coalition_id] = coalition
        
        print(f"   ✓ Coalition formed: {len(coalition_members)} members, leader: {leader}")
        return coalition
    
    def execute_coalition_goal(self, coalition: Coalition) -> bool:
        """Execute coalition's goal."""
        print(f"\n   Executing: {coalition.name}")
        
        # Simulate work using LLM
        members_text = ", ".join(coalition.members)
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", "You are describing a coalition executing a goal."),
            ("human", f"Coalition members: {members_text}\n"
                     f"Leader: {coalition.leader}\n"
                     f"Goal: {coalition.goal.description}\n"
                     f"Required capabilities: {', '.join(coalition.goal.required_capabilities)}\n\n"
                     f"Describe how they work together (2 sentences).")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        
        print(f"   Work: {str(response.content)[:150]}...")
        
        # Distribute rewards
        share_per_member = coalition.goal.reward / len(coalition.members)
        
        print(f"   Distributing {coalition.goal.reward} reward:")
        for member in coalition.members:
            self.agents[member].rewards_earned += share_per_member
            print(f"   - {member}: +{share_per_member:.1f}")
        
        coalition.total_reward = coalition.goal.reward
        coalition.status = CoalitionStatus.COMPLETED
        
        return True
    
    def dissolve_coalition(self, coalition_id: str):
        """Dissolve coalition."""
        if coalition_id in self.coalitions:
            coalition = self.coalitions[coalition_id]
            
            # Remove from agent memberships
            for member in coalition.members:
                if member in self.agents:
                    self.agents[member].current_coalitions.discard(coalition_id)
            
            coalition.status = CoalitionStatus.DISSOLVED
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute coalition formation pattern."""
        # Form coalitions for all goals
        formed_coalitions = []
        for goal in self.goals:
            coalition = self.form_coalition(goal)
            if coalition:
                formed_coalitions.append(coalition)
        
        # Execute coalition goals
        for coalition in formed_coalitions:
            self.execute_coalition_goal(coalition)
        
        # Statistics
        total_rewards_distributed = sum(
            agent.rewards_earned for agent in self.agents.values()
        )
        
        return {
            "pattern": "CoalitionFormation",
            "total_agents": len(self.agents),
            "goals_available": len(self.goals),
            "coalitions_formed": len(formed_coalitions),
            "coalitions_completed": sum(
                1 for c in self.coalitions.values() 
                if c.status == CoalitionStatus.COMPLETED
            ),
            "total_rewards_distributed": total_rewards_distributed,
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Coalition Formation Pattern Demo ===\n")
    
    # Create pattern
    pattern = CoalitionFormationPattern()
    
    # Register agents with different capabilities
    print("1. Registering agents...")
    
    agents = [
        CoalitionAgent("Alice", ["programming", "architecture"], reputation=8.0),
        CoalitionAgent("Bob", ["programming", "testing"], reputation=7.0),
        CoalitionAgent("Carol", ["design", "frontend"], reputation=7.5),
        CoalitionAgent("Dave", ["devops", "cloud"], reputation=6.5),
        CoalitionAgent("Eve", ["data_analysis", "ml"], reputation=8.5),
    ]
    
    for agent in agents:
        pattern.register_agent(agent)
        print(f"   - {agent.name}: {', '.join(agent.capabilities)} (rep: {agent.reputation})")
    
    # Add goals requiring coalitions
    print("\n2. Adding coalition goals...")
    
    goals = [
        CoalitionGoal(
            "g1",
            "Build web application with ML features",
            required_capabilities=["programming", "frontend", "ml"],
            reward=1000
        ),
        CoalitionGoal(
            "g2",
            "Deploy cloud infrastructure",
            required_capabilities=["devops", "cloud", "architecture"],
            reward=750
        ),
        CoalitionGoal(
            "g3",
            "Create data dashboard",
            required_capabilities=["data_analysis", "design", "programming"],
            reward=800
        ),
    ]
    
    for goal in goals:
        pattern.add_goal(goal)
    
    # Run coalition formation
    print("\n3. Forming and executing coalitions...")
    summary = pattern.run("Project team coalition formation")
    
    # Final results
    print("\n4. Coalition Results:")
    print(f"   Goals: {summary['goals_available']}")
    print(f"   Coalitions Formed: {summary['coalitions_formed']}")
    print(f"   Completed: {summary['coalitions_completed']}")
    print(f"   Total Rewards: ${summary['total_rewards_distributed']:.2f}")
    
    print("\n5. Agent Earnings:")
    for agent_name, agent in pattern.agents.items():
        coalitions = len(agent.current_coalitions)
        print(f"   {agent_name}: ${agent.rewards_earned:.2f} ({coalitions} coalitions)")
    
    print("\n✓ Agents formed coalitions and achieved shared goals!")

    

Agentic Architecture Pattern: Coalition Formation

Pattern 26: 26 Coalition Formation – Line-by-Line Explanation

Overview

Pattern 26: Coalition Formation Pattern

Agents form temporary alliances (coalitions) to achieve mutual goals. Features coalition evaluation, membership negotiation, and benefit distribution.

Code Structure Analysis

File Information

• Pattern Number: 26
• Pattern Name: 26 Coalition Formation
• File Path: patterns\collaboration\26_coalition_formation.py
• Total Lines: 323

Detailed Code Explanation

Module Docstring

"""
Pattern 26: Coalition Formation Pattern

Agents form temporary alliances (coalitions) to achieve mutual goals.
Features coalition evaluation, membership negotiation, and benefit distribution.
"""

Explanation: This module docstring describes the purpose and key features of the 26 Coalition Formation pattern.

Imports Section

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  # 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: 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

Enumerations

CoalitionStatus (Line 16)

class CoalitionStatus(Enum):
    FORMING
    ACTIVE
    DISSOLVED
    COMPLETED

Purpose: Status of coalition.

Values: – FORMING: Represents forming – ACTIVE: Currently operational – DISSOLVED: Represents dissolved – COMPLETED: Successfully finished

Classes

CoalitionStatus (Line 16)

Description: Status of coalition.

CoalitionGoal (Line 25)

Description: Goal that coalition works toward.

Coalition (Line 35)

Description: Group of agents working together.

CoalitionAgent (Line 48)

Description: Agent that can form coalitions.

CoalitionFormationPattern (Line 57)

Description: Coalition Formation Pattern – Dynamic alliance formation.

Agents evaluate potential coalitions, negotiate membership, and work together to achieve shared goals with reward distribution.

Methods:

__init__ (Line 65)

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 71)

def register_agent(self, agent):
    """Register agent for coalition formation."""

Purpose: Register agent for coalition formation.

Parameters: – self: Instance reference – agent: Parameter for agent

add_goal (Line 75)

def add_goal(self, goal):
    """Add goal that requires coalition."""

Purpose: Add goal that requires coalition.

Parameters: – self: Instance reference – goal: Parameter for goal

evaluate_coalition_value (Line 79)

def evaluate_coalition_value(self, agent, coalition_goal, potential_members):
    """Evaluate value of joining coalition for agent."""

Purpose: Evaluate value of joining coalition for agent.

Parameters: – self: Instance reference – agent: Parameter for agent – coalition_goal: Parameter for coalition_goal – potential_members: Parameter for potential_members

form_coalition (Line 111)

def form_coalition(self, goal):
    """Attempt to form coalition for goal."""

Purpose: Attempt to form coalition for goal.

Parameters: – self: Instance reference – goal: Parameter for goal

execute_coalition_goal (Line 179)

def execute_coalition_goal(self, coalition):
    """Execute coalition's goal."""

Purpose: Execute coalition’s goal.

Parameters: – self: Instance reference – coalition: Parameter for coalition

dissolve_coalition (Line 213)

def dissolve_coalition(self, coalition_id):
    """Dissolve coalition."""

Purpose: Dissolve coalition.

Parameters: – self: Instance reference – coalition_id: Parameter for coalition_id

run (Line 225)

def run(self, scenario):
    """Execute coalition formation pattern."""

Purpose: Execute coalition formation 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 = CoalitionStatus(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

    """
Pattern 25: Voting Pattern

Agents vote to reach consensus on decisions.
Supports multiple voting methods and weighted votes.
"""

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import Counter
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI


class VotingMethod(Enum):
    """Type of voting method."""
    MAJORITY = "majority"  # Simple majority wins
    PLURALITY = "plurality"  # Most votes wins
    UNANIMOUS = "unanimous"  # All must agree
    WEIGHTED = "weighted"  # Votes have different weights
    RANKED_CHOICE = "ranked_choice"  # Rank preferences


@dataclass
class VoteOption:
    """Option that can be voted on."""
    option_id: str
    description: str
    proposed_by: str


@dataclass
class Vote:
    """Individual vote cast by an agent."""
    voter_name: str
    proposal_id: str
    choice: str  # option_id
    weight: float = 1.0
    ranking: Optional[List[str]] = None  # For ranked choice
    reasoning: str = ""
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class VotingResult:
    """Result of a vote."""
    proposal_id: str
    method: VotingMethod
    winner: Optional[str]
    all_votes: List[Vote]
    vote_counts: Dict[str, float]
    total_voters: int
    passed: bool


class VotingAgent:
    """Agent that participates in voting."""
    
    def __init__(self, name: str, role: str, expertise: List[str],
                 vote_weight: float = 1.0, model_name: str = "gpt-4"):
        self.name = name
        self.role = role
        self.expertise = expertise
        self.vote_weight = vote_weight
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
    
    def cast_vote(self, proposal_id: str, options: List[VoteOption],
                  method: VotingMethod = VotingMethod.MAJORITY) -> Vote:
        """Cast vote on proposal."""
        
        options_text = "\n".join([f"- {opt.option_id}: {opt.description}" 
                                  for opt in options])
        
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a {self.role} with expertise in {', '.join(self.expertise)}. "
                      f"Analyze the options and vote based on your expertise."),
            ("human", f"Proposal: {proposal_id}\n\n"
                     f"Options:\n{options_text}\n\n"
                     f"Choose the best option and explain why briefly.")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        content = str(response.content)
        
        # Parse choice from response
        choice = options[0].option_id  # Default
        for opt in options:
            if opt.option_id.lower() in content.lower():
                choice = opt.option_id
                break
        
        # For ranked choice, create ranking
        ranking = None
        if method == VotingMethod.RANKED_CHOICE:
            ranking = [opt.option_id for opt in options]
        
        vote = Vote(
            voter_name=self.name,
            proposal_id=proposal_id,
            choice=choice,
            weight=self.vote_weight,
            ranking=ranking,
            reasoning=content[:150]
        )
        
        return vote


class VotingPattern:
    """
    Voting Pattern - Democratic decision-making through voting.
    
    Agents cast votes on proposals using various voting methods.
    Supports weighted votes, ranked choice, and different consensus rules.
    """
    
    def __init__(self, voting_method: VotingMethod = VotingMethod.MAJORITY):
        self.voting_method = voting_method
        self.agents: Dict[str, VotingAgent] = {}
        self.voting_history: List[VotingResult] = []
    
    def register_agent(self, agent: VotingAgent):
        """Register agent as voter."""
        self.agents[agent.name] = agent
    
    def conduct_vote(self, proposal_id: str, options: List[VoteOption]) -> VotingResult:
        """Conduct vote among registered agents."""
        print(f"\n   Voting on: {proposal_id}")
        print(f"   Options: {', '.join([opt.option_id for opt in options])}")
        print(f"   Method: {self.voting_method.value}")
        
        # Collect votes
        votes: List[Vote] = []
        for agent in self.agents.values():
            vote = agent.cast_vote(proposal_id, options, self.voting_method)
            votes.append(vote)
            print(f"   - {agent.name} voted for: {vote.choice}")
        
        # Count votes based on method
        vote_counts: Dict[str, float] = {}
        
        if self.voting_method == VotingMethod.WEIGHTED:
            for vote in votes:
                vote_counts[vote.choice] = vote_counts.get(vote.choice, 0) + vote.weight
        else:
            for vote in votes:
                vote_counts[vote.choice] = vote_counts.get(vote.choice, 0) + 1
        
        # Determine winner
        winner = None
        passed = False
        
        if self.voting_method == VotingMethod.MAJORITY:
            total = sum(vote_counts.values())
            if vote_counts:
                winner = max(vote_counts, key=vote_counts.get)
                passed = vote_counts[winner] > total / 2
        
        elif self.voting_method == VotingMethod.PLURALITY:
            if vote_counts:
                winner = max(vote_counts, key=vote_counts.get)
                passed = True
        
        elif self.voting_method == VotingMethod.UNANIMOUS:
            if len(set(v.choice for v in votes)) == 1:
                winner = votes[0].choice
                passed = True
        
        elif self.voting_method == VotingMethod.WEIGHTED:
            if vote_counts:
                winner = max(vote_counts, key=vote_counts.get)
                total_weight = sum(vote_counts.values())
                passed = vote_counts[winner] > total_weight / 2
        
        if passed:
            print(f"   ✓ Winner: {winner} ({vote_counts.get(winner, 0)} votes)")
        else:
            print(f"   ✗ No winner (consensus not reached)")
        
        result = VotingResult(
            proposal_id=proposal_id,
            method=self.voting_method,
            winner=winner,
            all_votes=votes,
            vote_counts=vote_counts,
            total_voters=len(votes),
            passed=passed
        )
        
        self.voting_history.append(result)
        return result
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute voting pattern."""
        passed_votes = sum(1 for v in self.voting_history if v.passed)
        
        return {
            "pattern": "Voting",
            "voting_method": self.voting_method.value,
            "total_votes": len(self.voting_history),
            "passed": passed_votes,
            "failed": len(self.voting_history) - passed_votes,
            "registered_voters": len(self.agents),
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Voting Pattern Demo ===\n")
    
    # Create pattern with weighted voting
    pattern = VotingPattern(voting_method=VotingMethod.WEIGHTED)
    
    # Register voters with different weights
    print("1. Registering voters...")
    
    tech_lead = VotingAgent("Alice", "tech lead", ["architecture", "security"], vote_weight=2.0)
    senior_dev = VotingAgent("Bob", "senior developer", ["backend", "databases"], vote_weight=1.5)
    designer = VotingAgent("Carol", "designer", ["UI", "UX"], vote_weight=1.0)
    pm = VotingAgent("Dave", "product manager", ["business", "strategy"], vote_weight=1.5)
    
    for agent in [tech_lead, senior_dev, designer, pm]:
        pattern.register_agent(agent)
    
    # Vote on proposals
    print("\n2. Conducting votes...")
    
    proposal1_options = [
        VoteOption("opt1", "Use microservices architecture", "Alice"),
        VoteOption("opt2", "Use monolithic architecture", "Bob"),
        VoteOption("opt3", "Use serverless architecture", "Dave")
    ]
    
    proposal2_options = [
        VoteOption("opt1", "React for frontend", "Carol"),
        VoteOption("opt2", "Vue for frontend", "Bob"),
        VoteOption("opt3", "Angular for frontend", "Alice")
    ]
    
    result1 = pattern.conduct_vote("architecture_decision", proposal1_options)
    result2 = pattern.conduct_vote("frontend_framework", proposal2_options)
    
    # Summary
    print("\n3. Voting Summary:")
    summary = pattern.run("Team technical decisions")
    print(f"   Method: {summary['voting_method']}")
    print(f"   Total Votes: {summary['total_votes']}")
    print(f"   Passed: {summary['passed']}")
    print(f"   Failed: {summary['failed']}")
    print(f"   Registered Voters: {summary['registered_voters']}")
    
    print("\n✓ Voting pattern enabled democratic decision-making!")

    

Agentic Architecture Pattern: Voting

Pattern 25: 25 Voting – Line-by-Line Explanation

Overview

Pattern 25: Voting Pattern

Agents vote to reach consensus on decisions. Supports multiple voting methods and weighted votes.

Code Structure Analysis

File Information

• Pattern Number: 25
• Pattern Name: 25 Voting
• File Path: patterns\collaboration\25_voting.py
• Total Lines: 257

Detailed Code Explanation

Module Docstring

"""
Pattern 25: Voting Pattern

Agents vote to reach consensus on decisions.
Supports multiple voting methods and weighted votes.
"""

Explanation: This module docstring describes the purpose and key features of the 25 Voting pattern.

Imports Section

from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import Counter
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI  # 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

Enumerations

VotingMethod (Line 17)

class VotingMethod(Enum):
    MAJORITY
    PLURALITY
    UNANIMOUS
    WEIGHTED
    RANKED_CHOICE

Purpose: Type of voting method.

Values: – MAJORITY: Represents majority – PLURALITY: Represents plurality – UNANIMOUS: Represents unanimous – WEIGHTED: Represents weighted – RANKED_CHOICE: Represents ranked choice

Classes

VotingMethod (Line 17)

Description: Type of voting method.

VoteOption (Line 27)

Description: Option that can be voted on.

Vote (Line 35)

Description: Individual vote cast by an agent.

VotingResult (Line 47)

Description: Result of a vote.

VotingAgent (Line 58)

Description: Agent that participates in voting.

Methods:

__init__ (Line 61)

def __init__(self, name, role, expertise, vote_weight, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – role: Parameter for role – expertise: Parameter for expertise – vote_weight: Parameter for vote_weight – model_name: LLM model to use (e.g., “gpt-4”)

cast_vote (Line 69)

def cast_vote(self, proposal_id, options, method):
    """Cast vote on proposal."""

Purpose: Cast vote on proposal.

Parameters: – self: Instance reference – proposal_id: Parameter for proposal_id – options: Parameter for options – method: Parameter for method

VotingPattern (Line 112)

Description: Voting Pattern – Democratic decision-making through voting.

Agents cast votes on proposals using various voting methods. Supports weighted votes, ranked choice, and different consensus rules.

Methods:

__init__ (Line 120)

def __init__(self, voting_method):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – voting_method: Parameter for voting_method

register_agent (Line 125)

def register_agent(self, agent):
    """Register agent as voter."""

Purpose: Register agent as voter.

Parameters: – self: Instance reference – agent: Parameter for agent

conduct_vote (Line 129)

def conduct_vote(self, proposal_id, options):
    """Conduct vote among registered agents."""

Purpose: Conduct vote among registered agents.

Parameters: – self: Instance reference – proposal_id: Parameter for proposal_id – options: Parameter for options

run (Line 196)

def run(self, scenario):
    """Execute voting pattern."""

Purpose: Execute voting 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 = VotingMethod(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

    """
Pattern 24: Auction-Based Pattern

Allocates tasks or resources through competitive bidding.
Agents submit bids, and the best bid wins the right to perform the task.
"""

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


class AuctionType(Enum):
    """Type of auction."""
    FIRST_PRICE = "first_price"  # Highest bid wins, pays bid
    SECOND_PRICE = "second_price"  # Highest bid wins, pays second highest
    DUTCH = "dutch"  # Price decreases until someone accepts
    ENGLISH = "english"  # Price increases through competitive bidding


@dataclass
class AuctionItem:
    """Item being auctioned."""
    item_id: str
    task_description: str
    requirements: List[str]
    reserve_price: float = 0.0
    deadline: Optional[datetime] = None


@dataclass
class Bid:
    """Bid from an agent."""
    bidder_name: str
    item_id: str
    amount: float
    capabilities: List[str]
    proposed_timeline: str
    justification: str
    timestamp: datetime = field(default_factory=datetime.now)


@dataclass
class AuctionResult:
    """Result of an auction."""
    item_id: str
    winner: Optional[str]
    winning_bid: Optional[float]
    all_bids: List[Bid]
    auction_type: AuctionType
    success: bool


class AuctionAgent:
    """Agent that participates in auctions."""
    
    def __init__(self, name: str, role: str, capabilities: List[str], budget: float, model_name: str = "gpt-4"):
        self.name = name
        self.role = role
        self.capabilities = capabilities
        self.budget = budget
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
    
    def evaluate_and_bid(self, item: AuctionItem) -> Optional[Bid]:
        """Evaluate item and submit bid if interested."""
        # Check if agent has required capabilities
        has_capabilities = all(req in self.capabilities for req in item.requirements)
        
        if not has_capabilities:
            return None
        
        # Use LLM to determine bid amount and strategy
        prompt = ChatPromptTemplate.from_messages([
            ("system", f"You are {self.name}, a {self.role} with budget ${self.budget:.2f}. "
                      f"Your capabilities: {', '.join(self.capabilities)}. "
                      f"Decide how much to bid for this task."),
            ("human", f"Task: {item.task_description}\n"
                     f"Requirements: {', '.join(item.requirements)}\n"
                     f"Reserve Price: ${item.reserve_price:.2f}\n\n"
                     f"Provide:\n"
                     f"1. Bid amount (number only, less than your budget)\n"
                     f"2. Timeline estimate\n"
                     f"3. Brief justification")
        ])
        
        chain = prompt | self.llm
        response = chain.invoke({})
        content = str(response.content)
        
        # Parse response for bid amount
        try:
            import re
            # Look for dollar amounts or numbers
            amounts = re.findall(r'\$?(\d+(?:\.\d{2})?)', content)
            if amounts:
                bid_amount = float(amounts[0])
                bid_amount = min(bid_amount, self.budget)
                
                if bid_amount >= item.reserve_price:
                    bid = Bid(
                        bidder_name=self.name,
                        item_id=item.item_id,
                        amount=bid_amount,
                        capabilities=self.capabilities,
                        proposed_timeline="2 weeks",  # Could parse from response
                        justification=content[:200]
                    )
                    return bid
        except:
            pass
        
        return None


class AuctionBasedPattern:
    """
    Auction-Based Pattern - Task allocation through competitive bidding.
    
    Tasks are auctioned to agents who submit competitive bids.
    Various auction mechanisms determine winners and prices.
    """
    
    def __init__(self, auction_type: AuctionType = AuctionType.FIRST_PRICE):
        self.auction_type = auction_type
        self.agents: Dict[str, AuctionAgent] = {}
        self.auction_history: List[AuctionResult] = []
    
    def register_agent(self, agent: AuctionAgent):
        """Register agent to participate in auctions."""
        self.agents[agent.name] = agent
    
    def conduct_auction(self, item: AuctionItem) -> AuctionResult:
        """Conduct auction for item."""
        print(f"\n   Auctioning: {item.task_description}")
        print(f"   Reserve Price: ${item.reserve_price:.2f}")
        
        # Collect bids from agents
        bids: List[Bid] = []
        for agent in self.agents.values():
            bid = agent.evaluate_and_bid(item)
            if bid:
                bids.append(bid)
                print(f"   - {agent.name} bid: ${bid.amount:.2f}")
        
        # Determine winner based on auction type
        winner = None
        winning_bid_amount = None
        success = False
        
        if bids:
            # Sort bids by amount (descending)
            sorted_bids = sorted(bids, key=lambda b: b.amount, reverse=True)
            
            if self.auction_type == AuctionType.FIRST_PRICE:
                # Highest bid wins and pays their bid
                winner = sorted_bids[0].bidder_name
                winning_bid_amount = sorted_bids[0].amount
                success = True
            
            elif self.auction_type == AuctionType.SECOND_PRICE:
                # Highest bid wins but pays second-highest bid
                winner = sorted_bids[0].bidder_name
                winning_bid_amount = sorted_bids[1].amount if len(sorted_bids) > 1 else sorted_bids[0].amount
                success = True
            
            print(f"   ✓ Winner: {winner} at ${winning_bid_amount:.2f}")
        else:
            print(f"   ✗ No valid bids received")
        
        result = AuctionResult(
            item_id=item.item_id,
            winner=winner,
            winning_bid=winning_bid_amount,
            all_bids=bids,
            auction_type=self.auction_type,
            success=success
        )
        
        self.auction_history.append(result)
        return result
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute auction-based pattern."""
        return {
            "pattern": "AuctionBased",
            "auction_type": self.auction_type.value,
            "registered_agents": len(self.agents),
            "total_auctions": len(self.auction_history),
            "successful_auctions": sum(1 for a in self.auction_history if a.success),
            "scenario": scenario
        }


# Example usage
if __name__ == "__main__":
    print("=== Auction-Based Pattern Demo ===\n")
    
    # Create pattern
    pattern = AuctionBasedPattern(auction_type=AuctionType.FIRST_PRICE)
    
    # Register agents with different capabilities and budgets
    print("1. Registering agents...")
    agent1 = AuctionAgent("DevCorp", "development company", 
                         ["web_dev", "mobile_dev", "testing"], budget=5000)
    agent2 = AuctionAgent("DesignStudio", "design agency",
                         ["UI_design", "branding", "web_dev"], budget=3000)
    agent3 = AuctionAgent("TechSolutions", "tech consultancy",
                         ["web_dev", "cloud", "security"], budget=7000)
    
    for agent in [agent1, agent2, agent3]:
        pattern.register_agent(agent)
    
    # Auction tasks
    print("\n2. Conducting auctions...")
    
    task1 = AuctionItem(
        item_id="task1",
        task_description="Build e-commerce website",
        requirements=["web_dev"],
        reserve_price=2000.0
    )
    
    task2 = AuctionItem(
        item_id="task2",
        task_description="Design mobile app UI",
        requirements=["UI_design", "mobile_dev"],
        reserve_price=1500.0
    )
    
    # Run auctions
    result1 = pattern.conduct_auction(task1)
    result2 = pattern.conduct_auction(task2)
    
    # Summary
    print("\n3. Auction Summary:")
    summary = pattern.run("Task allocation via auction")
    print(f"   Total Agents: {summary['registered_agents']}")
    print(f"   Auctions Held: {summary['total_auctions']}")
    print(f"   Successful: {summary['successful_auctions']}")
    print(f"   Auction Type: {summary['auction_type']}")
    
    print("\n✓ Auction-based pattern allocated tasks through competitive bidding!")

    

Agentic Architecture Pattern: Auction Based

Pattern 24: 24 Auction Based – Line-by-Line Explanation

Overview

Pattern 24: Auction-Based Pattern

Allocates tasks or resources through competitive bidding. Agents submit bids, and the best bid wins the right to perform the task.

Code Structure Analysis

File Information

• Pattern Number: 24
• Pattern Name: 24 Auction Based
• File Path: patterns\collaboration\24_auction_based.py
• Total Lines: 246

Detailed Code Explanation

Module Docstring

"""
Pattern 24: Auction-Based Pattern

Allocates tasks or resources through competitive bidding.
Agents submit bids, and the best bid wins the right to perform the task.
"""

Explanation: This module docstring describes the purpose and key features of the 24 Auction Based 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
            import re  # 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

Enumerations

AuctionType (Line 16)

class AuctionType(Enum):
    FIRST_PRICE
    SECOND_PRICE
    DUTCH
    ENGLISH

Purpose: Type of auction.

Values: – FIRST_PRICE: Represents first price – SECOND_PRICE: Represents second price – DUTCH: Represents dutch – ENGLISH: Represents english

Classes

AuctionType (Line 16)

Description: Type of auction.

AuctionItem (Line 25)

Description: Item being auctioned.

Bid (Line 35)

Description: Bid from an agent.

AuctionResult (Line 47)

Description: Result of an auction.

AuctionAgent (Line 57)

Description: Agent that participates in auctions.

Methods:

__init__ (Line 60)

def __init__(self, name, role, capabilities, budget, model_name):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – name: Identifier or label – role: Parameter for role – capabilities: Parameter for capabilities – budget: Parameter for budget – model_name: LLM model to use (e.g., “gpt-4”)

evaluate_and_bid (Line 67)

def evaluate_and_bid(self, item):
    """Evaluate item and submit bid if interested."""

Purpose: Evaluate item and submit bid if interested.

Parameters: – self: Instance reference – item: Parameter for item

AuctionBasedPattern (Line 118)

Description: Auction-Based Pattern – Task allocation through competitive bidding.

Tasks are auctioned to agents who submit competitive bids. Various auction mechanisms determine winners and prices.

Methods:

__init__ (Line 126)

def __init__(self, auction_type):
    """No method docstring."""

Purpose: No method docstring.

Parameters: – self: Instance reference – auction_type: Parameter for auction_type

register_agent (Line 131)

def register_agent(self, agent):
    """Register agent to participate in auctions."""

Purpose: Register agent to participate in auctions.

Parameters: – self: Instance reference – agent: Parameter for agent

conduct_auction (Line 135)

def conduct_auction(self, item):
    """Conduct auction for item."""

Purpose: Conduct auction for item.

Parameters: – self: Instance reference – item: Parameter for item

run (Line 185)

def run(self, scenario):
    """Execute auction-based pattern."""

Purpose: Execute auction-based 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 = AuctionType(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