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. 环境准备

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

4. 项目结构

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

5. 核心代码实现

5.1 配置管理

1# config.py
2import os
3from dotenv import load_dotenv
4
5load_dotenv()
6
7# AutoGen 0.4+ 使用新的模型客户端 API
8GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
9OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
10
11# 默认使用 Gemini（免费额度）
12DEFAULT_MODEL = "gemini-2.0-flash"
13

5.2 单智能体对话

1# 01_single_agent.py
2"""
3最简单的 Agent 示例：一个 AI 助手 + 一个用户代理
4"""
5import asyncio
6from autogen_agentchat.agents import AssistantAgent
7from autogen_agentchat.ui import Console
8from autogen_ext.models.gemini import GeminiModelClient
9
10
11async def single_agent_chat():
12    # 创建 Gemini 模型客户端
13    model_client = GeminiModelClient(model="gemini-2.0-flash")
14
15    # 创建 AI 助手 Agent
16    assistant = AssistantAgent(
17        name="AI_Assistant",
18        model_client=model_client,
19        system_message=(
20            "你是一个专业的 Python 开发助手。"
21            "回答要简洁、准确，附带代码示例。"
22            "使用中文回答。"
23        ),
24    )
25
26    # 单次对话
27    result = await assistant.run(task="用 Python 实现一个 LRU 缓存，要求支持 O(1) 的 get 和 put")
28    print(result.messages[-1].content)
29
30    # 模型客户端需要关闭
31    await model_client.close()
32
33
34if __name__ == "__main__":
35    asyncio.run(single_agent_chat())
36

5.3 双智能体协作

1# 02_two_agent_chat.py
2"""
3两个 Agent 协作完成编码任务：
4- Coder: 负责编写代码
5- Reviewer: 负责审查代码
6"""
7import asyncio
8from autogen_agentchat.agents import AssistantAgent
9from autogen_agentchat.teams import RoundRobinGroupChat
10from autogen_agentchat.ui import Console
11from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
12from autogen_ext.models.gemini import GeminiModelClient
13
14
15async def two_agent_collaboration():
16    model_client = GeminiModelClient(model="gemini-2.0-flash")
17
18    # 编码 Agent
19    coder = AssistantAgent(
20        name="Coder",
21        model_client=model_client,
22        system_message=(
23            "你是一个资深 Python 工程师。"
24            "根据需求编写高质量代码，包含类型注解和错误处理。"
25            "当审查者提出修改意见时，你要更新代码。"
26            "请用中文回复。"
27        ),
28    )
29
30    # 审查 Agent
31    reviewer = AssistantAgent(
32        name="Reviewer",
33        model_client=model_client,
34        system_message=(
35            "你是一个严格的代码审查专家。"
36            "检查代码的正确性、性能、安全性、可读性。"
37            "如果代码满意，回复 'APPROVED'。"
38            "否则给出具体修改意见。"
39            "请用中文回复。"
40        ),
41    )
42
43    # 终止条件：出现 APPROVED 或超过 10 轮
44    termination = TextMentionTermination("APPROVED") | MaxMessageTermination(10)
45
46    # 轮询群聊：Coder -> Reviewer -> Coder -> ...
47    team = RoundRobinGroupChat(
48        participants=[coder, reviewer],
49        termination_condition=termination,
50    )
51
52    # 运行协作
53    await Console(
54        team.run_stream(
55            task="实现一个线程安全的 Python 连接池类，支持：\n"
56                 "1. 连接获取和归还\n"
57                 "2. 最大连接数限制\n"
58                 "3. 空闲连接超时回收\n"
59                 "4. 上下文管理器支持"
60        )
61    )
62
63    await model_client.close()
64
65
66if __name__ == "__main__":
67    asyncio.run(two_agent_collaboration())
68

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

