章节导语

一个人的能力有限，团队配合更高效。AI Agent也是这样——单个Agent能力有上限，多个专业Agent协作能完成更复杂的任务。

但多Agent协作不是简单地把多个Agent拼在一起。关键问题有四个：任务怎么分配？信息怎么共享？意见冲突怎么办？结果怎么汇总？

本文会系统讲解三种多Agent协作模式，每种都有完整代码和实战案例，让你真正掌握协作机制。

一、前置说明

1.1 学习路径

建议先掌握单Agent模式（Day1），再学习多Agent协作：

1.2 读者需要的基础

• 单Agent知识：理解Direct、ReAct等基本模式
• Python面向对象：会写类和实例
• API调用：知道怎么调用大模型

1.3 学习目标

学完本文，你将能够：

• 理解三种多Agent协作模式的原理
• 实现Agent之间的信息共享和任务分配
• 处理Agent间的冲突和结果汇总
• 构建一个多Agent客服系统

二、为什么需要多Agent协作

2.1 单Agent的局限

即使是最强大的GPT-4，单Agent也有明显局限：

• 知识有边界：一个Agent不可能精通所有领域
• 能力有专长：写代码强的，营销可能一般
• 效率天花板：一个Agent处理多任务要来回切换

2.2 多Agent的优势

多个专业Agent协作可以：

• 专业分工：研究员收集信息，文案师写稿，审核员把关
• 并行处理：多个子任务同时执行
• 互补不足：A不懂的B懂，B不会的C会
• 容错性强：一个Agent出错，其他可以补位

2.3 协作的核心问题

多Agent不是简单并排，需要解决4个核心问题：

三、三种协作模式概述

3.1 模式一览

3.2 怎么选模式

选择依据主要是任务复杂度和协调需求：

选Orchestra：任务类型明确、可分类，流程相对固定。

选Hierarchical：复杂业务流程，多层级，需要监督和协调。

选HiveMind：创意生成、多角度分析，没有标准答案。

四、Orchestra模式：有协调者的协作

4.1 原理讲解

Orchestra是最常用的多Agent模式。有一个协调者（Conductor）负责任务分配，其他Agent执行具体任务。

想象一个项目团队：

• 项目经理 = 协调者，接收任务、分配工作
• 程序员 = 执行技术任务
• 设计师 = 执行设计任务
• 测试工程师 = 执行测试任务

工作流程：

1. 协调者接收任务
2. 分析任务类型
3. 分配给合适的Agent
4. Agent执行任务
5. 结果返回协调者
6. 如需要，分配给下一个Agent

4.2 代码实现

下面实现一个简单的Orchestra系统：

class Agent:
    """专业Agent：负责执行具体任务
    
    每个Agent有自己的专长（specialty）
    根据专长来分配任务
    """
    
    def __init__(self, name, specialty):
        self.name = name          # Agent名称
        self.specialty = specialty  # 专长领域
    
    def work(self, task):
        """执行分配的任务
        
        参数:
            task: 要执行的任务描述
        返回:
            执行结果
        """
        # 模拟处理：实际应用中这里会调用大模型
        return f"[{self.name}] 已完成 {task[:30]}..."

class Conductor:
    """协调者：负责任务分配
    
    核心职责：
    - 分析任务类型
    - 选择合适的Agent
    - 控制工作流程
    """
    
    def __init__(self):
        # 创建Agent池（专业分工）
        self.agents = {
            "researcher": Agent("研究员", "信息收集、调研"),
            "writer": Agent("文案师", "内容创作、文案撰写"),
            "editor": Agent("编辑", "审核、修改、质量把控")
        }
    
    def assign(self, task):
        """根据任务类型分配Agent
        
        参数:
            task: 任务描述
        返回:
            分配到的Agent名称
        """
        # 简单规则匹配，实际可用AI做更智能的路由
        if any(k in task for k in ["分析", "调研", "收集", "查"]):
            return "researcher"
        if any(k in task for k in ["写", "创作", "生成"]):
            return "writer"
        if any(k in task for k in ["审核", "检查", "修改"]):
            return "editor"
        
        # 默认分配给研究员
        return "researcher"
    
    def conduct(self, task):
        """指挥完成一个任务
        
        参数:
            task: 任务描述
        返回:
            任务执行结果
        """
        print(f"\n🎯 收到任务: {task}")
        
        # 1. 分析并分配
        agent_type = self.assign(task)
        agent = self.agents[agent_type]
        print(f"📢 分配给: {agent.name}（专长: {agent.specialty}）")
        
        # 2. 执行任务
        result = agent.work(task)
        print(f"✅ 执行完成: {result[:50]}...")
        
        return result

