阶段 5：多 Agent 协同与角色体系
📖 本阶段目标
- 理解多 Agent 系统的架构模式
- 掌握 Manager-Worker、专家小组等协作模式
- 学习 Agent 间的通信和协调机制
- 实践构建多角色协作的 Agent 系统
🤝 多 Agent 系统架构
为什么需要多 Agent？
单一 Agent 的局限性： - ❌ 知识和能力有限 - ❌ 难以处理多领域任务 - ❌ 缺乏多角度思考 - ❌ 质量控制不足
多 Agent 系统的优势： - ✅ 专业化分工 - ✅ 并行处理能力 - ✅ 互相验证和改进 - ✅ 更好的可扩展性
🏗️ 常见多 Agent 模式
1. Manager-Worker 模式
┌──────────┐
│ Manager │
│ Agent │
└─────┬────┘
│
┌─────┴─────┐
▼ ▼
┌────────┐ ┌────────┐
│Worker 1│ │Worker 2│
└────────┘ └────────┘
适用场景：任务可以明确分解和分配
# examples/stage5/manager_worker.py
from openai import OpenAI
from typing import List, Dict
import json
client = OpenAI()
class ManagerAgent:
"""管理者 Agent - 负责任务分解和协调"""
def __init__(self):
self.workers = {}
def register_worker(self, name: str, worker):
"""注册工作者 Agent"""
self.workers[name] = worker
def decompose_task(self, task: str) -> List[Dict]:
"""分解任务"""
worker_descriptions = "\n".join([
f"- {name}: {worker.description}"
for name, worker in self.workers.items()
])
prompt = f"""
可用的工作者：
{worker_descriptions}
任务：{task}
请将任务分解为子任务，并分配给合适的工作者。
以 JSON 格式返回：
{{
"subtasks": [
{{
"id": 1,
"description": "子任务描述",
"assigned_to": "工作者名称",
"priority": "high/medium/low"
}},
...
]
}}
"""
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
result = json.loads(response.choices[0].message.content)
return result.get("subtasks", [])
def execute_task(self, task: str) -> Dict:
"""执行完整任务"""
print(f"📋 Manager: 收到任务 - {task}\n")
# 1. 分解任务
subtasks = self.decompose_task(task)
print(f"✂️ Manager: 已分解为 {len(subtasks)} 个子任务\n")
# 2. 分配并执行
results = []
for subtask in subtasks:
worker_name = subtask["assigned_to"]
if worker_name in self.workers:
print(f"👷 分配给 {worker_name}: {subtask['description']}")
worker = self.workers[worker_name]
result = worker.execute(subtask["description"])
results.append({
"subtask": subtask,
"result": result
})
print(f"✅ {worker_name} 完成\n")
# 3. 整合结果
final_result = self.synthesize_results(task, results)
return {
"task": task,
"subtasks": subtasks,
"results": results,
"final_result": final_result
}
def synthesize_results(self, original_task: str, results: List[Dict]) -> str:
"""整合各个工作者的结果"""
results_text = "\n\n".join([
f"子任务: {r['subtask']['description']}\n结果: {r['result']}"
for r in results
])
prompt = f"""
原始任务: {original_task}
各部分完成情况：
{results_text}
请将这些结果整合成一个完整、连贯的最终成果。
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
class WorkerAgent:
"""工作者 Agent 基类"""
def __init__(self, name: str, description: str, system_prompt: str):
self.name = name
self.description = description
self.system_prompt = system_prompt
def execute(self, task: str) -> str:
"""执行任务"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": task}
],
temperature=0.7
)
return response.choices[0].message.content
# 使用示例
if __name__ == "__main__":
# 创建管理者
manager = ManagerAgent()
# 创建专业工作者
researcher = WorkerAgent(
"Researcher",
"负责研究和收集信息",
"你是一个专业的研究员，擅长收集和整理信息。"
)
writer = WorkerAgent(
"Writer",
"负责撰写和编辑内容",
"你是一个专业的作家，擅长撰写清晰、引人入胜的内容。"
)
reviewer = WorkerAgent(
"Reviewer",
"负责审核和改进内容",
"你是一个专业的审核员，擅长发现问题并提出改进建议。"
)
# 注册工作者
manager.register_worker("Researcher", researcher)
manager.register_worker("Writer", writer)
manager.register_worker("Reviewer", reviewer)
# 执行任务
task = "写一篇关于 AI Agent 的博客文章"
result = manager.execute_task(task)
print("=" * 60)
print("最终成果：")
print("=" * 60)
print(result["final_result"])
2. 专家小组模式
# examples/stage5/expert_panel.py
from openai import OpenAI
from typing import List, Dict
import json
client = OpenAI()
class ExpertPanel:
"""专家小组 - 多个专家共同讨论决策"""
def __init__(self):
self.experts = {}
def add_expert(self, name: str, expertise: str, personality: str):
"""添加专家"""
self.experts[name] = {
"expertise": expertise,
"personality": personality,
"opinions": []
}
def get_expert_opinion(self, expert_name: str, topic: str,
previous_opinions: List[Dict] = None) -> str:
"""获取专家意见"""
expert = self.experts[expert_name]
system_prompt = f"""
你是 {expert_name}，一位{expert['expertise']}专家。
性格特点：{expert['personality']}
请基于你的专业知识给出意见。
"""
# 构建上下文（包括其他专家的意见）
context = f"讨论主题：{topic}\n\n"
if previous_opinions:
context += "其他专家的意见：\n"
for opinion in previous_opinions:
context += f"\n{opinion['expert']}：{opinion['content']}\n"
context += f"\n现在轮到你发表意见，你可以同意、反对或补充其他专家的观点。"
else:
context += "你是第一个发言的专家，请给出你的专业意见。"
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": context}
],
temperature=0.8
)
return response.choices[0].message.content
def discuss(self, topic: str, rounds: int = 2) -> Dict:
"""进行多轮讨论"""
print(f"🎯 讨论主题: {topic}\n")
print("=" * 60)
all_opinions = []
for round_num in range(rounds):
print(f"\n第 {round_num + 1} 轮讨论\n")
print("-" * 60)
round_opinions = []
for expert_name in self.experts.keys():
# 获取之前的意见作为上下文
previous = all_opinions if round_num > 0 else None
opinion = self.get_expert_opinion(
expert_name, topic, previous
)
opinion_dict = {
"round": round_num + 1,
"expert": expert_name,
"content": opinion
}
round_opinions.append(opinion_dict)
all_opinions.append(opinion_dict)
print(f"\n{expert_name}:")
print(opinion)
print()
# 生成总结
summary = self.generate_summary(topic, all_opinions)
return {
"topic": topic,
"rounds": rounds,
"opinions": all_opinions,
"summary": summary
}
def generate_summary(self, topic: str, opinions: List[Dict]) -> str:
"""生成讨论总结"""
opinions_text = "\n\n".join([
f"[第{op['round']}轮] {op['expert']}:\n{op['content']}"
for op in opinions
])
prompt = f"""
讨论主题: {topic}
专家意见:
{opinions_text}
请总结这次专家讨论：
1. 主要共识点
2. 主要分歧点
3. 最有价值的建议
4. 最终结论
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}],
temperature=0.5
)
return response.choices[0].message.content
# 使用示例
if __name__ == "__main__":
panel = ExpertPanel()
# 添加不同领域的专家
panel.add_expert(
"技术架构师",
"软件架构和系统设计",
"注重技术可行性和系统稳定性"
)
panel.add_expert(
"产品经理",
"产品规划和用户体验",
"注重用户需求和商业价值"
)
panel.add_expert(
"安全专家",
"信息安全和数据保护",
"注重安全风险和合规性"
)
# 开始讨论
topic = "是否应该在我们的产品中引入 AI Agent 功能？"
result = panel.discuss(topic, rounds=2)
print("\n" + "=" * 60)
print("讨论总结")
print("=" * 60)
print(result["summary"])
3. Self-Refine 模式
# examples/stage5/self_refine.py
from openai import OpenAI
client = OpenAI()
class SelfRefineAgent:
"""自我精炼 Agent - 通过多个角色迭代改进"""
def __init__(self, max_iterations=3):
self.max_iterations = max_iterations
def generate(self, task: str) -> str:
"""生成初始内容"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是一个内容创作者。"},
{"role": "user", "content": task}
],
temperature=0.7
)
return response.choices[0].message.content
def critique(self, content: str, task: str) -> str:
"""批评和评估"""
prompt = f"""
原始任务: {task}
生成的内容:
{content}
请作为一个严格的评审者，指出这个内容的问题和可以改进的地方：
1. 是否完成了任务要求？
2. 内容是否清晰、准确？
3. 结构是否合理？
4. 有哪些具体的改进建议？
请给出详细、建设性的反馈。
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是一个专业的评审者。"},
{"role": "user", "content": prompt}
],
temperature=0.5
)
return response.choices[0].message.content
def refine(self, content: str, critique: str, task: str) -> str:
"""根据反馈改进内容"""
prompt = f"""
原始任务: {task}
当前内容:
{content}
评审意见:
{critique}
请根据评审意见改进内容，生成更好的版本。
"""
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是一个内容创作者，擅长根据反馈改进作品。"},
{"role": "user", "content": prompt}
],
temperature=0.7
)
return response.choices[0].message.content
def run(self, task: str) -> Dict:
"""执行完整的自我精炼流程"""
print(f"📝 任务: {task}\n")
print("=" * 60)
# 初始生成
print("\n第 1 版本 - 初始生成")
print("-" * 60)
content = self.generate(task)
print(content)
history = [{
"version": 1,
"content": content,
"critique": None
}]
# 迭代改进
for i in range(self.max_iterations - 1):
print(f"\n{'=' * 60}")
print(f"第 {i + 2} 版本 - 评审和改进")
print("=" * 60)
# 评审
print("\n🔍 评审意见:")
print("-" * 60)
critique = self.critique(content, task)
print(critique)
# 改进
print("\n✨ 改进后的内容:")
print("-" * 60)
content = self.refine(content, critique, task)
print(content)
history.append({
"version": i + 2,
"content": content,
"critique": critique
})
return {
"task": task,
"history": history,
"final_content": content
}
# 使用示例
if __name__ == "__main__":
agent = SelfRefineAgent(max_iterations=3)
task = "写一段关于 AI Agent 应用价值的介绍（100字左右）"
result = agent.run(task)
print("\n\n" + "=" * 60)
print("最终版本")
print("=" * 60)
print(result["final_content"])
🎯 综合实践项目
多角色文档创作系统
# examples/stage5/document_creation_team.py
from openai import OpenAI
from typing import List, Dict
client = OpenAI()
class DocumentCreationTeam:
"""文档创作团队 - 多 Agent 协作"""
def __init__(self):
self.agents = {
"outliner": self.create_outliner(),
"researcher": self.create_researcher(),
"writer": self.create_writer(),
"editor": self.create_editor(),
"reviewer": self.create_reviewer()
}
def create_outliner(self):
"""创建大纲设计师"""
return {
"role": "大纲设计师",
"system_prompt": "你是一个专业的内容架构师，擅长设计文档大纲和结构。"
}
def create_researcher(self):
"""创建研究员"""
return {
"role": "研究员",
"system_prompt": "你是一个研究员，负责收集和组织相关信息。"
}
def create_writer(self):
"""创建写作者"""
return {
"role": "写作者",
"system_prompt": "你是一个专业作家，擅长将信息转化为清晰、引人入胜的文字。"
}
def create_editor(self):
"""创建编辑"""
return {
"role": "编辑",
"system_prompt": "你是一个编辑，负责优化文字表达和结构。"
}
def create_reviewer(self):
"""创建审核者"""
return {
"role": "审核者",
"system_prompt": "你是一个质量审核员，负责检查内容的准确性和完整性。"
}
def agent_execute(self, agent_key: str, task: str) -> str:
"""执行 Agent 任务"""
agent = self.agents[agent_key]
response = client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": agent["system_prompt"]},
{"role": "user", "content": task}
],
temperature=0.7
)
return response.choices[0].message.content
def create_document(self, topic: str) -> Dict:
"""完整的文档创作流程"""
print(f"📚 开始创作文档: {topic}\n")
# 1. 设计大纲
print("步骤 1: 设计大纲")
print("=" * 60)
outline = self.agent_execute(
"outliner",
f"为'{topic}'设计一个详细的文档大纲，包括主要章节和要点。"
)
print(outline)
print()
# 2. 研究收集
print("步骤 2: 研究和收集信息")
print("=" * 60)
research = self.agent_execute(
"researcher",
f"基于以下大纲，收集相关信息和要点：\n{outline}"
)
print(research)
print()
# 3. 撰写内容
print("步骤 3: 撰写文档")
print("=" * 60)
draft = self.agent_execute(
"writer",
f"基于以下研究内容撰写文档：\n大纲：{outline}\n研究内容：{research}"
)
print(draft[:500] + "...\n")
# 4. 编辑优化
print("步骤 4: 编辑优化")
print("=" * 60)
edited = self.agent_execute(
"editor",
f"优化以下文档的表达和结构：\n{draft}"
)
print(edited[:500] + "...\n")
# 5. 最终审核
print("步骤 5: 质量审核")
print("=" * 60)
review = self.agent_execute(
"reviewer",
f"审核以下文档，给出质量评估和改进建议：\n{edited}"
)
print(review)
print()
return {
"topic": topic,
"outline": outline,
"research": research,
"draft": draft,
"edited": edited,
"review": review,
"final_document": edited
}
# 使用示例
if __name__ == "__main__":
team = DocumentCreationTeam()
result = team.create_document("AI Agent 在企业中的应用")
print("\n" + "=" * 60)
print("最终文档")
print("=" * 60)
print(result["final_document"])
📋 最佳实践
1. Agent 间通信协议
class Message:
"""Agent 间消息"""
def __init__(self, sender, receiver, content, message_type="info"):
self.sender = sender
self.receiver = receiver
self.content = content
self.message_type = message_type
self.timestamp = datetime.now()
2. 协调机制
- 同步协调：按顺序执行
- 异步协调：并行处理
- 投票机制：多数决策
- 优先级调度：按重要性排序
3. 冲突解决
def resolve_conflict(opinions: List[Dict]) -> str:
"""解决 Agent 间的意见冲突"""
# 投票、加权平均、LLM 仲裁等
pass
✅ 阶段完成检查清单
- [ ] 理解多 Agent 系统的价值和架构
- [ ] 掌握 Manager-Worker 模式
- [ ] 实现专家小组讨论系统
- [ ] 了解 Self-Refine 迭代改进模式
- [ ] 完成多角色协作项目
- [ ] 理解 Agent 间通信和协调机制
- [ ] 掌握冲突解决策略
🎯 下一步
完成本阶段后，进入 阶段 6：成熟 Agent 框架实践，学习使用主流框架构建 Agent。
💡 小贴士：多 Agent 系统就像一个团队，合理的分工和协作能发挥 1+1>2 的效果。从简单的 2-3 个 Agent 开始，逐步增加复杂度。