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

# Hierarchical Swarm

> Director-worker pattern where a central director plans, delegates, and (optionally) judges or refines specialist agents through feedback loops

The `HierarchicalSwarm` implements a director-worker pattern. A central director agent analyzes the task, produces a structured plan (a `SwarmSpec`), distributes orders to specialist worker agents, and — optionally — provides feedback or runs a judge over the outputs to iterate further.

## When to Use

* **Complex project management** — multi-stage projects with specialized roles
* **Team coordination** — coordinating diverse specialist agents
* **Quality control** — iterative refinement through feedback loops
* **Hierarchical decisions** — tasks requiring oversight and delegation
* **Strategic planning** — breaking down complex goals into specialized subtasks

## Key Features

* Automatic director agent creation (or pass your own)
* Structured output via `SwarmSpec` (Pydantic schema with `plan` + `orders`)
* Multi-loop feedback and refinement
* Optional agent-as-judge phase for objective scoring
* Optional planning phase before order generation
* Optional team-awareness preamble for workers
* Parallel order execution (default)
* Interactive `rich` dashboard mode
* Conversation history tracking
* Autosave of conversation history to a workspace directory

## Basic Example

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

content_strategist = Agent(
    agent_name="Content-Strategist",
    system_prompt="Develop content strategies and editorial calendars.",
    model_name="gpt-5.4",
)

creative_director = Agent(
    agent_name="Creative-Director",
    system_prompt="Create compelling advertising concepts and visual direction.",
    model_name="gpt-5.4",
)

seo_specialist = Agent(
    agent_name="SEO-Specialist",
    system_prompt="Conduct keyword research and optimize content for search.",
    model_name="gpt-5.4",
)

swarm = HierarchicalSwarm(
    name="Marketing-Team",
    description="Comprehensive marketing team for product launches",
    agents=[content_strategist, creative_director, seo_specialist],
    max_loops=2,            # allow feedback + refinement
)

result = swarm.run(
    "Develop a marketing strategy for a new SaaS project management tool"
)
print(result)
```

## Architecture Flow

```
1. User task → Director Agent
2. (Optional) Director runs a planning pass
3. Director emits SwarmSpec:
   - plan:   overall strategy
   - orders: list of (agent_name, task) pairs
4. Orders execute (parallel by default)
5. Worker outputs go back to the director
6. (Optional) Director feedback OR judge agent scores outputs
7. Loop back to step 2 until max_loops, then synthesize and return
```

## SwarmSpec Structure

The director outputs a Pydantic-validated plan:

```python theme={null}
from pydantic import BaseModel
from typing import List

class HierarchicalOrder(BaseModel):
    agent_name: str   # worker agent that should execute
    task: str         # specific task for that agent

class SwarmSpec(BaseModel):
    plan: str                              # overall strategy
    orders: List[HierarchicalOrder]        # task assignments
