learn-claude-codes11 自主 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 章节docs/ja/s11-autonomous-agents.md讲解如何让多 Agent 团队中的成员从等指令干活升级为自己找活干。读完你可以掌握三件事WORK/IDLE 双阶段队友循环的设计与实现、基于.tasks/任务板的无锁竞争式自动领取auto-claim以及上下文压缩后通过身份块identity block防止 Agent 失忆的再注入技巧。一、问题Leader 逐一派活的模式无法规模化在 s09Agent 团队与 s10团队协议中团队成员只在被显式指示时才工作Leader 必须为每个具体任务 spawn 一个带专属 prompt 的队友。任务板上有 10 个未分配任务时也得由 Leader 手动逐一指派——这在任务规模增长时完全不可行。真正的自主性Autonomy指的是队友自己发现工作——扫描任务板、领取未占据的任务、执行完成后继续寻找下一个任务。原文档将其概括为队友自己看看板、自己领活——Leader 不需要逐一分配。这是 s11 在整个课程序列s01 agent 循环 → s02 工具使用 → … → s10 团队协议 →[s11 自主 Agent]→ s12中承担的核心职责把协作模式从Leader 指挥切换为自组织。另一个连带问题前序章节的上下文压缩机制compact可能让 Agent 丢失自己的身份信息我是谁、扮演什么角色、属于哪个团队。s11 用身份再注入identity re-injection解决它。二、团队员生命周期WORK / IDLE 双阶段模型s11 给每个队友定义了一个带空闲循环idle cycle的生命周期原文档中的流程示意如下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核心常量定义在 agents/s11_autonomous_agents.pyPOLL_INTERVAL 5 # 空闲期每 5 秒轮询一次 IDLE_TIMEOUT 60 # 连续 60 秒无工作则自动 shutdown即空闲阶段最多轮询60 / 5 12次每次轮询做两件事检查收件箱是否有新消息、扫描任务板是否有可领取的任务。两者皆无且超时则队友进入 SHUTDOWN 状态并退出线程。三、源码拆解队友主循环_loop实现入口在 agents/s11_autonomous_agents.py 的 TeammateManager._loop。其结构与原文档给出的骨架一致但源码中有几个值得注意的细节3.1 WORK 阶段标准 agent 循环 idle工具信号WORK 阶段就是标准的 LLM 工具循环限制最多 50 轮while True: # -- WORK PHASE: standard agent loop -- for _ in range(50): inbox BUS.read_inbox(name) # 处理收件箱消息shutdown_request 直接退出 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 # 执行工具调用 if idle_requested: break退出 WORK 阶段有两个触发条件与文档描述一致response.stop_reason ! tool_use——LLM 自然结束了一轮对话LLM 主动调用了idle工具表示我这边没有更多工作了。此时工具返回Entering idle phase. Will poll for new tasks.循环立即跳出进入空闲期。3.2 IDLE 阶段收件箱优先任务板兜底对应源码 agents/s11_autonomous_agents.py#L266-L302# -- IDLE PHASE: poll for inbox messages and unclaimed tasks -- self._set_status(name, idle) resume False polls IDLE_TIMEOUT // max(POLL_INTERVAL, 1) # 60 / 5 12 for _ in range(polls): time.sleep(POLL_INTERVAL) inbox BUS.read_inbox(name) if inbox: # shutdown_request 直接置为 shutdown 并退出 # 否则把消息追加进上下文恢复 WORK resume True break unclaimed scan_unclaimed_tasks() if unclaimed: task unclaimed[0] result claim_task(task[id], name) if result.startswith(Error:): continue # 被别的队友抢走了继续轮询下一个 # 注入 auto-claimed 任务 prompt恢复 WORK resume True break if not resume: self._set_status(name, shutdown) return self._set_status(name, working)注意claim_task失败时执行continue而不是退出如果第一个未领取任务刚被另一个队友抢走本轮轮询会继续尝试下一个可领取任务。状态变更working/idle/shutdown会实时写入.team/config.json这是后文/team命令能观测到每个成员状态的基础。3.3 任务板扫描scan_unclaimed_tasks扫描逻辑见 agents/s11_autonomous_agents.py#L128-L137条件与原文档完全一致——三个条件同时满足才算可领取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 unclaimedstatus pending任务尚未开始无owner还没有人领取无blockedBy没有未完成的依赖任务——这让队友天然尊重阻塞顺序依赖未满足的任务不会出现在扫描结果中。任务文件存储于工作目录下的.tasks/task_*.jsonsorted(glob)保证领取顺序稳定按任务 ID 排序先到先得。3.4 任务领取claim_task与互斥锁agents/s11_autonomous_agents.py#L140-L155 中的claim_task用threading.Lock模块级_claim_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}要点领取是检查—写入原子操作被抢的任务返回带Error:前缀的字符串调用方IDLE 轮询据此continue跳过。这一语义在仓库测试 tests/test_agent_teams_runtime.py 中有系统性覆盖如test_idle_claim_is_atomic_across_teammates验证了跨队友领取的原子性test_teammate_emits_result_then_idle_and_shuts_down验证了产出结果 → 空闲 → 关闭的完整生命周期tests/test_task_system.py 则覆盖了依赖未满足时claim_task返回Blocked by: [...]的行为。3.5 身份再注入压缩后不忘本上下文压缩s06会截断历史消息可能导致 Agent 忘记自己是alice、角色是frontend。s11 的对策是在恢复 WORK 前检查消息长度过短说明压缩发生过此时把身份块插到对话最前面。源码见 agents/s11_autonomous_agents.py#L158-L163 与主循环中的调用点def 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, } # 在 _loop 的 IDLE 阶段领取到新任务后 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最前索引 0 为用户消息、索引 1 为确认性 assistant 消息随后才追加auto-claimed任务 prompt。这样即使历史被压掉身份声明也始终处于上下文开头。四、工具面Leader 14 工具 vs 队友 10 工具原文档的对比表是 s10 → s11 的核心增量这里完整给出ComponentBefore (s10)After (s11)Tools1214idleclaim_taskAutonomyLead-directedSelf-organizingIdle phaseNonePoll inbox task boardTask claimingManual onlyAuto-claim unclaimed tasksIdentitySystem prompt re-injection after compressTimeoutNone60s idle - auto shutdown对应到源码工具定义分两套Leader 侧TOOL_HANDLERS 共 14 个——基础文件工具 4 个bash、read_file、write_file、edit_file与 s02 相同 团队工具 6 个spawn_teammate、list_teammates、send_message、read_inbox、broadcast、shutdown_request 协议工具 2 个shutdown_response、plan_approval s11 新增 2 个idle、claim_task。其中 Leader 的idle返回Lead does not idle.——Leader 本身不进入空闲轮询。队友侧_teammate_tools 共 10 个同样包含 s11 新增的idle与claim_task但不含spawn_teammate/broadcast等 Leader 专属能力。send_message的msg_type取值受 VALID_MSG_TYPES 约束message、broadcast、shutdown_request、shutdown_response、plan_approval_response。消息传递由 MessageBus 实现每个成员在.team/inbox/下有一个 JSONL 文件如alice.jsonlsend追加写入、read_inbox读取后清空drain 语义broadcast向除发送者外的所有成员逐条投递。五、动手运行5.1 环境与启动从源码头部 agents/s11_autonomous_agents.py#L46-L63 可以确认运行前置条件from anthropic import Anthropic from dotenv import load_dotenv load_dotenv(overrideTrue) if os.getenv(ANTHROPIC_BASE_URL): os.environ.pop(ANTHROPIC_AUTH_TOKEN, None) WORKDIR Path.cwd() client Anthropic(base_urlos.getenv(ANTHROPIC_BASE_URL)) MODEL os.environ[MODEL_ID] TEAM_DIR WORKDIR / .team TASKS_DIR WORKDIR / .tasks即需要Python 依赖anthropic、python-dotenv见仓库 requirements.txt环境变量MODEL_ID必需指定要调用的模型 ID可选ANTHROPIC_BASE_URL配置兼容网关时生效设置后会自动移除ANTHROPIC_AUTH_TOKEN避免冲突在仓库根目录启动.team/与.tasks/会按当前工作目录创建。cd learn-claude-code python agents/s11_autonomous_agents.py5.2 观察自主行为的推荐操作原文档给出的五条试验 prompt 与建议操作如下Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim.—— 创建 3 个任务后 spawn 两名队友观察他们自动领取任务Spawn a coder teammate and let it find work from the task board itself—— spawn 一个 coder 角色队友让它自己从任务板找活Create tasks with dependencies. Watch teammates respect the blocked order.—— 创建带依赖关系的任务观察队友遵守 blockedBy 顺序输入/tasks查看任务板——每个任务会带状态标记[ ]pending、[]in_progress、[x]completed与当前 owner输入/team实时查看每个队友是working还是idle。交互入口还有/inbox查看 Leader 收件箱源码入口 使用非清空方式读取输入q/exit或空行退出。/tasks的渲染逻辑见 agents/s11_autonomous_agents.py#L570-L577。六、小结与后续章节s11 的本质是把任务分配从 Leader 的显式行为下沉为一种文件系统层面的自组织协议队友之间不互相通信就能协调领取靠的是.tasks/中的 JSON 任务状态 线程锁保证的原子claim而 60 秒空闲超时让团队规模可以按需伸缩——没有活的队友自动退出不占用 LLM 调用。身份再注入则补上了压缩机制在长期运行 Agent场景下的最后一块拼图。在本仓库的课程结构中这套自主协作机制后续还会被 s12worktree 任务隔离继续扩展agents/目录下各章节实现可对照阅读agents/s10_team_protocols.py前一章、agents/s12_worktree_task_isolation.py后一章课程总入口见 README.md。【免费下载链接】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-codes11 自主 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 章节docs/ja/s11-autonomous-agents.md讲解如何让多 Agent 团队中的成员从等指令干活升级为自己找活干。读完你可以掌握三件事WORK/IDLE 双阶段队友循环的设计与实现、基于.tasks/任务板的无锁竞争式自动领取auto-claim以及上下文压缩后通过身份块identity block防止 Agent 失忆的再注入技巧。一、问题Leader 逐一派活的模式无法规模化在 s09Agent 团队与 s10团队协议中团队成员只在被显式指示时才工作Leader 必须为每个具体任务 spawn 一个带专属 prompt 的队友。任务板上有 10 个未分配任务时也得由 Leader 手动逐一指派——这在任务规模增长时完全不可行。真正的自主性Autonomy指的是队友自己发现工作——扫描任务板、领取未占据的任务、执行完成后继续寻找下一个任务。原文档将其概括为队友自己看看板、自己领活——Leader 不需要逐一分配。这是 s11 在整个课程序列s01 agent 循环 → s02 工具使用 → … → s10 团队协议 →[s11 自主 Agent]→ s12中承担的核心职责把协作模式从Leader 指挥切换为自组织。另一个连带问题前序章节的上下文压缩机制compact可能让 Agent 丢失自己的身份信息我是谁、扮演什么角色、属于哪个团队。s11 用身份再注入identity re-injection解决它。二、团队员生命周期WORK / IDLE 双阶段模型s11 给每个队友定义了一个带空闲循环idle cycle的生命周期原文档中的流程示意如下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核心常量定义在 agents/s11_autonomous_agents.pyPOLL_INTERVAL 5 # 空闲期每 5 秒轮询一次 IDLE_TIMEOUT 60 # 连续 60 秒无工作则自动 shutdown即空闲阶段最多轮询60 / 5 12次每次轮询做两件事检查收件箱是否有新消息、扫描任务板是否有可领取的任务。两者皆无且超时则队友进入 SHUTDOWN 状态并退出线程。三、源码拆解队友主循环_loop实现入口在 agents/s11_autonomous_agents.py 的 TeammateManager._loop。其结构与原文档给出的骨架一致但源码中有几个值得注意的细节3.1 WORK 阶段标准 agent 循环 idle工具信号WORK 阶段就是标准的 LLM 工具循环限制最多 50 轮while True: # -- WORK PHASE: standard agent loop -- for _ in range(50): inbox BUS.read_inbox(name) # 处理收件箱消息shutdown_request 直接退出 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 # 执行工具调用 if idle_requested: break退出 WORK 阶段有两个触发条件与文档描述一致response.stop_reason ! tool_use——LLM 自然结束了一轮对话LLM 主动调用了idle工具表示我这边没有更多工作了。此时工具返回Entering idle phase. Will poll for new tasks.循环立即跳出进入空闲期。3.2 IDLE 阶段收件箱优先任务板兜底对应源码 agents/s11_autonomous_agents.py#L266-L302# -- IDLE PHASE: poll for inbox messages and unclaimed tasks -- self._set_status(name, idle) resume False polls IDLE_TIMEOUT // max(POLL_INTERVAL, 1) # 60 / 5 12 for _ in range(polls): time.sleep(POLL_INTERVAL) inbox BUS.read_inbox(name) if inbox: # shutdown_request 直接置为 shutdown 并退出 # 否则把消息追加进上下文恢复 WORK resume True break unclaimed scan_unclaimed_tasks() if unclaimed: task unclaimed[0] result claim_task(task[id], name) if result.startswith(Error:): continue # 被别的队友抢走了继续轮询下一个 # 注入 auto-claimed 任务 prompt恢复 WORK resume True break if not resume: self._set_status(name, shutdown) return self._set_status(name, working)注意claim_task失败时执行continue而不是退出如果第一个未领取任务刚被另一个队友抢走本轮轮询会继续尝试下一个可领取任务。状态变更working/idle/shutdown会实时写入.team/config.json这是后文/team命令能观测到每个成员状态的基础。3.3 任务板扫描scan_unclaimed_tasks扫描逻辑见 agents/s11_autonomous_agents.py#L128-L137条件与原文档完全一致——三个条件同时满足才算可领取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 unclaimedstatus pending任务尚未开始无owner还没有人领取无blockedBy没有未完成的依赖任务——这让队友天然尊重阻塞顺序依赖未满足的任务不会出现在扫描结果中。任务文件存储于工作目录下的.tasks/task_*.jsonsorted(glob)保证领取顺序稳定按任务 ID 排序先到先得。3.4 任务领取claim_task与互斥锁agents/s11_autonomous_agents.py#L140-L155 中的claim_task用threading.Lock模块级_claim_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}要点领取是检查—写入原子操作被抢的任务返回带Error:前缀的字符串调用方IDLE 轮询据此continue跳过。这一语义在仓库测试 tests/test_agent_teams_runtime.py 中有系统性覆盖如test_idle_claim_is_atomic_across_teammates验证了跨队友领取的原子性test_teammate_emits_result_then_idle_and_shuts_down验证了产出结果 → 空闲 → 关闭的完整生命周期tests/test_task_system.py 则覆盖了依赖未满足时claim_task返回Blocked by: [...]的行为。3.5 身份再注入压缩后不忘本上下文压缩s06会截断历史消息可能导致 Agent 忘记自己是alice、角色是frontend。s11 的对策是在恢复 WORK 前检查消息长度过短说明压缩发生过此时把身份块插到对话最前面。源码见 agents/s11_autonomous_agents.py#L158-L163 与主循环中的调用点def 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, } # 在 _loop 的 IDLE 阶段领取到新任务后 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最前索引 0 为用户消息、索引 1 为确认性 assistant 消息随后才追加auto-claimed任务 prompt。这样即使历史被压掉身份声明也始终处于上下文开头。四、工具面Leader 14 工具 vs 队友 10 工具原文档的对比表是 s10 → s11 的核心增量这里完整给出ComponentBefore (s10)After (s11)Tools1214idleclaim_taskAutonomyLead-directedSelf-organizingIdle phaseNonePoll inbox task boardTask claimingManual onlyAuto-claim unclaimed tasksIdentitySystem prompt re-injection after compressTimeoutNone60s idle - auto shutdown对应到源码工具定义分两套Leader 侧TOOL_HANDLERS 共 14 个——基础文件工具 4 个bash、read_file、write_file、edit_file与 s02 相同 团队工具 6 个spawn_teammate、list_teammates、send_message、read_inbox、broadcast、shutdown_request 协议工具 2 个shutdown_response、plan_approval s11 新增 2 个idle、claim_task。其中 Leader 的idle返回Lead does not idle.——Leader 本身不进入空闲轮询。队友侧_teammate_tools 共 10 个同样包含 s11 新增的idle与claim_task但不含spawn_teammate/broadcast等 Leader 专属能力。send_message的msg_type取值受 VALID_MSG_TYPES 约束message、broadcast、shutdown_request、shutdown_response、plan_approval_response。消息传递由 MessageBus 实现每个成员在.team/inbox/下有一个 JSONL 文件如alice.jsonlsend追加写入、read_inbox读取后清空drain 语义broadcast向除发送者外的所有成员逐条投递。五、动手运行5.1 环境与启动从源码头部 agents/s11_autonomous_agents.py#L46-L63 可以确认运行前置条件from anthropic import Anthropic from dotenv import load_dotenv load_dotenv(overrideTrue) if os.getenv(ANTHROPIC_BASE_URL): os.environ.pop(ANTHROPIC_AUTH_TOKEN, None) WORKDIR Path.cwd() client Anthropic(base_urlos.getenv(ANTHROPIC_BASE_URL)) MODEL os.environ[MODEL_ID] TEAM_DIR WORKDIR / .team TASKS_DIR WORKDIR / .tasks即需要Python 依赖anthropic、python-dotenv见仓库 requirements.txt环境变量MODEL_ID必需指定要调用的模型 ID可选ANTHROPIC_BASE_URL配置兼容网关时生效设置后会自动移除ANTHROPIC_AUTH_TOKEN避免冲突在仓库根目录启动.team/与.tasks/会按当前工作目录创建。cd learn-claude-code python agents/s11_autonomous_agents.py5.2 观察自主行为的推荐操作原文档给出的五条试验 prompt 与建议操作如下Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim.—— 创建 3 个任务后 spawn 两名队友观察他们自动领取任务Spawn a coder teammate and let it find work from the task board itself—— spawn 一个 coder 角色队友让它自己从任务板找活Create tasks with dependencies. Watch teammates respect the blocked order.—— 创建带依赖关系的任务观察队友遵守 blockedBy 顺序输入/tasks查看任务板——每个任务会带状态标记[ ]pending、[]in_progress、[x]completed与当前 owner输入/team实时查看每个队友是working还是idle。交互入口还有/inbox查看 Leader 收件箱源码入口 使用非清空方式读取输入q/exit或空行退出。/tasks的渲染逻辑见 agents/s11_autonomous_agents.py#L570-L577。六、小结与后续章节s11 的本质是把任务分配从 Leader 的显式行为下沉为一种文件系统层面的自组织协议队友之间不互相通信就能协调领取靠的是.tasks/中的 JSON 任务状态 线程锁保证的原子claim而 60 秒空闲超时让团队规模可以按需伸缩——没有活的队友自动退出不占用 LLM 调用。身份再注入则补上了压缩机制在长期运行 Agent场景下的最后一块拼图。在本仓库的课程结构中这套自主协作机制后续还会被 s12worktree 任务隔离继续扩展agents/目录下各章节实现可对照阅读agents/s10_team_protocols.py前一章、agents/s12_worktree_task_isolation.py后一章课程总入口见 README.md。【免费下载链接】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），仅供参考