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

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.

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()

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()

​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()

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