```

Example output:

```json theme={null}
{
    "plan": "Create comprehensive marketing strategy with content, creative, and SEO components",
    "orders": [
        {"agent_name": "Content-Strategist", "task": "Develop 90-day content calendar"},
        {"agent_name": "Creative-Director", "task": "Create brand visual identity"},
        {"agent_name": "SEO-Specialist", "task": "Research keywords for SaaS project management"}
    ]
}
```

## Key Parameters

<ParamField path="name" type="str" default="HierarchicalAgentSwarm">
  Name for the swarm instance.
</ParamField>

<ParamField path="description" type="str" default="Distributed task swarm">
  Description of the swarm's purpose.
</ParamField>

<ParamField path="agents" type="List[Agent]" required>
  Specialist worker agents.
</ParamField>

<ParamField path="director" type="Agent">
  Custom director agent. If `None`, a default director is created using `director_model_name` + `director_system_prompt`.
</ParamField>

<ParamField path="max_loops" type="int" default="1">
  Maximum number of plan → execute → feedback iterations.
</ParamField>

<ParamField path="output_type" type="OutputType" default="dict-all-except-first">
  Format of the returned conversation history.
</ParamField>

<ParamField path="director_name" type="str" default="Director">
  Display name for the director agent.
</ParamField>

<ParamField path="director_model_name" type="str" default="gpt-5.4">
  Model used by the director.
</ParamField>

<ParamField path="director_system_prompt" type="str" default="HIEARCHICAL_SWARM_SYSTEM_PROMPT">
  System prompt for the director. Override to customize director behavior.
</ParamField>

<ParamField path="director_temperature" type="float" default="0.7">
  Sampling temperature for the director.
</ParamField>

<ParamField path="director_top_p" type="float" default="0.9">
  Top-p for the director.
</ParamField>

<ParamField path="feedback_director_model_name" type="str" default="gpt-5.4">
  Model used when the director generates feedback on worker outputs.
</ParamField>

<ParamField path="director_feedback_on" type="bool" default="True">
  Enable director feedback after each loop.
</ParamField>

<ParamField path="planning_enabled" type="bool" default="True">
  Run an explicit planning pass before generating orders.
</ParamField>

<ParamField path="agent_as_judge" type="bool" default="False">
  Run a judge agent that scores worker outputs against the plan.
</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. These values override the legacy director parameters. Set `planning_system_prompt` in this dictionary to customize the optional planning pass.
</ParamField>

<ParamField path="max_agent_retries" type="int" default="1">
  Number of retries after a worker's initial execution fails. When retries are exhausted, the worker is marked unavailable in the shared swarm conversation.
</ParamField>

<ParamField path="max_reassignment_attempts" type="int" default="1">
  Maximum number of recovery rounds in which the director can move failed tasks to healthy workers. A failed task does not stop successful workers or abort the swarm.
</ParamField>

<ParamField path="add_collaboration_prompt" type="bool" default="True">
  Inject the multi-agent collaboration preamble into agents.
</ParamField>

<ParamField path="multi_agent_prompt_improvements" type="bool" default="False">
  Augment every worker's system prompt with team awareness (other agents + roles).
</ParamField>

<ParamField path="parallel_execution" type="bool" default="True">
  Execute multi-agent orders concurrently within a loop.
</ParamField>

<ParamField path="interactive" type="bool" default="False">
  Enable the interactive `rich` dashboard.
</ParamField>

<ParamField path="autosave" type="bool" default="True">
  Persist conversation history to a workspace directory.
</ParamField>

<ParamField path="verbose" type="bool" default="False">
  Enable info-level logging.
</ParamField>

<Note>
  The swarm enforces `output_type="final"` on all configurable director and worker agents so only final responses pass between agents. The swarm-level `output_type` still determines the value returned from `run()`.
</Note>

## Methods

### `run(task, img=None)`

Execute the swarm end-to-end.

```python theme={null}
result = swarm.run(
    task="Develop product launch strategy",
    img=None,
)
```

### `batched_run(tasks, ...)`

Run the swarm sequentially over a batch of tasks.

```python theme={null}
results = swarm.batched_run([
    "Strategy for product A",
    "Strategy for product B",
    "Strategy for product C",
])
```

### `run_stream(task, ...)`

Stream output tokens as they arrive from the director and workers. Useful for live UI.

### `display_hierarchy()`

Visualize the swarm's tree with `rich.Tree`.

```python theme={null}
swarm.display_hierarchy()
# Hierarchical Swarm: Marketing-Team
# Director: Director (gpt-5.4)
# ├─ Content-Strategist
# ├─ Creative-Director
# └─ SEO-Specialist
```

### `feedback_director(outputs)`

Manually trigger the feedback director on a list of worker outputs.

### `run_judge_agent(outputs)`

Manually score worker outputs with the judge agent (requires `agent_as_judge=True`).

## Advanced Configuration

### Custom Director

```python theme={null}
custom_director = Agent(
    agent_name="Chief-Strategy-Officer",
    system_prompt="You are a senior executive coordinating teams.",
    model_name="claude-sonnet-4-20250514",
)