1# 03_group_chat.py
2"""
3模拟软件开发团队：产品经理 + 架构师 + 开发 + 测试
4通过 GroupChat Manager 协调多个 Agent 协作
5"""
6import asyncio
7from autogen_agentchat.agents import AssistantAgent
8from autogen_agentchat.teams import SelectorGroupChat
9from autogen_agentchat.ui import Console
10from autogen_agentchat.conditions import TextMentionTermination, MaxMessageTermination
11from autogen_ext.models.gemini import GeminiModelClient
12
13
14async def software_dev_team():
15    model_client = GeminiModelClient(model="gemini-2.0-flash")
16
17    # 产品经理
18    pm = AssistantAgent(
19        name="PM",
20        model_client=model_client,
21        system_message=(
22            "你是产品经理。负责：\n"
23            "1. 理解用户需求并拆解为具体任务\n"
24            "2. 定义功能边界和验收标准\n"
25            "3. 确认最终交付物满足需求\n"
26            "满意时回复 'TASK_DONE'。用中文回复。"
27        ),
28    )
29
30    # 架构师
31    architect = AssistantAgent(
32        name="Architect",
33        model_client=model_client,
34        system_message=(
35            "你是系统架构师。负责：\n"
36            "1. 设计模块划分和类图\n"
37            "2. 选择合适的设计模式\n"
38            "3. 定义接口规范\n"
39            "用中文回复。"
40        ),
41    )
42
43    # 开发工程师
44    developer = AssistantAgent(
45        name="Developer",
46        model_client=model_client,
47        system_message=(
48            "你是高级开发工程师。负责：\n"
49            "1. 根据架构设计编写代码\n"
50            "2. 代码必须包含类型注解和文档字符串\n"
51            "3. 遵循 SOLID 原则\n"
52            "用中文回复，代码放在 ```python 代码块中。"
53        ),
54    )
55
56    # 测试工程师
57    tester = AssistantAgent(
58        name="Tester",
59        model_client=model_client,
60        system_message=(
61            "你是测试工程师。负责：\n"
62            "1. 为代码编写单元测试\n"
63            "2. 覆盖正常流程和边界情况\n"
64            "3. 使用 pytest 框架\n"
65            "用中文回复，测试代码放在 ```python 代码块中。"
66        ),
67    )
68
69    # Selector 选择器：由 LLM 决定下一步由哪个 Agent 发言
70    selector_prompt = (
71        "你是一个团队协调者。根据当前对话进展，选择最合适的下一位发言者。\n"
72        "需求分析阶段 -> PM\n"
73        "架构设计阶段 -> Architect\n"
74        "编码阶段 -> Developer\n"
75        "测试阶段 -> Tester\n\n"
76        "可选的发言者: {participants}\n"
77        "当前对话历史:\n{history}\n\n"
78        "请只返回发言者的名字。"
79    )
80
81    termination = TextMentionTermination("TASK_DONE") | MaxMessageTermination(20)
82
83    team = SelectorGroupChat(
84        participants=[pm, architect, developer, tester],
85        model_client=model_client,
86        termination_condition=termination,
87        selector_prompt=selector_prompt,
88    )
89
90    await Console(
91        team.run_stream(
92            task="开发一个 Python 任务调度器，支持：\n"
93                 "1. 定时执行任务（cron 表达式）\n"
94                 "2. 任务优先级排序\n"
95                 "3. 失败重试机制（指数退避）\n"
96                 "4. 任务状态追踪（pending/running/success/failed）\n"
97                 "5. 异步并发执行"
98        )
99    )
100
101    await model_client.close()
102
103
104if __name__ == "__main__":
105    asyncio.run(software_dev_team())
106

5.5 带工具调用的 Agent

1# 04_tool_agent.py
2"""
3Agent 配备工具：让 AI 能执行实际操作
4"""
5import asyncio
6import json
7import aiohttp
8from autogen_agentchat.agents import AssistantAgent
9from autogen_agentchat.ui import Console
10from autogen_ext.models.gemini import GeminiModelClient
11from autogen_core.tools import FunctionTool
12
13
14# ---- 定义工具函数 ----
15
16async def fetch_webpage(url: str) -> str:
17    """获取网页内容"""
18    async with aiohttp.ClientSession() as session:
19        async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
20            return await resp.text()[:5000]
21
22
23def calculate(expression: str) -> str:
24    """安全计算数学表达式"""
25    import ast
26    import operator
27
28    ops = {
29        ast.Add: operator.add,
30        ast.Sub: operator.sub,
31        ast.Mult: operator.mul,
32        ast.Div: operator.truediv,
33        ast.Pow: operator.pow,
34    }
35
36    try:
37        tree = ast.parse(expression, mode="eval")
38        result = eval(compile(tree, "<string>", "eval"),
39                      {"__builtins__": {}}, ops)
40        return str(result)
41    except Exception as e:
42        return f"计算错误: {e}"
43
44
45def search_codebase(query: str, directory: str = ".") -> str:
46    """在代码库中搜索关键词"""
47    import subprocess
48    try:
49        result = subprocess.run(
50            ["grep", "-rn", query, directory, "--include=*.py"],
51            capture_output=True, text=True, timeout=10
52        )
53        return result.stdout[:3000] if result.stdout else "未找到匹配"
54    except Exception as e:
55        return f"搜索失败: {e}"
56
57
58async def tool_agent_demo():
59    model_client = GeminiModelClient(model="gemini-2.0-flash")
60
61    # 注册工具
62    fetch_tool = FunctionTool(fetch_webpage, description="获取指定 URL 的网页内容")
63    calc_tool = FunctionTool(calculate, description="计算数学表达式")
64    search_tool = FunctionTool(search_codebase, description="在代码库中搜索关键词")
65
66    # 创建带工具的 Agent
67    agent = AssistantAgent(
68        name="ToolAgent",
69        model_client=model_client,
70        tools=[fetch_tool, calc_tool, search_tool],
71        system_message=(
72            "你是一个有工具使用能力的 AI 助手。"
73            "你可以获取网页、执行计算、搜索代码。"
74            "请根据用户需求选择合适的工具。用中文回复。"
75        ),
76    )
77
78    # 测试工具调用
79    result = await agent.run(
80        task="请帮我计算 (1024 * 768) / (16 * 16) 的结果"
81    )
82    print(result.messages[-1].content)
83
84    await model_client.close()
85
86
87if __name__ == "__main__":
88    asyncio.run(tool_agent_demo())
89

