AutoGen 精通教程：从零到企业级多 Agent 系统架构师

1. AutoGen 是什么

1.1 概述

AutoGen 是微软研究院推出的一款开源多 Agent 对话框架，旨在简化多个 LLM Agent（大语言模型智能体）之间的协作。它让 AI Agent 之间可以相互对话、交换信息、协作完成复杂任务。

1.2 核心定位

维度	说明
框架类型	多 Agent 对话编排框架
主要语言	Python
核心思想	Agent 通过对话（Conversation）进行协作
适用场景	复杂任务分解、多角色协作、代码生成、数据分析、自动化工作流
特色能力	人类反馈介入、代码执行沙箱、工具调用、群组聊天

1.3 为什么选择 AutoGen？

• ✅ 企业级: 微软出品，有微软云生态支持
• ✅ 灵活: 支持多种 Agent 类型和对话模式
• ✅ 可扩展: 自定义 Agent、工具、模型
• ✅ 人类参与: 支持人在回路（Human-in-the-Loop）
• ✅ 代码执行: 内置代码执行沙箱
• ✅ 活跃社区: GitHub 45k+ Stars

2. 核心概念与架构

2.1 核心组件

2.2 关键概念一览

概念	说明	类比
Agent	一个 AI 智能体，配有 LLM、工具、系统提示词	一个「员工」
Conversation	Agent 之间的消息交换	员工之间的「对话」
Message	对话中的一条消息（文本/函数调用/结果）	一句「话」
Tool	Agent 可以调用的函数/API	员工使用的「工具」
Team	一组 Agent 的编排执行单元	一个「项目组」
GroupChat	多 Agent 群聊模式	一个「群聊」
RoundRobin	轮流发言的对话模式	轮流「发言」
SelectorGroupChat	由选择器决定谁下一个发言	主持人「点名」
UserProxyAgent	代表人类用户的代理，可执行代码	人类的「秘书」
AssistantAgent	由 LLM 驱动的助手 Agent	一个「专家」
CodeExecutorAgent	专门执行代码的 Agent	「程序员」

2.3 消息生命周期

用户输入
    │
    ▼
┌─────────────┐
│  Agent A     │──► 思考（调用 LLM）
│  (接收消息)  │──► 决定（是否调用工具）
└──────┬──────┘
       │
       ├──► 调用工具 ──► 返回结果 ──► 继续思考
       │
       ▼
    发送消息给 Agent B
       │
       ▼
┌─────────────┐
│  Agent B     │──► 类似流程
└──────┬──────┘
       │
       ▼
    最终输出给用户

---

3. 环境搭建

3.1 安装

# 基础安装
pip install pyautogen

# 安装全部依赖（推荐）
pip install "pyautogen[all]"

# 按需安装特定功能
pip install "pyautogen[retrievechat]"   # RAG 检索对话
pip install "pyautogen[mathchat]"       # 数学问题求解
pip install "pyautogen[blendsearch]"    # 超参数调优
pip install "pyautogen[teachable]"      # 可教学 Agent
pip install "pyautogen[lmm]"            # 多模态支持
pip install "pyautogen[websockets]"     # WebSocket 支持
pip install "pyautogen[graph]"          # 图功能（实验性）
pip install "pyautogen[redis]"          # Redis 支持
pip install "pyautogen[long_context]"   # 长上下文支持

3.2 验证安装

import autogen

print(autogen.__version__)
# 应输出类似 0.4.0 的版本号

3.3 配置 LLM

方式一：环境变量（推荐）

export OPENAI_API_KEY="sk-your-key-here"
# 或其他厂商
export OPENAI_BASE_URL="https://api.example.com/v1"

方式二：配置文件 OAI_CONFIG_LIST

[
    {
        "model": "gpt-4o",
        "api_key": "sk-your-key-here",
        "tags": ["gpt-4", "default"]
    },
    {
        "model": "gpt-4o-mini",
        "api_key": "sk-your-key-here",
        "tags": ["gpt-4", "fast", "cheap"]
    },
    {
        "model": "deepseek-chat",
        "api_key": "sk-deepseek-key",
        "base_url": "https://api.deepseek.com/v1",
        "api_type": "openai",
        "tags": ["deepseek", "alternative"]
    },
    {
        "model": "qwen2.5-72b-instruct",
        "api_key": "your-api-key",
        "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
        "api_type": "openai",
        "tags": ["qwen", "china"]
    }
]

方式三：代码中直接配置

config_list = [
    {
        'model': 'gpt-4o',
        'api_key': 'sk-xxx',
    }
]

3.4 快速测试

import autogen

config_list = autogen.config_list_from_json(
    "OAI_CONFIG_LIST",
    filter_dict={"tags": ["gpt-4"]},  # 可选过滤
)

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

user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "coding", "use_docker": False},
)

user_proxy.initiate_chat(
    assistant,
    message="Python 的列表推导式和生成器表达式有什么区别？",
)

---

4. 基础实战：第一个多 Agent 系统

4.1 最简单的两个 Agent 对话

import autogen

# 配置 LLM
llm_config = {
    "config_list": [
        {"model": "gpt-4o", "api_key": "sk-xxx"}
    ],
    "temperature": 0.7,
}

# 创建助手 Agent
assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config=llm_config,
    system_message="你是一个 Python 编程专家。",
)

# 创建用户代理（模拟人类）
user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",  # 从不请求人类输入
    max_consecutive_auto_reply=10,
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False,   # 不使用 Docker
    },
    system_message="你是用户代理，可以执行代码。",
)

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

