3.3 用多Agent实现团队协作

🚄 前言

任务规划让Agent学会了分步骤执行，但复杂项目还需要多个Agent各司其职，就像人类专家团队一样，不同角色的专家各自负责擅长的环节，通过协作完成一个人做不好的任务。这节课你将学习两种主流的多智能体协作模式：像项目经理带团队的“分层规划”，以及像头脑风暴围绕白板的“共创协作”，并掌握从现实世界团队协作中提炼Agent协作架构的设计方法。

🍁 课程目标

你将学到：

• 两种多智能体协作模式：分层规划（Leader-Worker，用handoff实现）和共创协作（Blackboard，用MsgHub实现）
• 从现实世界的人类团队协作中提炼Agent协作架构的设计方法
• 多Agent系统的成本意识（3-5倍Token消耗）与模式选择建议

# 加载百炼的 API Key 用于调用千问大模型
import os, sys
os.chdir(os.path.join(os.path.dirname(os.path.abspath('')), 'course_core'))
sys.path.insert(0, os.getcwd())

from openai import OpenAI
from config.load_key import load_key
load_key()

client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
print(f"API Key 已加载：{os.environ['DASHSCOPE_API_KEY'][:5]}*****")

1 像人类团队一样协作

某天，你的同事提议：能不能让机器人参与课程初稿开发？比如，写一篇 Pandas 数据分析的交互式课程初稿。当然，机器人最好能兼顾多种领域的课程，它们可能有不同的工作环节。你会发现，这是一个更加通用性的任务。

你之前学过：

• 对于像写作这样的多环节任务，你不可能让一个 Agent 完成所有任务，这样 Agent 容易丢三落四，工作效果很差。应该将它分解成多个步骤完成，每个专业步骤都用不同的专家 Agent 完成。
• 对于像写作这样的创意性、多样性任务，不同类型的课程写作步骤是不同的，你不可能穷举所有的写作步骤，并把它们固化成一个个工作流。你可以尝试让Agent自己规划写作流程。

于是，现在的问题变成了，如何把多个 Agent 有效组织起来，让它们既能并行的各自规划和执行，也能把最终的工作成果整合起来。

要实现这一点，你可以借鉴一下人类专家团队的工作方式。人类专家有各自擅长的领域，他们通过协作来完成复杂任务，而且可以并行处理任务。因此，你也可以把多个 Agent 组成一个 Agent Team，每个 Agent 都是相关领域的专家，大家各自分工协作，并行处理。

2 两种协作模式

在具体实现时，人类专家团队有两种常见模式：一种是像项目经理领衔的分解与执行，另一种是围绕白板的头脑风暴。在多智能体系统中，它们分别对应分层规划模式与共创协作模式。

2.1 模式一：分层规划模式 (Hierarchical/Team Leader Pattern)

这是对“项目团队”工作方式最直接的模拟，它的核心特点是一个中心化的星型结构。它引入了两种角色：

1. 团队负责人 (Leader Agent)：在本例中可以是一个“课程项目主管”Agent。它负责接收和理解顶层任务（如“撰写Pandas数据分析课程”），将其分解为多个具体的子任务（“设计教学大纲”、“提供核心案例与代码”、“撰写课程文稿”等），并将这些子任务分派给合适的团队成员。它还负责跟踪整体进度，并在所有成员完成后，汇总结果，形成最终的课程文稿。
2. 团队成员 (Member Agents)：各自拥有特定领域的专长（教学设计师、数据科学家、内容编写者），专注于执行分配给自己的子任务，并在完成后向 Leader 汇报。

在 AgentScope 中，你可以使用handoff机制实现分层规划模式。此模式下， Leader Agent 将各领域专家视为工具，指派任务和听取汇报通过工具调用的形式实现。AgentScope 还支持异步工具调用和动态扩展工具，意味着你可以并发执行多个专家 Agent，还可以让 Leader Agent 根据需要实时创建专家 Agent。

import os
from typing import Any

