Skip to main content

​
Overview

The HierarchicalSwarm class implements a hierarchical architecture where a director agent creates plans and distributes tasks to worker agents. The director can provide feedback and iterate on results through multiple loops to achieve desired outcomes while maintaining conversation history throughout the process.

​
Class Definition

from swarms.structs.hiearchical_swarm import HierarchicalSwarm

​
Parameters

​
name
str
default:"HierarchicalAgentSwarm"
The name identifier for this swarm instance
​
description
str
default:"Distributed task swarm"
A description of the swarm’s purpose and capabilities
​
director
Optional[Union[Agent, Callable, Any]]
default:"None"
The director agent that coordinates the swarm. If None, a default director will be created
​
agents
AgentListType
default:"None"
List of worker agents available for task execution. Must not be empty
​
max_loops
int
default:"1"
Maximum number of feedback loops the swarm can perform (must be > 0)
​
output_type
OutputType
default:"dict-all-except-first"
Format for the final output of the swarm
​
feedback_director_model_name
str
default:"gpt-4o-mini"
Model name for the feedback director
​
director_name
str
default:"Director"
Name identifier for the director agent
​
director_model_name
str
default:"gpt-4o-mini"
Model name for the main director agent
​
add_collaboration_prompt
bool
default:"True"
Whether to add collaboration prompts to agents
​
director_feedback_on
bool
default:"True"
Whether director feedback is enabled
​
interactive
bool
default:"False"
Enable interactive dashboard with real-time monitoring
​
director_system_prompt
str
default:"HIEARCHICAL_SWARM_SYSTEM_PROMPT"
Custom system prompt for the director agent
​
multi_agent_prompt_improvements
bool
default:"False"
Enable enhanced multi-agent collaboration prompts for worker agents
​
director_temperature
float
default:"0.7"
Temperature parameter for director agent’s LLM
​
director_top_p
float
default:"0.9"
Top-p parameter for director agent’s LLM
​
planning_enabled
bool
default:"True"
Whether to enable the planning phase before order distribution
​
autosave
bool
default:"True"
Whether to enable autosaving of conversation history to workspace directory
​
verbose
bool
default:"False"
Enable verbose logging output

​
Methods

​
run()

def run(
    self,
    task: Optional[str] = None,
    img: Optional[str] = None,
    streaming_callback: Optional[Callable[[str, str, bool], None]] = None,
    *args,
    **kwargs,
) -> Any
Executes the hierarchical swarm for the specified number of feedback loops. Parameters:
​
task
str
The initial task to be processed by the swarm. If None and interactive mode is enabled, will prompt for input
​
img
str
Optional image input for the agents
​
streaming_callback
Callable[[str, str, bool], None]
Callback function for streaming agent outputs. Parameters are (agent_name, chunk, is_final)
Returns:
​
result
Any
The formatted conversation history as output, formatted according to output_type configuration
Workflow:
  1. Director creates a plan and distributes orders to agents
  2. Agents execute tasks and report back to director
  3. Director evaluates results and issues new orders if needed (up to max_loops)
  4. All context and conversation history is preserved throughout
  5. Returns final output formatted per output_type

​
step()

def step(
    self,
    task: str,
    img: str = None,
    streaming_callback: Optional[Callable[[str, str, bool], None]] = None,
    *args,
    **kwargs,
) -> Any
Executes a single step of the hierarchical swarm workflow. Parameters:
​
task
str
required
The task to be processed in this step
​
img
str
Optional image input for the task
​
streaming_callback
Callable
Callback for streaming outputs
Returns:
​
feedback
Any
The results from this step, either agent outputs or director feedback
Process:
  1. Director runs to create plan and orders
  2. Orders are parsed and distributed
  3. Agents execute assigned tasks
  4. Optional director feedback on results

​
display_hierarchy()

def display_hierarchy(self) -> None
Displays the hierarchical structure of the swarm using Rich Tree visualization. Shows the Director at the top level and all worker agents as children branches with their configurations.

​
reliability_checks()

def reliability_checks(self) -> None
Performs validation checks to ensure the swarm is properly configured. Validates:
  • At least one agent is provided
  • max_loops is greater than 0
  • Director is available (creates default if needed)
Raises:
  • ValueError: If swarm configuration is invalid

​
Data Models

​
HierarchicalOrder

class HierarchicalOrder(BaseModel):
    agent_name: str  # Name of agent assigned to execute task
    task: str        # Specific task to be executed

​
SwarmSpec

class SwarmSpec(BaseModel):
    plan: str                           # Director's overall plan
    orders: List[HierarchicalOrder]     # Task assignments to agents

​
Interactive Dashboard

When interactive=True, the HierarchicalSwarm displays a real-time dashboard with:
  • Operations Status: Swarm name, description, current loop, agent count
  • Director Operations: Current plan and active orders
  • Agent Monitoring Matrix: Real-time agent status, tasks, and outputs
  • Progress Tracking: Loop completion and runtime metrics
The dashboard uses Swarms Corporation styling with red/black color scheme and provides professional monitoring of swarm operations.

​
Usage Example

from swarms.structs.agent import Agent
from swarms.structs.hiearchical_swarm import HierarchicalSwarm

# Create worker agents
research_agent = Agent(
    agent_name="Research-Specialist",
    system_prompt="Expert in research and data gathering",
    model_name="gpt-4o"
)

analysis_agent = Agent(
    agent_name="Analysis-Specialist",
    system_prompt="Expert in data analysis",
    model_name="gpt-4o"
)

writing_agent = Agent(
    agent_name="Writing-Specialist",
    system_prompt="Expert in writing and communication",
    model_name="gpt-4o"
)

# Create hierarchical swarm
swarm = HierarchicalSwarm(
    name="Research-Analysis-Team",
    description="A hierarchical team for research and analysis",
    agents=[research_agent, analysis_agent, writing_agent],
    max_loops=2,
    director_model_name="gpt-4o",
    interactive=True,
    verbose=True
)

# Display the hierarchy
swarm.display_hierarchy()

# Execute a task
result = swarm.run(
    task="Analyze the impact of AI on healthcare and create a comprehensive report"
)

print(result)

​
Conversation Autosave

When autosave=True, conversation history is automatically saved to: workspace_dir/swarms/HierarchicalSwarm/{swarm-name}-{timestamp}/conversation_history.json This enables:
  • Post-execution analysis
  • Debugging and monitoring
  • Historical tracking of swarm operations

​
Source Code

View the source code on GitHub

Build docs developers (and LLMs) love

Get started for freeTalk to us