AutoGen多Agent系统：构建智能协作的AI应用

引言

随着大语言模型（LLM）能力的不断提升，单一Agent已难以满足复杂任务的需求。Microsoft AutoGen 是一个开源框架，专门用于构建和编排多个AI Agent协同工作的系统。本文将深入介绍AutoGen的核心概念、架构设计以及实战应用。

一、AutoGen 核心概念

1.1 什么是 AutoGen？

AutoGen 是由微软研究院开发的开源框架，旨在简化多Agent对话系统的构建。它提供了一套高层抽象，让开发者能够：

• 快速构建可对话的AI Agent
• 灵活编排多Agent协作流程
• 无缝集成人类参与（Human-in-the-loop）
• 高效利用LLM能力

1.2 核心架构

┌─────────────────────────────────────────────────────────┐
│                    AutoGen Framework                     │
├─────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐ │
│  │ Conversable │  │   GroupChat │  │  Human-in-the   │ │
│  │   Agent     │  │   Manager   │  │     Loop        │ │
│  └─────────────┘  └─────────────┘  └─────────────────┘ │
├─────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐ │
│  │ LLM Backend │  │   Tools     │  │   Memory Store  │ │
│  │ (OpenAI/...)│  │  (Function  │  │  (Context Mgmt) │ │
│  │             │  │   Calling)  │  │                 │ │
│  └─────────────┘  └─────────────┘  └─────────────────┘ │
└─────────────────────────────────────────────────────────┘

二、核心组件详解

2.1 ConversableAgent

ConversableAgent 是AutoGen中最基础的Agent类，所有其他Agent类型都继承自它。

from autogen import ConversableAgent

# 基础配置
config_list = [
    {
        "model": "gpt-4",
        "api_key": "your-api-key"
    }
]

# 创建基础Agent
agent = ConversableAgent(
    name="chatbot",
    system_message="你是一个有帮助的AI助手。",
    llm_config={"config_list": config_list},
    human_input_mode="NEVER"  # 从不请求人类输入
)

关键属性：

2.2 UserProxyAgent

UserProxyAgent 代表人类用户，可以执行代码、调用工具，并在需要时请求人类输入。

from autogen import UserProxyAgent

# 创建用户代理
user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="TERMINATE",  # 在终止时请求输入
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False,  # 生产环境建议使用Docker
    },
)

代码执行配置：

# 安全代码执行（推荐）
code_execution_config = {
    "work_dir": "workspace",           # 工作目录
    "use_docker": True,                # 使用Docker隔离
    "timeout": 120,                    # 执行超时（秒）
    "last_n_messages": 3,              # 检查最近N条消息
}

2.3 AssistantAgent

AssistantAgent 是专门为LLM对话设计的Agent，默认不能执行代码，但可以编写代码和建议。

from autogen import AssistantAgent

# 创建助手Agent
assistant = AssistantAgent(
    name="coding_assistant",
    system_message="""你是一个专业的Python程序员。
    你的任务是编写高质量、可维护的代码。
    完成任务后，请回复 TERMINATE。""",
    llm_config={"config_list": config_list}
)

三、多Agent对话模式

3.1 两Agent对话（Two-Agent Chat）

最简单的多Agent交互模式：

from autogen import UserProxyAgent, AssistantAgent

# 初始化Agent
assistant = AssistantAgent(
    name="assistant",
    system_message="你是一个有帮助的AI助手。",
    llm_config={"config_list": config_list}
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: "TERMINATE" in x.get("content", ""),
    code_execution_config={"work_dir": "workspace"}
)

# 开始对话
user_proxy.initiate_chat(
    assistant,
    message="写一个Python函数，计算斐波那契数列的前n项。",
)

3.2 顺序对话（Sequential Chat）

多个Agent按顺序接力完成任务：

# 定义多个专业Agent
data_analyst = AssistantAgent(
    name="data_analyst",
    system_message="你是数据分析专家，专注于数据清洗和统计分析。"
)

viz_expert = AssistantAgent(
    name="viz_expert", 
    system_message="你是数据可视化专家，擅长使用matplotlib和plotly。"
)

reporter = AssistantAgent(
    name="reporter",
    system_message="你是报告撰写专家，负责生成清晰的数据分析报告。"
)