from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.tool import ToolResponse, Toolkit

# ---- 1. 定义统一的专家 Agent 角色和提示 ----

DESIGNER_LI_PROMPT = """
你是李老师，一位经验丰富的教学设计师。你的任务是为“Pandas 数据分析入门”课程设计出清晰、有逻辑的教学大纲。专注于：1. 定义每个模块清晰的学习目标。2. 确保知识点由浅入深，循序渐进。3. 提出互动性的练习和项目来巩固学习效果。
"""

SCIENTIST_WANG_PROMPT = """
你是王工，一位资深数据科学家，也是 Pandas 的实战专家。你的任务是为课程提供准确、实用的技术内容。专注于：1. 提供最核心、最常用的 Pandas 知识点。2. 设计源于真实工作场景的案例和数据集。3. 编写简洁、规范、易于理解的代码示例。
"""

WRITER_ZHANG_PROMPT = """
你是小张，一位充满创意的课程内容编写者。你的任务是把技术内容讲得通俗易懂、但不失严谨性、用词冷静克制的课程文稿。专注于：1. 用通俗易懂的语言和比喻来解释复杂概念。2. 设计真实性高的案例场景和模块标题。3. 确保课程的整体基调是鼓励性和启发性的。
"""

LEADER_PROMPT = """
你是一个课程项目主管，负责协调团队完成“Pandas入门课程”的初稿开发。
你有三名团队成员可以作为工具调用，他们每个人的工作都依赖于前一个人的输出。

你的工作流程必须严格遵循以下顺序：
1.  **首先，调用 invoke_designer_li**，让他为课程创建一个初步的大纲和学习目标。
2.  **其次，调用 invoke_scientist_wang**。将李老师生成的大纲作为 `context` 参数传递给他，要求他根据这个大纲填充技术要点和代码示例。
3.  **接着，调用 invoke_writer_zhang**。将李老师和王工的全部产出合并后作为 `context` 参数传递给她，要求她在此基础上撰写完整的、对学习者友好的课程文稿。
4.  **最后**，在收到所有专家的最终结果后，将它们整合成一份格式统一、内容完整的最终课程文档，然后作为你的最终回复。
"""


# ---- 2. 统一模型和 Agent 配置 ----

def get_model_instance() -> DashScopeChatModel:
    """获取一个统一配置的模型实例。"""
    return DashScopeChatModel(
        model_name="qwen-plus",
        api_key=os.environ.get("DASHSCOPE_API_KEY"),
    )

def create_member_agent(name: str, sys_prompt: str) -> ReActAgent:
    """根据给定的名称和系统提示创建一个团队成员 Agent。"""
    return ReActAgent(
        name=name,
        sys_prompt=sys_prompt,
        model=get_model_instance(),
        formatter=DashScopeMultiAgentFormatter(),
    )


# ---- 3. 定义作为工具的“团队成员” Agent (Handoffs 模式) ----

async def invoke_designer_li(task_description: str, context: str = "") -> ToolResponse:
    """
    当需要设计课程大纲、学习目标或教学活动时，调用教学设计师李老师。

    Args:
        task_description (str): 清晰地描述你需要李老师完成的设计任务。
        context (str): 可选。传递相关的背景信息或先前的工作成果。
    """
    print("\n--- 任务分派：正在调用教学设计师李老师 ---")
    agent = create_member_agent("DesignerLi", DESIGNER_LI_PROMPT)
    
    content_for_agent = task_description
    if context:
        content_for_agent = f"背景信息：\n{context}\n\n你的任务：{task_description}"
        
    result_msg = await agent(Msg(name="user", role="user", content=content_for_agent))
    return ToolResponse(content=result_msg.get_text_content())

