AI Agent 智能体开发入门：AutoGen 多智能体协作实战教程

本文通过 AutoGen 框架，从单智能体到多智能体协作，循序渐进地讲解如何构建 AI Agent 系统，包含完整的代码示例和架构设计。

1. 多智能体协作架构

通过

需修改

测试失败

测试通过

用户输入

编排者 Agent
GroupChat Manager

规划 Agent
分解任务

编码 Agent
编写代码

审查 Agent
代码审查

测试 Agent
执行测试

总结 Agent
汇总报告

2. AutoGen 核心概念

45%25%20%10%AutoGen Agent 类型使用分布AssistantAgent (LLM 驱动)UserProxyAgent (人工介入)GroupChat (多 Agent 协作)自定义 Agent

3. 环境准备

pip install"autogen-agentchat[gemini]>=0.4""autogen-ext[gemini]" python-dotenv 

4. 项目结构

autogen-demo/ ├── 01_single_agent.py # 单智能体对话 ├── 02_two_agent_chat.py # 双智能体协作 ├── 03_group_chat.py # 多智能体群聊 ├── 04_code_generation.py # 代码生成工作流 ├── 05_custom_agent.py # 自定义 Agent ├── config.py # 配置管理 └── .env 

5. 核心代码实现

5.1 配置管理

# config.pyimport os from dotenv import load_dotenv load_dotenv()# AutoGen 0.4+ 使用新的模型客户端 API GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")# 默认使用 Gemini（免费额度） DEFAULT_MODEL ="gemini-2.0-flash"

5.2 单智能体对话

# 01_single_agent.py""" 最简单的 Agent 示例：一个 AI 助手 + 一个用户代理 """import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.ui import Console from autogen_ext.models.gemini import GeminiModelClient asyncdefsingle_agent_chat():# 创建 Gemini 模型客户端 model_client = GeminiModelClient(model="gemini-2.0-flash")# 创建 AI 助手 Agent assistant = AssistantAgent( name="AI_Assistant", model_client=model_client, system_message=("你是一个专业的 Python 开发助手。""回答要简洁、准确，附带代码示例。""使用中文回答。"),)# 单次对话 result =await assistant.run(task="用 Python 实现一个 LRU 缓存，要求支持 O(1) 的 get 和 put")print(result.messages[-1].content)# 模型客户端需要关闭await model_client.close()if __name__ =="__main__": asyncio.run(single_agent_chat())

5.3 双智能体协作

# 02_two_agent_chat.py""" 两个 Agent 协作完成编码任务： - Coder: 负责编写代码 - Reviewer: 负责审查代码 """import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.teams import RoundRobinGroupChat from autogen_agentchat.ui import Console from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination from autogen_ext.models.gemini import GeminiModelClient asyncdeftwo_agent_collaboration(): model_client = GeminiModelClient(model="gemini-2.0-flash")# 编码 Agent coder = AssistantAgent( name="Coder", model_client=model_client, system_message=("你是一个资深 Python 工程师。""根据需求编写高质量代码，包含类型注解和错误处理。""当审查者提出修改意见时，你要更新代码。""请用中文回复。"),)# 审查 Agent reviewer = AssistantAgent( name="Reviewer", model_client=model_client, system_message=("你是一个严格的代码审查专家。""检查代码的正确性、性能、安全性、可读性。""如果代码满意，回复 'APPROVED'。""否则给出具体修改意见。""请用中文回复。"),)# 终止条件：出现 APPROVED 或超过 10 轮 termination = TextMentionTermination("APPROVED")| MaxMessageTermination(10)# 轮询群聊：Coder -> Reviewer -> Coder -> ... team = RoundRobinGroupChat( participants=[coder, reviewer], termination_condition=termination,)# 运行协作await Console( team.run_stream( task="实现一个线程安全的 Python 连接池类，支持：\n""1. 连接获取和归还\n""2. 最大连接数限制\n""3. 空闲连接超时回收\n""4. 上下文管理器支持"))await model_client.close()if __name__ =="__main__": asyncio.run(two_agent_collaboration())

5.4 多智能体群聊（软件开发团队）

