SwarmRouter

Overview

The SwarmRouter class dispatches tasks to any supported multi-agent architecture through a single, uniform interface. Set swarm_type to choose the architecture; pass the same agent list and call run(task) regardless of which architecture you select. It also handles:

Validated construction of the underlying swarm (reliability_check)
Fallback to other swarm types when the primary one fails (fallback_swarms)
Token usage summed across every agent the swarm ran (usage)
Optional autosave of config / state / metadata to the workspace
Multi-agent collaboration prompt injection
O(1) factory dispatch with internal caching

Import

from swarms import SwarmRouter, SwarmType
from swarms.structs.swarm_router import (
    SwarmRouterConfig,
    SwarmRouterRunError,
    SwarmRouterConfigError,
)

`SwarmType`

SwarmType is a typing.Literal enumerating every accepted swarm_type string. Any value outside this set raises SwarmRouterConfigError at construction time.

SwarmType = Literal[
    "AgentRearrange",
    "MixtureOfAgents",
    "SequentialWorkflow",
    "ConcurrentWorkflow",
    "GroupChat",
    "MultiAgentRouter",
    "HierarchicalSwarm",
    "MajorityVoting",
    "CouncilAsAJudge",
    "HeavySwarm",
    "LLMCouncil",
    "DebateWithJudge",
    "RoundRobin",
    "PlannerWorkerSwarm",
]

"AutoSwarmBuilder" is not in the SwarmType Literal — passing it raises SwarmRouterConfigError at construction. Neither "auto" nor "BatchedGridWorkflow" is a member either (14 values total) — passing either raises SwarmRouterConfigError at construction time, the same as any other invalid value.

Constructor

SwarmRouter(
    id: Optional[str] = None,  # resolves to generate_id("swarm-router") when not set
    name: str = "swarm-router",
    description: str = "Routes your task to the desired swarm",
    max_loops: int = 1,
    agents: List[Union[Agent, Callable]] = [],
    swarm_type: SwarmType = "SequentialWorkflow",
    autosave: bool = False,
    rearrange_flow: Optional[str] = None,
    fallback_swarms: Optional[List[SwarmType]] = None,
    output_type: OutputType = "dict",
    multi_agent_collab_prompt: bool = False,
    list_all_agents: bool = False,
    conversation: Any = None,
    agents_config: Optional[Dict[Any, Any]] = None,
    heavy_swarm_question_agent_model_name: str = "gpt-5.4",
    heavy_swarm_worker_model_name: str = "gpt-5.4",
    heavy_swarm_swarm_show_output: bool = True,
    heavy_swarm_variant: SwarmVariant = "default",
    heavy_swarm_max_loops: int = 1,
    heavy_swarm_timeout: int = 900,
    council_judge_model_name: str = "gpt-5.4",
    verbose: bool = False,
    worker_tools: Optional[List[Callable]] = None,
    chairman_model: str = "gpt-5.1",
    autosave_use_timestamp: bool = True,
    director_model_name: str = "gpt-5.4",
    director_settings: Optional[Dict[str, Any]] = None,
    *args,
    **kwargs,
)

Core parameters

Optional[str]

default:"None"

Unique identifier for the router instance. When left as None, resolves to generate_id("swarm-router"). Used for autosave directory naming and telemetry.

str

default:"swarm-router"

Human-readable name. Used for autosave directory naming.

str

default:"Routes your task to the desired swarm"

Free-form description of the router’s purpose.

List[Union[Agent, Callable]]

default:"[]"

Agents the router will pass to the selected swarm. Some swarm types require a specific minimum count (e.g. DebateWithJudge needs at least three).

SwarmType

default:"SequentialWorkflow"

Which architecture to instantiate. See the SwarmType Literal above.

int

default:"1"

Maximum execution loops where supported by the underlying swarm (e.g. HierarchicalSwarm, GroupChat). For MixtureOfAgents it is the number of layers. HeavySwarm uses heavy_swarm_max_loops instead, and CouncilAsAJudge, LLMCouncil, and MultiAgentRouter do not receive it. Cannot be 0.

OutputType

