Hierarchical Swarm

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

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

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

SwarmSpec Structure

The director outputs a Pydantic-validated plan:

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:

{
    "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

name

str

default:"HierarchicalAgentSwarm"

Name for the swarm instance.

description

str

default:"Distributed task swarm"

Description of the swarm’s purpose.

agents

List[Agent]

required

Specialist worker agents.

director

Agent

Custom director agent. If None, a default director is created using director_model_name + director_system_prompt.

max_loops

int

default:"1"

Maximum number of plan → execute → feedback iterations.

output_type

OutputType

default:"dict-all-except-first"

Format of the returned conversation history.

director_name

str

default:"Director"

Display name for the director agent.

director_model_name

str

default:"gpt-5.4"

Model used by the director.

director_system_prompt

str

default:"HIEARCHICAL_SWARM_SYSTEM_PROMPT"

System prompt for the director. Override to customize director behavior.

director_temperature

float

default:"0.7"

Sampling temperature for the director.

director_top_p

float

default:"0.9"

Top-p for the director.

feedback_director_model_name

str

default:"gpt-5.4"

Model used when the director generates feedback on worker outputs.

director_feedback_on

bool

default:"True"

Enable director feedback after each loop.

planning_enabled

bool

default:"True"

Run an explicit planning pass before generating orders.

agent_as_judge

bool

default:"False"

Run a judge agent that scores worker outputs against the plan.

judge_agent_model_name

str

default:"gpt-5.4"

Model used by the judge agent when agent_as_judge=True.

add_collaboration_prompt

bool

default:"True"

Inject the multi-agent collaboration preamble into agents.

multi_agent_prompt_improvements

bool

default:"False"

Augment every worker’s system prompt with team awareness (other agents + roles).

parallel_execution

bool

default:"True"

Execute multi-agent orders concurrently within a loop.

interactive

bool

default:"False"

Enable the interactive rich dashboard.

autosave

bool

default:"True"

Persist conversation history to a workspace directory.

verbose

bool

default:"False"

Enable info-level logging.

Methods

`run(task, img=None)`

Execute the swarm end-to-end.

result = swarm.run(
    task="Develop product launch strategy",
    img=None,
)

`batched_run(tasks, ...)`

Run the swarm sequentially over a batch of tasks.

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.

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

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

swarm = HierarchicalSwarm(
    name="Research-Team",
    agents=researchers,
    planning_enabled=True,
    max_loops=3,
)

With planning_enabled=True:

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

Agent-as-Judge

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

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

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

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

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

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

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"
)

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

swarm = HierarchicalSwarm(
    agents=workers,
    max_loops=3,
    director_feedback_on=True,
)

Each loop:

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

Order Execution

Concurrent Execution (default)

Orders for multiple agents run in parallel via run_agents_concurrently:

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:

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

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

Specialized workers — each agent should have an obviously different area of expertise; overlapping workers waste tokens.
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.
Loop balance — every extra loop costs an additional director pass + every worker run. Budget accordingly.
Planning phase — enable for complex tasks that benefit from a strategy step.
Judge phase — enable when output quality is uneven and you want explicit scoring/feedback.
Dashboard — useful for demos and debugging.

Each loop runs the director and every selected worker. Costs grow roughly linearly with max_loops × len(agents).

Error Handling

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

run_agents_concurrently runs all agents in a loop simultaneously when parallel_execution=True:

results = run_agents_concurrently(
    agents=agents_to_run,
    task=self.conversation.get_str(),
)

Conversation Context

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

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

Heavy Swarm — fixed multi-phase analysis with question generation
Mixture of Agents — parallel experts + aggregator
Agent Rearrange — custom flows without a director
Sequential Workflow — simple linear flows
Structures Catalog — full enumeration of every multi-agent structure

​When to Use

​Key Features

​Basic Example

​Architecture Flow

​SwarmSpec Structure

​Key Parameters

​Methods

​run(task, img=None)

​batched_run(tasks, ...)

​run_stream(task, ...)

​display_hierarchy()

​feedback_director(outputs)

​run_judge_agent(outputs)

​Advanced Configuration

​Custom Director

​Planning + Multiple Loops

​Agent-as-Judge

​Team-Awareness Preamble

​Interactive Dashboard

​Autosave

​Use Cases

​Software Development Team

​Research Team

​Marketing Campaign

​Multi-Loop Refinement

​Order Execution

​Concurrent Execution (default)

​Sequential Execution

​Director System Prompt

​Best Practices

​Error Handling

​Performance Considerations

​Parallel Order Execution

​Conversation Context

​Related Architectures

When to Use

Key Features

Basic Example

Architecture Flow

SwarmSpec Structure

Key Parameters

Methods

`run(task, img=None)`

`batched_run(tasks, ...)`

`run_stream(task, ...)`

`display_hierarchy()`

`feedback_director(outputs)`

`run_judge_agent(outputs)`

Advanced Configuration

Custom Director

Planning + Multiple Loops

Agent-as-Judge

Team-Awareness Preamble

Interactive Dashboard

Autosave

Use Cases

Software Development Team

Research Team

Marketing Campaign

Multi-Loop Refinement

Order Execution

Concurrent Execution (default)

Sequential Execution

Director System Prompt

Best Practices

Error Handling

Performance Considerations

Parallel Order Execution

Conversation Context

Related Architectures