Nested Worker Hierarchies

Building Multi-Level Agent Systems

Nested worker hierarchies in AgentRouter enable the creation of sophisticated, multi-tiered agent systems. Each worker agent can have its own sub-workers, creating a tree-like structure that maintains clean separation of concerns while allowing complex task delegation.

🎯 Multi-Level Architecture

📊 Hierarchy Levels Explained

Level 0: Manager Agent (Root)

The entry point and orchestrator of the entire system:

ceo = ManagerAgent(
    name="ceo",
    api_key="key",
    goal="Oversee entire operation",
    backstory="Strategic leader with vision",
    instruction="Make high-level decisions and delegate appropriately"
)

Level 1: Direct Workers

Specialized agents reporting directly to the manager:

cto = WorkerAgent(
    name="cto",
    role="Manages technology strategy and oversees technical teams",
    goal="Manage technology initiatives",
    backstory="Tech visionary and architect"
)

cfo = WorkerAgent(
    name="cfo",
    role="Handles financial planning, budgeting, and fiscal operations",
    goal="Manage financial operations",
    backstory="Financial strategist and analyst"
)

# Attach to manager
ceo.attach_worker(cto)
ceo.attach_worker(cfo)

Level 2: Sub-Workers

Workers can have their own workers:

backend_lead = WorkerAgent(
    name="backend_lead",
    role="Develops server-side architecture and APIs",
    goal="Develop backend systems"
)

frontend_lead = WorkerAgent(
    name="frontend_lead",
    role="Creates user interfaces and client-side functionality",
    goal="Create user interfaces"
)

# Attach to CTO worker
cto.attach_worker(backend_lead)
cto.attach_worker(frontend_lead)

Level 3+: Deep Nesting

Continue nesting as deep as needed:

api_developer = WorkerAgent(
    name="api_developer",
    role="Builds and maintains REST APIs and microservices",
    goal="Build and maintain APIs"
)

database_expert = WorkerAgent(
    name="database_expert",
    role="Optimizes database queries and manages data architecture",
    goal="Optimize database performance"
)

# Attach to backend lead
backend_lead.attach_worker(api_developer)
backend_lead.attach_worker(database_expert)

🔐 Isolation & Encapsulation

Complete Execution Isolation

Each agent's execution cycle is completely isolated:

# Parent agent calls child
result = await child_agent.run(
    task="Analyze data",
    additional_context="Focus on Q3 metrics"
)

# What happens internally:
# 1. Child receives task
# 2. Child's complete workflow executes (Plan → Tool → Iterate)
# 3. Only final result returns to parent
# 4. Internal messages, plans, iterations are NOT exposed

Message History Management

Workers can optionally access parent message history:

# Configure history access
worker = WorkerAgent(
    name="analyst",
    role="Analyzes business data and generates insights",
    access_parent_history=True,      # Access parent's messages
    trim_last_parent_user_message=True  # Remove last user message
)

# When executed:
# - If access_parent_history=True: Inherits parent's conversation
# - If trim_last_parent_user_message=True: Removes last user message
# - Task is added as new user message

🔄 Configuration Inheritance

Cascading Configuration

Configuration flows down the hierarchy automatically:

# Manager configuration
manager = ManagerAgent(
    name="manager",
    api_key="key",
    model="usf-mini",
    max_iterations=30,
    api_timeout=60,
    retry_delay=1.0
)

# Worker inherits if not specified
worker = WorkerAgent(
    name="worker",
    role="Executes specialized tasks and operations",
    # Automatically inherits:
    # - api_key from manager
    # - model from manager
    # - max_iterations from manager
    # - api_timeout from manager
    # - retry_delay from manager
    goal="Specific task execution"
)
manager.attach_worker(worker)

# Sub-worker with override
sub_worker = WorkerAgent(
    name="sub_worker",
    role="Performs detailed data processing tasks",
    model="usf-mini",  # Override parent model
    max_iterations=10,       # Override parent iterations
    # Still inherits: api_key, api_timeout, retry_delay
)
worker.attach_worker(sub_worker)

Configuration Priority

1. Explicit Configuration: Directly set on agent
2. Parent Configuration: Inherited from parent
3. Default Values: Framework defaults

Real-World Hierarchy Examples

Example 1: E-Commerce Platform

# Level 0: Platform Manager
platform_manager = ManagerAgent(
    name="platform_manager",
    api_key="key",
    goal="Manage entire e-commerce platform",
    max_iterations=30
)

# Level 1: Department Managers
order_manager = WorkerAgent(
    name="order_manager",
    role="Processes orders and manages fulfillment pipeline",
    goal="Handle order processing"
)

