第 15 章 / 共 21 章
Agent 团队：常驻队友与文件信箱
上一章留下的问题：第6章的子代理是一次性的——接一个任务、跑完、返回一句结论、消失。它们之间没法通信，也没法在等一件事的同时干另一件事。
现在设想一个真实规模的重构：认证模块、数据库层、API 路由、测试套件，四块要同时动，而且改动之间会互相影响（改了认证的接口，路由那边要跟着改）。一个 Agent 装不下这么多上下文；四个互不通信的子代理会各改各的，最后拼不起来。
你需要的是常驻的队友，以及它们之间的通信通道。
15.1 用文件当信箱
通信机制选得越简单越好。这里用文件：每个 Agent 一个 .jsonl 收件箱，发消息就是往对方的文件里追加一行 JSON，读消息就是读完整个文件然后删掉。
import json, time
from pathlib import Path
MAILBOX_DIR = Path(".mailbox"); MAILBOX_DIR.mkdir(exist_ok=True)
class MessageBus:
def send(self, from_agent: str, to_agent: str, msg_type: str,
content: str = "", metadata: dict | None = None) -> None:
msg = {"from": from_agent, "to": to_agent, "type": msg_type,
"content": content, "metadata": metadata or {}, "ts": time.time()}
inbox = MAILBOX_DIR / f"{to_agent}.jsonl"
with inbox.open("a") as f:
f.write(json.dumps(msg, ensure_ascii=False) + "\n")
def read_inbox(self, agent: str) -> list[dict]:
inbox = MAILBOX_DIR / f"{agent}.jsonl"
if not inbox.exists():
return []
msgs = [json.loads(line) for line in inbox.read_text().splitlines() if line.strip()]
inbox.unlink() # 读完即清空
return msgs
BUS = MessageBus()选文件而不是内存队列，有一个教学之外的实际好处：**你可以直接 cat .mailbox/alice.jsonl 看到它们在说什么。**多智能体系统最难的部分是调试，而一个能用 cat 看的通信层，比一个藏在进程内存里的队列友好太多。
15.2 派生常驻队友
TEAMMATE_TOOLS = ["bash", "read_file", "write_file", "edit_file", "glob",
"send_message", "claim_task", "complete_task", "list_tasks"]
def spawn_teammate(name: str, role: str) -> str:
def teammate_loop():
messages = [{"role": "user", "content": f"你的角色：{role}。开始工作。"}]
system = (f"你是团队成员 {name}，职责是：{role}。"
f"需要与他人协作时用 send_message 联系 lead 或其他成员。"
f"完成阶段性工作后，向 lead 汇报。")
for _ in range(10): # 教学版的轮次上限
inbox = BUS.read_inbox(name)
if inbox:
messages.append({"role": "user",
"content": f"<inbox>{json.dumps(inbox, ensure_ascii=False)}</inbox>"})
response = client.messages.create(
model=MODEL, system=system, messages=messages,
tools=[t for t in TOOLS if t["name"] in TEAMMATE_TOOLS], max_tokens=8000)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
break
# …与主循环相同的工具执行（权限钩子照常触发）…
BUS.send(name, "lead", "report", extract_text(messages[-1]["content"]))
threading.Thread(target=teammate_loop, daemon=True).start()
return f"队友 {name} 已启动，职责：{role}"队友的循环和主循环是同一个形状——调模型、看 stop_reason、执行工具、回灌。只多了两件事：开头读收件箱，结尾向 lead 汇报。
15.3 Lead 侧的收件箱注入
Lead 的循环里，每轮末尾检查一次收件箱：
inbox = BUS.read_inbox("lead")
if inbox:
messages.append({"role": "user",
"content": f"<inbox>{json.dumps(inbox, ensure_ascii=False)}</inbox>"})这一步是让并行工作”被看见”的关键。队友在自己的线程里跑，Lead 的模型看不到它们的过程；只有通过收件箱注入，队友的产出才进入 Lead 的视野，Lead 才能据此调整安排。
注意这个结构和第13章的后台通知是同构的：异步发生的事情，都要通过某种”注入消息”的机制才能进入模型的世界。
15.4 走一遍：一次四人重构
你：把这个项目的认证模块从 session 改成 JWT，涉及认证、路由、测试三块。
Lead：
create_task("重写认证核心逻辑") → task_A
create_task("更新受保护路由", blockedBy=["task_A"]) → task_B
create_task("更新集成测试", blockedBy=["task_A"]) → task_C
spawn_teammate("alice", "负责认证核心逻辑")
spawn_teammate("bob", "负责路由层")
spawn_teammate("carol", "负责测试")| 时刻 | alice | bob | carol | lead |
|---|---|---|---|---|
| T0 | 认领 task_A，开始改 | task_B 被阻塞，等待 | task_C 被阻塞，等待 | 等待汇报 |
| T1 | 完成 A，send_message 告知新接口签名 | 收到消息，认领 B | 收到消息，认领 C | 收到 alice 的汇报 |
| T2 | 空闲 | 改路由 | 改测试 | 观察进度 |
| T3 | — | 汇报完成 | 汇报完成 | 汇总结果给你 |
T1 那一步是团队协作真正的价值所在：alice 主动把”新接口长什么样”广播出去，bob 和 carol 不需要各自去读代码猜。通信省下的返工，才是多智能体的收益来源——不是”人多力量大”，而是”信息传递比重新发现便宜”。
15.5 工程现实
教学版有几个明显的简化：
read_inbox有竞态：读文件和删文件不是原子操作，两次读之间到达的消息会丢。生产实现要加文件锁。- 固定 10 轮上限：真实队友应该在做完手头工作后进入空闲等待，而不是转够 10 圈就退出。这正是第17章要解决的。
- 权限上报（permission bubbling）：队友遇到需要人工确认的操作时，应该把请求冒泡给 Lead 或用户，而不是自己在后台线程里
input()（那会挂死）。 - 没有优雅关闭：现在杀掉一个队友线程，它可能正写到文件一半。这是第16章的题目。
- 进程模型不同：一些成熟实现用独立进程/终端面板承载队友，而不是线程——隔离性更好，也更容易观察。
15.6 子代理与队友的分工
| 子代理（第6章） | 队友（本章） | |
|---|---|---|
| 生命周期 | 一次性，跑完即销毁 | 常驻，可持续接活 |
| 通信 | 无，只返回一句结论 | 通过信箱双向通信 |
| 上下文 | 全新，用完丢弃 | 持续累积（也因此需要压缩） |
| 适合 | 结论明确、过程冗长的一次性调研 | 长期负责一块领域、需要协调的工作 |
不要用队友做子代理的活。“帮我在代码库里找一下所有用了旧接口的地方”——这是一次性调研，开一个子代理就够了，派一个常驻队友是浪费。