AgentRearrange

Overview

The AgentRearrange class enables complex workflows where multiple agents can work sequentially or concurrently based on a defined flow pattern. It supports both sequential execution (using ->) and concurrent execution (using ,) within the same workflow, providing maximum flexibility for agent orchestration.

Key Features

Flexible Flow Syntax: Define sequential (->) and concurrent (,) agent execution in one flow
Custom Flow Patterns: Mix sequential and concurrent execution patterns
Team Awareness: Agents can be aware of their position in the workflow
Batch Processing: Process multiple tasks with the same flow
Concurrent Execution: Run multiple tasks in parallel
Async Support: Asynchronous execution for non-blocking operations

Installation

pip install -U swarms

Class Definition

class AgentRearrange:
    def __init__(
        self,
        id: str = None,
        name: str = "AgentRearrange",
        description: str = "A swarm of agents for rearranging tasks.",
        agents: List[Union[Agent, Callable]] = None,
        flow: str = None,
        max_loops: int = 1,
        verbose: bool = False,
        memory_system: Any = None,
        output_type: OutputType = "all",
        autosave: bool = True,
        team_awareness: bool = False,
        time_enabled: bool = False,
        message_id_on: bool = False,
        collab_prompt: Optional[str] = None,
    )

Earlier versions accepted human_in_the_loop, custom_human_in_the_loop, and rules parameters (and an H token in the flow string for human review steps). These have been removed — AgentRearrange no longer supports inline human-in-the-loop steps or a rules argument.

Parameters

str

default:"auto-generated"

Unique identifier for the agent rearrange system. Auto-generated via generate_id("agent-rearrange") if not provided, producing agent-rearrange-<32 hex chars>.

str

default:"AgentRearrange"

Human-readable name for the system

str

default:"A swarm of agents for rearranging tasks."

Description of the system’s purpose

List[Union[Agent, Callable]]

required

List of agents to include in the system. Can be Agent objects or callable functions.

str

required

Flow pattern defining agent execution order. Uses -> for sequential and , for concurrent execution. Example: "agent1 -> agent2, agent3 -> agent4"

int

default:"1"

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

bool

default:"False"

Whether to enable verbose logging

Any

default:"None"

Accepted for backwards compatibility. The value is stored on the instance and is not read by the workflow. Configure memory on the individual agents with persistent_memory instead.

OutputType

default:"all"

Format of the returned conversation. One of "list", "dict", "dictionary", "str", "string", "all", "final", "last", "json", "yaml", "dict-all-except-first", "str-all-except-first", "dict-final", "list-final". The default "all" returns the transcript as "role: content" text. Any other value raises ValueError at construction.

bool

default:"True"

Whether to automatically save execution data

bool

default:"False"

If True, the step-by-step flow structure (see get_sequential_flow_structure()) is added to the shared conversation as a system message at the start of each run. Agents in single-agent steps always receive the names of the agents ahead and behind them as a system turn, whether or not this is set.

str

Guidance prepended to every agent’s messages as a system turn for the duration of the run. It is not written to the shared conversation, so other agents never see it, and the caller’s Agent objects are not modified. SequentialWorkflow passes its collaboration preamble through this parameter.

bool

default:"False"

Whether to track timestamps in conversations

bool

default:"False"

Whether to include message IDs in conversations

Flow Syntax

The flow pattern defines how agents execute:

Sequential: agent1 -> agent2 -> agent3 (agents run one after another)
Concurrent: agent1, agent2, agent3 (agents run simultaneously)
Mixed: agent1 -> agent2, agent3 -> agent4 (agent1 first, then agent2 and agent3 concurrently, then agent4)

Methods

`run(task, img=None, *args, **kwargs)`

Execute the agent rearrangement task. Each call starts a fresh shared conversation, so a reused instance does not carry the previous task’s transcript. Every agent’s final answer, not its whole transcript, is added to that conversation. The flow is not validated up front: a step that names an unregistered agent raises ValueError when it is reached, after earlier steps have run.

str

required

The task to execute through the agent workflow

Optional[str]

default:"None"

Path to input image if required by any agents

Union[str, List[str], Dict[str, str]]

The processed output in the format specified by output_type

`batch_run(tasks, img=None, batch_size=10, *args, **kwargs)`