default:"dict"

Output formatter applied to the conversation. One of 'list', 'dict', 'dictionary', 'string', 'str', 'final', 'last', 'json', 'all', 'yaml', 'dict-all-except-first', 'str-all-except-first', 'basemodel', 'dict-final', 'list-final'. HeavySwarm, DebateWithJudge, CouncilAsAJudge, and LLMCouncil receive it too; RoundRobin does not.

Optional[List[SwarmType]]

default:"None"

Swarm types to try, in order, when the primary swarm_type raises while being built or run. Each fallback is built from the same agents and configuration and receives the same payload; the first to complete wins. If every swarm fails, SwarmRouterRunError is raised listing each attempt, with the last error as its cause. Validated at construction: it must be a list of valid SwarmType strings, and rearrange_flow must be set if it includes "AgentRearrange". When None, the primary swarm’s error propagates unchanged. See Fallback swarms.

Behavior modifiers

bool

default:"False"

Deliver the built-in collaboration prompt to each agent so they understand they are part of a multi-agent team. It is passed to the swarm at construction and reaches the agents as a system turn at run time; the caller’s Agent objects are never modified.Only SequentialWorkflow and AgentRearrange can deliver it. For any other swarm_type a warning is logged and the flag has no effect.

bool

default:"False"

When True, every agent receives a manifest of its peers (name + description) in the same system turn as the collaboration prompt. Like the collaboration prompt, it is only delivered by SequentialWorkflow and AgentRearrange; other swarm types ignore it without a warning.

Any

Stored as router.conversation. It is not passed to the underlying swarm. After each run, router.conversation is repointed at the swarm’s own conversation.

Optional[Dict[Any, Any]]

Stored as router.agents_config. No swarm factory reads it.

bool

default:"False"

Emit detailed logs during construction and execution.

Autosave

bool

default:"False"

When enabled, persist config.json on construction and state.json + metadata.json after each run to {workspace_dir}/swarms/SwarmRouter/{swarm-name}-{stamp}/.

bool

default:"True"

Use a YYYYmmdd_HHMMSS timestamp in the directory name when True; use a 12-character UUID fragment when False.

Per-architecture parameters

These are only consulted when the corresponding swarm_type is selected.

AgentRearrange

str

required

Flow DSL string, e.g. "researcher -> writer, editor". Required for swarm_type="AgentRearrange".

GroupChat

GroupChat is asynchronous and self-selecting — there are no speaker-selection functions. The router passes agents, max_loops, output_type, and verbose through to it; agents decide on their own whether to speak. See the GroupChat API reference for threshold / idle_timeout tuning (configure those by constructing the GroupChat directly).

HeavySwarm

str

default:"gpt-5.4"

Model used by the question-generation agent.

str

default:"gpt-5.4"

Model used by every worker agent.

bool

default:"True"

Print per-agent output during execution (agent_prints_on). The live dashboard is forced off when HeavySwarm is created through the router.

SwarmVariant ("default" | "medium" | "heavy")

default:"\"default\""

Which agent line-up to instantiate: "default" → 5 agents, "medium" → 4 agents (Captain + Harper / Benjamin / Lucas), "heavy" → 16 agents (Grok captain + 15 specialists). SwarmVariant is exported from swarms.agents.heavy_swarm_agents.

int

default:"1"

Number of iterative-refinement loops for HeavySwarm.

int

default:"900"

Per-worker wall-clock cap, in seconds, for HeavySwarm.

List[Callable]

Tools made available to every worker agent in HeavySwarm.

HierarchicalSwarm

str

default:"gpt-5.4"

Model used by the auto-created director agent. Forwarded to HierarchicalSwarm by _create_hierarchical_swarm().

Dict[str, Any]

Extra Agent keyword arguments for the auto-created director. Every key is forwarded to the director’s Agent; agent_name, model_name, system_prompt, temperature, and top_p override the corresponding defaults — including director_model_name. A planning_system_prompt key sets the planning director’s prompt instead.

CouncilAsAJudge

str

default:"gpt-5.4"

