Swarm Router

from swarms import SwarmType

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

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)

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

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

router = SwarmRouter(
    swarm_type="HierarchicalSwarm",
    agents=worker_agents,
    max_loops=2,    # feedback loops
    director_model_name="claude-sonnet-4-6",
    director_settings={
        "agent_name": "LeadDirector",
        "temperature": 0.2,
    },
)

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)

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

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

result = router("Analyze market trends")

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

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

router.run("Analyze market trends")
print(router.usage["total_tokens"])

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

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)

from swarms.structs.swarm_router import SwarmRouterRunError

router = SwarmRouter(
    swarm_type="HierarchicalSwarm",
    fallback_swarms=["SequentialWorkflow", "ConcurrentWorkflow"],
    agents=agents,
    autosave=True,
)

try:
    result = router.run(task)
    print(f"Served by {router.active_swarm_type}")
    for attempt in router.fallback_attempts:
        print(f"Failed first: {attempt['swarm_type']}: {attempt['error']}")
except SwarmRouterRunError as e:
    print(f"Every swarm failed: {e}")

# 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 factory-backed SwarmType ...
    # Note: "auto" and "AutoSwarmBuilder" have no entry here — see the
    # warning under "Supported Swarm Types" above.
}

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

# 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
# - fallback_swarms, if set, is a list of valid swarm types

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

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

try:
    router = SwarmRouter(
        swarm_type="HierarchicalSwarm",
        fallback_swarms=["SequentialWorkflow"],
        agents=agents,
    )
except SwarmRouterConfigError as e:
    print(f"Configuration error: {e}")

try:
    result = router.run("Task")
except SwarmRouterRunError as e:
    print(f"Every swarm failed: {e}")

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",
    fallback_swarms=["SequentialWorkflow"],
    autosave=True,
    autosave_use_timestamp=True,
    director_model_name="gpt-5.4",
    verbose=True,
)