swarm = HierarchicalSwarm(
    name="Executive-Team",
    agents=workers,
    director=custom_director,
)
```

### Planning + Multiple Loops

```python theme={null}
swarm = HierarchicalSwarm(
    name="Research-Team",
    agents=researchers,
    planning_enabled=True,
    max_loops=3,
)
```

With `planning_enabled=True`:

1. Director runs a planning pass first
2. The plan is added to the conversation context
3. Director then emits concrete orders
4. Workers execute with full plan context

### Agent-as-Judge

```python theme={null}
swarm = HierarchicalSwarm(
    name="Reviewed-Team",
    agents=workers,
    agent_as_judge=True,
    judge_agent_model_name="claude-sonnet-4-20250514",
    max_loops=2,
)
```

When enabled, after worker outputs are gathered the judge agent emits an `AgentScore` / `JudgeReport` Pydantic object summarizing per-agent quality, which feeds the next loop's director.

### Team-Awareness Preamble

```python theme={null}
swarm = HierarchicalSwarm(
    agents=workers,
    multi_agent_prompt_improvements=True,
)
```

Each worker's `system_prompt` is extended with a description of every other team member so workers can write coherent hand-offs.

### Interactive Dashboard

```python theme={null}
swarm = HierarchicalSwarm(
    name="Development-Team",
    agents=developers,
    interactive=True,
    verbose=True,
)

result = swarm.run("Build authentication system")
```

The dashboard shows:

* Swarm metadata (name, description, loops)
* Director status and current plan
* Per-agent status matrix with loop tracking
* Real-time progress updates

### Autosave

```python theme={null}
swarm = HierarchicalSwarm(
    agents=workers,
    autosave=True,
)
```

When enabled, the swarm writes conversation history under
`$WORKSPACE_DIR/swarms/HierarchicalSwarm/{name}-{timestamp}/`.
If `WORKSPACE_DIR` is unset it defaults to `./agent_workspace`.

## Use Cases

### Software Development Team

```python theme={null}
architect    = Agent(agent_name="Architect", ...)
frontend_dev = Agent(agent_name="Frontend-Dev", ...)
backend_dev  = Agent(agent_name="Backend-Dev", ...)
tester       = Agent(agent_name="QA-Tester", ...)

dev_swarm = HierarchicalSwarm(
    name="Development-Team",
    agents=[architect, frontend_dev, backend_dev, tester],
    max_loops=2,
)

code = dev_swarm.run("Build user authentication with OAuth")
```

### Research Team

```python theme={null}
literature_researcher = Agent(agent_name="Literature-Researcher", ...)
data_analyst          = Agent(agent_name="Data-Analyst", ...)
statistician          = Agent(agent_name="Statistician", ...)
writer                = Agent(agent_name="Research-Writer", ...)

research_swarm = HierarchicalSwarm(
    name="Research-Team",
    description="Academic research team",
    agents=[literature_researcher, data_analyst, statistician, writer],
    max_loops=3,
)

paper = research_swarm.run("Research the impact of AI on job markets")
```

### Marketing Campaign

```python theme={null}
brand_strategist = Agent(agent_name="Brand-Strategist", ...)
copywriter       = Agent(agent_name="Copywriter", ...)
designer         = Agent(agent_name="Designer", ...)
media_buyer      = Agent(agent_name="Media-Buyer", ...)

marketing_swarm = HierarchicalSwarm(
    name="Campaign-Team",
    agents=[brand_strategist, copywriter, designer, media_buyer],
    director_model_name="claude-sonnet-4-20250514",
    max_loops=2,
)