Passed to CouncilAsAJudge as model_name, so it sets the model for every dimension judge. The aggregator keeps CouncilAsAJudge’s own aggregation_model_name default ("gpt-5.4"), which SwarmRouter does not expose.

LLMCouncil

str

default:"gpt-5.1"

Model used by the chairman that synthesizes the council members’ responses.

Additional kwargs

The constructor accepts arbitrary *args / **kwargs for forward-compatibility, but they are not stored on the router or forwarded to the underlying swarm — extra constructor kwargs are effectively ignored. Configure the underlying swarm through the router’s explicit parameters listed above. For MixtureOfAgents, the aggregator is not passed as a kwarg; the last agent in agents is used as the aggregator automatically:

router = SwarmRouter(
    swarm_type="MixtureOfAgents",
    agents=[worker_1, worker_2, aggregator],  # aggregator = agents[-1]
)

Exceptions

Exception

Raised by reliability_check() when construction inputs are invalid: swarm_type is None, not a string, or not a SwarmType member; rearrange_flow is missing for AgentRearrange; max_loops == 0; or fallback_swarms is not a list of valid swarm types.

Exception

Raised by run() only when fallback_swarms is set and every swarm in the chain fails. The message names each attempt, and __cause__ is the last error. Without fallback_swarms, run() re-raises whatever the swarm raised.

If the selected factory itself raises while building the swarm, the error is wrapped in a RuntimeError ("Failed to create swarm ..."). batch_run() wraps any per-task failure in a RuntimeError.

Methods

`run(task=None, img=None, tasks=None, *args, **kwargs)`

Execute a single task using the configured swarm.

str

The task to be executed. Required for most swarm_type values.

str

Optional image input forwarded to vision-capable agents.

List[str]

Batch-style task list, forwarded as tasks= to the underlying swarm’s run() in place of task when provided. Only meaningful for swarm types whose run() accepts a tasks argument.

Extra **kwargs are forwarded to the underlying swarm’s run(). img is forwarded only when it is set. Each call is recorded as a SwarmRouter.run trace span when telemetry is enabled. Returns: Any — shape depends on output_type and the underlying swarm. Raises: the swarm’s own exception, or SwarmRouterRunError when fallback_swarms is set and every swarm fails.

result = router.run("Analyse Q1 market trends.")

`call(task, img=None, imgs=None, *args, **kwargs)`

Alias for run. Lets you treat a router as a callable. imgs is forwarded to the swarm’s run() as imgs= only when you pass it, so swarms whose run() has no imgs parameter still work.

result = router("Analyse Q1 market trends.")

`batch_run(tasks, img=None, imgs=None, *args, **kwargs)`

Execute many tasks sequentially by calling run() once per task. The cached swarm is reused across tasks.

List[str]

required

Tasks to process.

str

Single image applied to every task.

List[str]

Forwarded unchanged as imgs= to every run() call, and only when given. It is not split per task; use concurrent_run() for one image per task.

Returns: List[Any] — results in the same order as tasks. Raises: RuntimeError wrapping the first failure.

results = router.batch_run([
    "Summarize Q1.",
    "Summarize Q2.",
    "Summarize Q3.",
])

`concurrent_run(tasks, imgs=None, *args, **kwargs)`

Execute tasks through run() in parallel on a thread pool sized to os.cpu_count(). The concurrent counterpart to batch_run(). The tracing context follows each task into its worker thread.

List[str]

required

Tasks to execute in parallel.

List[str]

One image per task, paired by position. Must be the same length as tasks, or ValueError is raised.

Returns: List[Any] — one result per task, in task order.

results = router.concurrent_run([
    "Summarize Q1.",
    "Summarize Q2.",
])

`reliability_check()`

Validates the configuration (swarm type, rearrange_flow, max_loops, fallback_swarms) and then runs setup(). Called automatically at the end of __init__; you do not normally call it yourself. Raises SwarmRouterConfigError on invalid input.

`setup()`

Logs a warning when multi_agent_collab_prompt=True is set for a swarm type that cannot deliver it. It does not modify the agents: the collaboration preamble is passed to the swarm when it is built. Called automatically during construction via reliability_check(). The underlying swarm itself is created lazily on the first run().

