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


    
通过
需修改
测试失败
测试通过
用户输入
45%25%20%10%AutoGen Agent 类型使用分布AssistantAgent (LLM 驱动)UserProxyAgent (人工介入)GroupChat (多 Agent 协作)自定义 Agent
pip install"autogen-agentchat[gemini]>=0.4""autogen-ext[gemini]" python-dotenv 
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.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 连接池类，支持： 
 
”“1. 连接获取和归还 ““2. 最大连接数限制 ““3. 空闲连接超时回收 ““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=(“你是产品经理。负责： ““1. 理解用户需求并拆解为具体任务 ““2. 定义功能边界和验收标准 ““3. 确认最终交付物满足需求 ““满意时回复 ‘TASK_DONE’。用中文回复。”),)# 架构师 architect = AssistantAgent( name=“Architect”, model_client=model_client, system_message=(“你是系统架构师。负责： ““1. 设计模块划分和类图 ““2. 选择合适的设计模式 ““3. 定义接口规范 ““用中文回复。”),)# 开发工程师 developer = AssistantAgent( name=“Developer”, model_client=model_client, system_message=(“你是高级开发工程师。负责： ““1. 根据架构设计编写代码 ““2. 代码必须包含类型注解和文档字符串 ““3. 遵循 SOLID 原则 ““用中文回复，代码放在 python 代码块中。"),)# 测试工程师 tester = AssistantAgent( name="Tester", model_client=model_client, system_message=("你是测试工程师。负责： ""1. 为代码编写单元测试 ""2. 覆盖正常流程和边界情况 ""3. 使用 pytest 框架 ""用中文回复，测试代码放在python 代码块中。”),)# Selector 选择器：由 LLM 决定下一步由哪个 Agent 发言 selector_prompt =(“你是一个团队协调者。根据当前对话进展，选择最合适的下一位发言者。 ““需求分析阶段 -> PM ““架构设计阶段 -> Architect ““编码阶段 -> Developer ““测试阶段 -> Tester 
”“可选的发言者: {participants} ““当前对话历史: {history}
 
”“请只返回发言者的名字。”) 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 任务调度器，支持： ““1. 定时执行任务（cron 表达式） ““2. 任务优先级排序 ““3. 失败重试机制（指数退避） ““4. 任务状态追踪（pending/running/success/failed） ““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,” 
  
    
    
      ”,“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”““definit(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 =” 历史记忆: “for m in self.memory[-self.max_memory:]: memory_context +=f”- {m[‘role’]}: {m[‘content’][:100]} “ full_prompt =(f”{self.system_message} 
“f”{memory_context} “f”用户: {user_message} “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” 用户: {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())

模式三: 管道

Planner

Coder

Reviewer

Tester

模式二: 选择器

Selector

Agent A

Agent B

Agent C

模式一: 轮询

Agent A

Agent B

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

========== 软件开发团队协作 ========== [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 ✅ 交付物满足所有需求

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

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

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

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