章节2：多智能体系统

本章导学

单个Agent的能力始终有限——就像一个人不可能精通所有领域。多智能体系统（Multi-Agent System, MAS）通过让多个专业化Agent协作、分工、通信，实现"1+1>2"的效果。本章将系统讲解多Agent的协作模式、角色分工策略、通信协调机制，并深度分析AutoGPT、BabyAGI的原理以及AutoGen、CrewAI等主流框架的实战应用。

---

学习目标

• 理解多Agent协作的四种主要模式
• 掌握角色分工与任务分解的策略
• 了解Agent通信与协调的核心机制
• 理解AutoGPT和BabyAGI的设计原理
• 能够使用AutoGen和CrewAI搭建多Agent系统

---

2.1 多Agent协作模式

2.1.1 为什么需要多Agent

单Agent的限制	多Agent的优势
单一角色，能力有限	专业化分工，各司其职
缺少制衡，易产生幻觉	互相审核，提升准确性
单点故障	容错性强
无法处理多视角问题	多角度分析，结论更全面

2.1.2 四种核心协作模式

模式1：顺序协作（Sequential）

Agent按固定顺序流水线执行，前一个的输出作为后一个的输入。

用户输入 → Agent A(分析) → Agent B(搜索) → Agent C(撰写) → 最终输出

适合场景：有明确流程的任务，如审批流水线、文档处理管线。

class SequentialAgentPipeline:
    """顺序执行的多Agent流水线"""
    
    def __init__(self):
        self.agents = []
    
    def add_agent(self, name: str, role: str, system_prompt: str):
        self.agents.append({
            "name": name,
            "role": role,
            "system_prompt": system_prompt
        })
    
    def run(self, user_input: str) -> str:
        current_input = user_input
        for agent in self.agents:
            print(f"\n【{agent['name']}】（角色：{agent['role']}）处理中...")
            current_input = f"[{agent['name']}处理结果] 基于'{current_input}'的分析"
        return current_input

# 使用
pipeline = SequentialAgentPipeline()
pipeline.add_agent("分析员", "需求分析师", "分析用户需求")
pipeline.add_agent("研究员", "信息研究员", "搜索相关信息")
pipeline.add_agent("写作员", "报告撰写员", "撰写最终报告")
result = pipeline.run("请分析AI Agent市场趋势")
print(f"最终结果: {result}")

模式2：层级协作（Hierarchical）

一个管理者Agent负责任务分解和调度，多个工作者Agent执行具体任务。

             ┌─────────────┐
             │  管理者Agent  │（决策、分配、协调）
             └──────┬──────┘
                    │
       ┌────────────┼────────────┐
       │            │            │
   ┌───┴───┐   ┌───┴───┐   ┌───┴───┐
   │ Worker1│   │ Worker2│   │ Worker3│
   │ (搜索) │   │ (分析) │   │ (写作) │
   └───────┘   └───────┘   └───────┘

适合场景：复杂项目管理、研发团队模拟。

class HierarchicalAgentSystem:
    """层级协作多Agent系统"""
    
    class ManagerAgent:
        """管理者Agent：负责任务分解和分配"""
        def assign_task(self, objective: str, workers: list) -> list:
            print(f"管理：分析目标「{objective}」")
            assignments = []
            for i, worker in enumerate(workers):
                assignments.append({
                    "worker": worker["name"],
                    "task": f"子任务{i+1}: {worker['role']}相关处理",
                    "context": objective
                })
            return assignments
        
        def aggregate_results(self, results: list) -> str:
            summary = "## 最终汇总报告\n\n"
            for r in results:
                summary += f"### {r['worker']} 的贡献\n{r['result']}\n\n"
            return summary
    
    class WorkerAgent:
        """工作者Agent：执行具体任务"""
        def __init__(self, name: str, role: str):
            self.name = name
            self.role = role
        
        def execute(self, task: dict) -> dict:
            print(f"  {self.name}（{self.role}）：执行任务...")
            return {
                "worker": self.name,
                "result": f"[{self.role}处理结果] 基于'{task['context']}'的分析输出"
            }
    
    def run(self, objective: str):
        manager = self.ManagerAgent()
        workers = [
            self.WorkerAgent("搜索专家", "信息检索"),
            self.WorkerAgent("数据分析师", "数据挖掘"),
            self.WorkerAgent("报告撰写员", "内容生成")
        ]
        
        assignments = manager.assign_task(objective, [
            {"name": w.name, "role": w.role} for w in workers
        ])
        
        results = []
        for worker, assignment in zip(workers, assignments):
            result = worker.execute(assignment)
            results.append(result)
        
        return manager.aggregate_results(results)

