# s11: Autonomous Agents `s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > [ s11 ] s12` > *"チームメイトが自らボードを見て、仕事を取る"* -- リーダーが逐一割り振る必要はない。 > > **Harness 層**: 自律 -- 指示なしで仕事を見つけるモデル。 ## 問題 s09-s10では、チームメイトは明示的に指示された時のみ作業する。リーダーは各チームメイトを特定のプロンプトでspawnしなければならない。タスクボードに未割り当てのタスクが10個あっても、リーダーが手動で各タスクを割り当てる。これはスケールしない。 真の自律性とは、チームメイトが自分で作業を見つけること: タスクボードをスキャンし、未確保のタスクを確保し、作業し、完了したら次を探す。 もう1つの問題: コンテキスト圧縮(s06)後にエージェントが自分の正体を忘れる可能性がある。アイデンティティ再注入がこれを解決する。 ## 解決策 ``` 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) ``` ## 仕組み 1. チームメイトのループはWORKとIDLEの2フェーズ。LLMがツール呼び出しを止めた時(または`idle`ツールを呼んだ時)、IDLEフェーズに入る。 ```python def _loop(self, name, role, prompt): while True: # -- WORK PHASE -- messages = [{"role": "user", "content": prompt}] for _ in range(50): response = client.messages.create(...) if response.stop_reason != "tool_use": break # execute tools... if idle_requested: break # -- IDLE PHASE -- self._set_status(name, "idle") resume = self._idle_poll(name, messages) if not resume: self._set_status(name, "shutdown") return self._set_status(name, "working") ``` 2. IDLEフェーズがインボックスとタスクボードをポーリングする。 ```python 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": f"{inbox}"}) return True unclaimed = scan_unclaimed_tasks() if unclaimed: claim_task(unclaimed[0]["id"], name) messages.append({"role": "user", "content": f"Task #{unclaimed[0]['id']}: " f"{unclaimed[0]['subject']}"}) return True return False # timeout -> shutdown ``` 3. タスクボードスキャン: pendingかつ未割り当てかつブロックされていないタスクを探す。 ```python def scan_unclaimed_tasks() -> list: 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 unclaimed ``` 4. アイデンティティ再注入: コンテキストが短すぎる(圧縮が起きた)場合にアイデンティティブロックを挿入する。 ```python if len(messages) <= 3: messages.insert(0, {"role": "user", "content": f"You are '{name}', role: {role}, " f"team: {team_name}. Continue your work."}) messages.insert(1, {"role": "assistant", "content": f"I am {name}. Continuing."}) ``` ## s10からの変更点 | Component | Before (s10) | After (s11) | |----------------|------------------|----------------------------| | Tools | 12 | 14 (+idle, +claim_task) | | Autonomy | Lead-directed | Self-organizing | | Idle phase | None | Poll inbox + task board | | Task claiming | Manual only | Auto-claim unclaimed tasks | | Identity | System prompt | + re-injection after compress| | Timeout | None | 60s idle -> auto shutdown | ## 試してみる ```sh cd learn-claude-code python agents/s11_autonomous_agents.py ``` 1. `Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim.` 2. `Spawn a coder teammate and let it find work from the task board itself` 3. `Create tasks with dependencies. Watch teammates respect the blocked order.` 4. `/tasks`と入力してオーナー付きのタスクボードを確認する 5. `/team`と入力して誰が作業中でアイドルかを監視する