# 03_group_chat.py""" 模拟软件开发团队：产品经理 + 架构师 + 开发 + 测试 通过 GroupChat Manager 协调多个 Agent 协作 """import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.teams import SelectorGroupChat from autogen_agentchat.ui import Console from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination from autogen_ext.models.gemini import GeminiModelClient asyncdefsoftware_dev_team(): model_client = GeminiModelClient(model="gemini-2.0-flash")# 产品经理 pm = AssistantAgent( name="PM", model_client=model_client, system_message=("你是产品经理。负责：\n""1. 理解用户需求并拆解为具体任务\n""2. 定义功能边界和验收标准\n""3. 确认最终交付物满足需求\n""满意时回复 'TASK_DONE'。用中文回复。"),)# 架构师 architect = AssistantAgent( name="Architect", model_client=model_client, system_message=("你是系统架构师。负责：\n""1. 设计模块划分和类图\n""2. 选择合适的设计模式\n""3. 定义接口规范\n""用中文回复。"),)# 开发工程师 developer = AssistantAgent( name="Developer", model_client=model_client, system_message=("你是高级开发工程师。负责：\n""1. 根据架构设计编写代码\n""2. 代码必须包含类型注解和文档字符串\n""3. 遵循 SOLID 原则\n""用中文回复，代码放在 ```python 代码块中。"),)# 测试工程师 tester = AssistantAgent( name="Tester", model_client=model_client, system_message=("你是测试工程师。负责：\n""1. 为代码编写单元测试\n""2. 覆盖正常流程和边界情况\n""3. 使用 pytest 框架\n""用中文回复，测试代码放在 ```python 代码块中。"),)# Selector 选择器：由 LLM 决定下一步由哪个 Agent 发言 selector_prompt =("你是一个团队协调者。根据当前对话进展，选择最合适的下一位发言者。\n""需求分析阶段 -> PM\n""架构设计阶段 -> Architect\n""编码阶段 -> Developer\n""测试阶段 -> Tester\n\n""可选的发言者: {participants}\n""当前对话历史:\n{history}\n\n""请只返回发言者的名字。") termination = TextMentionTermination("TASK_DONE")| MaxMessageTermination(20) team = SelectorGroupChat( participants=[pm, architect, developer, tester], model_client=model_client, termination_condition=termination, selector_prompt=selector_prompt,)await Console( team.run_stream( task="开发一个 Python 任务调度器，支持：\n""1. 定时执行任务（cron 表达式）\n""2. 任务优先级排序\n""3. 失败重试机制（指数退避）\n""4. 任务状态追踪（pending/running/success/failed）\n""5. 异步并发执行"))await model_client.close()if __name__ =="__main__": asyncio.run(software_dev_team())

5.5 带工具调用的 Agent

# 04_tool_agent.py""" Agent 配备工具：让 AI 能执行实际操作 """import asyncio import json import aiohttp from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.ui import Console from autogen_ext.models.gemini import GeminiModelClient from autogen_core.tools import FunctionTool # ---- 定义工具函数 ----asyncdeffetch_webpage(url:str)->str:"""获取网页内容"""asyncwith aiohttp.ClientSession()as session:asyncwith session.get(url, timeout=aiohttp.ClientTimeout(total=10))as resp:returnawait resp.text()[:5000]defcalculate(expression:str)->str:"""安全计算数学表达式"""import ast import operator ops ={ ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow,}try: tree = ast.parse(expression, mode="eval") result =eval(compile(tree,"<string>","eval"),{"__builtins__":{}}, ops)returnstr(result)except Exception as e:returnf"计算错误: {e}"defsearch_codebase(query:str, directory:str=".")->str:"""在代码库中搜索关键词"""import subprocess try: result = subprocess.run(["grep","-rn", query, directory,"--include=*.py"], capture_output=True, text=True, timeout=10)return result.stdout[:3000]if result.stdout else"未找到匹配"except Exception as e:returnf"搜索失败: {e}"asyncdeftool_agent_demo(): model_client = GeminiModelClient(model="gemini-2.0-flash")# 注册工具 fetch_tool = FunctionTool(fetch_webpage, description="获取指定 URL 的网页内容") calc_tool = FunctionTool(calculate, description="计算数学表达式") search_tool = FunctionTool(search_codebase, description="在代码库中搜索关键词")# 创建带工具的 Agent agent = AssistantAgent( name="ToolAgent", model_client=model_client, tools=[fetch_tool, calc_tool, search_tool], system_message=("你是一个有工具使用能力的 AI 助手。""你可以获取网页、执行计算、搜索代码。""请根据用户需求选择合适的工具。用中文回复。"),)# 测试工具调用 result =await agent.run( task="请帮我计算 (1024 * 768) / (16 * 16) 的结果")print(result.messages[-1].content)await model_client.close()if __name__ =="__main__": asyncio.run(tool_agent_demo())

5.6 自定义 Agent