# 使用
system = HierarchicalAgentSystem()
result = system.run("2025年AI趋势分析")
print(result)

模式3：投票协作（Voting）

多个Agent对同一问题给出答案，通过投票或评分选出最佳结果。

class VotingMultiAgent:
    """投票式多Agent协作"""
    
    def __init__(self, num_agents: int = 3):
        self.num_agents = num_agents
    
    def generate_answers(self, question: str) -> list:
        answers = []
        for i in range(self.num_agents):
            answer = f"Agent{i+1}的回答：关于「{question}」的详细分析..."
            answers.append({"agent": f"Agent{i+1}", "answer": answer})
        return answers
    
    def vote(self, answers: list) -> str:
        print("投票环节：评估各Agent的回答")
        best = max(answers, key=lambda x: len(x["answer"]))
        print(f"选中：{best['agent']}")
        return best["answer"]
    
    def run(self, question: str) -> str:
        answers = self.generate_answers(question)
        return self.vote(answers)

# 使用
voting = VotingMultiAgent(3)
result = voting.run("这段代码有bug吗？")

模式4：辩论协作（Debate）

Agent之间通过辩论和讨论达成共识。

class DebateMultiAgent:
    """辩论式多Agent协作"""
    
    def __init__(self):
        self.agents = [
            {"name": "乐观派", "bias": "重点分析机遇和优势"},
            {"name": "悲观派", "bias": "重点分析风险和挑战"},
            {"name": "中立派", "bias": "客观综合评估"}
        ]
    
    def run_debate(self, topic: str, rounds: int = 3) -> str:
        statements = {}
        
        for r in range(rounds):
            print(f"\n--- 辩论第{r+1}轮 ---")
            for agent in self.agents:
                if agent["name"] not in statements:
                    statements[agent["name"]] = f"[{agent['name']}] 关于「{topic}」，我的立场是{agent['bias']}。初步分析如下..."
                else:
                    others = [s for n, s in statements.items() if n != agent["name"]]
                    statements[agent["name"]] = f"[{agent['name']}] 回复：我同意部分观点，但需要补充{agent['bias']}..."
                print(f"  {statements[agent['name']]}")
        
        return self._synthesize(statements, topic)
    
    def _synthesize(self, statements: dict, topic: str) -> str:
        output = f"\n## 辩论结论：{topic}\n\n"
        output += "### 各方观点\n"
        for n, s in statements.items():
            output += f"- {n}: {s[:50]}...\n"
        output += "\n### 综合结论\n经过多轮辩论讨论后的最终结论..."
        return output

# 使用
debate = DebateMultiAgent()
result = debate.run_debate("企业是否应该全面采用AI Agent替代客服人员？", rounds=2)
print(result)

2.1.3 协作模式选择指南

协作模式	适合场景	优点	缺点
顺序	有明确流水线的任务	简单可控	单点故障
层级	复杂项目管理	结构清晰	管理者可能成为瓶颈
投票	需要质量把关	结果相对可靠	成本高
辩论	需要深度分析	结论全面	耗时长

---

2.2 角色分工与任务分解策略

2.2.1 角色设计原则

原则	说明	示例
单一职责	每个Agent专注一个领域	搜索Agent只负责搜索
明确边界	清晰定义职责范围	不越界处理
互补性	角色之间互补而非重叠	分析师+写作员+审查员
可协作	角色间有清晰接口	输出格式标准化

2.2.2 常用角色模板

