自主 Agent —— 让 Agent 自己决定做什么

最优秀的下属不是"你说一步他做一步"，而是"他看到有什么活该干，自己就去干了"。

---

从"要我干"到"我要干"

回顾之前的演进：

• s01-s06：Agent 被动响应。你说什么，它做什么。你不说，它不动。
• s07-s08：Agent 开始有计划。它能把大任务拆成小任务，但仍然是"你给了任务它才做"。
• s09-s10：Agent 之间能协作，但 Leader 还是"下指令"的模式：spawn → 分配任务 → 等待完成。

都是"要我干"。

s11 把最后一层依赖也去掉了。 Agent 可以：

1. 完成工作后进入 IDLE 状态
2. 在 IDLE 状态下，定时扫描任务板
3. 发现未认领的任务 → 自动认领
4. 认领后自动恢复 WORKING 状态

这是一个从"被动执行者"到"主动工作者"的跃迁。

---

三阶段生命周期

s11 的 Agent 生命周期变成了三个阶段：

+-------+
| SPAWN |  ← 被 Lead spawn
+---+---+
    |
    v
+-------+  tool_use    +-------+
| WORK  | ←─────────── |  LLM  |  工作阶段：正常 agent loop
+---+---+              +-------+
    |
    | stop_reason != tool_use
    v
+--------+
| IDLE   |  轮询最多 60 秒，每 5 秒一次
+---+----+
    |
    ├──→ 检查 inbox → 有消息？→ 恢复 WORK
    |
    ├──→ 扫描 .tasks/ → 有未认领任务？→ 认领 → 恢复 WORK
    |
    └──→ 60 秒超时 → SHUTDOWN

代码实现比描述更直白：

class TeammateManager:
    def _loop(self, name: str, role: str, prompt: str):
        while True:
            # ── WORK PHASE ──
            for _ in range(50):  # 最多 50 轮工具调用
                inbox = BUS.read_inbox(name)
                for msg in inbox:
                    if msg.get("type") == "shutdown_request":
                        self._set_status(name, "shutdown")
                        return  # 直接退出
                    messages.append(...)

                response = client.messages.create(...)
                messages.append(response)

                if response.stop_reason != "tool_use":
                    break  # 进入 IDLE

                # 执行工具...
                if block.name == "idle":
                    idle_requested = True  # 模型主动要求闲置

            # ── IDLE PHASE ──
            self._set_status(name, "idle")
            for _ in range(polls):  # polls = IDLE_TIMEOUT / POLL_INTERVAL
                time.sleep(POLL_INTERVAL)  # 每 5 秒醒一次
                inbox = BUS.read_inbox(name)
                if inbox:
                    resume = True
                    break
                unclaimed = scan_unclaimed_tasks()
                if unclaimed:
                    task = unclaimed[0]
                    result = claim_task(task["id"], name)
                    if not result.startswith("Error:"):
                        resume = True
                        break

            if not resume:
                self._set_status(name, "shutdown")  # 60 秒无动静 → 关机
                return
            self._set_status(name, "working")  # 找到活 → 恢复工作

---

“idle” 工具：模型可以主动说"我没事干了"

一个特别有意思的细节是那个 idle 工具：

{"name": "idle", "description": "Signal that you have no more work. Enters idle polling phase.",
 "input_schema": {"type": "object", "properties": {}}}

这不是强迫 Agent 闲置的机制——是让 Agent 主动说"我干完了"的工具。

当 Agent 完成当前任务，没有更多 TODO 可以做，它会主动调用 idle：

Agent: 完成了支付模块重构，提交了 PR。进入 idle。
  → 调用 idle 工具
  → 循环进入 IDLE PHASE
  → 每 5 秒扫描一次任务板
  → 发现 task #42 "优化数据库查询" 无人认领
  → 自动认领，注入消息 "Claimed task #42"
  → 恢复 WORKING

这就像你团队里最积极的同事。他干完手里的活，不会干坐着等——他会自己去 backlog 里找下一件该做的事。

模型自己决定什么时候"干完了"，而不是 Harness 替它判断。

---

安全认领：Claim 锁

多 Agent 同时扫任务板时，可能两个人同时看中了同一个任务。s11 用了一个简单的线程级锁来防止重复认领：

_claim_lock = threading.Lock()