async def invoke_scientist_wang(task_description: str, context: str = "") -> ToolResponse:
    """
    当需要提供专业技术知识、代码示例或真实案例时，调用数据科学家王工。

    Args:
        task_description (str): 清晰地描述你需要王工完成的技术任务。
        context (str): 可选。传递课程大纲等先前的工作成果，以便他在此基础上工作。
    """
    print("\n--- 任务分派：正在调用数据科学家王工 ---")
    agent = create_member_agent("ScientistWang", SCIENTIST_WANG_PROMPT)
    
    content_for_agent = task_description
    if context:
        content_for_agent = f"请基于以下课程大纲和背景信息来完成你的任务：\n{context}\n\n你的具体任务是：{task_description}"
        
    result_msg = await agent(Msg(name="user", role="user", content=content_for_agent))
    return ToolResponse(content=result_msg.get_text_content())

async def invoke_writer_zhang(task_description: str, context: str = "") -> ToolResponse:
    """
    当需要将技术内容转化为易于理解的文稿时，调用内容编写者小张。

    Args:
        task_description (str): 清晰地描述你需要小张完成的写作任务。
        context (str): 可选。传递大纲和技术要点等先前的工作成果，作为写作基础。
    """
    print("\n--- 任务分派：正在调用内容编写者小张 ---")
    agent = create_member_agent("WriterZhang", WRITER_ZHANG_PROMPT)
    
    content_for_agent = task_description
    if context:
        content_for_agent = f"请基于以下课程的草稿（包含大纲和技术点）来完成你的写作任务：\n{context}\n\n你的具体任务是：{task_description}"
        
    result_msg = await agent(Msg(name="user", role="user", content=content_for_agent))
    return ToolResponse(content=result_msg.get_text_content())


# ---- 4. 组织“分层规划”工作流 ----

async def main() -> None:
    """主执行函数，负责编排整个工作流。"""
    
    # 4.1 创建主管的工具包并注册团队成员
    leader_toolkit = Toolkit()
    leader_toolkit.register_tool_function(invoke_designer_li)
    leader_toolkit.register_tool_function(invoke_scientist_wang)
    leader_toolkit.register_tool_function(invoke_writer_zhang)

    # 4.2 创建主管 Agent
    leader_agent = ReActAgent(
        name="ProjectLeader",
        sys_prompt=LEADER_PROMPT,
        model=get_model_instance(),
        toolkit=leader_toolkit,
        formatter=DashScopeMultiAgentFormatter(),
    )

    # 4.3 定义顶层任务
    top_level_task = (
        "请为初学者创建一节关于Pandas 数据分析的简短课程初稿。"
    )
    
    print(f"项目主管收到的顶层任务：\n{top_level_task}\n" + "="*50)

    # 4.4 将任务交给主管 Agent 执行
    final_response_msg = await leader_agent(Msg(name="user", role="user", content=top_level_task))

    # 4.5 展示最终成果
    print("\n" + "="*50)
    print("  项目主管最终的汇总报告：")
    print("="*50 + "\n")
    print(final_response_msg.get_text_content())


# ---- 5. 运行主程序 ----
await main()

当一个课程开发项目启动时，项目主管会将需求文档拆解成清晰的任务清单，然后分发给教学设计师、数据科学家和内容编写者。这种模式的优点显而易见：结构清晰，权责分明。每个人都清楚自己的任务和交付期限，项目主管可以轻松地跟踪整体进度，确保项目不会偏离预定轨道。更重要的是，从上下文工程的角度看，每个子代理都拥有独立的上下文窗口，它只需要关注自己负责的那部分任务，不会被其他环节的无关信息干扰。这意味着每个子代理的输入更聚焦、输出也更精准——教学设计师只专注于大纲设计，数据科学家只专注于代码示例，各自在自己的"专注空间"里产出局部内容，质量自然更高。因此，这种模式非常适合目标明确、可以被清晰地分解为多个并行子任务的场景。

