Agent团队系统多智能体协同工作开发进阶指南

Agent 开发进阶（十五）：Agent 团队系统，让多个智能体协同工作

先问一个问题

当一项任务复杂到单凭个人无法完成时，你是否也曾设想过这样的场景——

• 一名专职的编码 Agent 负责编写代码
• 一名专职的测试 Agent 负责运行检测
• 一名专职的文档 Agent 负责整理资料
• 而且这些 Agent 并非用完即毁，而是长期在线、随时待命

如果你点头认同了，那就说明你真正需要的，正是一套 Agent 团队系统。

单 Agent 的「能力边界」问题

走到这个阶段，你的 Agent 已经集齐了各种核心能力：核心循环、工具分发、规划机制、子智能体、技能加载、上下文压缩、权限系统、Hook 系统、Memory 系统、提示词组装、错误恢复、任务系统、后台任务、定时调度……单看这份清单确实挺唬人。

然而，当真正面对复杂任务时，单兵作战的短板就会暴露无遗：

• 能力有限——一个 Agent 无法精通所有领域
• 上下文冲突——不同类型任务挤在一起互相污染
• 并行度低——只能串联处理，一路走到黑
• 可维护性差——代码和逻辑堆积如山，难以维护

之前介绍的 Subagent 虽然能拆解小任务，但它的生命周期非常短暂：创建 → 执行 → 返回摘要 → 消失。这种模式适用于一次性小委派，但如果要实现长期默契协作，显然不够用。

因此，在这一阶段，我们需要一个更成熟的方案：Agent 团队系统——一批有身份、能长期存在、可以反复协作的队友。

Agent 团队系统的核心设计：名册、邮箱、独立循环

先来看一下核心流程示意：

lead|+-- spawn alice (coder)+-- spawn bob (tester)|+-- send message --> alice inbox+-- send message --> bob inboxalice|+-- 自己的 messages+-- 自己的 inbox+-- 自己的 agent loopbob|+-- 自己的 messages+-- 自己的 inbox+-- 自己的 agent loop

核心精髓只有三个组件：

1. 名册：记录团队成员的身份与状态
2. 邮箱：队友之间的通信渠道
3. 独立循环：每个队友都拥有自己的 Agent Loop

几个必须搞懂的概念

队友（Teammate）

这里的 teammate 指的不是临时打工仔，而是拥有独立身份、独立记忆、独立生命周期的长期协作伙伴。

名册（Roster）

名册就是团队成员的花名册。它回答的问题很直白：队伍里现在有谁？每个人是什么角色？每个人当前是空闲、工作中，还是已关闭？

邮箱（Inbox）

邮箱是每个队友的专属收件箱。别人把消息塞进去，队友在下一轮工作时会先查看收件箱，再开始干活。

消息信封（Message Envelope）

envelope 这个词字面意思是“信封”。在程序里，它表示一条消息的包装——包含了发送者、接收者、内容、时间戳等元信息，就像寄信一样。

最小实现

1. Team Manager