class RoleTemplates:
    """预定义角色模板"""
    
    @staticmethod
    def get_researcher_role() -> dict:
        return {
            "name": "研究员",
            "system_prompt": """你是专业研究员，擅长：
- 深入搜索和收集信息
- 验证信息的准确性
- 整理和标注信息来源
- 提取关键发现和洞察""",
            "tools": ["search_web", "read_document", "summarize"]
        }
    
    @staticmethod
    def get_analyst_role() -> dict:
        return {
            "name": "分析师",
            "system_prompt": """你是数据分析专家，擅长：
- 分析数据和信息的模式
- 发现趋势和关联
- 生成数据可视化建议
- 提供数据驱动的洞察""",
            "tools": ["analyze_data", "create_chart", "statistical_analysis"]
        }
    
    @staticmethod
    def get_writer_role() -> dict:
        return {
            "name": "撰稿人",
            "system_prompt": """你是专业内容创作者，擅长：
- 将复杂信息转化为易懂内容
- 撰写结构清晰、语言优美的文档
- 根据不同受众调整写作风格
- 确保内容的准确性和连贯性""",
            "tools": ["write_document", "format_text", "grammar_check"]
        }
    
    @staticmethod
    def get_reviewer_role() -> dict:
        return {
            "name": "审查员",
            "system_prompt": """你是质量审查专家，擅长：
- 检查内容的准确性和完整性
- 发现逻辑漏洞和错误
- 提供具体的改进建议
- 把控最终输出质量""",
            "tools": ["fact_check", "consistency_check", "quality_score"]
        }

2.2.3 任务分解策略

策略1：自上而下（Top-Down）

总体目标
    ├── 子目标1
    │   ├── 任务1.1
    │   └── 任务1.2
    ├── 子目标2
    └── 子目标3

策略2：自下而上（Bottom-Up）

先列举所有需要完成的具体任务，再组合成更大的模块。

策略3：基于依赖关系

识别任务之间的依赖关系，按拓扑排序执行。

class TaskDecomposer:
    """任务分解器"""
    
    def decompose(self, objective: str, strategy: str = "top_down") -> list:
        if strategy == "top_down":
            return self._top_down_decompose(objective)
        elif strategy == "bottom_up":
            return self._bottom_up_decompose(objective)
        else:
            return self._dependency_decompose(objective)
    
    def _top_down_decompose(self, objective: str) -> list:
        tasks = [
            {"id": 1, "name": "信息收集", "subtasks": [
                {"id": "1.1", "name": "搜索资料"},
                {"id": "1.2", "name": "整理文献"}
            ]},
            {"id": 2, "name": "数据分析", "subtasks": [
                {"id": "2.1", "name": "数据清洗"},
                {"id": "2.2", "name": "趋势分析"}
            ]},
            {"id": 3, "name": "报告撰写", "subtasks": [
                {"id": "3.1", "name": "大纲编写"},
                {"id": "3.2", "name": "正文撰写"},
                {"id": "3.3", "name": "审查修订"}
            ]}
        ]
        return tasks

# 使用
decomposer = TaskDecomposer()
tasks = decomposer.decompose("生成市场分析报告", "top_down")

---

2.3 Agent通信与协调机制

2.3.1 通信模式

模式1：消息传递（Message Passing）

from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import uuid

@dataclass
class Message:
    """Agent消息体"""
    sender: str
    receiver: str
    content: str
    msg_type: str
    msg_id: str = None
    timestamp: str = None
    
    def __post_init__(self):
        self.msg_id = self.msg_id or str(uuid.uuid4())[:8]
        self.timestamp = self.timestamp or datetime.now().isoformat()

class MessageBus:
    """消息总线：负责Agent之间的消息路由"""
    
    def __init__(self):
        self.inboxes = {}
        self.message_log = []
    
    def register(self, agent_name: str):
        if agent_name not in self.inboxes:
            self.inboxes[agent_name] = []
    
    def send(self, message: Message):
        if message.receiver in self.inboxes:
            self.inboxes[message.receiver].append(message)
        self.message_log.append(message)
        print(f"📨 [{message.timestamp[11:19]}] {message.sender} -> {message.receiver}: {message.content[:50]}...")
    
    def receive(self, agent_name: str) -> Optional[Message]:
        if agent_name in self.inboxes and self.inboxes[agent_name]:
            return self.inboxes[agent_name].pop(0)
        return None
    
    def broadcast(self, sender: str, content: str, receivers: list):
        for receiver in receivers:
            if receiver != sender:
                self.send(Message(sender, receiver, content, "broadcast"))

# 使用
bus = MessageBus()
bus.register("研究员")
bus.register("分析师")
bus.register("撰稿人")

bus.send(Message("研究员", "分析师", "已收集到2025年AI市场数据", "response"))
bus.send(Message("分析师", "撰稿人", "分析完成，核心结论：市场规模年增长35%", "request"))

模式2：会话管理（Session Management）

from datetime import datetime
import uuid