# 测试
if __name__ == "__main__":
    conductor = Conductor()
    
    # 测试不同任务
    conductor.conduct("分析竞品A的市场定位")
    conductor.conduct("写一篇关于AI的文章")
    conductor.conduct("检查这篇文章的语法错误")

4.3 运行结果

$ python orchestra.py

🎯 收到任务: 分析竞品A的市场定位
📢 分配给: 研究员（专长: 信息收集、调研）
✅ 执行完成: [研究员] 已完成 分析竞品A的市场...

🎯 收到任务: 写一篇关于AI的文章
📢 分配给: 文案师（专长: 内容创作、文案撰写）
✅ 执行完成: [文案师] 已完成 写一篇关于AI的...

🎯 收到任务: 检查这篇文章的语法错误
📢 分配给: 编辑（专长: 审核、修改、质量把控）
✅ 执行完成: [编辑] 已完成 检查这篇文章的...

4.4 进阶：顺序协作链

有时候一个任务需要多个Agent顺序执行。比如：研究→写作→审核。

class OrchestraChain(Conductor):
    """Orchestra扩展：支持顺序协作链
    
    继承自Conductor，新增多Agent顺序协作功能
    """
    
    def conduct_chain(self, task, chain):
        """顺序执行多个Agent
        
        参数:
            task: 初始任务
            chain: Agent顺序列表，如 ["researcher", "writer", "editor"]
        返回:
            最终执行结果
        """
        print(f"\n🔗 开始协作链: {' → '.join(chain)}")
        print(f"🎯 初始任务: {task}\n")
        
        current_result = task
        
        for i, agent_type in enumerate(chain, 1):
            agent = self.agents[agent_type]
            print(f"[步骤{i}] {agent.name} 处理中...")
            
            # 上一个Agent的结果作为下一个的输入
            current_result = agent.work(current_result)
            print(f"[步骤{i}] {agent.name} 完成: {current_result[:40]}...\n")
        
        print("✅ 协作链执行完成")
        return current_result

# 测试
if __name__ == "__main__":
    orchestra = OrchestraChain()
    
    # 完整流程：调研 → 写作 → 审核
    result = orchestra.conduct_chain(
        task="关于AI大模型的发展趋势",
        chain=["researcher", "writer", "editor"]
    )
    print(f"\n最终结果: {result}")

4.5 避坑指南

坑1：任务分配不合理

如果分配给错误的Agent，可能白费功夫。解决方案：建立更智能的路由机制。

坑2：Agent之间信息丢失

上一个Agent的结果没有正确传给下一个。解决方案：明确输入输出格式。

坑3：单点故障

如果协调者崩溃，整个系统瘫痪。解决方案：添加备份协调者。

4.6 小结

Orchestra模式适合任务明确、可分类的场景。协调者负责分配，专业Agent执行，流程清晰简单。

4.7 练习题

基础题：添加一个”设计师”Agent，能处理设计相关任务。

进阶题：实现并行协作——多个Agent同时执行不同子任务。

挑战题：添加Agent”超时”和”重试”机制。

五、Hierarchical模式：层级治理

5.1 原理讲解

Hierarchical是更复杂的多Agent模式。采用层级结构，上层管规划，下层管执行。

想象一家公司：

• CEO = Manager，制定全局战略
• 部门经理 = Supervisor，监督执行、协调问题
• 员工 = Worker，执行具体任务

工作流程：

1. Manager接收任务，做全局规划
2. 分解成子任务，分配给Supervisor
3. Supervisor分配给Worker执行
4. Worker执行任务，逐层上报结果
5. Manager汇总结果，给出最终输出

