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