"""
Pattern 34: Holonic Agent Pattern

Agents that are simultaneously autonomous wholes and parts of larger wholes (holarchies).
Each holon can act independently yet also function as a component of a super-holon.
"""

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


@dataclass
class Holon:
    """
    A holon - both a whole and a part.
    Can act autonomously and be part of a larger structure.
    """
    name: str
    role: str
    level: int  # Level in holarchy (0=atomic, higher=composite)
    sub_holons: List['Holon'] = field(default_factory=list)
    parent_holon: Optional[str] = None
    capabilities: List[str] = field(default_factory=list)
    state: Dict[str, Any] = field(default_factory=dict)
    
    def is_atomic(self) -> bool:
        """Check if holon is atomic (no sub-holons)."""
        return len(self.sub_holons) == 0
    
    def is_composite(self) -> bool:
        """Check if holon is composite (has sub-holons)."""
        return len(self.sub_holons) > 0


@dataclass
class HolonicTask:
    """Task that can be executed by holons at different levels."""
    description: str
    required_capabilities: List[str]
    complexity: int  # 1-5, determines if decomposition needed
    decomposed: bool = False
    subtasks: List['HolonicTask'] = field(default_factory=list)


class HolonicAgentPattern:
    """
    Holonic Agent Pattern - Recursive agent composition.
    
    Features:
    - Holons are both autonomous and part of larger structures
    - Self-similar at different scales
    - Task decomposition and composition
    - Emergent behavior from holon interactions
    - Dynamic holarchy restructuring
    """
    
    def __init__(self, model_name: str = "gpt-4"):
        self.llm = ChatOpenAI(model=model_name, temperature=0.7)
        self.holons: Dict[str, Holon] = {}
        self.task_history: List[Dict[str, Any]] = []
    
    def create_atomic_holon(self, name: str, role: str, capabilities: List[str]) -> Holon:
        """Create an atomic holon (no sub-holons)."""
        holon = Holon(name=name, role=role, level=0, capabilities=capabilities)
        self.holons[name] = holon
        print(f"✓ Created atomic holon: {name} ({role}) - Capabilities: {', '.join(capabilities)}")
        return holon
    
    def create_composite_holon(self, name: str, role: str, 
                              sub_holon_names: List[str]) -> Holon:
        """Create a composite holon from existing holons."""
        sub_holons = [self.holons[n] for n in sub_holon_names if n in self.holons]
        
        # Aggregate capabilities from sub-holons
        aggregated_capabilities = set()
        max_level = 0
        for sub in sub_holons:
            aggregated_capabilities.update(sub.capabilities)
            max_level = max(max_level, sub.level)
            sub.parent_holon = name
        
        holon = Holon(
            name=name,
            role=role,
            level=max_level + 1,
            sub_holons=sub_holons,
            capabilities=list(aggregated_capabilities)
        )
        self.holons[name] = holon
        
        print(f"✓ Created composite holon: {name} (Level {holon.level}) from "
              f"{', '.join(sub_holon_names)}")
        print(f"  Aggregated capabilities: {', '.join(holon.capabilities)}")
        return holon
    
    def decompose_task(self, task: HolonicTask) -> List[HolonicTask]:
        """Decompose complex task into subtasks."""
        if task.complexity  Dict[str, Any]:
        """Assign task to holon - may delegate to sub-holons."""
        if holon_name not in self.holons:
            return {"success": False, "error": "Holon not found"}
        
        holon = self.holons[holon_name]
        
        # Check if holon has required capabilities
        has_capabilities = all(cap in holon.capabilities 
                              for cap in task.required_capabilities)
        
        if not has_capabilities:
            return {
                "success": False,
                "holon": holon_name,
                "error": "Missing required capabilities"
            }
        
        # If atomic holon, execute directly
        if holon.is_atomic():
            result = {
                "success": True,
                "holon": holon_name,
                "type": "atomic",
                "task": task.description,
                "execution": "direct"
            }
        else:
            # Composite holon: decompose and delegate
            if task.complexity > 1 and not task.decomposed:
                subtasks = self.decompose_task(task)
            else:
                subtasks = [task]
            
            # Delegate to sub-holons
            delegations = []
            for subtask in subtasks:
                # Find best sub-holon for subtask
                for sub_holon in holon.sub_holons:
                    if all(cap in sub_holon.capabilities 
                          for cap in subtask.required_capabilities):
                        delegations.append({
                            "sub_holon": sub_holon.name,
                            "subtask": subtask.description
                        })
                        break
            
            result = {
                "success": True,
                "holon": holon_name,
                "type": "composite",
                "task": task.description,
                "execution": "delegated",
                "delegations": delegations
            }
        
        self.task_history.append(result)
        return result
    
    def get_holarchy_structure(self) -> Dict[str, Any]:
        """Get the hierarchical structure of holons."""
        def build_tree(holon: Holon) -> Dict[str, Any]:
            tree = {
                "name": holon.name,
                "role": holon.role,
                "level": holon.level,
                "type": "composite" if holon.is_composite() else "atomic",
                "capabilities": holon.capabilities
            }
            if holon.sub_holons:
                tree["sub_holons"] = [build_tree(sub) for sub in holon.sub_holons]
            return tree
        
        # Find root holons (no parent)
        roots = [h for h in self.holons.values() if h.parent_holon is None]
        return {"holarchy": [build_tree(root) for root in roots]}
    
    def run(self, scenario: str) -> Dict[str, Any]:
        """Execute holonic organization scenario."""
        return {
            "pattern": "Holonic Agent",
            "scenario": scenario,
            "total_holons": len(self.holons),
            "atomic_holons": sum(1 for h in self.holons.values() if h.is_atomic()),
            "composite_holons": sum(1 for h in self.holons.values() if h.is_composite()),
            "tasks_executed": len(self.task_history)
        }


