Swarm Router

The SwarmRouter is the highest-level multi-agent abstraction in the framework. Pass it a list of agents and a swarm_type — it builds the matching orchestrator (SequentialWorkflow, ConcurrentWorkflow, HierarchicalSwarm, MixtureOfAgents, HeavySwarm, etc.) and forwards run() calls to it. To switch architectures, change one string. The underlying swarm is built lazily on the first run() call and cached on the instance — repeated calls reuse it (keyed by swarm_type, agent identities, and construction-time config).

When to Use

Flexible orchestration — switch between swarm types without rewriting code
Strategy comparison — A/B different architectures on the same task
Unified interface — one API for every supported swarm
Dynamic selection — choose swarm_type at runtime
Production deployments — standardized swarm management with optional autosave

Key Features

17 swarm types supported (plus "auto" for automatic routing)
Factory pattern with O(1) lookup and per-instance swarm cache
Pre-flight reliability checks
Optional autosave of config.json / state.json / metadata.json
Shared memory injection across agents
Multi-agent collaboration prompt injection
Per-swarm-type specialized parameters (HeavySwarm, AgentRearrange, GroupChat, etc.)
Inherits SerializableMixin — to_dict() is available for telemetry / persistence

Supported Swarm Types

from swarms.structs.swarm_router import SwarmType

# SwarmType is a Literal — pass the matching string:
"SequentialWorkflow"
"ConcurrentWorkflow"
"AgentRearrange"
"MixtureOfAgents"
"GroupChat"
"MultiAgentRouter"
"AutoSwarmBuilder"
"HierarchicalSwarm"
"MajorityVoting"
"CouncilAsAJudge"
"HeavySwarm"
"BatchedGridWorkflow"
"LLMCouncil"
"DebateWithJudge"
"RoundRobin"
"PlannerWorkerSwarm"
"auto"             # let the router decide

Basic Example

from swarms import Agent, SwarmRouter

writer = Agent(
    agent_name="Writer",
    system_prompt="You are a creative writer.",
    model_name="gpt-5.4",
)

editor = Agent(
    agent_name="Editor",
    system_prompt="You are an expert editor.",
    model_name="gpt-5.4",
)

reviewer = Agent(
    agent_name="Reviewer",
    system_prompt="You are a quality reviewer.",
    model_name="gpt-5.4",
)

agents = [writer, editor, reviewer]
task = "Write a short story about AI"

# Sequential
seq = SwarmRouter(swarm_type="SequentialWorkflow", agents=agents)
seq_result = seq.run(task)

# Concurrent
conc = SwarmRouter(swarm_type="ConcurrentWorkflow", agents=agents)
conc_result = conc.run(task)

# Mixture of Agents (last agent acts as aggregator)
moa = SwarmRouter(swarm_type="MixtureOfAgents", agents=agents)
moa_result = moa.run(task)

Key Parameters

SwarmType

default:"\"SequentialWorkflow\""

Which orchestrator to instantiate. See the list above.

List[Union[Agent, Callable]]

required

The agent roster the swarm will use. The role of each agent depends on swarm_type (e.g. for DebateWithJudge the first two are debaters and the third is the judge; for MixtureOfAgents the last agent is the aggregator).

str

Stable identifier for this router instance. Auto-generated if omitted.

str

default:"\"swarm-router\""

Human-readable name. Used for log lines and autosave directory naming.

str

default:"\"Routes your task to the desired swarm\""

Free-text description of what this router is for.

int

default:"1"

Iteration count for the underlying swarm. Semantics depend on swarm_type (e.g. for MixtureOfAgents this is the number of layers).

OutputType

default:"\"dict-all-except-first\""

How the final swarm output is formatted.

bool

default:"False"

When True, save config.json at init and state.json + metadata.json after each run.

bool

default:"True"

If True, use a timestamp in the autosave directory name; otherwise use a UUID.

str

Required when swarm_type="AgentRearrange". Flow DSL like "A -> B, C -> D".

Any

Memory backend injected into every agent’s long_term_memory.

bool

default:"True"

Append the multi-agent collaboration preamble to every agent’s system prompt.

bool

default:"False"

When True, every agent is told about every other agent at the start of a run.

Any

Pre-existing conversation object to seed the swarm with.

Dict[Any, Any]

Optional per-agent config overrides.

bool

default:"False"

When True, snapshot each agent’s config into self.agent_config.

bool

default:"False"

Emit info / debug logs (reliability check, cache hits, swarm creation).

Swarm-Specific Parameters

`AgentRearrange`

str

required

Flow DSL (e.g. "researcher -> writer, editor").

router = SwarmRouter(
    swarm_type="AgentRearrange",
    agents=agents,
    rearrange_flow="researcher -> writer, editor",
)

`HeavySwarm`

str

default:"\"gpt-5.4\""

Model for the HeavySwarm question agent.

str

default:"\"gpt-5.4\""

Model for HeavySwarm workers.

bool

default:"True"

Print per-agent output for HeavySwarm.

