Team Pipeline

TeamPipeline is a pipeline of one leader and several members: the leader assigns tasks to members through a tool call, each member works in its own context, and its final reply comes back to the leader as the tool result. The assignment flow looks like this:

                ┌── TeamAssign(member="Researcher") ──▶ Researcher ──┐
                │                                                    │
input ─▶ leader ┤                                                    ├──▶ tool results ─▶ leader ─▶ output
                │                                                    │
                └── TeamAssign(member="Coder") ───────▶ Coder ───────┘

Unlike the fixed loop of the Goal Pipeline, a team pipeline only fixes who can be assigned work and where the results go. Whom to assign, and how many times, is up to the leader. Its main properties are:

Property	Description
Results only	The leader receives only each member’s final reply, never its intermediate reasoning or tool calls, so members’ work does not flood the leader’s context
Concurrent execution	Tasks assigned to different members in the same round run concurrently; several tasks assigned to the same member run one after another in call order
No member-to-member talk	Members never talk to each other directly; their results always go back to the leader
Human-in-the-loop per member	When a member stops on a tool authorization, the request streams out as is, the answer is routed back to that member by `reply_id`, and the leader continues once the member finishes

The pipeline module is experimental, and its interfaces may change in later releases.

Run the Pipeline

The following example builds a team of a leader, a researcher and a programmer, where the leader decides whom to assign each task to:

team_pipeline.py

import asyncio
import os

from agentscope.agent import Agent
from agentscope.console import launch_console
from agentscope.credential import DashScopeCredential
from agentscope.model import DashScopeChatModel
from agentscope.pipeline import TeamMember, TeamPipeline
from agentscope.tool import Toolkit
from agentscope.workspace import LocalWorkspace


async def main() -> None:
    async with LocalWorkspace(workdir="./workspace") as workspace:
        model = DashScopeChatModel(
            credential=DashScopeCredential(
                api_key=os.getenv("DASHSCOPE_API_KEY"),
            ),
            model="qwen3.8-max",
        )

        # The leader breaks down, assigns and summarizes tasks. It must have
        # a toolkit, where the pipeline registers the TeamAssign tool
        leader = Agent(
            name="Leader",
            system_prompt="You're a team leader named 'Leader'.",
            model=model,
            toolkit=Toolkit(),
        )

        # Members are ordinary agents; the leader refers to them by name
        researcher = Agent(
            name="Researcher",
            system_prompt="You're a researcher named 'Researcher'.",
            model=model,
            toolkit=Toolkit(tools=await workspace.list_tools()),
            offloader=workspace,
        )
        coder = Agent(
            name="Coder",
            system_prompt="You're a programmer named 'Coder'.",
            model=model,
            toolkit=Toolkit(tools=await workspace.list_tools()),
            offloader=workspace,
        )

        pipe = TeamPipeline(
            leader=leader,
            members=[
                # The description goes into the TeamAssign tool description,
                # which the leader reads to decide whom to assign
                TeamMember(
                    agent=researcher,
                    description="Investigates a question and reports findings.",
                ),
                TeamMember(
                    agent=coder,
                    description="Writes and runs code in the workspace.",
                ),
            ],
        )

        # The pipeline satisfies PipelineProtocol, so the console can run it
        await launch_console(agent=pipe)


asyncio.run(main())

When members need to collaborate through files, let them share one workspace, like the LocalWorkspace the researcher and programmer share above, so files one member produces can be read directly by another.

Constructor Arguments

TeamPipeline takes the following arguments:

Argument	Type	Description
`leader`	`Agent`	The leader that assigns tasks; it must have a toolkit
`members`	`list[TeamMember]`	The members the leader can assign tasks to
`reset_members`	`bool`, defaults to `True`	Whether to clear every member’s context after each leader reply

Each member is described by a TeamMember with these fields:

Field	Type	Description
`agent`	`Agent`	The member agent; its `name` is how the leader refers to it
`description`	`str`	What the member is good at, when to assign to it and what it returns, shown to the leader in the tool description

Member names must be unique and must differ from the leader’s name, otherwise the constructor raises ValueError.

Assign Tasks

On its first run, the pipeline registers an external tool named TeamAssign in the leader’s toolkit. Its parameters are:

Parameter	Description
`member`	The member’s name, one of the names in `members`
`prompt`	The complete task for the member. Members cannot see the leader’s context, so everything the task needs must be written here

A full assignment goes through these steps:

The Leader Assigns

The leader calls TeamAssign with a member and a task. This tool skips permission checks; the member’s own tools are still checked under their own permissions.

The Member Works

The pipeline hands prompt to the member as a user message. The member runs in its own context, and its events stream out of the pipeline’s reply_stream.

The Result Returns

The member’s final reply (text and data blocks) goes back to the leader as the TeamAssign tool result. Results of several assignments in one round are ordered by the leader’s calls.

The Leader Continues

After reading the results, the leader keeps reasoning: it can assign again or give its final reply.

How a member’s reply ends determines the state of the tool result:

Member reply ends with	Tool result state
Normal completion, or reaching the maximum iterations	`SUCCESS`
Interruption	`INTERRUPTED`
An error	`ERROR`

reset_members decides what context members start the next assignment with. It defaults to True: within one leader reply, the leader can follow up with a member on the same matter; once the leader reply ends, every member’s context and summary are cleared, and assignments in the next reply start fresh. Set it to False to let members keep their conversations across replies.

Interrupt and Resume

When anyone in the team stops on a tool authorization, it is handled just like a single agent: reply_stream ends once every participant has finished or parked, and the developer feeds the answer back in to continue.

Resume after a pause

# A member emits RequireUserConfirmEvent midway, and the event stream ends
async for event in pipe.reply_stream(user_msg):
    ...

# Feed the user's answer back; the pipeline routes it to the parked member by reply_id
async for event in pipe.reply_stream(user_confirm_result_event):
    ...

While a member is parked, the leader also waits on that assignment; once the member finishes, its reply comes back as the tool result and the leader continues. Members in the same round that did not park run to completion as usual, and results that are already finished reach the leader first. reply_stream accepts these inputs:

Input type	Meaning
`Msg` / `list[Msg]`	Sent to the leader to start a new reply
`UserConfirmResultEvent`	The user’s answer to a tool authorization, routed to the parked participant by `reply_id`
`ExternalExecutionResultEvent`	The result of an external execution, routed to the parked participant by `reply_id`
`UserInterruptEvent`	Interrupts every parked member and the leader

When the pipeline’s coroutine is cancelled, it closes the parked leader and members as if it had received a UserInterruptEvent, then re-raises the cancellation only if the leader’s interruption_raise_cancelled_error is set.

Get Started

Building Blocks

Agent as Service

Others

Run the Pipeline

Constructor Arguments

Assign Tasks

Interrupt and Resume

​Run the Pipeline

​Constructor Arguments

​Assign Tasks

​Interrupt and Resume

Run the Pipeline

Constructor Arguments

Assign Tasks

Interrupt and Resume