5.6 自定义 Agent

1# 05_custom_agent.py
2"""
3自定义 Agent：实现一个带记忆的智能客服
4"""
5import asyncio
6from typing import AsyncGenerator, List
7from autogen_agentchat.agents import BaseChatAgent
8from autogen_agentchat.base import Response
9from autogen_agentchat.messages import ChatMessage, TextMessage
10from autogen_core import CancellationToken
11from autogen_ext.models.gemini import GeminiModelClient
12
13
14class MemoryEnabledAgent(BaseChatAgent):
15    """带长期记忆的自定义 Agent"""
16
17    def __init__(self, name: str, model_client: GeminiModelClient,
18                 system_message: str, max_memory: int = 10):
19        super().__init__(name=name, description="带记忆的智能客服")
20        self.model_client = model_client
21        self.system_message = system_message
22        self.memory: List[dict] = []
23        self.max_memory = max_memory
24
25    async def on_messages(self, messages: List[ChatMessage],
26                          cancellation_token: CancellationToken) -> Response:
27        # 获取最新用户消息
28        user_message = messages[-1].content if messages else ""
29
30        # 构建带记忆的 prompt
31        memory_context = ""
32        if self.memory:
33            memory_context = "\n历史记忆:\n"
34            for m in self.memory[-self.max_memory:]:
35                memory_context += f"- {m['role']}: {m['content'][:100]}\n"
36
37        full_prompt = (
38            f"{self.system_message}\n\n"
39            f"{memory_context}\n"
40            f"用户: {user_message}\n"
41            f"助手:"
42        )
43
44        # 调用模型
45        from autogen_core.models import UserMessage
46        response = await self.model_client.create(
47            messages=[UserMessage(content=full_prompt, source="user")]
48        )
49
50        # 存入记忆
51        self.memory.append({"role": "user", "content": user_message})
52        self.memory.append({"role": "assistant", "content": response.content})
53
54        return Response(
55            chat_message=TextMessage(content=response.content, source=self.name)
56        )
57
58    async def on_reset(self, cancellation_token: CancellationToken) -> None:
59        self.memory.clear()
60
61
62async def custom_agent_demo():
63    model_client = GeminiModelClient(model="gemini-2.0-flash")
64
65    agent = MemoryEnabledAgent(
66        name="SmartCustomerService",
67        model_client=model_client,
68        system_message=(
69            "你是一个电商平台的智能客服。"
70            "记住用户提到过的所有信息（商品、订单号、偏好等）。"
71            "友好、专业地回答问题。用中文回复。"
72        ),
73    )
74
75    # 模拟多轮对话
76    questions = [
77        "我想买一台笔记本电脑，预算 6000 左右",
78        "有什么推荐的品牌吗？我之前用过联想",
79        "好的，那帮我查一下ThinkPad系列的",
80        "对了，刚才我说我的预算是多少来着？",  # 测试记忆
81    ]
82
83    for q in questions:
84        print(f"\n用户: {q}")
85        result = await agent.on_messages(
86            [TextMessage(content=q, source="user")],
87            CancellationToken(),
88        )
89        print(f"客服: {result.chat_message.content}")
90
91    await model_client.close()
92
93
94if __name__ == "__main__":
95    asyncio.run(custom_agent_demo())
96

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

模式	适用场景	Agent 数量	特点
RoundRobin	简单双人协作	2-3	严格轮流发言
Selector	复杂团队协作	3-8	LLM 智能选择发言者
Pipeline	流水线任务	3-5	按阶段依次执行
Nested	大型项目	8+	多层级嵌套团队

8. 运行效果示例

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

[Tester] 编写测试：

1@pytest.mark.asyncio
2async def test_scheduler_retry():
3    ...
4

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

9. 最佳实践

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

10. 总结

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

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

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

《AI Agent 智能体开发入门：AutoGen 多智能体协作实战教程》是转载文章，点击查看原文。