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)
Optional autosave of config / state / metadata to the workspace
Multi-agent collaboration prompt injection
Optional rules / shared-memory injection
O(1) factory dispatch with internal caching

Import

from swarms.structs.swarm_router import (
    SwarmRouter,
    SwarmType,
    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",
    "AutoSwarmBuilder",
    "HierarchicalSwarm",
    "auto",
    "MajorityVoting",
    "CouncilAsAJudge",
    "HeavySwarm",
    "BatchedGridWorkflow",
    "LLMCouncil",
    "DebateWithJudge",
    "RoundRobin",
    "PlannerWorkerSwarm",
]

"auto" and "AutoSwarmBuilder" are listed in the Literal but are not dispatched by the current factory (it handles 15 of the 17 values). Selecting either passes construction validation, then raises ValueError during run(). Use one of the other 15 values.

Constructor

SwarmRouter(
    id: str = generate_api_key(prefix="swarm-router"),
    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,
    shared_memory_system: Any = None,
    output_type: OutputType = "dict-all-except-first",
    multi_agent_collab_prompt: bool = True,
    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,
    telemetry_enabled: bool = False,
    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,
    *args,
    **kwargs,
)

Core parameters

str

default:"generate_api_key(prefix='swarm-router')"

Unique identifier for the router instance. 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).

OutputType

default:"dict-all-except-first"

Output formatter applied to the conversation. One of 'str', 'string', 'list', 'json', 'dict', 'dict-all-except-first', 'yaml', 'xml'.

bool

default:"False"

When True, snapshot each agent’s serialized config into self.agent_config during the reliability check.

Behavior modifiers

bool

default:"True"

Inject the built-in collaboration prompt into each agent so they understand they are part of a multi-agent team. Applied during setup() via update_system_prompt_for_agent_in_swarm().

Any

Shared memory backend (e.g. a vector store) attached to every agent’s long_term_memory.

bool

default:"False"

When True, embed a manifest of every agent (name + description) into the conversation so each agent can see its peers.

Any

Pre-existing Conversation instance to reuse instead of starting fresh.

Optional[Dict[Any, Any]]

Per-agent configuration overrides applied during swarm setup.

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 an ISO timestamp in the directory name when True; use a UUID 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.

CouncilAsAJudge

str

default:"gpt-5.4"

Model used by the judge agent.

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: unknown swarm_type, missing required parameter for the chosen type (e.g. rearrange_flow for AgentRearrange), max_loops <= 0, etc.

Exception

Raised by run() / batch_run() / concurrent_run() when task execution fails inside the underlying swarm. Wraps the original traceback.

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]

List of tasks. Consumed by BatchedGridWorkflow.

Returns: Any — shape depends on output_type and the underlying swarm. Raises: SwarmRouterRunError on execution failure.

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

`call(task, *args, **kwargs)`

Alias for run. Lets you treat a router as a callable.

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

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

Execute many tasks sequentially. Each task gets a fresh execution against the same router instance.

List[str]

required

Tasks to process.

str

Single image applied to every task.

List[str]

Per-task images aligned to tasks by index.

Returns: List[Any] — results in the same order as tasks.

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

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

Execute one task in a separate thread using a ThreadPoolExecutor. Useful when calling the router from synchronous code that should not block.

result = router.concurrent_run("Summarize Q1.")

`reliability_check()`

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

`setup()`

Applies pre-run configuration to the agents: auto-prompt-engineering (auto_generate_prompts), shared-memory activation (shared_memory_system), rules injection (rules), and the multi-agent collaboration prompt (multi_agent_collab_prompt). 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 conversation as a single formatted string.

`activate_shared_memory()` · `update_system_prompt_for_agent_in_swarm()` · `list_agents_to_eachother()`

Internal helpers invoked by setup(): attach shared_memory_system to every agent’s long_term_memory, append the multi-agent collaboration prompt to each agent, and (when list_all_agents=True) add the agent roster to the swarm conversation. agent_config() returns a {agent_name: agent.to_dict()} snapshot used when telemetry_enabled=True. Rarely called directly.

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)
`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)
`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`
`BatchedGridWorkflow`	`agents`, `tasks` passed to `run`
`RoundRobin`	`agents` (≥2)
`MultiAgentRouter`	`agents`
`PlannerWorkerSwarm`	`agents` (workers); planner is created automatically
`AutoSwarmBuilder`	Reserved — not dispatched by the current factory (raises `ValueError` at run)
`auto`	Reserved — not dispatched by the current factory (raises `ValueError` at run)

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 resolves to the SWARMS_WORKSPACE_DIR env var if set, otherwise to the configured Swarms workspace.

Usage Example

from swarms.structs.agent import Agent
from swarms.structs.swarm_router import 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
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

CouncilAsAJudge

LLMCouncil

Additional kwargs

Exceptions

Methods

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

`call(task, *args, **kwargs)`

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

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

`reliability_check()`

`setup()`

`to_dict()`

`fetch_message_history_as_string()`

`activate_shared_memory()` · `update_system_prompt_for_agent_in_swarm()` · `list_agents_to_eachother()`

Required parameters by `swarm_type`

Autosave layout

Usage Example

Source

​Overview

​Import

​SwarmType

​Constructor

​Core parameters

​Behavior modifiers

​Autosave

​Per-architecture parameters

​AgentRearrange

​GroupChat

​HeavySwarm

​CouncilAsAJudge

​LLMCouncil

​Additional kwargs

​Exceptions

​Methods

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

​__call__(task, *args, **kwargs)

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

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

​reliability_check()

​setup()

​to_dict()

​fetch_message_history_as_string()

​activate_shared_memory() · update_system_prompt_for_agent_in_swarm() · list_agents_to_eachother()

​Required parameters by swarm_type

​Autosave layout

​Usage Example

​Related Pages

​Source

Overview

Import

`SwarmType`

Constructor

Core parameters

Behavior modifiers

Autosave

Per-architecture parameters

AgentRearrange

GroupChat

HeavySwarm

CouncilAsAJudge

LLMCouncil

Additional kwargs

Exceptions

Methods

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

`call(task, *args, **kwargs)`

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

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

`reliability_check()`

`setup()`

`to_dict()`

`fetch_message_history_as_string()`

`activate_shared_memory()` · `update_system_prompt_for_agent_in_swarm()` · `list_agents_to_eachother()`

Required parameters by `swarm_type`

Autosave layout

Usage Example

Related Pages

Source