Forest Swarm Pattern

Organize agents into a tree structure with specialized sub-teams. Ideal for hierarchical problem decomposition and divide-and-conquer strategies.

Overview

Forest Swarm creates a tree of agent teams where each node can have its own sub-swarm. Problems are recursively broken down, with specialized teams handling different aspects, and results are aggregated up the tree.

When to Use

• Complex problem decomposition: Break large problems into sub-problems
• Hierarchical organization: Natural tree structure to the work
• Specialized sub-teams: Different expertise needed at different levels
• Divide-and-conquer: Parallel execution of independent sub-tasks

Basic Usage

from azcore import ForestSwarm, Agent

# Create leaf agents for specific tasks
researcher1 = Agent(
    agent_name="Market Researcher",
    system_prompt="Research market trends",
    llm=llm,
)

researcher2 = Agent(
    agent_name="Competitor Researcher",
    system_prompt="Research competitors",
    llm=llm,
)

# Create a sub-swarm
research_team = ForestSwarm(
    agents=[researcher1, researcher2],
    max_loops=1,
)

# Create higher-level agents
analyst = Agent(
    agent_name="Market Analyst",
    system_prompt="Analyze market research findings",
    llm=llm,
)

strategist = Agent(
    agent_name="Strategist",
    system_prompt="Develop market strategy",
    llm=llm,
)

# Create top-level swarm
market_swarm = ForestSwarm(
    agents=[research_team, analyst, strategist],
    max_loops=1,
)

# Run the hierarchical workflow
result = market_swarm.run("Analyze the AI education market")
print(result)

Configuration Options

max_loops

Control iterations at each level:

swarm = ForestSwarm(
    agents=[team1, team2, agent1],
    max_loops=2,  # Run twice at this level
)

verbose

Enable detailed logging:

swarm = ForestSwarm(
    agents=[team1, team2],
    verbose=True,  # Log hierarchical execution
)

Advanced Examples

Software Development Project

from azcore import ForestSwarm, Agent

# Frontend sub-team
ui_designer = Agent(
    agent_name="UI Designer",
    system_prompt="Design user interface and components",
    llm=llm,
)

frontend_dev = Agent(
    agent_name="Frontend Developer",
    system_prompt="Implement frontend code",
    llm=llm,
    tools=[code_tools],
)

frontend_team = ForestSwarm(
    agents=[ui_designer, frontend_dev],
    max_loops=1,
)

# Backend sub-team
api_designer = Agent(
    agent_name="API Designer",
    system_prompt="Design API endpoints and data models",
    llm=llm,
)

backend_dev = Agent(
    agent_name="Backend Developer",
    system_prompt="Implement backend services",
    llm=llm,
    tools=[code_tools],
)

db_specialist = Agent(
    agent_name="Database Specialist",
    system_prompt="Design and optimize database schema",
    llm=llm,
)

backend_team = ForestSwarm(
    agents=[api_designer, backend_dev, db_specialist],
    max_loops=1,
)

# Testing sub-team
test_designer = Agent(
    agent_name="Test Designer",
    system_prompt="Design test strategy and test cases",
    llm=llm,
)

qa_engineer = Agent(
    agent_name="QA Engineer",
    system_prompt="Implement and run tests",
    llm=llm,
    tools=[testing_tools],
)

testing_team = ForestSwarm(
    agents=[test_designer, qa_engineer],
    max_loops=1,
)

# Project leadership
architect = Agent(
    agent_name="Architect",
    system_prompt="Oversee architecture and integration",
    llm=llm,
)

project_manager = Agent(
    agent_name="Project Manager",
    system_prompt="Coordinate teams and ensure delivery",
    llm=llm,
)

# Top-level project swarm
project_swarm = ForestSwarm(
    agents=[frontend_team, backend_team, testing_team, architect, project_manager],
    max_loops=1,
)

# Execute project
result = project_swarm.run("Build a task management application")

Research Project

# Literature review sub-team
paper_finder = Agent(
    agent_name="Paper Finder",
    system_prompt="Find relevant academic papers",
    llm=llm,
    tools=[scholar_search],
)

paper_analyzer = Agent(
    agent_name="Paper Analyzer",
    system_prompt="Analyze and summarize papers",
    llm=llm,
)

citation_manager = Agent(
    agent_name="Citation Manager",
    system_prompt="Organize citations and references",
    llm=llm,
)

literature_team = ForestSwarm(
    agents=[paper_finder, paper_analyzer, citation_manager],
    max_loops=1,
)

# Methodology sub-team
experiment_designer = Agent(
    agent_name="Experiment Designer",
    system_prompt="Design experiments and methodology",
    llm=llm,
)

data_collector = Agent(
    agent_name="Data Collector",
    system_prompt="Collect and organize data",
    llm=llm,
    tools=[data_tools],
)