# 顺序执行
chat_results = user_proxy.initiate_chats([
    {"recipient": data_analyst, "message": "分析这个数据集...", "summary_method": "reflection_with_llm"},
    {"recipient": viz_expert, "message": "根据分析结果创建可视化...", "summary_method": "reflection_with_llm"},
    {"recipient": reporter, "message": "生成最终报告...", "summary_method": "last_msg"},
])

3.3 群聊模式（GroupChat）

最灵活的多Agent协作模式：

from autogen import GroupChat, GroupChatManager

# 创建多个专业Agent
project_manager = AssistantAgent(
    name="pm",
    system_message="""你是项目经理。负责协调团队，分配任务，
    确保项目按时完成。在任务分配清楚后回复 TERMINATE。"""
)

coder = AssistantAgent(
    name="coder",
    system_message="你是Python开发专家。编写高质量、带注释的代码。"
)

tester = AssistantAgent(
    name="tester",
    system_message="你是测试工程师。编写单元测试，检查代码质量。"
)

reviewer = AssistantAgent(
    name="reviewer",
    system_message="你是代码审查专家。检查代码风格和潜在问题。"
)

# 配置群聊
groupchat = GroupChat(
    agents=[user_proxy, project_manager, coder, tester, reviewer],
    messages=[],
    max_round=20,
    speaker_selection_method="auto",  # 自动选择下一个发言者
    allow_repeat_speaker=False,        # 禁止连续发言
)

# 创建群聊管理器
manager = GroupChatManager(
    groupchat=groupchat,
    llm_config={"config_list": config_list}
)

# 启动群聊
user_proxy.initiate_chat(
    manager,
    message="开发一个待办事项管理应用，包含增删改查功能。"
)

四、实战案例：智能软件开发团队

4.1 完整代码实现

import autogen
from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

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

# 定义系统提示词
SYSTEM_PROMPTS = {
    "pm": """你是经验丰富的项目经理。
    职责：
    1. 分析用户需求，制定开发计划
    2. 协调开发、测试、审查团队
    3. 跟踪项目进度，确保质量
    4. 在任务明确分配后回复 TERMINATE
    """,
    
    "architect": """你是软件架构师。
    职责：
    1. 设计系统架构和技术方案
    2. 定义模块接口和数据结构
    3. 评估技术选型的合理性
    """,
    
    "developer": """你是高级Python开发工程师。
    职责：
    1. 根据架构设计编写实现代码
    2. 遵循PEP8规范，编写清晰注释
    3. 处理异常情况和边界条件
    """,
    
    "tester": """你是QA工程师。
    职责：
    1. 编写全面的单元测试
    2. 进行边界测试和异常测试
    3. 确保代码覆盖率>80%
    """,
    
    "reviewer": """你是代码审查专家。
    职责：
    1. 检查代码质量和设计模式
    2. 识别潜在bug和安全问题
    3. 提出改进建议
    """
}

# 创建Agent工厂
def create_agents():
    """创建软件开发团队的所有Agent"""
    
    agents = {}
    
    # 项目经理
    agents["pm"] = AssistantAgent(
        name="ProjectManager",
        system_message=SYSTEM_PROMPTS["pm"],
        llm_config={"config_list": config_list},
    )
    
    # 架构师
    agents["architect"] = AssistantAgent(
        name="Architect",
        system_message=SYSTEM_PROMPTS["architect"],
        llm_config={"config_list": config_list},
    )
    
    # 开发者
    agents["developer"] = AssistantAgent(
        name="Developer",
        system_message=SYSTEM_PROMPTS["developer"],
        llm_config={"config_list": config_list},
    )
    
    # 测试工程师
    agents["tester"] = AssistantAgent(
        name="Tester",
        system_message=SYSTEM_PROMPTS["tester"],
        llm_config={"config_list": config_list},
    )
    
    # 代码审查
    agents["reviewer"] = AssistantAgent(
        name="CodeReviewer",
        system_message=SYSTEM_PROMPTS["reviewer"],
        llm_config={"config_list": config_list},
    )
    
    return agents

# 创建用户代理
def create_user_proxy():
    """创建用户代理"""
    return UserProxyAgent(
        name="User",
        human_input_mode="TERMINATE",
        max_consecutive_auto_reply=15,
        is_termination_msg=lambda x: "TERMINATE" in x.get("content", ""),
        code_execution_config={
            "work_dir": "software_project",
            "use_docker": False,
            "timeout": 300,
        },
    )