# 05_custom_agent.py""" 自定义 Agent：实现一个带记忆的智能客服 """import asyncio from typing import AsyncGenerator, List from autogen_agentchat.agents import BaseChatAgent from autogen_agentchat.base import Response from autogen_agentchat.messages import ChatMessage, TextMessage from autogen_core import CancellationToken from autogen_ext.models.gemini import GeminiModelClient classMemoryEnabledAgent(BaseChatAgent):"""带长期记忆的自定义 Agent"""def__init__(self, name:str, model_client: GeminiModelClient, system_message:str, max_memory:int=10):super().__init__(name=name, description="带记忆的智能客服") self.model_client = model_client self.system_message = system_message self.memory: List[dict]=[] self.max_memory = max_memory asyncdefon_messages(self, messages: List[ChatMessage], cancellation_token: CancellationToken)-> Response:# 获取最新用户消息 user_message = messages[-1].content if messages else""# 构建带记忆的 prompt memory_context =""if self.memory: memory_context ="\n历史记忆:\n"for m in self.memory[-self.max_memory:]: memory_context +=f"- {m['role']}: {m['content'][:100]}\n" full_prompt =(f"{self.system_message}\n\n"f"{memory_context}\n"f"用户: {user_message}\n"f"助手:")# 调用模型from autogen_core.models import UserMessage response =await self.model_client.create( messages=[UserMessage(content=full_prompt, source="user")])# 存入记忆 self.memory.append({"role":"user","content": user_message}) self.memory.append({"role":"assistant","content": response.content})return Response( chat_message=TextMessage(content=response.content, source=self.name))asyncdefon_reset(self, cancellation_token: CancellationToken)->None: self.memory.clear()asyncdefcustom_agent_demo(): model_client = GeminiModelClient(model="gemini-2.0-flash") agent = MemoryEnabledAgent( name="SmartCustomerService", model_client=model_client, system_message=("你是一个电商平台的智能客服。""记住用户提到过的所有信息（商品、订单号、偏好等）。""友好、专业地回答问题。用中文回复。"),)# 模拟多轮对话 questions =["我想买一台笔记本电脑，预算 6000 左右","有什么推荐的品牌吗？我之前用过联想","好的，那帮我查一下ThinkPad系列的","对了，刚才我说我的预算是多少来着？",# 测试记忆]for q in questions:print(f"\n用户: {q}") result =await agent.on_messages([TextMessage(content=q, source="user")], CancellationToken(),)print(f"客服: {result.chat_message.content}")await model_client.close()if __name__ =="__main__": asyncio.run(custom_agent_demo())

6. Agent 协作模式对比

模式三: 管道

Planner

Coder

Reviewer

Tester

模式二: 选择器

Selector

Agent A

Agent B

Agent C

模式一: 轮询

Agent A

Agent B

7. 各模式适用场景

35%30%25%10%Agent 协作模式适用场景轮询 RoundRobin选择器 Selector管道 Pipeline嵌套 Nested

8. 运行效果示例

========== 软件开发团队协作 ========== [PM] 需求分析： 用户需要一个任务调度器。我将需求拆解为： 1. TaskScheduler 核心类 - 管理任务生命周期 2. CronParser - 解析 cron 表达式 3. RetryPolicy - 失败重试策略 验收标准：支持异步、优先级、重试、状态追踪 [Architect] 架构设计： 采用以下设计模式： - 策略模式：RetryPolicy 可替换 - 观察者模式：状态变更通知 - 优先级队列：heapq 实现 类图：Scheduler -> Task -> RetryPolicy [Developer] 编码实现： ```python class TaskScheduler: ... 

[Tester] 编写测试：

@pytest.mark.asyncioasyncdeftest_scheduler_retry():...

[PM] TASK_DONE ✅ 交付物满足所有需求

9. 最佳实践

1. System Message 要精确 —— 明确每个 Agent 的职责边界，避免角色重叠
2. 设置终止条件 —— 用 MaxMessageTermination 兜底，防止无限循环
3. 控制 Agent 数量 —— 建议 3-5 个，过多会导致上下文过长和选择困难
4. 工具要类型安全 —— 使用 FunctionTool 包装，确保参数校验
5. 记忆窗口有限 —— 长对话注意上下文截断，重要信息可存入外部存储
6. 模型选择 —— 简单任务用 Flash，复杂推理用 Pro/GPT-4o

10. 总结

本文从零到一构建了多智能体协作系统：

1. 单 Agent —— 最基础的 AI 对话能力
2. 双 Agent 协作 —— Coder + Reviewer 模式实现代码审查
3. 多 Agent 群聊 —— 模拟完整软件开发团队（PM + 架构师 + 开发 + 测试）
4. 工具调用 —— 赋予 Agent 执行实际操作的能力
5. 自定义 Agent —— 带记忆的智能客服实现

AutoGen 0.4+ 已全面支持异步，推荐使用 asyncio 运行。Gemini Flash 模型免费额度充足，适合学习和开发。