Swarm

Swarm provides autonomous, collaborative agent orchestration. Agents work together as a team with shared working memory, deciding among themselves who handles which part of a task.

Unlike Graph, there's no fixed execution order — agents coordinate dynamically based on the task at hand.

Basic example

from elsai import Agent
from elsai.multiagent import Swarm

coordinator = Agent(
    name="coordinator",
    system_prompt="You coordinate a team of specialists. "
                  "Break down tasks and delegate to the right expert."
)

data_analyst = Agent(
    name="data_analyst",
    system_prompt="You analyse numerical data and produce insights.",
)

writer = Agent(
    name="writer",
    system_prompt="You write clear, engaging reports based on data insights.",
)

swarm = Swarm(agents=[coordinator, data_analyst, writer])
result = swarm("Analyse our Q4 sales data and write an executive summary")
print(result)

How it works

1. The initial task goes to all agents as shared context
2. A coordinator agent (first agent or explicitly designated) assigns sub-tasks
3. Agents execute their sub-tasks and share results in a shared workspace
4. The swarm continues until the coordinator is satisfied with the outcome
5. The final output is returned

Shared working memory

All agents in a swarm share a working memory (conversation context):

swarm = Swarm(
    agents=[coordinator, researcher, coder],
    shared_state={"project": "MyApp", "deadline": "2025-12-01"},
)

Async execution

import asyncio
from elsai.multiagent import Swarm

async def main():
    result = await swarm.invoke_async("Build a REST API spec for a todo app")
    print(result)

asyncio.run(main())

Hooks

from elsai.hooks import BeforeNodeCallEvent

def track_agents(event: BeforeNodeCallEvent) -> None:
    print(f"Agent {event.node.name} is taking a turn")

swarm = Swarm(agents=[...], hooks=[track_agents])

When to use Swarm vs Graph

	Swarm	Graph
Execution order	Dynamic (agents decide)	Fixed (edges define order)
Coordination	Autonomous	Developer-controlled
Best for	Open-ended, creative tasks	Structured pipelines
Predictability	Lower	Higher
Debugging	Harder	Easier

SwarmResult

result = swarm("Do a complex task")

print(result.output)       # Final combined output
print(result.status)       # "completed" | "failed" | "interrupted"
print(result.node_results) # Each agent's contributions
print(result.metrics)      # Total token usage