【从零手写 ClaudeCode：learn-claude-code 项目实战笔记】（9）Agent Teams (智能体团队)

第九章 Agent Teams (智能体团队)

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > [ s09 ] s10 > s11 > s12

"本专栏基于开源项目learn-claude-code的官方文档。原文档非常硬核，为了方便像我一样的新手小白理解，我对文档进行了逐行精读，并加入了很多中文注释、大白话解释和踩坑记录。希望这套'咀嚼版'教程能帮你推开 AI Agent 开发的大门。"

项目地址：shareAI-lab/learn-claude-code: Bash is all you need - A nano Claude Code–like agent, built from 0 to 1

"任务太大一个人干不完, 要能分给队友"-- 持久化队友 + JSONL 邮箱。

一、问题：子智能体是一次性的、后台任务是无脑的

s04 的问题：子智能体是"一次性"的

子智能体的生命周期是：

生成 → 干活 → 返回摘要 → 被销毁

它没有身份、没有名字、不能跨轮对话存活、不能互相通信。父智能体说"去读五个文件总结一下"，子智能体做完就没了，下次需要再重新生成一个全新的。

对应的代码在s04里可以看到：

def run_subagent(prompt: str) -> str: sub_messages = [{"role": "user", "content": prompt}] # 全新的空上下文 for _ in range(30): response = client.messages.create(...) if response.stop_reason != "tool_use": break # execute tools... return "".join(b.text for b in response.content...) # 返回摘要，函数结束 # 函数返回后，sub_messages 被丢弃，这个子智能体就"死了"

没有client.messages.create()的循环调用，就没有"脑子"。s08 的后台线程就更直接了：

s08 的问题：后台线程是"哑执行器"

def _execute(self, task_id, command): r = subprocess.run(command, shell=True, ...) # 直接跑 shell 命令 output = (r.stdout + r.stderr).strip()[:50000] self._notification_queue.append({"task_id": task_id, "result": output[:500]}) # 函数结束，没有任何 LLM 调用

整个函数里没有client.messages.create()，所以它不会思考、不会做决策、不会根据输出调整下一步。命令是什么就跑什么，跑完就完了。

到底缺什么？

s09 要做的就是：既有 s04 的"脑子"（LLM 循环），又有 s08 的"持久性"（线程不销毁），再加上智能体之间的通信能力。

二、解决方案

Teammate lifecycle: spawn -> WORKING -> IDLE -> WORKING -> ... -> SHUTDOWN Communication: .team/ config.json <- 团队名册 + 状态 inbox/ alice.jsonl <- append-only, drain-on-read bob.jsonl lead.jsonl +--------+ send("alice","bob","...") +--------+ | alice | -----------------------------> | bob | | loop | bob.jsonl << {json_line} | loop | +--------+ +--------+ ^ | | BUS.read_inbox("alice") | +---- alice.jsonl -> read + drain ---------+

三个核心组件：

1. TeammateManager— 管理队友的创建、生命周期、线程
2. MessageBus— 基于 JSONL 文件的消息收发系统
3. _teammate_loop— 每个队友的"大脑循环"（包含 LLM 调用）

三、工作原理

3.1 TeammateManager：团队管理器

TeammateManager通过config.json维护团队名册，记录每个队友的 name、role、status。

class TeammateManager: def __init__(self, team_dir: Path): self.dir = team_dir self.dir.mkdir(exist_ok=True) 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 _save_config(self): self.config_path.write_text(json.dumps(self.config, indent=2)) # 持久化到磁盘

config.json长这样：

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

为什么用文件而不是内存？因为配置文件持久化到磁盘，进程重启后团队名册还在。

3.2 spawn()：创建队友

spawn()做三件事：注册到 config → 标记为 working → 启动线程。

def spawn(self, name: str, role: str, prompt: str) -> str: member = self._find_member(name) if member: # 已存在的队友，只有 idle（空闲）/shutdown（关闭） 状态才能重新 spawn if member["status"] not in ("idle", "shutdown"): return f"Error: '{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() # 写入 config.json # 关键：在新线程中启动 teammate_loop thread = threading.Thread( target=self._teammate_loop, args=(name, role, prompt), daemon=True, # 守护线程，主进程退出时自动结束 ) self.threads[name] = thread thread.start() return f"Spawned '{name}' (role: {role})"