Process multiple tasks in batches. Batches run one after another; the tasks inside a batch run concurrently. Each task runs on an isolated clone with a fresh conversation and deep-copied agents (falling back to the original agent when it cannot be copied).

List[str]

required

List of tasks to process through the agent workflow

Optional[List[str]]

default:"None"

Optional list of images corresponding to tasks. Must be the same length as tasks, or ValueError is raised.

int

default:"10"

Number of tasks to process simultaneously in each batch. Must be positive, or ValueError is raised.

List[str]

List of results in the same order as the input tasks

`concurrent_run(tasks, img=None, max_workers=None, *args, **kwargs)`

Process multiple tasks concurrently on a thread pool. Each task runs on an isolated clone with a fresh conversation, so concurrent tasks never share history.

List[str]

required

List of tasks to process through the agent workflow

Optional[List[str]]

default:"None"

Optional list of images corresponding to tasks

Optional[int]

default:"None"

Maximum number of worker threads. Defaults to os.cpu_count() when None.

List[str]

List of results in the same order as the input tasks

`run_async(task, img=None, *args, **kwargs)`

Asynchronously execute a task.

str

required

The task to be executed through the agent workflow

Optional[str]

default:"None"

Optional image input for the task

Any

The result of the task execution

`set_custom_flow(flow)`

Set a custom flow pattern for agent execution. The new flow is validated immediately; if it is empty or names an unregistered agent, ValueError is raised and the previous flow is kept.

str

required

The new flow pattern to use for agent execution

`add_agent(agent)`

Add an agent to the swarm.

Agent

required

The agent to be added

`add_agents(agents)`

Add multiple agents to the swarm at once.

List[Agent]

required

A list of Agent objects to be added

`remove_agent(agent_name)`

Remove an agent from the swarm. Raises ValueError if no agent has that name.

str

required

The name of the agent to be removed

`explain(return_str=False)`

Print (or return) the resolved execution plan for the current flow, listing every step in order and marking each as sequential or parallel. Does not invoke any agents or LLMs. Validates the flow first and raises if it is invalid.

bool

default:"False"

When True, returns the plan as a string instead of printing it

Optional[str]

The plan string if return_str=True; otherwise None

`get_agent_sequential_awareness(agent_name)`

Get the sequential awareness information (agents immediately ahead/behind) for a specific agent in the current flow.

str

required

The name of the agent to get awareness for

str

A string describing the agents ahead and behind in the sequence

`get_sequential_flow_structure()`

Get a string describing the complete sequential flow structure (step-by-step breakdown of the flow).

str

A string describing the overall sequential flow structure

`run_stream(task=None, img=None, with_events=False, **kwargs)`

Sync generator that streams tokens from each agent in flow order as they are generated. Sequential segments stream one agent at a time; parallel (comma) segments interleave tokens fairly across concurrent agents.

str

default:"None"

Initial task fed into the first agent in the flow

Optional[str]

default:"None"

Optional image input forwarded to every agent

bool

default:"False"

When False, yields (agent_name, token) tuples. When True, yields structured event dicts (agent_start, token, agent_end)

Not yet supported in streaming mode: max_loops > 1, custom_tasks. Use run() for those.

`arun_stream(task=None, img=None, with_events=False, **kwargs)`

Async generator version of run_stream, with the same parameters and yield semantics.

`validate_flow()`

Validate the flow pattern.

bool

True if the flow pattern is valid

Raises:

ValueError: If the flow pattern is incorrectly formatted or contains unregistered agents

`to_dict()`

Convert all attributes to a dictionary for serialization.

Dict[str, Any]

Dictionary representation of all class attributes

Attributes

Attribute	Type	Description
`id`	str	Unique identifier for the system
`name`	str	Human-readable name
`description`	str	Description of the system’s purpose
`agents`	List[Union[Agent, Callable]]	The agents in the system, as a list. Look agents up by `agent_name`, not by indexing with a name
`flow`	str	Flow pattern defining agent execution order
`steps`	List[List[str]]	Read-only property: the parsed flow, one list of agent names per `->` step. `"a -> b, c"` gives `[["a"], ["b", "c"]]`
`max_loops`	int	Number of times the flow runs per call
`output_type`	OutputType	Format of the returned conversation
`team_awareness`	bool	Whether the flow structure is added to the conversation
`collab_prompt`	Optional[str]	Guidance prepended to every agent’s messages
`conversation`	Conversation	The latest run’s shared conversation; replaced at the start of every run