Literal["default", "medium", "heavy"]

default:"\"default\""

HeavySwarm architecture variant. See the Heavy Swarm docs.

int

default:"1"

Iteration count for HeavySwarm multi-loop refinement.

int

default:"900"

Per-worker wall-clock cap (seconds) for HeavySwarm.

List[Callable]

Tools passed to HeavySwarm workers.

router = SwarmRouter(
    swarm_type="HeavySwarm",
    agents=agents,
    heavy_swarm_worker_model_name="claude-sonnet-4-6",
    heavy_swarm_question_agent_model_name="gpt-5.4",
    heavy_swarm_variant="medium",
    heavy_swarm_max_loops=2,
)

`HierarchicalSwarm`

router = SwarmRouter(
    swarm_type="HierarchicalSwarm",
    agents=worker_agents,
    max_loops=2,    # feedback loops
)

`CouncilAsAJudge`

str

default:"\"gpt-5.4\""

Model used as the council judge.

`LLMCouncil`

str

default:"\"gpt-5.1\""

Chairman model for LLMCouncil.

Advanced Features

Shared Memory

from swarms.memory import ChromaDB

memory = ChromaDB()

router = SwarmRouter(
    swarm_type="SequentialWorkflow",
    agents=agents,
    shared_memory_system=memory,
)

Autosave

router = SwarmRouter(
    swarm_type="HierarchicalSwarm",
    agents=agents,
    autosave=True,
    autosave_use_timestamp=True,
)

# Saves to: $WORKSPACE_DIR/swarms/SwarmRouter/{swarm-name}-{timestamp}/
#   config.json    (on initialization)
#   state.json     (after each run)
#   metadata.json  (after each run)

Tell Every Agent About Every Other Agent

router = SwarmRouter(
    swarm_type="SequentialWorkflow",
    agents=agents,
    list_all_agents=True,
)

Methods

`run(task=None, img=None, tasks=None, ...)`

Execute the configured swarm with a single task (or a list of tasks, when the underlying swarm accepts one).

result = router.run(
    task="Analyze market trends",
    img=None,
)

`call(task, img=None, imgs=None, ...)`

The router is directly callable as a shortcut for run().

result = router("Analyze market trends")

`batch_run(tasks, img=None, imgs=None, ...)`

Process multiple tasks sequentially. Re-uses the cached underlying swarm.

tasks = ["Task 1", "Task 2", "Task 3"]
results = router.batch_run(tasks)

`concurrent_run(...)`

Run multiple tasks concurrently.

`to_dict()`

Inherited from SerializableMixin. Returns a JSON-friendly snapshot of the router configuration.

Use Cases

Strategy Comparison

strategies = [
    "SequentialWorkflow",
    "ConcurrentWorkflow",
    "MixtureOfAgents",
]

results = {}
for strategy in strategies:
    router = SwarmRouter(swarm_type=strategy, agents=agents)
    results[strategy] = router.run(task)

for strategy, result in results.items():
    print(f"\n{strategy}:\n{result}")

Dynamic Swarm Selection

def select_swarm_type(task_complexity: str) -> str:
    if task_complexity == "simple":
        return "SequentialWorkflow"
    elif task_complexity == "parallel":
        return "ConcurrentWorkflow"
    elif task_complexity == "complex":
        return "HierarchicalSwarm"
    return "MixtureOfAgents"

task = "Complex analysis required"
complexity = analyze_complexity(task)

router = SwarmRouter(
    swarm_type=select_swarm_type(complexity),
    agents=agents,
)

result = router.run(task)

Production Pipeline with Fallback

class ProductionSwarmRouter:
    def __init__(self, agents):
        self.router = SwarmRouter(
            swarm_type="HierarchicalSwarm",
            agents=agents,
            autosave=True,
            verbose=True,
        )

    def process(self, task):
        try:
            return self.router.run(task)
        except Exception as e:
            print(f"Error with HierarchicalSwarm: {e}")
            fallback = SwarmRouter(
                swarm_type="SequentialWorkflow",
                agents=self.router.agents,
            )
            return fallback.run(task)

Factory Pattern

The router maintains a per-instance factory dispatch table and a swarm cache:

# Internal layout
self._swarm_factory = {
    "SequentialWorkflow": self._create_sequential_workflow,
    "ConcurrentWorkflow": self._create_concurrent_workflow,
    "AgentRearrange":     self._create_agent_rearrange,
    "MixtureOfAgents":    self._create_mixture_of_agents,
    "HierarchicalSwarm":  self._create_hierarchical_swarm,
    "GroupChat":          self._create_group_chat,
    "HeavySwarm":         self._create_heavy_swarm,
    # ... one entry per SwarmType ...
    "auto":               self._create_auto,
}

# First run() builds, subsequent runs reuse:
swarm = self._swarm_cache.get(cache_key) or factory(*args, **kwargs)
self._swarm_cache[cache_key] = swarm

Reliability Checks

reliability_check() runs automatically during construction:

