Agent Rearrange

The AgentRearrange system enables sophisticated multi-agent orchestration through custom flow patterns. Define how agents communicate using simple syntax: -> for sequential execution and , for concurrent execution.

When to Use

Flexible workflows: Mix sequential and parallel execution
Dynamic routing: Tasks need different paths through agents
Complex coordination: Multiple agents with custom relationships
Adaptive workflows: Flow changes based on task requirements
Team awareness: Agents need context about team structure

Flow Syntax

agent1 -> agent2: Sequential execution (agent2 runs after agent1)
agent1, agent2: Concurrent execution (both run simultaneously)
agent1 -> agent2, agent3: Combined (agent1 first, then agent2 and agent3 in parallel)

Basic Example

from swarms import Agent, AgentRearrange

# Define specialized agents
researcher = Agent(
    agent_name="researcher",
    system_prompt="Research topics and gather information.",
    model_name="gpt-5.4",
)

writer = Agent(
    agent_name="writer",
    system_prompt="Write engaging content based on research.",
    model_name="gpt-5.4",
)

reviewer = Agent(
    agent_name="reviewer",
    system_prompt="Review and provide feedback on content.",
    model_name="gpt-5.4",
)

# Define flow: researcher first, then writer and reviewer in parallel
flow = "researcher -> writer, reviewer"

# Create the system
rearrange = AgentRearrange(
    agents=[researcher, writer, reviewer],
    flow=flow,
    max_loops=1,
)

# Execute
result = rearrange.run("Analyze quantum computing trends")
print(result)

Complex Flow Patterns

Fan-Out Pattern

One agent distributes to multiple agents:

# Data collector sends to three analysts simultaneously
flow = "data_collector -> technical_analyst, fundamental_analyst, sentiment_analyst"

rearrange = AgentRearrange(
    agents=[data_collector, technical_analyst, fundamental_analyst, sentiment_analyst],
    flow=flow,
)

Fan-In Pattern

Multiple agents converge to one:

# Multiple researchers feed into synthesizer
flow = "researcher1, researcher2, researcher3 -> synthesizer"

rearrange = AgentRearrange(
    agents=[researcher1, researcher2, researcher3, synthesizer],
    flow=flow,
)

Multi-Stage Pipeline

# Research → parallel analysis → synthesis
flow = "researcher -> analyst1, analyst2, analyst3 -> synthesizer"

rearrange = AgentRearrange(
    agents=[researcher, analyst1, analyst2, analyst3, synthesizer],
    flow=flow,
)

Key Parameters

str

default:"AgentRearrange"

Name for the agent rearrange system

List[Agent]

required

List of agents to orchestrate

str

required

Flow pattern defining agent execution (e.g., “agent1 -> agent2, agent3”)

int

default:"1"

Number of times the whole flow runs per run() call. Values below 1 are treated as 1.

bool

default:"False"

Add the step-by-step flow structure to the shared conversation at the start of each run

OutputType

default:"all"

Output format, for example "all", "final", "list", "dict" or "str-all-except-first". An unknown value raises ValueError at construction.

Any

default:"None"

Accepted for backwards compatibility and stored, but never read. Use persistent_memory=True on individual agents instead.

bool

default:"False"

Log every flow step and agent transition. Pass verbose=True while debugging a flow.

bool

default:"True"

Persist workflow state. Defaults to True.

bool

default:"False"

Record an ISO timestamp on every conversation message.

bool

default:"False"

Attach a unique id to every conversation message.

Methods

run()

Execute the defined flow with a task. Each call starts a fresh shared conversation, so a reused instance does not carry the previous task’s transcript.

result = rearrange.run(
    task="Analyze market trends",
    img=None,  # Optional image input
)

batch_run()

Process multiple tasks in batches. Batches run one after another, the tasks inside a batch run concurrently, and each task runs on its own clone of the flow. Results come back in input order.

tasks = ["Task 1", "Task 2", "Task 3"]
results = rearrange.batch_run(
    tasks=tasks,
    batch_size=10,
)

concurrent_run()

Run multiple tasks concurrently. Each task runs on its own clone of the flow, so tasks never share a conversation. Results come back in input order.

tasks = ["Task 1", "Task 2", "Task 3"]
results = rearrange.concurrent_run(
    tasks=tasks,
    max_workers=5,
)

run_async()

Asynchronous task execution.

import asyncio

async def main():
    result = await rearrange.run_async("Task description")
    return result

