09 — 智能体团队

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

Harness 层：团队邮箱 — 多个模型，通过文件协调。

课程路径：[[07-任务系统|s07 任务系统]] → [[08-后台任务|s08 后台任务]] → ⭐ s09 智能体团队

---

问题：一次性的子智能体

回顾 [[04-子智能体|s04 子智能体]]：父派一个孩子去干活，孩子做完回来报个摘要，然后消失。没有名字、没有记忆、下一次对话又是从零开始。对于"查一下这个 API"一类的任务这完全够用，但当我们需要真正的团队协作时，它的局限就暴露了：

局限	表现
没有身份	子智能体没有名字和角色，无法被其他人引用或找到
没有状态	做完即焚，无法保持 idle 等待下一项任务
没有通信	无法互相发消息，所有交互都经过父智能体中转
没有生命周期	只有"创建→运行→消亡"，没有 idle/working/idle 的循环

[[08-后台任务|s08 后台任务]] 往前走了一步：让智能体能在后台跑 shell 命令。但它跑的是脚本，不是LLM 循环——做不了需要模型实时推理的决策。

真正的团队协作需要三样东西：

1. 持久智能体 — 能跨多轮对话存活的队友，而非一次性工具
2. 身份和生命周期 — 每个队友有名字、角色、状态（idle/working）
3. 通信通道 — 队友之间能发消息，不经过领导者中转

解决方案：Team Directory + MessageBus

架构很简单：一个 .team/ 目录 + 一个 append-only 的邮件系统。

Teammate lifecycle:
  spawn -> WORKING -> IDLE -> WORKING -> ... -> SHUTDOWN

Communication:
  .team/
    config.json           <- 团队名册 + 状态
    inbox/
      alice.jsonl         <- append-only，读取即清空
      bob.jsonl
      lead.jsonl

每个队友是一个独立的 agent loop 线程，共享底层的工具集和文件系统。它们通过 JSONL 文件通信，不共享 messages[]——这是 [[04-子智能体|s04 上下文隔离]] 原则的延续，只不过现在隔离的主体从"子任务"变成了"队友"。

TeammateManager：名册管理

Manager 通过 config.json 维护队友列表，spawn() 创建新队友并启动线程：

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 spawn(self, name: str, role: str, prompt: str) -> str:
        member = {"name": name, "role": role, "status": "working"}
        self.config["members"].append(member)
        self._save_config()
        thread = threading.Thread(
            target=self._teammate_loop,
            args=(name, role, prompt), daemon=True)
        thread.start()
        return f"Spawned teammate '{name}' (role: {role})"

关键设计：daemon=True 意味着主线程退出时所有队友线程自动终止，不会留下孤儿进程。

MessageBus：文件即网络

每个队友有一个 JSONL 收件箱文件。send() 追加一行，read_inbox() 读取全部并清空：

class MessageBus:
    def send(self, sender, to, content, msg_type="message", extra=None):
        msg = {"type": msg_type, "from": sender,
               "content": content, "timestamp": time.time()}
        if extra:
            msg.update(extra)
        with open(self.dir / f"{to}.jsonl", "a") as f:
            f.write(json.dumps(msg) + "\n")

    def read_inbox(self, name):
        path = self.dir / f"{name}.jsonl"
        if not path.exists(): return "[]"
        msgs = [json.loads(l) for l in
                path.read_text().strip().splitlines() if l]
        path.write_text("")  # drain
        return json.dumps(msgs, indent=2)

几点设计考量：

• Append-only 写入：send() 用 "a" 模式追加，天然支持并发写入（POSIX 保证 append 不会互相覆盖）
• Drain-on-read：读取后清空文件，避免同一消息被多次消费。这是类似 Kafka 的"消费即删除"模型，简单可靠
• JSONL 格式：每行一个独立 JSON 对象，支持简单解析和逐行处理，比 XML 或整块 JSON 更适合流式场景

这与 [[07-任务系统|s07 的持久化任务图]] 共享同样的哲学：文件即数据层。任务图是"待办清单"，MessageBus 是"即时通讯"——一个管规划，一个管执行。

队友循环：带收件箱的 Agent Loop

每个队友跑自己的 agent loop，与 [[01-智能体循环|s01 基础 agent loop]] 唯一的不同是：每次 LLM 调用前先检查收件箱。

