实战：多 Agent 应用 🔥

前面两页分别讲了 Agent Service（用 create_app 启动多租户后端）和 Agent Team（Leader-Worker 多 Agent 协作）。本页把它们组合成一个完整可跑的多 Agent 应用：一个 Leader 协调，一个 explorer Worker 只读探索代码库，一个 writer Worker 产出文档，三者通过 Redis 消息总线异步协作，最终由 Agent Service 以 SSE 流把进度与汇总结果回推给前端。这是把"单 Agent 服务"升级成"多 Agent 服务" 的最小完整路径。

版本基线：本文以 Python agentscope 2.0.4（PyPI，2026-07-07）为准，要求 Python >=3.11。导入路径为 from agentscope.app import create_app, SubAgentTemplate，配套模块 agentscope.app.storage / agentscope.app.message_bus / agentscope.app.workspace_manager / agentscope.permission（源：src/agentscope/app/__init__.py 及各子包，已逐项核对）。2.x 文档站存在版本化陷阱：非版本化的 /en/ URL 会 404，本页所有官方链接均使用 versions/2.0.4/en/ 形式。

前置：Redis 必须先运行。本实战用 RedisStorage 做持久化、RedisMessageBus 做消息总线，两者都依赖一个可连接的 Redis 实例。启动服务前请确保 Redis 已就绪：
bash
# 方式一：本机直接启动
redis-server --port 6379

# 方式二：用 Docker 临时起一个
docker run -d --name agentscope-redis -p 6379:6379 redis:7
用 redis-cli ping 确认返回 PONG 再继续。InMemoryMessageBus 无法跨进程分发 wakeup，不能用于本实战。

⚠️ 费用提示：本实战运行时会真实调用 LLM。Leader 与每个 Worker 都会消耗 token，Worker 一经 AgentCreate 创建就立即在自己的 session 中执行，并发意味着并发计费。调试阶段建议先用便宜模型（如 qwen-turbo）并设置较小的 max_iterations，确认链路通畅后再切换生产模型。

目标：构建一个多 Agent 应用

我们要构建的应用做一件事：给定一个本地代码库路径与一句需求，自动产出一份结构化文档。职责拆分如下：

Leader：接收用户需求，调用 TeamCreate 建团队，调用 AgentCreate 生成 explorer 与 writer 两个 Worker，把探索任务派给 explorer、把写作任务派给 writer，等两者用 TeamSay 回报后汇总，最后把文档与协作过程经 SSE 流回推给前端。
explorer Worker：只读权限（PermissionMode.EXPLORE），在共享 workspace 中浏览代码库、读取文件、列出结构，把发现整理成探索报告，用 TeamSay 回报给 Leader。
writer Worker：可写权限（PermissionMode.ACCEPT_EDITS），接收 explorer 的报告与 Leader 的写作要求，产出 Markdown 文档并写回 workspace，用 TeamSay 回报文档路径与摘要。

关键约束：explorer 必须只读，writer 必须可写但只能写文档。这两条由 SubAgentTemplate 的 permission_context 显式声明，不依赖 prompt 自觉。这正是 AgentScope 2.x 把权限做进部署层的价值--角色边界是强制约束，不是建议。

前端类比

如果你来自前端，可以把这个应用类比为 带任务队列的 BFF + 两个专职微服务：

Agent Service 像一个 Next.js API Route 工厂：一行 create_app() 装配出可 uvicorn.run 的服务，差别是它装配的不是手写 controller，而是 Agent 运行时。
Leader 像一个 orchestrator 函数：拿到请求后不亲自干活，而是把子任务派发给两个专职 Worker，自己汇总。
explorer / writer 像两个 Web Worker：在独立线程（独立 session）中运行，通过 postMessage 等价的 TeamSay 与 Leader 通信，天然并发。
Redis 像消息队列 + 持久层：Leader 和 Worker 可以在不同进程甚至不同节点上，通信全部经 Redis 中转。

AgentScope 原生语义：这不是"Web Worker + 消息队列"的线程模型，也不是 microservice 的 RPC 模型。每个 Worker 是一个完整的 Agent Service session，拥有独立的 AgentState、上下文、权限上下文与 ReAct 循环。通信是异步消息传递（inbox + wakeup），Leader 不"调用"Worker 的方法，而是向其 inbox 推入 HintBlock，Worker 被唤醒后在下一轮推理前由 InboxMiddleware 排空 inbox。权限边界由 SubAgentTemplate.permission_context 在创建时锁定，prompt 无法绕过。

多 Agent 协作流程

