AutoGen 完整教程:从入门到进阶
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 会自动:
- 分析用户问题
- 决定是否需要调用工具
- 生成工具调用参数
- 执行工具并获取结果
- 基于结果生成回答
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 应用
参考资源
更多推荐

所有评论(0)