aishastr.live

Agentic Architecture Pattern: Hierarchical Agent

Written by

admin

in


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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts