作者逆境不可逃技术永无止境希望我的内容可以帮助到你大家吼 ! 我是 逆境不可逃 今天给大家带来文章《【与我学 ClaudeCode】协作篇 之Agent Teams 持久化队友与文件邮箱》.Learn-Claude-Code 官方地址 :shareAI-lab/learn-claude-code: Bash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1上一篇文章【与我学 ClaudeCode】并发篇 之 Background Tasks 守护线程与异步通知队列-CSDN博客Agent Teams 是迭代的第 9 个版本s09核心解决单一 Agent 无法处理超大型任务的问题。它构建了一个「持久化队友 文件系统邮箱」的多智能体协作框架实现了 Leader-Worker 模式的团队协作让任务可以被拆解、分发、并行执行。学习路线s01 s02 s03 s04 s05 s06| s07s08 s09 s10 s11 s12一、问题根源为什么一次性 Subagent 撑不起真正的团队协作s04 的 Subagent 是一次性的创建、执行一个任务、返回结果、消亡。它存在三个致命缺陷无身份与记忆没有跨调用的持久身份每次任务都要从零开始重复学习成本极高无法跨任务协作完成任务 A 后无法被分配任务 B无法携带之前学到的知识和上下文无通信机制多个 Agent 之间无法直接沟通无法实现复杂的任务分配和结果同步真正的团队协作需要三样东西能跨多轮对话存活的持久 Agent明确的身份和生命周期管理Agent 之间的可靠通信通道二、三大核心设计决策Agent Teams 通过三个关键设计构建了一个零依赖、可调试、易扩展的多智能体协作系统。1. 持久化队友 vs 一次性 Subagent核心设计s04 的 Subagent 是临时的任务完成后即销毁s09 的队友是具有身份名称、角色和配置文件的持久化线程。队友可以完成任务 A然后被分配任务 B并携带之前学到的所有知识。持久化队友积累项目知识理解已建立的模式不需要为每个任务重新阅读相同的文件。替代方案的致命缺陷一次性 Subagent 实现更简单提供完美的上下文隔离无任务污染风险但重复学习成本极高每个新任务都从零开始带共享内存 / 知识库的 Subagent 是折中方案但会增加复杂度却无法获得持久身份和状态的全部好处。2. 团队配置持久化到.teams/{name}/config.json核心设计团队结构成员名称、角色、状态存储在 JSON 配置文件中而非任何 Agent 的内存中。任何 Agent 都可以通过读取配置文件发现队友无需发现服务或共享内存。如果 Agent 崩溃并重启它读取配置即可知道团队中还有谁。文件系统就是协调层配置文件人类可读便于手动添加 / 移除成员、调试团队配置问题。替代方案的致命缺陷内存中的团队注册表速度更快但进程重启后会丢失且需要中心进程维护服务发现如 DNS 或发现服务器在大规模场景下更健壮但对于本地多智能体系统来说是过度设计基于文件的配置是最简单的跨独立进程工作的方法。3. 队友获得工具子集组长获得全部工具核心设计团队组长获得ALL_TOOLS包括spawn、send、read_inbox等而队友获得TEAMMATE_TOOLS专注于任务执行的精简工具集。这强制了清晰的职责分离队友专注于做事编码、测试、研究组长专注于协调创建任务、分配工作、管理沟通。给队友协调工具会让他们创建自己的子团队或重新分配任务破坏组长维持连贯计划的能力。替代方案的致命缺陷给所有 Agent 相同的工具实现更简单、更平等但在实践中会导致协调混乱 —— 多个 Agent 试图互相管理造成冲突的任务分配。基于角色的静态工具过滤是可预测且易于推理的方案。三、系统整体架构与工作原理1. 核心架构图.team/ config.json - 团队名册 状态 inbox/ alice.jsonl - 追加写入、读取清空的收件箱 bob.jsonl lead.jsonl -------- send(alice,bob,...) -------- | alice | ----------------------------- | bob | | loop | bob.jsonl {json_line} | loop | -------- -------- ^ | | BUS.read_inbox(alice) | ---- alice.jsonl - read drain ---------2. 队友生命周期spawn - WORKING - IDLE - WORKING - ... - SHUTDOWNspawn组长调用spawn_teammate创建队友线程初始化状态为workingWORKING队友执行任务处理收件箱消息IDLE任务完成后状态变为idle等待新任务SHUTDOWN收到关闭请求后线程终止3. 关键组件与实现细节(1) MessageBus文件系统邮箱核心class MessageBus: def __init__(self, inbox_dir: Path): self.dir inbox_dir self.dir.mkdir(parentsTrue, exist_okTrue) def send(self, sender: str, to: str, content: str, msg_type: str message, extra: dict None) - str: 发送消息追加写入目标的 JSONL 收件箱 if msg_type not in VALID_MSG_TYPES: return fError: Invalid type {msg_type}. Valid: {VALID_MSG_TYPES} msg { type: msg_type, from: sender, content: content, timestamp: time.time(), } if extra: msg.update(extra) inbox_path self.dir / f{to}.jsonl with open(inbox_path, a) as f: f.write(json.dumps(msg) \n) return fSent {msg_type} to {to} def read_inbox(self, name: str) - list: 读取收件箱读取全部消息并清空文件drain inbox_path self.dir / f{name}.jsonl if not inbox_path.exists(): return [] messages [] for line in inbox_path.read_text().strip().splitlines(): if line: messages.append(json.loads(line)) inbox_path.write_text() # 读取后清空避免重复处理 return messages def broadcast(self, sender: str, content: str, teammates: list) - str: 广播消息发送给所有队友 count 0 for name in teammates: if name ! sender: self.send(sender, name, content, broadcast) count 1 return fBroadcast to {count} teammates(2) TeammateManager队友管理核心class TeammateManager: def __init__(self, team_dir: Path): self.dir team_dir self.dir.mkdir(exist_okTrue) self.config_path self.dir / config.json self.config self._load_config() self.threads {} # 存储队友线程 def _load_config(self) - dict: 加载团队配置文件 if self.config_path.exists(): return json.loads(self.config_path.read_text()) return {team_name: default, members: []} def spawn(self, name: str, role: str, prompt: str) - str: 创建/唤醒队友线程 member self._find_member(name) if member: if member[status] not in (idle, shutdown): return fError: {name} is currently {member[status]} member[status] working member[role] role else: # 新队友添加到配置文件 member {name: name, role: role, status: working} self.config[members].append(member) self._save_config() # 创建守护线程运行队友的 agent loop thread threading.Thread( targetself._teammate_loop, args(name, role, prompt), daemonTrue, ) self.threads[name] thread thread.start() return fSpawned {name} (role: {role}) def _teammate_loop(self, name: str, role: str, prompt: str): 队友的 agent loop每次 LLM 调用前检查收件箱 sys_prompt ( fYou are {name}, role: {role}, at {WORKDIR}. fUse send_message to communicate. Complete your task. ) messages [{role: user, content: prompt}] tools self._teammate_tools() # 队友工具子集 for _ in range(50): # 读取并清空收件箱 inbox BUS.read_inbox(name) for msg in inbox: messages.append({role: user, content: json.dumps(msg)}) try: response client.messages.create( modelMODEL, systemsys_prompt, messagesmessages, toolstools, max_tokens8000, ) except Exception: break messages.append({role: assistant, content: response.content}) if response.stop_reason ! tool_use: break # 处理工具调用 results [] for block in response.content: if block.type tool_use: output self._exec(name, block.name, block.input) results.append({ type: tool_result, tool_use_id: block.id, content: str(output), }) messages.append({role: user, content: results}) # 任务完成状态变为 idle member self._find_member(name) if member and member[status] ! shutdown: member[status] idle self._save_config() def _teammate_tools(self) - list: 队友的工具子集无 spawn 工具仅包含执行和通信工具 return [ {name: bash, ...}, {name: read_file, ...}, {name: write_file, ...}, {name: edit_file, ...}, {name: send_message, ...}, {name: read_inbox, ...}, ](3) 组长工具与主循环# 组长工具集包含 spawn、list_teammates 等协调工具 TOOL_HANDLERS { bash: lambda **kw: _run_bash(kw[command]), read_file: lambda **kw: _run_read(kw[path], kw.get(limit)), write_file: lambda **kw: _run_write(kw[path], kw[content]), edit_file: lambda **kw: _run_edit(kw[path], kw[old_text], kw[new_text]), spawn_teammate: lambda **kw: TEAM.spawn(kw[name], kw[role], kw[prompt]), list_teammates: lambda **kw: TEAM.list_all(), send_message: lambda **kw: BUS.send(lead, kw[to], kw[content], kw.get(msg_type, message)), read_inbox: lambda **kw: json.dumps(BUS.read_inbox(lead), indent2), broadcast: lambda **kw: BUS.broadcast(lead, kw[content], TEAM.member_names()), } def agent_loop(messages: list): while True: # 组长读取收件箱处理队友的消息 inbox BUS.read_inbox(lead) if inbox: messages.append({ role: user, content: finbox{json.dumps(inbox, indent2)}/inbox, }) # 调用 LLM response client.messages.create( modelMODEL, systemSYSTEM, messagesmessages, toolsTOOLS, max_tokens8000, ) messages.append({role: assistant, content: response.content}) if response.stop_reason ! tool_use: return # 处理工具调用包括 spawn、send_message 等 results [] for block in response.content: if block.type tool_use: handler TOOL_HANDLERS.get(block.name) output handler(**block.input) if handler else fUnknown tool: {block.name} results.append({type: tool_result, tool_use_id: block.id, content: str(output)}) messages.append({role: user, content: results})(4) 执行流程四、与 Background Taskss08的关键变更对比组件之前s08 Background Tasks之后s09 Agent TeamsAgent 数量单一 Agent组长 N 个队友线程模型后台命令线程每个队友一个完整的 agent loop 线程持久化无config.json团队配置 JSONL 收件箱生命周期一次性命令idle - working - idle可复用状态通信机制无send_messagebroadcast文件邮箱工具集6 个工具组长 9 个工具含 spawn / 协调工具队友 6 个工具仅执行 / 通信核心优化耗时命令后台执行多 Agent 团队协作、任务分发与并行执行五、核心优势与创新点持久化队友避免重复学习队友线程可复用跨任务携带知识无需每次重新理解项目文件系统即协调层团队配置和收件箱都存储在文件中无需中心化服务可调试、可重启角色工具分离职责清晰组长专注协调队友专注执行避免工具权限混乱和任务冲突JSONL 邮箱简单可靠追加写入 读取清空的设计线程安全、无锁通信实现简单且无竞态条件可扩展的消息类型支持message、broadcast、shutdown_request等多种消息类型为后续版本的团队协作扩展打下基础六、运行示例假设用户输入开发一个用户登录功能需要实现接口和编写单元测试组长 Agent 的典型执行流程组长分析任务决定创建两个队友coder负责实现登录接口tester负责编写单元测试组长调用spawn_teammate(coder, backend developer, 实现用户登录接口使用 bcrypt 加密密码)创建队友线程组长调用spawn_teammate(tester, QA engineer, 为登录接口编写单元测试覆盖正常登录和密码错误场景)创建队友线程组长通过send_message给两个队友发送任务详情队友的收件箱收到消息队友每次 LLM 调用前读取收件箱获取任务指令并开始工作coder完成接口开发后通过send_message向组长发送「接口已完成代码路径auth.py」tester完成测试编写后向组长发送「测试已完成测试用例通过」组长读取收件箱整合两个队友的结果向用户报告任务完成情况七、可扩展方向任务依赖与分配优化为队友添加blockedBy依赖字段支持任务按依赖顺序分配避免无效工作队友状态管理扩展idle/working/shutdown状态增加paused状态支持任务暂停与恢复消息优先级为消息添加优先级字段队友优先处理高优先级任务指令团队权限控制为队友添加权限配置限制对特定文件 / 目录的读写实现更细粒度的安全控制队友自动扩缩容根据任务负载自动创建 / 销毁队友避免闲置资源浪费
