AutoGen 多智能体框架完全教程

从零掌握微软 AutoGen 多智能体协作框架，构建强大的 AI Agent 团队

📑 目录

1. AutoGen 概述与核心架构
2. ConversableAgent 基类详解
3. AssistantAgent 与 UserProxyAgent
4. 代码执行与 Docker 沙箱
5. 工具注册与函数调用
6. 对话模式：双人对话与群聊
7. GroupChatManager 团队编排
8. 嵌套对话 Nested Chat
9. 人机协作模式 Human-in-the-Loop
10. 自定义 Agent 开发
11. AutoGen Studio 可视化界面
12. 流式输出与实时反馈
13. 错误处理与重试机制
14. 生产部署与扩展
15. 实战项目 1：自动化代码审查系统
16. 实战项目 2：智能研究助手

1. AutoGen 概述与核心架构

1.1 什么是 AutoGen

AutoGen 是微软研究院开源的多智能体（Multi-Agent）协作框架。它的核心理念是：让多个 AI Agent 通过对话协作来完成复杂任务，而不是依赖单一 Agent 独自完成一切。

与单 Agent 框架（如 LangChain Agent）不同，AutoGen 的优势在于：

• 角色分工：每个 Agent 有明确的专业角色（程序员、审查员、规划师等）
• 对话驱动：Agent 之间通过自然语言对话协作，类似人类团队沟通
• 灵活编排：支持双人对话、群聊、嵌套对话等多种协作模式
• 人机融合：天然支持人类作为"团队成员"参与决策

1.2 核心架构

AutoGen 的架构围绕以下核心组件构建：

┌─────────────────────────────────────────┐
│            AutoGen Framework            │
├─────────────────────────────────────────┤
│                                         │
│  ConversableAgent (基类)                │
│    ├── AssistantAgent (AI 助手)         │
│    ├── UserProxyAgent (人类代理)        │
│    └── 自定义 Agent                     │
│                                         │
│  GroupChatManager (群聊管理器)          │
│    └── GroupChat (群聊容器)             │
│                                         │
│  工具层: 函数注册 / 代码执行 / 沙箱     │
│  模型层: OpenAI / Azure / 本地模型      │
│                                         │
└─────────────────────────────────────────┘

1.3 安装与环境配置

# 基础安装
pip install pyautogen

# 安装 Docker 支持（用于沙箱代码执行）
pip install docker

# 安装额外功能
pip install pyautogen[retrievechat]  # RAG 支持
pip install pyautogen[mathchat]      # 数学推理

配置 LLM：

import autogen

# 方式一：使用 OpenAI
config_list = [
    {
        "model": "gpt-4o",
        "api_key": "your-api-key",
    }
]

# 方式二：使用本地模型（如 Ollama）
config_list = [
    {
        "model": "llama3",
        "base_url": "http://localhost:11434/v1",
        "api_key": "not-needed",
    }
]

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

2. ConversableAgent 基类详解

ConversableAgent 是所有 Agent 的基类，理解它是掌握 AutoGen 的关键。

2.1 核心参数

agent = autogen.ConversableAgent(
    name="my_agent",                      # Agent 名称（用于对话标识）
    system_message="你是一个有帮助的助手。",  # 系统提示词
    llm_config={                          # LLM 配置
        "config_list": config_list,
        "temperature": 0.7,
        "timeout": 120,
        "cache_seed": 42,                 # 缓存种子（相同输入返回缓存结果）
    },
    human_input_mode="NEVER",             # 人机交互模式
    max_consecutive_auto_reply=10,        # 最大连续自动回复数
    is_termination_msg=lambda msg: msg.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config=False,          # 代码执行配置
    default_auto_reply="收到，请继续。",    # 默认自动回复
)

2.2 人机交互模式

human_input_mode 是核心配置之一：

2.3 消息格式

AutoGen 内部消息采用标准格式：

message = {
    "role": "assistant",          # 或 "user", "system"
    "content": "这是一条消息",     # 文本内容
    "name": "agent_name",         # 发送者名称
}

3. AssistantAgent 与 UserProxyAgent

3.1 AssistantAgent

AssistantAgent 是"纯 AI"角色，它不会执行代码，只负责思考和回复：

assistant = autogen.AssistantAgent(
    name="senior_engineer",
    system_message="""你是一位资深软件工程师。
你的职责是：
1. 分析需求并设计解决方案
2. 编写高质量代码
3. 解释你的设计决策
4. 在代码完成后说 TERMINATE""",
    llm_config={"config_list": config_list},
)