4.2 关键参数详解

human_input_mode — 人类介入模式：

模式	行为
"ALWAYS"	每次 Agent 回复前都询问人类
"TERMINATE"	当 Agent 发出终止信号时询问人类
"NEVER"	从不询问，完全自动

max_consecutive_auto_reply — 最大连续自动回复次数：

• 防止 Agent 无限对话下去
• 一般设置为 5-20

code_execution_config — 代码执行配置：

code_execution_config = {
    "work_dir": "coding",          # 工作目录
    "use_docker": False,           # 是否使用 Docker
    "timeout": 60,                 # 超时时间（秒）
    "last_n_messages": 3,          # 只考虑最后几条消息中的代码
}

4.3 理解对话流程

用户: 写一个计算斐波那契数列的函数
  │
  ▼
UserProxy (user_proxy)
  │  转发消息给 assistant
  ▼
Assistant (assistant)
  │  生成回复: 代码 + 说明
  │  示例回复: 建议使用Python代码实现...
  │  包含代码块: ```python ... ```
  ▼
UserProxy
  │  检测到代码块 → 执行代码
  │  返回执行结果
  ▼
Assistant
  │  分析结果 → 继续回复或终止
  ▼
...（循环直到终止）

---

5. Agent 类型详解

5.1 ConversableAgent（基类）

所有 Agent 的基类，提供对话的基本能力。

from autogen import ConversableAgent

agent = ConversableAgent(
    name="basic_agent",
    llm_config=None,          # 可选，不使用 LLM
    human_input_mode="NEVER",
    system_message="你是一个通用 Agent。",
)

5.2 AssistantAgent（助手 Agent）

由 LLM 驱动的助手，是使用最广泛的 Agent 类型。

assistant = autogen.AssistantAgent(
    name="coding_expert",
    system_message="""你是顶级 Python 开发者。
    请提供简洁、高效、有类型注解的代码。
    先解释思路，再提供完整代码。""",
    llm_config=llm_config,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
)

5.3 UserProxyAgent（用户代理 Agent）

代表人类用户的代理，可以执行代码、请求人类输入。

user_proxy = autogen.UserProxyAgent(
    name="admin",
    human_input_mode="TERMINATE",
    max_consecutive_auto_reply=5,
    code_execution_config={
        "work_dir": "workspace",
        "use_docker": "python:3-slim",  # 使用 Docker 镜像
        "timeout": 120,
    },
)

5.4 GroupChatManager（群聊管理器）

管理多个 Agent 之间的群组对话。

from autogen import GroupChat, GroupChatManager

groupchat = GroupChat(
    agents=[agent1, agent2, agent3],
    messages=[],
    max_round=12,
    speaker_selection_method="auto",  # 自动选择发言者
)

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

5.5 自定义 Agent

从基类继承创建自定义 Agent：

class MyCustomAgent(autogen.ConversableAgent):
    def __init__(self, name, **kwargs):
        super().__init__(name=name, **kwargs)
        self.custom_state = {}

    def process_message(self, message):
        """自定义消息处理逻辑"""
        # 在这里添加自定义逻辑
        return super().generate_reply(messages=[message])

# 使用
my_agent = MyCustomAgent(
    name="custom_agent",
    llm_config=llm_config,
)

---

6. 对话模式与流程控制

6.1 一对一对话（Two-Agent Chat）

# 最简单的对话模式
user_proxy.initiate_chat(
    assistant,
    message="帮我分析这段代码的性能瓶颈...",
    clear_history=True,  # 清除历史记录
)

对话控制方法：

# 发送后续消息
user_proxy.send(
    recipient=assistant,
    message="能否再优化一下？",
)

# 获取对话历史
history = user_proxy.chat_messages[assistant]

# 注册回复函数（自定义处理收到的消息）
def my_reply_func(recipient, messages, sender, config):
    content = messages[-1]["content"]
    return f"收到消息: {content[:50]}..."

user_proxy.register_reply(
    [autogen.AssistantAgent, autogen.UserProxyAgent],
    reply_func=my_reply_func,
    config={"custom_param": "value"},
    position=0,  # 优先级
)

6.2 群组对话（GroupChat）

import autogen
from autogen import GroupChat, GroupChatManager

# 创建多个 Agent
product_manager = autogen.AssistantAgent(
    name="product_manager",
    system_message="你是产品经理，负责定义需求。用中文回答。",
    llm_config=llm_config,
)

developer = autogen.AssistantAgent(
    name="developer",
    system_message="你是全栈开发者，负责实现功能。用中文回答。",
    llm_config=llm_config,
)

tester = autogen.AssistantAgent(
    name="tester",
    system_message="你是质量保证工程师，负责测试。用中文回答。",
    llm_config=llm_config,
)

# 创建群聊
groupchat = GroupChat(
    agents=[product_manager, developer, tester],
    messages=[],
    max_round=20,
    speaker_selection_method="auto",  # 可选: "auto", "round_robin", "random"
    allow_repeat_speaker=False,       # 不允许同一人连续发言
    enable_clear_history=True,        # 启用历史清理
)

# 创建群聊管理器
manager = GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config,
)

# 用户代理发起对话
user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="ALWAYS",
    code_execution_config=False,
)

user_proxy.initiate_chat(
    manager,
    message="我们需要开发一个 Todo 应用，请讨论并制定方案。",
)

6.3 发言者选择策略