5.2 代码实现

class Worker:
    """执行层Agent：最终执行任务
    
    最底层的Agent，直接执行具体任务
    """
    
    def __init__(self, name, skills):
        self.name = name
        self.skills = skills  # 能处理的技能列表
    
    def can_handle(self, task):
        """判断是否能处理这个任务"""
        return any(skill in task for skill in self.skills)
    
    def execute(self, task):
        """执行任务"""
        return f"[{self.name}] 执行完成: {task[:30}]..."

class Supervisor:
    """监督层Agent：分配任务给Worker
    
    中间层Agent，负责任务分配和进度监督
    """
    
    def __init__(self, name):
        self.name = name
        # 初始化Worker池
        self.workers = [
            Worker("技术员", ["代码", "bug", "技术", "开发"]),
            Worker("销售员", ["价格", "报价", "销售", "购买"]),
            Worker("客服", ["问题", "咨询", "帮助", "售后"])
        ]
    
    def assign(self, task):
        """分配任务给合适的Worker"""
        for worker in self.workers:
            if worker.can_handle(task):
                return worker
        return self.workers[0]  # 默认返回第一个
    
    def supervise(self, task):
        """监督执行流程"""
        worker = self.assign(task)
        print(f"  [{self.name}] 分配给: {worker.name}")
        return worker.execute(task)

class Manager:
    """管理层Agent：全局规划和协调
    
    最顶层Agent，负责任务分解和全局协调
    """
    
    def __init__(self):
        # 创建Supervisor池
        self.supervisors = {
            "tech": Supervisor("技术主管"),
            "biz": Supervisor("商务主管")
        }
    
    def plan(self, task):
        """分解任务为子任务
        
        实际应用中，可以用AI来做更智能的任务分解
        """
        print(f"\n[Manager] 接收任务: {task}")
        
        # 简化处理：根据关键词分解
        # 实际应用中可能更复杂
        if "代码" in task or "技术" in task:
            return task, "tech"
        else:
            return task, "biz"
    
    def manage(self, task):
        """管理工作流程"""
        # 1. 规划分解
        subtask, supervisor_type = self.plan(task)
        supervisor = self.supervisors[supervisor_type]
        
        print(f"[Manager] 分配给: {supervisor.name}")
        
        # 2. 监督执行
        result = supervisor.supervise(subtask)
        
        # 3. 汇总结果
        print(f"[Manager] 收到结果: {result}")
        
        return result

# 测试
if __name__ == "__main__":
    manager = Manager()
    
    print("=" * 50)
    print("测试：处理技术问题")
    print("=" * 50)
    manager.manage("用户的代码有bug怎么办")
    
    print("\n" + "=" * 50)
    print("测试：处理商务问题")
    print("=" * 50)
    manager.manage("这个产品多少钱")

5.3 运行结果

$ python hierarchical.py

==================================================
测试：处理技术问题
==================================================

[Manager] 接收任务: 用户的代码有bug怎么办
[Manager] 分配给: 技术主管
  [技术主管] 分配给: 技术员
  [技术主管] 执行完成: [技术员] 执行完成: 用户的代码...

[Manager] 收到结果: [技术主管] 执行完成: [技术员]...

==================================================
测试：处理商务问题
==================================================

[Manager] 接收任务: 这个产品多少钱
[Manager] 分配给: 商务主管
  [商务主管] 分配给: 销售员
  [商务主管] 执行完成: [商务主管] 执行完成: [销售员]...

5.4 适用场景

• 企业级应用：复杂的业务流程，多层级
• 需要监督机制：任务执行需要跟踪和协调
• 资源调度：多个Worker需要协调工作

5.5 避坑指南

坑1：层级过深

层级太多会导致信息传递失真、响应变慢。解决方案：尽量扁平化，控制在3层以内。

坑2：Supervisor成为瓶颈

如果Supervisor太忙，会拖慢整体进度。解决方案：多个Supervisor并行工作。

坑3：Worker闲置

某些Worker可能一直空闲，而其他的很忙。解决方案：实现动态任务分配。

5.6 小结