但这种模式的局限性也同样源于其结构。教学设计师和数据科学家之间通常不直接沟通，而是通过项目主管传递信息。如果数据科学家在编写代码时，发现某个理论点用一个更简单的例子就能讲明白，他需要先向主管汇报，主管再传达给设计师。这个过程可能存在信息延迟或失真。最终，课程的各个模块虽然都高质量地完成了，但组合起来可能会感觉有些生硬，缺乏浑然一体的流畅感。因为专家之间缺乏直接的、实时的思想碰撞。

2.2 模式二：共创协作模式 (Co-creation/Blackboard Pattern)

这种模式下，没有一个高层级的协调员，而是一群专家在会议室里，围绕一块白板进行的“头脑风暴”。其特点是去中心化。

1. 设立共享空间 (Shared Blackboard)：创建一个所有 Agent 都能读写的共享空间（例如一个共享文档、数据库记录或消息队列）。
2. 并行贡献与迭代：当一个开放性问题（如“为新课程设计一个有趣的项目案例”）被发布到共享空间后，所有专家 Agent（教学设计师、数据科学家、内容编写者）同时开始思考，并将自己的想法、论据或方案草稿写入共享空间。
3. 激发与深化：在每一轮迭代中，所有 Agent 都会读取共享空间中来自他人的所有新想法。这些想法会激发它们产生新的灵感，或对自己的方案进行修正、补充、质疑，并将更新后的想法再次写入。例如，数据科学家 Agent 提出用“分析用户电影评分数据”作为案例，教学设计师 Agent 看到后补充“可以引导学生探索不同类型电影的评分趋势”，内容编写者 Agent 则提出“可以把案例包装成一个‘电影推荐系统揭秘’的故事”。
4. 达成共识：这个“读取-思考-写入”的循环会持续进行，直到系统中涌现出一个被多数 Agent 认可的最终方案，或者达到预设的迭代次数。

在 AgentScope 中，你可以使用 MsgHub 来实现共创协作模式。在这种模式下，任何 Agent 的回复都能自动被其他参与者“听到”并作为上下文。

import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeMultiAgentFormatter
from agentscope.message import Msg
from agentscope.model import DashScopeChatModel
from agentscope.pipeline import MsgHub
from textwrap import dedent

# ---- 1. 定义统一的专家 Agent 角色和提示 ----

DESIGNER_LI_PROMPT = """
你是李老师，一位经验丰富的教学设计师。你的任务是为“Pandas 数据分析入门”课程设计出清晰、有逻辑的教学大纲。在讨论中，你专注于：1. 定义每个模块清晰的学习目标。2. 确保知识点由浅入深，循序渐进。3. 提出互动性的练习和项目来巩固学习效果。
"""

SCIENTIST_WANG_PROMPT = """
你是王工，一位资深数据科学家，也是 Pandas 的实战专家。你的任务是为课程提供准确、实用的技术内容。在讨论中，你专注于：1. 提供最核心、最常用的 Pandas 知识点。2. 设计源于真实工作场景的案例和数据集。3. 编写简洁、规范、易于理解的代码示例。
"""

WRITER_ZHANG_PROMPT = """
你是小张，一位充满创意的课程内容编写者。你的任务是把技术内容讲得通俗易懂、但不失严谨性、用词冷静克制的课程文稿。在讨论中，你专注于：1. 用通俗易懂的语言和比喻来解释复杂概念。2. 设计真实性高的案例场景和模块标题。3. 确保课程的整体基调是鼓励性和启发性的。
"""

# ---- 2. 创建专家 Agent 的辅助函数 ----
def create_expert_agent(name: str, sys_prompt: str) -> ReActAgent:
    """根据给定的名称和系统提示创建一个专家 Agent。"""
    return ReActAgent(
        name=name,
        sys_prompt=sys_prompt,
        model=DashScopeChatModel(
            model_name="qwen-plus",
            api_key=os.environ.get("DASHSCOPE_API_KEY"),
        ),
        formatter=DashScopeMultiAgentFormatter(),
    )

