第 6 章：Agent Team——让固定角色通过协议持续协作

← 上一章：Subagent · 下一章：Tool、Skill 与 MCP

开场：外包师傅和固定班底，差在哪？

上一章的 Subagent 像一次外包：交一件事，拿一次结果，局部上下文随后结束，师傅走了就不再来。

但很多真实任务不是一次外包。「开发者写代码 → Reviewer 审查 → 开发者修改 → Reviewer 复核」——这是同一个班底反复协作。如果每轮都重新创建角色、每次完事就把上下文销毁，上次讨论的未解决问题、Reviewer 记住的审查标准，全都要从零再来。

再想想餐厅的厨房。临时请来的外包师傅做完一道菜就走；固定班底是另一回事：主厨（Lead）派活、配菜的备料、质检的验收。做完一道菜，大家不是被开除，而是回到 idle 等着下一单。可这时候最微妙的问题出现了——一群人同时在厨房里，为什么还不是一个「团队」？

必须有人回答这些问题：谁负责什么？消息怎样送达（而不是靠吼）？什么共享、什么各自留着？哪个成员现在真的在线？两个人都说「这道菜我验收过了」，听谁的？以及最关键——这顿饭什么时候算整体结束？

这一章把「同时跑几个模型」升级成真正的 Agent Team。本章结束时，你能回答：为什么 asyncio.gather 不是 Agent Team？

让三个模型同时跑，为什么还不是一个 Agent Team？

• A：它们缺少角色身份、可寻址的消息通道、持续状态和团队级结束条件
• B：它们跑得不够快
• C：需要更多模型
• D：它们没有系统提示

并发任务为什么不等于团队

先排除一个最容易被误认成团队的东西——并发：

results = await asyncio.gather(fetch_a(), fetch_b(), fetch_c())

这段代码有并发，但没有：成员身份和角色、独立的长期上下文、可寻址的 Inbox、工作后 idle 并等待下一条消息、团队花名册与管理状态、Lead 与任务板、团队终止协议。所以它是「并发任务集合」，不是 Agent Team。

「多个模型」不自动构成团队。 就像厨房里站着十个人，如果没有分工、没有消息传递、没有总控、没有「什么时候收工」的约定，那只是十个人在同一个房间。

那么一个最小 Team 到底由什么组成？四层状态，一层都不能少：

层	保存什么	例子
成员私有状态	自己的 messages、草稿、局部预算	Reviewer 的审查上下文
协作状态	消息、任务板、共享 Artifact 引用	report.md 的版本 ID
管理状态	成员、角色、在线状态、配额	alice/coder/idle
终止状态	成功标准、未完成工作、取消与 ack	首次有效报告后停止其他搜索

私有和共享必须分开。 不要把所有人的完整 messages 拼成一个「团队大脑」。私有的归私有：每个成员的 system prompt、history、草稿、工具结果。共享的走共享通道：任务板、消息信封、经过版本化的产物引用。成员交换 artifact://review/42，而不是在消息里复制五千行 Diff——这样谁也不会被别人的上下文淹没。

花名册、信封与成员生命周期

成员花名册回答「谁在队里、什么状态」。名字是寻址标识，角色决定职责：

from dataclasses import dataclass

@dataclass
class Member:
    name: str
    role: str
    status: str = "offline"
    current_task_id: str | None = None
    last_seen_sequence: int = 0

状态必须反映真实运行时，而不是只读配置文件。 教学仓库里专门有一个 _mark_stale_members_offline()：进程重启后，配置里还写着 idle 的成员，其实执行线程早已不存在——必须把它纠正成 offline，不能假装它还在线。idle 表示「运行时存活、等待消息」；offline 表示「没有活跃执行者」。 两者混用，就会出现「程序重启后配置还在，但实际干活的人已经消失」的幻觉。

成员生命周期是一条链：

offline → starting → idle → working → failed → shutdown_requested → shutdown

消息要带信封，不能裸丢一句话：

from dataclasses import dataclass

@dataclass(frozen=True)
class Envelope:
    message_id: str
    sequence: int
    sender: str
    recipient: str
    type: str
    content: str
    artifact_refs: tuple[str, ...] = ()

生产消息还应加三样东西：correlation_id 把提问和回复关联起来、idempotency_key 保证重复投递不重复产生副作用、ack 与重试次数让系统知道「这条消息有没有被确实处理」。

成员完成当前任务、等着下一条消息，此时的状态是？

• A：shutdown——它已经结束了
• B：idle——运行时存活，等待消息
• C：offline——配置文件里没有它
• D：working——它还在处理

