11 - 自治智能体

s01 > s02 > s03 > s04 > s05 > s06 | [[../09-agent-teams/09-智能体团队|s09]] > [[../10-team-protocols/10-团队协议|s10]] > [ s11 ] s12

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

Harness 层：自治 —— 模型自己找活干，无需指派。

学习目标

• 理解 WORK/IDLE 双阶段循环的设计模式
• 掌握任务看板自动扫描与认领机制
• 学会上下文压缩后的身份重注入策略
• 了解空闲超时与自动关机的工程意义

核心概念

问题

[[../09-agent-teams/09-智能体团队|s09]] 和 [[../10-team-protocols/10-团队协议|s10]] 中，队友只在被明确指派时才动。领导得给每个队友写 prompt，任务看板上 10 个未认领的任务得手动分配。这扩展不了。

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

一个细节：上下文压缩（[[../06-context-compact/06-上下文压缩|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 双阶段循环

队友循环分两个阶段：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")

🔍 Deep Dive：为什么是 50 次迭代上限？

for _ in range(50) 并非任意选择。这是防止模型陷入无限工具调用循环的安全阀。实际产品中应作为可配置参数（max_consecutive_tool_calls），根据任务的工具调用密度动态调整。50 对于代码任务（编译、测试、修复循环）通常是足够的。

2. 空闲阶段：轮询收件箱 + 任务看板

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. 收件箱优先 —— 队友协作消息比任务看板优先级更高
2. 任务看板次之 —— 没有直接消息时才扫描可用任务
3. 超时即关机 —— 60 秒无动静就释放资源

3. 任务看板扫描：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

扫描条件确保队友不会：

• 捡走已完成或进行中的任务（必须 pending）
• 抢走已被认领的任务（必须无 owner）
• 启动依赖未就绪的任务（必须无 blockedBy）

🔍 Deep Dive：竞态条件与认领原子性

多个队友同时扫描看板时可能看到同一个未认领任务。代码中 claim_task 的实现需要原子性：先写 owner 再检查是否已被抢占。在实际系统中，建议用文件锁（fcntl.flock）或专门的状态转移文件操作，防止 double-claim。

场景：Alice 和 Bob 同时扫描
Alice: 读取 task_03.json → 无 owner → 写入 owner="alice"
Bob:   读取 task_03.json → 无 owner（Alice 还没写完） → 写入 owner="bob" ❌
                                    ↓
解决方案：写 owner 后立即读回验证，如不匹配则回退

4. 身份重注入

上下文过短（说明发生了压缩）时，在开头插入身份块：

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."})

这个机制解决了[[../06-context-compact/06-上下文压缩|s06 上下文压缩]]引入的副作用：压缩后模型失去了"我是谁、在哪个团队"的上下文。重注入用 <identity> 标签包裹身份信息，帮助模型快速恢复状态。

5. 空闲超时与自动关机

60 秒空闲后队友自动退出。这一设计避免了：

• 僵尸进程占用系统资源
• 多个队友同时执行同一任务
• 空转的 LLM API 费用

要重建队友只需重新 spawn，新实例会从看板读取最新的任务状态。

相对 s10 的变更

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

架构演进一览

s09 单体团队  ───→  s10 协议驱动  ───→  s11 自组织
  领导派活           领导派活 + 协议     队友自己看板认领
  无空闲阶段          无空闲阶段          双阶段循环
  无超时              无超时              60s idle timeout
  身份只在 prompt     同 left             压缩后重注入

设计决策

决策	选择	理由
轮询 vs 事件驱动	轮询	简化实现，LLM 调用的秒级延迟可接受
5s 间隔	经验值	平衡响应速度与磁盘 I/O
50 次 WORK 上限	安全阀	防止无限工具调用循环
3 条消息阈值	启发式	压缩后通常只剩 1-3 条消息
inbox 优先	协作优先	避免队友忽略队友消息

试一试

cd learn-claude-code
python agents/s11_autonomous_agents.py

试试这些 prompt（英文 prompt 对 LLM 效果更好，也可以用中文）：

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 查看带 owner 的任务看板
5. 输入 /team 监控谁在工作、谁在空闲

延展思考

• 调度策略：当前实现是简单的先到先得。更高级的策略可以引入优先级排序、能力匹配（role-based claiming）、负载均衡
• 事件驱动替代：如果延迟要求高，可以用 fswatch / inotify 监听任务目录变化，替代定时轮询
• 弹性扩缩：空闲超时后自动销毁 + 队列积压时自动 spawn 新队友 = 基础的弹性扩缩模式
• 与 s12 的关系：[[../12-worktree-isolation/12-工作区隔离|s12 Worktree 隔离]] 为自治队友提供了安全的执行沙箱，防止队友间的文件系统冲突

讨论问题

1. 轮询 vs 事件驱动：当前用 5s 定时轮询，如果改成 inotify/fswatch 监听文件变化，会有什么优缺点？
2. 竞态条件：两个队友同时扫描看板发现同一个未认领任务，谁的写操作会赢？如何保证认领的原子性？
3. 60s 空闲超时是否合理？如果任务启动需要 30s 安装依赖，队友是否应该在安装期间保持 WORK 状态？
4. 身份重注入用 len(messages) <= 3 作为压缩检测条件 —— 这个阈值的误判率如何？什么时候会误触发？
5. 从 s09 到 s11，控制权从领导逐步转移到队友自身。这种去中心化设计的优缺点各是什么？

延伸阅读

• Learn Claude Code 项目 — s11 练习代码
• [[../09-agent-teams/09-智能体团队|← s09 智能体团队]] · [[../10-team-protocols/10-团队协议|← s10 团队协议]] · [[../12-worktree-isolation/12-工作区隔离|s12 工作区隔离 →]]
• [[../00-课程概览/综述|返回课程综述]]