AutoGen 学习笔记教程

1. AutoGen 是什么

AutoGen 是微软系的一个 Agent 开发框架，用来构建基于大模型的单 Agent 或多 Agent 应用。它最早出名的点，不是“写一个聊天机器人”，而是“让多个 Agent 以消息对话的方式协同完成复杂任务”。

一句话理解：

AutoGen 是一个偏多智能体协作、消息驱动、可接工具和代码执行能力的 Python Agent 框架。

如果以前写的是普通 LLM 应用，那么 AutoGen 关注的是：

• 如何定义一个 Agent
• 如何让 Agent 调工具
• 如何让多个 Agent 对话协作
• 如何在流程中插入人类确认
• 如何把代码执行、浏览器、MCP 等能力接进来

3. AutoGen 适合解决什么问题

AutoGen 更适合以下任务：

• 多角色协作任务
• 需要“规划 -> 执行 -> 审核”链路的任务
• 需要工具调用的任务
• 需要代码执行、网页操作、文件处理的任务
• 需要人类介入确认的复杂流程

典型例子：

• 一个 Agent 负责规划
• 一个 Agent 负责写代码
• 一个 Agent 负责 Review
• 一个 Agent 负责执行工具
• 最后由人类决定是否继续

如果只是一个很简单的问答助手，AutoGen 往往偏重。

4. AutoGen 的核心思想

4.1 Agent 不是“有灵魂的对象”，而是 Prompt + Model + Tool + Message Loop

在工程上，一个 AutoGen Agent 的行为主要由这几部分决定：

• 模型客户端：决定调用哪个模型
• system_message 或 instructions：决定角色与行为边界
• tools：决定它能调用哪些外部能力
• 对话上下文：决定它当前知道什么
• 运行循环：决定它是否继续思考、继续调工具或停止

所以 AutoGen 的 Agent 并不是“神秘智能体”，而是一个被约束好的、能持续收发消息并调用工具的执行单元。

4.2 AutoGen 最有代表性的能力是多 Agent 对话

这是 AutoGen 和很多“单 Agent + Prompt”框架最不同的地方。

你可以让多个 Agent：

• 轮流发言
• 指定谁先说
• 用 Team / Group Chat 方式协作
• 在消息中共享上下文
• 在中途接入人类反馈

这让它非常适合做“角色协作型 Agent 系统”。

4.3 复杂任务不要全压给一个 Prompt

AutoGen 很适合解决“复杂任务”，但不是靠一个超长 Prompt 解决，而是靠：

• 多个 Agent 拆职责
• 多轮消息传递
• 工具调用
• 明确的协作模式

这是它比“单提示词大包大揽”更有工程感的地方。

5. AutoGen 的分层结构

根据官方文档，AutoGen 现在的结构可以理解为三层。

5.1 AgentChat

最常用、最容易上手的一层。

适合：

• 快速写一个 Agent
• 做多 Agent 对话原型
• 做 Team / Group Chat 演示

如果你是第一次学 AutoGen，通常从这一层开始。

5.2 Core

更底层，偏事件驱动、消息驱动。

适合：

• 做更复杂的事件流
• 做分布式 Agent 运行时
• 做强控制的多 Agent 系统

如果你把 Agent 当“框架编程模型”而不是“Prompt 封装”，这一层会更有价值。

5.3 Extensions

这一层负责接具体能力和外部实现，例如：

• OpenAI 模型客户端
• Azure OpenAI 模型客户端
• MCP 能力接入
• 代码执行器
• Docker 执行器

你可以把它理解成 AutoGen 的“适配器和外设层”。

6. 安装与环境准备

AutoGen 官方当前常见安装方式如下：

pip install -U "autogen-agentchat" "autogen-ext[openai]"

基础要求：

• Python 3.10+
• 可访问的模型服务
• 常见情况下需要 OPENAI_API_KEY

例如在 Windows PowerShell 中设置环境变量：

$env:OPENAI_API_KEY="你的 key"

如果你用 .env，也可以配合 python-dotenv 使用。

7. 第一个 Hello World 示例

