learn claude code S11 自主 Agent 详解笔记

S11 自主 Agent 详解笔记基于s11_autonomous_agents.py源码逐行分析配合s11-autonomous-agents.md设计思路。一、问题队友需要有人持续指派任务s09-s10 的 teammate 有一个尴尬的空白期完成当前任务后进入idle然后呢Lead 必须主动发消息给它新 prompt。如果 Lead 很忙或本身就是另一个 agentteammate 就闲置着。更糟的是s06 的压缩可能丢掉 Lead 的消息teammate 醒来发现 inbox 是空的什么都不做。idle 不是终点是等别人告诉我该干什么的等待状态。但谁说一定要别人告诉s11 的核心思想让 agent 自己找到工作。不是 Lead 分配任务而是 teammate 扫描任务板看到没人认领的就去领。二、解决方案空闲轮询 自动认领s11 在 teammate 的循环中加入了idle 阶段WORK 阶段标准 agent loop ↓ stop_reason ! tool_use IDLE 阶段最多 60 秒 ├── 每 5 秒检查 inbox → 有消息 → 恢复 WORK ├── 每 5 秒扫 .tasks/ → 有未认领任务 → 认领 → 恢复 WORK └── 超时60 秒→ shutdown同时还解决了压缩后的身份丢失问题identity block 确保压缩后 teammate 仍然知道我是谁、我的角色是什么。三、和 s10 相比多了什么组件s10s11任务认领无scan_unclaimed_tasks()claim_task()idle 模式无直接 shutdown/idle轮询 inbox 任务板身份保持无make_identity_block()压缩后恢复队友工具8 个10 个idle,claim_task生命周期spawn → work → idlespawn → work ⇄ idle循环 → shutdown四、任务板扫描与认领4.1scan_unclaimed_tasks()— 找到可领的任务defscan_unclaimed_tasks()-list:TASKS_DIR.mkdir(exist_okTrue)unclaimed[]forfinsorted(TASKS_DIR.glob(task_*.json)):taskjson.loads(f.read_text())# 三个条件pending 无 owner 无依赖阻塞if(task.get(status)pendingandnottask.get(owner)andnottask.get(blockedBy)):unclaimed.append(task)returnunclaimed三个条件缺一不可pending没被完成或废弃!owner没被其他 agent 认领!blockedBy前置依赖已全部完成如果还有依赖blockedBy不为空返回的是所有未认领任务。teammate 只取第一个unclaimed[0]——它一次只接一个任务做完再去领下一个。4.2claim_task()— 原子认领defclaim_task(task_id:int,owner:str)-str:with_claim_lock:# 关键线程锁pathTASKS_DIR/ftask_{task_id}.jsontaskjson.loads(path.read_text())# 三重防御检查iftask.get(owner):returnfError: Task{task_id}has already been claimed by{task[owner]}iftask.get(status)!pending:returnfError: Task{task_id}cannot be claimed (status:{task[status]})iftask.get(blockedBy):returnfError: Task{task_id}is blockedtask[owner]owner task[status]in_progresspath.write_text(json.dumps(task,indent2))returnfClaimed task #{task_id}for{owner}_claim_lock是核心防护假设 Alice 和 Bob 同时扫描到 task 5 没人认领同时调用claim_task(5, ...)。没有锁的话两人都可能读到一个owner为空的状态然后都设定自己为 owner——任务被认领了两次。with _claim_lock:确保读-检查-写是一个原子操作。第一个线程读、检查、写完成后第二个线程才进入——此时owner已经不是空了检查失败。五、身份块压缩后记得自己是谁s06 的 auto_compact 会把 messages 替换为 LLM 的摘要。摘要可能丢掉关键的身份信息——teammate 醒来后不再知道我叫 Alice角色是 backend developer。defmake_identity_block(name:str,role:str,team_name:str)-dict:return{role:user,content:(fidentityYou are {name}, role:{role}, fteam:{team_name}. Continue your work./identity),}在 idle 阶段认领新任务后如果 messages 很短len(messages) 3说明刚被压缩在 messages 开头插入身份块iflen(messages)3:messages.insert(0,make_identity_block(name,role,team_name))messages.insert(1,{role:assistant,content:fI am{name}. Continuing.})len(messages) 3是压缩的指纹——auto_compact 后 messages 只有 1 条摘要消息正常对话至少有几轮交互。这是一个简单但有效的启发式判断。六、Teammate 的完整生命周期6.1 WORK 阶段和 s10 的 agent_loop 几乎相同。唯一的新增是idle工具ifblock.nameidle:idle_requestedTrueoutputEntering idle phase. Will poll for new tasks.模型判断我手上的活做完了没有新指令调用idle。循环检测到idle_requested后 break 进入 idle。6.2 IDLE 阶段POLL_INTERVAL5# 每 5 秒轮询一次IDLE_TIMEOUT60# 60 秒后放弃self._set_status(name,idle)pollsIDLE_TIMEOUT//max(POLL_INTERVAL,1)# 60/5 12 次for_inrange(polls):time.sleep(POLL_INTERVAL)# 检查 inboxinboxBUS.read_inbox(name)ifinbox:formsgininbox:ifmsg.get(type)shutdown_request:self._set_status(name,shutdown)returnmessages.append({role:user,content:json.dumps(msg)})resumeTruebreak# 扫描任务板unclaimedscan_unclaimed_tasks()ifunclaimed:taskunclaimed[0]resultclaim_task(task[id],name)ifresult.startswith(Error:):continue# 认领失败可能被抢了继续轮询# 认领成功 → 注入任务作为新 prompttask_promptfauto-claimedTask #{task[id]}:{task[subject]}\n{task[description]}/auto-claimediflen(messages)3:messages.insert(0,make_identity_block(name,role,team_name))messages.insert(1,{role:assistant,content:fI am{name}. Continuing.})messages.append({role:user,content:task_prompt})messages.append({role:assistant,content:fClaimed task #{task[id]}. Working on it.})resumeTruebreak# 12 次都没结果 → shutdownifnotresume:self._set_status(name,shutdown)return# 有结果 → 恢复 WORKself._set_status(name,working)两个退出 IDLE 的路径inbox 有消息有人发来了指令 → 恢复 WORK 处理消息认领到任务任务板上找到了工作 → 注入任务描述恢复 WORK一个退出程序的路径3.超时12 次轮询60 秒什么都没等到 → shutdown清除线程七、完整流程走读初始状态Lead 创建了 3 个任务在.tasks/下task 1设计 API、task 2实现 API、task 3写测试。task 2 的blockedBy: [1]task 3 的blockedBy: [2]。都无 owner。Lead 调用spawn_teammate(alice, backend, 处理 backend 相关任务)。Alice 的 WORK 阶段Alice 收到 prompt “处理 backend 相关任务”开始工作。她读任务板看到 task 1 是设计 API于是开始设计。完成后调用idle。Alice 的 IDLE 阶段第 1 次轮询scan_unclaimed_tasks()返回 task 2task 1 已完成不再是 blocking。Alice 调claim_task(2, alice)认领成功。注入任务描述恢复 WORK。压缩发生在 Alice 处理 task 2 的过程中上下文超过阈值。auto_compact 触发messages 被替换为一行摘要。Alice 的 identity 在摘要中可能被省略。压缩后Alice 完成 task 2再次 idle。scan_unclaimed_tasks()返回 task 3。认领成功。此时len(messages) 3压缩的指纹所以注入 identity block “我是 Alice继续工作” task 3 的描述。Alice 虽然忘了之前的细节但知道自己是谁、该干什么。超时退出如果 Alice 在 idle 阶段等了 60 秒没有新消息也没有未认领任务状态切为shutdown线程结束。整个团队的任务都做完了。八、设计洞察8.1 拉取 vs 推送s09-s10 的任务分配是推送Lead 发消息 → teammate 被动接收。s11 加入了拉取teammate 自己扫描任务板看到就领。两种模式互补推送适合紧急、特定目标的指令“Alice这个 bug 你立刻修”拉取适合常规、非特定目标的工作“有人需要处理 backlog 里的任务”8.2 认领锁是分布式系统的最小实现claim_task的_claim_lock只是一个 Python 线程锁——在这个单进程多线程的 demo 中够用。但如果 agent 分布在多台机器上未来方向就需要数据库的行级锁或分布式锁。s11 的锁指出了一个正确方向实现可以随着规模升级。8.3 自我修复的身份压缩是不可预测的——不知道什么时候触发、摘要里会保留什么。make_identity_block和len(messages) 3的启发式检测是一个被动修复机制不阻止压缩发生但在压缩后尽快修复关键信息。这和 s06 的信息没有真正丢失只是移出了活跃上下文一脉相承。8.4 轮询间隔的工程权衡POLL_INTERVAL 5秒IDLE_TIMEOUT 60秒。为什么是 5 和 60太短1 秒空转浪费 CPU且不断读文件系统太长30 秒队友闲置太久用户以为卡住了5 秒是够快但不浪费的平衡点60 秒是给足够机会但不无限等待的平衡点这些数字没有理论最优解是在体验和效率之间的工程折中。