def _teammate_loop(self, name, role, prompt):
    messages = [{"role": "user", "content": prompt}]
    for _ in range(50):
        inbox = BUS.read_inbox(name)
        if inbox != "[]":
            messages.append({"role": "user",
                "content": f"<inbox>{inbox}</inbox>"})
            messages.append({"role": "assistant",
                "content": "Noted inbox messages."})
        response = client.messages.create(...)
        if response.stop_reason != "tool_use":
            break
        # execute tools, append results...
    self._find_member(name)["status"] = "idle"

循环结束后，队友状态变为 idle。下次有新消息进收件箱时，由管理者或另一个队友重新唤醒。

团队通信模式

基于这三个新工具（spawn、send、read_inbox），可以实现多种协作模式：

模式	流程	示例
指令下发	Lead → Alice	send("lead","alice","实现这个功能")
汇报回传	Alice → Lead	send("alice","lead","已完成")
同级协作	Alice → Bob	send("alice","bob","帮我 review 代码")
广播	Lead → all	遍历名册逐人 send()
轮询	Lead → inbox	read_inbox("lead") 检查是否有回复

关键设计决策

1. 线程而非进程：队友之间共享 Python 进程空间，通信成本极低，但也意味着 GIL 限制。对于 I/O 密集的 LLM 调用来说这通常不是问题
2. 状态轮转：idle → working → idle 的循环让队友不需要持续消耗 token。只有在收到消息时才会触发 LLM 调用
3. 领导者模型：只有一个全功能智能体（lead）拥有 spawn/send/read_inbox 工具，队友仅有基础的 read/write/bash/task 工具。这是 [[04-子智能体|s04 父子工具隔离]] 的自然延伸
4. 安全上限：每个队友循环上限 50 次迭代，防止单个队友失控。这对于后续 [[11-自主智能体|s11 自主智能体]] 的自适应循环设计是一个重要参考

相对 s08 的变更

组件	之前 (s08)	之后 (s09)
Tools	6	9（+spawn/send/read_inbox）
智能体数量	单一	领导 + N 个队友
持久化	无	config.json + JSONL 收件箱
线程	后台命令	每线程完整 agent loop
生命周期	一次性	idle → working → idle
通信	无	message + broadcast

从代码量上看变化不大——核心的 agent loop 没有改动，只加了三个新工具和一个队友管理器。但这标志着 Harness 从一个"单人执行器"变成了"多人协作平台"。

⭐ 核心要点

1. 文件即通信 — JSONL 收件箱是零依赖的队友通信方案，append-only 写入天然支持并发
2. 身份与生命周期 — 每个队友有名字、角色、状态（idle/working），不再是一次性的工具调用
3. 共享工具，隔离上下文 — 队友共享文件系统和工具集，但每个队友有独立的 messages[]，延续 [[04-子智能体|s04 上下文隔离]] 原则
4. 领导者代理 — 只有 lead 有团队管理工具，队友只有执行工具，防止权限扩散
5. 线程安全 — POSIX append 语义 + drain-on-read 模式避免了大部分竞态条件
6. 从单人走向多人 — 这是从"单智能体系统"到"多智能体系统"的第一步

试一试

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 手动检查领导的收件箱

思考题

1. 当前设计是"领导者审批制"——只有 lead 能 spawn。如果允许队友自行 spawn 新队友，会出现什么问题？
2. JSONL 收件箱的 drain-on-read 模式意味着消息只能被消费一次。如果需要"广播确认"（确保所有队友都读了消息），应该如何扩展？
3. 队友之间的消息目前是纯文本。如果需要支持"发送工具调用结果"或"发送文件引用"，消息协议应该如何设计？
4. 当队友数量增长到几十个时，config.json 的单一文件架构会成为瓶颈吗？如何设计可扩展的团队注册表？

下节预告

团队有了，但队友之间怎么协调工作？目前的消息传递还是"我说你听"的点对点模式。[[10-团队协议|s10 团队协议]] 将引入结构化通信协议——指令、汇报、请求、确认——让智能体团队像人类团队一样有章法地协作。

延伸阅读

• Learn Claude Code 项目 — 练习代码和完整文档
• [[07-任务系统|s07 任务系统]] — 团队的待办清单
• [[00-课程概览/教学大纲|教学大纲]] — 查看完整课程地图
• [[00-课程概览/综述|课程综述]] — Agent Harness 工程导论

---

「团队协作不是消息传递，而是身份认同」