这是一个最小可运行示例，帮助你理解 AutoGen 最基础的模型：一个 Agent 接收任务并返回结果。

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4.1")

    agent = AssistantAgent(
        name="assistant",
        model_client=model_client,
        system_message="你是一个简洁、可靠的中文助手。"
    )

    result = await agent.run(task="请用三句话解释什么是 AutoGen")
    print(result)

    await model_client.close()


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

这个例子里发生了什么：

1. 创建模型客户端 OpenAIChatCompletionClient
2. 创建一个 AssistantAgent
3. 给它传入 task
4. Agent 调模型生成回答
5. 输出结果并关闭模型客户端

这个层面上，你可以把它理解成“比直接调 OpenAI SDK 更像 Agent 的封装”。

8. 一个更像 Agent 的例子：带工具调用

真正体现 Agent 特性的，不是单次回答，而是“模型可以决定是否调用工具”。

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient


def get_weather(city: str) -> str:
    """返回天气信息。真实项目中这里可以接天气 API。"""
    mock = {
        "北京": "晴，26 度",
        "上海": "多云，24 度",
        "深圳": "小雨，28 度",
    }
    return mock.get(city, f"未找到 {city} 的天气信息")


async def main() -> None:
    model_client = OpenAIChatCompletionClient(model="gpt-4.1")

    agent = AssistantAgent(
        name="weather_assistant",
        model_client=model_client,
        system_message="你是天气助手。需要天气时优先调用工具，不要编造。",
        tools=[get_weather],
        max_tool_iterations=5,
    )

    result = await agent.run(task="帮我查一下北京天气，并顺便给出穿衣建议")
    print(result)

    await model_client.close()


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

这个例子里最关键的是：

• tools=[get_weather] 给了 Agent 一个可调用能力
• 模型不只是“回答”，还会判断是否需要用工具
• max_tool_iterations 控制工具调用与思考轮次，避免无限循环

这已经比普通 Prompt 工程更接近“可执行系统”。

9. 多 Agent 协作是 AutoGen 的招牌能力

AutoGen 的核心魅力在这里。

你可以把一个复杂问题拆给多个 Agent，例如：

• planner：负责拆任务
• researcher：负责查资料
• writer：负责生成文章
• reviewer：负责审稿

这样做的好处是：

• Prompt 更短、更清晰
• 每个 Agent 职责更明确
• 更容易调试哪一步出了问题
• 更适合复杂任务分治

下面是完整的多 Agent 协作示例，展示启动、链路安排和通信机制。

9.1 方式一：使用 RoundRobinGroupChat（轮询模式）

"""  
    RoundRobinGroupChat 示例  
    演示如何使用 RoundRobinGroupChat 创建一个协作流程，多个 Agent 按顺序执行。  
"""  
  
import asyncio  
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  
  
  
async def main() -> None:  
    model_client = OpenAIChatCompletionClient(  
        model="deepseek-v4-flash",  
        base_url="https://api.deepseek.com",  
        api_key="xxxxxxxxxxxx",  
        model_info={  
            "vision": False,  # DeepSeek V4 Flash 不支持视觉  
            "function_calling": True,  # 支持函数调用  
            "json_output": True,  # 支持 JSON 输出  
            "family": "deepseek",  # 模型家族  
            "structured_output": True  # 支持结构化输出  
        }  
    )  
  
    # 1. 创建多个 Agent    planner = AssistantAgent(  
        name="planner",  
        model_client=model_client,  
        system_message="你负责拆解任务，给出可执行步骤。完成后说NEXT_AGENT")  
  
    writer = AssistantAgent(  
        name="writer",  
        model_client=model_client,  
        system_message="你负责根据规划撰写内容。完成后说NEXT_AGENT")  
  
    reviewer = AssistantAgent(  
        name="reviewer",  
        model_client=model_client,  
        system_message="你负责检查内容是否清晰、完整、无明显错误。完成后说TERMINATE")  
  
    # 2. 创建 Team，定义协作链路  
    # RoundRobinGroupChat: planner -> writer -> reviewer 按顺序执行  
    team = RoundRobinGroupChat(  
        participants=[planner, writer, reviewer],  
        termination_condition=TextMentionTermination("TERMINATE"),  # 终止条件  
        max_turns=10  # 最大轮次，防止无限循环  
    )  
  
    # 3. 启动协作流程  
    result = await team.run(task="写一篇关于AIAgent的500字科普文章")  
  
    # 4. 查看协作历史  
    print("\n == = 协作历史 == =")  
    for message in result.messages:  
        print(f"\n[{message.source}]: {message.content}")  
  
        await model_client.close()  
  
  
if __name__ == "__main__":  
    asyncio.run(main())

关键点：

• 启动：team.run(task=...) 启动协作流程
• 链路：RoundRobinGroupChat 按 participants 顺序轮询执行
• 通信：Agent 通过 result.messages 共享消息历史，每个 Agent 都能看到前面的对话
• 终止：TextMentionTermination(“TERMINATE”) 检测到关键词后停止

9.2 方式二：使用 SelectorGroupChat（选择器模式）

import asyncio  
from autogen_agentchat.agents import AssistantAgent  
from autogen_agentchat.teams import SelectorGroupChat  
from autogen_agentchat.conditions import MaxMessageTermination  
from autogen_ext.models.openai import OpenAIChatCompletionClient  
  
  
async def main() -> None:  
    model_client = OpenAIChatCompletionClient(  
        model="deepseek-v4-flash",  
        base_url="https://api.deepseek.com",  
        api_key="xxxxxx",  
        model_info={  
            "vision": False,  # DeepSeek V4 Flash 不支持视觉  
            "function_calling": True,  # 支持函数调用  
            "json_output": True,  # 支持 JSON 输出  
            "family": "deepseek",  # 模型家族  
            "structured_output": True  # 支持结构化输出  
        }  
    )  
  
    # 1. 创建专家 Agent    researcher = AssistantAgent(  
        name="researcher",  
        model_client=model_client,  
        system_message="你负责查找资料和事实核查。"  
    )  
  
    writer = AssistantAgent(  
        name="writer",  
        model_client=model_client,  
        system_message="你负责撰写内容。"  
    )  
  
    reviewer = AssistantAgent(  
        name="reviewer",  
        model_client=model_client,  
        system_message="你负责审核内容质量。"  
    )  
  
    # 2. 创建选择器 Agent（决定下一个发言者）  
    selector = AssistantAgent(  
        name="selector",  
        model_client=model_client,  
        system_message=  
        """  
        你是协调者，根据当前任务状态决定下一个发言者：  
        - 需要查资料时选择 researcher        - 需要写作时选择 writer        - 需要审核时选择 reviewer        - 任务完成时选择 TERMINATE  
        你必须从以下选项中选择一个：researcher, writer, reviewer, TERMINATE  
        只返回选择的名称，不要添加其他内容。  
        """    )  
  
    # 3. 创建 Team，由 selector 动态决定链路  
    team = SelectorGroupChat(  
        participants=[researcher, writer, reviewer],  
        model_client=model_client,  
        selector_func=selector,  # 关联选择器 Agent        termination_condition=MaxMessageTermination(20)  
    )  
  
    # 4. 启动协作  
    result = await team.run(task="写一篇关于量子计算的文章，需要准确的技术细节")  
  
    print("\n=== 协作链路 ===")  
    for i, msg in enumerate(result.messages):  
        print(f"{i + 1}. [{msg.source}] -> {msg.content[:100]}...")  
  
        await model_client.close()  
  
  
if __name__ == "__main__":  
    asyncio.run(main())

关键点：

• 启动：同样使用 team.run(task=...)
• 链路：SelectorGroupChat 由 selector Agent 动态决定下一个发言者
• 通信：所有 Agent 共享完整对话历史，selector 根据历史做决策
• 优势：更灵活，适合复杂任务的动态协作

9.3 方式三：使用 Swarm（群体智能模式）

import asyncio  
import logging  
import os  
  
from autogen_core import EVENT_LOGGER_NAME, TRACE_LOGGER_NAME  
from autogen_agentchat.agents import AssistantAgent  
from autogen_agentchat.conditions import MaxMessageTermination  
from autogen_agentchat.teams import Swarm  
from autogen_agentchat.ui import Console  
  