result = asyncio.run(main())

run_stream() / arun_stream()

Stream tokens as agents execute, in flow order. Sequential segments (A -> B) stream one agent at a time; parallel segments (A, B) interleave tokens from concurrent agents fairly.

# Sync generator
for agent_name, token in rearrange.run_stream("Analyse quantum computing trends"):
    print(f"[{agent_name}] {token}", end="", flush=True)

# Async generator
import asyncio

async def main():
    async for agent_name, token in rearrange.arun_stream("Analyse quantum computing trends"):
        print(f"[{agent_name}] {token}", end="", flush=True)

asyncio.run(main())

Pass with_events=True to receive structured agent_start / token / agent_end event dicts instead of (agent_name, token) tuples.

max_loops > 1 and custom_tasks are not supported in streaming mode. Use run() for those.

explain()

Print or return the resolved execution plan for the current flow. It validates the flow, then lists every step in order and marks each as sequential or parallel. No agents or LLMs are invoked, which makes it cheap enough for CI smoke tests and pre-flight checks.

bool

default:"False"

When True, return the plan as a string. When False, print it and return None.

Optional[str]

The plan string when return_str=True; otherwise None.

rearrange = AgentRearrange(
    agents=[ingestor, tech, business, legal, synthesizer],
    flow="Ingestor -> Tech, Business, Legal -> Synthesizer",
)

rearrange.explain()
# Flow: Ingestor -> Tech, Business, Legal -> Synthesizer
#
# Step 1: Ingestor               [sequential]
# Step 2: Tech, Business, Legal  [parallel, 3 agents]
# Step 3: Synthesizer            [sequential]
#
# 3 steps, 5 agent invocations across 1 loop(s).

plan = rearrange.explain(return_str=True)

explain() validates the flow first and raises if it is invalid — the same error run() would raise.

Team Awareness

Enable agents to understand their position in the workflow:

rearrange = AgentRearrange(
    agents=agents,
    flow="agent1 -> agent2 -> agent3",
    team_awareness=True,  # Agents know who comes before/after
)

Every agent in a single-agent step is told its neighbours as a system turn, for example "Sequential awareness: Agent ahead: agent1 | Agent behind: agent3", whether or not team_awareness is set. With team_awareness=True, the shared conversation also starts with the full sequential flow structure, so every agent sees the whole pipeline.

Use Cases

Content Creation Pipeline

# Research → Write → (Edit, Fact-Check) → Publish
flow = "researcher -> writer -> editor, fact_checker -> publisher"

pipeline = AgentRearrange(
    agents=[researcher, writer, editor, fact_checker, publisher],
    flow=flow,
)

article = pipeline.run("AI in healthcare")

Software Development

# Design → (Frontend, Backend) → Testing → Review
flow = "architect -> frontend_dev, backend_dev -> tester -> reviewer"

dev_pipeline = AgentRearrange(
    agents=[architect, frontend_dev, backend_dev, tester, reviewer],
    flow=flow,
)

code = dev_pipeline.run("Build user authentication system")

Market Analysis

# Data Collection → (Technical, Fundamental, Sentiment) → Synthesis
flow = "collector -> tech_analyst, fund_analyst, sent_analyst -> synthesizer"

analysis = AgentRearrange(
    agents=[collector, tech_analyst, fund_analyst, sent_analyst, synthesizer],
    flow=flow,
)

report = analysis.run("NVIDIA stock analysis")

Dynamic Flow Management

Change Flow at Runtime

rearrange = AgentRearrange(agents=agents, flow="agent1 -> agent2")

# Update flow dynamically
rearrange.set_custom_flow("agent1 -> agent2, agent3")

result = rearrange.run("New task")

Add/Remove Agents

# Add new agent
new_agent = Agent(agent_name="new_agent", ...)
rearrange.add_agent(new_agent)

# Remove agent; raises ValueError if no agent has that name
rearrange.remove_agent("old_agent")

agents is a plain list. Look an agent up by agent_name rather than indexing it with a name, and update the flow with set_custom_flow() after removing an agent it references.

Sequential Awareness

Agents can understand their workflow position:

# Get awareness info for specific agent
awareness = rearrange.get_agent_sequential_awareness("agent2")
print(awareness)
# Output: "Sequential awareness: Agent ahead: agent1 | Agent behind: agent3"

