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

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 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 (lines 10-44), the logic iterates through the agent list and awaits each call sequentially:

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 (lines 46-104), each agent receives a deepcopy of the original message to prevent side effects between concurrent executions:

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

The core logic resides in 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

The object-oriented interface lives in 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:

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:

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:

# 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 (lines 10-90), while core logic lives in 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, 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 (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. 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}