AutoGen 深度解析：微软多 Agent 框架实战

概述

AutoGen 是微软研究院开源的多 Agent 对话框架，核心设计理念是通过 Agent 之间的对话来完成复杂任务。不同于单 Agent 的工具调用模式，AutoGen 让多个具有不同角色和能力的 Agent 协作对话，形成"群体智能"。

2024 年底 AutoGen 发布了 0.4 版本（AutoGen v0.4 / AG2），进行了重大架构重构。本文同时覆盖经典版本和新架构。

核心架构

核心概念

AutoGen 架构
    |
    ├── Agent（智能体）
    │   ├── ConversableAgent - 基础对话 Agent
    │   ├── AssistantAgent - 带 LLM 的助手
    │   ├── UserProxyAgent - 代表用户的 Agent
    │   └── GroupChatManager - 群聊管理者
    |
    ├── Conversation Pattern（对话模式）
    │   ├── Two-Agent Chat - 两个 Agent 对话
    │   ├── Sequential Chat - 顺序对话链
    │   ├── Group Chat - 多 Agent 群聊
    │   └── Nested Chat - 嵌套对话
    |
    └── Code Execution（代码执行）
        ├── LocalCommandLine - 本地执行
        ├── DockerCommandLine - Docker 隔离执行
        └── Jupyter - Notebook 执行

Agent 类型

Agent 类型	角色	关键配置
AssistantAgent	LLM 驱动的助手	system_message, llm_config
UserProxyAgent	代表人类	human_input_mode, code_execution_config
GroupChatManager	群聊协调者	groupchat, llm_config
自定义 Agent	领域专家	继承 ConversableAgent

安装与配置

pip install autogen-agentchat~=0.2

LLM 配置

import autogen

# 配置列表（支持多模型 fallback）
config_list = [
    {
        "model": "gpt-4o",
        "api_key": "sk-xxx",
    },
    {
        "model": "gpt-4o-mini",
        "api_key": "sk-xxx",
    },
]

# 或从环境变量/文件加载
config_list = autogen.config_list_from_json(
    "OAI_CONFIG_LIST",
    filter_dict={"model": ["gpt-4o", "gpt-4o-mini"]},
)

llm_config = {
    "config_list": config_list,
    "temperature": 0,
    "cache_seed": 42,  # 缓存 LLM 响应
}

对话模式实战

模式一：双 Agent 对话

最基础的模式：一个 Assistant 和一个 UserProxy 对话。

import autogen

# 助手 Agent（有 LLM 能力）
assistant = autogen.AssistantAgent(
    name="analyst",
    system_message="""你是一个数据分析专家。
当需要分析数据时，编写 Python 代码来处理。
使用 pandas 和 matplotlib 进行数据分析和可视化。
代码必须包含 print() 输出结果或 plt.savefig() 保存图表。""",
    llm_config=llm_config,
)

# 用户代理（可执行代码）
user_proxy = autogen.UserProxyAgent(
    name="user",
    human_input_mode="NEVER",   # 全自动，不需要人工输入
    max_consecutive_auto_reply=10,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={
        "work_dir": "coding_output",
        "use_docker": False,     # 生产环境建议用 True
    },
)

# 启动对话
user_proxy.initiate_chat(
    assistant,
    message="请分析附件 sales_data.csv 中的月度销售趋势，"
            "并找出销售额最高的产品类别。生成趋势图。",
)

执行流程：

1. UserProxy 发送任务给 Assistant
2. Assistant 生成 Python 代码
3. UserProxy 自动执行代码
4. 执行结果返回给 Assistant
5. Assistant 分析结果，可能生成更多代码
6. 循环直到 Assistant 输出 "TERMINATE"

模式二：群聊（Group Chat）

多个专业 Agent 协作讨论。

# 创建多个专业 Agent
researcher = autogen.AssistantAgent(
    name="researcher",
    system_message="""你是一个技术研究员。
你的职责是搜索和整理技术方案，提供客观的技术分析。
不要写代码，只做调研和分析。""",
    llm_config=llm_config,
)

architect = autogen.AssistantAgent(
    name="architect",
    system_message="""你是一个系统架构师。
基于研究员的调研结果，设计系统架构方案。
输出架构图描述和关键设计决策。
不要写代码。""",
    llm_config=llm_config,
)

developer = autogen.AssistantAgent(
    name="developer",
    system_message="""你是一个高级开发者。
基于架构师的设计，编写实现代码。
代码必须可直接运行，包含完整的错误处理。""",
    llm_config=llm_config,
)

