HierarchicalSwarm

Overview

The HierarchicalSwarm class implements a hierarchical architecture where a director agent distributes tasks to worker agents as structured orders. The director can provide feedback and iterate on results through multiple loops to achieve desired outcomes while maintaining conversation history throughout the process. Each step prints a panel with the director’s orders, and its plan when the director returns one. Set print_on=False to hide it.

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 is created that returns an OrderBatch through a function call. A custom director must return output that parse_orders() can read: an OrderBatch or SwarmSpec, a JSON string or dict with an orders key, or a tool call carrying one. See Custom director

AgentListType

default:"None"

List of worker agents available for task execution. Must not be empty. Nested orchestrators work too; they are matched by their name

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 for the separate agent that writes director feedback when director_feedback_on=True

str

default:"Director"

Name identifier for the director agent

str

default:"gpt-5.4"

Model name for the main director agent

bool

default:"True"

Forwarded to the helper that adds the worker roster to the shared conversation. The roster is added before the collaboration prompt would be appended, so this flag currently has no effect

bool

default:"False"

Whether director feedback is enabled. Feedback runs only when max_loops > 1, never on the final loop, and not when agent_as_judge=True

bool

default:"False"

When True, run() called without a task prompts for one on the console, and every worker’s print_on is set to False at construction

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

The auto-created director sends both director_temperature and director_top_p. Anthropic’s current models reject the two together, so for a Claude director set one of them to None, for example director_settings={"model_name": "claude-sonnet-4-6", "top_p": None}.

bool

default:"False"

Whether to run a separate planning pass before the director writes orders. The plan is recorded in the shared conversation under the director’s name

bool

default:"False"

Whether to enable autosaving of conversation history to workspace directory

bool

default:"False"

Log the autosave directory and each save. It does not control the orders panel; see print_on

bool

default:"True"

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

int

default:"None"

Thread pool size used when parallel_execution is True. Defaults to 95% of available CPU cores. Must be greater than zero if provided.

bool

default:"False"

When True, a judge agent scores the worker outputs after each step and returns a JudgeReport. It replaces director feedback, and its report becomes the previous-loop result the next loop builds on.

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.

bool

default:"True"

Whether to print the director’s orders, and its plan when it returns one, to the console at each step. Orders are shown in full, and the panel is titled with the director’s name. This is separate from verbose, which controls logging

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,
    *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

Returns:

Any

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

Workflow:

Clears the conversation and every agent’s delivery cursor, then re-adds the worker roster and the task, so each call starts clean
Director distributes orders to agents (after an optional planning pass)
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). Each later loop gets the previous loop’s result and the original task
Records a --- Loop N/M completed --- or --- Loop N/M failed: <error> --- marker after every loop
Returns final output formatted per swarm-level output_type, and saves the conversation when autosave=True

A loop that raises, for example because the director is unreachable or its output cannot be parsed, is recorded as failed and the next loop still runs. If every loop fails, run() re-raises the last error instead of returning a transcript.

`step()`

def step(
    self,
    task: str,
    img: str = None,
    *args,
    is_final_loop: bool = False,
    **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

bool

default:"False"

Skips director feedback, which could not inform another loop

Returns:

Any

The results from this step: the judge report when agent_as_judge=True, director feedback when it runs, otherwise the list of worker outputs. An empty list when the director issued no orders

Process:

Director runs (after the planning pass, if enabled) and returns orders
parse_orders() extracts the plan and orders; the panel prints them when print_on=True
Agents execute assigned tasks
Optional judge scoring or director feedback on results

Errors are not swallowed. If the director fails or its output cannot be parsed, step() raises and run() records the loop as failed.

`parse_orders()`

def parse_orders(self, output)

Parses a director response and returns a (plan, orders) tuple, where orders is a list of HierarchicalOrder. Accepts an OrderBatch, a SwarmSpec, a JSON string (decoded twice if double-encoded), a dict, an OpenAI-style tool call, or a list of tool calls or conversation turns. plan is None unless the output carries one. Raises ValueError when no orders can be found. The logic lives in swarms/structs/hierarchical_order_parser.py.

`batched_run()`

def batched_run(
    self,
    tasks: List[str],
    *args,
    img: Optional[Union[str, List[Optional[str]]]] = None,
    imgs: Optional[List[Optional[str]]] = None,
    max_workers: Optional[int] = None,
    return_agent_output_dict: bool = False,
    return_exceptions: bool = False,
    **kwargs,
)

Executes the hierarchical swarm for multiple tasks, calling run() once per task via the shared batch utility. Parameters:

List[str]

required

Tasks to execute

Optional[Union[str, List[Optional[str]]]]

One image applied to every task, or one image per task

Optional[List[Optional[str]]]

Images paired with tasks

Optional[int]

default:"None"

Concurrent task limit; None runs tasks sequentially

bool

default:"False"

Return results keyed by task instead of as a list

bool

default:"False"

Return exceptions instead of raising them

Returns:

list | dict

Results in task order, or keyed by task when return_agent_output_dict=True

Each task starts from a clean conversation and delivery cursor, because run() resets both.

All tasks share this one instance. With max_workers set, concurrent tasks clear and write the same conversation. Keep the default sequential mode unless you use one swarm per task.

`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
max_workers is greater than 0 when provided
Director is available (creates default if needed)

Raises:

ValueError: If swarm configuration is invalid

Data Models

The schemas live in swarms.schemas.hs_schemas. SwarmSpec is also importable from swarms.structs.hiearchical_swarm.

HierarchicalOrder

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

OrderBatch

The schema the default director returns.

class OrderBatch(BaseModel):
    orders: List[HierarchicalOrder]

SwarmSpec

The legacy schema with a plan. A custom director can still return it; the plan is then shown in the orders panel.

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

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,
        "top_p": None,
        "max_tokens": 4000,
    },
    max_agent_retries=1,
    max_reassignment_attempts=1,
)

# 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, the directory is created at construction and the conversation history is saved at the end of every run(), including a failed one: $WORKSPACE_DIR/swarms/HierarchicalSwarm/{swarm-name}-{timestamp}/conversation_history.json WORKSPACE_DIR defaults to ./agent_workspace. Each run overwrites the same file. 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()`

`parse_orders()`

`batched_run()`

`display_hierarchy()`

`reliability_checks()`

Data Models

HierarchicalOrder

OrderBatch

SwarmSpec

JudgeReport

Usage Example

Conversation Autosave

Source Code

​Overview

​Class Definition

​Parameters

​Methods

​run()

​step()

​parse_orders()

​batched_run()

​display_hierarchy()

​reliability_checks()

​Data Models

​HierarchicalOrder

​OrderBatch

​SwarmSpec

​JudgeReport

​Usage Example

​Conversation Autosave

​Source Code

Overview

Class Definition

Parameters

Methods

`run()`

`step()`

`parse_orders()`

`batched_run()`

`display_hierarchy()`

`reliability_checks()`

Data Models

HierarchicalOrder

OrderBatch

SwarmSpec

JudgeReport

Usage Example

Conversation Autosave

Source Code