Hierarchical模式适合复杂业务流程。通过层级管理，每层各司其职，但系统复杂度也更高。

5.7 练习题

基础题：添加一个新的Supervisor，负责人力资源相关任务。

进阶题：实现Supervisor之间的任务转移——当一个Supervisor太忙时，转移任务给其他。

挑战题：添加”异常上报”机制——Worker遇到无法处理的问题时，自动上报给Supervisor。

六、HiveMind模式：去中心化协作

6.1 原理讲解

HiveMind是激进的多Agent模式。没有中心协调者，所有Agent完全对等，通过投票或共识做决策。

就像一场头脑风暴会议：

• 每个人都可以发言
• 观点没有对错
• 通过讨论说服彼此
• 最后投票决定或达成共识

适用场景：

• 创意生成
• 多角度分析
• 复杂决策（没有标准答案）

6.2 代码实现

from collections import Counter

class HiveAgent:
    """去中心化Agent：独立思考，分享观点
    
    每个Agent有自己的视角（perspective）
    独立思考，然后分享观点
    """
    
    def __init__(self, name, perspective):
        self.name = name
        self.perspective = perspective  # 视角：技术/商业/用户
    
    def think(self, topic):
        """独立思考，给出观点
        
        参数:
            topic: 讨论话题
        返回:
            个人观点
        """
        # 模拟思考过程
        templates = {
            "技术专家": "从技术角度分析，{topic}需要考虑架构设计、可扩展性、技术栈选择等因素。",
            "商业专家": "从商业角度分析，{topic}需要考虑市场规模、竞争对手、盈利模式、用户增长等因素。",
            "用户专家": "从用户角度分析，{topic}需要考虑用户需求、使用体验、定价策略、客户服务等因素。"
        }
        
        template = templates.get(self.name, "{topic}是一个复杂问题，需要综合考虑。")
        return template.format(topic=topic)

class HiveMind:
    """去中心化协作：投票决策
    
    没有中心协调者
    每个Agent独立思考后，通过投票或共识做决策
    """
    
    def __init__(self):
        # 创建多角度Agent团队
        self.agents = [
            HiveAgent("技术专家", "技术实现"),
            HiveAgent("商业专家", "市场商业"),
            HiveAgent("用户专家", "用户体验")
        ]
    
    def discuss(self, topic):
        """讨论模式：各抒己见
        
        参数:
            topic: 讨论话题
        返回:
            各Agent的观点列表
        """
        print(f"\n💭 议题: {topic}\n")
        print("-" * 50)
        
        opinions = []
        for agent in self.agents:
            print(f"[{agent.name}] 思考中...")
            opinion = agent.think(topic)
            opinions.append(opinion)
            print(f"  → {opinion}\n")
        
        return opinions
    
    def vote(self, topic):
        """投票模式：选最佳方案
        
        参数:
            topic: 投票议题
        返回:
            投票结果
        """
        opinions = self.discuss(topic)
        
        # 简化投票：让AI判断哪个观点最全面
        print("-" * 50)
        print("🗳️ 开始投票...\n")
        
        # 这里简化处理，实际可以用更复杂的投票机制
        votes = Counter(opinions)
        winning_opinion = votes.most_common(1)[0][0]
        
        print(f"📊 投票结果: {winning_opinion}")
        return winning_opinion

# 测试
if __name__ == "__main__":
    hivemind = HiveMind()
    
    # 测试讨论模式
    print("=" * 50)
    print("讨论模式")
    print("=" * 50)
    hivemind.discuss("要不要上线AI Agent功能")
    
    # 测试投票模式
    print("\n" + "=" * 50)
    print("投票模式")
    print("=" * 50)
    hivemind.vote("选择哪个技术方案")

6.3 运行结果

$ python hivemind.py

==================================================
讨论模式
==================================================

💭 议题: 要不要上线AI Agent功能

--------------------------------------------------
[技术专家] 思考中...
  → 从技术角度分析，要不要上线AI Agent功能需要考虑架构设计、可扩展性、技术栈选择等因素。

[商业专家] 思考中...
  → 从商业角度分析，要不要上线AI Agent功能需要考虑市场规模、竞争对手、盈利模式、用户增长等因素。

