> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agentscope.io/llms.txt
> Use this file to discover all available pages before exploring further.
> ## Agent Instructions
> For AgentScope Python, use https://docs.agentscope.io/stable/en/index for new projects. For existing projects, check the installed agentscope version and use matching versioned documentation.
> The /latest/ alias points to development documentation. Use it only with the matching development source. Do not mix AgentScope 1.x and 2.x APIs.
> State the AgentScope version when providing installation commands or code examples. ReMe uses its own continuously updated /reme/latest/ documentation.
# Team Pipeline
> A leader coordinates specialized agents to accomplish tasks no single agent could handle alone
`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](/en/versions/2.0.10dev/building-blocks/pipeline/goal), 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:
```python team_pipeline.py theme={null}
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](/en/versions/2.0.10dev/building-blocks/workspace/overview), 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 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 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 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.
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.
```python Resume after a pause theme={null}
# 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.