下面这张时序图说明一次完整的多 Agent 协作：前端通过 Agent Service 向 Leader 发送需求，Leader 创建团队、生成 explorer 与 writer 两个 Worker 并派发任务，两个 Worker 在各自独立 session 中并发执行，完成后用 TeamSay 回报，Leader 汇总后把结果经 SSE 回推前端。图前先说明：所有跨 session 通信都经过 Redis 消息总线中转，Worker 一经创建就立即并发执行，不是 Leader 的嵌套协程。

图后解释关键阶段：

请求进入：前端 POST /chat 触发 Leader session 运行。/chat 立即返回 session_id，前端再用 GET /sessions/{id}/stream 订阅 SSE 流，不等长轮询。
创建团队与 Worker：Leader 调用 TeamCreate 建团队，然后两次 AgentCreate 生成 explorer 与 writer。两个 Worker 一经创建就立即开始执行，不需要额外启动信号。Leader 的 chat model 与 workspace 绑定会被 Worker 继承，但权限上下文由各自 SubAgentTemplate 决定。
并发执行：explorer 与 writer 在各自独立 session 中跑 ReAct 循环，彼此不阻塞、不嵌套。 explorer 用 EXPLORE 只读模式浏览代码库，writer 用 ACCEPT_EDITS 写文档。
回报与汇总：两个 Worker 完成后各自 TeamSay 向 Leader 回报。Leader 被 wakeup 唤醒后， InboxMiddleware 排空 inbox，团队消息以 HintBlockEvent 进入 Leader 上下文，Leader 据此汇总。
回流前端：汇总结果写入 Leader session 的 SSE 流，前端通过 /sessions/{id}/stream 收到。由于同一会话的 /chat 触发、Worker TeamSay 回报都汇聚到这一条流，前端无需为不同触发源分别建连。

用到的构建块

本实战组合了五个构建块，下表给出每个块在本应用中的具体作用：

构建块	本实战作用
Agent Service	用 `create_app` 启动 FastAPI 多租户后端，对外暴露 `/chat` 与 `/sessions/{id}/stream`
Agent Team	Leader-Worker 协作模式，Leader 用 `TeamCreate` / `AgentCreate` / `TeamSay` 编排两个 Worker
SubAgentTemplate	注册 `explorer` 与 `writer` 两种 Worker 蓝图，锁定各自的 system prompt 与权限模式
Permission	explorer 用 `PermissionMode.EXPLORE` 只读，writer 用 `PermissionMode.ACCEPT_EDITS` 可写
Redis（Storage + Bus）	`RedisStorage` 落盘 agents/sessions/messages/teams，`RedisMessageBus` 承载 inbox/wakeup/锁/replay

完整代码

把下面这段保存为 multi_agent_app.py，先确认 Redis 已运行、环境变量已注入，再 python multi_agent_app.py 启动。代码里没有任何硬编码密钥，全部通过环境变量读取。

python

import os

import uvicorn
from agentscope.app import create_app, SubAgentTemplate
from agentscope.app.storage import RedisStorage
from agentscope.app.message_bus import RedisMessageBus
from agentscope.app.workspace_manager import LocalWorkspaceManager
from agentscope.permission import PermissionContext, PermissionMode

# Redis 连接信息从环境变量读取，不要写死在代码里。
# 启动前确保 redis-server 已运行且 redis-cli ping 返回 PONG。
REDIS_HOST = os.environ.get("REDIS_HOST", "localhost")
REDIS_PORT = int(os.environ.get("REDIS_PORT", "6379"))
REDIS_PASSWORD = os.environ.get("REDIS_PASSWORD", None)

# 持久化层：agents / sessions / credentials / messages / schedules / teams。
# 连接池在 app 启动时打开、关闭时释放（lifespan 管理）。
storage = RedisStorage(
    host=REDIS_HOST,
    port=REDIS_PORT,
    password=REDIS_PASSWORD,
)

# 消息总线：会话锁、replay 日志、inbox 队列、wakeup 信号。
# Agent Team 的所有跨成员通信都经它中转，是 Worker 并发与跨进程分发的前提。
message_bus = RedisMessageBus(
    host=REDIS_HOST,
    port=REDIS_PORT,
    password=REDIS_PASSWORD,
)

# 工作区生命周期：explorer 与 writer 共享 Leader 的工作区绑定。
# LocalWorkspaceManager 无强隔离，仅用于开发；生产请换 DockerWorkspaceManager 或 E2BWorkspaceManager。
workspace_manager = LocalWorkspaceManager(
    basedir=os.environ.get("WORKSPACE_BASEDIR", "/data/workspaces"),
    ttl=3600.0,
)