import osimport jsonimport threadingimport timeimport uuidfrom pathlib import Pathclass TeamManager:"""团队管理器"""def __init__(self, team_dir=".team"):self.team_dir = Path(team_dir)self.team_dir.mkdir(exist_ok=True)# 创建必要的目录self.config_dir = self.team_dirself.inbox_dir = self.team_dir / "inbox"self.inbox_dir.mkdir(exist_ok=True)# 加载配置self.config_path = self.config_dir / "config.json"self.config = self._load_config()# 存储队友线程self.teammate_threads = {}def _load_config(self):"""加载配置"""if self.config_path.exists():try:return json.loads(self.config_path.read_text(encoding="utf-8"))except Exception as e:print(f"加载配置失败: {e}")return {"team_name": "default", "members": []}def _sa ve_config(self):"""保存配置"""self.config_path.write_text(json.dumps(self.config, indent=2, ensure_ascii=False),encoding="utf-8")def spawn(self, name, role, prompt):"""创建队友"""# 检查是否已存在for member in self.config["members"]:if member["name"] == name:return f"队友 {name} 已存在"# 添加到配置member = {"name": name,"role": role,"status": "working"}self.config["members"].append(member)self._sa ve_config()# 启动队友线程thread = threading.Thread(target=self._teammate_loop,args=(name, role, prompt),daemon=True,)thread.start()self.teammate_threads[name] = threadreturn f"队友 {name} ({role}) 已创建"def list(self):"""列出所有队友"""if not self.config["members"]:return "团队暂无成员"lines = ["# 团队成员n"]for member in self.config["members"]:lines.append(f"- **{member['name']}** ({member['role']}) [{member['status']}]")return "n".join(lines)def send(self, sender, to, content):"""发送消息"""# 检查接收者是否存在recipient_exists = Falsefor member in self.config["members"]:if member["name"] == to:recipient_exists = Truebreakif not recipient_exists:return f"队友 {to} 不存在"# 发送消息到邮箱inbox_file = self.inbox_dir / f"{to}.jsonl"message = {"type": "message","from": sender,"content": content,"timestamp": time.time(),}with open(inbox_file, "a", encoding="utf-8") as f:f.write(json.dumps(message, ensure_ascii=False) + "n")return f"消息已发送给 {to}"def _read_inbox(self, name):"""读取邮箱"""inbox_file = self.inbox_dir / f"{name}.jsonl"if not inbox_file.exists():return []messages = []try:with open(inbox_file, "r", encoding="utf-8") as f:for line in f:line = line.strip()if line:messages.append(json.loads(line))# 清空邮箱inbox_file.write_text("", encoding="utf-8")except Exception as e:print(f"读取邮箱失败 {name}: {e}")return messagesdef _teammate_loop(self, name, role, prompt):"""队友循环"""print(f"队友 {name} ({role}) 已启动")# 初始化消息messages = [{"role": "system","content": f"你是 {name}，一个 {role}。请专注于你的职责，完成任务后等待新的指示。"}, {"role": "user","content": prompt}]while True:# 读取邮箱inbox = self._read_inbox(name)if inbox:print(f"队友 {name} 收到 {len(inbox)} 条消息")for msg in inbox:messages.append({"role": "user","content": f"来自 {msg['from']} 的消息: {msg['content']}"})# 这里应该调用模型，但为了简化，我们模拟一下if messages:print(f"队友 {name} 正在处理任务...")# 模拟处理时间time.sleep(2)# 模拟回复response = f"{name} 已收到并处理了消息"messages.append({"role": "assistant", "content": response})# 等待一段时间再检查邮箱time.sleep(5)def shutdown(self):"""关闭团队"""# 这里可以添加清理逻辑pass

2. 团队工具

def create_team_tools(team_manager):"""创建团队相关的工具"""def team_spawn(name, role, prompt):"""创建队友"""return team_manager.spawn(name, role, prompt)def team_list():"""列出所有队友"""return team_manager.list()def team_send(to, content):"""发送消息"""return team_manager.send("lead", to, content)return {"team_spawn": team_spawn,"team_list": team_list,"team_send": team_send,}

3. 集成到 Agent Loop

def agent_loop_with_team(state):"""带团队系统的 Agent Loop"""# 初始化团队管理器team_manager = TeamManager()# 创建团队工具team_tools = create_team_tools(team_manager)state["tools"] = state.get("tools", []) + [{"name": "team_spawn","description": "创建队友","parameters": {"name": {"type": "string", "description": "队友名字"},"role": {"type": "string", "description": "队友角色"},"prompt": {"type": "string", "description": "队友提示词"}}},{"name": "team_list","description": "列出所有队友","parameters": {}},{"name": "team_send","description": "发送消息给队友","parameters": {"to": {"type": "string", "description": "接收者名字"},"content": {"type": "string", "description": "消息内容"}}}]# 主循环while True:# 调用模型response = call_model(state["messages"])if response.stop_reason != "tool_use":return response.contentresults = []for block in response.content:if hasattr(block, "type") and block.type == "tool_use":tool_name = block.nametool_input = block.input# 执行团队工具if tool_name in team_tools:output = team_tools[tool_name](**tool_input)else:# 执行其他工具output = run_tool(tool_name, tool_input)results.append({"type": "tool_result","tool_use_id": block.id,"content": output})if results:state["messages"].append({"role": "user", "content": results})

核心功能说明

1. 创建队友

举个例子，创建编码队友：

team_manager.spawn("alice", "coder", "你是 Alice，一名专业的 Python 开发者。请负责编写高质量的代码。")

再来一个测试队友：

team_manager.spawn("bob", "tester", "你是 Bob，一名专业的测试工程师。请负责测试代码的质量和功能。")

2. 管理团队

列出所有队友：

team_manager.list()

发消息给队友：

team_manager.send("lead", "alice", "请编写一个 JSON 解析器")