inventory_manager = WorkerAgent(
    name="inventory_manager",
    role="Tracks stock levels and manages product availability",
    goal="Manage product inventory"
)

customer_manager = WorkerAgent(
    name="customer_manager",
    role="Handles customer inquiries and support tickets",
    goal="Handle customer relations"
)

platform_manager.attach_worker(order_manager)
platform_manager.attach_worker(inventory_manager)
platform_manager.attach_worker(customer_manager)

# Level 2: Order Processing Team
payment_processor = WorkerAgent(
    name="payment_processor",
    role="Processes payments and validates transactions",
    goal="Process payments securely"
)

shipping_coordinator = WorkerAgent(
    name="shipping_coordinator",
    role="Coordinates logistics and tracks deliveries",
    goal="Coordinate shipping and delivery"
)

order_manager.attach_worker(payment_processor)
order_manager.attach_worker(shipping_coordinator)

# Level 3: Payment Specialists
fraud_detector = WorkerAgent(
    name="fraud_detector",
    role="Analyzes transactions for fraudulent patterns",
    goal="Detect and prevent fraudulent transactions"
)

refund_handler = WorkerAgent(
    name="refund_handler",
    role="Processes refund requests and chargebacks",
    goal="Process refunds and chargebacks"
)

payment_processor.attach_worker(fraud_detector)
payment_processor.attach_worker(refund_handler)

Example 2: Content Creation Agency

# Level 0: Creative Director
creative_director = ManagerAgent(
    name="creative_director",
    api_key="key",
    goal="Oversee all creative projects"
)

# Level 1: Content Types
blog_manager = WorkerAgent(
    name="blog_manager",
    role="Manages blog content pipeline and editorial calendar",
    goal="Manage blog content creation"
)

video_manager = WorkerAgent(
    name="video_manager",
    role="Produces and edits video content for multiple platforms",
    goal="Produce video content"
)

social_manager = WorkerAgent(
    name="social_manager",
    role="Creates and schedules social media content",
    goal="Manage social media presence"
)

creative_director.attach_worker(blog_manager)
creative_director.attach_worker(video_manager)
creative_director.attach_worker(social_manager)

# Level 2: Blog Team
researcher = WorkerAgent(
    name="researcher",
    role="Researches topics and gathers credible sources",
    goal="Research topics and gather information"
)

writer = WorkerAgent(
    name="writer",
    role="Writes engaging blog posts and articles",
    goal="Write engaging blog posts"
)

editor = WorkerAgent(
    name="editor",
    role="Edits content for clarity and quality",
    goal="Edit and polish content"
)

blog_manager.attach_worker(researcher)
blog_manager.attach_worker(writer)
blog_manager.attach_worker(editor)

🔄 Execution Flow in Nested Hierarchies

Task Delegation Flow

When a task flows through the hierarchy:

# User request to CEO
response = await ceo.run("Develop new mobile app")

# Internal execution flow:
# 1. CEO's Plan API analyzes task
# 2. CEO's Tool Call API suggests delegating to CTO
# 3. CTO worker executes (isolated workflow)
# 4. CTO's Plan API analyzes sub-task
# 5. CTO's Tool Call API delegates to Frontend Lead
# 6. Frontend Lead executes (isolated workflow)
# 7. Results bubble back up through the hierarchy
# 8. CEO generates final response to user

Worker Timeout Considerations

# Important: Worker timeouts should accommodate sub-worker execution
manager = ManagerAgent(
    name="manager",
    api_key="key",
    worker_timeout=180.0  # 3 minutes for complex hierarchies
)

# Level 1 worker
worker = WorkerAgent(
    name="worker",
    role="Coordinates team operations and task distribution",
    worker_timeout=120.0  # 2 minutes for sub-workers
)

# Level 2 sub-worker
sub_worker = WorkerAgent(
    name="sub_worker",
    role="Executes specific technical tasks",
    worker_timeout=60.0  # 1 minute for leaf workers
)

💡 Best Practices for Nested Hierarchies

1. Reasonable Depth Limits

# Recommended: Keep hierarchy depth reasonable (3-5 levels)
# Deep hierarchies increase execution time and complexity

# Good structure
manager → department_head → team_lead → specialist

# Too deep (avoid)
manager → director → manager → lead → senior → mid → junior

2. Clear Responsibility Boundaries

# Good: Clear, non-overlapping responsibilities
order_processor = WorkerAgent(
    name="order_processor",
    role="Validates and processes customer orders",
    goal="Process and validate orders only",
    instruction="Focus on order validation and processing"
)