reviewer = autogen.AssistantAgent(
    name="reviewer",
    system_message="""你是一个代码审查专家。
审查开发者的代码，关注：
1. 安全漏洞
2. 性能问题
3. 代码质量
4. 边界情况
提出具体的改进建议。""",
    llm_config=llm_config,
)

user_proxy = autogen.UserProxyAgent(
    name="admin",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=3,
    code_execution_config={"work_dir": "project", "use_docker": True},
)

# 创建群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, researcher, architect, developer, reviewer],
    messages=[],
    max_round=20,
    speaker_selection_method="auto",  # LLM 自动选择下一个发言者
    allow_repeat_speaker=False,       # 禁止连续发言
)

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

# 启动群聊
user_proxy.initiate_chat(
    manager,
    message="我们需要设计一个实时数据处理 pipeline，"
            "要求：接收 Kafka 消息，用 Flink 处理，结果写入 ClickHouse。"
            "请团队协作完成调研、设计、实现和审查。",
)

发言者选择策略

# 方式一：LLM 自动选择（默认）
speaker_selection_method = "auto"

# 方式二：轮询
speaker_selection_method = "round_robin"

# 方式三：随机
speaker_selection_method = "random"

# 方式四：自定义选择函数
def custom_speaker_selection(last_speaker, groupchat):
    """基于规则的发言者选择"""
    messages = groupchat.messages

    if last_speaker.name == "researcher":
        return architect  # 研究员之后是架构师
    elif last_speaker.name == "architect":
        return developer  # 架构师之后是开发者
    elif last_speaker.name == "developer":
        return reviewer   # 开发者之后是审查员
    elif last_speaker.name == "reviewer":
        if "approved" in messages[-1]["content"].lower():
            return None    # 审查通过，结束
        return developer   # 审查不通过，回到开发者

    return "auto"  # 其他情况用 LLM 决定

groupchat = autogen.GroupChat(
    agents=agents,
    messages=[],
    speaker_selection_method=custom_speaker_selection,
)

模式三：顺序对话链

# 定义对话链
chat_results = user_proxy.initiate_chats([
    {
        "recipient": researcher,
        "message": "调研当前主流的向量数据库方案",
        "max_turns": 3,
        "summary_method": "reflection_with_llm",
        "summary_args": {
            "summary_prompt": "总结调研的关键发现，用要点列表",
        },
    },
    {
        "recipient": architect,
        "message": "基于调研结果，设计 RAG 系统架构",
        "max_turns": 3,
        "summary_method": "reflection_with_llm",
    },
    {
        "recipient": developer,
        "message": "基于架构设计，实现核心模块",
        "max_turns": 5,
        "summary_method": "last_msg",
    },
])

# 每步的上下文会自动传递给下一步

工具注册与使用

from typing import Annotated

# 定义工具函数
def search_web(
    query: Annotated[str, "搜索关键词"],
    max_results: Annotated[int, "最大结果数"] = 5,
) -> str:
    """在网络上搜索信息"""
    # 实际实现...
    results = web_search_api(query, max_results)
    return json.dumps(results, ensure_ascii=False)

def query_database(
    sql: Annotated[str, "SQL 查询语句"],
) -> str:
    """执行 SQL 查询"""
    # 实际实现...
    result = db.execute(sql)
    return json.dumps(result)

def calculate(
    expression: Annotated[str, "数学表达式"],
) -> float:
    """计算数学表达式"""
    return eval(expression)  # 生产环境应使用安全的表达式求值

# 注册工具到 Agent
autogen.register_function(
    search_web,
    caller=assistant,      # 谁来调用（生成调用请求）
    executor=user_proxy,   # 谁来执行（实际运行函数）
    description="Search the web for information",
)

autogen.register_function(
    query_database,
    caller=assistant,
    executor=user_proxy,
    description="Execute SQL query on the database",
)

autogen.register_function(
    calculate,
    caller=assistant,
    executor=user_proxy,
    description="Calculate a mathematical expression",
)

代码执行安全

# 方式一：Docker 隔离执行（推荐）
code_execution_config = {
    "work_dir": "coding_output",
    "use_docker": "python:3.11-slim",
    "timeout": 60,
    "last_n_messages": "auto",
}

# 方式二：本地执行（开发用）
code_execution_config = {
    "work_dir": "coding_output",
    "use_docker": False,
    "timeout": 30,
}

# 方式三：Jupyter 执行
from autogen.coding import JupyterCodeExecutor

