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"

Maximum number of execution loops

bool

default:"False"

Enable agents to know their position in workflow

OutputType

default:"all"

Output format (all, final, list, dict)

Any

default:"None"

Optional memory system for persistence

bool

default:"True"

Log every flow step and agent transition. This defaults to True, so a fresh AgentRearrange is noisy out of the box — pass verbose=False for quiet runs.

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.

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

batch_run()

Process multiple tasks in batches.

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

concurrent_run()

Run multiple tasks concurrently.

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
)

With team awareness, agents receive context like:

“Agent ahead: agent1”
“Agent behind: agent3”
Sequential flow structure information

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
rearrange.remove_agent("old_agent")

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

# Override task for specific agent
custom_tasks = {
    "researcher": "Focus on recent developments",
}

result = rearrange.run(
    task="Main task",
    custom_tasks=custom_tasks,
)

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() will also raise a similar ValueError at execution time if it reaches a step referencing an unregistered agent, so validation happens automatically before any agent work is wasted — just not at construction time.

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