class SessionManager:
    """会话管理器"""
    
    def __init__(self):
        self.sessions = {}
    
    def create_session(self, objective: str) -> str:
        session_id = str(uuid.uuid4())[:8]
        self.sessions[session_id] = {
            "objective": objective,
            "messages": [],
            "participants": [],
            "status": "active",
            "artifacts": {}
        }
        return session_id
    
    def add_participant(self, session_id: str, agent_name: str):
        if session_id in self.sessions:
            self.sessions[session_id]["participants"].append(agent_name)
    
    def add_message(self, session_id: str, sender: str, content: str):
        if session_id in self.sessions:
            self.sessions[session_id]["messages"].append({
                "sender": sender,
                "content": content,
                "timestamp": datetime.now().isoformat()
            })
    
    def get_context(self, session_id: str) -> str:
        session = self.sessions.get(session_id, {})
        context = f"会话目标：{session.get('objective', '未知')}\n\n"
        context += "历史对话：\n"
        for msg in session.get("messages", [])[-10:]:
            context += f"{msg['sender']}: {msg['content']}\n"
        return context

2.3.2 协调机制

机制	说明	实现方式
集中式协调	一个中央协调Agent管理所有通信	管理/工作者模式
分布式协调	Agent之间直接通信	消息总线、P2P
黑板模式	共享一个"黑板"，Agent在上面读写	共享数据库/文档
合同网协议	Agent竞标任务	招标/投标机制

---

2.4 AutoGPT / BabyAGI原理分析

2.4.1 AutoGPT

AutoGPT 是2023年最火爆的开源项目之一，它是一个全自主的AI Agent——只需给定一个目标，它会自动生成任务、执行、评估、再生成新任务，循环直到目标完成。

核心架构

目标输入
    │
    ▼
┌──────────────┐
│  任务生成器   │ <- LLM根据目标和新信息生成任务列表
└──────┬───────┘
       │
       ▼
┌──────────────┐
│  任务优先级队列 │ <- 对任务排序（可持久化）
└──────┬───────┘
       │
       ▼
┌──────────────┐
│  任务执行引擎  │ <- 调用工具执行当前任务
└──────┬───────┘
       │
       ▼
┌──────────────┐
│  记忆存储     │ <- 保存执行结果到长期记忆
└──────┬───────┘
       │
       ▼ (循环)
┌──────────────┐
│  结果评估     │ <- 判断目标是否完成
└──────┬───────┘
       │ 完成/未完成
       ▼
    结束或继续

简化实现

class SimpleAutoGPT:
    """简化版AutoGPT实现"""
    
    def __init__(self, objective: str, model="gpt-4"):
        self.objective = objective
        self.model = model
        self.tasks = []
        self.completed_tasks = []
        self.memory = []
    
    def generate_tasks(self, last_result: str = "") -> list:
        prompt = f"""目标：{self.objective}
已完成任务：{self.completed_tasks[-3:] if self.completed_tasks else '无'}
最新结果：{last_result or '初始状态'}
请列出接下来需要完成的具体任务（最多5个）。"""
        tasks = [
            f"任务{i+1}：执行目标相关的子任务第{i+1}步" 
            for i in range(3)
        ]
        return tasks
    
    def execute_task(self, task: str) -> str:
        print(f"  执行：{task}")
        return f"[完成] {task} - 执行成功"
    
    def evaluate_progress(self) -> bool:
        return len(self.completed_tasks) >= 5
    
    def run(self, max_iterations: int = 10):
        print(f"目标：{self.objective}\n")
        
        for i in range(max_iterations):
            print(f"\n--- 迭代 {i+1} ---")
            last_result = self.completed_tasks[-1] if self.completed_tasks else ""
            new_tasks = self.generate_tasks(last_result)
            self.tasks.extend(new_tasks)
            
            if self.tasks:
                current_task = self.tasks.pop(0)
                result = self.execute_task(current_task)
                self.completed_tasks.append(result)
                self.memory.append(result)
            
            if self.evaluate_progress():
                print(f"\n目标完成！共执行{len(self.completed_tasks)}个任务")
                return
            
        print(f"\n达到最大迭代次数，共执行{len(self.completed_tasks)}个任务")

# 使用
agent = SimpleAutoGPT("分析2025年AI Agent市场趋势并生成报告")
agent.run(max_iterations=6)

2.4.2 BabyAGI

BabyAGI 由Yohei Nakajima创建，核心思想是模拟人类"学习-执行-学习"的循环。

BabyAGI vs AutoGPT