3.2 UserProxyAgent

UserProxyAgent 是人类的"代理"，它可以执行代码并代表人类发言：

user_proxy = autogen.UserProxyAgent(
    name="user",
    human_input_mode="NEVER",      # 自动模式
    max_consecutive_auto_reply=5,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
    code_execution_config={
        "work_dir": "workspace",    # 代码执行工作目录
        "use_docker": True,         # 使用 Docker 沙箱
        "timeout": 60,
        "last_n_messages": 3,       # 检查最近 N 条消息中的代码
    },
    system_message="你是一个代码执行代理，负责运行助手提供的代码。",
)

3.3 最简单的双人对话

import autogen

config_list = autogen.config_list_from_json("OAI_CONFIG_LIST")

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config={"config_list": config_list},
)

user_proxy = autogen.UserProxyAgent(
    name="user",
    human_input_mode="TERMINATE",
    code_execution_config={"work_dir": "workspace"},
)

# 发起对话
user_proxy.initiate_chat(
    assistant,
    message="帮我写一个 Python 函数，计算斐波那契数列的第 n 项，并测试 n=10。",
)

4. 代码执行与 Docker 沙箱

4.1 执行模式

AutoGen 支持三种代码执行模式：

# 模式 1：不执行代码（仅对话）
code_execution_config = False

# 模式 2：在本地执行（有安全风险）
code_execution_config = {
    "work_dir": "./workspace",
    "use_docker": False,
}

# 模式 3：Docker 沙箱执行（推荐）
code_execution_config = {
    "work_dir": "./workspace",
    "use_docker": True,          # True = 使用默认镜像
    # "use_docker": "python:3.11-slim",  # 或指定镜像
    "timeout": 120,
    "last_n_messages": 3,
}

4.2 Docker 沙箱配置

code_execution_config = {
    "work_dir": "/tmp/autogen_workspace",
    "use_docker": "python:3.11-slim",
    "timeout": 180,
    "last_n_messages": 5,
    # 自定义 Docker 参数
    "docker_extra_args": [
        "--memory=512m",
        "--cpus=1",
        "--network=none",  # 禁用网络（更安全）
    ],
}

4.3 禁用自动代码执行

有时候你只需要 Agent 生成代码，不需要自动执行：

# 方式一：整个 Agent 不执行
user_proxy = autogen.UserProxyAgent(
    name="user",
    code_execution_config=False,
)

# 方式二：在对话中动态控制
def custom_code_executor(code_blocks):
    """自定义代码执行器"""
    for code, lang in code_blocks:
        if lang in ["python", "bash"]:
            # 自定义执行逻辑
            print(f"执行 {lang} 代码:\n{code}")
    return 0, "代码已记录但未执行。"

user_proxy.register_reply(
    autogen.Agent,
    reply_func=custom_code_executor,
    position=0,
)

5. 工具注册与函数调用

5.1 注册函数作为工具

AutoGen 支持将 Python 函数注册为 Agent 可调用的工具：

import json
from typing import Annotated

# 定义工具函数
def search_web(
    query: Annotated[str, "搜索关键词"],
    max_results: Annotated[int, "最大结果数"] = 5,
) -> str:
    """搜索互联网并返回结果。"""
    # 模拟搜索结果
    results = [
        {"title": f"结果 {i}", "url": f"https://example.com/{i}", "snippet": f"关于{query}的内容..."}
        for i in range(max_results)
    ]
    return json.dumps(results, ensure_ascii=False)

def read_file(
    file_path: Annotated[str, "文件路径"],
) -> str:
    """读取指定文件的内容。"""
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            return f.read()
    except Exception as e:
        return f"读取文件失败: {str(e)}"

# 注册工具到 Agent
assistant.register_for_llm(
    name="search_web",
    description="搜索互联网获取最新信息",
)(search_web)

user_proxy.register_for_execution(
    name="search_web",
)(search_web)

assistant.register_for_llm(
    name="read_file",
    description="读取本地文件内容",
)(read_file)

user_proxy.register_for_execution(
    name="read_file",
)(read_file)

5.2 使用装饰器注册

@assistant.register_for_llm(
    name="calculate",
    description="执行数学计算"
)
@user_proxy.register_for_execution()
def calculate(
    expression: Annotated[str, "数学表达式，如 '2 + 3 * 4'"],
) -> str:
    """安全地执行数学计算。"""
    try:
        # 限制可用函数确保安全
        allowed_names = {"abs": abs, "round": round, "min": min, "max": max}
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

