AutoGen 完整教程:从入门到进阶

AutoGen 是微软开源的多智能体对话框架,核心理念是用对话驱动协作——让多个 AI Agent 通过自然语言对话来协同完成复杂任务。

本教程基于 AutoGen 0.7.x(AgentChat API),从零基础到高级用法,循序渐进。


目录


第一篇:入门基础

1.1 AutoGen 是什么

一句话定义

AutoGen = 一个让多个 AI Agent 通过对话协同工作的 Python 框架

核心思想

想象一个软件开发团队:

  • 产品经理 分析需求,制定方案
  • 工程师 根据方案写代码
  • 代码审查员 检查代码质量
  • 测试员 验证功能是否正常

他们通过开会讨论来协作。AutoGen 就是把这种协作模式搬到了 AI 世界——每个 Agent 扮演一个角色,通过"对话"来协同完成任务。

架构总览

┌─────────────────────────────────────────────────┐
│                你的应用程序                       │
├─────────────────────────────────────────────────┤
│           autogen-agentchat (高层 API)           │
│  AssistantAgent / UserProxyAgent / GroupChat     │
├─────────────────────────────────────────────────┤
│            autogen-core (底层引擎)                │
│  消息传递 / Agent 运行时 / 事件驱动               │
├─────────────────────────────────────────────────┤
│            autogen-ext (扩展)                    │
│  OpenAI 客户端 / Azure / 代码执行 / MCP          │
└─────────────────────────────────────────────────┘
层级 包名 作用
高层 API autogen-agentchat 快速构建多 Agent 应用(本教程重点)
底层引擎 autogen-core 消息传递、Agent 运行时、分布式支持
扩展 autogen-ext LLM 客户端、代码执行器等

AutoGen vs 其他框架

特性 AutoGen LangChain CrewAI
核心理念 对话驱动协作 链式调用 角色扮演
多 Agent ✅ 原生支持 需要 LangGraph ✅ 原生支持
群聊模式 3 种模式 有限 单一模式
代码执行 ✅ 内置沙箱 需额外配置 需额外配置
异步支持 ✅ 原生 async 部分支持 有限
学习曲线 中等 较高 较低

1.2 安装与环境配置

基础安装

# 安装 AgentChat + OpenAI 扩展
pip install -U "autogen-agentchat" "autogen-ext[openai]"

# 安装 dotenv(管理环境变量)
pip install python-dotenv

环境变量配置

在项目根目录创建 .env 文件:

# .env
LLM_API_KEY=your-api-key-here
LLM_BASE_URL=https://api.openai.com/v1
LLM_MODEL_ID=gpt-4o-mini

使用国产模型?BASE_URL 改为对应平台的地址即可:

  • 通义千问:https://dashscope.aliyuncs.com/compatible-mode/v1
  • DeepSeek:https://api.deepseek.com/v1
  • 智谱 GLM:https://open.bigmodel.cn/api/paas/v4

验证安装

import autogen_agentchat
import autogen_ext
print(f"AgentChat 版本: {autogen_agentchat.__version__}")
print("✅ 安装成功!")

1.3 核心概念速览

在写代码之前,先理解 4 个核心概念:

概念图

┌──────────────────────────────────────────────┐
│                GroupChat (群聊)               │
│                                              │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ Assistant │  │ Assistant │  │UserProxy │  │
│  │  Agent    │  │  Agent    │  │  Agent   │  │
│  │ (AI大脑)  │  │ (AI大脑)  │  │ (人类代理)│  │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  │
│       │              │              │        │
│       └──────────────┼──────────────┘        │
│                      │                       │
│              模型客户端 (LLM)                 │
└──────────────────────────────────────────────┘
概念 类比 说明
AssistantAgent 团队中的专家 由 LLM 驱动,能思考、说话、使用工具
UserProxyAgent 代表用户的项目经理 不用 LLM,代表人类做决策、执行代码
GroupChat 会议室 管理多个 Agent 的对话流程
模型客户端 专家的大脑 连接 LLM 服务(OpenAI、DeepSeek 等)

