HierarchicalSwarm

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 import HierarchicalSwarm

Parameters

str

default:"HierarchicalAgentSwarm"

The name identifier for this swarm instance

str

default:"Distributed task swarm"

A description of the swarm’s purpose and capabilities

Optional[Union[Agent, Callable, Any]]

default:"None"

The director agent that coordinates the swarm. If None, a default director will be created

AgentListType

default:"None"

List of worker agents available for task execution. Must not be empty

int

default:"1"

Maximum number of feedback loops the swarm can perform (must be > 0)

OutputType

default:"dict-all-except-first"

Format for the final output of the swarm

str

default:"gpt-5.4"

Model name for the feedback director

str

default:"Director"

Name identifier for the director agent

str

default:"gpt-5.4"

Model name for the main director agent

bool

default:"True"

Whether to add collaboration prompts to agents

bool

default:"True"

Whether director feedback is enabled

bool

default:"False"

Enable interactive dashboard with real-time monitoring

str

default:"HIEARCHICAL_SWARM_SYSTEM_PROMPT"

Custom system prompt for the director agent

bool

default:"False"

Enable enhanced multi-agent collaboration prompts for worker agents

float

default:"0.7"

Temperature parameter for director agent’s LLM

float

default:"0.9"

Top-p parameter for director agent’s LLM

bool

default:"True"

Whether to enable the planning phase before order distribution

bool

default:"True"

Whether to enable autosaving of conversation history to workspace directory

bool

default:"False"

Enable verbose logging output

bool

default:"True"

When True, worker agents assigned in a single director order are executed concurrently; when False, they run one after another.

bool

default:"False"

When True, a judge agent evaluates the worker outputs to inform the director’s feedback loop.

str

default:"gpt-5.4"

Model used by the judge agent when agent_as_judge=True.

Optional[Dict[str, Any]]

default:"None"

Additional Agent constructor settings for the automatically created director. Values in this dictionary override the legacy director parameters. Use planning_system_prompt to customize the optional planning pass. The swarm always forces the director’s output_type to "final".

int

default:"1"

Number of retry attempts after a worker’s initial execution fails. After all attempts are exhausted, the worker is marked unavailable in the shared conversation and its failure is reported to the director. Must be greater than or equal to 0.

int

default:"1"

Maximum number of recovery rounds in which the director can reassign failed tasks to healthy workers. The swarm continues running when recovery cannot complete a task. Must be greater than or equal to 0.

HierarchicalSwarm forces every configurable director and worker agent to use output_type="final". The swarm-level output_type parameter still controls the format returned by HierarchicalSwarm.run().

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:

str

The initial task to be processed by the swarm. If None and interactive mode is enabled, will prompt for input

str

Optional image input for the agents

Callable[[str, str, bool], None]

Callback function for streaming agent outputs. Parameters are (agent_name, chunk, is_final)

Returns:

Any

The formatted conversation history as output, formatted according to output_type configuration

Workflow:

Director creates a plan and distributes orders to agents
Agents execute tasks and report back to director
Failed workers are retried up to max_agent_retries
Exhausted workers are marked unavailable in shared context
Director reassigns failed tasks to healthy workers, up to max_reassignment_attempts
Director evaluates results and issues new orders if needed (up to max_loops)
All context and conversation history is preserved throughout
Returns final output formatted per swarm-level 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:

str

required

The task to be processed in this step

str

Optional image input for the task

Callable

Callback for streaming outputs

Returns:

Any

The results from this step, either agent outputs or director feedback

Process:

Director runs to create plan and orders
Orders are parsed and distributed
Agents execute assigned tasks
Optional director feedback on results

`arun()`

async def arun(
    self,
    task: Optional[str] = None,
    img: Optional[str] = None,
    streaming_callback: Optional[Callable[[str, str, bool], None]] = None,
    *args,
    **kwargs,
) -> Any

Async entry point that wraps the synchronous run() in asyncio.to_thread(). Accepts the same parameters as run() and returns the same result.

`batched_run()`

def batched_run(
    self,
    tasks: List[str],
    img: str = None,
    streaming_callback: Optional[Callable[[str, str, bool], None]] = None,
    *args,
    **kwargs,
) -> list

Executes the hierarchical swarm for multiple tasks in sequence, calling run() once per task and returning a list of results (one per task).

`arun_stream()`

async def arun_stream(
    self,
    task: Optional[str] = None,
    img: Optional[str] = None,
    with_events: bool = False,
    **kwargs,
)

Async generator that streams tokens from every phase of the swarm — director planning, worker execution, and feedback/judge aggregation. When with_events=False (default) it yields (agent_name, token) tuples; when True it yields structured event dicts tagged with role (director / worker / aggregator / swarm) and loop index, with event types swarm_start, director_start, token, director_end, worker_start, worker_end, aggregator_start, aggregator_end, swarm_end.

`run_stream()`

def run_stream(
    self,
    task: Optional[str] = None,
    img: Optional[str] = None,
    with_events: bool = False,
    **kwargs,
)

Synchronous generator version of arun_stream(). Bridges the async generator to a sync iterator using a background thread. Use arun_stream() directly when already inside a running event loop (e.g. a FastAPI handler).

`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
max_agent_retries is greater than or equal to 0
max_reassignment_attempts is greater than or equal to 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

JudgeReport

Used when agent_as_judge=True. A one-shot judge agent scores each worker agent’s output instead of the director issuing free-form feedback.

class AgentScore(BaseModel):
    agent_name: str
    score: int          # 0-10
    reasoning: str
    suggestions: str

class JudgeReport(BaseModel):
    overall_quality: int    # 0-10
    scores: List[AgentScore]
    summary: str

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 import Agent, HierarchicalSwarm

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

analysis_agent = Agent(
    agent_name="Analysis-Specialist",
    system_prompt="Expert in data analysis",
    model_name="claude-sonnet-4-6"
)

writing_agent = Agent(
    agent_name="Writing-Specialist",
    system_prompt="Expert in writing and communication",
    model_name="claude-sonnet-4-6"
)

# 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_settings={
        "model_name": "claude-sonnet-4-6",
        "temperature": 0.2,
        "max_tokens": 4000,
        "persistent_memory": False,
    },
    max_agent_retries=1,
    max_reassignment_attempts=1,
    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

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

HierarchicalSwarm

Overview

Class Definition

Parameters

Methods

`run()`

`step()`

`arun()`

`batched_run()`

`arun_stream()`

`run_stream()`

`display_hierarchy()`

`reliability_checks()`

Data Models

HierarchicalOrder

SwarmSpec

JudgeReport

Interactive Dashboard

Usage Example

Conversation Autosave

Source Code

​Overview

​Class Definition

​Parameters

​Methods

​run()

​step()

​arun()

​batched_run()

​arun_stream()

​run_stream()

​display_hierarchy()

​reliability_checks()

​Data Models

​HierarchicalOrder

​SwarmSpec

​JudgeReport

​Interactive Dashboard

​Usage Example

​Conversation Autosave

​Source Code

Overview

Class Definition

Parameters

Methods

`run()`

`step()`

`arun()`

`batched_run()`

`arun_stream()`

`run_stream()`

`display_hierarchy()`

`reliability_checks()`

Data Models

HierarchicalOrder

SwarmSpec

JudgeReport

Interactive Dashboard

Usage Example

Conversation Autosave

Source Code