from deepseek_reasoning_client import DeepSeekReasoningChatCompletionClient  
  
  
def setup_logging() -> None:  
    """Enable AutoGen runtime logs when AUTOGEN_LOG_LEVEL is configured."""  
    log_level_name = os.getenv("AUTOGEN_LOG_LEVEL", "").upper()  
    if not log_level_name:  
        return  
  
    log_level = getattr(logging, log_level_name, logging.INFO)  
    logging.basicConfig(  
        level=log_level,  
        format="%(asctime)s %(levelname)s [%(name)s] %(message)s",  
        force=True,  
    )  
    logging.getLogger(EVENT_LOGGER_NAME).setLevel(log_level)  
    logging.getLogger(TRACE_LOGGER_NAME).setLevel(log_level)  
  
  
async def main() -> None:  
    # DeepSeek thinking mode requires reasoning_content to be sent back.  
    setup_logging()  
  
    model_client = DeepSeekReasoningChatCompletionClient(  
        model="deepseek-v4-flash",  
        base_url="https://api.deepseek.com",  
        api_key="xxxxxx",  
        model_info={  
            "vision": False,  
            "function_calling": True,  
            "json_output": True,  
            "family": "deepseek",  
            "structured_output": True,  
        },  
    )  
  
    planner = AssistantAgent(  
        name="planner",  
        model_client=model_client,  
        handoffs=["writer"],  
        system_message=(  
            "You are the planner. "  
            "Your only job is to create a brief outline for the article. "            "Do not write the full article yourself. "            "After you finish the outline, you must hand off to writer immediately."        ),  
    )  
  
    writer = AssistantAgent(  
        name="writer",  
        model_client=model_client,  
        handoffs=["reviewer"],  
        system_message=(  
            "You are the writer. "  
            "Write the full article in English based on the plan. "            "Keep the article under 300 words. "            "Do not apologize, do not explain your process, and do not repeat the article. "            "After the draft is complete, you must hand off to reviewer immediately."        ),  
    )  
  
    reviewer = AssistantAgent(  
        name="reviewer",  
        model_client=model_client,  
        system_message=(  
            "You are the reviewer. "  
            "Review the draft and output exactly one clean final version of the article. "            "Keep it under 300 words. "            "Do not include notes such as word counts, glitch apologies, revision offers, or duplicate copies. "            "Your response must be the final article only."        ),  
    )  
  
    team = Swarm(  
        participants=[planner, writer, reviewer],  
        termination_condition=MaxMessageTermination(6),  
**关键点：**
- 使用 DeepSeek 的 thinking 模式模型时，如果模型返回了推理内容，后续多轮请求必须把上一轮 assistant 的 `reasoning_content` 原样带回；否则就会报错：`The reasoning_content in the thinking mode must be passed back to the API.`。这类问题的本质不是模型不能用，而是客户端在消息转换时只保留了最终回答，没有把推理字段继续传回。
- 在 AutoGen 的 `Swarm` 模式里，链路能否切换到下一个 Agent，关键不在于“提示词里写了交给谁”，而在于当前 Agent 是否真的发出了 handoff。`handoffs=["writer"]` 表示当前 Agent 允许把控制权移交给 `name="writer"` 的 Agent；如果没有产生真正的 handoff message，当前 speaker 就会继续自己说，从而出现重复输出、反复道歉、一直不切换到下一个 Agent 的现象。
- 因此，排查 `Swarm` 重复输出时，要先分清两层：第一层是 handoff 配置是否正确，是否真的对应下一个 Agent 的 `name`；第二层才是提示词是否足够明确。实践上，固定流水线任务通常要把提示词收紧为“当前 Agent 只做本阶段工作，完成后必须立即 handoff”，必要时再配合消息数终止条件，避免异常情况下无休止自说自话。
  
  
if __name__ == "__main__":  
    asyncio.run(main())

from __future__ import annotations  
  
from typing import Any, Literal, Mapping, Sequence  
  
from autogen_core.models import AssistantMessage, FunctionExecutionResultMessage, LLMMessage  
from autogen_core.tools import Tool, ToolSchema  
from autogen_ext.models.openai import OpenAIChatCompletionClient  
  
  
class DeepSeekReasoningChatCompletionClient(OpenAIChatCompletionClient):  
    """补齐 DeepSeek thinking 模式要求的 reasoning_content 回传。"""  
  
    def _process_create_args(  
        self,  
        messages: Sequence[LLMMessage],  
        tools: Sequence[Tool | ToolSchema],  
        tool_choice: Tool | Literal["auto", "required", "none"],  
        json_output: bool | type[Any] | None,  
        extra_create_args: Mapping[str, Any],  
    ) -> Any:  
        create_params = super()._process_create_args(  
            messages,  
            tools,  
            tool_choice,  
            json_output,  
            extra_create_args,  
        )  
        self._patch_reasoning_messages(messages, create_params.messages)  
        return create_params  
  
    @staticmethod  
    def _patch_reasoning_messages(messages: Sequence[LLMMessage], oai_messages: list[dict[str, Any]]) -> None:  
        """把 AssistantMessage.thought 映射为 DeepSeek 需要的 reasoning_content。"""  
  
        message_index = 0  
        for message in messages:  
            if isinstance(message, FunctionExecutionResultMessage):  
                message_index += len(message.content)  
                continue  
  
            if isinstance(message, AssistantMessage) and message.thought:  
                oai_message = oai_messages[message_index]  
                oai_message["reasoning_content"] = message.thought  
  
                # AutoGen 默认会把 thought 塞进 content；DeepSeek thinking 模式要求独立字段。  
                if "tool_calls" in oai_message and oai_message.get("content") == message.thought:  
                    oai_message["content"] = None  
  
            message_index += 1

9.4 Agent 间通信机制详解

# Agent 间通信的核心数据结构
class ChatMessage:
    source: str        # 发送者 Agent 名称
    content: str       # 消息内容
    tool_calls: list   # 工具调用记录
    metadata: dict     # 元数据（时间戳、上下文等）

# 通信流程：
# 1. Agent A 生成消息 -> 添加到 team.messages
# 2. Team 决定下一个 Agent B
# 3. Agent B 接收完整 messages 历史作为上下文
# 4. Agent B 生成新消息 -> 添加到 team.messages
# 5. 循环直到满足终止条件

9.5 完整协作流程图

11. AutoGen 和 CrewAI、LangGraph 的区别

11.1 AutoGen vs CrewAI

CrewAI 更强调：

• 角色分工
• Crew / Task 编排
• 用较直观的方式描述多个 Agent 协作

AutoGen 更强调：

• 多 Agent 消息交互
• 对话式协作
• 事件驱动的 Agent 编程模型

如果你从体验上感受：

• CrewAI 更像“多角色任务编排”
• AutoGen 更像“多智能体消息协作框架”

11.2 AutoGen vs LangGraph

LangGraph 更强调：

• 状态机
• 显式工作流图
• 节点和边的可控编排
• 长流程、可恢复、可插人工节点

AutoGen 更强调：

• Agent 对话
• 消息驱动
• Team / GroupChat 型协作

如果你很在意“流程严格受控”，LangGraph 往往更稳。
如果你想研究“多 Agent 如何交互协作”，AutoGen 非常经典。

12. AutoGen 的优势

12.1 多 Agent 表达能力强

它非常擅长表达：

• 专家协作
• 审核与反驳
• 计划与执行分离
• 人机共治

12.2 研究味浓，适合做原型和实验

很多多 Agent 模式的讨论，早期都受 AutoGen 影响很大。

12.3 与工具、代码执行、MCP 结合方便

这让它不止是“聊天”，而是更接近“可执行 Agent 系统”。

13. AutoGen 的局限

13.1 容易失控

如果你把任务描述得太模糊，多 Agent 之间可能：

• 重复讨论
• 无限来回
• 偏题
• 工具调用失控

13.2 调试比普通代码难

你调试的不只是 Python 代码，还有：

• Prompt
• 消息历史
• 工具调用过程
• 模型决策路径

13.3 复杂任务不能只靠 Prompt

复杂任务必须结合：

• 明确拆分职责
• 限制工具权限
• 控制轮次
• 固定输出格式
• 必要时加入人工确认

13.4 官方主线已经迁移

这不是技术缺陷，但对选型非常重要：

• AutoGen 现在不是微软新项目主推路线
• 新项目要考虑 Microsoft Agent Framework