1.4 第一个 AutoGen 程序

最简示例:两个 Agent 对话

"""
第一个 AutoGen 程序:一个助手 Agent 回答问题
"""
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from dotenv import load_dotenv
import os

load_dotenv()

async def main():
    # 1. 创建模型客户端
    model = OpenAIChatCompletionClient(
        model=os.getenv("LLM_MODEL_ID"),
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL"),
    )

    # 2. 创建助手 Agent
    assistant = AssistantAgent(
        name="helper",
        model_client=model,
        system_message="你是一个有帮助的AI助手。回答完问题后说 'TERMINATE'。",
    )

    # 3. 组建团队(只有1个Agent)
    team = RoundRobinGroupChat(
        participants=[assistant],
        termination_condition=TextMentionTermination("TERMINATE"),
    )

    # 4. 运行
    await Console(team.run_stream(task="用一句话解释什么是人工智能"))

asyncio.run(main())

运行结果

---------- user ----------
用一句话解释什么是人工智能
---------- helper ----------
人工智能是让计算机模拟人类智能行为(如学习、推理、决策)的技术。 TERMINATE

你已经成功运行了第一个 AutoGen 程序! 🎉


第二篇:核心组件深入

2.1 AssistantAgent 详解

AssistantAgent 是 AutoGen 中最核心的组件——一个由 LLM 驱动的智能 Agent。

完整参数

agent = AssistantAgent(
    name="my_agent",                    # Agent 名称(必需)
    model_client=model,                 # 模型客户端(必需)
    system_message="你是...",           # 系统提示词
    tools=[my_tool],                    # 可用工具列表
    description="这个Agent负责...",      # Agent 描述(用于群聊选择)
    handoffs=["other_agent"],           # 可移交给的 Agent 列表
)

system_message 的重要性

system_message 是 Agent 的"灵魂",决定了 Agent 的:

  • 角色定位:你是谁,负责什么
  • 行为规范:应该怎么说话,遵循什么格式
  • 协作规则:什么时候把任务交给别人
# 好的 system_message 示例
system_message = """
你是一个代码审查专家。

## 职责
- 检查代码的安全性、可读性、性能
- 提出具体的改进建议

## 输出格式
1. 问题列表(严重程度:高/中/低)
2. 改进建议
3. 总体评价

## 协作规则
- 审查完成后,将任务移交给 engineer 让他修改
"""

2.2 UserProxyAgent 详解

UserProxyAgent 代表人类用户,不使用 LLM,用于:

  • 人工审核和决策
  • 执行代码
  • 终止对话
from autogen_agentchat.agents import UserProxyAgent

user_proxy = UserProxyAgent(
    name="user",
    description="代表人类用户,负责审核和批准",
)

2.3 模型客户端配置

OpenAI 模型

from autogen_ext.models.openai import OpenAIChatCompletionClient

model = OpenAIChatCompletionClient(
    model="gpt-4o-mini",
    api_key="sk-...",
    base_url="https://api.openai.com/v1",
)

国产模型(DeepSeek / 通义千问 / 智谱)

国产模型需要额外提供 model_info

from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.models import ModelInfo

model = OpenAIChatCompletionClient(
    model="deepseek-chat",
    api_key="your-key",
    base_url="https://api.deepseek.com/v1",
    model_info=ModelInfo(
        vision=False,           # 是否支持图片输入
        function_calling=True,  # 是否支持工具调用
        json_output=True,       # 是否支持 JSON 输出
        family="unknown",       # 模型家族
        structured_output=True, # 是否支持结构化输出
    ),
)

2.4 终止条件

控制对话何时结束,避免无限循环:

from autogen_agentchat.conditions import (
    TextMentionTermination,
    MaxMessageTermination,
    TimeoutTermination,
    HandoffTermination,
    TokenUsageTermination,
    ExternalTermination,
)