`to_dict()`

Serialize the router’s configuration and current state to a dictionary suitable for logging or persistence. Returns: Dict[str, Any].

config = router.to_dict()

`fetch_message_history_as_string()`

Return the underlying swarm’s conversation as a single string, excluding the first message. Call it after at least one run(). Returns None if no history is available.

`list_agents_to_eachother()`

Internal helper. Points self.conversation at the underlying swarm’s conversation. Called after each run, because the swarm is built lazily on the first run() and several structures replace their conversation object per task. Rarely called directly.

update_system_prompt_for_agent_in_swarm() was removed. It appended the collaboration prompt to each agent’s system_prompt after the LLM had been built, so the text never reached the model, while accumulating on the caller’s live agents on every construction. The preamble is now passed through the swarm constructor and delivered as a system turn instead.

Required parameters by `swarm_type`

`swarm_type`	Additional required parameters
`SequentialWorkflow`	`agents` (≥1)
`ConcurrentWorkflow`	`agents` (≥1)
`AgentRearrange`	`agents`, `rearrange_flow`
`MixtureOfAgents`	`agents` (workers + aggregator as the last element; the full list, aggregator included, also answers as workers)
`HierarchicalSwarm`	`agents` (workers); director is created automatically
`GroupChat`	`agents` (≥2; asynchronous, self-selecting — no speaker function)
`MajorityVoting`	`agents` (≥3 recommended)
`CouncilAsAJudge`	`council_judge_model_name` (builds its own agents; `agents` is not passed)
`LLMCouncil`	`agents` (council members), `chairman_model`
`DebateWithJudge`	`agents` (≥3: pro, con, judge)
`HeavySwarm`	`heavy_swarm_worker_model_name`, `heavy_swarm_question_agent_model_name`, `heavy_swarm_variant`
`RoundRobin`	`agents` (≥2)
`MultiAgentRouter`	`agents`
`PlannerWorkerSwarm`	`agents` (workers); planner is created automatically

"AutoSwarmBuilder", "auto", and "BatchedGridWorkflow" are not members of the SwarmType Literal — passing any of them raises SwarmRouterConfigError at construction time, before run() is ever reached.

Autosave layout

When autosave=True, the router writes to:

{workspace_dir}/swarms/SwarmRouter/{name}-{timestamp_or_uuid}/
├── config.json    # written on __init__
├── state.json     # rewritten after each run
└── metadata.json  # rewritten after each run

workspace_dir is the WORKSPACE_DIR environment variable. When it is unset, the router sets it to ./agent_workspace under the current directory. metadata.json records the task, the swarm_type that served the run, and any fallback_attempts.

Fallback swarms

Set fallback_swarms to keep a run alive when the primary architecture fails. The router tries each type in order and stops at the first one that completes.

from swarms import Agent, SwarmRouter

agents = [
    Agent(agent_name="Researcher", model_name="gpt-5.4", max_loops=1),
    Agent(agent_name="Analyst", model_name="gpt-5.4", max_loops=1),
    Agent(agent_name="Writer", model_name="gpt-5.4", max_loops=1),
]

router = SwarmRouter(
    name="resilient-router",
    agents=agents,
    swarm_type="HierarchicalSwarm",
    fallback_swarms=["SequentialWorkflow", "ConcurrentWorkflow"],
)

result = router.run("Assess whether a SaaS company should adopt usage-based pricing.")

print(router.active_swarm_type)
for attempt in router.fallback_attempts:
    print(attempt["swarm_type"], attempt["error"])

Each swarm type is cached under its own key, so later runs reuse whichever swarms were already built.

Attributes

dict

Provider token usage summed over every distinct Agent the router has run: the agents you passed plus any agent a built swarm holds on its own, such as a HierarchicalSwarm director, a MixtureOfAgents aggregator, or a judge. Keys: input_tokens, output_tokens, cached_tokens, reasoning_tokens, total_tokens. Agent totals are lifetime totals, so an agent shared with another router contributes what it spent there too. Read-only property. See Token usage.