3. 队友循环

每个队友都运行着自己的独立循环：先检查邮箱是否有新消息，有就处理，没有就继续等待。这样一来，队友可以长期在线，随时接手新任务。

4. 持久化

团队配置会被保存到 .team/config.json 文件，内容大致如下：

{"team_name": "default","members": [{"name": "alice","role": "coder","status": "working"},{"name": "bob","role": "tester","status": "working"}]}

这样一来，即使程序重启，团队成员的信息也不会丢失，系统能够从容恢复。

Teammate vs Subagent vs Runtime Task 的边界

选择建议非常清晰：

• 长期需要的角色 → 使用 Teammate
• 一次性的小任务 → 使用 Subagent
• 耗时较长的后台命令 → 使用 Runtime Task

新手最容易犯的4个错

1. 把队友当成「名字不同的 subagent」

# ❌ 错误# 生命周期还是执行完就销毁def spawn_subagent(name, task):# 执行任务result = execute_task(task)# 返回结果后销毁return result# ✅ 正确# 队友长期存在def spawn_teammate(name, role, prompt):# 创建队友member = {"name": name, "role": role, "status": "working"}# 启动独立循环thread = threading.Thread(target=teammate_loop, args=(name, role, prompt))thread.start()# 队友继续存在return f"队友 {name} 已创建"

2. 队友之间共用同一份 messages

# ❌ 错误# 所有队友共用一个消息列表class TeamManager:def __init__(self):self.messages = []def spawn(self, name):# 所有队友都用 self.messagespass# ✅ 正确# 每个队友有自己的消息列表def teammate_loop(name, role, prompt):# 每个队友有自己的 messagesmessages = [{"role": "system","content": f"你是 {name}，一个 {role}。"}]# 处理自己的消息while True:pass

3. 没有持久名册

# ❌ 错误# 名册只存在内存中class TeamManager:def __init__(self):self.members = []# ✅ 正确# 名册持久化到磁盘class TeamManager:def __init__(self, team_dir=".team"):self.team_dir = Path(team_dir)self.config_path = self.team_dir / "config.json"self.config = self._load_config()def _load_config(self):if self.config_path.exists():return json.loads(self.config_path.read_text(encoding="utf-8"))return {"team_name": "default", "members": []}

4. 没有邮箱，靠共享变量直接喊话

# ❌ 错误# 直接使用共享变量通信shared_messages = {}def send_message(to, content):if to not in shared_messages:shared_messages[to] = []shared_messages[to].append(content)def get_messages(to):return shared_messages.get(to, [])# ✅ 正确# 使用邮箱系统通信def send(self, sender, to, content):inbox_file = self.inbox_dir / f"{to}.jsonl"message = {"type": "message","from": sender,"content": content,"timestamp": time.time(),}with open(inbox_file, "a", encoding="utf-8") as f:f.write(json.dumps(message, ensure_ascii=False) + "n")

为什么这很重要

一个真正强大的系统，背后往往是多个专业角色在默契配合。Agent 团队系统带来的实际好处非常明显：

1. 分工协作——各司其职，每个人只专注于自己的领域
2. 并行处理——多个任务同时推进，效率成倍提升
3. 上下文隔离——不同任务的对话不会互相干扰
4. 长期存在——队友在线就绪，无需每次都重新创建
5. 可扩展——需要新角色时，随时添加成员

推荐的实现步骤

1. 先实现 TeamManager 类——管理团队成员和持久化配置
2. 实现 队友创建和管理——支持 spawn、list、shutdown
3. 实现 邮箱系统——每个队友一个文件收件箱
4. 实现 队友独立循环——持续监听邮箱并处理任务
5. 创建 团队工具——将团队操作封装成模型可调用的工具
6. 集成到 Agent Loop——让主 Agent 能动态创建和管理团队

Agent 团队系统与后续章节的关系

• s15 Agent 团队：解决团队成员如何长期存在、互相发消息的问题
• s16 团队协议：解决团队成员之间如何进行结构化协作的问题
• s17 自主智能体：解决智能体如何自主工作、主动认领任务的问题

可以说，Agent 团队系统是构建复杂智能体系统的重要基础。

下一章预告

有了 Agent 团队系统，你的多个 Agent 已经能够长期共存、互相写信了。下一章我们将讨论团队协议系统，让成员之间的协作更加结构化、可追踪。一句话总结这一章的核心：Subagent 是一次性外包助手，Teammate 是长期在线队友。