# Get full flow structure
structure = rearrange.get_sequential_flow_structure()
print(structure)
# Output:
# Sequential Flow Structure:
# Step 1: agent1 (leads to: agent2)
# Step 2: agent2 (follows: agent1) (leads to: agent3)
# Step 3: agent3 (follows: agent2)

Advanced Features

Custom tasks for specific agents

run() accepts a custom_tasks dict, but the current implementation splices the custom task text into the flow as if it were an agent name. A run with custom_tasks raises ValueError: either the agent name is not found in the flow, or the spliced text is “not registered” when its step is reached. Put per-agent instructions in each agent’s system_prompt instead.

Output Formatting

# Different output types
rearrange_all = AgentRearrange(
    agents=agents,
    flow=flow,
    output_type="all",  # All agent responses
)

rearrange_final = AgentRearrange(
    agents=agents,
    flow=flow,
    output_type="final",  # Only final agent's response
)

rearrange_list = AgentRearrange(
    agents=agents,
    flow=flow,
    output_type="list",  # List of responses
)

Best Practices

Flow Design: Start simple and add complexity as needed. Test with “agent1 -> agent2” before complex patterns.

Clear Flow Logic: Ensure flow makes sense for your task
Agent Naming: Use descriptive names for clarity in flow definitions
Validate Flow: Use validate_flow() before production
Team Awareness: Enable when agents benefit from position context
Start Simple: Begin with sequential, add concurrency where beneficial

Flow validation happens at runtime - ensure all agent names in flow exist in the agents list

Flow Validation

Construction only checks that flow is a non-empty string — it does not verify that every agent name in the flow is registered. Call validate_flow() explicitly (or explain(), which calls it internally) to catch typos before running:

rearrange = AgentRearrange(
    agents=[agent1, agent2],
    flow="agent1 -> agent3",  # agent3 doesn't exist
)

try:
    rearrange.validate_flow()
except ValueError as e:
    print(f"Invalid flow: {e}")
    # Output: "Agent 'agent3' is not registered."

run() does not validate the whole flow first. It raises a similar ValueError only when it reaches the step that references the unregistered agent, so every earlier step has already run. Validate up front to avoid wasting that work. set_custom_flow() validates the new flow immediately and keeps the old one if the new one is invalid.

Sequential Workflow - Simpler linear flows
Concurrent Workflow - Pure parallel execution
Graph Workflow - DAG-based complex flows
Social Algorithms - Custom communication patterns

Getting Started

Core Concepts

Agent Development

Reasoning Agents

Multi-Agent Architectures

Deployment & Operations

Integrations

CLI Reference

Changelog

Agent Rearrange

When to Use

Flow Syntax

Basic Example

Complex Flow Patterns

Fan-Out Pattern

Fan-In Pattern

Multi-Stage Pipeline

Key Parameters

Methods

run()

batch_run()

concurrent_run()

run_async()

run_stream() / arun_stream()

explain()

Team Awareness

Use Cases

Content Creation Pipeline

Software Development

Market Analysis

Dynamic Flow Management

Change Flow at Runtime

Add/Remove Agents

Sequential Awareness

Advanced Features

Custom tasks for specific agents

Output Formatting

Best Practices

Flow Validation

​When to Use

​Flow Syntax

​Basic Example

​Complex Flow Patterns

​Fan-Out Pattern

​Fan-In Pattern

​Multi-Stage Pipeline

​Key Parameters

​Methods

​run()

​batch_run()

​concurrent_run()

​run_async()

​run_stream() / arun_stream()

​explain()

​Team Awareness

​Use Cases

​Content Creation Pipeline

​Software Development

​Market Analysis

​Dynamic Flow Management

​Change Flow at Runtime

​Add/Remove Agents

​Sequential Awareness

​Advanced Features

​Custom tasks for specific agents

​Output Formatting

​Best Practices

​Flow Validation

​Related Architectures

When to Use

Flow Syntax

Basic Example

Complex Flow Patterns

Fan-Out Pattern

Fan-In Pattern

Multi-Stage Pipeline

Key Parameters

Methods

run()

batch_run()

concurrent_run()

run_async()

run_stream() / arun_stream()

explain()

Team Awareness

Use Cases

Content Creation Pipeline

Software Development

Market Analysis

Dynamic Flow Management

Change Flow at Runtime

Add/Remove Agents

Sequential Awareness

Advanced Features

Custom tasks for specific agents

Output Formatting

Best Practices

Flow Validation

Related Architectures