str

The swarm type that served the most recent run. Equal to swarm_type unless a fallback was used.

List[dict]

One {"swarm_type": str, "error": Exception} entry per swarm that failed during the most recent run. Empty when the primary swarm succeeded.

Any

The underlying swarm that served the most recent run. None until the first run().

Optional[str]

The autosave directory, or None when autosave=False.

before = router.usage
router.run("Should a fifty-person engineering org adopt a monorepo?")
after = router.usage
print({k: after[k] - before[k] for k in after})

Usage Example

from swarms import Agent, SwarmRouter

researcher = Agent(
    agent_name="Research-Agent",
    system_prompt="You are a research specialist.",
    model_name="claude-sonnet-4-6",
)

analyst = Agent(
    agent_name="Analysis-Agent",
    system_prompt="You analyze research findings into actionable insights.",
    model_name="claude-sonnet-4-6",
)

router = SwarmRouter(
    name="research-analysis-swarm",
    description="A swarm for research + analysis tasks",
    agents=[researcher, analyst],
    swarm_type="SequentialWorkflow",
    max_loops=1,
    autosave=True,
    verbose=True,
)

result = router.run("Research recent trends in AI hardware and analyze them.")
print(result)

SwarmRouter architecture overview
SwarmRouter example walkthrough
Token usage — how Agent.usage is counted
MultiAgentRouter API reference — task-aware routing to one of N agents
AutoSwarmBuilder API reference — LLM-driven swarm construction

Source

swarms/structs/swarm_router.py on GitHub

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

Overview

Import

`SwarmType`

Constructor

Core parameters

Behavior modifiers

Autosave

Per-architecture parameters

AgentRearrange

GroupChat

HeavySwarm

HierarchicalSwarm

CouncilAsAJudge

LLMCouncil

Additional kwargs

Exceptions

Methods

`run(task=None, img=None, tasks=None, *args, **kwargs)`

`call(task, img=None, imgs=None, *args, **kwargs)`

`batch_run(tasks, img=None, imgs=None, *args, **kwargs)`

`concurrent_run(tasks, imgs=None, *args, **kwargs)`

`reliability_check()`

`setup()`

`to_dict()`

`fetch_message_history_as_string()`

`list_agents_to_eachother()`

Required parameters by `swarm_type`

Autosave layout

Fallback swarms

Attributes

Usage Example

Source

​Overview

​Import

​SwarmType

​Constructor

​Core parameters

​Behavior modifiers

​Autosave

​Per-architecture parameters

​AgentRearrange

​GroupChat

​HeavySwarm

​HierarchicalSwarm

​CouncilAsAJudge

​LLMCouncil

​Additional kwargs

​Exceptions

​Methods

​run(task=None, img=None, tasks=None, *args, **kwargs)

​__call__(task, img=None, imgs=None, *args, **kwargs)

​batch_run(tasks, img=None, imgs=None, *args, **kwargs)

​concurrent_run(tasks, imgs=None, *args, **kwargs)

​reliability_check()

​setup()

​to_dict()

​fetch_message_history_as_string()

​list_agents_to_eachother()

​Required parameters by swarm_type

​Autosave layout

​Fallback swarms

​Attributes

​Usage Example

​Related Pages

​Source

Overview

Import

`SwarmType`

Constructor

Core parameters

Behavior modifiers

Autosave

Per-architecture parameters

AgentRearrange

GroupChat

HeavySwarm

HierarchicalSwarm

CouncilAsAJudge

LLMCouncil

Additional kwargs

Exceptions

Methods

`run(task=None, img=None, tasks=None, *args, **kwargs)`

`call(task, img=None, imgs=None, *args, **kwargs)`

`batch_run(tasks, img=None, imgs=None, *args, **kwargs)`

`concurrent_run(tasks, imgs=None, *args, **kwargs)`

`reliability_check()`

`setup()`

`to_dict()`

`fetch_message_history_as_string()`

`list_agents_to_eachother()`

Required parameters by `swarm_type`

Autosave layout

Fallback swarms

Attributes

Usage Example

Related Pages

Source