成员循环与可靠消息

成员的日常工作是一条持续循环：

while member is alive:
  message = await inbox.receive(member)
  set status=working
  result = await member.runner.run(message)
  publish artifact/report
  ack message
  set status=idle

这里有一个教学实现会踩的坑：「读取即清空」不是可靠消息语义。 成员读到 inbox.jsonl、文件被清空、然后进程在处理完成前崩溃——这条消息就永久丢了。正确的做法是把读取和确认分开：peek 读出但不删除，处理成功后 ack 才移除。崩溃重启后，未 ack 的消息还能重新读到。

教学仓库的 Inbox 是文件（.team/config.json + inbox/*.jsonl），价值在于可观察、可 diff、可调试；但它的「读取即清空」是故意留下的教学简化。真正的消息总线还需要原子写入、顺序保证和消息系统。你应当看出这两者的边界。

教学版 Inbox「读取后立即清空」的缺陷是什么？

• A：太占磁盘空间
• B：消息太多会乱序
• C：成员读到消息后进程崩溃，消息就永久丢了——读取与确认必须分离
• D：没人能读到消息

结算、终止与团队完成

两个成员同时宣布「我成功了」，系统不能重复结算。这需要幂等结算：对 commit SHA + report type 之类的键只结算一次，由单一 Lead/Coordinator 持锁发布。迟到结果保留为事件，但不能重复产生副作用。

团队不能因为某个成员发来「我完成了」就整体结束。 Lead 必须核对团队完成谓词：

必需任务全部进入成功终态
AND 产物有确定性验证
AND 没有未解决的阻断性问题
AND 没有仍在 working 的必需成员
AND 终止广播已发送且 ack 已收敛

终止也不是说一句「大家停」。要广播终止、等待成员在安全点停止并 ack；超时后强制取消并记录。这保证了「找到答案后其他成员继续烧成本」「一个离线成员被误算作等待中」这类问题被兜住。

完整流程长这样——消息总线是控制平面，Artifact Store 是数据平面，不要用一个超长消息同时承担两者：

一个成员发来「我完成了」，团队就能整体结束吗？

• A：能，成员说完成就完成
• B：不能，Lead 要核对必需任务终态、产物验证、无 working 成员、终止广播已 ack
• C：永远不能结束
• D：看用户心情

动手：让两个失败发生，再用一个测试收住

理论和手感之间隔着一道墙，亲手撞一次才会留下印象。下面两个实验都是故意做错的。

失败一：读取即清空，崩溃后消息没了。

# 教学版 Inbox：读取后立即清空——危险
class Inbox:
    def __init__(self):
        self.queue = ["review-commit-3"]
    def read(self):
        msg = self.queue.pop(0)   # 读到消息
        self.queue.clear()        # 立刻清空
        return msg

inbox = Inbox()
msg = inbox.read()   # 成员读到消息……
# ……成员进程在这时崩溃，还没来得及处理……
print("已读消息：", msg)
print("重启后还能读到吗？", "不能——消息已经没了")

失败二：两个成员同时宣布成功，重复结算。

import asyncio

async def main():
    state = {"settlements": 0}
    async def report(name):
        await asyncio.sleep(0)                     # 两个协程几乎同时到达
        if state["settlements"] == 0:              # 这不是原子检查
            state["settlements"] += 1              # 并发下可能都被执行
            return f"{name} settled"
        return f"{name} late"
    results = await asyncio.gather(report("reviewer-a"), report("reviewer-b"))
    print(state, results)

asyncio.run(main())

没有锁的检查-然后-写入在并发下可能结算两次：写两条 MR 评论、两次改变门禁、两次广播终止。修复是 asyncio.Lock + 幂等结算键，由单一 Lead 持锁发布。

跑完后用现成测试验证「模型提议 → 确定性证据核实」这道闸真实有效——Mini Emperor 的 AsyncReviewPipeline 还不是持久 Team，但它的这一环正是 Team 里 Reviewer/Verifier 协作的雏形：

cd "$(git rev-parse --show-toplevel)"
uv run --python 3.12 --extra dev pytest \
  mini-emperor/backend/tests/test_review.py::test_async_reviewer_filters_findings_without_diff_evidence \
  -v

如果它不过，按这个顺序排查：

• 模型提议了但没有证据 → 确定性 Verifier 有没有对照真实 Diff；
• 证据指向的行号不存在 → 越界查找有没有被拦截；
• 虚假证据没被过滤 → 核实的代码路径有没有真的跑。

门禁·代码：跑通上面这条测试，把输出保存为证据。

本章小结

这一章把「一群人」升级成了「一个团队」。四件事连起来看：

asyncio.gather 不是 Agent Team。 并发只有「同时跑」；Team 有身份、消息通道、持续状态和团队级结束条件。多个模型不自动构成团队。

固定班底的关键是「完成后再等待」。 成员做完一件事进入 idle 而不是销毁，才能响应后续消息。idle 和 offline 是两个概念，重启后必须纠正虚假在线。

消息要可靠，结算要幂等。 「读取即清空」会丢消息，读取与 ack 要分离；两个成员同时「成功」要由单一 Lead 持锁只结算一次。

团队结束由谓词决定，不由成员的一句话决定。 Lead 核对必需任务终态、产物验证、无 working 成员、终止广播 ack 收敛，才宣布完成。

一句话记住这一章：

固定角色、可寻址消息、私有与共享分界、真实在线状态、幂等结算、团队完成谓词——缺一样，都只是「一起跑的模型」，不是团队。

如果上面这些话你自己也能说得出来（不看资料），就说明这一章的知识已经连成一片。

复述与证据

合上资料，画出成员的生命周期：offline → idle → working → idle → shutdown。再画出「Lead 派活 → 成员 Inbox → 私有循环 → 回报 → Lead 结算」的完整闭环，标出控制平面和数据平面分别是谁。

再追问三个问题：

1. idle 与 offline 为什么不能混用？
2. 「读取即清空」会在哪个时序丢数据？
3. 为什么首次命中要幂等结算？

门禁·证据：提交一份成员状态快照（含 team_id、member/role、status、task_id、ack_count），并说明团队为什么还没到能结束的状态。

24 小时后：不看资料，写一个最小 Team Registry + Inbox + Settlement，模拟两个成员并发回报和一次终止广播。

有兴趣可以继续看

以下资料按优先级排。至少完成一项 P0，并保存完成证据。

视频P020 分钟 · 视频 06 [00:00:00–00:05:15] 生命周期，再看 [00:05:59–00:08:37] 协作闭环与解散

预计

20 分钟

位置

视频 06 [00:00:00–00:05:15] 生命周期，再看 [00:05:59–00:08:37] 协作闭环与解散

阅读任务

回答成员完成一次任务后为何仍不销毁、重启后为何变 offline。

完成证据

画 spawn → working → idle → message → working → shutdown 状态图。

教学仓库P060 分钟 · Step 10，重点 MessageBus、TeammateManager.spawn()、_teammate_loop()、_mark_stale_members_offline()

预计

60 分钟

位置

Step 10，重点 MessageBus、TeammateManager.spawn()、_teammate_loop()、_mark_stale_members_offline()

阅读任务

区分哪些状态保存在文件、哪些只存在进程内，不一致时谁是真相。

完成证据

运行后查看 config 与 inbox；重启并证明旧 working/idle 变 offline。

书与实验P050 分钟 · ai-agent-book 第 10 章：上下文共享轴、协作拓扑、失败模式

预计

50 分钟

位置

ai-agent-book 第 10 章：上下文共享轴、协作拓扑、失败模式

阅读任务

判断教学实现属于哪种上下文与拓扑组合；换成对等模式要改什么。

完成证据

为 Team 标注控制平面、数据平面和三个失败注入点。

论文P175 分钟 · CAMEL 与 AutoGen 的角色设定与多 Agent 对话

预计

75 分钟

位置

CAMEL 与 AutoGen 的角色设定与多 Agent 对话

阅读任务

观察角色 Prompt 解决了什么，哪些生产边界仍必须由 Harness 强制。

完成证据

选一个论文模式，补上 member registry、message ID、artifact version、budget 和 termination predicate。

本文提到的代码与出处（有兴趣逐条核对时用）：

• claude-agent-examples@54a18980334541773f13940aa0c0475728d30ee0:build-agent-example/code/step10_agent_team.py
• claude-agent-examples@54a18980334541773f13940aa0c0475728d30ee0:build-agent-example/doc/step10_agent_team.md
• ai-agent-book@1c18370279f8f0457bf2c44dfa585d08d4d5f281:book/chapter10.md
• ai-agent-book@1c18370279f8f0457bf2c44dfa585d08d4d5f281:chapter10/parallel-web-research/agents.py
• mini-emperor/backend/src/mini_emperor/review.py
• mini-emperor/backend/tests/test_review.py

← 上一章：Subagent · 下一章：Tool、Skill 与 MCP