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

# Agent Rearrange

> Define complex multi-agent workflows with custom flow patterns using arrow and comma syntax

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

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

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

```python theme={null}
# Multiple researchers feed into synthesizer
flow = "researcher1, researcher2, researcher3 -> synthesizer"

rearrange = AgentRearrange(
    agents=[researcher1, researcher2, researcher3, synthesizer],
    flow=flow,
)
```

### Multi-Stage Pipeline

```python theme={null}
# Research → parallel analysis → synthesis
flow = "researcher -> analyst1, analyst2, analyst3 -> synthesizer"

rearrange = AgentRearrange(
    agents=[researcher, analyst1, analyst2, analyst3, synthesizer],
    flow=flow,
)
```

## Key Parameters

<ParamField path="name" type="str" default="AgentRearrange">
  Name for the agent rearrange system
</ParamField>

<ParamField path="agents" type="List[Agent]" required>
  List of agents to orchestrate
</ParamField>

<ParamField path="flow" type="str" required>
  Flow pattern defining agent execution (e.g., "agent1 -> agent2, agent3")
</ParamField>

<ParamField path="max_loops" type="int" default="1">
  Maximum number of execution loops
</ParamField>

<ParamField path="team_awareness" type="bool" default="False">
  Enable agents to know their position in workflow
</ParamField>

<ParamField path="output_type" type="OutputType" default="all">
  Output format (all, final, list, dict)
</ParamField>

<ParamField path="memory_system" type="Any" default="None">
  Optional memory system for persistence
</ParamField>

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

<ParamField path="autosave" type="bool" default="True">
  Persist workflow state. Defaults to `True`.
</ParamField>

<ParamField path="time_enabled" type="bool" default="False">
  Record an ISO timestamp on every conversation message.
</ParamField>

<ParamField path="message_id_on" type="bool" default="False">
  Attach a unique id to every conversation message.
</ParamField>

## Methods

### run()

Execute the defined flow with a task.

```python theme={null}
result = rearrange.run(
    task="Analyze market trends",
    img=None,  # Optional image input
)
```

### batch\_run()

Process multiple tasks in batches.

```python theme={null}
tasks = ["Task 1", "Task 2", "Task 3"]
results = rearrange.batch_run(
    tasks=tasks,
    batch_size=10,
)
```

### concurrent\_run()

Run multiple tasks concurrently.

```python theme={null}
tasks = ["Task 1", "Task 2", "Task 3"]
results = rearrange.concurrent_run(
    tasks=tasks,
    max_workers=5,
)
```

### run\_async()

Asynchronous task execution.

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

```python theme={null}
# Sync generator
for agent_name, token in rearrange.run_stream("Analyse quantum computing trends"):
    print(f"[{agent_name}] {token}", end="", flush=True)
```

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

<Note>
  `max_loops > 1` and `custom_tasks` are not supported in streaming mode. Use `run()` for those.
</Note>

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

<ParamField path="return_str" type="bool" default="False">
  When `True`, return the plan as a string. When `False`, print it and return `None`.
</ParamField>

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

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

<Warning>
  `explain()` validates the flow first and raises if it is invalid — the same error `run()` would raise.
</Warning>

## Team Awareness

Enable agents to understand their position in the workflow:

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

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

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

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

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

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

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

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

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

### Output Formatting

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

<Note>
  **Flow Design**: Start simple and add complexity as needed. Test with "agent1 -> agent2" before complex patterns.
</Note>

1. **Clear Flow Logic**: Ensure flow makes sense for your task
2. **Agent Naming**: Use descriptive names for clarity in flow definitions
3. **Validate Flow**: Use `validate_flow()` before production
4. **Team Awareness**: Enable when agents benefit from position context
5. **Start Simple**: Begin with sequential, add concurrency where beneficial

<Warning>
  Flow validation happens at runtime - ensure all agent names in flow exist in the agents list
</Warning>

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

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

## Related Architectures

* [Sequential Workflow](/architectures/sequential-workflow) - Simpler linear flows
* [Concurrent Workflow](/architectures/concurrent-workflow) - Pure parallel execution
* [Graph Workflow](/architectures/graph-workflow) - DAG-based complex flows
* [Social Algorithms](/architectures/social-algorithms) - Custom communication patterns