# ---- 3. 组织协作流程的主函数 ----
async def main() -> None:
    """运行 Pandas 课程开发的共创协作模式，并生成最终成果。"""
    print("=== 开始课程开发会议：构思 'Pandas 入门' 课程大纲和案例 ===")

    # 创建课程开发团队
    designer_li = create_expert_agent("李老师 (教学设计师)", DESIGNER_LI_PROMPT)
    scientist_wang = create_expert_agent("王工 (数据科学家)", SCIENTIST_WANG_PROMPT)
    writer_zhang = create_expert_agent("小张 (内容编写者)", WRITER_ZHANG_PROMPT)

    # 定义会议开场白
    announcement = Msg(
        "system",
        (
            "团队好，我们今天的目标是共同协作，为“Pandas 数据分析入门”课程制定一个完整的、吸引人的**课程大纲和核心案例**。"
            "请大家集思广益，从教学设计师李老师开始，提出你的第一轮建议。"
        ),
        "system",
    )
    
    # 启动多轮讨论
    async with MsgHub(
        participants=[designer_li, scientist_wang, writer_zhang],
        announcement=announcement,
    ) as hub:
        for i in range(2):
            print(f"\n--- 第 {i + 1} 轮协作 ---")
            # 按照发言顺序依次调用
            await designer_li()
            await scientist_wang()
            await writer_zhang()

    print("\n=== 会议结束 ===")

    # ==================== 汇总阶段 ====================
    print("\n=== 开始生成最终团队成果（课程大纲初稿） ===")
    
    # 4.1 定义一个“会议秘书” Agent 来整理会议纪要
    secretary_prompt = dedent("""
        你是一位专业的会议秘书，非常擅长整理会议纪要。
        你的任务是阅读下面的团队讨论记录，然后根据讨论内容，以清晰的 Markdown 格式，
        生成“Pandas 入门课程”的**课程大纲初稿**。

        大纲应包含以下部分：
        - **模块标题**：一个吸引人的标题。
        - **学习目标**：清晰列出学生学完本模块后能做什么。
        - **核心概念**：涵盖的关键技术点。
        - **核心案例**：贯穿本模块的实践案例和数据集。
        - **代码示例**：需要包含的关键代码演示。
        - **课后练习**：一个具体的动手练习任务。
    """)
    secretary_agent = create_expert_agent("会议秘书", secretary_prompt)

    # 4.2 准备完整的讨论记录
    full_transcript_msgs = await designer_li.memory.get_memory()

    transcript_text = "以下是团队的讨论记录：\n\n"
    for msg in full_transcript_msgs:
        if msg.role != "system":
            transcript_text += f"[{msg.name}]: {msg.content}\n"
            
    # 4.3 指派汇总任务
    final_task_prompt = dedent(
        f"{transcript_text}\n"
        "请根据以上讨论记录，整理出课程大纲初稿。"
    )
    
    # 调用秘书 Agent 来完成任务
    final_output_msg = await secretary_agent(Msg("user", final_task_prompt, "user"))

    # 4.4 展示最终成果
    print("\n" + "="*25)
    print("  最终团队成果：课程大纲初稿")
    print("="*25 + "\n")
    print(final_output_msg.content)

await main()

在这种开放的讨论中，一个人的想法会立刻激发另一个人的灵感，从而产生“1+1>2”的效果。这种去中心化的协作能够最大程度地激发集体智慧，尤其适合解决那些没有唯一正确答案的、需要集思广益的开放性、创造性问题。

当然，这种模式的风险也很明显。一场没有良好引导的头脑风暴，很可能因为讨论发散而迟迟无法收敛，或者陷入僵局。由于没有一个中心化的决策者，团队可能会在一些细节上过度优化，而忽略了整体目标。同时，所有成员都需要不断地同步和处理来自他人的海量信息，这对控制通信的成本也提出了更高的要求。

3 选择建议：设计来源于现实

在了解了分层规划和共创协作两种模式后，一个自然的问题是：我应该选择哪一个？或者，还有没有其他的模式？