methodology_team = ForestSwarm(
    agents=[experiment_designer, data_collector],
    max_loops=1,
)

# Analysis sub-team
statistician = Agent(
    agent_name="Statistician",
    system_prompt="Perform statistical analysis",
    llm=llm,
    tools=[stats_tools],
)

visualizer = Agent(
    agent_name="Visualizer",
    system_prompt="Create charts and visualizations",
    llm=llm,
    tools=[plotting_tools],
)

analysis_team = ForestSwarm(
    agents=[statistician, visualizer],
    max_loops=1,
)

# Research leadership
principal_investigator = Agent(
    agent_name="Principal Investigator",
    system_prompt="Oversee research direction and quality",
    llm=llm,
)

academic_writer = Agent(
    agent_name="Academic Writer",
    system_prompt="Write research paper and documentation",
    llm=llm,
)

# Top-level research swarm
research_swarm = ForestSwarm(
    agents=[literature_team, methodology_team, analysis_team, principal_investigator, academic_writer],
    max_loops=1,
)

# Conduct research
paper = research_swarm.run("Research the effectiveness of mixture of agents")

Marketing Campaign

# Content creation sub-team
copywriter = Agent(
    agent_name="Copywriter",
    system_prompt="Write marketing copy",
    llm=llm,
)

designer = Agent(
    agent_name="Designer",
    system_prompt="Create visual designs",
    llm=llm,
)

video_producer = Agent(
    agent_name="Video Producer",
    system_prompt="Create video content",
    llm=llm,
)

content_team = ForestSwarm(
    agents=[copywriter, designer, video_producer],
    max_loops=1,
)

# Channel management sub-team
social_media_manager = Agent(
    agent_name="Social Media Manager",
    system_prompt="Manage social media presence",
    llm=llm,
    tools=[social_tools],
)

email_marketer = Agent(
    agent_name="Email Marketer",
    system_prompt="Create and send email campaigns",
    llm=llm,
    tools=[email_tools],
)

seo_specialist = Agent(
    agent_name="SEO Specialist",
    system_prompt="Optimize for search engines",
    llm=llm,
    tools=[seo_tools],
)

channel_team = ForestSwarm(
    agents=[social_media_manager, email_marketer, seo_specialist],
    max_loops=1,
)

# Analytics sub-team
data_analyst = Agent(
    agent_name="Data Analyst",
    system_prompt="Analyze campaign performance",
    llm=llm,
    tools=[analytics_tools],
)

conversion_optimizer = Agent(
    agent_name="Conversion Optimizer",
    system_prompt="Optimize conversion rates",
    llm=llm,
)

analytics_team = ForestSwarm(
    agents=[data_analyst, conversion_optimizer],
    max_loops=1,
)

# Campaign leadership
campaign_manager = Agent(
    agent_name="Campaign Manager",
    system_prompt="Oversee campaign execution",
    llm=llm,
)

brand_strategist = Agent(
    agent_name="Brand Strategist",
    system_prompt="Ensure brand consistency",
    llm=llm,
)

# Top-level campaign swarm
campaign_swarm = ForestSwarm(
    agents=[content_team, channel_team, analytics_team, campaign_manager, brand_strategist],
    max_loops=1,
)

# Execute campaign
result = campaign_swarm.run("Launch product campaign for AI writing tool")

Enterprise System Design

# Security sub-team
security_architect = Agent(
    agent_name="Security Architect",
    system_prompt="Design security architecture",
    llm=llm,
)

pen_tester = Agent(
    agent_name="Penetration Tester",
    system_prompt="Identify vulnerabilities",
    llm=llm,
    tools=[security_tools],
)

compliance_officer = Agent(
    agent_name="Compliance Officer",
    system_prompt="Ensure regulatory compliance",
    llm=llm,
)

security_team = ForestSwarm(
    agents=[security_architect, pen_tester, compliance_officer],
    max_loops=1,
)

# Infrastructure sub-team
cloud_architect = Agent(
    agent_name="Cloud Architect",
    system_prompt="Design cloud infrastructure",
    llm=llm,
)

devops_engineer = Agent(
    agent_name="DevOps Engineer",
    system_prompt="Set up CI/CD and deployment",
    llm=llm,
    tools=[devops_tools],
)

sre = Agent(
    agent_name="SRE",
    system_prompt="Design for reliability and monitoring",
    llm=llm,
)

infrastructure_team = ForestSwarm(
    agents=[cloud_architect, devops_engineer, sre],
    max_loops=1,
)

# Application sub-team
app_architect = Agent(
    agent_name="Application Architect",
    system_prompt="Design application architecture",
    llm=llm,
)

lead_developer = Agent(
    agent_name="Lead Developer",
    system_prompt="Oversee development",
    llm=llm,
)