# Validates:
# - swarm_type is not None
# - swarm_type is a string
# - swarm_type is one of the valid SwarmType values
# - rearrange_flow is set when swarm_type="AgentRearrange"
# - max_loops > 0

try:
    router = SwarmRouter(
        swarm_type="InvalidType",
        agents=agents,
    )
except SwarmRouterConfigError as e:
    print(e)
    # Includes the offending value and the list of valid types

Error Handling

from swarms.structs.swarm_router import (
    SwarmRouterConfigError,
    SwarmRouterRunError,
)

try:
    result = router.run("Task")
except SwarmRouterRunError as e:
    print(f"Execution failed: {e}")
    # The exception body includes:
    # - The reason for failure
    # - A formatted traceback
    # - Troubleshooting hints
except SwarmRouterConfigError as e:
    print(f"Configuration error: {e}")
    # Invalid swarm_type, missing required params, etc.

Best Practices

Start simple: begin with SequentialWorkflow, then escalate to a heavier topology only when you can name the failure mode it fixes.

Match swarm_type to the task — pipelines for known shapes, ensembles for quality, hierarchies for decomposition.
Validate required params — e.g. AgentRearrange needs rearrange_flow; HeavySwarm honors the heavy_swarm_* knobs.
Wrap run() in try/except — catch SwarmRouterRunError and SwarmRouterConfigError explicitly in production.
Test with simple types first — confirm agents and tools work, then swap in heavier swarms.
Enable autosave for production — durable config.json / state.json / metadata.json make incident analysis far cheaper.

Some swarm types have specific requirements: AgentRearrange requires rearrange_flow; MixtureOfAgents consumes the last agent in the list as the aggregator; DebateWithJudge consumes the third agent as the judge.

Configuration Reference

Complete example with the most common options:

router = SwarmRouter(
    id="my-swarm-123",
    name="Production-Swarm",
    description="Production multi-agent system",
    swarm_type="HierarchicalSwarm",
    agents=agents,
    max_loops=2,
    output_type="dict",
    autosave=True,
    autosave_use_timestamp=True,
    multi_agent_collab_prompt=True,
    list_all_agents=True,
    shared_memory_system=memory,
    telemetry_enabled=True,
    verbose=True,
)

Sequential Workflow
Concurrent Workflow
Agent Rearrange
Heavy Swarm
Hierarchical Swarm
Structures Catalog — full enumeration of every multi-agent structure
All Architectures Overview

Getting Started

Core Concepts

Agent Development

Reasoning Agents

Multi-Agent Architectures

Deployment & Operations

Integrations

CLI Reference

Changelog

When to Use

Key Features

Supported Swarm Types

Basic Example

Key Parameters

Swarm-Specific Parameters

`AgentRearrange`

`HeavySwarm`

`HierarchicalSwarm`

`CouncilAsAJudge`

`LLMCouncil`

Advanced Features

Shared Memory

Autosave

Tell Every Agent About Every Other Agent

Methods

`run(task=None, img=None, tasks=None, ...)`

`call(task, img=None, imgs=None, ...)`

`batch_run(tasks, img=None, imgs=None, ...)`

`concurrent_run(...)`

`to_dict()`

Use Cases

Strategy Comparison

Dynamic Swarm Selection

Production Pipeline with Fallback

Factory Pattern

Reliability Checks

Error Handling

Best Practices

Configuration Reference

​When to Use

​Key Features

​Supported Swarm Types

​Basic Example

​Key Parameters

​Swarm-Specific Parameters

​AgentRearrange

​HeavySwarm

​HierarchicalSwarm

​CouncilAsAJudge

​LLMCouncil

​Advanced Features

​Shared Memory

​Autosave

​Tell Every Agent About Every Other Agent

​Methods

​run(task=None, img=None, tasks=None, ...)

​__call__(task, img=None, imgs=None, ...)

​batch_run(tasks, img=None, imgs=None, ...)

​concurrent_run(...)

​to_dict()

​Use Cases

​Strategy Comparison

​Dynamic Swarm Selection

​Production Pipeline with Fallback

​Factory Pattern

​Reliability Checks

​Error Handling

​Best Practices

​Configuration Reference

​Related Architectures

When to Use

Key Features

Supported Swarm Types

Basic Example

Key Parameters

Swarm-Specific Parameters

`AgentRearrange`

`HeavySwarm`

`HierarchicalSwarm`

`CouncilAsAJudge`

`LLMCouncil`

Advanced Features

Shared Memory

Autosave

Tell Every Agent About Every Other Agent

Methods

`run(task=None, img=None, tasks=None, ...)`

`call(task, img=None, imgs=None, ...)`

`batch_run(tasks, img=None, imgs=None, ...)`

`concurrent_run(...)`

`to_dict()`

Use Cases

Strategy Comparison

Dynamic Swarm Selection

Production Pipeline with Fallback

Factory Pattern

Reliability Checks

Error Handling

Best Practices

Configuration Reference

Related Architectures