Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.swarms.world/llms.txt

Use this file to discover all available pages before exploring further.

​
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.
FunctionTopologyPer-call shape
circular_swarmRing — every agent processes every task in order(agents, tasks)
grid_swarmAgents laid out in a √N×√N grid; one task per cell(agents, tasks)
one_to_oneA → B handoff for max_loops turns(sender, receiver, task)
broadcastOne 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]]
​
agents
AgentListType
required
Flat list of Agent instances (or list-of-lists; the function flattens once).
​
tasks
List[str]
required
Tasks processed sequentially. Each agent runs each task.
​
output_type
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.

​
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]]
​
sender
Agent
required
Agent that processes the task first each loop.
​
receiver
Agent
required
Agent that processes the sender’s output.
​
task
str
required
Task for the sender on the first turn.
​
max_loops
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
from swarms.structs.swarming_architectures import 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
from swarms.structs.swarming_architectures import 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
from swarms.structs.swarming_architectures import 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

PatternWhen to use
circular_swarmEach agent should see every task and the accumulating context
grid_swarmTasks map cleanly to a 2D layout you already have
one_to_oneTwo-agent loops — generator/critic, proposer/judge
broadcastFan-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.