# 配置群聊
def setup_group_chat(agents, user_proxy):
    """配置群聊管理器"""
    
    groupchat = GroupChat(
        agents=[user_proxy, agents["pm"], agents["architect"], 
                agents["developer"], agents["tester"], agents["reviewer"]],
        messages=[],
        max_round=30,
        speaker_selection_method="auto",
        allow_repeat_speaker=False,
    )
    
    manager = GroupChatManager(
        groupchat=groupchat,
        llm_config={"config_list": config_list}
    )
    
    return manager

# 主函数
def main():
    """运行软件开发团队"""
    
    # 创建所有Agent
    agents = create_agents()
    user_proxy = create_user_proxy()
    manager = setup_group_chat(agents, user_proxy)
    
    # 项目需求
    requirement = """
    开发一个RESTful API服务，实现用户认证系统：
    
    功能需求：
    1. 用户注册（邮箱验证）
    2. 用户登录（JWT Token）
    3. 密码重置
    4. 用户信息管理
    
    技术要求：
    - 使用FastAPI框架
    - 使用SQLAlchemy + SQLite
    - 使用Pydantic进行数据验证
    - 包含完整的错误处理
    
    请团队协作完成这个项目。
    """
    
    # 启动项目
    user_proxy.initiate_chat(
        manager,
        message=requirement
    )

if __name__ == "__main__":
    main()

4.2 运行结果示例

User (to chat_manager):
开发一个RESTful API服务，实现用户认证系统...

ProjectManager (to chat_manager):
好的，我来分析这个需求并制定开发计划。

阶段1：架构设计 - @Architect 请设计系统架构
阶段2：后端开发 - @Developer 实现API接口  
阶段3：测试验证 - @Tester 编写测试用例
阶段4：代码审查 - @Reviewer 审查代码质量

Architect (to chat_manager):
我设计了以下架构...
[架构设计内容]

Developer (to chat_manager):
根据架构设计，我实现了以下代码...
[代码实现]

Tester (to chat_manager):
我编写了以下测试用例...
[测试代码]

CodeReviewer (to chat_manager):
代码审查结果...
[审查意见]

ProjectManager (to chat_manager):
项目已完成，所有阶段都通过验收。TERMINATE

五、高级特性

5.1 Function Calling（函数调用）

让Agent调用外部工具和API：

from autogen import register_function
import requests

# 定义工具函数
def get_weather(location: str) -> str:
    """获取指定城市的天气信息"""
    # 调用天气API
    api_key = "your-weather-api-key"
    url = f"https://api.weather.com/v1/current?city={location}&appid={api_key}"
    response = requests.get(url)
    return response.json()

def search_database(query: str) -> list:
    """搜索数据库"""
    # 数据库查询逻辑
    results = db.execute(f"SELECT * FROM products WHERE name LIKE '%{query}%'")
    return results

# 注册函数到Agent
assistant = AssistantAgent(
    name="assistant",
    system_message="你可以使用工具函数获取信息。",
    llm_config={
        "config_list": config_list,
        "functions": [
            {
                "name": "get_weather",
                "description": "获取指定城市的天气",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {"type": "string", "description": "城市名称"}
                    },
                    "required": ["location"]
                }
            },
            {
                "name": "search_database",
                "description": "搜索产品数据库",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "搜索关键词"}
                    },
                    "required": ["query"]
                }
            }
        ]
    }
)

# 注册实际函数
register_function(
    get_weather,
    caller=assistant,
    executor=user_proxy,
    name="get_weather",
    description="获取指定城市的天气"
)

5.2 自定义 Speaker Selection

控制群聊中的发言顺序：

def custom_speaker_selection(last_speaker, groupchat):
    """自定义发言者选择逻辑"""
    
    messages = groupchat.messages
    
    # 根据对话历史智能选择下一个发言者
    if len(messages) < 2:
        return groupchat.agent_by_name("ProjectManager")
    
    last_message = messages[-1]["content"]
    
    # 如果提到架构设计，让架构师发言
    if "架构" in last_message or "design" in last_message.lower():
        return groupchat.agent_by_name("Architect")
    
    # 如果提到代码实现，让开发者发言
    if "代码" in last_message or "实现" in last_message:
        return groupchat.agent_by_name("Developer")
    
    # 如果提到测试，让测试工程师发言
    if "测试" in last_message or "bug" in last_message.lower():
        return groupchat.agent_by_name("Tester")
    
    # 默认使用轮询
    return groupchat.next_agent(last_speaker)

