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
Memory System Support: Optional persistent memory across agent interactions
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 = swarm_id(),
        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 = True,
        memory_system: Any = None,
        output_type: OutputType = "all",
        autosave: bool = True,
        team_awareness: bool = False,
        time_enabled: bool = False,
        message_id_on: bool = False,
    )

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 if not provided.

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"

Maximum number of execution loops. Must be greater than 0.

bool

default:"True"

Whether to enable verbose logging

Any

default:"None"

Optional memory system for persistence across interactions

OutputType

default:"all"

Format for output results. Options: “all”, “final”, “list”, “dict”

bool

default:"True"

Whether to automatically save execution data

bool

default:"False"

Whether agents should be aware of team structure and sequential flow

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.

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.

List[str]

required

List of tasks to process through the agent workflow

Optional[List[str]]

default:"None"

Optional list of images corresponding to tasks

int

default:"10"

Number of tasks to process simultaneously in each batch

List[str]

List of results corresponding to input tasks

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

Process multiple tasks concurrently using ThreadPoolExecutor.

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. Uses default ThreadPoolExecutor behavior if None.

List[str]

List of results corresponding to 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.

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

`remove_agent(agent_name)`

Remove an agent from the swarm.

str

required

The name of the agent to be removed

`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`	Dict[str, Agent]	Dictionary mapping agent names to Agent objects
`flow`	str	Flow pattern defining agent execution order
`conversation`	Conversation	Conversation history management

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",
    llm=llm,
    system_prompt="You collect and organize data."
)

technical_analyst = Agent(
    agent_name="technical_analyst",
    llm=llm,
    system_prompt="You analyze technical aspects."
)

business_analyst = Agent(
    agent_name="business_analyst",
    llm=llm,
    system_prompt="You analyze business aspects."
)

synthesizer = Agent(
    agent_name="synthesizer",
    llm=llm,
    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 Memory System

from swarms_memory import ChromaDB

# Create memory system
memory = ChromaDB(
    output_dir="./agent_memory",
    n_results=5
)

system = AgentRearrange(
    agents=[researcher, writer],
    flow="researcher -> writer",
    memory_system=memory  # Persist conversation across runs
)

result = system.run("Analyze AI trends")

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"  # Return all agent outputs
)

# 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"  # Return list of outputs
)

# Return as dict
system_dict = AgentRearrange(
    agents=[researcher, writer],
    flow="researcher -> writer",
    output_type="dict"  # Return dict mapping agent names to outputs
)

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 Systems: Use for tasks requiring persistent context
Flow Validation: Always validate flows before production use
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)

​remove_agent(agent_name)

​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 Memory System

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

`remove_agent(agent_name)`

`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 Memory System

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