多Agent协作系统：从单Agent到Agent Swarm

单Agent能力有限，复杂任务需要多个Agent协作完成。本文将介绍多Agent系统的设计思想、主流框架和实战代码。

一、为什么需要多Agent？

单Agent面临的问题：

• 任务过于复杂，一个Agent难以兼顾所有方面
• 角色冲突，同一个Agent既要写代码又要做测试
• 上下文过载，长任务导致记忆混乱

多Agent的优势：

• 专业分工：每个Agent负责一个领域
• 并行处理：多任务同时进行
• 质量检查：Agent之间互相审核
• 容错能力：单个Agent失败不影响整体

二、多Agent协作模式

模式 说明 | 示例 ------------|------顺序协作Agent A -> Agent B -> Agent C | 需求分析 -> 编码 -> 测试层级管理管理者分配任务给执行者 | 项目经理 + 程序员讨论协商多个Agent讨论达成共识 | 辩论式决策竞争投票多个Agent生成方案，投票选择 | 代码生成 + 评审Swarm去中心化，Agent自组织协作 | 蚁群算法式任务分配

三、CrewAI 实战

CrewAI 是基于角色的多Agent框架，适合复杂任务分解。

安装

pip install crewai crewai-tools

基础示例：内容创作团队

from crewai import Agent, Task, Crew, Processfrom langchain_openai import ChatOpenAI

定义大模型

llm = ChatOpenAI(model="gpt-4", temperature=0.7)

1. 定义Agent角色

researcher = Agent( role="研究员", goal="收集和分析最新的AI技术趋势", backstory="你是一位资深技术研究员，擅长追踪和分析前沿技术动态。", llm=llm, verbose=True)

writer = Agent( role="技术写手", goal="将研究成果转化为高质量的技术文章", backstory="你是一位经验丰富的技术写手，擅长将复杂技术概念通俗易懂地表达。", llm=llm, verbose=True)

editor = Agent( role="编辑", goal="审核文章质量，确保内容准确、逻辑清晰", backstory="你是一位严格的编辑，关注文章结构、语法和事实准确性。", llm=llm, verbose=True)

2. 定义任务

research_task = Task( description="调研2025年AI Agent领域的最新发展趋势，列出3个最重要的方向。", expected_output="一份结构化的研究报告，包含趋势分析和数据支持。", agent=researcher)

writing_task = Task( description="根据研究员的报告，撰写一篇2000字的技术文章。", expected_output="一篇完整的Markdown格式文章，包含标题、小标题和代码示例。", agent=writer)

editing_task = Task( description="审核技术文章，修正语法错误和逻辑问题，给出修改建议。", expected_output="修改后的文章和修改意见列表。", agent=editor)

3. 组建Crew并执行

crew = Crew( agents=[researcher, writer, editor], tasks=[research_task, writing_task, editing_task], process=Process.sequential, # 顺序执行 verbose=True)

result = crew.kickoff()print(result)

进阶：并行任务与层级管理

from crewai import Crew, Process

并行执行多个独立任务

crew = Crew( agents=[researcher, writer, editor, designer], tasks=[task1, task2, task3, task4], process=Process.parallel, # 并行执行 memory=True # 启用共享记忆)

层级管理：Manager 分配任务

crew = Crew( agents=[researcher, writer, editor], tasks=[task1, task2], process=Process.hierarchical, # 层级管理 manager_llm=ChatOpenAI(model="gpt-4"), # 管理者使用更强的模型 function_calling_llm=ChatOpenAI(model="gpt-3.5-turbo"))

四、AutoGen 实战

AutoGen 是微软开源的多Agent对话框架，核心是Agent之间的消息传递。

安装

pip install pyautogen

基础示例：代码生成与调试

from autogen import ConversableAgent, GroupChat, GroupChatManager

配置LLM

config_list = [{ "model": "gpt-4", "api_key": "your-api-key"}]

创建Agent

coder = ConversableAgent( name="coder", system_message="你是一个Python程序员。编写简洁、高效的代码。", llm_config={"config_list": config_list})

tester = ConversableAgent( name="tester", system_message="你是一个代码测试员。检查代码的正确性和边界情况。", llm_config={"config_list": config_list})

reviewer = ConversableAgent( name="reviewer", system_message="你是一个代码审查员。关注代码风格、可读性和性能。", llm_config={"config_list": config_list})

群聊协作

group_chat = GroupChat( agents=[coder, tester, reviewer], messages=[], max_round=10)

manager = GroupChatManager(groupchat=group_chat, llm_config={"config_list": config_list})

启动对话

coder.initiate_chat( manager, message="请写一个Python函数，实现快速排序算法。coder先写代码，tester和reviewer随后审查。")

自定义对话终止条件

def custom_termination(message): return "APPROVED" in message.get("content", "")

使用自定义终止

group_chat = GroupChat( agents=[coder, tester, reviewer], messages=[], max_round=10, speaker_selection_method="round_robin" # 轮流发言)

五、自研轻量多Agent框架

如果不想依赖第三方框架，可以用简单Python实现：

class MultiAgentSystem: def __init__(self): self.agents = {} self.message_queue = [] def register(self, name, agent, role): self.agents[name] = {"agent": agent, "role": role, "inbox": []} def send(self, from_agent, to_agent, message): """发送消息""" self.agents[to_agent]["inbox"].append({ "from": from_agent, "content": message }) def broadcast(self, from_agent, message): """广播消息""" for name in self.agents: if name != from_agent: self.send(from_agent, name, message) def run(self, task, max_rounds=10): """运行多Agent协作""" # 初始化：广播任务 self.broadcast("system", task) for round_num in range(max_rounds): for name, data in self.agents.items(): if data["inbox"]: messages = data["inbox"] data["inbox"] = [] # Agent处理消息并回复 response = data["agent"].process(messages) print(f"[{name}] {response}") # 根据角色决定回复对象 if data["role"] == "worker": self.send(name, "manager", response) elif data["role"] == "manager": self.broadcast(name, response) return "协作完成"

使用

system = MultiAgentSystem()system.register("manager", manager_agent, "manager")system.register("coder", coder_agent, "worker")system.register("tester", tester_agent, "worker")system.run("实现一个用户登录系统")

六、多Agent设计最佳实践

1. 角色定义清晰

每个Agent应该有明确的职责边界，避免角色重叠。

# 角色模糊researcher = Agent(role="全能助手", ...)

角色清晰

researcher = Agent(role="数据分析师", goal="分析数据趋势", ...)writer = Agent(role="报告撰写员", goal="撰写分析报告", ...)

2. 控制对话轮次

多Agent对话容易无限循环，必须设置终止条件。

# 设置最大轮次group_chat = GroupChat(agents=..., max_round=10)

设置终止条件

def should_stop(messages): return any("FINAL ANSWER" in m["content"] for m in messages)

3. 成本管理

多Agent意味着多倍的Token消耗：

• 使用较便宜的模型处理简单任务
• 设置上下文长度限制
• 启用缓存避免重复计算

七、总结

多Agent系统是解决复杂任务的强大范式。CrewAI适合角色明确的流水线任务，AutoGen适合开放式对话协作。选择框架时要考虑任务特性、成本和可维护性。

---

下一篇将讨论Agent 的安全与伦理问题，包括幻觉、提示注入等风险及防护策略。