对比 s04：s04 的run_subagent()是一个函数调用，阻塞等待完成。s09 的spawn()启动线程后立即返回，队友在后台独立运行。

3.3 MessageBus：消息收件箱

所有数据都存在磁盘上的 JSONL 文件中：

.team/inbox/ alice.jsonl <- 每行一条 JSON 消息 bob.jsonl lead.jsonl

class MessageBus: def __init__(self, inbox_dir: Path): self.dir = inbox_dir self.dir.mkdir(parents=True, exist_ok=True) def send(self, sender: str, to: str, content: str, msg_type: str = "message", extra: dict = None) -> str: if msg_type not in VALID_MSG_TYPES: return f"Error: Invalid type '{msg_type}'. Valid: {VALID_MSG_TYPES}" msg = { "type": msg_type, "from": sender, "content": content, "timestamp": time.time(), } if extra: msg.update(extra) # 往收件人的 .jsonl 文件末尾追加一行 inbox_path = self.dir / f"{to}.jsonl" with open(inbox_path, "a") as f: # "a" = append 模式 f.write(json.dumps(msg) + "\n") return f"Sent {msg_type} to {to}" def read_inbox(self, name: str) -> list: 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("") # drain：读完就清空 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 f"Broadcast to {count} teammates"

为什么选 JSONL 而不是 JSON 数组？

• JSONL 的append操作文件系统级别保证原子性，不需要加锁
• 无需先读出整个数组再写回，直接追加一行即可
• 每个队友有独立文件，互相之间不会冲突

3.4 _teammate_loop：队友的"大脑"（这是核心！）

这是 s09 和 s04/s08 最关键的区别。每个队友跑的是完整的 agent loop，里面有 LLM 调用。

def _teammate_loop(self, name: str, role: str, prompt: str): # 设置系统提示，告诉 LLM "你是谁、你是什么角色" sys_prompt = ( f"You are '{name}', role: {role}, at {WORKDIR}. " f"Use send_message to communicate. Complete your task." ) messages = [{"role": "user", "content": prompt}] tools = self._teammate_tools() # 这个队友能用的所有工具 for _ in range(50): # 最多 50 轮，防止死循环 # 第一步：检查收件箱，看看有没有别人发来的消息 inbox = BUS.read_inbox(name) for msg in inbox: messages.append({"role": "user", "content": json.dumps(msg)}) # 第二步：调用 LLM —— 这就是"脑子"所在 try: response = client.messages.create( model=MODEL, system=sys_prompt, messages=messages, tools=tools, max_tokens=8000, ) except Exception: break messages.append({"role": "assistant", "content": response.content}) # LLM 没有调用工具 = 任务完成，退出循环 if response.stop_reason != "tool_use": break # 第三步：执行 LLM 请求的工具调用 results = [] for block in response.content: if block.type == "tool_use": output = self._exec(name, block.name, block.input) print(f" [{name}] {block.name}: {str(output)[:120]}") results.append({ "type": "tool_result", "tool_use_id": block.id, "content": str(output), }) # 把工具执行结果喂回给 LLM，让它决定下一步 messages.append({"role": "user", "content": results}) # 循环结束，标记为 idle member = self._find_member(name) if member and member["status"] != "shutdown": member["status"] = "idle" self._save_config()

这个循环和 s04 的run_subagent()几乎一样，区别在于：

• s04 的子智能体执行完返回摘要就销毁了
• s09 的队友执行完后只是变成idle状态，线程结束，但config.json里保留了它的记录，下次可以spawn()唤醒它

3.5 队友能用哪些工具？