5.3 批量注册工具

# 使用 FunctionTool 批量管理
from autogen import register_function

tools = [
    {
        "function": search_web,
        "name": "search_web",
        "description": "搜索互联网",
    },
    {
        "function": read_file,
        "name": "read_file",
        "description": "读取文件",
    },
]

register_function(
    search_web,
    caller=assistant,
    executor=user_proxy,
    name="search_web",
    description="搜索互联网获取最新信息",
)

6. 对话模式：双人对话与群聊

6.1 双人对话（Two-Agent Chat）

最基础的模式，两个 Agent 交替对话：

# 发起对话
result = user_proxy.initiate_chat(
    assistant,
    message="分析这段代码的性能问题并优化：\n```python\nimport time\ndef slow_sum(n):\n    result = 0\n    for i in range(n):\n        time.sleep(0.001)\n        result += i\n    return result\n```",
    max_turns=6,  # 最大轮数
)

# 查看对话历史
for msg in result.chat_history:
    print(f"[{msg['role']}]: {msg['content'][:100]}...")

6.2 连续对话

# 在现有对话基础上继续提问
user_proxy.send(
    "请给优化后的代码添加单元测试。",
    assistant,
)

6.3 群聊模式预览

群聊模式将在下一节详细展开：

agent1 = autogen.AssistantAgent(name="planner", ...)
agent2 = autogen.AssistantAgent(name="coder", ...)
agent3 = autogen.AssistantAgent(name="reviewer", ...)
user = autogen.UserProxyAgent(name="user", ...)

# 群聊将在 GroupChat 中启动

7. GroupChatManager 团队编排

7.1 基本群聊

import autogen

# 创建团队成员
planner = autogen.AssistantAgent(
    name="planner",
    system_message="你是项目规划师。负责分解任务、制定计划、协调团队。在计划完成后说 TERMINATE。",
    llm_config={"config_list": config_list},
)

coder = autogen.AssistantAgent(
    name="coder",
    system_message="你是 Python 开发者。根据规划师的计划编写代码。只写代码，不做规划。",
    llm_config={"config_list": config_list},
)

reviewer = autogen.AssistantAgent(
    name="reviewer",
    system_message="你是代码审查员。审查代码质量、安全性和性能。提出改进建议。",
    llm_config={"config_list": config_list},
)

user_proxy = autogen.UserProxyAgent(
    name="user",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "workspace", "use_docker": True},
)

# 创建群聊
groupchat = autogen.GroupChat(
    agents=[planner, coder, reviewer, user_proxy],
    messages=[],
    max_round=20,           # 最大轮数
    speaker_selection_method="auto",  # 自动选择发言者
)

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

# 启动群聊
user_proxy.initiate_chat(
    manager,
    message="请设计并实现一个简单的待办事项 API，使用 FastAPI。",
)

7.2 发言者选择策略

groupchat = autogen.GroupChat(
    agents=[planner, coder, reviewer, user_proxy],
    messages=[],
    max_round=15,
    # 策略 1：自动选择（LLM 决定谁说话）
    speaker_selection_method="auto",
    
    # 策略 2：轮流发言
    # speaker_selection_method="round_robin",
    
    # 策略 3：随机选择
    # speaker_selection_method="random",
    
    # 策略 4：自定义函数
    # speaker_selection_method=my_custom_selector,
)

# 自定义发言者选择函数
def my_custom_selector(last_speaker, groupchat):
    """根据上一个发言者决定下一个"""
    if last_speaker == planner:
        return coder       # 规划后让程序员写代码
    elif last_speaker == coder:
        return reviewer    # 写完代码后让审查员检查
    elif last_speaker == reviewer:
        return planner     # 审查后让规划师确认
    return user_proxy      # 默认返回用户

7.3 允许重复发言

groupchat = autogen.GroupChat(
    agents=[planner, coder, reviewer],
    messages=[],
    max_round=20,
    speaker_selection_method="auto",
    allow_repeat_speaker=True,  # 允许同一 Agent 连续发言
)

8. 嵌套对话 Nested Chat

嵌套对话允许一个 Agent 在回复前先与其他 Agent 进行内部"讨论"。

8.1 基本嵌套对话

# 内部审查 Agent（不直接与用户交互）
internal_reviewer = autogen.AssistantAgent(
    name="internal_reviewer",
    system_message="你是内部代码审查员。仔细检查代码的 bug 和安全问题。给出具体的修复建议。",
    llm_config={"config_list": config_list},
)