def claim_task(task_id: int, owner: str) -> str:
    with _claim_lock:  # ← 只有一个线程能进入
        path = TASKS_DIR / f"task_{task_id}.json"
        if not path.exists():
            return f"Error: Task {task_id} not found"
        task = json.loads(path.read_text())
        if task.get("owner"):
            return f"Error: Task {task_id} already claimed by {task['owner']}"
        if task.get("status") != "pending":
            return f"Error: Task {task_id} status is '{task['status']}'"
        if task.get("blockedBy"):
            return f"Error: Task {task_id} is blocked by other task(s)"

        task["owner"] = owner
        task["status"] = "in_progress"
        path.write_text(json.dumps(task, indent=2))
    return f"Claimed task #{task_id} for {owner}"

三重检查 + 文件写入：检查 ownership → 检查状态 → 检查依赖 → 原子写入。

注意：这里的"原子性"是 Python 的 GIL + 文件系统提供的，不是真正的分布式锁。但在单进程多线程的场景下够用了。如果多进程，需要用文件锁（fcntl/msvct），但那是 s12 的事了。

---

身份重建：压缩后别失忆

s11 还有一个重要的细节——identity re-injection。

上下文压缩后，Agent 的 messages 被替换成了一个摘要。但如果摘要里没有包含身份信息，Agent 醒来后会失忆：

[Conversation compressed]
完成了支付模块重构。当前正在等待 code review。
→ "我是谁？我是什么角色？我属于哪个团队？"
→ 茫然中

s11 的解法是：在压缩后插入身份块：

def make_identity_block(name: str, role: str, team_name: str) -> dict:
    return {
        "role": "user",
        "content": f"<identity>You are '{name}', role: {role}, team: {team_name}. Continue your work.</identity>",
    }

当 Agent 从 IDLE 恢复并认领了新任务时：

if len(messages) <= 3:  # 说明刚被压缩过
    messages.insert(0, make_identity_block(name, role, team_name))
    messages.insert(1, {"role": "assistant", "content": f"I am {name}. Continuing."})

这个检查 len(messages) <= 3 的意思是：如果消息列表很短（刚被压缩），插入身份信息。如果消息列表很长（还没压缩过），不需要。

这个设计很小，但很关键。 没有它，自主 Agent 会在第一次压缩后变成一个"没有记忆的打工机器"——它会知道自己要干活，但不知道自己是谁、有什么专长、在团队里扮演什么角色。

---

自主性带来了什么？

给 Agent 自主性，不只是"更酷了"。它带来了实际的工程价值：

1. 利用率提升
以前 Agent 干完活需要人来派新活。现在它会自己去 task board 上找。人只需要往 board 上扔任务，Agent 群体会自动消化。

2. 弹性伸缩
任务多的时候，更多 Agent 会处于 WORKING 状态。任务少的时候，Agent 做完活自动 IDLE 再 SHUTDOWN。不需要手动管理"扩缩容"。

3. 故障自愈
如果一个 Agent 在处理任务时崩溃了（比如 API 超时），任务会保持在 in_progress 状态。其他 Agent 不会碰它。但如果任务加了超时机制，超时后任务回到 pending，其他 Agent 可以重新认领。

4. 人只需要定义目标
不需要告诉 Agent"第二步做什么、第三步做什么"。只需要把任务扔到 board 上，合格的 Agent 自然会认领和处理。

---

但自主性有个坑

自主性的反面是不可预测性。

当你的 Agent 开始自己决定做什么的时候，你可能会遇到：

• Agent A 和 B 同时看上了一个任务，争抢（锁机制解决了这个）
• Agent 认领了一个它其实不擅长的任务（没有匹配机制）
• Agent 在 IDLE 扫描时疯狂轮询 API（s11 用 5 秒间隔限制了这个）
• 所有 Agent 都去抢简单任务，复杂任务没人动（没有优先级机制）

s11 的解决方案是最简的——它只解决了"能不能自主"的问题，没解决"自主得好不好"的问题。那些是留给生产系统的优化空间。

---

下集预告

s11 让 Agent 可以自主干活，但所有 Agent 共享同一个工作目录。有两个 Agent 同时改同一个文件怎么办？

s12 用 git worktree 解决了这个问题——每个任务有自己独立的目录，通过 worktree 隔离，永不碰撞。

下一篇：Worktree 任务隔离 —— 目录即边界