learn-claude-code s11 自治 Agent空闲轮询、任务自动认领与身份重注入的设计与实现【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code本文围绕 learn-claude-code 课程的 s11 章节Autonomous Agents展开讲解如何让多 Agent 团队中的队友从被动等派工转变为自主找活干队友在空闲时循环轮询收件箱与共享任务看板自动认领未被占用的任务并在上下文压缩后通过身份重注入保持我是谁的一致性。读完本文你可以理解agents/s11_autonomous_agents.py中 WORK/IDLE 双阶段循环的完整调用链并掌握在本地运行、观察自组织行为的具体步骤。从领导指派到自组织s11 解决的问题在 s09Agent Teams与 s10Team Protocols中队友只在被明确指派时才工作Lead 需要为每个队友编写 prompt任务看板上未认领的任务只能手动分配。当任务数量增多时这种中心化派工模式扩展不动——Lead 成为瓶颈任务看板上的 pending 任务越多手动分配成本越高。s11 引入真正的自治语义队友自己扫描任务看板认领没人做的任务做完再找下一个Lead 不再逐个指派。此外还有一个隐藏问题s06 的 Context Compact上下文压缩发生后队友可能忘记自己是谁——名字、角色、所属团队都会丢失。s11 用身份重注入机制解决这一点。对应源码入口为 s11_autonomous_agents.py它构建在 s10_team_protocols.py 的收件箱通信与请求-响应协议之上文件头部注释直接点出核心洞察The agent finds work itself.队友生命周期WORK 与 IDLE 双阶段s11 中每个队友运行一个持久循环状态在 WORK工作和 IDLE空闲之间切换文档中的生命周期状态机如下Teammate lifecycle with idle cycle: ------- | spawn | ------ | v ------- tool_use ------- | WORK | ------------- | LLM | ------ ------- | | stop_reason ! tool_use (or idle tool called) v -------- | IDLE | poll every 5s for up to 60s ------- | --- check inbox -- message? ---------- WORK | --- scan .tasks/ -- unclaimed? ------- claim - WORK | --- 60s timeout ---------------------- SHUTDOWN Identity re-injection after compression: if len(messages) 3: messages.insert(0, identity_block)在 TeammateManager._loop 中这个状态机的实际实现要点如下def _loop(self, name: str, role: str, prompt: str): team_name self.config[team_name] sys_prompt ( fYou are {name}, role: {role}, team: {team_name}, at {WORKDIR}. fUse idle tool when you have no more work. You will auto-claim new tasks. ) messages [{role: user, content: prompt}] tools self._teammate_tools() while True: # -- WORK PHASE: standard agent loop -- for _ in range(50): inbox BUS.read_inbox(name) for msg in inbox: if msg.get(type) shutdown_request: self._set_status(name, shutdown) return messages.append({role: user, content: json.dumps(msg)}) response client.messages.create( modelMODEL, systemsys_prompt, messagesmessages, toolstools, max_tokens8000, ) messages.append({role: assistant, content: response.content}) if response.stop_reason ! tool_use: break # ...执行工具块若模型调用了 idle 工具则置 idle_requested if idle_requested: break # -- IDLE PHASE: poll for inbox messages and unclaimed tasks -- self._set_status(name, idle) # ...轮询逻辑见下文 if not resume: self._set_status(name, shutdown) return self._set_status(name, working)这里有几个值得注意的实现细节WORK 阶段的进入条件LLM 返回stop_reason ! tool_use模型自然收工或模型主动调用了idle工具两种情况都会退出 WORK 进入 IDLE。idle工具的存在让模型可以用显式信号表达我干完了而不是等它无话可说。WORK 阶段每次迭代前先排空收件箱新消息包括shutdown_request会在下一轮 LLM 调用前注入 messages收到关机请求时立即置状态为shutdown并退出线程。WORK 循环有 50 轮上限for _ in range(50)防止单个工作阶段无限循环超限后同样进入 IDLE 重新决策。状态持久化_set_status会更新.team/config.json中该成员的status字段因此随时可以通过/team命令观察谁在 working、谁在 idle、谁已 shutdown。API 异常即退出client.messages.create抛异常时网络错误、鉴权失败等队友把状态置为idle后直接 return不会无限重试。IDLE 阶段5 秒轮询60 秒超时空闲阶段是整个自治机制的核心。源码中的两个常量定义了轮询节奏常量定义常量值含义POLL_INTERVAL5秒每 5 秒检查一次收件箱和任务板IDLE_TIMEOUT60秒空闲累计 60 秒无新活则自动关机轮询次数为IDLE_TIMEOUT // max(POLL_INTERVAL, 1)即最多 12 次。每轮轮询的优先级是先看收件箱再扫任务板——控制类消息关机、审批响应、Lead 直接指令必须先于自己找活这与 s13 章节中关机、计划审批和 Lead 的直接指令应该先于临时发现的工作的运行时语义一致。文档中给出的_idle_poll简化版本def _idle_poll(self, name, messages): for _ in range(IDLE_TIMEOUT // POLL_INTERVAL): # 60s / 5s 12 time.sleep(POLL_INTERVAL) inbox BUS.read_inbox(name) if inbox: messages.append({role: user, content: finbox{inbox}/inbox}) return True unclaimed scan_unclaimed_tasks() if unclaimed: claim_task(unclaimed[0][id], name) messages.append({role: user, content: fauto-claimedTask #{unclaimed[0][id]}: f{unclaimed[0][subject]}/auto-claimed}) return True return False # timeout - shutdown而 真实实现 内联在_loop的 IDLE 分支中比文档版本多处理了两个边界情况# -- IDLE PHASE: poll for inbox messages and unclaimed tasks -- self._set_status(name, idle) resume False polls IDLE_TIMEOUT // max(POLL_INTERVAL, 1) for _ in range(polls): time.sleep(POLL_INTERVAL) inbox BUS.read_inbox(name) if inbox: for msg in inbox: if msg.get(type) shutdown_request: self._set_status(name, shutdown) return messages.append({role: user, content: json.dumps(msg)}) resume True break unclaimed scan_unclaimed_tasks() if unclaimed: task unclaimed[0] result claim_task(task[id], name) if result.startswith(Error:): continue # 认领失败比如被别人抢先了继续轮询下一个候选 task_prompt ( fauto-claimedTask #{task[id]}: {task[subject]}\n f{task.get(description, )}/auto-claimed ) if len(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.}) resume True break if not resume: self._set_status(name, shutdown) return self._set_status(name, working)与文档版本的差异在于认领失败不终止轮询claim_task返回Error:前缀例如任务已被其他队友抢先认领时执行continue队友继续等待下一轮扫描而不是带着一条失败消息回到 WORK 阶段空转。认领成功后立即补两条消息auto-claimed任务提示 一条 Claimed task #N. Working on it. 的 assistant 回执保证 WORK 阶段从已认领这一事实出发推理。超时即关机12 轮轮询都没有新消息、没有可认领任务时状态被置为shutdown线程退出。这解决了空闲队友无限占用上下文的资源问题。任务看板扫描与认领任务看板是.tasks/目录下的task_*.json文件TASKS_DIR WORKDIR / .tasks。扫描逻辑只找满足三个条件的任务状态为pending、没有owner、且没有blockedBy前置依赖def scan_unclaimed_tasks() - list: TASKS_DIR.mkdir(exist_okTrue) unclaimed [] for f in sorted(TASKS_DIR.glob(task_*.json)): task json.loads(f.read_text()) if (task.get(status) pending and not task.get(owner) and not task.get(blockedBy)): unclaimed.append(task) return unclaimedsorted()保证多个队友看到候选列表时的顺序一致倾向于先认领编号最小的任务。blockedBy条件则实现了依赖顺序前置任务未完成仍有blockedBy记录时后续任务对所有人不可见这就是文档中创建带依赖的任务观察队友尊重 blocked 顺序这一实验的底层机制。claim_task 负责把发现候选变成确认拥有整个过程包在_claim_lockthreading.Lock中_claim_lock threading.Lock() def claim_task(task_id: int, owner: str) - str: with _claim_lock: path TASKS_DIR / ftask_{task_id}.json if not path.exists(): return fError: Task {task_id} not found task json.loads(path.read_text()) if existing_owner : task.get(owner): return fError: Task {task_id} has already been claimed by {existing_owner} if (status : task.get(status)) ! pending: return fError: Task {task_id} cannot be claimed because its status is {status} if task.get(blockedBy): return fError: Task {task_id} is blocked by other task(s) and cannot be claimed yet task[owner] owner task[status] in_progress path.write_text(json.dumps(task, indent2)) return fClaimed task #{task_id} for {owner}设计上有三点值得注意扫描与认领是两步以认领时的校验为准。scan_unclaimed_tasks返回的候选只是某一时刻的快照多个队友线程可能同时看到同一个任务claim_task在锁内重新读取文件并复核owner/status/blockedBy只有第一个通过复核的调用才能把任务推进到in_progress其余调用拿到Error: ... has already been claimed by ...。锁的适用范围由于 s11 中所有队友都是同一进程内的threading.Thread进程内的threading.Lock足以保证互斥。这一点与后续 s13 章节不同——那里用文件锁把互斥扩展到跨进程。认领同时是状态迁移pending - in_progress且写入owner/tasks命令输出中的[]标记和owner后缀都来自这两个字段。身份重注入压缩后不忘本s06 的 Context Compact 会把长上下文压缩成摘要压缩后消息历史可能只剩寥寥几条。对 Lead 而言这不是问题用户输入还在但对队友是致命的——它的系统提示词里的名字和角色可能不再被记起甚至不知道自己属于哪个团队。s11 的解法是一个极简启发式消息列表过短说明发生过压缩时在开头插入身份块。独立的工厂函数是 make_identity_blockdef make_identity_block(name: str, role: str, team_name: str) - dict: return { role: user, content: fidentityYou are {name}, role: {role}, team: {team_name}. Continue your work./identity, }在 IDLE 阶段成功认领新任务时触发_loop 中的判断if len(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意味着历史中最多只剩初始 prompt 加一两条记录此时插入两条消息user 角色的identity声明和一条 I am {name}. Continuing. 的 assistant 回应。之所以用认领任务这个时机做检查而不是每轮 LLM 调用前都检查从源码结构看是出于成本考虑——每轮注入身份块会白白消耗 token而在重新开工的节点注入一次正好覆盖压缩后的失忆窗口。工具集变更12 到 14 个工具文档给出了相对 s10 的完整变更对照表组件之前 (s10)之后 (s11)Tools1214 (idle, claim_task)自治性领导指派自组织空闲阶段无轮询收件箱 任务看板任务认领仅手动自动认领未分配任务身份系统提示 压缩后重注入超时无60 秒空闲 - 自动关机对照源码可以验证这个计数s10 的 TOOL_HANDLERS 恰好是 12 个条目s11 的 TOOL_HANDLERS 在其基础上增加了idle和claim_task共 14 个工具说明bash/read_file/write_file/edit_file基础文件与 Shell 工具自 s02 起保持不变spawn_teammate生成或复用一个自治队友启动独立线程list_teammates列出所有队友及当前 statussend_message/read_inbox/broadcast基于.team/inbox/*.jsonl的收件箱通信shutdown_request/shutdown_responses10 的关机握手协议request_id 关联plan_approvals10 的计划审批协议request_id 关联idle新增模型显式宣告没有更多活进入空闲轮询对 Lead 调用返回 Lead does not idle.claim_task新增按 ID 认领任务Lead 也可以用来认领owner 记为lead队友侧的工具集_teammate_tools共 10 个是 Lead 工具集的子集加上idle与claim_task基础 4 个文件工具、send_message、read_inbox、shutdown_response、plan_approval、idle、claim_task。队友不能spawn_teammate或broadcast生成权仍集中在 Lead。运行与实验运行 s11 需要两个环境变量源码在模块加载时读取见 环境初始化MODEL_ID必填指定模型 IDANTHROPIC_BASE_URL可选指向兼容 Anthropic Messages API 的服务设置后会显式移除ANTHROPIC_AUTH_TOKEN由 API Key 走标准的ANTHROPIC_API_KEY通道。依赖见 requirements.txtanthropic0.25.0、python-dotenv1.0.0、pyyaml6.0。启动命令来自文档的试一试章节cd learn-claude-code python agents/s11_autonomous_agents.pyCLI 提示符为s11 支持三条内置命令主循环处理/team列出团队名、每个队友的角色与状态working / idle / shutdown/inbox查看 Lead 自己的收件箱不排空clearFalse/tasks查看.tasks/看板[ ]表示 pending、[]表示 in_progress、[x]表示 completed并显示owner。输入q、exit或空行退出。文档推荐的实验 prompt英文对 LLM 效果更好也可以用中文Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim.Spawn a coder teammate and let it find work from the task board itselfCreate tasks with dependencies. Watch teammates respect the blocked order.输入/tasks查看带 owner 的任务看板输入/team监控谁在工作、谁在空闲实验中可以直接观察磁盘上的状态变化.team/config.json里每个成员的status字段随 WORK/IDLE/SHUTDOWN 切换.team/inbox/name.jsonl逐行追加消息.tasks/task_*.json的owner与status在认领瞬间被写入。关键设计要点小结自治不靠新循环靠新入口队友发现任务不需要第二个 Agent Loop——认领成功后只是往同一个 messages 里追加auto-claimed提示复用既有 WORK 循环执行。优先级由轮询顺序保证IDLE 每轮先读收件箱再扫任务板关机与直接指令天然优先于自主找活。发现与确认分离scan_unclaimed_tasks提供有序候选claim_task在锁内二次校验并原子落盘配合失败后continue的轮询策略容忍多队友竞争同一任务。身份是运行时维护的名字、角色、团队写进系统提示词还不够压缩事件会清空模型对自身的记忆len(messages) 3时的身份重注入是最低成本的补救。空闲是有代价边界的POLL_INTERVAL5s/IDLE_TIMEOUT60s两个常量同时决定了响应延迟最坏 5 秒发现新活与资源回收60 秒无活自动关机按需调整即可改变团队的行为节奏。本章是课程后半程Agent Teams 方向的过渡点s11 用线程 文件收件箱 任务 JSON实现了最小可运行的自治团队后续章节如s13_agent_teams/的完整团队运行时在此基础上把收件箱投递、任务认领、工作目录隔离与协议状态进一步工程化可以对照阅读 docs/zh/s10-team-protocols.md 回顾 s10 的协议基础以及 agents/s11_autonomous_agents.py 全文核对本文引用的每一处实现。【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 learn-claude-code s11 自治 Agent：空闲轮询、任务自动认领与身份重注入的设计与实现
learn-claude-code s11 自治 Agent空闲轮询、任务自动认领与身份重注入的设计与实现【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code本文围绕 learn-claude-code 课程的 s11 章节Autonomous Agents展开讲解如何让多 Agent 团队中的队友从被动等派工转变为自主找活干队友在空闲时循环轮询收件箱与共享任务看板自动认领未被占用的任务并在上下文压缩后通过身份重注入保持我是谁的一致性。读完本文你可以理解agents/s11_autonomous_agents.py中 WORK/IDLE 双阶段循环的完整调用链并掌握在本地运行、观察自组织行为的具体步骤。从领导指派到自组织s11 解决的问题在 s09Agent Teams与 s10Team Protocols中队友只在被明确指派时才工作Lead 需要为每个队友编写 prompt任务看板上未认领的任务只能手动分配。当任务数量增多时这种中心化派工模式扩展不动——Lead 成为瓶颈任务看板上的 pending 任务越多手动分配成本越高。s11 引入真正的自治语义队友自己扫描任务看板认领没人做的任务做完再找下一个Lead 不再逐个指派。此外还有一个隐藏问题s06 的 Context Compact上下文压缩发生后队友可能忘记自己是谁——名字、角色、所属团队都会丢失。s11 用身份重注入机制解决这一点。对应源码入口为 s11_autonomous_agents.py它构建在 s10_team_protocols.py 的收件箱通信与请求-响应协议之上文件头部注释直接点出核心洞察The agent finds work itself.队友生命周期WORK 与 IDLE 双阶段s11 中每个队友运行一个持久循环状态在 WORK工作和 IDLE空闲之间切换文档中的生命周期状态机如下Teammate lifecycle with idle cycle: ------- | spawn | ------ | v ------- tool_use ------- | WORK | ------------- | LLM | ------ ------- | | stop_reason ! tool_use (or idle tool called) v -------- | IDLE | poll every 5s for up to 60s ------- | --- check inbox -- message? ---------- WORK | --- scan .tasks/ -- unclaimed? ------- claim - WORK | --- 60s timeout ---------------------- SHUTDOWN Identity re-injection after compression: if len(messages) 3: messages.insert(0, identity_block)在 TeammateManager._loop 中这个状态机的实际实现要点如下def _loop(self, name: str, role: str, prompt: str): team_name self.config[team_name] sys_prompt ( fYou are {name}, role: {role}, team: {team_name}, at {WORKDIR}. fUse idle tool when you have no more work. You will auto-claim new tasks. ) messages [{role: user, content: prompt}] tools self._teammate_tools() while True: # -- WORK PHASE: standard agent loop -- for _ in range(50): inbox BUS.read_inbox(name) for msg in inbox: if msg.get(type) shutdown_request: self._set_status(name, shutdown) return messages.append({role: user, content: json.dumps(msg)}) response client.messages.create( modelMODEL, systemsys_prompt, messagesmessages, toolstools, max_tokens8000, ) messages.append({role: assistant, content: response.content}) if response.stop_reason ! tool_use: break # ...执行工具块若模型调用了 idle 工具则置 idle_requested if idle_requested: break # -- IDLE PHASE: poll for inbox messages and unclaimed tasks -- self._set_status(name, idle) # ...轮询逻辑见下文 if not resume: self._set_status(name, shutdown) return self._set_status(name, working)这里有几个值得注意的实现细节WORK 阶段的进入条件LLM 返回stop_reason ! tool_use模型自然收工或模型主动调用了idle工具两种情况都会退出 WORK 进入 IDLE。idle工具的存在让模型可以用显式信号表达我干完了而不是等它无话可说。WORK 阶段每次迭代前先排空收件箱新消息包括shutdown_request会在下一轮 LLM 调用前注入 messages收到关机请求时立即置状态为shutdown并退出线程。WORK 循环有 50 轮上限for _ in range(50)防止单个工作阶段无限循环超限后同样进入 IDLE 重新决策。状态持久化_set_status会更新.team/config.json中该成员的status字段因此随时可以通过/team命令观察谁在 working、谁在 idle、谁已 shutdown。API 异常即退出client.messages.create抛异常时网络错误、鉴权失败等队友把状态置为idle后直接 return不会无限重试。IDLE 阶段5 秒轮询60 秒超时空闲阶段是整个自治机制的核心。源码中的两个常量定义了轮询节奏常量定义常量值含义POLL_INTERVAL5秒每 5 秒检查一次收件箱和任务板IDLE_TIMEOUT60秒空闲累计 60 秒无新活则自动关机轮询次数为IDLE_TIMEOUT // max(POLL_INTERVAL, 1)即最多 12 次。每轮轮询的优先级是先看收件箱再扫任务板——控制类消息关机、审批响应、Lead 直接指令必须先于自己找活这与 s13 章节中关机、计划审批和 Lead 的直接指令应该先于临时发现的工作的运行时语义一致。文档中给出的_idle_poll简化版本def _idle_poll(self, name, messages): for _ in range(IDLE_TIMEOUT // POLL_INTERVAL): # 60s / 5s 12 time.sleep(POLL_INTERVAL) inbox BUS.read_inbox(name) if inbox: messages.append({role: user, content: finbox{inbox}/inbox}) return True unclaimed scan_unclaimed_tasks() if unclaimed: claim_task(unclaimed[0][id], name) messages.append({role: user, content: fauto-claimedTask #{unclaimed[0][id]}: f{unclaimed[0][subject]}/auto-claimed}) return True return False # timeout - shutdown而 真实实现 内联在_loop的 IDLE 分支中比文档版本多处理了两个边界情况# -- IDLE PHASE: poll for inbox messages and unclaimed tasks -- self._set_status(name, idle) resume False polls IDLE_TIMEOUT // max(POLL_INTERVAL, 1) for _ in range(polls): time.sleep(POLL_INTERVAL) inbox BUS.read_inbox(name) if inbox: for msg in inbox: if msg.get(type) shutdown_request: self._set_status(name, shutdown) return messages.append({role: user, content: json.dumps(msg)}) resume True break unclaimed scan_unclaimed_tasks() if unclaimed: task unclaimed[0] result claim_task(task[id], name) if result.startswith(Error:): continue # 认领失败比如被别人抢先了继续轮询下一个候选 task_prompt ( fauto-claimedTask #{task[id]}: {task[subject]}\n f{task.get(description, )}/auto-claimed ) if len(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.}) resume True break if not resume: self._set_status(name, shutdown) return self._set_status(name, working)与文档版本的差异在于认领失败不终止轮询claim_task返回Error:前缀例如任务已被其他队友抢先认领时执行continue队友继续等待下一轮扫描而不是带着一条失败消息回到 WORK 阶段空转。认领成功后立即补两条消息auto-claimed任务提示 一条 Claimed task #N. Working on it. 的 assistant 回执保证 WORK 阶段从已认领这一事实出发推理。超时即关机12 轮轮询都没有新消息、没有可认领任务时状态被置为shutdown线程退出。这解决了空闲队友无限占用上下文的资源问题。任务看板扫描与认领任务看板是.tasks/目录下的task_*.json文件TASKS_DIR WORKDIR / .tasks。扫描逻辑只找满足三个条件的任务状态为pending、没有owner、且没有blockedBy前置依赖def scan_unclaimed_tasks() - list: TASKS_DIR.mkdir(exist_okTrue) unclaimed [] for f in sorted(TASKS_DIR.glob(task_*.json)): task json.loads(f.read_text()) if (task.get(status) pending and not task.get(owner) and not task.get(blockedBy)): unclaimed.append(task) return unclaimedsorted()保证多个队友看到候选列表时的顺序一致倾向于先认领编号最小的任务。blockedBy条件则实现了依赖顺序前置任务未完成仍有blockedBy记录时后续任务对所有人不可见这就是文档中创建带依赖的任务观察队友尊重 blocked 顺序这一实验的底层机制。claim_task 负责把发现候选变成确认拥有整个过程包在_claim_lockthreading.Lock中_claim_lock threading.Lock() def claim_task(task_id: int, owner: str) - str: with _claim_lock: path TASKS_DIR / ftask_{task_id}.json if not path.exists(): return fError: Task {task_id} not found task json.loads(path.read_text()) if existing_owner : task.get(owner): return fError: Task {task_id} has already been claimed by {existing_owner} if (status : task.get(status)) ! pending: return fError: Task {task_id} cannot be claimed because its status is {status} if task.get(blockedBy): return fError: Task {task_id} is blocked by other task(s) and cannot be claimed yet task[owner] owner task[status] in_progress path.write_text(json.dumps(task, indent2)) return fClaimed task #{task_id} for {owner}设计上有三点值得注意扫描与认领是两步以认领时的校验为准。scan_unclaimed_tasks返回的候选只是某一时刻的快照多个队友线程可能同时看到同一个任务claim_task在锁内重新读取文件并复核owner/status/blockedBy只有第一个通过复核的调用才能把任务推进到in_progress其余调用拿到Error: ... has already been claimed by ...。锁的适用范围由于 s11 中所有队友都是同一进程内的threading.Thread进程内的threading.Lock足以保证互斥。这一点与后续 s13 章节不同——那里用文件锁把互斥扩展到跨进程。认领同时是状态迁移pending - in_progress且写入owner/tasks命令输出中的[]标记和owner后缀都来自这两个字段。身份重注入压缩后不忘本s06 的 Context Compact 会把长上下文压缩成摘要压缩后消息历史可能只剩寥寥几条。对 Lead 而言这不是问题用户输入还在但对队友是致命的——它的系统提示词里的名字和角色可能不再被记起甚至不知道自己属于哪个团队。s11 的解法是一个极简启发式消息列表过短说明发生过压缩时在开头插入身份块。独立的工厂函数是 make_identity_blockdef make_identity_block(name: str, role: str, team_name: str) - dict: return { role: user, content: fidentityYou are {name}, role: {role}, team: {team_name}. Continue your work./identity, }在 IDLE 阶段成功认领新任务时触发_loop 中的判断if len(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意味着历史中最多只剩初始 prompt 加一两条记录此时插入两条消息user 角色的identity声明和一条 I am {name}. Continuing. 的 assistant 回应。之所以用认领任务这个时机做检查而不是每轮 LLM 调用前都检查从源码结构看是出于成本考虑——每轮注入身份块会白白消耗 token而在重新开工的节点注入一次正好覆盖压缩后的失忆窗口。工具集变更12 到 14 个工具文档给出了相对 s10 的完整变更对照表组件之前 (s10)之后 (s11)Tools1214 (idle, claim_task)自治性领导指派自组织空闲阶段无轮询收件箱 任务看板任务认领仅手动自动认领未分配任务身份系统提示 压缩后重注入超时无60 秒空闲 - 自动关机对照源码可以验证这个计数s10 的 TOOL_HANDLERS 恰好是 12 个条目s11 的 TOOL_HANDLERS 在其基础上增加了idle和claim_task共 14 个工具说明bash/read_file/write_file/edit_file基础文件与 Shell 工具自 s02 起保持不变spawn_teammate生成或复用一个自治队友启动独立线程list_teammates列出所有队友及当前 statussend_message/read_inbox/broadcast基于.team/inbox/*.jsonl的收件箱通信shutdown_request/shutdown_responses10 的关机握手协议request_id 关联plan_approvals10 的计划审批协议request_id 关联idle新增模型显式宣告没有更多活进入空闲轮询对 Lead 调用返回 Lead does not idle.claim_task新增按 ID 认领任务Lead 也可以用来认领owner 记为lead队友侧的工具集_teammate_tools共 10 个是 Lead 工具集的子集加上idle与claim_task基础 4 个文件工具、send_message、read_inbox、shutdown_response、plan_approval、idle、claim_task。队友不能spawn_teammate或broadcast生成权仍集中在 Lead。运行与实验运行 s11 需要两个环境变量源码在模块加载时读取见 环境初始化MODEL_ID必填指定模型 IDANTHROPIC_BASE_URL可选指向兼容 Anthropic Messages API 的服务设置后会显式移除ANTHROPIC_AUTH_TOKEN由 API Key 走标准的ANTHROPIC_API_KEY通道。依赖见 requirements.txtanthropic0.25.0、python-dotenv1.0.0、pyyaml6.0。启动命令来自文档的试一试章节cd learn-claude-code python agents/s11_autonomous_agents.pyCLI 提示符为s11 支持三条内置命令主循环处理/team列出团队名、每个队友的角色与状态working / idle / shutdown/inbox查看 Lead 自己的收件箱不排空clearFalse/tasks查看.tasks/看板[ ]表示 pending、[]表示 in_progress、[x]表示 completed并显示owner。输入q、exit或空行退出。文档推荐的实验 prompt英文对 LLM 效果更好也可以用中文Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim.Spawn a coder teammate and let it find work from the task board itselfCreate tasks with dependencies. Watch teammates respect the blocked order.输入/tasks查看带 owner 的任务看板输入/team监控谁在工作、谁在空闲实验中可以直接观察磁盘上的状态变化.team/config.json里每个成员的status字段随 WORK/IDLE/SHUTDOWN 切换.team/inbox/name.jsonl逐行追加消息.tasks/task_*.json的owner与status在认领瞬间被写入。关键设计要点小结自治不靠新循环靠新入口队友发现任务不需要第二个 Agent Loop——认领成功后只是往同一个 messages 里追加auto-claimed提示复用既有 WORK 循环执行。优先级由轮询顺序保证IDLE 每轮先读收件箱再扫任务板关机与直接指令天然优先于自主找活。发现与确认分离scan_unclaimed_tasks提供有序候选claim_task在锁内二次校验并原子落盘配合失败后continue的轮询策略容忍多队友竞争同一任务。身份是运行时维护的名字、角色、团队写进系统提示词还不够压缩事件会清空模型对自身的记忆len(messages) 3时的身份重注入是最低成本的补救。空闲是有代价边界的POLL_INTERVAL5s/IDLE_TIMEOUT60s两个常量同时决定了响应延迟最坏 5 秒发现新活与资源回收60 秒无活自动关机按需调整即可改变团队的行为节奏。本章是课程后半程Agent Teams 方向的过渡点s11 用线程 文件收件箱 任务 JSON实现了最小可运行的自治团队后续章节如s13_agent_teams/的完整团队运行时在此基础上把收件箱投递、任务认领、工作目录隔离与协议状态进一步工程化可以对照阅读 docs/zh/s10-team-protocols.md 回顾 s10 的协议基础以及 agents/s11_autonomous_agents.py 全文核对本文引用的每一处实现。【免费下载链接】learn-claude-codeBash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1项目地址: https://gitcode.com/GitHub_Trending/an/learn-claude-code创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考