Usage Examples

Sequential Flow

from swarms import Agent, AgentRearrange

# Create agents
researcher = Agent(
    agent_name="researcher",
    model_name="claude-sonnet-4-6",
    system_prompt="You are a research specialist."
)

writer = Agent(
    agent_name="writer",
    model_name="claude-sonnet-4-6",
    system_prompt="You are a content writer."
)

editor = Agent(
    agent_name="editor",
    model_name="claude-sonnet-4-6",
    system_prompt="You are an editor."
)

# Define sequential flow
flow = "researcher -> writer -> editor"

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

# Execute task
result = system.run("Write a blog post about AI")
print(result)

Concurrent Flow

# Define concurrent flow - all agents run simultaneously
flow = "researcher, writer, editor"

system = AgentRearrange(
    agents=[researcher, writer, editor],
    flow=flow
)

result = system.run("Analyze this topic from different angles")

Mixed Sequential and Concurrent Flow

# Create more agents
data_collector = Agent(
    agent_name="data_collector",
    model_name="gpt-5.4-mini",
    system_prompt="You collect and organize data."
)

technical_analyst = Agent(
    agent_name="technical_analyst",
    model_name="gpt-5.4-mini",
    system_prompt="You analyze technical aspects."
)

business_analyst = Agent(
    agent_name="business_analyst",
    model_name="gpt-5.4-mini",
    system_prompt="You analyze business aspects."
)

synthesizer = Agent(
    agent_name="synthesizer",
    model_name="gpt-5.4",
    system_prompt="You synthesize multiple perspectives."
)

# Mixed flow: collect data, then analyze concurrently, then synthesize
flow = "data_collector -> technical_analyst, business_analyst -> synthesizer"

system = AgentRearrange(
    agents=[data_collector, technical_analyst, business_analyst, synthesizer],
    flow=flow,
    team_awareness=True  # Agents know about each other
)

result = system.run("Analyze the market opportunity for AI assistants")

Batch Processing

# Process multiple tasks through the same flow
tasks = [
    "Analyze healthcare AI trends",
    "Analyze education AI trends",
    "Analyze finance AI trends"
]

system = AgentRearrange(
    agents=[researcher, writer, editor],
    flow="researcher -> writer -> editor"
)

results = system.batch_run(tasks, batch_size=2)

for task, result in zip(tasks, results):
    print(f"Task: {task}")
    print(f"Result: {result}")
    print("-" * 80)

Concurrent Task Execution

tasks = [
    "Research AI in healthcare",
    "Research AI in education",
    "Research AI in finance"
]

# Run all tasks in parallel
results = system.concurrent_run(tasks, max_workers=3)
print(f"Processed {len(results)} tasks concurrently")

Async Execution

import asyncio

async def process_async():
    result = await system.run_async("Analyze quantum computing trends")
    return result

result = asyncio.run(process_async())

Dynamic Flow Modification

# Create system with initial flow
system = AgentRearrange(
    agents=[researcher, writer, editor],
    flow="researcher -> writer -> editor"
)

# Run with initial flow
result1 = system.run("Task 1")

# Change flow dynamically
system.set_custom_flow("researcher -> editor -> writer")

# Run with new flow
result2 = system.run("Task 2")

With Team Awareness

# Agents will know about their position in the workflow
system = AgentRearrange(
    agents=[researcher, writer, editor],
    flow="researcher -> writer -> editor",
    team_awareness=True,  # Enable team awareness
    verbose=True
)

result = system.run("Create comprehensive analysis")
# Each agent will receive information about agents ahead and behind

Different Output Types

# Return all outputs
system_all = AgentRearrange(
    agents=[researcher, writer],
    flow="researcher -> writer",
    output_type="all"  # Whole transcript as "role: content" text
)

# Return only final output
system_final = AgentRearrange(
    agents=[researcher, writer],
    flow="researcher -> writer",
    output_type="final"  # Return only last agent's output
)

# Return as list
system_list = AgentRearrange(
    agents=[researcher, writer],
    flow="researcher -> writer",
    output_type="list"  # List of {"role", "content"} message dicts
)