# Example usage
if __name__ == "__main__":
    print("=== Holonic Agent Pattern Demo ===\n")
    
    pattern = HolonicAgentPattern()
    
    # Create atomic holons (Level 0)
    print("--- Creating Atomic Holons (Level 0) ---")
    pattern.create_atomic_holon("DataCollector", "Data Collector", ["data_collection", "API"])
    pattern.create_atomic_holon("DataCleaner", "Data Cleaner", ["data_cleaning", "validation"])
    pattern.create_atomic_holon("ModelTrainer", "Model Trainer", ["ML", "training"])
    pattern.create_atomic_holon("ModelEvaluator", "Model Evaluator", ["ML", "evaluation"])
    
    # Create composite holons (Level 1)
    print("\n--- Creating Composite Holons (Level 1) ---")
    pattern.create_composite_holon(
        "DataPipeline", 
        "Data Processing Pipeline",
        ["DataCollector", "DataCleaner"]
    )
    pattern.create_composite_holon(
        "MLPipeline",
        "ML Training Pipeline",
        ["ModelTrainer", "ModelEvaluator"]
    )
    
    # Create super-holon (Level 2)
    print("\n--- Creating Super-Holon (Level 2) ---")
    pattern.create_composite_holon(
        "MLSystem",
        "Complete ML System",
        ["DataPipeline", "MLPipeline"]
    )
    
    # Assign tasks at different levels
    print("\n--- Task Assignment ---")
    
    # Task to atomic holon
    task1 = HolonicTask(
        description="Collect data from API endpoint",
        required_capabilities=["data_collection"],
        complexity=1
    )
    result1 = pattern.assign_task("DataCollector", task1)
    print(f"\n✓ Atomic execution:")
    print(f"  Holon: {result1['holon']}")
    print(f"  Task: {result1['task']}")
    print(f"  Execution: {result1['execution']}")
    
    # Task to composite holon (Level 1)
    task2 = HolonicTask(
        description="Process raw data into clean dataset",
        required_capabilities=["data_collection", "data_cleaning"],
        complexity=3
    )
    result2 = pattern.assign_task("DataPipeline", task2)
    print(f"\n✓ Composite execution (Level 1):")
    print(f"  Holon: {result2['holon']}")
    print(f"  Task: {result2['task']}")
    print(f"  Execution: {result2['execution']}")
    print(f"  Delegations:")
    for delegation in result2['delegations']:
        print(f"    - {delegation['sub_holon']}: {delegation['subtask']}")
    
    # Task to super-holon (Level 2)
    task3 = HolonicTask(
        description="Build and deploy complete ML model",
        required_capabilities=["data_collection", "data_cleaning", "ML", "training", "evaluation"],
        complexity=5
    )
    result3 = pattern.assign_task("MLSystem", task3)
    print(f"\n✓ Super-holon execution (Level 2):")
    print(f"  Holon: {result3['holon']}")
    print(f"  Task: {result3['task']}")
    print(f"  Execution: {result3['execution']}")
    print(f"  Delegations:")
    for delegation in result3['delegations']:
        print(f"    - {delegation['sub_holon']}: {delegation['subtask']}")
    
    # Show holarchy structure
    print("\n--- Holarchy Structure ---")
    structure = pattern.get_holarchy_structure()
    
    def print_tree(node, indent=0):
        prefix = "  " * indent
        print(f"{prefix}├─ {node['name']} ({node['type']}, Level {node['level']})")
        print(f"{prefix}   Role: {node['role']}")
        if node.get('sub_holons'):
            for sub in node['sub_holons']:
                print_tree(sub, indent + 1)
    
    for root in structure['holarchy']:
        print_tree(root)
    
    # Summary
    print("\n--- Summary ---")
    result = pattern.run("Holonic ML system")
    print(f"Pattern: {result['pattern']}")
    print(f"Total Holons: {result['total_holons']}")
    print(f"Atomic Holons: {result['atomic_holons']}")
    print(f"Composite Holons: {result['composite_holons']}")
    print(f"Tasks Executed: {result['tasks_executed']}")

    

