Swarming Architectures

Overview

swarms.structs.swarming_architectures is a set of lightweight functions implementing common message-passing topologies between agents. Each function takes a list of Agent objects, a task or task list, and an output_type, then drives the agents through the topology and returns a formatted conversation history. These are functional building blocks — no shared class, no orchestration object. Reach for them when you want a one-off interaction pattern without instantiating a full swarm class. All seven functions are exported from the top-level swarms package.

Function	Topology	Per-call shape
`circular_swarm`	Ring — every agent processes every task in order	`(agents, tasks)`
`grid_swarm`	Agents laid out in a √N×√N grid; one task per cell	`(agents, tasks)`
`star_swarm`	Star — a central agent processes every task first, then every other agent processes it too	`(agents, tasks)`
`mesh_swarm`	Mesh — agents pull from a shared FIFO task queue until it’s empty	`(agents, tasks)`
`pyramid_swarm`	Pyramid — agents arranged in triangular levels; one task per cell	`(agents, tasks)`
`one_to_one`	A → B handoff for `max_loops` turns	`(sender, receiver, task)`
`broadcast`	One sender → many receivers, async	`(sender, agents, task)`

All functions return whatever shape output_type selects via history_output_formatter — "dict" (default), "list", "str", etc.

Installation

pip install -U swarms

circular_swarm()

Every agent runs every task, in order. Each agent sees the running conversation context.

def circular_swarm(
    agents: AgentListType,
    tasks: List[str],
    output_type: OutputType = "dict",
) -> Union[Dict[str, Any], List[str]]

AgentListType

required

Flat list of Agent instances (or list-of-lists; the function flattens once).

List[str]

required

Tasks processed sequentially. Each agent runs each task.

OutputType

default:"\"dict\""

Format for the returned conversation history.

Raises: ValueError if agents or tasks is empty.

grid_swarm()

Place agents in a √N×√N grid, pop tasks one by one, assign to cells row-by-row.

def grid_swarm(
    agents: AgentListType,
    tasks: List[str],
    output_type: OutputType = "dict",
) -> Union[Dict[str, Any], List[str]]

Best when len(agents) is a perfect square and len(tasks) matches the cell count. Extra tasks are dropped; extra agents are idle.

star_swarm()

agents[0] acts as the central agent: it processes each task first (seeing the running conversation), then every other agent processes the same task independently.

def star_swarm(
    agents: AgentListType,
    tasks: List[str],
    output_type: OutputType = "dict",
) -> Union[Dict[str, Any], List[str]]

Raises: ValueError if agents or tasks is empty.

mesh_swarm()

All tasks are pushed onto a shared FIFO queue. Agents repeatedly loop over the agents list, each popping the next task off the front of the queue, until the queue is empty. (Despite the name, task assignment is deterministic FIFO order, not randomized.)

def mesh_swarm(
    agents: AgentListType,
    tasks: List[str],
    output_type: OutputType = "dict",
) -> Union[Dict[str, Any], List[str]]

Raises: ValueError if agents or tasks is empty.

pyramid_swarm()

Agents are arranged into triangular levels (level i holds i + 1 agents), and tasks are popped one at a time and assigned to each cell in the pyramid, level by level.

def pyramid_swarm(
    agents: AgentListType,
    tasks: List[str],
    output_type: OutputType = "dict",
) -> Union[Dict[str, Any], List[str]]

Best when len(agents) is a triangular number (1, 3, 6, 10, …) and len(tasks) matches the cell count. Extra tasks are dropped; extra agents are idle. Raises: ValueError if agents or tasks is empty.

one_to_one()

Two-agent handoff: sender processes the task, receiver processes the sender’s output, repeat for max_loops.

def one_to_one(
    sender: Agent,
    receiver: Agent,
    task: str,
    max_loops: int = 1,
    output_type: OutputType = "dict",
) -> Union[Dict[str, Any], List[str]]

Agent

required

Agent that processes the task first each loop.

Agent

required

Agent that processes the sender’s output.

str

required

Task for the sender on the first turn.

int

default:"1"

Number of sender→receiver round trips.

broadcast()

Async function. One sender produces a message based on the conversation context; every agent in agents then processes the sender’s broadcast independently.

async def broadcast(
    sender: Agent,
    agents: AgentListType,
    task: str,
    output_type: OutputType = "dict",
) -> Union[Dict[str, Any], List[str]]

Must be awaited:

import asyncio
result = asyncio.run(broadcast(sender, agents, task))

Raises: ValueError if sender, agents, or task is empty.

Usage Examples

Circular: Every Agent Reviews Every Document

from swarms import Agent, circular_swarm

reviewers = [
    Agent(agent_name="Legal", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Security", model_name="claude-sonnet-4-6", max_loops=1),
    Agent(agent_name="Privacy", model_name="claude-sonnet-4-6", max_loops=1),
]

result = circular_swarm(
    agents=reviewers,
    tasks=[
        "Review draft TOS section 4.",
        "Review draft TOS section 5.",
    ],
)

Each reviewer sees the running conversation, so later reviewers can build on earlier comments.

One-to-One: Generator/Critic Loop

from swarms import Agent, one_to_one

generator = Agent(agent_name="Generator", model_name="claude-sonnet-4-6", max_loops=1)
critic = Agent(agent_name="Critic", model_name="claude-opus-4-6", max_loops=1)

result = one_to_one(
    sender=generator,
    receiver=critic,
    task="Draft a one-paragraph announcement for v12.1",
    max_loops=3,
)

Broadcast: Notify a Pool of Workers

import asyncio
from swarms import Agent, broadcast

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

result = asyncio.run(
    broadcast(
        sender=dispatcher,
        agents=workers,
        task="Status check: report your current workload.",
    )
)

Choosing a Topology

Pattern	When to use
`circular_swarm`	Each agent should see every task and the accumulating context
`grid_swarm`	Tasks map cleanly to a 2D layout you already have
`star_swarm`	One agent should weigh in first on every task, then the rest respond independently
`mesh_swarm`	A pool of interchangeable agents should drain a shared task queue
`pyramid_swarm`	Tasks map cleanly to a triangular/hierarchical layout you already have
`one_to_one`	Two-agent loops — generator/critic, proposer/judge
`broadcast`	Fan-out where one message needs to reach many workers

For richer orchestration (planning, voting, hierarchies) see SwarmRouter, HierarchicalSwarm, or MajorityVoting.

Source Code

View the source on GitHub.

Core Classes

Multi-Agent Harnesses

Tools & Utilities

Deployment

Swarming Architectures

Overview

Installation

circular_swarm()

grid_swarm()

star_swarm()

mesh_swarm()

pyramid_swarm()

one_to_one()

broadcast()

Usage Examples

Circular: Every Agent Reviews Every Document

One-to-One: Generator/Critic Loop

Broadcast: Notify a Pool of Workers

Choosing a Topology

Source Code

​Overview

​Installation

​circular_swarm()

​grid_swarm()

​star_swarm()

​mesh_swarm()

​pyramid_swarm()

​one_to_one()

​broadcast()

​Usage Examples

​Circular: Every Agent Reviews Every Document

​One-to-One: Generator/Critic Loop

​Broadcast: Notify a Pool of Workers

​Choosing a Topology

​Source Code

Overview

Installation

circular_swarm()

grid_swarm()

star_swarm()

mesh_swarm()

pyramid_swarm()

one_to_one()

broadcast()

Usage Examples

Circular: Every Agent Reviews Every Document

One-to-One: Generator/Critic Loop

Broadcast: Notify a Pool of Workers

Choosing a Topology

Source Code