> ## Documentation Index
> Fetch the complete documentation index at: https://docs.swarms.world/llms.txt
> Use this file to discover all available pages before exploring further.

# HierarchicalSwarm

> A hierarchical multi-agent orchestrator that coordinates agents through a director

## 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

```python theme={null}
from swarms import HierarchicalSwarm
```

## Parameters

<ParamField path="name" type="str" default="HierarchicalAgentSwarm">
  The name identifier for this swarm instance
</ParamField>

<ParamField path="description" type="str" default="Distributed task swarm">
  A description of the swarm's purpose and capabilities
</ParamField>

<ParamField path="director" type="Optional[Union[Agent, Callable, Any]]" default="None">
  The director agent that coordinates the swarm. If None, a default director will be created
</ParamField>

<ParamField path="agents" type="AgentListType" default="None">
  List of worker agents available for task execution. Must not be empty
</ParamField>

<ParamField path="max_loops" type="int" default="1">
  Maximum number of feedback loops the swarm can perform (must be > 0)
</ParamField>

<ParamField path="output_type" type="OutputType" default="dict-all-except-first">
  Format for the final output of the swarm
</ParamField>

<ParamField path="feedback_director_model_name" type="str" default="gpt-5.4">
  Model name for the feedback director
</ParamField>

<ParamField path="director_name" type="str" default="Director">
  Name identifier for the director agent
</ParamField>

<ParamField path="director_model_name" type="str" default="gpt-5.4">
  Model name for the main director agent
</ParamField>

<ParamField path="add_collaboration_prompt" type="bool" default="True">
  Whether to add collaboration prompts to agents
</ParamField>

<ParamField path="director_feedback_on" type="bool" default="True">
  Whether director feedback is enabled
</ParamField>

<ParamField path="interactive" type="bool" default="False">
  Enable interactive dashboard with real-time monitoring
</ParamField>

<ParamField path="director_system_prompt" type="str" default="HIEARCHICAL_SWARM_SYSTEM_PROMPT">
  Custom system prompt for the director agent
</ParamField>

<ParamField path="multi_agent_prompt_improvements" type="bool" default="False">
  Enable enhanced multi-agent collaboration prompts for worker agents
</ParamField>

<ParamField path="director_temperature" type="float" default="0.7">
  Temperature parameter for director agent's LLM
</ParamField>

<ParamField path="director_top_p" type="float" default="0.9">
  Top-p parameter for director agent's LLM
</ParamField>

<ParamField path="planning_enabled" type="bool" default="True">
  Whether to enable the planning phase before order distribution
</ParamField>

<ParamField path="autosave" type="bool" default="True">
  Whether to enable autosaving of conversation history to workspace directory
</ParamField>

<ParamField path="verbose" type="bool" default="False">
  Enable verbose logging output
</ParamField>

<ParamField path="parallel_execution" type="bool" default="True">
  When `True`, worker agents assigned in a single director order are executed concurrently; when `False`, they run one after another.
</ParamField>

<ParamField path="agent_as_judge" type="bool" default="False">
  When `True`, a judge agent evaluates the worker outputs to inform the director's feedback loop.
</ParamField>

<ParamField path="judge_agent_model_name" type="str" default="gpt-5.4">
  Model used by the judge agent when `agent_as_judge=True`.
</ParamField>

<ParamField path="director_settings" type="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"`.
</ParamField>

<ParamField path="max_agent_retries" type="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`.
</ParamField>

<ParamField path="max_reassignment_attempts" type="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`.
</ParamField>

<Note>
  `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()`.
</Note>

## Methods

### `run()`

```python theme={null}
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:**

<ParamField path="task" type="str" optional>
  The initial task to be processed by the swarm. If None and interactive mode is enabled, will prompt for input
</ParamField>

<ParamField path="img" type="str" optional>
  Optional image input for the agents
</ParamField>

<ParamField path="streaming_callback" type="Callable[[str, str, bool], None]" optional>
  Callback function for streaming agent outputs. Parameters are (agent\_name, chunk, is\_final)
</ParamField>

**Returns:**

<ResponseField name="result" type="Any">
  The formatted conversation history as output, formatted according to output\_type configuration
</ResponseField>

**Workflow:**

1. Director creates a plan and distributes orders to agents
2. Agents execute tasks and report back to director
3. Failed workers are retried up to `max_agent_retries`
4. Exhausted workers are marked unavailable in shared context
5. Director reassigns failed tasks to healthy workers, up to `max_reassignment_attempts`
6. Director evaluates results and issues new orders if needed (up to `max_loops`)
7. All context and conversation history is preserved throughout
8. Returns final output formatted per swarm-level `output_type`

***

### `step()`

```python theme={null}
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:**

<ParamField path="task" type="str" required>
  The task to be processed in this step
</ParamField>

<ParamField path="img" type="str" optional>
  Optional image input for the task
</ParamField>

<ParamField path="streaming_callback" type="Callable" optional>
  Callback for streaming outputs
</ParamField>

**Returns:**

<ResponseField name="feedback" type="Any">
  The results from this step, either agent outputs or director feedback
</ResponseField>

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

***

### `arun()`

```python theme={null}
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()`

```python theme={null}
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()`

```python theme={null}
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()`

```python theme={null}
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()`

```python theme={null}
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()`

```python theme={null}
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

```python theme={null}
class HierarchicalOrder(BaseModel):
    agent_name: str  # Name of agent assigned to execute task
    task: str        # Specific task to be executed
```

### SwarmSpec

```python theme={null}
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.

```python theme={null}
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

```python theme={null}
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](https://github.com/kyegomez/swarms/blob/master/swarms/structs/hiearchical_swarm.py)