# 子 Agent 模板：explorer 只读探索，writer 可写产出文档。
# 占位符 {team_name} / {team_description} / {member_name} / {member_description} / {leader_name}
# 在 AgentCreate 时由服务层填充，不要写成 ${...} 或 {{...}}。
sub_agent_templates = [
    SubAgentTemplate(
        type="explorer",
        description=(
            "Read-only agent specialized in codebase exploration. "
            "Use this type when you need to investigate files, structure, "
            "or implementation details without making any changes."
        ),
        system_prompt_template=(
            "You are {member_name}, an explorer agent in team '{team_name}' "
            "led by {leader_name}.\n\n"
            "Team purpose: {team_description}\n\n"
            "Your role: {member_description}\n\n"
            "## Responsibilities\n"
            "- Complete the exploration tasks assigned by {leader_name}.\n"
            "- You are read-only: you may inspect files and the codebase, but "
            "you must never modify, create, or delete anything.\n"
            "- Structure your findings as a concise exploration report.\n\n"
            "## Reporting\n"
            "- Always report the result back to {leader_name} using the "
            "TeamSay tool, whether the task succeeds or fails."
        ),
        permission_context=PermissionContext(
            mode=PermissionMode.EXPLORE,
        ),
    ),
    SubAgentTemplate(
        type="writer",
        description=(
            "Agent specialized in producing Markdown documentation. "
            "Use this type when you need to turn exploration findings into "
            "a structured document written back to the workspace."
        ),
        system_prompt_template=(
            "You are {member_name}, a writer agent in team '{team_name}' "
            "led by {leader_name}.\n\n"
            "Team purpose: {team_description}\n\n"
            "Your role: {member_description}\n\n"
            "## Responsibilities\n"
            "- Receive exploration findings and writing requirements from "
            "{leader_name}.\n"
            "- Produce a well-structured Markdown document and write it to "
            "the workspace.\n"
            "- You may only create or edit documentation files; do not "
            "modify source code.\n\n"
            "## Reporting\n"
            "- Always report the document path and a short summary back to "
            "{leader_name} using the TeamSay tool, whether the task "
            "succeeds or fails."
        ),
        permission_context=PermissionContext(
            mode=PermissionMode.ACCEPT_EDITS,
        ),
    ),
]

app = create_app(
    storage=storage,
    message_bus=message_bus,
    workspace_manager=workspace_manager,
    sub_agent_templates=sub_agent_templates,
)

if __name__ == "__main__":
    # 启动后访问 http://localhost:8000/docs 查看 OpenAPI 文档。
    # 前端接入流程：POST /agent 注册 Leader -> POST /credential 存 API Key ->
    # POST /sessions 建会话并绑定模型 -> POST /chat 触发运行 ->
    # GET /sessions/{id}/stream 订阅 SSE 流。
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", "8000")))

几个要点：

两个模板的权限是强制的：explorer 用 PermissionMode.EXPLORE 只读，writer 用 PermissionMode.ACCEPT_EDITS 可写。这不是 prompt 层的"请只读"，而是 Permission 三态决策在工具执行前的硬拦截。即使 explorer 的 LLM 试图调用写文件工具，也会被 deny。
Worker 继承 Leader 的 model 与 workspace：你不需要在 Worker 模板里配置模型，Leader 用什么模型、绑定哪个 workspace，Worker 就用什么。需要改的是 permission_context、context_config、 react_config 这些 Worker 专属配置。
sub_agent_templates 触发 AgentCreate schema 扩展：注册后，Leader 的 AgentCreate 工具自动增加 subagent_type 枚举字段（含 "default"、"explorer"、"writer"），Leader 看到每种类型的 description 后自行选择。不注册任何模板则不暴露该字段。

运行与期望

启动服务：

bash

# 1. 先起 Redis
docker run -d --name agentscope-redis -p 6379:6379 redis:7

# 2. 注入模型凭据（以 DashScope 为例，其他厂商同理）
export DASHSCOPE_API_KEY='你的密钥'

# 3. 注入 Redis 连接信息（本机默认可省略）
export REDIS_HOST=localhost
export REDIS_PORT=6379

# 4. 启动服务
python multi_agent_app.py

服务起来后，按以下顺序调用 REST 端点（完整字段见 http://localhost:8000/docs）：

POST /credential 存入模型 API Key。
POST /agent 注册一个 Leader Agent，在其 system prompt 里说明"可以创建 explorer 与 writer 团队来完成任务"。
POST /sessions 建会话并绑定模型与 Leader Agent。
POST /chat 发送需求，例如"探索 /data/repo 这个代码库，产出一份架构说明文档"。
GET /sessions/{id}/stream 订阅 SSE 流。