策略	说明	适用场景
"auto"	LLM 自动选择下一个发言人	灵活、复杂的讨论
"round_robin"	轮流发言	结构化的评审流程
"random"	随机选择	探索性讨论
"manual"	人类手动选择	需要人类控制的场景

6.4 终止条件控制

# 方式一：消息内容终止
assistant = autogen.AssistantAgent(
    name="assistant",
    llm_config=llm_config,
    is_termination_msg=lambda x: x.get("content", "").rstrip().endswith("TERMINATE"),
)

# 方式二：最大轮数终止
groupchat = GroupChat(
    agents=[...],
    messages=[],
    max_round=10,  # 最多10轮后自动终止
)

# 方式三：自定义终止函数
def custom_termination(message):
    content = message.get("content", "")
    if "任务完成" in content:
        return True
    if message.get("role") == "user" and "停止" in content:
        return True
    return False

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

6.5 嵌套对话（Sequential Chat）

from autogen import SequentialGroupChat

# 按顺序执行多个对话阶段
stage1 = SequentialGroupChat(
    agents=[product_manager, developer],
    max_round=5,
)

stage2 = SequentialGroupChat(
    agents=[developer, tester],
    max_round=5,
)

# 顺序执行
stage1.run(message="定义需求")
stage2.run(message="Review 代码")

---

7. 工具与函数调用

7.1 注册工具函数

import autogen
from typing import Annotated

# 定义工具函数
def get_weather(
    city: Annotated[str, "城市名称"],
    date: Annotated[str, "日期，格式 YYYY-MM-DD"],
) -> str:
    """获取指定城市的天气信息"""
    # 模拟天气查询
    weather_data = {
        "北京": {"2026-05-26": "晴，25°C"},
        "上海": {"2026-05-26": "多云，28°C"},
        "深圳": {"2026-05-26": "雷阵雨，30°C"},
    }
    return weather_data.get(city, {}).get(date, "暂无数据")

def search_database(
    query: Annotated[str, "搜索关键词"],
    limit: Annotated[int, "返回结果数量"] = 5,
) -> str:
    """搜索本地数据库"""
    # 模拟数据库查询
    results = [
        f"结果{i}: 关于{query}的相关记录" for i in range(limit)
    ]
    return "\n".join(results)

# 创建带工具的 AssistantAgent
assistant = autogen.AssistantAgent(
    name="assistant_with_tools",
    llm_config={
        "config_list": [...],
        "tools": [
            {
                "function": get_weather,
                "name": "get_weather",
                "description": "获取天气信息",
            },
            {
                "function": search_database,
                "name": "search_database",
                "description": "搜索数据库",
            },
        ],
    },
)

7.2 使用 register_for_llm 和 register_for_execution

from autogen import register_function

# 定义一个函数
def calculate_bmi(weight_kg: float, height_m: float) -> float:
    """计算 BMI 指数"""
    return weight_kg / (height_m ** 2)

# 创建 Agent
assistant = autogen.AssistantAgent(
    name="assistant", llm_config=llm_config
)
user_proxy = autogen.UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    code_execution_config=False,
)

# 注册函数（同时注册给 LLM 和执行器）
register_function(
    calculate_bmi,
    caller=assistant,      # 调用者（LLM 会知道这个函数）
    executor=user_proxy,    # 执行者（实际执行函数）
    name="calculate_bmi",   # 函数名称
    description="计算身体质量指数",
)

7.3 异步工具

import asyncio
from typing import Annotated

async def async_fetch_data(url: Annotated[str, "API URL"]) -> str:
    """异步获取数据"""
    await asyncio.sleep(1)  # 模拟网络请求
    return f"从 {url} 获取的数据"

# 创建支持异步工具的 Agent
assistant = autogen.AssistantAgent(
    name="async_assistant",
    llm_config={
        "config_list": [...],
        "tools": [{
            "function": async_fetch_data,
            "name": "fetch_data",
            "description": "异步获取数据",
        }],
    },
)

7.4 工具使用最佳实践

# ✅ 好的实践：提供详细类型注解和描述
def search_products(
    keyword: Annotated[str, "搜索关键词，如'笔记本电脑'"],
    category: Annotated[str, "商品分类，可选: 电子产品/服装/食品/图书"] = "全部",
    min_price: Annotated[float, "最低价格"] = 0.0,
    max_price: Annotated[float, "最高价格"] = 99999.0,
) -> list:
    """搜索商品"""
    ...

# ⚠️ 避免：没有类型注解和描述
def search_products(keyword, category, min_price, max_price):
    ...

---

8. 团队协作模式（Team）

8.1 Team 架构概述

AutoGen 0.4+ 引入了 Team 概念，作为更高层次的编排单元。

from autogen import AssistantAgent, UserProxyAgent
from autogen.agentchat.teams import RoundRobinGroupChat, SelectorGroupChat
from autogen.agentchat.teams import MagenticOneGroupChat

8.2 RoundRobinGroupChat（轮流发言）

from autogen.agentchat.teams import RoundRobinGroupChat

# 创建团队
team = RoundRobinGroupChat(
    participants=[agent1, agent2, agent3],
    termination_condition="max_messages",  # 或自定义条件
    max_turns=10,
)

# 运行团队任务
result = await team.run(task="分析这份市场报告并给出建议")
print(result.messages)  # 输出所有消息
print(result.summary)   # 输出总结

8.3 SelectorGroupChat（选择器发言）

from autogen.agentchat.teams import SelectorGroupChat