[用户专家] 思考中...
  → 从用户角度分析，要不要上线AI Agent功能需要考虑用户需求、使用体验、定价策略、客户服务等因素。

==================================================
投票模式
==================================================

[技术专家] 思考中...
[商业专家] 思考中...
[用户专家] 思考中...
🗳️ 开始投票...

📊 投票结果: 从技术角度分析...

6.4 进阶：共识机制

更高级的HiveMind是让Agent互相讨论、说服，最终达成共识。

class ConsensusHiveMind(HiveMind):
    """HiveMind扩展：共识机制
    
    Agent之间可以互相讨论
    通过多轮讨论，逐步达成共识
    """
    
    def reach_consensus(self, topic, rounds=3):
        """达成共识
        
        参数:
            topic: 讨论话题
            rounds: 讨论轮数
        返回:
            共识结论
        """
        print(f"\n🔄 开始共识流程（共{rounds}轮）\n")
        
        # 初始观点
        opinions = [agent.think(topic) for agent in self.agents]
        
        for round_num in range(1, rounds + 1):
            print(f"--- 第{round_num}轮 ---")
            
            # 每轮：每个Agent评论其他人的观点
            new_opinions = []
            for i, agent in enumerate(self.agents):
                other_opinions = [op for j, op in enumerate(opinions) if j != i]
                print(f"[{agent.name}] 评论其他人的观点...")
                
                # 模拟：综合其他人观点形成新观点
                new_opinion = f"[{agent.name}] 综合考虑后认为：{topic}需要平衡技术可行性和商业价值。"
                new_opinions.append(new_opinion)
                print(f"  → {new_opinion}\n")
            
            opinions = new_opinions
        
        # 最终共识
        consensus = f"\n✅ 达成共识：{topic}需要在技术、商业、用户体验三方面找到平衡点。"
        print(consensus)
        return consensus

# 测试
if __name__ == "__main__":
    consensus_hivemind = ConsensusHiveMind()
    consensus_hivemind.reach_consensus("AI产品的定价策略")

6.5 适用场景

• 战略决策：没有标准答案，需要多方权衡
• 产品规划：需要技术、商业、用户多角度审视
• 创意头脑风暴：激发不同角度的想法

6.6 避坑指南

坑1：讨论无果而终

大家各执己见，永远达不成共识。解决方案：设置最大讨论轮数，超时后强制投票。

坑2：群体思维

有人主导讨论，其他人不敢发言。解决方案：匿名提交观点。

坑3：投票不公

某些Agent的投票权重过高。解决方案：平等投票，或根据相关性加权。

6.7 小结

HiveMind适合需要多角度分析的创意/决策场景。通过去中心化的讨论和投票，得出更全面的结论。

6.8 练习题

基础题：添加一个”法律专家”Agent，从法律合规角度分析。

进阶题：实现更复杂的投票机制——Agent可以根据可信度加权投票。

挑战题：实现多轮讨论+观点演变记录，追踪每轮的共识变化。

七、实战：多Agent客服系统

7.1 需求分析

构建一个多Agent客服系统，能处理：

• 技术问题（代码、bug、功能）
• 商务问题（价格、购买、合同）
• 一般问题（使用帮助、账户）

7.2 系统架构

• 协调者：接收消息，识别意图
• 技术Agent：处理技术问题
• 销售Agent：处理商务问题
• 客服Agent：处理一般问题
• 记忆系统：保存对话历史

7.3 完整代码

import re
from datetime import datetime

class Agent:
    """客服Agent基类"""
    def __init__(self, name, keywords):
        self.name = name
        self.keywords = keywords  # 识别关键词
    
    def can_handle(self, message):
        """判断是否能处理这个消息"""
        return any(kw in message for kw in self.keywords)
    
    def handle(self, message):
        """处理消息"""
        return f"[{self.name}] 您好，关于{message[:20]}，我来帮您处理。"

class Memory:
    """对话记忆系统"""
    def __init__(self):
        self.history = []
    
    def add(self, role, content):
        """添加对话记录"""
        self.history.append({
            "role": role,
            "time": datetime.now(),
            "content": content
        })
    
    def get_recent(self, limit=5):
        """获取最近N条记录"""
        return self.history[-limit:]

