S17: Autonomous Agents — 代码解析

相对于 S16 新增:scan_unclaimed_tasks + idle_poll + WORK→IDLE→SHUTDOWN 生命周期 + Teammate 任务自取工具


1. 整体功能概览

01
解决什么问题

S16 的 teammate 依赖 Lead 主动分配任务。S17 让 teammate 自主发现并认领任务:完成当前工作 → 进入 IDLE → 扫描任务板 → 自动 claim 未认领任务 → 回到 WORK。无任务可做时等待 60s 后自行退出(timeout)。

02
三阶段生命周期

WORK:最多 10 轮 LLM + tool_use 迭代。完成后进入 IDLE。
IDLE:每 5s 轮询 inbox + 任务板,最长等 60s。
SHUTDOWN:收到 shutdown_request 或 timeout,发 summary 给 Lead,退出线程。

03
新增模块

IDLE_POLL_INTERVAL = 5IDLE_TIMEOUT = 60scan_unclaimed_tasks()idle_poll(),Teammate 额外工具 list_tasks / claim_task / complete_task,身份重注入逻辑(if len(messages) <= 3)。


2. 常量定义

IDLE_POLL_INTERVAL = 5 # seconds 模块级常量;int,单位秒;IDLE 阶段每次轮询间隔
IDLE_TIMEOUT = 60 # seconds IDLE 最长等待时间;60s / 5s = 12 次轮询后 timeout
为何用常量而不是硬编码?
易于调整(测试时可改小);搜索时一目了然;S18/S19/S20 直接继承这两个常量。

3. scan_unclaimed_tasks — 任务板扫描

新增遍历 .tasks/ 目录,找出可被 teammate 自动认领的任务。

def scan_unclaimed_tasks() -> list[dict]: 返回 list[dict](原始 JSON dict,非 Task dataclass);避免在 idle_poll 中额外做 Task 转换
unclaimed = [] 空列表累积结果
for f in sorted(TASKS_DIR.glob("task_*.json")): Path.glob() 返回 generator;sorted() 保证确定性顺序(任务 ID 含时间戳,sorted = 按创建时间)
task = json.loads(f.read_text()) 读取 JSON 为 dict;json.loads() 解析字符串
if (task.get("status") == "pending" dict.get() 安全取值;检查任务状态
and not task.get("owner") owner 为 None 或空字符串时 falsy;未认领任务
and can_start(task["id"])): 检查依赖是否全部完成;三个条件 AND,全满足才加入结果
unclaimed.append(task) 追加到结果列表
return unclaimed 返回所有满足条件的任务(可能为空列表)
示例输入/输出:
.tasks/ 目录下有 3 个文件:
task_001.json: {status:"completed", owner:"lead"}
task_002.json: {status:"pending", owner:null, blockedBy:["task_001"]}
task_003.json: {status:"pending", owner:"alice"}

scan_unclaimed_tasks() 调用:
task_001: status=completed → 跳过
task_002: status=pending, owner=None, can_start=True (task_001完成) → ✓ 加入
task_003: owner="alice" → not task.get("owner") = False → 跳过
返回: [{"id":"task_002","subject":"Build tests",...}]

4. idle_poll — IDLE 阶段轮询器

新增Teammate 完成 WORK 阶段后调用此函数。它轮询 inbox 和任务板,返回下一步行动指令。

