> ## Documentation Index
> Fetch the complete documentation index at: https://docs.swarms.world/llms.txt
> Use this file to discover all available pages before exploring further.

# AgentRearrange

> A sophisticated multi-agent system for dynamic task orchestration with custom flow patterns

## 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

```bash theme={null}
pip install -U swarms
```

## Class Definition

```python theme={null}
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,
    )
```

<Note>
  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.
</Note>

## Parameters

<ParamField path="id" type="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>`.
</ParamField>

<ParamField path="name" type="str" default="AgentRearrange">
  Human-readable name for the system
</ParamField>

<ParamField path="description" type="str" default="A swarm of agents for rearranging tasks.">
  Description of the system's purpose
</ParamField>

<ParamField path="agents" type="List[Union[Agent, Callable]]" required>
  List of agents to include in the system. Can be Agent objects or callable functions.
</ParamField>

<ParamField path="flow" type="str" required>
  Flow pattern defining agent execution order. Uses `->` for sequential and `,` for concurrent execution.
  Example: `"agent1 -> agent2, agent3 -> agent4"`
</ParamField>

<ParamField path="max_loops" type="int" default="1">
  Maximum number of execution loops. Must be greater than 0.
</ParamField>

<ParamField path="verbose" type="bool" default="True">
  Whether to enable verbose logging
</ParamField>

<ParamField path="memory_system" type="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.
</ParamField>

<ParamField path="output_type" type="OutputType" default="all">
  Format for output results. Options: "all", "final", "list", "dict"
</ParamField>

<ParamField path="autosave" type="bool" default="True">
  Whether to automatically save execution data
</ParamField>

<ParamField path="team_awareness" type="bool" default="False">
  Whether agents should be aware of team structure and sequential flow
</ParamField>

<ParamField path="collab_prompt" type="str" optional>
  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.
</ParamField>

<ParamField path="time_enabled" type="bool" default="False">
  Whether to track timestamps in conversations
</ParamField>

<ParamField path="message_id_on" type="bool" default="False">
  Whether to include message IDs in conversations
</ParamField>

## 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.

<ParamField path="task" type="str" required>
  The task to execute through the agent workflow
</ParamField>

<ParamField path="img" type="Optional[str]" default="None">
  Path to input image if required by any agents
</ParamField>

<ResponseField name="result" type="Union[str, List[str], Dict[str, str]]">
  The processed output in the format specified by output\_type
</ResponseField>

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

Process multiple tasks in batches.

<ParamField path="tasks" type="List[str]" required>
  List of tasks to process through the agent workflow
</ParamField>

<ParamField path="img" type="Optional[List[str]]" default="None">
  Optional list of images corresponding to tasks
</ParamField>

<ParamField path="batch_size" type="int" default="10">
  Number of tasks to process simultaneously in each batch
</ParamField>

<ResponseField name="results" type="List[str]">
  List of results corresponding to input tasks
</ResponseField>

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

Process multiple tasks concurrently using ThreadPoolExecutor.

<ParamField path="tasks" type="List[str]" required>
  List of tasks to process through the agent workflow
</ParamField>

<ParamField path="img" type="Optional[List[str]]" default="None">
  Optional list of images corresponding to tasks
</ParamField>

<ParamField path="max_workers" type="Optional[int]" default="None">
  Maximum number of worker threads. Uses default ThreadPoolExecutor behavior if None.
</ParamField>

<ResponseField name="results" type="List[str]">
  List of results corresponding to input tasks
</ResponseField>

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

Asynchronously execute a task.

<ParamField path="task" type="str" required>
  The task to be executed through the agent workflow
</ParamField>

<ParamField path="img" type="Optional[str]" default="None">
  Optional image input for the task
</ParamField>

<ResponseField name="result" type="Any">
  The result of the task execution
</ResponseField>

### `set_custom_flow(flow)`

Set a custom flow pattern for agent execution.

<ParamField path="flow" type="str" required>
  The new flow pattern to use for agent execution
</ParamField>

### `add_agent(agent)`

Add an agent to the swarm.

<ParamField path="agent" type="Agent" required>
  The agent to be added
</ParamField>

### `add_agents(agents)`

Add multiple agents to the swarm at once.

<ParamField path="agents" type="List[Agent]" required>
  A list of Agent objects to be added
</ParamField>

### `remove_agent(agent_name)`

Remove an agent from the swarm.

<ParamField path="agent_name" type="str" required>
  The name of the agent to be removed
</ParamField>

### `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.

<ParamField path="return_str" type="bool" default="False">
  When `True`, returns the plan as a string instead of printing it
</ParamField>

<ResponseField name="plan" type="Optional[str]">
  The plan string if `return_str=True`; otherwise `None`
</ResponseField>

### `get_agent_sequential_awareness(agent_name)`

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

<ParamField path="agent_name" type="str" required>
  The name of the agent to get awareness for
</ParamField>

<ResponseField name="awareness" type="str">
  A string describing the agents ahead and behind in the sequence
</ResponseField>

### `get_sequential_flow_structure()`

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

<ResponseField name="structure" type="str">
  A string describing the overall sequential flow structure
</ResponseField>

### `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.

<ParamField path="task" type="str" default="None">
  Initial task fed into the first agent in the flow
</ParamField>

<ParamField path="img" type="Optional[str]" default="None">
  Optional image input forwarded to every agent
</ParamField>

<ParamField path="with_events" type="bool" default="False">
  When `False`, yields `(agent_name, token)` tuples. When `True`, yields structured event dicts (`agent_start`, `token`, `agent_end`)
</ParamField>

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

### `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.

<ResponseField name="is_valid" type="bool">
  True if the flow pattern is valid
</ResponseField>

**Raises:**

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

### `to_dict()`

Convert all attributes to a dictionary for serialization.

<ResponseField name="dict_representation" type="Dict[str, Any]">
  Dictionary representation of all class attributes
</ResponseField>

## 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]] | List of agents or callables in the system   |
| `flow`         | str                            | Flow pattern defining agent execution order |
| `conversation` | Conversation                   | Conversation history management             |

## Usage Examples

### Sequential Flow

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# 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

```python theme={null}
# 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

```python theme={null}
# 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:

```python theme={null}
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

```python theme={null}
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

1. **Flow Design**: Carefully design your flow to match your task requirements
2. **Team Awareness**: Enable for complex flows where context matters
3. **Error Handling**: Always wrap execution in try-except blocks
4. **Logging**: Enable verbose mode during development
5. **Memory Systems**: Use for tasks requiring persistent context
6. **Flow Validation**: Always validate flows before production use
7. **Agent Naming**: Use clear, descriptive agent names in flows

## Common Flow Patterns

### Fan-Out Pattern

```python theme={null}
# One agent feeds multiple agents
flow = "collector -> analyst1, analyst2, analyst3 -> synthesizer"
```

### Pipeline Pattern

```python theme={null}
# Linear processing chain
flow = "stage1 -> stage2 -> stage3 -> stage4"
```

### Review Pattern

```python theme={null}
# Create, review, revise
flow = "creator -> reviewer -> reviser"
```

### Ensemble Pattern

```python theme={null}
# Multiple independent analyses
flow = "agent1, agent2, agent3, agent4 -> aggregator"
```

## Related Classes

* [SequentialWorkflow](/api/sequential-workflow): For simple sequential execution
* [ConcurrentWorkflow](/api/concurrent-workflow): For parallel agent execution
* [GraphWorkflow](/api/graph-workflow): For complex DAG-based workflows
* [Agent](/api/agent): The base agent class used in workflows