executor = JupyterCodeExecutor(
    jupyter_server_url="http://localhost:8888",
    kernel_name="python3",
    timeout=120,
    output_dir="./jupyter_output",
)

人工介入控制

# human_input_mode 三种模式
user_proxy_auto = autogen.UserProxyAgent(
    name="auto_user",
    human_input_mode="NEVER",     # 全自动
    max_consecutive_auto_reply=10,
)

user_proxy_confirm = autogen.UserProxyAgent(
    name="confirm_user",
    human_input_mode="TERMINATE",  # 仅在终止时询问
)

user_proxy_always = autogen.UserProxyAgent(
    name="manual_user",
    human_input_mode="ALWAYS",    # 每步都询问
)

# 自定义终止条件
def should_terminate(message):
    content = message.get("content", "")
    return (
        "TERMINATE" in content
        or "任务完成" in content
        or message.get("tool_calls") is None  # 没有工具调用时停止
    )

user_proxy = autogen.UserProxyAgent(
    name="user",
    is_termination_msg=should_terminate,
    max_consecutive_auto_reply=15,
)

AutoGen v0.4（AG2）新架构

v0.4 引入了更现代的架构，基于事件驱动和异步消息传递。

# AutoGen v0.4 (autogen-agentchat >= 0.4)
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_ext.models.openai import OpenAIChatCompletionClient

# 模型客户端
model = OpenAIChatCompletionClient(
    model="gpt-4o",
    api_key="sk-xxx",
)

# 创建 Agent
planner = AssistantAgent(
    "planner",
    model_client=model,
    system_message="你是项目规划专家，将任务分解为可执行的步骤。",
)

executor = AssistantAgent(
    "executor",
    model_client=model,
    system_message="你是执行专家，按照计划实现具体步骤。",
)

critic = AssistantAgent(
    "critic",
    model_client=model,
    system_message="你是质量审查专家，检查输出质量并提出改进建议。"
                   "当你认为任务完成时，回复 'APPROVE'。",
)

# 终止条件
termination = TextMentionTermination("APPROVE")

# 创建团队
team = RoundRobinGroupChat(
    [planner, executor, critic],
    termination_condition=termination,
    max_turns=15,
)

# 运行
import asyncio

async def main():
    result = await team.run(task="设计一个 URL 短链服务的系统架构")
    print(result)

asyncio.run(main())

v0.2 vs v0.4 对比

维度	v0.2 (经典)	v0.4 (AG2)
消息传递	同步函数调用	异步事件驱动
代码执行	内置到 Agent	独立的 CodeExecutor
团队协作	GroupChat	Team (RoundRobin/Selector/Swarm)
模型绑定	config_list	ModelClient 接口
可观测性	日志	结构化事件流
扩展性	继承	组合 + 插件

生产环境最佳实践

成本控制

# 设置 token 预算
llm_config = {
    "config_list": config_list,
    "temperature": 0,
    "cache_seed": 42,       # 缓存相同输入的响应
    "max_tokens": 2000,     # 限制单次输出
}

# 限制对话轮数
user_proxy = autogen.UserProxyAgent(
    name="user",
    max_consecutive_auto_reply=5,  # 最多 5 轮自动回复
)

# 对话后查看用量
print(f"Total cost: ${assistant.client.total_usage['total_cost']:.4f}")

错误处理

# 重试和 fallback 配置
config_list = [
    {"model": "gpt-4o", "api_key": "sk-xxx"},
    {"model": "gpt-4o-mini", "api_key": "sk-xxx"},  # fallback
]

# 自定义错误处理
class RobustAssistant(autogen.AssistantAgent):
    def generate_reply(self, messages=None, sender=None, **kwargs):
        try:
            return super().generate_reply(messages, sender, **kwargs)
        except Exception as e:
            return f"Error occurred: {str(e)}. Let me try a different approach."

典型应用场景

场景	Agent 配置	对话模式
数据分析	Analyst + UserProxy	双 Agent
代码开发	PM + Dev + Tester	群聊
研究报告	Researcher + Writer + Editor	顺序链
客服系统	Router + Specialists + Escalation	嵌套对话
决策支持	Pros_Agent + Cons_Agent + Judge	群聊辩论

总结

AutoGen 的核心价值在于将"多 Agent 对话"作为解决复杂问题的第一性原理。通过角色分工、轮次控制和代码执行的有机组合，AutoGen 能够处理从数据分析到系统设计的广泛任务。选择 AutoGen 的关键考量：需要多个角色协作、任务需要代码执行能力、需要人工可介入的自动化流程。

---

Maurice | [email protected]