class CustomerServiceSystem:
    """多Agent客服系统
    
    采用Orchestra模式
    协调者负责任务分配
    """
    
    def __init__(self):
        # 创建Agent池
        self.agents = {
            "tech": Agent("技术支持", ["代码", "bug", "报错", "技术", "如何实现"]),
            "sales": Agent("销售顾问", ["价格", "多少钱", "购买", "报价", "收费"]),
            "service": Agent("在线客服", ["帮助", "怎么", "如何", "问题", "使用"])
        }
        
        # 默认Agent
        self.default_agent = self.agents["service"]
        
        # 对话记忆
        self.memory = Memory()
    
    def route(self, message):
        """智能路由：选择合适的Agent"""
        for name, agent in self.agents.items():
            if agent.can_handle(message):
                return agent
        return self.default_agent
    
    def chat(self, user_message):
        """处理对话
        
        参数:
            user_message: 用户消息
        返回:
            系统回复
        """
        print(f"\n👤 用户: {user_message}")
        
        # 1. 记录用户消息
        self.memory.add("user", user_message)
        
        # 2. 路由选择Agent
        agent = self.route(user_message)
        print(f"📢 分配给: {agent.name}")
        
        # 3. Agent处理
        response = agent.handle(user_message)
        print(f"🤖 {agent.name}: {response}")
        
        # 4. 记录回复
        self.memory.add("assistant", response)
        
        return response
    
    def show_history(self):
        """显示对话历史"""
        print("\n📜 最近对话：")
        for item in self.memory.get_recent(4):
            marker = "👤" if item["role"] == "user" else "🤖"
            print(f"  {marker} {item['content'][:40]}...")

# 测试
if __name__ == "__main__":
    cs = CustomerServiceSystem()
    
    # 模拟对话
    dialogues = [
        "代码运行报错了怎么办？",
        "你们的价格是多少？",
        "怎么开通账户？",
        "这个功能怎么使用？"
    ]
    
    print("=" * 50)
    print("多Agent客服系统演示")
    print("=" * 50)
    
    for msg in dialogues:
        cs.chat(msg)
    
    cs.show_history()

7.4 运行结果

$ python customer_service.py

==================================================
多Agent客服系统演示
==================================================

👤 用户: 代码运行报错了怎么办？
📢 分配给: 技术支持
🤖 技术支持: [技术支持] 您好，关于代码运行报错了...

👤 用户: 你们的价格是多少？
📢 分配给: 销售顾问
🤖 销售顾问: [销售顾问] 您好，关于你们的价格是...

👤 用户: 怎么开通账户？
📢 分配给: 在线客服
🤖 在线客服: [在线客服] 您好，关于怎么开通账户...

👤 用户: 这个功能怎么使用？
📢 分配给: 在线客服
🤖 在线客服: [在线客服] 您好，关于这个功能怎么...

📜 最近对话：
  👤 代码运行报错了怎么办？...
  🤖 [技术支持] 您好...
  👤 你们的价格是多少？...
  🤖 [销售顾问] 您好...

7.5 扩展方向

• 接入真实LLM：用大模型生成自然语言回复
• 转人工机制：Agent无法处理时转人工
• 情感识别：识别用户情绪调整策略
• 知识库集成：结合RAG回答专业问题

八、总结与练习

8.1 要点回顾

• Orchestra：有协调者，任务分配简单直接
• Hierarchical：层级治理，Manager→Supervisor→Worker
• HiveMind：去中心化，投票/共识决策
• 核心问题：任务分配、信息共享、冲突解决、结果汇总

8.2 模式对比

8.3 选型指南

8.4 延伸阅读

• LangChain Agents：提供了多Agent协作的封装
• AutoGen：微软的多Agent对话框架，支持Agent间协作
• CrewAI：专注于多Agent协作的框架

8.5 课后练习

基础题：为Orchestra添加”设计师”Agent。

进阶题：实现Hierarchical的任务转移机制。

挑战题：设计一个多Agent项目管理团队（项目经理+开发+测试+产品）。