期望行为：Leader 收到需求后，调用 TeamCreate 建团队，调用 AgentCreate 生成一个 explorer Worker（派发探索任务）与一个 writer Worker（派发写作任务，但 writer 通常会等 explorer 的报告）。两个 Worker 在各自独立 session 中并发执行。explorer 用只读工具浏览代码库，把发现整理成报告用 TeamSay 回报给 Leader；Leader 再把报告转给 writer，writer 产出 Markdown 文档写回 workspace，用 TeamSay 回报文档路径与摘要。Leader 汇总两者结果，经 SSE 流回推前端。

整个过程中，前端只需持续订阅同一条 /sessions/{id}/stream，Leader 的推理、Worker 的创建与回报、最终汇总，全部汇聚到这一条流。

扩展点

用 DockerWorkspace 隔离 Worker：把 LocalWorkspaceManager 换成 DockerWorkspaceManager，每个 Worker 跑在独立 Docker 容器里，host 工作目录 bind-mount 进容器。代码只改一行，其余不动。详见部署 Agent Service。
加 TracingMiddleware 观测：用 extra_agent_middlewares 给每个 Agent 挂载 Tracing 中间件，观察 Leader 与 Worker 的工具调用、token 消耗与延迟。多 Agent 场景下可观测性尤其重要--并发 Worker 的失败面是单 Agent 的数倍。详见可观测性。
换 E2B 云沙箱：把 LocalWorkspaceManager 换成 E2BWorkspaceManager，Worker 在远程云 VM 中执行，密钥通过 E2B_API_KEY 环境变量注入。适合需要强隔离与弹性扩缩的场景。
加更多专职 Worker：注册 reviewer（只读，审查文档）、tester（可写，跑测试）等模板， Leader 按需组合。模板类型名必须唯一，重复会在启动时 ValueError。
协议适配：若前端用 AG-UI 协议，通过 extra_middlewares 挂载 AGUIProtocolMiddleware，把 SSE 帧的 AgentEvent 转成 AG-UI 格式。详见部署 Agent Service。

常见错误

Redis 未启就启动服务：RedisStorage 与 RedisMessageBus 在 lifespan 启动时会尝试连接 Redis，连不上会直接抛连接错误。先用 redis-cli ping 确认返回 PONG。InMemoryMessageBus 无法跨进程分发 wakeup，不能用于本实战。
Worker 权限过宽：未注册自定义模板时，所有 Worker 使用内置默认权限上下文（DEFAULT 模式），可能在写文件时频繁触发 HITL 确认，或意外修改源码。本实战显式给 explorer 设 EXPLORE、writer 设 ACCEPT_EDITS，就是用强制权限边界替代 prompt 自觉。根据角色选 PermissionMode，详见权限系统。
误以为 SDK 层有 Pipeline：2.x SDK 层没有 Pipeline、MsgHub、Workflow、routing 或 handoff。这些是 1.x 概念，2.x 已移除。多 Agent 协作的唯一官方方案是部署层的 Agent Team。不要在 SDK 层寻找这些原语。
⚠️ 1.x 仅作历史对照：1.x 的 Pipeline / MsgHub / ReActAgent / sys_prompt= 在 2.x 全部移除，不能作为 2.x 代码使用。迁移参见 1.x 到 2.x 迁移。
误以为 Worker 是 Leader 的嵌套协程：Worker 在独立 session 中并发运行，不是 Leader 协程内的子任务。Leader 不会被 Worker 阻塞。如果期望"调用 AgentCreate 后同步等待 Worker 完成"，这是错误的心智模型--应该让 Worker 完成后用 TeamSay 回报，Leader 被 wakeup 唤醒后汇总。
system_prompt_template 占位符写错：占位符是 Python format 字符串语法（{team_name} 等）。写成 ${team_name} 或都不会正确替换。可用占位符为 {team_name}、 {team_description}、{member_name}、{member_description}、{leader_name}。
忘记 Worker 会真实计费：Worker 一经 AgentCreate 创建就立即执行，并发意味着并发计费。调试阶段先用便宜模型并设小 max_iterations，确认链路通畅后再切生产模型。

先修

部署 Agent Service - 理解 create_app、Redis 存储、消息总线与 workspace manager，本实战建立在此基础上
Agent Team - 理解 Leader-Worker 模式、SubAgentTemplate 与内置 team tools，本实战直接组合这些能力

下一步

回到概览复盘四大支柱如何在这一实战中协同
参考 1.x 到 2.x 迁移理解为何 2.x 把多 Agent 收归到部署层

实战：多 Agent 应用 🔥 ​

目标：构建一个多 Agent 应用 ​

前端类比 ​

多 Agent 协作流程 ​

用到的构建块 ​

完整代码 ​

运行与期望 ​

扩展点 ​

常见错误 ​

先修 ​

下一步 ​

官方参考 ​