# Agents' answers only, without the task
system_answers = AgentRearrange(
    agents=[researcher, writer],
    flow="researcher -> writer",
    output_type="str-all-except-first"
)

Convenience Function

The rearrange() function provides a quick way to create and execute:

from swarms import rearrange

result = rearrange(
    name="Quick Analysis",
    agents=[researcher, writer, editor],
    flow="researcher -> writer -> editor",
    task="Analyze AI trends in 2024"
)

Error Handling

try:
    system = AgentRearrange(
        agents=[researcher, writer],
        flow="researcher -> writer"
    )
    result = system.run("Process this task")
except ValueError as e:
    print(f"Configuration error: {e}")
except Exception as e:
    print(f"Execution error: {e}")

Best Practices

Flow Design: Carefully design your flow to match your task requirements
Team Awareness: Enable for complex flows where context matters
Error Handling: Always wrap execution in try-except blocks
Logging: Enable verbose mode during development
Memory: memory_system is not read; set persistent_memory=True on individual agents for context that survives restarts
Flow Validation: Call validate_flow() or explain() before production use, since construction does not check agent names
Agent Naming: Use clear, descriptive agent names in flows

Common Flow Patterns

Fan-Out Pattern

# One agent feeds multiple agents
flow = "collector -> analyst1, analyst2, analyst3 -> synthesizer"

Pipeline Pattern

# Linear processing chain
flow = "stage1 -> stage2 -> stage3 -> stage4"

Review Pattern

# Create, review, revise
flow = "creator -> reviewer -> reviser"

Ensemble Pattern

# Multiple independent analyses
flow = "agent1, agent2, agent3, agent4 -> aggregator"

SequentialWorkflow: For simple sequential execution
ConcurrentWorkflow: For parallel agent execution
GraphWorkflow: For complex DAG-based workflows
Agent: The base agent class used in workflows

​Overview

​Key Features

​Installation

​Class Definition

​Parameters

​Flow Syntax

​Methods

​run(task, img=None, *args, **kwargs)

​batch_run(tasks, img=None, batch_size=10, *args, **kwargs)

​concurrent_run(tasks, img=None, max_workers=None, *args, **kwargs)

​run_async(task, img=None, *args, **kwargs)

​set_custom_flow(flow)

​add_agent(agent)

​add_agents(agents)

​remove_agent(agent_name)

​explain(return_str=False)

​get_agent_sequential_awareness(agent_name)

​get_sequential_flow_structure()

​run_stream(task=None, img=None, with_events=False, **kwargs)

​arun_stream(task=None, img=None, with_events=False, **kwargs)

​validate_flow()

​to_dict()

​Attributes

​Usage Examples

​Sequential Flow

​Concurrent Flow

​Mixed Sequential and Concurrent Flow

​Batch Processing

​Concurrent Task Execution

​Async Execution

​Dynamic Flow Modification

​With Team Awareness

​Different Output Types

​Convenience Function

​Error Handling

​Best Practices

​Common Flow Patterns

​Fan-Out Pattern

​Pipeline Pattern

​Review Pattern

​Ensemble Pattern

​Related Classes

Overview

Key Features

Installation

Class Definition

Parameters

Flow Syntax

Methods

`run(task, img=None, *args, **kwargs)`

`batch_run(tasks, img=None, batch_size=10, *args, **kwargs)`

`concurrent_run(tasks, img=None, max_workers=None, *args, **kwargs)`

`run_async(task, img=None, *args, **kwargs)`

`set_custom_flow(flow)`

`add_agent(agent)`

`add_agents(agents)`

`remove_agent(agent_name)`

`explain(return_str=False)`

`get_agent_sequential_awareness(agent_name)`

`get_sequential_flow_structure()`

`run_stream(task=None, img=None, with_events=False, **kwargs)`

`arun_stream(task=None, img=None, with_events=False, **kwargs)`

`validate_flow()`

`to_dict()`

Attributes

Usage Examples

Sequential Flow

Concurrent Flow

Mixed Sequential and Concurrent Flow

Batch Processing

Concurrent Task Execution

Async Execution

Dynamic Flow Modification

With Team Awareness

Different Output Types

Convenience Function

Error Handling

Best Practices

Common Flow Patterns

Fan-Out Pattern

Pipeline Pattern

Review Pattern

Ensemble Pattern

Related Classes