team = SelectorGroupChat(
    participants=[pm_agent, dev_agent, qa_agent],
    model_client=model_client,  # 用于选择下一个发言者的模型
    termination_condition="max_messages",
    max_turns=15,
    selector_prompt="""请根据当前讨论进展，选择下一个最合适的发言者。
    产品经理: 当需要定义需求或决策时
    开发者: 当需要技术实现或代码时
    QA: 当需要测试或质量检查时
    """,
)

result = await team.run(task="设计并实现一个登录功能")

8.4 MagenticOneGroupChat（动态领导模式）

from autogen.agentchat.teams import MagenticOneGroupChat

team = MagenticOneGroupChat(
    participants=[analyst, coder, reviewer],
    model_client=model_client,
    max_turns=20,
    # MagenticOne 使用动态领导模式
    # 一个 Agent 作为"指挥"协调其他人
)

result = await team.run(task="从 API 获取数据、分析并生成报告")

8.5 团队终止条件

from autogen.agentchat.teams import (
    MaxMessagesTermination,
    TextMentionTermination,
    TimeoutTermination,
    OrTermination,
    AndTermination,
)

# 单个条件
msg_termination = MaxMessagesTermination(max_messages=50)
text_termination = TextMentionTermination(text="TERMINATE")
timeout = TimeoutTermination(timeout=300)  # 5分钟超时

# 组合条件（或）
combined = OrTermination(
    [msg_termination, text_termination, timeout]
)

# 组合条件（与）
strict = AndTermination(
    [text_termination, MaxMessagesTermination(max_messages=10)]
)

team = RoundRobinGroupChat(
    participants=[...],
    termination_condition=combined,
)

8.6 任务与结果

# 同步运行
result = team.run(task="分析数据")

# 异步运行
result = await team.run(task="分析数据")

print(f"任务状态: {result.status}")
print(f"总结: {result.summary}")
print(f"消息数: {len(result.messages)}")
print(f"耗时: {result.duration}秒")

# 流式运行（实时查看消息）
async for message in team.run_stream(task="分析数据"):
    print(f"{message.source}: {message.content[:100]}")

---

9. 模型配置与 LLM 集成

9.1 支持的 LLM 提供商

AutoGen 原生支持 OpenAI 兼容 API 格式的所有服务商：

提供商	API 类型	配置示例
OpenAI (GPT-4, GPT-4o)	openai	model: gpt-4o
Azure OpenAI	azure	api_type: azure
Anthropic (Claude)	anthropic	api_type: anthropic
Google (Gemini)	google	api_type: google
DeepSeek	openai	base_url: https://api.deepseek.com/v1
通义千问 (Qwen)	openai	base_url: https://dashscope.aliyuncs.com/compatible-mode/v1
Moonshot (月之暗面)	openai	base_url: https://api.moonshot.cn/v1
智谱 GLM	openai	base_url: https://open.bigmodel.cn/api/paas/v4
百度文心	openai	base_url: https://aip.baidubce.com
本地 Ollama	openai	base_url: http://localhost:11434/v1
本地 vLLM	openai	base_url: http://localhost:8000/v1
Groq	openai	base_url: https://api.groq.com/openai/v1

9.2 配置过滤与标签

# 从 JSON 文件加载并过滤
config_list = autogen.config_list_from_json(
    "OAI_CONFIG_LIST",
    filter_dict={
        "model": ["gpt-4o", "gpt-4o-mini"],
        "tags": ["default"],
    },
)

# 从环境变量加载
config_list = autogen.config_list_from_dotenv(
    ".env",
    filter_dict={"model": ["gpt-4o"]},
)

# 从特定模型列表
config_list = autogen.config_list_from_models(
    model_list=["gpt-4o", "gpt-4o-mini"],
    api_key="sk-xxx",
)

9.3 模型客户端（Model Client）

AutoGen 0.4+ 支持 ModelClient 抽象：

from autogen import OpenAIWrapperClient

# 创建模型客户端
model_client = OpenAIWrapperClient(
    model="gpt-4o",
    api_key="sk-xxx",
    temperature=0.7,
    max_tokens=4096,
)

# 在 Agent 中使用
assistant = autogen.AssistantAgent(
    name="assistant",
    model_client=model_client,
)

9.4 多模型 Fallback 策略

# 配置多个模型作为后备
llm_config = {
    "config_list": [
        {"model": "gpt-4o", "api_key": "sk-xxx", "tags": ["primary"]},
        {"model": "gpt-4o-mini", "api_key": "sk-xxx", "tags": ["fallback"]},
        {"model": "deepseek-chat", "api_key": "sk-ds", "base_url": "https://api.deepseek.com/v1", "tags": ["fallback"]},
    ],
    "temperature": 0.7,
    "timeout": 60,
    "max_retries": 3,
}

# AutoGen 会自动在前一个失败时尝试下一个

9.5 自定义模型客户端

from autogen import ModelClient

class MyCustomClient(ModelClient):
    def __init__(self, config, **kwargs):
        self.config = config
        self.model = config["model"]

    def send(self, messages, **kwargs):
        # 实现自定义模型调用逻辑
        response = ...  # 调用你的模型
        return response

    def parse(self, response):
        # 解析响应
        return {"content": response.text, "role": "assistant"}

# 注册使用
assistant = autogen.AssistantAgent(
    name="custom_model_agent",
    llm_config={
        "config_list": [{"model": "my-model", "model_client_cls": MyCustomClient}],
    },
)

---

10. 高级话题

10.1 代码执行沙箱

Docker 沙箱（推荐生产环境）

