# SequentialPipeline vs FanoutPipeline in AgentScope: Key Differences and When to Use Each

> Understand the core differences between AgentScope's SequentialPipeline and FanoutPipeline. Learn when to use each for efficient agent communication and workflow management.

- Repository: [AgentScope-AI/agentscope](https://github.com/agentscope-ai/agentscope)
- Tags: deep-dive
- Published: 2026-03-09

---

**SequentialPipeline chains agents linearly so each agent's output becomes the next agent's input, while FanoutPipeline broadcasts the same input to all agents concurrently and returns a list of their responses.**

AgentScope provides these two high-level pipeline abstractions in [`src/agentscope/pipeline/_class.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_class.py) to simplify composing multi-agent workflows without managing async boilerplate manually. Understanding the difference between SequentialPipeline and FanoutPipeline is essential for designing efficient agent architectures, whether you need sequential refinement or parallel consensus gathering.

## Execution Model and Data Flow

The fundamental distinction lies in how messages propagate through the agent collection.

### SequentialPipeline Linear Execution

In `SequentialPipeline`, agents execute **one after another** in the order provided to the constructor. The pipeline maintains a single message pointer that mutates as it passes through each agent. According to the functional implementation in [`src/agentscope/pipeline/_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_functional.py) (lines 10-44), the logic iterates through the agent list and awaits each call sequentially:

```python
for agent in agents:
    msg = await agent(msg)
return msg

```

This design means the output of `agent[i]` becomes the input of `agent[i+1]`. The pipeline returns only the **final output** from the last agent, which can be a `Msg`, `list[Msg]`, or `None`.

### FanoutPipeline Parallel Broadcasting

`FanoutPipeline` takes the opposite approach by sending the **same input message to all agents simultaneously**. As implemented in [`src/agentscope/pipeline/_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_functional.py) (lines 46-104), each agent receives a `deepcopy` of the original message to prevent side effects between concurrent executions:

```python
if enable_gather:
    tasks = [asyncio.create_task(agent(deepcopy(msg), **kwargs)) for agent in agents]
    return await asyncio.gather(*tasks)
else:
    return [await agent(deepcopy(msg), **kwargs) for agent in agents]

```

Unlike its sequential counterpart, `FanoutPipeline` returns a **list containing every agent's output** (`list[Msg]`), enabling you to aggregate multiple perspectives on the same input.

## Implementation Details in Source Code

Both pipeline classes are thin wrappers around functional helpers defined in the AgentScope codebase.

### Functional Layer in [`_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/_functional.py)

The core logic resides in [`src/agentscope/pipeline/_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_functional.py). The `sequential_pipeline` function (lines 10-44) handles the iterative awaiting pattern, while `fanout_pipeline` (lines 46-104) manages the branching logic based on the `enable_gather` parameter. The functional layer handles message copying—`FanoutPipeline` explicitly uses `deepcopy` to isolate agent inputs, whereas `SequentialPipeline` allows in-place mutation since agents execute in isolation.

### Class Wrappers in [`_class.py`](https://github.com/agentscope-ai/agentscope/blob/main/_class.py)

The object-oriented interface lives in [`src/agentscope/pipeline/_class.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_class.py). `SequentialPipeline` (lines 10-41) stores the agent list and delegates to `sequential_pipeline` when called. `FanoutPipeline` (lines 43-90) additionally stores the `enable_gather` boolean flag, allowing the same pipeline instance to toggle between concurrent and sequential execution modes without reconstruction.

## Concurrency and Performance Characteristics

Performance characteristics diverge significantly based on the execution strategy.

- **SequentialPipeline** offers no concurrency controls—agents always run sequentially. This is ideal when steps have dependencies (e.g., "retrieve documents → summarize content → generate answer").

- **FanoutPipeline** defaults to **concurrent execution** using `asyncio.gather()` when `enable_gather=True`. This spawns independent tasks for each agent, making it suitable for parallelizable workloads like ensemble voting or multi-model comparison.

- **Controlled Sequentialism in Fanout**: Setting `enable_gather=False` forces `FanoutPipeline` to run agents sequentially via list comprehension, which is useful when you need the list-aggregation behavior of Fanout but must avoid resource contention or rate limits.

## Practical Code Examples

### Building a Linear Workflow with SequentialPipeline

Use `SequentialPipeline` when the output of one agent must feed into the next, such as a retrieve-then-summarize workflow:

```python
from agentscope.pipeline import SequentialPipeline
from agentscope.message import Msg

# Initialize agents (ReActAgent or custom implementations)

retriever = ReActAgent(...)
summarizer = ReActAgent(...)
formatter = ReActAgent(...)

# Chain them sequentially

pipeline = SequentialPipeline([retriever, summarizer, formatter])

# Execute: retriever receives input, passes to summarizer, then formatter

input_msg = Msg("user", "Analyze the latest quarterly earnings report", "user")
final_output = await pipeline(input_msg)
print(final_output.content)  # Output from formatter only

```

### Parallel Processing with FanoutPipeline

Use `FanoutPipeline` to query multiple agents with the same question and collect all responses for voting or consensus:

```python
from agentscope.pipeline import FanoutPipeline
from agentscope.message import Msg

# Three different translation agents (or same model with different configs)

translator_en = ReActAgent(...)
translator_fr = ReActAgent(...)
translator_de = ReActAgent(...)

# Execute concurrently (default)

fanout = FanoutPipeline(
    [translator_en, translator_fr, translator_de],
    enable_gather=True
)

input_msg = Msg("user", "Translate: Hello, how are you?", "user")
responses = await fanout(input_msg)  # Returns list[Msg]

for i, response in enumerate(responses, 1):
    print(f"Agent {i}: {response.content}")

```

### Sequential Aggregation with FanoutPipeline

When you need all outputs but want to avoid concurrent API calls, disable gathering:

```python

# Same pipeline, but runs one-by-one to respect rate limits

sequential_fanout = FanoutPipeline(
    [translator_en, translator_fr, translator_de],
    enable_gather=False
)

responses = await sequential_fanout(input_msg)  # Same list[Msg] return type

```

## Summary

- **SequentialPipeline** executes agents in strict order, mutates the message progressively, and returns only the final agent's output—ideal for dependent, multi-step workflows.
- **FanoutPipeline** broadcasts the input to all agents, returns a list of all outputs, and supports both concurrent (`asyncio.gather`) and sequential execution modes via the `enable_gather` flag.
- **Implementation location**: Class wrappers reside in [`src/agentscope/pipeline/_class.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_class.py) (lines 10-90), while core logic lives in [`src/agentscope/pipeline/_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_functional.py) (lines 10-104).
- **Message handling**: Sequential pipelines pass the same message object forward; fanout pipelines create `deepcopy` instances for each agent to prevent cross-contamination.

## Frequently Asked Questions

### When should I use SequentialPipeline over FanoutPipeline?

**Use SequentialPipeline when your workflow requires linear dependency chains**, such as data extraction followed by analysis followed by formatting. According to the AgentScope source code in [`src/agentscope/pipeline/_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_functional.py), this pipeline mutates the message in-place as it flows through agents, making it unsuitable for parallel execution but perfect for progressive refinement tasks.

### Does FanoutPipeline always run agents in parallel?

**No, FanoutPipeline supports both concurrent and sequential execution modes.** By default, it uses `asyncio.gather()` for concurrent execution when `enable_gather=True`. However, as implemented in [`src/agentscope/pipeline/_class.py`](https://github.com/agentscope-ai/agentscope/blob/main/src/agentscope/pipeline/_class.py) (lines 43-90), setting `enable_gather=False` switches to a list-comprehension loop that awaits each agent sequentially while still returning the aggregated list of all outputs.

### How does message passing differ between the two pipelines?

**SequentialPipeline passes the same message object forward**, allowing each agent to mutate the state for the next agent. In contrast, **FanoutPipeline creates a `deepcopy` of the input for each agent**, as shown in the functional implementation at lines 46-104 of [`_functional.py`](https://github.com/agentscope-ai/agentscope/blob/main/_functional.py). This ensures concurrent agents cannot interfere with each other's input state.

### Can I nest SequentialPipeline inside FanoutPipeline?

**Yes, both pipelines are callable objects that accept and return `Msg` objects**, allowing arbitrary composition. You can place a `SequentialPipeline` instance inside the agent list passed to `FanoutPipeline`, or vice versa, to create complex workflows like "parallel preprocessing streams that each feed into sequential processing chains." Both classes implement the `__call__` method that accepts a `Msg` and optional `**kwargs`.