campaign = marketing_swarm.run(
    "Launch campaign for eco-friendly water bottle"
)
```

## Multi-Loop Refinement

With `max_loops > 1`, the director can refine outputs across iterations:

```python theme={null}
swarm = HierarchicalSwarm(
    agents=workers,
    max_loops=3,
    director_feedback_on=True,
)
```

Each loop:

1. Director creates a new plan informed by previous results
2. Director issues orders (possibly to different agents or with different tasks)
3. Workers execute
4. Results — and any feedback/judge output — are appended to the conversation
5. The next loop sees the complete context

## Order Execution

### Concurrent Execution (default)

Orders for multiple agents run in parallel via a `ThreadPoolExecutor` (`execute_orders`), one worker thread per order:

```python theme={null}
orders = [
    {"agent_name": "Agent1", "task": "Analyze data"},
    {"agent_name": "Agent2", "task": "Research market"},
    {"agent_name": "Agent3", "task": "Review competitors"},
]
# All three execute simultaneously when parallel_execution=True
```

### Sequential Execution

Set `parallel_execution=False` to run orders one after another within a loop. Useful when each order depends on the previous one's output.

## Director System Prompt

The default `HIEARCHICAL_SWARM_SYSTEM_PROMPT` instructs the director to:

* Analyze the task
* Identify required expertise
* Create a comprehensive plan
* Distribute work appropriately
* Evaluate results
* Provide constructive feedback

You can fully override it:

```python theme={null}
custom_prompt = """
You are a senior project director coordinating a team of specialists.

Responsibilities:
1. Analyze complex tasks and break them down.
2. Assign work to appropriate specialists.
3. Ensure team coordination and communication.
4. Review outputs and provide feedback.
5. Synthesize results into cohesive outcomes.

Always consider dependencies and optimal sequencing.
"""

swarm = HierarchicalSwarm(
    agents=workers,
    director_system_prompt=custom_prompt,
)
```

## Best Practices

<Note>
  **Loop count:** start at 1 loop for simple coordination; 2–3 for quality refinement. Avoid >3 unless you're seeing clear improvement per loop.
</Note>

1. **Specialized workers** — each agent should have an obviously different area of expertise; overlapping workers waste tokens.
2. **Director model** — use a stronger model than the workers (e.g. Claude Sonnet, GPT-5.4). The director's planning quality bounds the whole swarm.
3. **Loop balance** — every extra loop costs an additional director pass + every worker run. Budget accordingly.
4. **Planning phase** — enable for complex tasks that benefit from a strategy step.
5. **Judge phase** — enable when output quality is uneven and you want explicit scoring/feedback.
6. **Dashboard** — useful for demos and debugging.

<Warning>
  Each loop runs the director **and** every selected worker. Costs grow roughly linearly with `max_loops × len(agents)`.
</Warning>

## Error Handling

```python theme={null}
try:
    result = swarm.run("Task")
except ValueError as e:
    print(f"Configuration error: {e}")
    # e.g. empty agent list, invalid max_loops
except Exception as e:
    print(f"Execution error: {e}")
    # e.g. runtime errors from individual agents
```

`reliability_checks()` runs at construction and validates:

* At least one agent is provided
* `max_loops > 0`
* Director agent can be created or is already valid

## Performance Considerations

### Parallel Order Execution

`execute_orders` runs all orders for a loop simultaneously when `parallel_execution=True`, submitting each order to a `ThreadPoolExecutor`:

```python theme={null}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
    futures_map = {
        executor.submit(
            self.call_single_agent,
            order.agent_name,
            order.task,
        ): order
        for order in orders
    }
```

### Conversation Context

The full conversation flows through every loop, giving the director complete state:

```python theme={null}
output = self.run_director(
    task=f"History: {self.conversation.get_str()} \n\n Task: {task}"
)
```

## Related Architectures

* [Heavy Swarm](/architectures/heavy-swarm) — fixed multi-phase analysis with question generation
* [Mixture of Agents](/architectures/mixture-of-agents) — parallel experts + aggregator
* [Agent Rearrange](/architectures/agent-rearrange) — custom flows without a director
* [Sequential Workflow](/architectures/sequential-workflow) — simple linear flows
* [Structures Catalog](/architectures/structures-catalog) — full enumeration of every multi-agent structure