application_team = ForestSwarm(
    agents=[app_architect, lead_developer],
    max_loops=1,
)

# Enterprise leadership
enterprise_architect = Agent(
    agent_name="Enterprise Architect",
    system_prompt="Oversee enterprise architecture",
    llm=llm,
)

cto = Agent(
    agent_name="CTO",
    system_prompt="Make technology decisions",
    llm=llm,
)

# Top-level enterprise swarm
enterprise_swarm = ForestSwarm(
    agents=[security_team, infrastructure_team, application_team, enterprise_architect, cto],
    max_loops=1,
)

# Design system
design = enterprise_swarm.run("Design enterprise data platform")

Tree Structure

Basic Tree

Top-Level Swarm
├── Sub-Swarm 1
│   ├── Agent A
│   └── Agent B
├── Sub-Swarm 2
│   ├── Agent C
│   └── Agent D
└── Agent E

Execution Flow

1. Top-level swarm receives task
2. Task distributed to sub-swarms and agents
3. Sub-swarms recursively execute their agents
4. Results bubble up and aggregate
5. Top-level agents process aggregated results

Best Practices

1. Design Clear Hierarchies

Organize by natural decomposition:

# Good: Logical hierarchy
project_swarm = ForestSwarm([
    frontend_team,  # Sub-swarm for frontend
    backend_team,   # Sub-swarm for backend
    architect,      # High-level coordination
])

# Bad: Flat structure (use Concurrent instead)
swarm = ForestSwarm([agent1, agent2, agent3, agent4])

2. Balance Tree Depth

Avoid too deep or too shallow:

# Good: 2-3 levels
# Top → Teams → Specialists

# Too deep: 5+ levels
# Hard to manage and debug

# Too shallow: 1 level
# Use simpler pattern instead

3. Size Sub-Teams Appropriately

Keep sub-teams manageable:

# Good: 2-4 agents per sub-team
sub_team = ForestSwarm(agents=[agent1, agent2, agent3])

# Too large: 10+ agents in one sub-team
# Consider further decomposition

4. Clear Responsibilities

Each level should have distinct purpose:

# Level 1: Strategic oversight
# Level 2: Team coordination
# Level 3: Task execution

State Management

State flows down and results bubble up:

# Top-level task → Sub-swarms
# Sub-swarm results → Top-level aggregation

Custom state handling:

class CustomForestSwarm(ForestSwarm):
    def pre_execution_hook(self, state):
        # Modify state before passing to sub-swarms
        return state

    def aggregate_results(self, results, state):
        # Custom result aggregation
        return aggregated_result

Error Handling

Handle failures at different levels:

try:
    result = swarm.run("Task")
except Exception as e:
    print(f"Swarm failed: {e}")
    # Identify which sub-team failed
    # Implement recovery strategy

Graceful degradation:

class RobustForestSwarm(ForestSwarm):
    def handle_subswarm_failure(self, subswarm, error):
        # Log and continue with other sub-swarms
        logger.error(f"Sub-swarm {subswarm.name} failed: {error}")
        return None

Performance Considerations

Parallelization

Sub-swarms can run in parallel:

# Sub-swarms execute concurrently
# Total time ≈ max(sub-swarm times) + coordination time

Complexity Management

# Total agents = sum of all sub-swarm agents
# Monitor total agent count for cost control

Optimization Tips

1. Parallel execution: Independent sub-swarms run concurrently
2. Caching: Reuse results from similar sub-tasks
3. Lazy evaluation: Only execute needed sub-swarms
4. Result streaming: Stream results as they become available

Comparison with Other Patterns

Debugging

Monitor hierarchical execution:

swarm = ForestSwarm(
    agents=[team1, team2, agent1],
    verbose=True,  # Log all levels
)

result = swarm.run("Task")

# Inspect sub-swarm outputs
for subswarm in swarm.subswarms:
    print(f"Sub-swarm {subswarm.name}: {subswarm.last_output}")

# Check top-level result
print(f"Final: {result}")

Visualize tree structure:

def visualize_swarm(swarm, level=0):
    indent = "  " * level
    print(f"{indent}{swarm.name}")
    for agent in swarm.agents:
        if isinstance(agent, ForestSwarm):
            visualize_swarm(agent, level + 1)
        else:
            print(f"{indent}  - {agent.agent_name}")

visualize_swarm(project_swarm)

Common Patterns

Divide-and-Conquer

Break problem into independent parts:

# Top level: Problem decomposition
# Middle level: Parallel sub-problems
# Bottom level: Task execution

Hierarchical Planning

Multi-level planning and execution:

# Strategic planning team
# Tactical execution teams
# Operational workers

Specialized Teams

Domain-specific sub-teams:

# Frontend team, Backend team, QA team
# Each with specialized agents