# 当某个 Agent 说出 "TERMINATE" 时结束
term1 = TextMentionTermination("TERMINATE")

# 消息总数达到 20 条时结束
term2 = MaxMessageTermination(20)

# 超过 300 秒时结束
term3 = TimeoutTermination(300)

# 组合条件(满足任一即结束)
from autogen_agentchat.conditions import OrTerminationCondition
combined = OrTerminationCondition(term1, term2)

第三篇:工具使用

3.1 什么是工具

工具 = Agent 能调用的外部能力:

  • 查天气、搜索网页、执行代码
  • 调用 API、读写文件、操作数据库

没有工具的 Agent 只能"说",有了工具才能"做"。


3.2 用 @tool 装饰器定义工具

from autogen_core.tools import FunctionTool

# 定义工具函数
def get_weather(city: str) -> str:
    """查询指定城市的天气信息。"""
    # 这里用模拟数据,实际可调用真实 API
    weather_data = {
        "北京": "晴天,25°C",
        "上海": "多云,22°C",
        "广州": "阵雨,28°C",
    }
    return weather_data.get(city, f"未找到 {city} 的天气信息")

# 包装为 AutoGen 工具
weather_tool = FunctionTool(
    func=get_weather,
    description="查询指定城市的天气",
)

关键点

  • 函数必须有类型注解city: str
  • 函数必须有 docstring(工具的描述)
  • FunctionTool 包装后才能被 Agent 使用

3.3 让 Agent 使用工具

assistant = AssistantAgent(
    name="weather_expert",
    model_client=model,
    system_message="你是一个天气查询专家。使用工具查询天气后回答用户。",
    tools=[weather_tool],  # 传入工具列表
)

Agent 会自动:

  1. 分析用户问题
  2. 决定是否需要调用工具
  3. 生成工具调用参数
  4. 执行工具并获取结果
  5. 基于结果生成回答

3.4 实战:带工具的旅行助手

"""
实战:带工具的旅行助手
"""
import asyncio
import requests
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.tools import FunctionTool
from dotenv import load_dotenv
import os

load_dotenv()

# --- 定义工具 ---

def get_weather(city: str) -> str:
    """查询指定城市的实时天气"""
    try:
        url = f"https://wttr.in/{city}?format=j1"
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        data = response.json()
        current = data['current_condition'][0]
        desc = current['weatherDesc'][0]['value']
        temp = current['temp_C']
        return f"{city}当前天气:{desc},气温{temp}°C"
    except Exception as e:
        return f"查询失败:{e}"

def get_attraction(city: str) -> str:
    """根据城市推荐旅游景点"""
    attractions = {
        "北京": ["故宫", "长城", "颐和园", "天坛"],
        "上海": ["外滩", "东方明珠", "豫园", "迪士尼"],
        "西安": ["兵马俑", "大雁塔", "城墙", "回民街"],
    }
    spots = attractions.get(city, ["暂无推荐"])
    return f"{city}推荐景点:{'、'.join(spots)}"

# --- 包装工具 ---

weather_tool = FunctionTool(
    func=get_weather,
    description="查询指定城市的实时天气",
)

attraction_tool = FunctionTool(
    func=get_attraction,
    description="根据城市推荐旅游景点",
)

# --- 主程序 ---

async def main():
    model = OpenAIChatCompletionClient(
        model=os.getenv("LLM_MODEL_ID"),
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL"),
    )

    # 创建旅行助手 Agent
    travel_agent = AssistantAgent(
        name="travel_assistant",
        model_client=model,
        system_message="""你是一个专业的旅行助手。

工作流程:
1. 先用 get_weather 查询目的地天气
2. 再用 get_attraction 推荐景点
3. 综合天气和景点给出旅行建议

回答完毕后说 TERMINATE。""",
        tools=[weather_tool, attraction_tool],
    )

    # 组建团队并运行
    team = RoundRobinGroupChat(
        participants=[travel_agent],
        termination_condition=TextMentionTermination("TERMINATE") | MaxMessageTermination(10),
    )

    await Console(team.run_stream(task="我想去北京旅游,帮我看看天气和推荐景点"))