def idle_poll(agent_name: str, messages: list, name: str, role: str) -> str: 返回 "work" | "shutdown" | "timeout";调用方据此决定下一阶段
for _ in range(IDLE_TIMEOUT // IDLE_POLL_INTERVAL): // 整除运算;60//5=12 次循环;for 变量用 _ 表示不需要值
time.sleep(IDLE_POLL_INTERVAL) 阻塞当前线程 5s;在 daemon 线程中安全使用
inbox = BUS.read_inbox(agent_name) 破坏性读取:读后删除 .jsonl 文件
if inbox: 非空列表为 truthy
for msg in inbox: if msg.get("type") == "shutdown_request": IDLE 中优先处理 shutdown;否则注入消息回 WORK
req_id = msg.get("metadata",{}).get("request_id","") 链式 dict.get();防止 metadata 缺失时 KeyError
BUS.send(name,"lead","Shutting down.", "shutdown_response", {"request_id":req_id,"approve":True}) 回传 shutdown 响应给 Lead;approve=True 表示同意
return "shutdown" 立即返回;调用方退出 WORK/IDLE 循环
messages.append({"role":"user", "content":"<inbox>" + json.dumps(inbox) + "</inbox>"}) 非 shutdown 消息:包装成 XML-like 标签注入到 messages;LLM 下轮能读到
return "work" 有新消息 → 回 WORK 阶段
unclaimed = scan_unclaimed_tasks() 无 inbox 消息时再扫描任务板
if unclaimed: task = unclaimed[0] result = claim_task(task["id"], agent_name) 取第一个(最早)未认领任务;claim_task 带 owner=agent_name
if "Claimed" in result: 字符串包含检查;claim_task 成功时返回 "Claimed task_xxx..."
messages.append({"role":"user", "content":f"<auto-claimed>Task {task['id']}: " f"{task['subject']}</auto-claimed>"}) 告知 LLM 自动认领了新任务;XML 标签帮助 LLM 识别这是系统通知
return "work" 有新任务 → 立即回 WORK
print(f" [idle] {name} timeout ({IDLE_TIMEOUT}s)") return "timeout" 12 次轮询结束仍无任务/消息 → timeout;调用方让 teammate 退出
示例 — auto-claim 场景:
alice 完成 task_001,进入 IDLE
第 1 次轮询 (t+5s): inbox 空,scan_unclaimed → [task_002]
claim_task("task_002", "alice") → "Claimed task_002 (Build tests)"
"Claimed" in result → True
messages.append("<auto-claimed>Task task_002: Build tests</auto-claimed>")
返回: "work"
alice 回 WORK 阶段,LLM 看到 auto-claimed 提示,开始执行 Build tests

5. WORK → IDLE → SHUTDOWN 生命周期

S17 将 spawn_teammate_thread 的内部 run() 函数重构为两层嵌套循环。

外层循环结构

# Outer loop: WORK → IDLE cycle while True: 无限循环;由内部 break 或 idle_result 判断退出
# Identity re-injection (s17) if len(messages) <= 3: messages 很短时(刚压缩后)重注入身份;见 section 7
# WORK phase should_shutdown = False for _ in range(10): WORK 最多 10 轮 LLM 迭代;should_shutdown 作为 break 信号
if should_shutdown: break WORK 阶段收到 shutdown → 退出外层循环
# IDLE phase (s17 new) idle_result = idle_poll(name, messages, name, role) 调用 idle_poll;可能阻塞最长 60s
if idle_result == "shutdown": break if idle_result == "timeout": break 两种退出条件;"work" 时 continue(回到外层循环顶部)

生命周期状态图

┌─────────────────┐ │ WORK PHASE │ │ (max 10 turns) │ └────────┬────────┘ │ stop_reason != "tool_use" │ (或10轮满) ▼ ┌─────────────────┐ │ IDLE PHASE │ │ (轮询 60s) │ └────────┬────────┘ ┌───────────┼──────────────┐ │ │ │ inbox unclaimed timeout (消息) (任务) (60s到) │ │ │ "work" "work" "timeout" │ │ │ └───────────┘ │ │ SHUTDOWN 回 WORK (发 summary) (外层循环顶部)

6. Teammate 新增工具:list_tasks / claim_task / complete_task

新增S16 teammate 只有 bash/read_file/write_file/send_message/submit_plan(5个工具)。S17 增加 3 个任务板工具(共8个),让 teammate 能自主管理任务。

list_tasks(teammate 版)

def _run_list_tasks(): tasks = list_tasks() if not tasks: return "No tasks." return "\n".join( f" {t.id}: {t.subject} [{t.status}]" for t in tasks) 内嵌闭包(closure);捕获外层 name/role 变量;简化版输出(无 owner、无 blockedBy)适合 LLM 阅读

claim_task(teammate 版)

def _run_claim_task(task_id: str): return claim_task(task_id, owner=name) 关键:owner=name(teammate 自己的名字),不是 "agent";区分哪个 teammate 在做这个任务

complete_task(teammate 版)

def _run_complete_task(task_id: str): return complete_task(task_id) 调用全局 complete_task;完成后其他 teammate 的 scan_unclaimed_tasks 可能发现新任务(解除 blockedBy)
sub_handlers = { ... "list_tasks": _run_list_tasks, "claim_task": _run_claim_task, "complete_task": _run_complete_task, } dict 字面量;将工具名映射到处理函数;LLM 调用 "list_tasks" 时执行 _run_list_tasks()
示例 — teammate 自主工作流:
LLM 输出 tool_use: {name:"list_tasks"} → _run_list_tasks()
→ " task_002: Build tests [pending]"
LLM 输出 tool_use: {name:"claim_task", input:{task_id:"task_002"}}
→ _run_claim_task("task_002") → claim_task("task_002", owner="alice")
→ "Claimed task_002 (Build tests)"
... (执行任务) ...
LLM 输出 tool_use: {name:"complete_task", input:{task_id:"task_002"}}
→ _run_complete_task("task_002")
→ "Completed task_002 (Build tests)\nUnblocked: Write docs"

7. 身份重注入 — 防止 LLM 上下文压缩后忘记身份

新增当 messages 被压缩到只剩几条时,LLM 可能忘记自己是谁(哪个 teammate,什么角色)。S17 在外层循环顶部检查并补充身份提示。

# Identity re-injection (s17) if len(messages) <= 3: messages 长度 ≤ 3 表明刚刚经历压缩或刚启动
messages.insert(0, {"role": "user", "content": f"<identity>You are '{name}', role: {role}. " f"Continue your work.</identity>"}) list.insert(0, ...) 插入到开头(最早的位置);XML 标签帮助 LLM 识别这是系统提示而非用户消息
为什么用 insert(0, ...) 而不是 append?
插入到开头 = 最早的 context;LLM 按时间顺序读 messages,开头的信息优先级最低但最稳定。
如果 append 到末尾,它可能覆盖最近的工作内容,干扰 LLM 的短期推理。
注意:messages[-20:] 切片(send to API 时)仍会截断开头,因此这个注入更像一个"early anchor"而非保证。

8. 完整自主 Agent 流程

ASCII 生命周期(详细版): spawn_teammate_thread("alice", "backend engineer", "Help with tasks") │ └─ Thread.start() → run() │ ├─ [OUTER LOOP 1] │ 身份检查 (len(messages)<=3?) → 注入 <identity> │ ┌─ [WORK PHASE, 最多10轮] │ │ read_inbox → 协议分发 │ │ LLM call → tool_use? │ │ │ Yes → 执行 tool → 回顶 │ │ │ No → 退出 WORK │ └─ WORK 完成 │ [IDLE PHASE] │ ┌─ 轮询 (max 12次 × 5s) │ │ read_inbox → shutdown? → return "shutdown" │ │ scan_unclaimed → 有任务? → claim → return "work" │ │ (等待 5s) → repeat │ └─ timeout → return "timeout" │ idle_result == "work"? │ Yes → 回外层循环顶 (OUTER LOOP 2) │ No → break │ ├─ [OUTER LOOP 2] (有新任务/消息) │ 身份检查 → WORK → IDLE → ... │ └─ [退出] BUS.send("alice", "lead", summary, "result") active_teammates.pop("alice")