Social Swarm Patterns

Low-level communication patterns compose custom multi-agent flows before you reach full workflows like SequentialWorkflow or GraphWorkflow. Every function and class on this page is exported from the top-level swarms package. For parameters and return shapes, see the Swarming Architectures reference.

pip install -U swarms

For higher-level orchestration, see Social Algorithms and Agent Rearrange.

One sender speaks, then each receiver replies in turn on the shared conversation, seeing the sender and the replies before its own. The broadcast function is async; the Broadcast class has a synchronous run:

import asyncio
from swarms import Agent, broadcast

sender = Agent(agent_name="Announcer", model_name="claude-sonnet-4-6", max_loops=1)
receivers = [
    Agent(agent_name="Analyst-A", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Analyst-B", model_name="claude-sonnet-4-6", max_loops=1),
]

async def main():
    results = await broadcast(
        sender=sender,
        agents=receivers,
        task="Summarize Q4 priorities for your domain.",
    )
    print(results)

asyncio.run(main())

Agents pass work in a ring with shared conversation history:

from swarms import Agent, circular_swarm

agents = [
    Agent(agent_name="Researcher", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Analyst", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Writer", model_name="claude-sonnet-4-6", max_loops=1),
]

result = circular_swarm(agents=agents, tasks=["Draft a one-page market brief."])
print(result)

Shared task queue; workers drain tasks in round-robin order:

from swarms import Agent, mesh_swarm

agents = [
    Agent(agent_name=f"Worker-{i}", model_name="claude-sonnet-4-6", max_loops=1)
    for i in range(3)
]
tasks = ["Task A", "Task B", "Task C", "Task D"]

results = mesh_swarm(agents=agents, tasks=tasks)
print(results)

Agents in a square grid (list length should be a perfect square, e.g. 4 or 9):

from swarms import Agent, grid_swarm

agents = [
    Agent(agent_name=f"A-{i}", model_name="claude-sonnet-4-6", max_loops=1)
    for i in range(4)
]
result = grid_swarm(agents=agents, tasks=["Grid task 1", "Grid task 2"])
print(result)

First agent is the hub; it processes each task before the others:

from swarms import Agent, star_swarm

agents = [
    Agent(agent_name="Hub", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Spoke-1", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Spoke-2", model_name="claude-sonnet-4-6", max_loops=1),
]

result = star_swarm(agents=agents, tasks=["Coordinate subtasks and merge."])
print(result)

Agents arranged in a pyramid; tasks flow level by level:

from swarms import Agent, pyramid_swarm

agents = [
    Agent(agent_name="Director", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Worker-1", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Worker-2", model_name="claude-sonnet-4-6", max_loops=1),
]
result = pyramid_swarm(agents=agents, tasks=["Planning task"])
print(result)

Sender and receiver alternate on a single task, max_loops times. OneToOne wraps the same logic in a reusable class:

from swarms import Agent, one_to_one

sender = Agent(agent_name="Sender", model_name="claude-sonnet-4-6", max_loops=1)
receiver = Agent(agent_name="Receiver", model_name="claude-sonnet-4-6", max_loops=1)

result = one_to_one(sender=sender, receiver=receiver, task="Review this draft section.")
print(result)

One sender, then exactly three receivers reply in turn. Any other receiver count raises ValueError:

from swarms import Agent, OneToThree

founder = Agent(agent_name="Founder", model_name="claude-sonnet-4-6", max_loops=1)
investors = [
    Agent(agent_name=name, model_name="claude-sonnet-4-6", max_loops=1)
    for name in ["Growth-VC", "Value-Investor", "Angel"]
]

panel = OneToThree(sender=founder, receivers=investors)
result = panel.run("Pitch a marketplace for renting professional camera gear.")
print(result)

Run workers concurrently, then synthesize with an aggregator agent:

from swarms import Agent, aggregate

workers = [
    Agent(agent_name=f"Expert-{i}", model_name="claude-sonnet-4-6", max_loops=1)
    for i in range(3)
]

merged = aggregate(workers=workers, task="Give one bullet each on risk, then merge.")
print(merged)

Choosing a pattern

Pattern	Best for
Broadcast	One announcement, then replies from many agents in turn
Circular	Sequential refinement with shared history
Mesh	Many independent tasks, worker pool
Grid	One task per agent, laid out as a square
Star	A lead answer on every task, plus independent answers from the rest
Pyramid	One task per agent, laid out in triangular levels
One-to-one	Back-and-forth between two agents
One-to-three	One message reviewed by a fixed panel of three
Aggregate	Fan-in synthesis of parallel outputs

Index

Basic Examples

Model Providers

Single Agent

Tree of Thoughts

Multi-Agent Examples

Auto Agent Builder

Scheduled Agents

Applications

Research

Technical Analysis

Use Cases

Finance

Voice Agents

Integrations

MCP Tool Integrations

Deployment

CLI

Social Swarm Patterns

Choosing a pattern

​Choosing a pattern

​Related

Choosing a pattern

Related