asyncio.run(main())

第四篇:群聊协作模式

群聊是 AutoGen 最强大的特性——让多个 Agent 通过对话协作。

4.1 RoundRobinGroupChat(轮询群聊)

最简单的群聊模式:Agent 按固定顺序轮流发言。

Agent A → Agent B → Agent C → Agent A → Agent B → ...
from autogen_agentchat.teams import RoundRobinGroupChat

team = RoundRobinGroupChat(
    participants=[agent_a, agent_b, agent_c],
    termination_condition=TextMentionTermination("TERMINATE"),
    max_turns=20,  # 最大轮次
)

适用场景

  • 流水线式任务(A 做完 → B 接着做 → C 接着做)
  • 角色固定、顺序明确的场景

4.2 SelectorGroupChat(选择器群聊)

智能选择:由 LLM 根据当前对话内容,动态选择下一个发言的 Agent。

对话内容 → LLM 分析 → 选择最合适的 Agent → 该 Agent 发言
from autogen_agentchat.teams import SelectorGroupChat

team = SelectorGroupChat(
    participants=[planner, coder, reviewer, tester],
    model_client=model,  # 选择器使用的 LLM
    termination_condition=TextMentionTermination("TERMINATE"),
)

适用场景

  • 不确定谁应该下一步发言
  • 需要根据上下文动态决策
  • 复杂的多角色协作

4.3 Swarm(蜂群模式)

动态移交:Agent 可以主动把任务移交给其他 Agent。

from autogen_agentchat.teams import Swarm

# Agent 定义时指定可移交的目标
triage_agent = AssistantAgent(
    name="triage",
    model_client=model,
    system_message="你是接待员。分析问题类型,移交给合适的专家。",
    handoffs=["tech_support", "billing", "general"],
)

tech_agent = AssistantAgent(
    name="tech_support",
    model_client=model,
    system_message="你是技术支持专家。解决技术问题后移交给 triage。",
    handoffs=["triage"],
)

team = Swarm(
    participants=[triage_agent, tech_agent, billing_agent],
    termination_condition=TextMentionTermination("TERMINATE"),
)

适用场景

  • 客服系统(接待→分流→处理→回访)
  • 需要动态路由的任务

第五篇:实战项目

5.1 项目一:软件开发团队

这是本项目 code/chapter6/AutoGenDemo/ 中的完整案例。

团队组成

┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│ ProductManager│ →  │  Engineer   │ →  │ CodeReviewer │ →  │  UserProxy  │
│   产品经理    │     │   工程师    │     │  代码审查员  │     │   用户代理   │
└─────────────┘     └─────────────┘     └─────────────┘     └─────────────┘

完整代码

"""
AutoGen 软件开发团队
"""
import asyncio
from autogen_agentchat.agents import AssistantAgent, UserProxyAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from dotenv import load_dotenv
import os

load_dotenv()

async def main():
    # 模型客户端
    model = OpenAIChatCompletionClient(
        model=os.getenv("LLM_MODEL_ID"),
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL"),
    )

    # === 定义团队成员 ===

    product_manager = AssistantAgent(
        name="ProductManager",
        model_client=model,
        system_message="""你是产品经理。
- 分析用户需求,制定产品方案
- 输出清晰的功能需求列表和开发计划
- 完成后将任务交给 Engineer""",
    )

    engineer = AssistantAgent(
        name="Engineer",
        model_client=model,
        system_message="""你是高级工程师。
- 根据产品经理的需求编写代码
- 代码要求:可运行、有注释、结构清晰
- 完成后将任务交给 CodeReviewer""",
    )

    code_reviewer = AssistantAgent(
        name="CodeReviewer",
        model_client=model,
        system_message="""你是代码审查专家。
- 审查代码的安全性、可读性、性能
- 列出问题和改进建议
- 如果代码质量OK,说 APPROVED
- 完成后将任务交给 UserProxy""",
    )

    user_proxy = UserProxyAgent(
        name="UserProxy",
        description="代表用户进行最终验收",
    )

    # === 组建团队 ===

    team = RoundRobinGroupChat(
        participants=[product_manager, engineer, code_reviewer, user_proxy],
        termination_condition=TextMentionTermination("TERMINATE"),
        max_turns=20,
    )

    # === 运行 ===

    task = "创建一个 Python Streamlit 应用,实时显示比特币价格"

    await Console(team.run_stream(task=task))