# 主助手 Agent，附加嵌套对话
main_assistant = autogen.AssistantAgent(
    name="main_assistant",
    system_message="你是一个高级 Python 开发者。你写的每段代码都会经过内部审查后才提交。",
    llm_config={"config_list": config_list},
)

# 注册嵌套对话：主助手在每次回复前，先和审查员讨论
main_assistant.register_nested_chats(
    [
        {
            "recipient": internal_reviewer,
            "message": lambda recipient, messages, sender, config: (
                f"请审查以下代码/回复，找出问题并提供改进建议：\n\n"
                f"{messages[-1]['content'] if messages else '暂无内容'}"
            ),
            "summary_method": "last_msg",    # 取最后一条消息作为总结
            "max_turns": 2,
        },
    ],
    trigger=autogen.UserProxyAgent,  # 当收到用户代理的消息时触发
)

8.2 多层嵌套

# 安全审查（最内层）
security_agent = autogen.AssistantAgent(
    name="security",
    system_message="你是安全专家。检查代码的安全漏洞。",
    llm_config={"config_list": config_list},
)

# 代码审查（中间层，内嵌安全审查）
code_reviewer = autogen.AssistantAgent(
    name="reviewer",
    system_message="你是代码审查员。审查代码质量。",
    llm_config={"config_list": config_list},
)
code_reviewer.register_nested_chats(
    [{"recipient": security_agent, "max_turns": 1, "summary_method": "last_msg"}],
    trigger=autogen.AssistantAgent,
)

# 主助手（最外层，内嵌代码审查）
main = autogen.AssistantAgent(
    name="main",
    system_message="你是开发者。",
    llm_config={"config_list": config_list},
)
main.register_nested_chats(
    [{"recipient": code_reviewer, "max_turns": 2, "summary_method": "last_msg"}],
    trigger=autogen.UserProxyAgent,
)

9. 人机协作模式 Human-in-the-Loop

9.1 按需人类输入

# TERMINATE 模式：Agent 完成后询问人类是否满意
user_proxy = autogen.UserProxyAgent(
    name="human",
    human_input_mode="TERMINATE",
    code_execution_config={"work_dir": "workspace"},
    is_termination_msg=lambda msg: "TERMINATE" in msg.get("content", ""),
)

9.2 代码执行审批

def ask_human_approval(code: str) -> bool:
    """在执行代码前请求人类批准"""
    print(f"\n即将执行以下代码:\n{'='*50}\n{code}\n{'='*50}")
    response = input("是否允许执行？(y/n): ").strip().lower()
    return response == "y"

user_proxy = autogen.UserProxyAgent(
    name="user",
    human_input_mode="NEVER",
    code_execution_config={
        "work_dir": "workspace",
        "use_docker": True,
        "timeout": 60,
    },
)

9.3 混合模式

# 关键决策点需要人类确认，其他自动执行
class HybridUserProxy(autogen.UserProxyAgent):
    def get_human_input(self, prompt: str) -> str:
        """自定义人类输入逻辑"""
        if "确认" in prompt or "批准" in prompt:
            return input(f"[需要确认] {prompt}: ")
        return ""  # 空字符串 = 不需要人类输入

10. 自定义 Agent 开发

10.1 创建自定义 Agent

import autogen
from typing import Dict, List, Optional, Union

class DataAnalystAgent(autogen.ConversableAgent):
    """数据分析师 Agent —— 专注于数据分析任务"""
    
    DEFAULT_SYSTEM_MESSAGE = """你是一位专业的数据分析师。
你的核心能力：
1. 数据清洗与预处理
2. 统计分析与可视化
3. 趋势预测与异常检测
4. 撰写数据分析报告

输出规范：
- 使用 pandas 进行数据处理
- 使用 matplotlib/seaborn 生成图表
- 分析结论必须有数据支撑
- 报告格式清晰、结论明确"""
    
    def __init__(
        self,
        name: str = "data_analyst",
        llm_config: Optional[Dict] = None,
        **kwargs,
    ):
        super().__init__(
            name=name,
            system_message=self.DEFAULT_SYSTEM_MESSAGE,
            llm_config=llm_config,
            **kwargs,
        )
        self._register_default_tools()
    
    def _register_default_tools(self):
        """注册数据分析师默认工具"""
        
        def load_dataset(
            file_path: str,
            file_format: str = "csv",
        ) -> str:
            """加载数据集"""
            import pandas as pd
            loaders = {
                "csv": pd.read_csv,
                "excel": pd.read_excel,
                "json": pd.read_json,
            }
            if file_format not in loaders:
                return f"不支持的格式: {file_format}"
            df = loaders[file_format](file_path)
            return f"数据集加载成功。形状: {df.shape}\n列名: {list(df.columns)}\n前5行:\n{df.head().to_string()}"
        
        self.register_for_llm(
            name="load_dataset",
            description="加载数据集文件（支持 CSV、Excel、JSON）",
        )(load_dataset)