def _teammate_tools(self) -> list: return [ # 基础工具（继承自 s02） {"name": "bash", ...}, # 跑 shell 命令 {"name": "read_file", ...}, # 读文件 {"name": "write_file", ...}, # 写文件 {"name": "edit_file", ...}, # 编辑文件 # s09 新增的通信工具 {"name": "send_message", ...}, # 给其他队友发消息 {"name": "read_inbox", ...}, # 读取自己的收件箱 ]

注意：队友没有spawn_teammate工具，只有 leader 才能创建新队友。这避免了无限递归生成。

3.6 领导的 9 个工具

领导（lead）比队友多了 3 个管理工具：队友的工具都是定义在定义在 TeammateManager 类里，而领导的工具是在TOOL_HANDLERS中。

TOOL_HANDLERS = { # 基础工具（4 个） "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"]), # 团队管理工具（3 个） "spawn_teammate": lambda **kw: TEAM.spawn(kw["name"], kw["role"], kw["prompt"]), "list_teammates": lambda **kw: TEAM.list_all(), "broadcast": lambda **kw: BUS.broadcast("lead", kw["content"], TEAM.member_names()), # 通信工具（2 个） "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"), indent=2), }

3.7 数据流完整示例

假设用户说 "Spawn alice and bob, have alice write a hello function, bob test it"：

用户 prompt │ ▼ Lead agent loop │ ├─ 调用 spawn_teammate("alice", "coder", "write a hello function") │ → 新线程启动 alice 的 teammate_loop │ → config.json: alice.status = "working" │ ├─ 调用 spawn_teammate("bob", "tester", "test the hello function") │ → 新线程启动 bob 的 teammate_loop │ → config.json: bob.status = "working" │ ▼ alice 线程（独立运行） bob 线程（独立运行） │ │ ├─ LLM 决定: 写 hello.py ├─ LLM 决定: 等 alice 完成 ├─ 调用 write_file ├─ 调用 read_inbox（空的） ├─ 调用 send_message("bob", ├─ LLM 决定: 等消息 │ "hello function written") │ │ → bob.jsonl 追加一行 JSON │ ├─ LLM 决定: 完成 │ ├─ status = "idle" │ ├─ read_inbox 读到消息 ├─ LLM 决定: 读 hello.py 并写测试 ├─ 调用 read_file("hello.py") ├─ 调用 write_file("test_hello.py") ├─ 调用 bash("pytest test_hello.py") ├─ LLM 决定: 完成 ├─ status = "idle"

四、相对 s08 的变更

五、关键代码对比

s08 后台线程：没有 LLM，不能思考

def _execute(self, task_id, command): r = subprocess.run(command, shell=True, ...) # ↑ 只有 shell 调用，没有 client.messages.create() # ↓ 执行完就完了，不会根据结果做决策 output = (r.stdout + r.stderr).strip() self._notification_queue.append({"task_id": task_id, "result": output})

s09 队友线程：有 LLM，能思考

def _teammate_loop(self, name, role, prompt): messages = [{"role": "user", "content": prompt}] for _ in range(50): inbox = BUS.read_inbox(name) # 检查收件箱 for msg in inbox: messages.append({"role": "user", "content": json.dumps(msg)}) response = client.messages.create(...) # ← 这就是"脑子" # LLM 看到上下文后自主决定： # - 要不要跑 shell？ # - 跑完结果不对要不要换个方式？ # - 要不要给别的队友发消息？ # - 要不要读个文件看看？ if response.stop_reason != "tool_use": break # 执行工具，把结果喂回 LLM，继续循环... self._find_member(name)["status"] = "idle"

一句话总结：s08 是"执行一段命令"，s09 是"让一个有脑子的 agent 去完成一个任务"。

六、5 种消息类型

代码中定义了 5 种消息类型（s09 中实现了 2 种，其余为 s10 预留）：

VALID_MSG_TYPES = { "message", # 普通文本消息（s09 实现） "broadcast", # 群发给所有队友（s09 实现） "shutdown_request", # 请求优雅关闭（s10 预留） "shutdown_response", # 批准/拒绝关闭请求（s10 预留） "plan_approval_response", # 批准/拒绝计划（s10 预留） }

七、试一试

cd learn-claude-code python agents/s09_agent_teams.py

试试这些 prompt（英文 prompt 对 LLM 效果更好，也可以用中文）：

1. Spawn alice (coder) and bob (tester). Have alice send bob a message.
2. Broadcast "status update: phase 1 complete" to all teammates
3. Check the lead inbox for any messages
4. 输入/team查看团队名册和状态
5. 输入/inbox手动检查领导的收件箱