asyncio.run(main())

对话流程分析

用户: 创建一个 Streamlit 应用,实时显示比特币价格

ProductManager: 需求分析 → 功能列表 → 开发计划
    ↓
Engineer: 编写代码 → 输出完整 Python 文件
    ↓
CodeReviewer: 审查代码 → 发现问题 → 建议修改
    ↓ (如果需要修改,下一轮 Engineer 会修改)
UserProxy: 验收测试 → 说 TERMINATE

5.2 项目二:深度研究助手

"""
深度研究助手:搜索 + 分析 + 总结
"""
import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import SelectorGroupChat
from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
from autogen_agentchat.ui import Console
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_core.tools import FunctionTool
from dotenv import load_dotenv
import os

load_dotenv()

# --- 工具定义 ---

def search_web(query: str) -> str:
    """模拟网络搜索"""
    # 实际项目中可接入 Tavily、Bing 等搜索 API
    return f"搜索结果:关于 '{query}' 的最新信息..."

def save_report(content: str, filename: str) -> str:
    """保存研究报告到文件"""
    with open(filename, "w", encoding="utf-8") as f:
        f.write(content)
    return f"报告已保存到 {filename}"

# --- 主程序 ---

async def main():
    model = OpenAIChatCompletionClient(
        model=os.getenv("LLM_MODEL_ID"),
        api_key=os.getenv("LLM_API_KEY"),
        base_url=os.getenv("LLM_BASE_URL"),
    )

    searcher = AssistantAgent(
        name="searcher",
        model_client=model,
        system_message="你是搜索专家。根据主题搜索相关信息,整理成要点。",
        tools=[FunctionTool(search_web, description="搜索网络信息")],
    )

    analyst = AssistantAgent(
        name="analyst",
        model_client=model,
        system_message="你是分析师。基于搜索结果进行深入分析,提出见解。",
    )

    writer = AssistantAgent(
        name="writer",
        model_client=model,
        system_message="你是报告撰写者。将分析结果整理成结构化的研究报告。完成后说 TERMINATE。",
        tools=[FunctionTool(save_report, description="保存报告")],
    )

    team = SelectorGroupChat(
        participants=[searcher, analyst, writer],
        model_client=model,
        termination_condition=TextMentionTermination("TERMINATE") | MaxMessageTermination(15),
    )

    await Console(team.run_stream(task="研究 2025 年 AI Agent 技术发展趋势"))

asyncio.run(main())

第六篇:进阶高级用法

6.1 自定义 Agent

当内置 Agent 不能满足需求时,可以创建自定义 Agent:

from autogen_agentchat.agents import BaseChatAgent
from autogen_agentchat.messages import ChatMessage, TextMessage
from autogen_core import CancellationToken

class MathAgent(BaseChatAgent):
    """一个专门做数学计算的 Agent"""

    def __init__(self, name: str):
        super().__init__(name=name, description="数学计算专家")

    @property
    def produced_message_types(self):
        return [TextMessage]

    async def on_messages(
        self, messages: list[ChatMessage], cancellation_token: CancellationToken
    ):
        # 获取最后一条消息
        last_msg = messages[-1]

        # 简单的数学处理逻辑
        response = f"数学计算结果: {last_msg.content}"

        return TextMessage(content=response, source=self.name)