维度	AutoGPT	BabyAGI
核心机制	任务驱动的自主循环	任务生成+优先级+执行
记忆管理	使用向量数据库存储	简短的任务列表
复杂度	较高，需要API密钥和配置	较轻量，代码更简洁
定位	通用自主Agent	研究任务管理与执行
持久化	支持文件/数据库持久化	内存中运行

class SimpleBabyAGI:
    """简化版BabyAGI实现"""
    
    def __init__(self, objective: str, first_task: str):
        self.objective = objective
        self.task_list = [{"id": 1, "name": first_task, "status": "pending"}]
        self.task_id_counter = 1
        self.results = []
    
    def add_task(self, task_name: str):
        self.task_id_counter += 1
        self.task_list.append({
            "id": self.task_id_counter,
            "name": task_name,
            "status": "pending"
        })
    
    def get_priority_task(self) -> dict:
        pending = [t for t in self.task_list if t["status"] == "pending"]
        if pending:
            return pending[0]
        return None
    
    def execute_task(self, task: dict) -> str:
        print(f"  执行任务{task['id']}: {task['name']}")
        task["status"] = "completed"
        result = f"任务{task['id']}「{task['name']}」执行完成"
        self.results.append(result)
        return result
    
    def create_new_tasks(self, last_result: str) -> list:
        new_tasks = [
            f"分析「{last_result[:20]}」的详细内容",
            f"基于分析结果撰写相关文档"
        ]
        return new_tasks
    
    def run(self, max_iterations: int = 10):
        print(f"BabyAGI 目标：{self.objective}\n")
        
        for i in range(max_iterations):
            print(f"\n--- 迭代 {i+1} ---")
            task = self.get_priority_task()
            if not task:
                print("所有任务已完成！")
                break
            
            result = self.execute_task(task)
            new_tasks = self.create_new_tasks(result)
            for t in new_tasks:
                self.add_task(t)
                print(f"  新增任务: {t}")
        
        print(f"\n共执行{len(self.results)}个任务")

# 使用
baby_agi = SimpleBabyAGI(
    objective="学习Python多线程编程",
    first_task="搜索Python多线程入门教程"
)
baby_agi.run(5)

2.4.3 两者的共同设计模式

1. 持续循环：任务生成→执行→评估→再生成
2. 目标驱动：所有行动围绕初始目标展开
3. 动态规划：不预设完整计划，边走边看
4. 记忆反馈：历史结果影响后续决策

---

2.5 AutoGen / CrewAI框架实战

2.5.1 AutoGen框架

AutoGen 是微软推出的多Agent对话框架，核心特点是Agent之间通过对话协作。

核心概念

概念	说明
AssistantAgent	由LLM驱动的智能体，负责推理和生成
UserProxyAgent	代表人类用户的代理，可以执行代码和工具
GroupChat	多Agent群聊环境
GroupChatManager	群聊管理者，控制对话流程

实战代码：AutoGen多Agent对话

# pip install pyautogen

import autogen

# 1. 配置LLM
llm_config = {
    "config_list": [
        {
            "model": "gpt-4",
            "api_key": "your-api-key"
        }
    ],
    "temperature": 0.7
}

# 2. 创建Agent
analyst = autogen.AssistantAgent(
    name="分析员",
    system_message="""你是一名专业的数据分析师。
你擅长：分析数据趋势，从数据中提取洞察，提供数据驱动的建议。
回答要简洁、专业，使用数据说话。""",
    llm_config=llm_config
)

writer = autogen.AssistantAgent(
    name="写作员",
    system_message="""你是一名专业的内容撰稿人。
你擅长：将复杂信息转化为清晰易懂的文档。
确保内容结构完整、语言流畅、读者友好。""",
    llm_config=llm_config
)

reviewer = autogen.AssistantAgent(
    name="审查员",
    system_message="""你是一名严格的质量审查员。
你擅长：检查内容的准确性、完整性和一致性。
发现任何问题都要明确指出，并给出改进建议。
只有当内容完全符合标准时才说'通过'。""",
    llm_config=llm_config
)

user_proxy = autogen.UserProxyAgent(
    name="用户",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("通过"),
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False
    }
)

# 3. 创建群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, analyst, writer, reviewer],
    messages=[],
    max_round=15
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config
)