10.2 消息钩子

class LoggingAgent(autogen.AssistantAgent):
    """带日志记录的 Agent"""
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.conversation_log = []
    
    def receive(
        self,
        message: Union[Dict, str],
        sender: autogen.Agent,
        request_reply: Optional[bool] = None,
        silent: Optional[bool] = False,
    ):
        """重写 receive 方法，添加日志"""
        self.conversation_log.append({
            "from": sender.name,
            "to": self.name,
            "message": message if isinstance(message, str) else message.get("content", ""),
            "timestamp": __import__("datetime").datetime.now().isoformat(),
        })
        return super().receive(message, sender, request_reply, silent)
    
    def generate_reply(
        self,
        messages: Optional[List[Dict]] = None,
        sender: Optional[autogen.Agent] = None,
        **kwargs,
    ):
        """重写生成回复方法，添加日志"""
        reply = super().generate_reply(messages, sender, **kwargs)
        if reply:
            self.conversation_log.append({
                "from": self.name,
                "to": sender.name if sender else "unknown",
                "message": reply if isinstance(reply, str) else str(reply),
                "timestamp": __import__("datetime").datetime.now().isoformat(),
            })
        return reply
    
    def export_log(self, file_path: str):
        """导出对话日志"""
        import json
        with open(file_path, "w", encoding="utf-8") as f:
            json.dump(self.conversation_log, f, ensure_ascii=False, indent=2)

11. AutoGen Studio 可视化界面

11.1 安装与启动

# 安装 AutoGen Studio
pip install autogenstudio

# 启动 Web 界面
autogenstudio ui --port 8080 --host 0.0.0.0

启动后访问 http://localhost:8080，你将看到可视化界面。

11.2 核心功能

AutoGen Studio 提供以下可视化能力：

• Agent 配置：图形化创建和配置 Agent
• 工作流设计：拖拽式设计多 Agent 协作流程
• 对话测试：实时测试 Agent 对话效果
• 历史记录：查看和回放历史对话
• Gallery：社区分享的工作流模板

11.3 通过 API 创建工作流

import requests

# 通过 REST API 创建 Agent
BASE_URL = "http://localhost:8080/api"

# 创建 Agent 配置
agent_config = {
    "config": {
        "name": "code_reviewer",
        "agent_type": "assistant",
        "system_message": "你是一位代码审查专家。",
        "llm_config": {
            "config_list": [{"model": "gpt-4o"}],
        },
    }
}

response = requests.post(f"{BASE_URL}/agents", json=agent_config)
agent_id = response.json()["id"]

# 创建工作流
workflow = {
    "config": {
        "name": "code_review_workflow",
        "description": "自动化代码审查流程",
        "participants": [agent_id],
    }
}

response = requests.post(f"{BASE_URL}/workflows", json=workflow)

12. 流式输出与实时反馈

12.1 启用流式输出

import autogen

# 方式一：全局流式配置
llm_config = {
    "config_list": config_list,
    "stream": True,  # 启用流式输出
}

assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config=llm_config,
)

# 方式二：使用流式打印
from autogen.io import IOStream, IOConsole

# 设置自定义 IO 流
class CustomIOStream(IOConsole):
    def print(self, *objects, sep=" ", end="\n", flush=False):
        """自定义打印，可以接入 WebSocket 等"""
        content = sep.join(str(obj) for obj in objects)
        # 在这里可以将内容推送到前端
        super().print(*objects, sep=sep, end=end, flush=True)

IOStream.set_global(CustomIOStream())

12.2 自定义流式处理

import asyncio
from typing import AsyncGenerator