Agentic Architecture Pattern: Holonic Agent

Pattern 34: 34 Holonic Agent – Line-by-Line Explanation

Overview

Pattern 34: Holonic Agent Pattern

Agents that are simultaneously autonomous wholes and parts of larger wholes (holarchies). Each holon can act independently yet also function as a component of a super-holon.

Code Structure Analysis

File Information

• Pattern Number: 34
• Pattern Name: 34 Holonic Agent
• File Path: patterns\organizational\34_holonic_agent.py
• Total Lines: 327

Detailed Code Explanation

Module Docstring

"""
Pattern 34: Holonic Agent Pattern

Agents that are simultaneously autonomous wholes and parts of larger wholes (holarchies).
Each holon can act independently yet also function as a component of a super-holon.
"""

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

Classes

Holon (Line 16)

Description: A holon – both a whole and a part. Can act autonomously and be part of a larger structure.

Methods:

is_atomic (Line 29)

def is_atomic(self):
    """Check if holon is atomic (no sub-holons)."""

Purpose: Check if holon is atomic (no sub-holons).

Parameters: – self: Instance reference

is_composite (Line 33)

def is_composite(self):
    """Check if holon is composite (has sub-holons)."""

Purpose: Check if holon is composite (has sub-holons).

Parameters: – self: Instance reference

HolonicTask (Line 39)

Description: Task that can be executed by holons at different levels.

HolonicAgentPattern (Line 48)

Description: Holonic Agent Pattern – Recursive agent composition.

Features: – Holons are both autonomous and part of larger structures – Self-similar at different scales – Task decomposition and composition – Emergent behavior from holon interactions – Dynamic holarchy restructuring

Methods:

__init__ (Line 60)

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

create_atomic_holon (Line 65)

def create_atomic_holon(self, name, role, capabilities):
    """Create an atomic holon (no sub-holons)."""

Purpose: Create an atomic holon (no sub-holons).

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

create_composite_holon (Line 72)

def create_composite_holon(self, name, role, sub_holon_names):
    """Create a composite holon from existing holons."""

Purpose: Create a composite holon from existing holons.

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

decompose_task (Line 99)

def decompose_task(self, task):
    """Decompose complex task into subtasks."""

Purpose: Decompose complex task into subtasks.

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

assign_task (Line 136)

def assign_task(self, holon_name, task):
    """Assign task to holon - may delegate to sub-holons."""

Purpose: Assign task to holon – may delegate to sub-holons.

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

get_holarchy_structure (Line 195)

def get_holarchy_structure(self):
    """Get the hierarchical structure of holons."""

Purpose: Get the hierarchical structure of holons.

Parameters: – self: Instance reference

run (Line 213)

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

Purpose: Execute holonic 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 = Holon(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