user_proxy = autogen.UserProxyAgent(
    name="sandbox_agent",
    code_execution_config={
        "work_dir": "coding",
        "use_docker": "python:3.11-slim",  # 指定 Docker 镜像
        "timeout": 60,
        "last_n_messages": 3,
    },
)

自定义 Docker 镜像

# Dockerfile
FROM python:3.11-slim
RUN pip install numpy pandas matplotlib requests scikit-learn

本地执行（开发环境）

code_execution_config = {
    "work_dir": "coding",
    "use_docker": False,  # 本地执行
    "timeout": 30,
}

10.2 人在回路（Human-in-the-Loop）

# 方式一：每次询问人类
agent = autogen.UserProxyAgent(
    name="human",
    human_input_mode="ALWAYS",  # 每次回复前都询问
)

# 方式二：只在终止时询问
agent = autogen.UserProxyAgent(
    name="human",
    human_input_mode="TERMINATE",  # Agent 想终止时询问
)

# 方式三：自定义人类输入函数
def custom_human_input(prompt):
    """自定义人类输入逻辑"""
    # 可以从队列、WebSocket、API 等获取输入
    response = input(f"[自定义输入] {prompt}")
    return response

agent = autogen.UserProxyAgent(
    name="human",
    human_input_func=custom_human_input,
)

10.3 记忆与上下文管理

# 自动摘要（防止上下文溢出）
assistant = autogen.AssistantAgent(
    name="memory_agent",
    llm_config={
        "config_list": [...],
        "max_consecutive_auto_reply": 20,
    },
    # 使用总结功能管理长对话
    summarizer_config={
        "model": "gpt-4o-mini",  # 用于总结的模型
        "max_tokens": 1024,
        "summary_role": "system",
    },
)

# 手动清理历史
user_proxy.initiate_chat(
    assistant,
    message="新任务...",
    clear_history=True,  # 清除历史开始新对话
)

10.4 反思与自我修正

# 创建具有反思能力的 Agent
reflective_assistant = autogen.AssistantAgent(
    name="reflective_assistant",
    system_message="""你是一个具有反思能力的 AI 助手。
    每次回答时，分三步：
    1. 思考：分析问题和可能的方案
    2. 回答：给出你的答案
    3. 反思：检查你的答案，指出可能的不足并改进
    
    如果发现错误，主动承认并修正。""",
    llm_config=llm_config,
)

# 创建"批评家" Agent 进行同行评审
critic = autogen.AssistantAgent(
    name="critic",
    system_message="""你是批评家。请严格审查对方的工作，
    指出问题、逻辑漏洞和改进建议。
    用中文回答。""",
    llm_config=llm_config,
)

# 让批评家评审
user_proxy.initiate_chat(
    critic,
    message=f"请评审以下内容：\n{assistant_output}",
)

10.5 多模态支持

# 使用支持多模态的模型
llm_config = {
    "config_list": [
        {
            "model": "gpt-4o",  # 多模态模型
            "api_key": "sk-xxx",
        }
    ],
}

# Agent 可以处理图片
assistant = autogen.AssistantAgent(
    name="multimodal_assistant",
    llm_config=llm_config,
)

# 在消息中包含图片
user_proxy.initiate_chat(
    assistant,
    message="""
    请分析这张图片中的内容：
    <image>https://example.com/photo.jpg</image>
    或者使用本地图片路径
    """,
)

10.6 RAG（检索增强生成）

from autogen import RetrieveUserProxyAgent, RetrieveAssistantAgent

# 创建检索助手
rag_assistant = RetrieveAssistantAgent(
    name="rag_assistant",
    system_message="你是一个基于文档库的问答助手。",
    llm_config=llm_config,
)

# 创建检索用户代理
rag_user_proxy = RetrieveUserProxyAgent(
    name="rag_user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,
    retrieve_config={
        "task": "qa",       # 任务类型
        "docs_path": "./documents/",  # 文档路径
        "chunk_token_size": 2000,
        "model": llm_config["config_list"][0]["model"],
        "embedding_model": "text-embedding-3-small",
        "vector_db": "chroma",  # 可选: chroma, qdrant, pgvector
        "overwrite": False,
        "get_or_create": True,
    },
)

# 开始对话
rag_user_proxy.initiate_chat(
    rag_assistant,
    message="根据文档库，解释什么是 Agentic RAG？",
    n_results=3,  # 返回最相关的3个文档块
)

10.7 错误处理与重试

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class RobustAssistant(autogen.AssistantAgent):
    """带自动重试的稳健 Agent"""
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
    )
    def generate_reply(self, messages, sender=None, **kwargs):
        try:
            return super().generate_reply(messages, sender=sender, **kwargs)
        except Exception as e:
            print(f"调用失败: {e}，正在重试...")
            raise

# 使用
robust_agent = RobustAssistant(
    name="robust_assistant",
    llm_config=llm_config,
)

10.8 日志与监控