# 应用自定义选择器
groupchat = GroupChat(
    agents=agents,
    messages=[],
    speaker_selection_method=custom_speaker_selection
)

5.3 对话记忆管理

from autogen import ChatCompletion

# 启用对话摘要
assistant = AssistantAgent(
    name="assistant",
    system_message="你是一个有记忆的助手。",
    llm_config={
        "config_list": config_list,
        "summary_prompt": "Summarize the conversation so far:",
        "summary_method": "reflection_with_llm",  # 使用LLM生成摘要
    }
)

# 手动管理上下文
user_proxy.initiate_chat(
    assistant,
    message="你好",
    clear_history=True,  # 清空历史
    silent=False         # 显示详细输出
)

六、最佳实践

6.1 系统设计原则

1. 角色定义清晰

   • 每个Agent有明确的职责边界
   • 系统提示词详细描述角色和能力
   • 避免角色重叠导致冲突

2. 对话流程可控

   • 设置合理的 max_round 防止无限循环
   • 定义明确的终止条件
   • 使用 allow_repeat_speaker 避免单一Agent垄断

3. 错误处理完善

   • 代码执行使用Docker隔离
   • 设置执行超时时间
   • 捕获并处理异常

6.2 性能优化

# 1. 使用缓存减少LLM调用
from functools import lru_cache

@lru_cache(maxsize=100)
def cached_llm_call(prompt):
    return llm_completion(prompt)

# 2. 异步执行提高效率
import asyncio

async def parallel_agent_execution(tasks):
    results = await asyncio.gather(*tasks)
    return results

# 3. 合理设置上下文长度
llm_config = {
    "config_list": config_list,
    "max_tokens": 2000,      # 限制生成长度
    "context_window": 8000,  # 上下文窗口
}

6.3 安全建议

# 1. 代码执行安全配置
code_execution_config = {
    "work_dir": "sandbox",
    "use_docker": True,           # 必须使用Docker
    "timeout": 60,                # 限制执行时间
    "env": {                      # 限制环境变量
        "PATH": "/usr/bin",
    }
}

# 2. 输入验证
def sanitize_input(user_input):
    # 过滤危险命令
    dangerous = ["rm -rf", "mkfs", "dd if"]
    for cmd in dangerous:
        if cmd in user_input:
            raise ValueError(f"Dangerous command detected: {cmd}")
    return user_input

# 3. API密钥管理
import os
from dotenv import load_dotenv

load_dotenv()
config_list = [{
    "api_key": os.getenv("OPENAI_API_KEY"),
    "model": "gpt-4"
}]

七、常见问题与解决方案

7.1 Agent循环对话

问题： Agent之间反复讨论无法终止

解决：

# 设置更严格的终止条件
is_termination_msg = lambda x: (
    "TERMINATE" in x.get("content", "") or
    "任务完成" in x.get("content", "") or
    len(groupchat.messages) > 20
)

7.2 代码执行失败

问题： 代码执行环境配置错误

解决：

# 检查环境
code_execution_config = {
    "work_dir": "workspace",
    "use_docker": False,
    "timeout": 120,
}

# 确保工作目录存在
import os
os.makedirs("workspace", exist_ok=True)

7.3 Token超限

问题： 对话历史太长导致Token超限

解决：

# 定期总结对话
from autogen import summarize_chat

# 或使用更小的上下文窗口
llm_config = {
    "config_list": config_list,
    "summary_method": "last_n_msg",
    "summary_args": {"last_messages": 5}
}

八、总结

Microsoft AutoGen 为构建多Agent系统提供了强大而灵活的框架。通过本文的介绍，你应该已经掌握了：

1. 核心概念：ConversableAgent、UserProxyAgent、AssistantAgent的区别和用法
2. 对话模式：两Agent对话、顺序对话、群聊模式的适用场景
3. 实战技能：构建完整的软件开发团队Agent系统
4. 高级特性：函数调用、自定义发言选择、记忆管理
5. 最佳实践：安全、性能、可维护性的设计原则

AutoGen 正在快速发展，建议关注官方GitHub仓库获取最新特性。多Agent协作是AI应用的重要趋势，掌握AutoGen将帮助你构建更智能、更强大的AI系统。

参考资源

• AutoGen GitHub
• AutoGen 官方文档
• AutoGen 论文
• 示例代码库

本文档最后更新于 2024年5月，基于 AutoGen v0.2.x 版本编写。