class StreamingAgent(autogen.AssistantAgent):
    """支持流式回调的 Agent"""
    
    def __init__(self, on_token_callback=None, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.on_token = on_token_callback
    
    async def a_generate_reply(
        self,
        messages=None,
        sender=None,
        **kwargs,
    ):
        """异步生成回复，支持流式回调"""
        # 调用 LLM 的流式 API
        response = await self.a_client.create(
            messages=messages or self._messages,
            stream=True,
        )
        
        full_response = ""
        async for chunk in response:
            if chunk.choices and chunk.choices[0].delta.content:
                token = chunk.choices[0].delta.content
                full_response += token
                if self.on_token:
                    await self.on_token(token)
        
        return full_response

# 使用示例
async def token_handler(token: str):
    print(token, end="", flush=True)

agent = StreamingAgent(
    name="streamer",
    on_token_callback=token_handler,
    llm_config={"config_list": config_list},
)

13. 错误处理与重试机制

13.1 基础错误处理

import autogen

# 配置重试参数
llm_config = {
    "config_list": config_list,
    "timeout": 120,                # 请求超时
    "retry_on_error": True,        # 错误时重试
    "max_retries": 3,              # 最大重试次数
    "cache_seed": None,            # 禁用缓存（调试时）
}

13.2 自定义错误处理

from autogen import Agent

class ResilientAgent(autogen.AssistantAgent):
    """带重试和降级机制的 Agent"""
    
    def generate_reply(self, messages=None, sender=None, **kwargs):
        max_attempts = 3
        for attempt in range(max_attempts):
            try:
                reply = super().generate_reply(messages, sender, **kwargs)
                return reply
            except Exception as e:
                error_msg = str(e)
                print(f"[{self.name}] 第 {attempt + 1} 次尝试失败: {error_msg}")
                
                if attempt < max_attempts - 1:
                    # 等待后重试
                    import time
                    wait_time = 2 ** attempt  # 指数退避
                    print(f"等待 {wait_time} 秒后重试...")
                    time.sleep(wait_time)
                else:
                    # 最后一次尝试失败，返回降级回复
                    return f"[错误] Agent {self.name} 无法生成回复。错误: {error_msg}"

13.3 对话级错误处理

def safe_initiate_chat(
    user_proxy,
    assistant,
    message: str,
    max_retries: int = 3,
):
    """带重试的对话发起"""
    for attempt in range(max_retries):
        try:
            result = user_proxy.initiate_chat(
                assistant,
                message=message,
                max_turns=10,
            )
            return result
        except Exception as e:
            print(f"对话失败 (第 {attempt + 1} 次): {e}")
            if attempt < max_retries - 1:
                # 重置对话历史后重试
                user_proxy.reset()
                assistant.reset()
            else:
                raise RuntimeError(f"对话在 {max_retries} 次尝试后仍然失败") from e

14. 生产部署与扩展

14.1 异步执行

import asyncio

async def run_async_chat():
    """异步运行对话"""
    assistant = autogen.AssistantAgent(
        name="assistant",
        llm_config={"config_list": config_list},
    )
    
    user = autogen.UserProxyAgent(
        name="user",
        human_input_mode="NEVER",
        code_execution_config=False,
    )
    
    # 异步发起对话
    result = await user.a_initiate_chat(
        assistant,
        message="写一个异步爬虫程序。",
        max_turns=5,
    )
    return result

# 运行
result = asyncio.run(run_async_chat())

14.2 日志与监控

import logging
from autogen.runtime_logging import start_logging, stop_logging

# 启用结构化日志
logging_session_id = start_logging(
    logger_type="sqlite",     # 或 "file"
    config={"dbname": "logs.db"},
)

# 运行 Agent 对话...

# 停止日志
stop_logging()

# 查询日志
import sqlite3
conn = sqlite3.connect("logs.db")
cursor = conn.execute("SELECT * FROM chat_completions LIMIT 10")
for row in cursor.fetchall():
    print(row)

14.3 对话状态持久化

import json
from pathlib import Path

class PersistentAgent(autogen.UserProxyAgent):
    """支持状态持久化的 Agent"""
    
    def __init__(self, state_dir: str = "./agent_states", *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.state_dir = Path(state_dir)
        self.state_dir.mkdir(parents=True, exist_ok=True)
    
    def save_state(self):
        """保存当前对话状态"""
        state = {
            "name": self.name,
            "chat_messages": {
                agent.name: messages
                for agent, messages in self._oai_messages.items()
            },
        }
        path = self.state_dir / f"{self.name}_state.json"
        with open(path, "w", encoding="utf-8") as f:
            json.dump(state, f, ensure_ascii=False, indent=2)
    
    def load_state(self):
        """加载对话状态"""
        path = self.state_dir / f"{self.name}_state.json"
        if path.exists():
            with open(path, "r", encoding="utf-8") as f:
                return json.load(f)
        return None

15. 实战项目 1：自动化代码审查系统

构建一个完整的多 Agent 代码审查系统，包含安全检查、性能分析和代码质量评估。

import autogen
import json
from datetime import datetime

# ===== LLM 配置 =====
config_list = autogen.config_list_from_json(
    "OAI_CONFIG_LIST",
    filter_dict={"model": ["gpt-4o"]},
)

llm_config = {"config_list": config_list, "temperature": 0.3}

# ===== 创建审查团队 =====

# 1. 安全审查员
security_reviewer = autogen.AssistantAgent(
    name="security_reviewer",
    system_message="""你是安全审查专家。审查代码时关注：
1. SQL 注入、XSS、CSRF 等安全漏洞
2. 敏感信息泄露（硬编码密码、API Key 等）
3. 不安全的依赖和函数调用
4. 权限控制问题
5. 输入验证缺失

输出格式：
- 🚨 严重问题（必须修复）
- ⚠️ 中等问题（建议修复）
- 💡 低风险建议（可选优化）
每条都要给出修复方案。""",
    llm_config=llm_config,
)

# 2. 性能分析师
performance_analyst = autogen.AssistantAgent(
    name="performance_analyst",
    system_message="""你是性能分析专家。审查代码时关注：
1. 时间复杂度和空间复杂度
2. 不必要的循环和递归
3. 数据库查询优化（N+1 问题等）
4. 内存泄漏风险
5. 并发/异步优化机会

输出格式：
- 📊 性能评分（1-10）
- 🔍 瓶颈分析
- ⚡ 优化建议（附代码示例）""",
    llm_config=llm_config,
)

# 3. 代码质量审查员
quality_reviewer = autogen.AssistantAgent(
    name="quality_reviewer",
    system_message="""你是代码质量审查员。审查代码时关注：
1. 代码风格和命名规范
2. 设计模式的合理使用
3. SOLID 原则遵守情况
4. 文档和注释完整性
5. 测试覆盖率建议

输出格式：
- 📝 代码质量评分（1-10）
- ✅ 做得好的地方
- ❌ 需要改进的地方
- 🔧 重构建议""",
    llm_config=llm_config,
)

# 4. 审查汇总员
summarizer = autogen.AssistantAgent(
    name="review_lead",
    system_message="""你是代码审查团队负责人。
你的任务是：
1. 汇总所有审查员的意见
2. 按优先级排序所有问题
3. 给出最终的代码质量评分
4. 提供明确的「通过/需要修改/拒绝」结论
5. 输出结构化的审查报告

报告格式：

代码审查报告

总体评分：X/10

结论：通过/需要修改/拒绝

关键问题（X 个）

建议改进（X 个）

亮点

    llm_config=llm_config,
)

# 5. 用户代理
user_proxy = autogen.UserProxyAgent(
    name="code_submitter",
    human_input_mode="NEVER",
    code_execution_config=False,
    is_termination_msg=lambda msg: "审查报告" in msg.get("content", ""),
)

# ===== 创建群聊 =====
groupchat = autogen.GroupChat(
    agents=[user_proxy, security_reviewer, performance_analyst, quality_reviewer, summarizer],
    messages=[],
    max_round=8,
    speaker_selection_method="auto",
)

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

# ===== 启动审查 =====
code_to_review = """
import sqlite3
import os

API_KEY = "sk-1234567890abcdef"  # API Key

def get_user(user_id):
    conn = sqlite3.connect("users.db")
    cursor = conn.cursor()
    query = f"SELECT * FROM users WHERE id = '{user_id}'"
    cursor.execute(query)
    result = cursor.fetchone()
    conn.close()
    return result

def process_all_users():
    users = get_all_users()
    results = []
    for user in users:
        data = get_user(user[0])
        results.append(data)
    return results

def get_all_users():
    conn = sqlite3.connect("users.db")
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    results = cursor.fetchall()
    conn.close()
    return results

if __name__ == "__main__":
    user = get_user("1' OR '1'='1")
    print(user)
"""

user_proxy.initiate_chat(
    manager,
    message=f"请对以下 Python 代码进行全面审查：\n```python\n{code_to_review}\n```\n请各位审查员依次发表意见，最后由审查负责人汇总。",
)

16. 实战项目 2：智能研究助手

构建一个能够自主搜索、分析、总结并生成研究报告的多 Agent 系统。

import autogen
import json
from typing import Annotated

# ===== 配置 =====
config_list = autogen.config_list_from_json("OAI_CONFIG_LIST")
llm_config = {"config_list": config_list, "temperature": 0.5}

# ===== 定义工具 =====

def web_search(
    query: Annotated[str, "搜索关键词"],
    num_results: Annotated[int, "返回结果数量"] = 5,
) -> str:
    """模拟网络搜索（实际可接入搜索 API）"""
    results = [
        {
            "title": f"搜索结果 {i+1}: {query}相关研究",
            "url": f"https://scholar.example.com/paper/{i+1}",
            "snippet": f"关于{query}的最新研究成果表明...",
        }
        for i in range(num_results)
    ]
    return json.dumps(results, ensure_ascii=False)

def summarize_paper(
    url: Annotated[str, "论文 URL"],
) -> str:
    """模拟论文摘要提取"""
    return f"论文摘要: 本文研究了相关领域的重要问题，提出了创新的解决方案..."

def save_report(
    title: Annotated[str, "报告标题"],
    content: Annotated[str, "报告内容"],
    format: Annotated[str, "输出格式 (md/pdf)"] = "md",
) -> str:
    """保存研究报告"""
    filename = f"report_{title.replace(' ', '_')}.{format}"
    with open(filename, "w", encoding="utf-8") as f:
        f.write(content)
    return f"报告已保存: {filename}"

# ===== 创建研究团队 =====

# 研究规划师
planner = autogen.AssistantAgent(
    name="research_planner",
    system_message="""你是研究规划师。
收到研究主题后：
1. 分解研究问题为 3-5 个子问题
2. 为每个子问题制定搜索策略
3. 确定研究方法和时间安排
4. 与搜索分析师协调执行

输出结构化的研究计划。""",
    llm_config=llm_config,
)

# 搜索分析师
searcher = autogen.AssistantAgent(
    name="search_analyst",
    system_message="""你是搜索分析师。
根据研究计划：
1. 使用 web_search 工具搜索相关资料
2. 使用 summarize_paper 工具提取论文摘要
3. 筛选和评估信息质量
4. 整理搜索结果并标注来源""",
    llm_config=llm_config,
)

# 研究撰写员
writer = autogen.AssistantAgent(
    name="research_writer",
    system_message="""你是研究撰写员。
根据搜索分析师提供的资料：
1. 撰写结构化的研究报告
2. 包含摘要、正文、结论、参考文献
3. 确保引用准确、论述有据
4. 使用 save_report 工具保存最终报告""",
    llm_config=llm_config,
)

# 质量审核员
qa_reviewer = autogen.AssistantAgent(
    name="qa_reviewer",
    system_message="""你是质量审核员。
审核研究报告：
1. 检查论述逻辑是否连贯
2. 验证引用是否准确
3. 评估研究深度是否足够
4. 提出改进建议
5. 满意后说 TERMINATE""",
    llm_config=llm_config,
)

# 用户代理
user = autogen.UserProxyAgent(
    name="user",
    human_input_mode="NEVER",
    code_execution_config={"work_dir": "research_output", "use_docker": False},
)

# 注册工具
for agent in [planner, searcher, writer, qa_reviewer]:
    agent.register_for_llm(name="web_search", description="搜索互联网获取研究资料")(web_search)
    agent.register_for_llm(name="summarize_paper", description="提取论文摘要")(summarize_paper)
    agent.register_for_llm(name="save_report", description="保存研究报告")(save_report)

user.register_for_execution(name="web_search")(web_search)
user.register_for_execution(name="summarize_paper")(summarize_paper)
user.register_for_execution(name="save_report")(save_report)

# ===== 群聊配置 =====
groupchat = autogen.GroupChat(
    agents=[user, planner, searcher, writer, qa_reviewer],
    messages=[],
    max_round=15,
    speaker_selection_method="auto",
)

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

# ===== 启动研究 =====
user.initiate_chat(
    manager,
    message="请研究「大语言模型在医疗领域的应用」，生成一份完整的研究报告。重点关注：诊断辅助、药物发现、医学影像分析三个方向。",
)

print("\n✅ 研究报告生成完成！")

总结

AutoGen 通过多 Agent 对话协作，将复杂的 AI 任务分解为可管理的子任务。核心要点：

下一步学习建议：

1. 从双人对话开始，理解基本交互模式
2. 尝试 GroupChat，体验多 Agent 协作
3. 结合实际场景开发自定义 Agent
4. 部署到生产环境，关注错误处理和监控

官方文档：https://microsoft.github.io/autogen/ GitHub：https://github.com/microsoft/autogen

本教程基于 AutoGen 0.2+ 版本编写。框架仍在快速迭代，建议关注官方更新日志获取最新 API 变更。