第11课：自治Agent——自组织任务认领
队友自己看看板，有活就认领。WORK-IDLE双阶段循环+身份重注入，让多Agent团队从手动指派走向自组织协作。
系列导读
这是**《12课拆解Claude Code架构》系列的第 11 课，也是阶段四（多 Agent 协作）的最后一课**。
前十课我们从一个 while 循环出发，一路叠加工具注册、规划系统、Subagent、技能加载、上下文压缩、任务持久化、后台执行、多 Agent 团队、消息传递。到第 10 课，队友之间已经能通过收件箱互发消息了。
但有一个问题越来越刺眼——队友只在被明确指派时才干活。
第 11 课的格言：
"队友自己看看板，有活就认领"
这一课，我们让 Agent 从"等领导分配"变成"自己找活干"。
手动指派的瓶颈
第 9-10 课的团队模式下，领导（orchestrator）是唯一的任务分配者：
领导看到10个待办任务
↓
给coder写prompt → 给tester写prompt → 给reviewer写prompt
↓
还剩7个任务没人认领，等队友做完再一个一个分配……| 瓶颈 | 表现 |
|---|---|
| 领导成为单点 | 所有任务必须经过领导分配 |
| 指派延迟 | 队友做完一个任务后要等领导轮到自己 |
| 扩展不了 | 10 个队友就要写 10 次 prompt |
| 空闲浪费 | 队友做完活只能干等，不会自己找下一个 |
现实世界里，好的团队不是这样运作的。你不会每写一张便利贴就走到某个人工位上贴到他脸上。 你贴在看板上，谁有空谁拿。
WORK-IDLE 双阶段循环
核心思路：每个队友的循环不再只有 WORK，增加一个 IDLE 阶段——做完活就自动去看板找下一个。
┌──────────────────────────┐
│ Agent 启动 │
└────────────┬─────────────┘
▼
┌──────────────────────────┐
┌──→│ IDLE 阶段 │
│ │ 轮询收件箱 + 扫描看板 │
│ │ (每5s，最多60s) │
│ └────────┬────────┬────────┘
│ │ │
│ 找到任务 60s超时
│ ▼ ▼
│ ┌──────────┐ ┌──────────┐
│ │ claim │ │ SHUTDOWN │
│ │ 认领任务 │ └──────────┘
│ └─────┬────┘
│ ▼
│ ┌──────────────────────────┐
│ │ WORK 阶段 │
│ │ 正常 Agent Loop 执行任务 │
│ └────────────┬─────────────┘
│ 任务完成
└────────────────┘- WORK：正常的 Agent Loop，和前面课程完全一致
- IDLE：轮询收件箱 + 扫描看板，每 5 秒一次，最多 60 秒
找到活 → claim → 进入 WORK。60 秒没活 → 自动 SHUTDOWN。
核心拆解：四个关键机制
机制一：idle_poll——轮询等待
def idle_poll(agent_id: str, task_board: TaskBoard,
inbox: Inbox, timeout: int = 60) -> Optional[str]:
deadline = time.time() + timeout
while time.time() < deadline:
# 先检查收件箱——有人直接指派给我的优先
msg = inbox.poll(agent_id)
if msg and msg.get("task_id"):
return msg["task_id"]
# 再扫描看板——找无主任务
task_id = scan_unclaimed_tasks(agent_id, task_board)
if task_id:
return task_id
time.sleep(5)
return None # 超时，准备关机优先级：直接指派 > 自由认领 > 等待 > 超时关机。
机制二：scan_unclaimed_tasks——扫描看板
def scan_unclaimed_tasks(agent_id: str,
task_board: TaskBoard) -> Optional[str]:
for task in task_board.list_tasks():
if (task["status"] == "pending"
and task.get("owner") is None
and not is_blocked(task, task_board)):
return task["id"]
return None
def is_blocked(task: dict, task_board: TaskBoard) -> bool:
for dep_id in task.get("blocked_by", []):
dep_task = task_board.get_task(dep_id)
if dep_task and dep_task["status"] != "completed":
return True
return False三个过滤条件全部满足才能认领：pending + 无 owner + 未被阻塞。blocked_by 是任务间的依赖——"写测试"依赖"写代码"完成，代码没写完就不能拿测试任务。
机制三：claim——原子认领
def claim_task(agent_id: str, task_id: str,
task_board: TaskBoard) -> bool:
with task_board.lock:
task = task_board.get_task(task_id)
if (task and task["status"] == "pending"
and task.get("owner") is None):
task["status"] = "in_progress"
task["owner"] = agent_id
task_board.update_task(task_id, task)
return True
return False加锁。 两个 Agent 可能同时看到同一个无主任务，锁保证只有一个能认领成功。认领失败的会在下一轮继续扫描。这就是去中心化调度的核心：不需要中央分配器，冲突通过锁 + 重试解决。
机制四：身份重注入
第 6 课的 Context Compact 会压缩消息历史，Agent 可能丢失关键信息——包括自己是谁。
压缩前: [0] system: "你是coder" → [1-39] 大量工具调用
压缩后: [0] assistant: "之前讨论了..." ← system提示词被压缩掉了
Agent忘了自己是coder，开始做tester的活解决方案：消息列表过短时，在开头插入身份块。
IDENTITY_THRESHOLD = 3
def maybe_reinject_identity(agent_id: str, role: str,
messages: list,
identity_prompt: str) -> list:
if len(messages) >= IDENTITY_THRESHOLD:
return messages
identity_block = {
"role": "user",
"content": (f"<identity>\n你是 {agent_id}，角色是 {role}。\n"
f"{identity_prompt}\n</identity>"),
}
return [identity_block] + messages触发条件是消息列表长度——Compact 后消息数量骤降，阈值检查能准确捕捉到"刚被压缩过"的时刻。
组装：自治 Agent 循环
四个机制拼进 Agent Loop，外层 while True 是 IDLE-WORK 切换，内层 while True 是原有的 Agent Loop：
def autonomous_agent_loop(agent_id, role, identity_prompt,
task_board, inbox, tools, bg_manager):
messages = []
while True:
# ——— IDLE 阶段 ———
task_id = idle_poll(agent_id, task_board, inbox)
if task_id is None:
print(f"[{agent_id}] 60秒无任务，自动关机")
break
if not claim_task(agent_id, task_id, task_board):
continue # 被别人抢了，继续找
task = task_board.get_task(task_id)
messages.append({"role": "user",
"content": f"任务: {task['title']}\n{task['description']}"})
# ——— WORK 阶段（原有 Agent Loop，一行不改） ———
while True:
messages = maybe_reinject_identity(
agent_id, role, messages, identity_prompt)
response = client.messages.create(
model=MODEL, system=identity_prompt,
messages=messages, tools=tools, max_tokens=8000)
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 = dispatch_tool(block.name, block.input)
results.append({"type": "tool_result",
"tool_use_id": block.id, "content": output})
messages.append({"role": "user", "content": results})
task_board.update_task(task_id, {"status": "completed", "owner": agent_id})
# 回到外层 while True → 继续 IDLEWORK 阶段的 Agent Loop 从第 1 课到第 11 课，一行不改。 自治逻辑全部在循环之外。
实际运行：三个 Agent 自组织协作
看板上 5 个任务，3 个 Agent 同时启动：
看板: T1(pending) T2(pending,blocked_by=T1) T3(pending,blocked_by=T2)
T4(pending) T5(pending)
[coder] IDLE → T1无阻塞 → claim → WORK
[tester] IDLE → T2被阻塞 → T4无阻塞 → claim → WORK
[devops] IDLE → T5无阻塞 → claim → WORK
[devops] 完成T5 → IDLE → 无活（T2/T3被阻塞）
[tester] 完成T4 → IDLE → 无活
[coder] 完成T1 → IDLE → T2解除阻塞 → claim → WORK
[coder] 完成T2 → IDLE → T3解除阻塞 → claim → WORK
[devops] IDLE超时 → SHUTDOWN
[tester] IDLE超时 → SHUTDOWN
[coder] 完成T3 → IDLE → 无活 → IDLE超时 → SHUTDOWN没有任何中央调度。 依赖自动通过 blocked_by 解决，空闲 Agent 自动退出。
两个设计洞见
洞见一：去中心化调度的代价
每个 Agent 每 5 秒扫一次看板，3 个 Agent 是 3 次/5s，100 个 Agent 是 100 次/5s。当前规模下可忽略（内存字典读操作不花时间），但原则上：去中心化用轮询换掉了中央推送。 规模上去后可优化为事件驱动，但不改变架构。
洞见二：身份重注入是 Agent 的"名牌"
Context Compact 会无差别压缩——包括身份信息。身份丢失的后果不只是"忘了自己是谁"，更是角色越权：coder 开始审查代码，tester 开始写业务逻辑。重注入把角色边界重新划清。
用 <identity> 标签而不是修改 system prompt——因为 system prompt 在 API 层面只设一次，身份重注入需要在对话过程中动态触发。
五分钟跑起来
# 进入项目目录
cd learn-claude-code
# 启动第十一课
python agents/s11_autonomous_agents.py任务一：自动认领 + 依赖解锁 + 冲突拒绝
创建 3 个任务，spawn 两个队友，观察他们自组织：
s11 >> Create 3 tasks on the board, then spawn alice and bob.
Watch them auto-claim.实际执行记录（精简）：
> bash: ✅ Tasks 8, 9, 10 created!
#8 Add farewell() to greet.py blockedBy: []
#9 Create CHANGELOG.md blockedBy: []
#10 Write test for farewell blockedBy: [8] ← 等 #8 完成
> spawn_teammate: Spawned 'alice' (role: developer)
> spawn_teammate: Spawned 'bob' (role: developer)两个 Agent 同时扫描看板，自主认领任务：
[bob] claim_task: Claimed task #8 for bob ← bob 先抢到 #8
[alice] write_file: CHANGELOG.md ← alice 拿了 #9
[bob] claim_task: Error: Task 9 has already been claimed by alice
← bob 想抢 #9，被拒绝！
[alice] edit_file: greet.py → 添加 farewell()
[alice] bash: Goodbye, Alice! ← 验证 farewell 函数alice 完成 #8 后，_clear_dependency 触发，#10 解锁：
→ Task 10 unblocked! ← 依赖自动解锁
[bob] claim_task: Claimed task #10 for bob ← bob 抢到解锁的 #10
[bob] bash: pytest → 3 tests, all passing ✅两个 Agent 完成所有任务后自动进入空闲：
[alice] idle: Entering idle phase. Will poll for new tasks.
[bob] send_message: Sent message to lead ← 汇报完成关键行为验证：
- 自动认领：无人指派，alice 和 bob 自己扫看板抢任务
- 冲突拒绝：bob 尝试认领 alice 已拿的 #9，被
claim_task拒绝 - 依赖解锁：#10 等 #8 完成才可认领，
_clear_dependency精确生效 - 自动空闲：任务做完，进入 idle 轮询
任务二：完全自治——零指令执行全部任务
只 spawn 一个 coder，什么都不告诉它，让它自己找活：
s11 >> Spawn a coder teammate and let it find work from the task board itself实际执行记录（时间线）：
> spawn_teammate: Spawned 'coder' (role: developer)
[coder] 自主扫描 .tasks/ 目录...
[coder] claim_task: Claimed task #2 for coder ← 自动找到 pending 任务
[coder] write_file: src/calculator.py ← 写代码
[coder] claim_task: Claimed task #3 for coder ← 完成后立即找下一个
[coder] write_file: tests/test_calculator.py ← 写测试
[coder] bash: pytest → 25 tests passed ✅
[coder] claim_task: Claimed task #4 for coder ← Parse（#5/#6 的前置）
[coder] write_file: src/refactor/parse.py
[coder] claim_task: Claimed task #5 for coder ← Transform（#4完成后解锁）
[coder] write_file: src/refactor/transform.py
[coder] claim_task: Claimed task #6 for coder ← Emit（也是 #4 完成后解锁）
[coder] write_file: src/refactor/emit.py
[coder] claim_task: Claimed task #7 for coder ← Test（#5 和 #6 都完成后解锁）
[coder] write_file: tests/test_refactor.py
[coder] bash: pytest → 29 tests passed ✅
[coder] send_message: "All available tasks have been completed."
[coder] idle: Entering idle phase. Will poll for new tasks.一个 Agent，零指令，自动清空 6 个任务——扫看板、认领、执行、完成、找下一个，循环直到全部做完。
任务三：钻石依赖图——验证并行分叉和汇聚
最硬核的测试——创建钻石形依赖，两个 Agent 并行执行：
s11 >> Create tasks with dependencies. Watch teammates respect the blocked order.任务图结构：
#1 Schema
/ \
#2 Backend #3 Frontend ← 并行分叉
\ /
#4 Integration ← 汇聚：等 #2 和 #3 都完成
|
#5 Deploy ← 线性：等 #4实际执行时间线：
~8s 🔄 frank claims #1 (唯一未阻塞的任务)
→ eve 无任务可认领，等待
~23s ✅ #1 done → #2 和 #3 同时解锁（钻石分叉）
🔄 eve claims #2 (Backend)
🔄 eve claims #3 (Frontend) ← eve 手快，两个都抢到
~43s ✅ #2 + #3 done → #4 解锁（钻石汇聚）
🔄 eve claims #4 (Integration)
→ frank 轮询发现 #4 已被 eve 认领
~68s ✅ #4 done → #5 解锁
🔄 eve claims #5 (Deploy)
~93s ✅ #5 done — 全部完成
[eve] send_message: "All tasks completed"
[frank] send_message: "All tasks completed"
→ 两个 Agent 都进入 idle验证最终状态：
s11 >> /tasks
[x] #1: Design the data schema @frank
[x] #2: Build the backend API layer @eve
[x] #3: Build the frontend formatter layer @eve
[x] #4: Integration: wire backend + frontend together @eve
[x] #5: Deploy: generate final report @eve
s11 >> /team
eve (developer): shutdown
frank (developer): shutdown关键行为验证：
- 钻石分叉：#2 和 #3 在 #1 完成后同时解锁
- 钻石汇聚：#4 等 #2 和 #3 都完成才解锁（不是只等一个）
- 线性依赖：#5 等 #4 完成
- 抢占式认领：eve 手快抢了大部分任务，frank 轮询发现没活就继续等
- 自动退出：全部做完，两个 Agent 自动 shutdown
总结：你刚造了什么
| 组件 | 之前（s09-s10） | 之后（s11） |
|---|---|---|
| 任务分配 | 领导手动指派 | 自组织认领 |
| Agent 空闲时 | 干等领导分配 | 轮询收件箱 + 扫描看板 |
| 任务认领 | 仅手动 | 自动 claim（加锁防冲突） |
| 身份保持 | 无机制 | Context Compact 后重注入 |
| 退出机制 | 领导控制 | 60 秒无活自动 SHUTDOWN |
| 新增代码 | — | ~80 行（idle_poll + claim + identity） |
核心变化：从中央调度到自组织。 领导不再是瓶颈，Agent 自己找活、自己认领、自己退出。
下一课预告
第 11 课让 Agent 团队能自组织协作。但所有 Agent 还共享同一个工作目录——两个 Agent 同时改同一个文件怎么办？
第 12 课：Worktree Isolation —— 用 Git Worktree 给每个 Agent 一个独立的工作空间。
# 预告：s12 的 worktree 隔离
worktree = create_worktree(agent_id, branch="feature/user-auth")
agent_loop(cwd=worktree.path, ...)
merge_worktree(worktree)并行不冲突。每个 Agent 有自己的沙箱。
这是《12课拆解Claude Code架构：从零掌握Agent Harness工程》系列的第 11 课。关注Claw开发者，不错过后续更新。
完整代码和交互式学习平台：github.com/shareAI-lab/learn-claude-code
如果这篇文章对你有帮助，欢迎转发给你的技术团队。
系列目录
- 第1课：用20行Python造出你的第一个AI Agent
- 第2课：给Agent加工具 —— dispatch map模式详解
- 第3课：TodoWrite —— 让Agent先想后做：规划系统
- 第4课：Subagent —— 拆解大任务，上下文隔离
- 第5课：按需加载领域知识——Skill机制
- 第6课：无限对话——上下文压缩三层策略
- 第7课：任务持久化——文件级DAG任务图
- 第8课：后台执行——异步任务与通知队列
- 第9课：Agent Teams——多Agent协作：团队与邮箱系统
- 第10课：团队协议——状态机驱动的协商
- 第11课：自治Agent——自组织任务认领（本文）
- 第12课：终极隔离——Worktree并行执行