payment_handler = WorkerAgent(
    name="payment_handler",
    role="Handles payment transactions and verification",
    goal="Handle payment transactions only",
    instruction="Focus on payment processing and verification"
)

# Bad: Overlapping responsibilities
general_handler = WorkerAgent(
    name="general_handler",
    role="Handles everything related to orders",  # Too broad
    goal="Handle orders, payments, and customer service"
    # Too broad - will cause confusion in task delegation
)

3. Configuration Management

# Use consistent configuration across hierarchy
base_config = {
    "api_key": "your-key",
    "model": "usf-mini",
    "max_iterations": 20,
    "api_timeout": 60.0
}

# Create agents with consistent config
manager = ManagerAgent(name="manager", **base_config)
worker1 = WorkerAgent(name="worker1", role="Handles primary operations", **base_config)
worker2 = WorkerAgent(name="worker2", role="Manages secondary tasks", **base_config)

🔍 Monitoring Nested Execution

Using the Pipeline Tracer

The built-in tracer visualizes nested execution:

from agentrouter.visualization import PipelineTracer

# Enable tracing
tracer = PipelineTracer()
manager.set_tracer(tracer)

# Execute with tracing
result = await manager.run("Complex hierarchical task")

# Display execution trace
tracer.display()

# Shows:
# - Manager execution
# - Worker delegations
# - Sub-worker delegations
# - API calls at each level
# - Execution timing

Understanding Execution Depth

The tracer tracks execution depth:

# Tracer output example:
# [Manager] Plan API Call #1
# [Manager] Tool Call API #1 → worker: "cto"
#   [Worker: cto] Plan API Call #1 (depth: 1)
#   [Worker: cto] Tool Call API #1 → worker: "backend_lead"
#     [Worker: backend_lead] Plan API Call #1 (depth: 2)
#     [Worker: backend_lead] Final Response
#   [Worker: cto] Plan API Call #2
#   [Worker: cto] Final Response
# [Manager] Plan API Call #2
# [Manager] Final Response

🚀 Performance Considerations

Execution Time

Each level adds execution overhead:

• Manager: Full workflow (Plan → Tool → Iterate)
• Each Worker: Full isolated workflow
• Deep hierarchies multiply execution time

Memory Usage

• Each agent maintains its own message history
• Nested execution accumulates messages at each level
• Consider access_parent_history=False for memory efficiency

Iteration Limits

# Balance iteration limits across hierarchy
manager = ManagerAgent(
    name="manager",
    max_iterations=30  # Higher for complex orchestration
)

worker = WorkerAgent(
    name="worker",
    role="Manages team workflows and delegates tasks",
    max_iterations=20  # Moderate for sub-tasks
)

specialist = WorkerAgent(
    name="specialist",
    role="Executes specialized technical operations",
    max_iterations=10  # Lower for specific tasks
)

🎯 Common Patterns

Specialist Pattern

Create highly focused workers for specific tasks:

# Specialist workers with narrow focus
sql_expert = WorkerAgent(
    name="sql_expert",
    role="Writes and optimizes SQL queries",
    goal="Write and optimize SQL queries",
    instruction="Focus only on SQL query generation"
)

api_expert = WorkerAgent(
    name="api_expert",
    role="Designs and implements REST APIs",
    goal="Design and implement REST APIs",
    instruction="Focus on RESTful API design"
)

Supervisor Pattern

Use intermediate workers to coordinate sub-teams:

# Supervisor coordinates multiple specialists
tech_supervisor = WorkerAgent(
    name="tech_supervisor",
    role="Coordinates technical teams and resource allocation",
    goal="Coordinate technical team"
)

tech_supervisor.attach_worker(sql_expert)
tech_supervisor.attach_worker(api_expert)

Sequential Processing Pattern

Chain workers for step-by-step processing:

# Data pipeline with sequential workers
data_collector = WorkerAgent(
    name="data_collector",
    role="Collects raw data from multiple sources",
    goal="Collect raw data"
)

data_cleaner = WorkerAgent(
    name="data_cleaner",
    role="Cleans and validates data integrity",
    goal="Clean and validate data"
)

data_analyzer = WorkerAgent(
    name="data_analyzer",
    role="Analyzes data and generates insights",
    goal="Analyze cleaned data"
)

# Attach in sequence for pipeline processing
pipeline_manager.attach_worker(data_collector)
data_collector.attach_worker(data_cleaner)
data_cleaner.attach_worker(data_analyzer)

---

Next: Production Deployment

Learn about deploying AgentRouter in production environments

Explore Production →