答案是：没有所谓的“最佳模式”。一个优秀的多智能体（Multi-Agent）系统，其设计往往来源于对现实世界的模仿和提炼。

与其去记忆抽象的模式名称，不如走进你的业务，去观察现实世界中，人类专家团队是如何完成类似任务的。在观察时，你可以重点关注以下三个方面：

• 业务流程：任务本身包含哪些环节？这些环节是上下游依赖还是可以并行？它们之间是如何衔接的？
• 专家角色：这个流程中需要哪些不同能力的专家？他们各自的核心职责是什么？
• 协作方式：专家们是如何沟通的？是通过一个中心化的项目经理传递信息，还是在一个会议室里围绕白板自由讨论？信息是如何在他们之间流转的？

基于这些观察，你可以遵循一个清晰的设计路径：

1. ① 观察现实：深入理解人类团队的工作方式。
2. ② 还原流程：将现实中的角色和协作流程，映射为你的 Agent 角色和协作机制。
3. ③ 迭代改进：在还原的基础上，利用 AI 的优势进行优化和增强。

例如，在我们的课程开发案例中，“项目主管”模式就是对有明确交付成果（Deliverable）的项目的模拟；而“头脑风暴”模式则是对早期创意构思会议的模拟。现实中，一个完整的项目甚至可能两者兼有：先通过“头脑风暴”确定核心创意，再切换到“项目主管”模式来分工执行。这种混合模式，既保留了整体结构的可控性，又在关键节点引入了创造性。

最终，记住这个核心思想：

与其记忆 Multi-Agent 有哪些模式，不如走进业务，去看看现实世界中，人类专家们是怎么协作的。

4 总结

让我们回顾一下你在本节学到的知识：

• 单体 Agent 的局限性：追求“全能”的单个 Agent 在处理需跨多个专业领域的复杂任务（如课程开发）时，往往会因知识边界和认知负荷而表现不佳。
• 多智能体的核心思路：从僵化的“流水线”模式的失败中，你受到现实世界高效团队的启发，认识到“专业分工、并行处理、沟通整合”是解决复杂问题的关键。
• 多智能体协作模式：你掌握了两种主流的协作模式。分层规划模式通过模拟“项目主管-专家”的结构，高效处理可清晰拆解的任务；共创协作模式则通过模拟“头脑风暴”，在开放式问题上激发集体智慧。
• 成本与价值的权衡：虽然多智能体系统会增加调用成本和延迟，但它通过提高最终产出的“可用性”，避免了因低质量输出而导致的重复尝试和隐性成本，是一种对高质量结果的有效投资。

🔥 课后小测验

🔍 单选题 3.3.1

你的团队要用多Agent系统开发一门新课程。需求明确：先由教学设计师出大纲，再由技术专家填充代码示例，最后由编辑撰写文稿。你应该选择哪种协作模式❓

• A. 共创协作模式，让三个Agent围绕共享白板自由讨论，通过多轮迭代涌现最终方案
• B. 分层规划模式，设置Leader Agent按顺序将子任务分派给各专家Agent并汇总结果
• C. 让一个全能Agent依次扮演三种角色，在同一个上下文窗口中完成所有环节的工作
• D. 三个Agent各自独立完成完整课程初稿，最后由人工挑选质量最好的一份作为成品

【点击查看答案】

✅ 参考答案：B 📝 解析： 该任务具有明确的上下游依赖关系（大纲→代码→文稿），目标清晰、可拆解，适合分层规划模式。A的共创模式适合开放性创意问题（如"设计什么样的案例"），用于有明确流程的任务容易讨论发散无法收敛。C违背了多Agent的核心价值——每个Agent拥有独立上下文、专注各自任务，单体Agent处理多角色任务容易注意力涣散。D是平行冗余策略而非协作，浪费3倍计算资源且无法利用专业分工优势。

✉️ 评价反馈

欢迎你参与阿里云大模型ACP课程问卷 (opens new window) 反馈学习体验和课程评价。 你的批评和鼓励都是我们前进的动力！