# 4. 启动对话
user_proxy.initiate_chat(
    manager,
    message="""请协作完成以下任务：
分析2025年AI Agent行业趋势，主要包括：
1. 市场规模和增长率
2. 主要应用场景
3. 关键技术突破
4. 未来展望

请分析员先分析，写作员撰写报告，审查员最后审核。"""
)

2.5.2 CrewAI框架

CrewAI 是一个更轻量级的多Agent框架，设计理念是"像管理团队一样管理Agent"。

核心概念

概念	说明
Agent	具有角色、目标、背景故事的智能体
Task	Agent需要完成的具体任务
Crew	一个由多个Agent组成的团队
Process	执行流程（顺序/层级）

实战代码：CrewAI多Agent团队

# pip install crewai

from crewai import Agent, Task, Crew, Process

# 1. 创建Agent
researcher = Agent(
    role="高级市场研究员",
    goal="深入研究AI Agent市场趋势并收集最新数据",
    backstory="你在科技市场研究领域有15年经验，擅长从海量数据中发现趋势和洞察。",
    verbose=True,
    allow_delegation=False
)

analyst = Agent(
    role="数据分析师",
    goal="分析收集到的数据并生成洞察",
    backstory="你是数据分析专家，精通统计分析和数据可视化，能将原始数据转化为有价值的洞察。",
    verbose=True,
    allow_delegation=False
)

writer = Agent(
    role="报告撰写员",
    goal="撰写结构清晰、内容专业的市场分析报告",
    backstory="你是资深技术作家，擅长将复杂的技术分析转化为易懂、有说服力的报告内容。",
    verbose=True,
    allow_delegation=False
)

# 2. 创建任务
research_task = Task(
    description="""收集2025年AI Agent市场的信息：
    - 市场规模和增长数据
    - 主要参与公司和产品
    - 技术发展趋势
    - 应用场景分布""",
    expected_output="一份包含数据来源的市场信息汇总",
    agent=researcher
)

analysis_task = Task(
    description="""分析收集到的市场数据：
    - 计算年复合增长率
    - 识别关键市场驱动因素
    - 分析竞争格局
    - 预测未来趋势""",
    expected_output="一份数据分析报告，包含关键发现和图表描述",
    agent=analyst
)

writing_task = Task(
    description="""基于分析结果撰写最终报告：
    - 执行摘要
    - 市场概况
    - 趋势分析
    - 结论与建议""",
    expected_output="一份完整的、可直接交付的市场分析报告",
    agent=writer
)

# 3. 组建团队
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    verbose=2,
    process=Process.sequential
)

# 4. 启动
result = crew.kickoff()
print("最终报告：")
print(result)

2.5.3 AutoGen vs CrewAI对比

维度	AutoGen	CrewAI
开发者	微软	社区驱动
核心机制	Agent对话协作	角色扮演+任务驱动
协作模式	群聊/双向对话	顺序/层级流程
复杂度	中高	中低
灵活性	高，可自定义对话流程	中，结构较固定
学习曲线	较陡	较平缓
适用场景	复杂推理/代码生成	结构化工作流

---

小结与练习

知识小结

1. 四种协作模式：顺序（流水线）、层级（管理+执行）、投票（聚合）、辩论（讨论）
2. 角色分工：单一职责、边界清晰、角色互补
3. 任务分解：自上而下、自下而上、依赖驱动
4. 通信机制：消息传递、会话管理、黑板模式
5. AutoGPT/BabyAGI：目标驱动的自主循环架构
6. AutoGen/CrewAI：主流多Agent框架，各有特色

练习

练习1：实现投票协作系统

实现一个由3个Agent组成的投票系统：
1. 每个Agent独立回答用户问题
2. 使用评分机制选出最佳答案
3. 输出最终答案和选择理由

练习2：层级多Agent系统

搭建一个管理-执行架构：
1. 管理者负责拆分任务
2. 至少3个不同类型的执行Agent
3. 管理者汇总结果并输出

练习3：AutoGen实战

使用AutoGen创建一个三Agent协作系统：
1. 编码Agent：写代码
2. 测试Agent：写测试用例
3. 审查Agent：审查代码质量
让它们协作完成一个简单的Python函数开发。

练习4：CrewAI团队搭建

使用CrewAI创建一个内容创作团队：
1. 研究员：收集资料
2. 撰稿人：撰写文章
3. 编辑：修改润色
让团队协作完成一篇技术博客。

---

下一章预告：章节3-企业级应用开发 - 将Agent技术落地到真实商业场景