Learn Claude Code
s11

Autonomous Agents

协作

Scan Board, Claim Tasks

506 LOC14 个工具Task board polling + timeout-based self-governance
机制概览

队友 agent 在没事干时进入"待命"状态,每隔 30 秒自动看一眼共享任务板,发现自己能做的任务就主动认领开始干。不需要你做指挥官——队友们自己看板自己接活。

本节贯穿示例
研究项目
耐心资本对企业 ESG 表现的实证项目 do10 稳健性矩阵要跑 A 面板(替换 PC 测度 × 7 套)+ B 面板(替换设定 × 6 套:ESG 评级版、PC 滞后 1 期、PC 滞后 2 期、行业+年份 FE、企业 FE+行业×年份 FE、双重聚类)+ C 面板(子样本 × 5 套:剔 2015、剔疫情、仅制造业、剔北上广、2013-2023)共 18 组回归。把这 18 组放上任务板
在这一节里做的事

夜里两个稳健性回归队友自主认领跑完,早上回来直接拿 06_结果输出/tables_tsv/table10A_pc_swap.csv、table10B_alt_specs.csv、table10C_subsamples.csv 三份汇总。

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > [ s11 ] s12

"队友自己看看板, 有活就认领" -- 不需要领导逐个分配, 自组织。

Harness 层: 自治 -- 模型自己找活干, 无需指派。

问题

s09-s10 中, 队友只在被明确指派时才动。领导得给每个队友写 prompt, 任务看板上 10 个未认领的任务得手动分配。这扩展不了。

真正的自治: 队友自己扫描任务看板, 认领没人做的任务, 做完再找下一个。

一个细节: Context Compact (s06) 后 Agent 可能忘了自己是谁。身份重注入解决这个问题。

解决方案

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。LLM 停止调用工具 (或调用了 idle) 时, 进入 IDLE。
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")
  1. 空闲阶段循环轮询收件箱和任务看板。
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>{inbox}</inbox>"})
            return True
        unclaimed = scan_unclaimed_tasks()
        if unclaimed:
            claim_task(unclaimed[0]["id"], name)
            messages.append({"role": "user",
                "content": f"<auto-claimed>Task #{unclaimed[0]['id']}: "
                           f"{unclaimed[0]['subject']}</auto-claimed>"})
            return True
    return False  # timeout -> shutdown
  1. 任务看板扫描: 找 pending 状态、无 owner、未被阻塞的任务。
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
  1. 身份重注入: 上下文过短 (说明发生了压缩) 时, 在开头插入身份块。
if len(messages) <= 3:
    messages.insert(0, {"role": "user",
        "content": f"<identity>You are '{name}', role: {role}, "
                   f"team: {team_name}. Continue your work.</identity>"})
    messages.insert(1, {"role": "assistant",
        "content": f"I am {name}. Continuing."})

相对 s10 的变更

组件之前 (s10)之后 (s11)
Tools1214 (+idle, +claim_task)
自治性领导指派自组织
空闲阶段轮询收件箱 + 任务看板
任务认领仅手动自动认领未分配任务
身份系统提示+ 压缩后重注入
超时60 秒空闲 -> 自动关机

试一试

cd claude-code-for-researchers
python agents/s11_autonomous_agents.py

试试这些 prompt (英文 prompt 对 LLM 效果更好, 也可以用中文):

  1. Create 13 robustness tasks (PC measure swaps, lag specs, subsample drops) on the board, then spawn 2 robust-runner teammates. Watch them auto-claim.
  2. Spawn a stata-runner teammate and let it find robustness work from the task board itself
  3. Create tasks with dependencies (do4 panel -> do6 baseline -> do10 robustness). Watch teammates respect the blocked order.
  4. 输入 /tasks 查看带 owner 的任务看板
  5. 输入 /team 监控谁在工作、谁在空闲