import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('autogen.log'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger("autogen")

# 在 Agent 中使用日志
class LoggingAgent(autogen.ConversableAgent):
    def receive(self, message, sender, request_reply=None, silent=False):
        logger.info(f"收到来自 {sender.name} 的消息: {message.get('content', '')[:100]}")
        reply = super().receive(message, sender, request_reply, silent)
        logger.info(f"回复给 {sender.name}: {reply.get('content', '')[:100]}")
        return reply

---

11. 企业级最佳实践

11.1 项目结构规范

project/
├── .env                          # 环境变量
├── OAI_CONFIG_LIST.json          # LLM 配置
├── autogen_config.py             # AutoGen 配置
├── agents/                       # 自定义 Agent
│   ├── __init__.py
│   ├── my_assistant.py
│   └── my_tools.py
├── teams/                        # 团队编排
│   ├── __init__.py
│   ├── research_team.py
│   └── coding_team.py
├── tools/                        # 工具函数
│   ├── __init__.py
│   ├── web_tools.py
│   ├── db_tools.py
│   └── api_tools.py
├── workflows/                    # 工作流
│   ├── __init__.py
│   └── daily_report.py
├── workspace/                    # 代码执行目录
├── documents/                    # 文档库（RAG）
├── logs/                         # 日志
├── tests/                        # 测试
├── requirements.txt
├── Dockerfile
└── main.py                       # 入口

11.2 配置管理

# autogen_config.py
import os
from dotenv import load_dotenv

load_dotenv()

def get_llm_config(model="gpt-4o"):
    """获取统一的 LLM 配置"""
    return {
        "config_list": [
            {
                "model": model,
                "api_key": os.getenv("OPENAI_API_KEY"),
                "base_url": os.getenv("OPENAI_BASE_URL"),
            }
        ],
        "temperature": 0.7,
        "max_tokens": 4096,
        "timeout": 120,
    }

def get_code_exec_config(work_dir="workspace"):
    """获取代码执行配置"""
    return {
        "work_dir": work_dir,
        "use_docker": os.getenv("USE_DOCKER", "False").lower() == "true",
        "timeout": 60,
    }

11.3 安全最佳实践

# 1. API 密钥管理 - 绝不硬编码
# ✅ 正确：使用环境变量
import os
api_key = os.getenv("OPENAI_API_KEY")

# ❌ 错误：硬编码
# api_key = "sk-xxxxxxxxxx"

# 2. 代码执行安全
code_exec_config = {
    "work_dir": "workspace",
    "use_docker": True,  # 生产环境务必使用 Docker
    "timeout": 30,
    "last_n_messages": 2,  # 限制检查的消息数
}

# 3. 输入验证
def sanitize_user_input(user_input: str) -> str:
    """清理用户输入"""
    # 移除危险字符
    import re
    dangerous = ["rm -rf", "drop table", "sudo"]
    for d in dangerous:
        user_input = user_input.replace(d, "[已过滤]")
    return user_input

# 4. 速率限制
from functools import wraps
import time

def rate_limit(max_calls: int, period: int):
    def decorator(func):
        calls = []
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            calls[:] = [c for c in calls if now - c < period]
            if len(calls) >= max_calls:
                raise Exception(f"速率限制：{period}秒内最多{max_calls}次调用")
            calls.append(now)
            return func(*args, **kwargs)
        return wrapper
    return decorator

11.4 单元测试

# tests/test_agents.py
import pytest
from unittest.mock import Mock, patch

class TestAssistantAgent:
    def test_generate_reply(self):
        """测试 Agent 能生成回复"""
        agent = autogen.AssistantAgent(
            name="test_agent",
            llm_config={
                "config_list": [{"model": "test", "api_key": "test"}],
            },
        )
        # 使用 mock 避免真实 API 调用
        with patch.object(agent, "_generate_reply") as mock_reply:
            mock_reply.return_value = {"content": "测试回复"}
            reply = agent.generate_reply(
                messages=[{"role": "user", "content": "你好"}]
            )
            assert reply["content"] == "测试回复"

    def test_tool_registration(self):
        """测试工具注册"""
        ...

# tests/test_teams.py
class TestTeam:
    async def test_team_execution(self):
        """测试团队运行"""
        ...

11.5 性能优化

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

@lru_cache(maxsize=100)
def cached_llm_call(prompt: str) -> str:
    """缓存相同输入的 LLM 响应"""
    ...

# 2. 并行执行独立任务
import asyncio

async def run_parallel_tasks(tasks: list):
    """并行运行多个独立任务"""
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# 3. 使用更快的模型进行简单任务
cheap_llm_config = {
    "config_list": [{"model": "gpt-4o-mini", ...}],
}
expensive_llm_config = {
    "config_list": [{"model": "gpt-4o", ...}],
}

# 4. 批量处理
class BatchProcessor:
    def __init__(self, batch_size=10):
        self.batch_size = batch_size
        self.queue = []
    
    async def add_and_process(self, item):
        self.queue.append(item)
        if len(self.queue) >= self.batch_size:
            return await self._flush()
        return None
    
    async def _flush(self):
        batch = self.queue[:self.batch_size]
        self.queue = self.queue[self.batch_size:]
        results = await asyncio.gather(*[
            self._process_item(item) for item in batch
        ])
        return results

11.6 部署方案

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制应用
COPY . .

# 设置环境变量
ENV PYTHONUNBUFFERED=1
ENV USE_DOCKER=False

# 运行
CMD ["python", "main.py"]

# docker-compose.yml
version: '3.8'

services:
  autogen-app:
    build: .
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - OPENAI_BASE_URL=${OPENAI_BASE_URL}
    volumes:
      - ./workspace:/app/workspace
      - ./logs:/app/logs
    restart: unless-stopped

  chroma-db:  # 向量数据库（用于 RAG）
    image: chromadb/chroma:latest
    ports:
      - "8000:8000"
    volumes:
      - chroma-data:/chroma/data

volumes:
  chroma-data:

# main.py (生产入口)
import asyncio
import logging
from autogen import AssistantAgent, UserProxyAgent

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def main():
    try:
        # 初始化 Agent
        assistant = AssistantAgent(
            name="assistant",
            llm_config=get_llm_config(),
        )
        
        user_proxy = UserProxyAgent(
            name="user_proxy",
            code_execution_config=get_code_exec_config(),
        )
        
        # 执行业务逻辑
        await user_proxy.a_initiate_chat(
            assistant,
            message="开始处理今日任务...",
        )
        
    except Exception as e:
        logger.error(f"任务执行失败: {e}")
        raise

if __name__ == "__main__":
    asyncio.run(main())

---

12. 常见问题与调试

12.1 常见错误及解决方案

错误	原因	解决方案
APIError	API 密钥无效或过期	检查 api_key 和环境变量
RateLimitError	请求频率过高	添加 timeout 和 max_retries 参数
TimeoutError	请求超时	增加 timeout，使用更快的模型
ContextLengthExceeded	上下文太长	使用 max_consecutive_auto_reply 限制轮数
ImportError	缺少依赖	安装 pyautogen[all]
Docker 错误	Docker 未运行	设置 use_docker=False 或启动 Docker

12.2 调试技巧

# 1. 开启详细日志
import logging
logging.basicConfig(level=logging.DEBUG)

# 2. 查看对话历史
messages = user_proxy.chat_messages[assistant]
for msg in messages:
    print(f"[{msg['role']}] {msg.get('content', '')[:200]}")

# 3. 检查 Agent 内部状态
print(f"助手名称: {assistant.name}")
print(f"LLM 配置: {assistant.llm_config}")
print(f"系统提示: {assistant.system_message}")
print(f"已注册回复函数: {assistant._reply_func_list}")

# 4. 使用简单的测试配置
test_llm_config = {
    "config_list": [
        {
            "model": "gpt-4o-mini",  # 更快更便宜
            "api_key": "sk-xxx",
        }
    ],
    "temperature": 0,  # 确定性输出，便于调试
}

# 5. 手动构造消息测试
def test_agent_response(agent, message):
    """手动测试 Agent 响应"""
    reply = agent.generate_reply(
        messages=[{"role": "user", "content": message}],
        sender=agent,
    )
    return reply

12.3 对话可视化

# 将对话导出为可读格式
def export_conversation(chat_messages, filename="conversation.md"):
    """导出对话为 Markdown 文件"""
    with open(filename, "w", encoding="utf-8") as f:
        f.write("# 对话记录\n\n")
        for msg in chat_messages:
            role = msg.get("role", "unknown")
            content = msg.get("content", "")
            f.write(f"## {role}\n\n{content}\n\n---\n\n")
    print(f"对话已导出到 {filename}")

# 使用
export_conversation(user_proxy.chat_messages[assistant])

---

13. 生态对比与选型建议

13.1 主流多 Agent 框架对比

特性	AutoGen	LangChain/LangGraph	CrewAI	MetaGPT	Semantic Kernel
开发方	微软	LangChain公司	-	香港中文大学	微软
核心思想	对话驱动	链/图驱动	角色驱动	SOP驱动	编排驱动
学习曲线	中	中-高	低	中	中
灵活度	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐
代码执行	✅ 内置Docker	❌ 需集成	❌ 需集成	✅ 内置	❌ 需集成
Human-in-Loop	✅ 原生支持	✅ 支持	⚠️ 有限	❌	⚠️ 有限
工具系统	✅ 完善	✅ 完善	✅ 支持	⚠️ 有限	✅ 完善
RAG支持	✅ 原生	✅ 完善	❌	❌	✅ 完善
企业级特性	✅ 强	✅ 强	⚠️ 一般	⚠️ 一般	✅ 强
社区规模	45k+⭐	100k+⭐	22k+⭐	30k+⭐	20k+⭐
微软生态	✅ Azure	❌	❌	❌	✅ Azure/C#

13.2 选型建议

┌─────────────────────────────────────────────────┐
│                选型决策树                         │
├─────────────────────────────────────────────────┤
│                                                   │
│ 需要多 Agent 协作？                               │
├── 是 ── 需要代码执行？                           │
│        ├── 是 ── AutoGen (最适合)                 │
│        └── 否 ── 需要复杂工作流？                  │
│                  ├── 是 ── LangGraph              │
│                  └── 否 ── CrewAI (最简单)        │
├── 否 ── 单个 Agent 就够了                         │
│        ├── 简单任务 ── 直接调 LLM API             │
│        ├── RAG 应用 ── LangChain + Chroma         │
│        └── 结构化输出 ── Instructor / Outlines    │
└─────────────────────────────────────────────────┘

推荐场景：

场景	推荐框架	原因
企业级多Agent系统	AutoGen	微软生态、代码执行、Human-in-Loop
复杂链式工作流	LangGraph	图结构编排、条件路由
快速原型开发	CrewAI	最简API、快速上手
软件开发团队模拟	MetaGPT	SOP驱动、角色专业
.NET/C# 生态	Semantic Kernel	语言原生支持

---

14. 学习路线图

📌 阶段一：入门（1-2周）

目标：理解 AutoGen 核心概念，能写基本的多 Agent 对话

• 完成本教程第1-4章
• 安装环境并跑通第一个示例
• 理解 ConversableAgent, AssistantAgent, UserProxyAgent
• 实践：写一个「问答助手」和「代码生成助手」
• 阅读官方文档: microsoft.github.io/autogen/

练习项目：

创建一个对话系统，包含：
1. 一个编程助手（AssistantAgent）
2. 一个用户代理（UserProxyAgent）
3. 让助手写一个 Web 爬虫并执行

📌 阶段二：进阶（2-4周）

目标：掌握群聊、工具调用、流程控制

• 第5-7章深入学习
• 实践 GroupChat 多 Agent 协作
• 学习注册和使用工具函数
• 理解发言者选择策略
• 实践：创建一个「产品开发团队」

练习项目：

创建一个软件开发的三人团队：
1. 产品经理（定义需求）
2. 开发者（编写代码）
3. 测试员（代码审查和测试）
通过 GroupChat 协作完成一个小功能开发

📌 阶段三：高级（1-2月）

目标：掌握企业级应用、性能优化、部署

• 第8-11章深入学习
• 学习 Team 高级编排模式
• 掌握 RAG + AutoGen 集成
• 学习 Docker 部署和安全加固
• 阅读 AutoGen 源码，理解内部机制

练习项目：

构建一个企业级智能客服系统：
1. 使用 RAG 检索知识库
2. 多 Agent 分工（问题分类、检索、回答、审核）
3. 人类介入模式（复杂问题转人工）
4. 日志监控和告警
5. Docker 部署

📌 阶段四：精通（持续）

目标：能设计和实现复杂的多 Agent 系统架构

• 深入研究 AutoGen 源码贡献
• 实践：设计行业特定解决方案
• 探索与其他框架（LangChain, Vector DB）的集成
• 关注 AutoGen 新版本特性和 RFC
• 参与社区讨论，分享最佳实践

进阶项目方向：

1. 自动化研究助手：联网搜索 -> 阅读论文 -> 生成摘要 -> 撰写报告
2. 自动化数据分析平台：连接数据库 -> 数据清洗 -> 分析建模 -> 可视化
3. 多语言翻译与本地化系统：翻译 -> 审校 -> 文化适配 -> 格式编排
4. Agent 训练与评估框架：自动生成测试用例 -> 运行测试 -> 评估性能 -> 优化提示

📚 推荐学习资源

类型	资源	链接
📖 官方文档	AutoGen 文档	microsoft.github.io/autogen/
📖 官方论文	AutoGen: Enabling Next-Gen LLM Applications via Multi-Agent Conversation	arxiv.org/abs/2308.08…
📺 教程	Microsoft AutoGen Playlist	YouTube
📺 实战	官方 Notebook 示例	github.com/microsoft/a…
📘 书籍	《多Agent系统实战》	各大书店
💬 社区	Discord	discord.gg/pAbn5JrE
💬 社区	GitHub Discussions	github.com/microsoft/a…

---

附录

A. 完整配置模板

{
    "llm_config": {
        "config_list": [
            {
                "model": "gpt-4o",
                "api_key": "${OPENAI_API_KEY}",
                "base_url": "${OPENAI_BASE_URL}",
                "tags": ["gpt4", "production"]
            },
            {
                "model": "gpt-4o-mini",
                "api_key": "${OPENAI_API_KEY}",
                "tags": ["fast", "cheap"]
            }
        ],
        "temperature": 0.7,
        "max_tokens": 4096,
        "timeout": 120,
        "max_retries": 3
    },
    "code_execution_config": {
        "work_dir": "workspace",
        "use_docker": true,
        "timeout": 60,
        "last_n_messages": 3
    }
}

B. 快速启动模板

#!/usr/bin/env python3
"""AutoGen 快速启动模板"""

import os
import autogen
from dotenv import load_dotenv

load_dotenv()

def main():
    # 配置
    llm_config = {
        "config_list": [
            {
                "model": "gpt-4o",
                "api_key": os.getenv("OPENAI_API_KEY"),
            }
        ],
        "temperature": 0.7,
    }
    
    # 创建 Agent
    assistant = autogen.AssistantAgent(
        name="assistant",
        llm_config=llm_config,
    )
    
    user_proxy = autogen.UserProxyAgent(
        name="user_proxy",
        human_input_mode="TERMINATE",
        max_consecutive_auto_reply=10,
        code_execution_config={
            "work_dir": "workspace",
            "use_docker": False,
        },
    )
    
    # 开始对话
    user_proxy.initiate_chat(
        assistant,
        message="输入你的问题或任务...",
    )

if __name__ == "__main__":
    main()

C. 术语表

术语	英文	解释
智能体	Agent	一个由 LLM 驱动的 AI 实体
对话	Conversation	Agent 之间的消息交换过程
消息	Message	Agent 之间传递的信息单元
工具	Tool	Agent 可以调用的外部函数
团队	Team	一组 Agent 的编排单元
群聊	GroupChat	多 Agent 的群组对话模式
发言者选择	Speaker Selection	决定群聊中下一个发言者的策略
终止条件	Termination	停止对话的条件
代码执行	Code Execution	Agent 执行生成的代码
人在回路	Human-in-the-Loop	人类参与决策过程
检索增强生成	RAG	结合检索和生成的问答技术
模型客户端	Model Client	与 LLM API 交互的客户端
系统提示词	System Message	Agent 的系统级指令
工具注册	Tool Registration	将函数注册为 Agent 可用工具

---

💡 最后建议

1. 从简单开始：先跑通 2-Agent 对话，再逐步增加复杂度
2. 善用调试：使用 logging.DEBUG、查看 chat_messages、导出对话
3. 关注成本：用 gpt-4o-mini 做简单任务，gpt-4o 做复杂推理
4. 安全第一：生产环境务必使用 Docker 沙箱
5. 持续学习：AutoGen 迭代很快，关注官方 GitHub Release Notes

Happy Building！🤖🚀"