6.2 流式输出

实时显示 Agent 的思考过程:

# 使用 Console 工具实现流式输出
from autogen_agentchat.ui import Console

# 这会实时打印每个 Agent 的消息
await Console(team.run_stream(task="你的任务"))

6.3 代码执行

让 Agent 生成并执行代码:

from autogen_ext.code_executor.docker_executor import DockerCodeExecutor

# 创建 Docker 沙箱执行器
executor = DockerCodeExecutor()

# 在 Agent 中使用
agent = AssistantAgent(
    name="coder",
    model_client=model,
    code_execution_config={"executor": executor},
)

6.4 MCP 集成

连接 MCP 服务器获取工具:

from autogen_ext.tools.mcp import McpToolAdapter, StdioServerParams

# 配置 MCP 服务器
server_params = StdioServerParams(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-filesystem", "/path/to/dir"],
)

# 获取工具列表
tools = await McpToolAdapter.from_server(server_params)

# 给 Agent 使用
agent = AssistantAgent(
    name="file_manager",
    model_client=model,
    tools=tools,
)

6.5 AutoGen Studio 可视化

AutoGen Studio 提供了一个 Web 界面来构建和测试 Agent:

# 安装
pip install -U autogenstudio

# 启动
autogenstudio ui --port 8080

打开浏览器访问 http://localhost:8080,可以:

  • 可视化创建 Agent 团队
  • 拖拽配置 Agent 关系
  • 实时测试对话效果
  • 导出为 Python 代码

第七篇:最佳实践与常见问题

最佳实践

1. System Message 设计原则

好的 system_message = 角色 + 职责 + 输出格式 + 协作规则
system_message = """
## 角色
你是一个 {角色名}。

## 职责
- 职责1
- 职责2

## 输出格式
1. xxx
2. xxx

## 协作规则
- 完成后移交给 {下一个Agent}
- 遇到 {情况} 时,移交给 {对应Agent}
"""

2. 选择合适的群聊模式

场景 推荐模式
固定流程(A→B→C) RoundRobinGroupChat
需要动态选择 SelectorGroupChat
客服/分流系统 Swarm
简单问答 单 Agent,不用群聊

3. 设置合理的终止条件

# 总是组合使用:内容条件 + 安全上限
termination = (
    TextMentionTermination("TERMINATE")  # 内容条件
    | MaxMessageTermination(20)           # 安全上限
)

4. 错误处理

try:
    result = await team.run_stream(task=task)
except Exception as e:
    print(f"Agent 执行出错: {e}")
    # 重试或降级处理

常见问题

Q1: Agent 一直循环不结束?

原因:没有设置终止条件,或 Agent 没有说 TERMINATE。

解决

# 组合终止条件
termination = TextMentionTermination("TERMINATE") | MaxMessageTermination(20)

Q2: 国产模型报错?

原因:需要提供 model_info

解决

from autogen_core.models import ModelInfo

model = OpenAIChatCompletionClient(
    model="deepseek-chat",
    api_key="...",
    base_url="...",
    model_info=ModelInfo(
        vision=False,
        function_calling=True,
        json_output=True,
        family="unknown",
        structured_output=True,
    ),
)

Q3: 工具不被调用?

原因:函数缺少类型注解或 docstring。

解决

def my_tool(param: str) -> str:  # 必须有类型注解
    """工具描述(必须有 docstring)"""  # 必须有 docstring
    return "result"

Q4: 如何调试?

import logging
logging.basicConfig(level=logging.DEBUG)

# 或者打印消息历史
async for message in team.run_stream(task=task):
    print(f"[{message.source}] {message.content}")

学习路径建议

入门 → 单 Agent + 简单工具
  ↓
进阶 → 多 Agent 群聊 + 复杂工具
  ↓
高级 → 自定义 Agent + MCP + 代码执行
  ↓
实战 → 构建完整的多 Agent 应用

参考资源


Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