·58 分钟
AutoGen 多智能体框架：构建协作式 AI 系统
AutoGen多智能体AI Agent协作系统LLMAI
什么是 AutoGen
AutoGen 是微软开源的多智能体框架，专注于构建多个 AI Agent 协作完成复杂任务的系统。
核心特性：
- 多智能体对话：多个 Agent 之间进行对话协作
- 灵活的对话模式：支持双人、群聊、嵌套等多种模式
- 人类参与：无缝集成人类反馈
- 代码执行：Agent 可以编写和执行代码
- 可定制性：高度可定制的 Agent 行为
环境搭建
bash
# 安装 AutoGen
pip install pyautogen
# 安装可选依赖
pip install autogen-agentchat # AgentChat 组件
pip install autogen-ext # 扩展组件
# 配置环境变量
export OPENAI_API_KEY="your-api-key"
核心概念
1. Agent（智能体）
python
from autogen import AssistantAgent, UserProxyAgent, ConversableAgent
# 助手 Agent（AI 驱动）
assistant = AssistantAgent(
name="assistant",
llm_config={
"model": "gpt-4o",
"temperature": 0.7,
"api_key": "your-api-key"
},
system_message="你是一个有用的 AI 助手。"
)
# 用户代理（人类参与）
user_proxy = UserProxyAgent(
name="user_proxy",
human_input_mode="ALWAYS", # 总是请求人类输入
code_execution_config=False # 不执行代码
)
# 可对话 Agent（自定义）
custom_agent = ConversableAgent(
name="custom",
llm_config={"model": "gpt-4o"},
system_message="你是一个专业的顾问。",
human_input_mode="NEVER", # 不请求人类输入
max_consecutive_auto_reply=10
)
2. 对话模式
python
from autogen import GroupChat, GroupChatManager
# 双人对话
assistant.initiate_chat(
user_proxy,
message="你好，有什么可以帮助你的吗？"
)
# 群聊
agents = [assistant, user_proxy, custom_agent]
group_chat = GroupChat(
agents=agents,
messages=[],
max_round=20,
speaker_selection_method="round_robin" # 轮流发言
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config={"model": "gpt-4o"}
)
# 启动群聊
user_proxy.initiate_chat(
manager,
message="我们需要讨论一下项目计划"
)
Agent 类型
1. AssistantAgent（助手）
python
from autogen import AssistantAgent
# 代码生成助手
code_assistant = AssistantAgent(
name="coder",
llm_config={
"model": "gpt-4o",
"temperature": 0
},
system_message="你是一个 Python 专家。请用 Python 编写代码来解决问题，并确保代码可以运行。"
)
# 文档写作助手
writer_assistant = AssistantAgent(
name="writer",
llm_config={"model": "gpt-4o"},
system_message="""你是一个技术写作专家。
请用清晰、简洁的语言撰写技术文档。
使用 Markdown 格式。"""
)
# 审查助手
reviewer_assistant = AssistantAgent(
name="reviewer",
llm_config={"model": "gpt-4o"},
system_message="""你是一个代码审查专家。
请检查代码的：
1. 正确性
2. 可读性
3. 性能
4. 安全性
给出具体的改进建议。"""
)
2. UserProxyAgent（用户代理）
python
from autogen import UserProxyAgent
# 人类参与模式
human_proxy = UserProxyAgent(
name="human",
human_input_mode="ALWAYS",
code_execution_config=False
)
# 自动模式（无人类参与）
auto_proxy = UserProxyAgent(
name="auto",
human_input_mode="NEVER",
code_execution_config={
"work_dir": "workspace",
"use_docker": True,
"timeout": 60,
"last_n_messages": 3
}
)
# 终止模式（特定条件下请求人类输入）
terminate_proxy = UserProxyAgent(
name="terminate",
human_input_mode="TERMINATE",
max_consecutive_auto_reply=5,
code_execution_config=False
)
3. 自定义 Agent
python
from autogen import ConversableAgent
class ResearchAgent(ConversableAgent):
"""研究助手 Agent"""
def __init__(self, name, llm_config):
super().__init__(
name=name,
llm_config=llm_config,
system_message="""你是一个研究助手。
你的任务是：
1. 分析问题
2. 搜索相关信息
3. 综合分析
4. 生成报告"""
)
self.research_results = []
def generate_reply(self, messages, sender, **kwargs):
"""自定义回复生成"""
# 分析输入
last_message = messages[-1]["content"]
# 执行研究
research = self.conduct_research(last_message)
self.research_results.append(research)
# 生成回复
reply = self.synthesize_findings(research)
return reply
def conduct_research(self, topic):
"""执行研究"""
# 这里可以集成搜索 API、数据库查询等
return f"关于 {topic} 的研究结果..."
def synthesize_findings(self, research):
"""综合研究发现"""
prompt = f"请综合以下研究发现，生成一个简洁的报告：\n{research}"
return self.llm.generate(prompt)
代码执行
1. 本地代码执行
python
from autogen import UserProxyAgent
# 配置本地代码执行
code_executor = UserProxyAgent(
name="executor",
human_input_mode="NEVER",
code_execution_config={
"work_dir": "./workspace",
"use_docker": False, # 不使用 Docker
"timeout": 120,
"last_n_messages": 3
}
)
# 让 Agent 执行代码
assistant = AssistantAgent(
name="coder",
llm_config={"model": "gpt-4o"}
)
# 启动对话
code_executor.initiate_chat(
assistant,
message="请写一个 Python 脚本，计算斐波那契数列的前 20 项"
)
2. Docker 代码执行
python
# 使用 Docker 执行代码（更安全）
code_executor = UserProxyAgent(
name="executor",
human_input_mode="NEVER",
code_execution_config={
"work_dir": "./workspace",
"use_docker": "python:3.11-slim", # 使用 Docker 镜像
"timeout": 120,
"docker_network": "bridge",
"docker_volume": "./workspace:/workspace"
}
)
3. 代码执行沙箱
python
from autogen.coding import DockerCommandLineCodeExecutor, LocalCommandLineCodeExecutor
# Docker 沙箱
docker_executor = DockerCommandLineCodeExecutor(
image="python:3.11-slim",
timeout=60,
work_dir="./workspace"
)
# 本地沙箱
local_executor = LocalCommandLineCodeExecutor(
timeout=60,
work_dir="./workspace"
)
# 使用沙箱
assistant = AssistantAgent(
name="coder",
llm_config={"model": "gpt-4o"},
code_execution_config={"executor": docker_executor}
)
对话流程控制
1. 自定义说话人选择
python
from autogen import GroupChat, GroupChatManager
def custom_speaker_selection(last_speaker, group_chat):
"""自定义说话人选择逻辑"""
messages = group_chat.messages
if len(messages) == 0:
# 第一条消息，选择主持人
return group_chat.agent_by_name("moderator")
last_message = messages[-1]["content"]
# 根据内容选择说话人
if "代码" in last_message or "实现" in last_message:
return group_chat.agent_by_name("coder")
elif "审查" in last_message or "检查" in last_message:
return group_chat.agent_by_name("reviewer")
elif "文档" in last_message:
return group_chat.agent_by_name("writer")
else:
# 默认轮询
current_idx = group_chat.agents.index(last_speaker)
next_idx = (current_idx + 1) % len(group_chat.agents)
return group_chat.agents[next_idx]
group_chat = GroupChat(
agents=[moderator, coder, reviewer, writer],
messages=[],
max_round=30,
speaker_selection_method=custom_speaker_selection
)
2. 终止条件
python
from autogen import GroupChat
# 自定义终止条件
def is_termination_met(messages):
"""检查是否满足终止条件"""
if len(messages) == 0:
return False
last_message = messages[-1]["content"].lower()
# 检查终止关键词
termination_keywords = ["完成", "结束", "终止", "done", "finished"]
for keyword in termination_keywords:
if keyword in last_message:
return True
# 检查消息数量
if len(messages) > 50:
return True
return False
group_chat = GroupChat(
agents=agents,
messages=[],
max_round=100,
speaker_selection_method="auto",
is_termination_msg=is_termination_met
)
3. 嵌套对话
python
from autogen import GroupChat, GroupChatManager
# 创建子群组
inner_group = GroupChat(
agents=[coder, reviewer],
messages=[],
max_round=10
)
inner_manager = GroupChatManager(
groupchat=inner_group,
name="dev_team",
llm_config={"model": "gpt-4o"}
)
# 主群组包含子群组的管理器
outer_group = GroupChat(
agents=[project_manager, inner_manager, writer],
messages=[],
max_round=20
)
outer_manager = GroupChatManager(
groupchat=outer_group,
llm_config={"model": "gpt-4o"}
)
# 启动嵌套对话
project_manager.initiate_chat(
outer_manager,
message="我们需要开发一个新功能"
)
工具集成
1. 函数注册
python
from autogen import register_function
# 定义工具函数
def search_web(query: str) -> str:
"""搜索互联网"""
# 实际实现
return f"搜索结果：{query}"
def read_file(file_path: str) -> str:
"""读取文件"""
with open(file_path, 'r') as f:
return f.read()
def write_file(file_path: str, content: str) -> str:
"""写入文件"""
with open(file_path, 'w') as f:
f.write(content)
return f"文件 {file_path} 已写入"
# 注册工具
register_function(
search_web,
caller=assistant,
executor=code_executor,
name="search_web",
description="搜索互联网获取信息"
)
register_function(
read_file,
caller=assistant,
executor=code_executor,
name="read_file",
description="读取文件内容"
)
2. 工具调用
python
from autogen import AssistantAgent, UserProxyAgent
# 创建 Agent
assistant = AssistantAgent(
name="assistant",
llm_config={
"model": "gpt-4o",
"tools": [
{
"type": "function",
"function": {
"name": "search_web",
"description": "搜索互联网获取信息",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
}
}
},
{
"type": "function",
"function": {
"name": "read_file",
"description": "读取文件内容",
"parameters": {
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "文件路径"
}
},
"required": ["file_path"]
}
}
}
]
}
)
executor = UserProxyAgent(
name="executor",
human_input_mode="NEVER",
code_execution_config=False
)
# 注册函数
register_function(search_web, caller=assistant, executor=executor)
register_function(read_file, caller=assistant, executor=executor)
实战案例
1. 软件开发团队
python
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager
class SoftwareDevTeam:
"""软件开发团队"""
def __init__(self):
self.agents = self.create_agents()
self.group_chat = self.create_group_chat()
def create_agents(self):
"""创建团队成员"""
# 产品经理
pm = AssistantAgent(
name="product_manager",
llm_config={"model": "gpt-4o"},
system_message="""你是产品经理。
负责需求分析和项目规划。
使用用户故事格式描述需求。"""
)
# 架构师
architect = AssistantAgent(
name="architect",
llm_config={"model": "gpt-4o"},
system_message="""你是软件架构师。
负责系统设计和技术选型。
给出清晰的架构图和组件说明。"""
)
# 开发者
developer = AssistantAgent(
name="developer",
llm_config={"model": "gpt-4o"},
system_message="""你是高级开发者。
负责代码实现。
编写高质量、可维护的代码。"""
)
# 测试工程师
tester = AssistantAgent(
name="tester",
llm_config={"model": "gpt-4o"},
system_message="""你是测试工程师。
负责测试策略和质量保证。
编写测试用例和测试报告。"""
)
# 用户代理（项目经理）
manager = UserProxyAgent(
name="project_manager",
human_input_mode="ALWAYS",
code_execution_config=False
)
return [pm, architect, developer, tester, manager]
def create_group_chat(self):
"""创建群聊"""
def check_termination(x):
return "项目完成" in x.get("content", "").lower()
group_chat = GroupChat(
agents=self.agents,
messages=[],
max_round=50,
speaker_selection_method="auto",
is_termination_msg=check_termination
)
return GroupChatManager(
groupchat=group_chat,
llm_config={"model": "gpt-4o"}
)
def develop_feature(self, feature_description: str):
"""开发新功能"""
return self.group_chat.initiate_chat(
self.agents[-1], # 项目经理发起
message=f"我们需要开发以下功能：{feature_description}"
)
# 使用示例
team = SoftwareDevTeam()
team.develop_feature("用户登录系统，支持邮箱和手机号登录")
2. 研究助手系统
python
from autogen import AssistantAgent, UserProxyAgent, GroupChat
class ResearchSystem:
"""研究助手系统"""
def __init__(self):
self.agents = self.create_agents()
def create_agents(self):
"""创建研究团队"""
# 研究主管
director = AssistantAgent(
name="research_director",
llm_config={"model": "gpt-4o"},
system_message="""你是研究主管。
负责研究方向把控和进度管理。
确保研究质量和创新性。"""
)
# 文献研究员
literature_researcher = AssistantAgent(
name="literature_researcher",
llm_config={"model": "gpt-4o"},
system_message="""你是文献研究专家。
负责文献搜索、阅读和综述。
提供全面的文献分析。"""
)
# 数据分析师
data_analyst = AssistantAgent(
name="data_analyst",
llm_config={"model": "gpt-4o"},
system_message="""你是数据分析专家。
负责数据收集、清洗和分析。
使用统计方法验证假设。"""
)
# 论文写手
writer = AssistantAgent(
name="paper_writer",
llm_config={"model": "gpt-4o"},
system_message="""你是学术写作专家。
负责论文撰写和修改。
确保语言规范、逻辑清晰。"""
)
# 用户
user = UserProxyAgent(
name="researcher",
human_input_mode="ALWAYS",
code_execution_config=False
)
return [director, literature_researcher, data_analyst, writer, user]
def conduct_research(self, topic: str):
"""执行研究"""
group_chat = GroupChat(
agents=self.agents,
messages=[],
max_round=40,
speaker_selection_method="auto"
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config={"model": "gpt-4o"}
)
return self.agents[-1].initiate_chat(
manager,
message=f"请对以下主题进行深入研究：{topic}"
)
# 使用示例
research = ResearchSystem()
research.conduct_research("大语言模型在教育领域的应用")
3. 内容创作团队
python
class ContentCreationTeam:
"""内容创作团队"""
def __init__(self):
self.setup_agents()
def setup_agents(self):
"""设置 Agent"""
# 内容策划
self.planner = AssistantAgent(
name="content_planner",
llm_config={"model": "gpt-4o"},
system_message="""你是内容策划专家。
负责选题、大纲设计和内容策略。
确保内容有吸引力且符合目标受众。"""
)
# 写手
self.writer = AssistantAgent(
name="writer",
llm_config={"model": "gpt-4o"},
system_message="""你是专业写手。
负责内容创作。
写作风格：清晰、有趣、专业。"""
)
# 编辑
self.editor = AssistantAgent(
name="editor",
llm_config={"model": "gpt-4o"},
system_message="""你是资深编辑。
负责内容审核和修改。
关注：语法、逻辑、可读性。"""
)
# SEO 专家
self.seo_expert = AssistantAgent(
name="seo_expert",
llm_config={"model": "gpt-4o"},
system_message="""你是 SEO 专家。
负责内容优化。
关注：关键词、标题、元描述。"""
)
self.user = UserProxyAgent(
name="client",
human_input_mode="ALWAYS",
code_execution_config=False
)
def create_content(self, topic: str, content_type: str = "blog"):
"""创建内容"""
agents = [self.planner, self.writer, self.editor, self.seo_expert, self.user]
group_chat = GroupChat(
agents=agents,
messages=[],
max_round=30,
speaker_selection_method="auto"
)
manager = GroupChatManager(
groupchat=group_chat,
llm_config={"model": "gpt-4o"}
)
return self.user.initiate_chat(
manager,
message=f"请创建一篇关于 {topic} 的{content_type}文章"
)
# 使用示例
team = ContentCreationTeam()
team.create_content("人工智能在医疗领域的应用", "技术博客")
高级特性
1. 嵌套聊天（Nested Chat）
python
from autogen import ConversableAgent
# 定义消息生成函数
def generate_review_message(recipient, messages, sender, config):
return f"请审查以下代码：\n{messages[-1]['content']}"
# 定义嵌套聊天配置
nested_chat_config = [
{
"recipient": reviewer,
"message": generate_review_message,
"summary_method": "last_msg",
"max_turns": 2
}
]
# 为主 Agent 添加嵌套聊天
assistant = AssistantAgent(
name="coder",
llm_config={"model": "gpt-4o"}
)
assistant.register_nested_chats(
nested_chat_config,
trigger=code_executor
)
2. 会话持久化
python
import json
from autogen import GroupChat
class PersistentGroupChat(GroupChat):
"""可持久化的群聊"""
def __init__(self, *args, save_path="chat_history.json", **kwargs):
super().__init__(*args, **kwargs)
self.save_path = save_path
self.load_history()
def append(self, message, speaker):
"""添加消息并保存"""
super().append(message, speaker)
self.save_history()
def save_history(self):
"""保存聊天历史"""
history = []
for msg, speaker in zip(self.messages, self.speakers):
history.append({
"content": msg["content"],
"speaker": speaker.name,
"role": msg.get("role", "assistant")
})
with open(self.save_path, 'w') as f:
json.dump(history, f, indent=2)
def load_history(self):
"""加载聊天历史"""
try:
with open(self.save_path, 'r') as f:
history = json.load(f)
for item in history:
speaker = self.agent_by_name(item["speaker"])
self.append(
{"content": item["content"], "role": item["role"]},
speaker
)
except FileNotFoundError:
pass
3. 流式输出
python
from autogen import AssistantAgent, UserProxyAgent
# 启用流式输出
assistant = AssistantAgent(
name="assistant",
llm_config={
"model": "gpt-4o",
"stream": True # 启用流式
}
)
# 自定义流式处理
def stream_callback(message):
"""流式消息回调"""
print(message, end="", flush=True)
user_proxy = UserProxyAgent(
name="user",
human_input_mode="NEVER"
)
# 启动流式对话
user_proxy.initiate_chat(
assistant,
message="请详细解释量子计算",
callback=stream_callback
)
与其他框架集成
1. LangChain 集成
python
from langchain_openai import ChatOpenAI
from autogen import AssistantAgent
# 使用 LangChain 的 LLM
llm = ChatOpenAI(model="gpt-4o")
# 包装为 AutoGen 配置
llm_config = {
"config_list": [
{
"model": "gpt-4o",
"api_key": llm.openai_api_key
}
]
}
assistant = AssistantAgent(
name="assistant",
llm_config=llm_config
)
2. 工具链集成
python
from autogen import register_function
from langchain.tools import Tool
# LangChain 工具转 AutoGen
langchain_tool = Tool(
name="search",
func=search_func,
description="搜索工具"
)
# 注册到 AutoGen
register_function(
langchain_tool.func,
caller=assistant,
executor=executor,
name=langchain_tool.name,
description=langchain_tool.description
)
最佳实践
1. Agent 设计原则
python
# 好的 Agent 设计
good_agent = AssistantAgent(
name="specialist",
llm_config={"model": "gpt-4o"},
system_message="""你是一个特定领域的专家。
你的职责：
1. 明确的任务边界
2. 清晰的输出格式
3. 具体的质量标准
请按照以下格式输出：
- 结论
- 证据
- 建议"""
)
# 避免的设计
bad_agent = AssistantAgent(
name="generalist",
llm_config={"model": "gpt-4o"},
system_message="你是一个全能助手，什么都做。" # 太模糊
)
2. 对话管理
python
from autogen import GroupChat
def check_termination(message):
"""检查终止条件"""
content = message.get("content", "").lower()
return any(keyword in content for keyword in ["完成", "结束", "done"])
def create_managed_group_chat(agents, max_rounds=30):
"""创建有管理的群聊"""
return GroupChat(
agents=agents,
messages=[],
max_round=max_rounds,
speaker_selection_method="auto",
is_termination_msg=check_termination,
allow_repeat_speaker=False # 避免重复发言
)
3. 错误处理
python
from autogen import AssistantAgent, GroupChatManager
class RobustGroupChatManager(GroupChatManager):
"""健壮的群聊管理器"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.max_retries = 3
def generate_reply(self, messages, sender, **kwargs):
"""带重试的回复生成"""
for attempt in range(self.max_retries):
try:
return super().generate_reply(messages, sender, **kwargs)
except Exception as e:
if attempt == self.max_retries - 1:
return f"抱歉，处理请求时遇到错误：{e}"
print(f"重试 {attempt + 1}/{self.max_retries}")
总结
AutoGen 核心优势：
- 多智能体协作：原生支持多个 Agent 对话
- 灵活的对话模式：双人、群聊、嵌套对话
- 人类参与：无缝集成人类反馈
- 代码执行：内置安全的代码执行环境
- 高度可定制：灵活的 Agent 行为定义
AutoGen 适合构建复杂的多 Agent 协作系统，是 LangChain 的有力补充。
延伸阅读：