HARNESS 111

Autonomous Agents

"队友自己看看板，有活就认领"

s09-s10 的队友是"被动型员工"——等领导派活才动。
s11 的队友是"自驱型员工"——自己扫描任务看板，看到没人做的就认领，做完继续找下一个。

s10 的队友有一个致命瓶颈——领导

在 s09-s10 中，每个队友只在被 Lead 明确 spawn 并给了具体 prompt 后才工作。工作完成后进入 idle，然后什么也不做——等 Lead 再次 spawn。

三个具体问题

① 领导是单点瓶颈：s07 的 .tasks/ 目录里有 20 个 unclaimed 任务，Alice 和 Bob 都在 idle 状态——但没人去做。为什么？因为没人告诉他们"去做"。Lead 必须手动检查每个任务、决定分配给谁、逐个 spawn。

② 无法扩展：3 个队友、10 个任务时 Lead 还能管理。30 个队友、100 个任务时 Lead 就是瓶颈了。系统的工作吞吐量被 Lead 的注意力上限限制。

③ 压缩后失去身份：s06 压缩后，Agent 的 messages 被替换为摘要。队友可能忘了"我是谁？我是什么角色？我擅长什么？" ——这些信息原本在 system prompt 和早期消息中，但压缩后可能丢失。

s11 要解决的问题：让队友在 idle 时自己找活干——扫描任务看板 → 找到没人做的 → 认领 → 开始工作 → 完成 → idle → 继续找。不需要 Lead 分配。领导从"工头"变成了"愿景制定者"。

两阶段生命周期：WORK ↔ IDLE

s11 的队友不再是"一次性任务执行器"。它们有一个持续的生命周期，在两个阶段之间循环，直到超时或收到关停信号。

⚡ WORK 工作阶段

↓ LLM 停止调用工具 / 调了 idle 工具

⏸️ IDLE 空闲阶段

每 5 秒检查一次：有收件箱消息？→ 重返 WORK

有未认领任务？→ 认领 → 重返 WORK

60 秒内什么都没有？→ SHUTDOWN

↑ 有新的工作 → 重返 WORK

WORK 阶段：和 s01 一样的 agent loop

WORK 阶段就是标准的 agent loop——读收件箱、调 LLM、执行工具、循环。唯一的区别是 LLM 现在有一个新的工具 idle——当 LLM 觉得"我没活干了"，它可以主动调 idle 进入空闲阶段。

IDLE 阶段：s11 的全新发明

IDLE 阶段不在调用 LLM——它只是一个轮询循环。每 5 秒做两件事：① 检查收件箱有没有新消息（有人找我？）；② 扫描 .tasks/ 目录有没有未认领的可执行任务（有活能干？）。

找到任何一个 → 立刻退出 IDLE，进入 WORK 阶段（带着新消息或新任务）。60 秒内什么都没找到 → 这个队友自动关机（shutdown）。

不调用 LLM 来轮询——这是一个关键设计选择。轮询是纯 Python 文件扫描，不消耗 LLM 推理。这保证了 idle Agent 的成本几乎为零。

案例：两个自主 Agent 抢任务

Lead 在 s07 的 .tasks/ 里创建了 5 个任务（搭建→前端→后端→测试→部署，有依赖关系），然后 spawn 了 Alice（前端）和 Bob（后端）。

自主协作过程

t=0s：Alice 和 Bob 同时从 spawn 进入 WORK 阶段。各自执行 Lead 给的 prompt。

t=30s：Alice 完成初始任务，调 idle 工具 → 进入 IDLE。每 5 秒扫描 .tasks/。

t=45s：Bob 也完成了，调 idle → 进入 IDLE。

t=50s：有人（可能是 Lead）把任务 #1 标为 completed → #2（前端）和 #3（后端）的 blockedBy 被清空。

t=55s：Alice 的 IDLE 轮询扫描到 #2（前端，unclaimed，unblocked）。调 claim_task(2, "alice") → 认领成功 → 进入 WORK，开始做前端。

t=55s：Bob 的 IDLE 轮询扫描到 #3（后端，unclaimed，unblocked）。调 claim_task(3, "bob") → 认领成功 → 进入 WORK，开始做后端。

t=120s：两个都完成了各自的任务 → idle → 继续扫描。

整个过程 Lead 没有介入——没有说"Alice 你做 #2，Bob 你做 #3"。Agent 自己看了看板，自己决定谁做什么。s07 的任务图提供了"什么可以做"的信息，s11 的自主循环让 Agent 自己行动。

如果两个 Agent 同时认领同一个任务怎么办？_claim_lock（线程锁）保证了原子性——只有一个 Agent 能成功认领。另一个会收到 "已被 xx 认领" 的错误，然后继续扫描下一个任务。和 s08 的 _lock 一样的模式。

任务扫描 & 原子认领

scan_unclaimed_tasks()：找到"在等主人"的任务

📝 中文注释版

def scan_unclaimed_tasks():
    TASKS_DIR.mkdir(exist_ok=True)
    unclaimed = []
    for f in sorted(TASKS_DIR.glob("task_*.json")):
        task = json.loads(f.read_text())
        if (task["status"] == "pending"       # 还没开始
            and not task.get("owner")         # 没人认领
            and not task.get("blockedBy")):   # 没有被阻塞
            unclaimed.append(task)
    return unclaimed
      

💻 英文代码原版

def scan_unclaimed_tasks():
    TASKS_DIR.mkdir(exist_ok=True)
    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（没有被阻塞）。三者缺一不可。这个简单的筛选表达了"这个任务现在就可以开始做"。

claim_task()：原子认领，防止抢同一个

📝 中文注释版

def claim_task(task_id, owner):
    with _claim_lock:                         # 🔒 原子操作：同一时间只能一个人认领
        task = json.loads(读文件)
        if task.get("owner"):                  # ⭐ 已经被别人抢了！
            return f"错误：已被 {task['owner']} 认领"
        if task["status"] != "pending":        # 状态不对
            return "错误：状态不是 pending"
        if task.get("blockedBy"):             # 还被阻塞
            return "错误：仍被阻塞"
        task["owner"] = owner                  # 盖上我的名字
        task["status"] = "in_progress"         # 标记为进行中
        写回文件(task)                         # 持久化
    return f"已认领任务 #{task_id}"
      

💻 英文代码原版

def claim_task(task_id, owner):
    with _claim_lock:
        task = json.loads(path.read_text())
        if task.get("owner"):
            return f"Error: claimed by {task['owner']}"
        if task.get("status") != "pending":
            return "Error: not pending"
        if task.get("blockedBy"):
            return "Error: still blocked"
        task["owner"] = owner
        task["status"] = "in_progress"
        path.write_text(json.dumps(task, indent=2))
    return f"Claimed task #{task_id}"
      

_claim_lock 的精妙之处：它保护的不仅是"写文件"，而是整个"读→检查→写"的原子操作。如果只锁写操作，Alice 和 Bob 可能同时读到 owner 为空 → 同时认为自己认领成功 → 最后写入的那个覆盖了前一个。锁包围整个 check-and-write 流程，保证了认领的正确性。

空闲轮询逻辑：不花钱的等待

IDLE 阶段的轮询循环

📝 中文注释版

# IDLE 阶段：每5秒检查，最多60秒
polls = IDLE_TIMEOUT // POLL_INTERVAL     # 60 / 5 = 12 次
for _ in range(polls):
    time.sleep(POLL_INTERVAL)              # 睡 5 秒

    # ① 检查收件箱：有人找我吗？
    inbox = BUS.read_inbox(name)
    if inbox:
        # 有人发消息（或是 shutdown_request）→ 注入到 messages，重返 WORK
        messages.append(inbox)
        resume = True
        break

    # ② 检查任务看板：有活吗？
    unclaimed = scan_unclaimed_tasks()
    if unclaimed:
        # 有！认领它 → 把任务内容注入 messages → 重返 WORK
        task = unclaimed[0]
        claim_task(task["id"], name)
        messages.append(任务内容)
        resume = True
        break

# 12 次都没找到 → 超时，shutdown
if not resume:
    self._set_status(name, "shutdown")
      

💻 英文代码原版

polls = IDLE_TIMEOUT // max(POLL_INTERVAL, 1)
for _ in range(polls):
    time.sleep(POLL_INTERVAL)
    inbox = BUS.read_inbox(name)
    if inbox:
        messages.append(json_msg)
        resume = True
        break
    unclaimed = scan_unclaimed_tasks()
    if unclaimed:
        task = unclaimed[0]
        claim_task(task["id"], name)
        messages.append(task_prompt)
        resume = True
        break
if not resume:
    self._set_status(name, "shutdown")
    return
      

POLL_INTERVAL=5, IDLE_TIMEOUT=60：这两个数字怎么选？

5 秒：太短浪费 CPU（每秒扫描 20 次文件系统），太长 Agent 反应迟钝（新任务等了 30 秒还没人认领）。5 秒是平衡点——对于代码 Agent 的工作节奏（完成任务通常需要几十秒到几分钟），5 秒的响应延迟几乎感觉不到。

60 秒：如果 60 秒内既没有新消息也没有新任务，说明"没活干了"。Agent 自动关机，释放线程资源。如果你的 Agent 团队需要 24/7 待命，可以把 IDLE_TIMEOUT 设为很大的值或无穷大。

身份重注入：压缩后还记得自己是谁

s06 的 auto_compact 可能把 messages 压缩到只剩一条摘要。如果 teammates 的 messages 被压缩，它可能失去关于"我是谁"的上下文。身份重注入解决了这个问题。

检测压缩 + 重新注入身份

# 在 teammates 认领新任务时，检查是否是压缩后的"失忆"状态
if len(messages) <= 3:                         # messages 很短 → 说明被压缩了
    messages.insert(0, {                          # 在最前面插入身份信息
        "role": "user",
        "content": f"<identity>你是 '{name}'，角色：{role}，团队：{team}。继续你的工作。</identity>"
    })
    messages.insert(1, {                          # 接着插入一个确认
        "role": "assistant",
        "content": f"我是 {name}，继续工作。"
    })
    

为什么是 messages <= 3？ 正常的 messages 至少有几条到几十条。如果只剩 1-3 条，说明被 auto_compact 压缩了（压缩后 messages 被替换为一条摘要 + 可能的系统消息）。这个简单的长度检查是判断"失忆"的启发式方法。

这个模式的价值：在长生命周期的 Agent 中，身份信息不应该只依赖 system prompt。system prompt 在压缩时可能被视为"已理解的信息"而被摘要模糊化。把身份信息显式地作为一条 user 消息插入，确保它始终在 messages 中可见。

代码拆解 & s10 vs s11

两阶段循环（~30 行）

WORK 阶段（标准 agent loop + idle 工具检测）+ IDLE 阶段（轮询循环）。最大的结构改变——while True 内嵌了 for 循环和 for 循环，而不是单一 agent loop。

任务扫描 + 原子认领（~30 行）

scan_unclaimed_tasks() 三个条件筛选 + claim_task() 的 check-and-write 原子操作。_claim_lock 是新增的锁，和 _tracker_lock 共享同一个保护模式。

身份重注入（~6 行）

长度检查 → 插入身份块。解决 s06 压缩后的"失忆"问题。

组件	s10	s11
生命周期	一次性：spawn→work→idle	循环：WORK ↔ IDLE
任务认领	仅 Lead 手动分配	Teammate 自动扫描+认领
空闲行为	无	每 5s 轮询，60s 超时关机
身份持久	仅 system prompt	压缩后重注入身份块
新工具	12	14 (+idle, +claim_task)
新锁	_tracker_lock	+ _claim_lock

从 s11 带走的四个核心认识

自治 = 自己找活

Agent 不需要等分配。扫描任务池 + 认领 + 执行。从"被动工具"变成"自主工作者"。

空闲轮询不烧钱

IDLE 阶段不调 LLM，纯文件扫描。5 秒一次，CPU 开销几乎为零。在等待和工作之间找到了经济平衡。

原子认领防冲突

_claim_lock 包围整个"读→检查→写"流程。多 Agent 抢同一个任务时，只有一个能成功。

身份需要显式保护

不依赖 system prompt 记住身份。压缩后主动重注入。长生命周期 Agent 需要"抗遗忘"机制。

落地案例：自主 Agent 在真实系统中的应用

🤖 AutoGPT / BabyAGI：自主 Agent 的先锋

2023 年引爆 AI Agent 热潮的 AutoGPT 和 BabyAGI，核心就是 s11 的"自主认领"模式——Agent 看到大目标 → 自动拆成子任务 → 逐个认领执行 → 根据结果决定下一步。

它们的弱点（无限循环、上下文爆炸、任务漂移）正是 s01-s12 逐步解决的问题——s06 压缩、s07 任务图、s10 协议、s11 超时关机。学完整个课程，你就理解了为什么那些早期自主 Agent"看起来很酷但用不起来"——它们缺少 harness 层的治理。

🔄 Kubernetes 的 Controller 模式

Kubernetes 的 Controller 就是一个自主 Agent——它不断轮询集群状态（= IDLE 阶段扫描），发现"期望状态"和"实际状态"不一致（= 发现了 unclaimed 任务），调谐到期望状态（= 认领并执行）。reconcile loop 就是 s11 的 WORK ↔ IDLE 循环。

📊 监控告警的 Auto-Remediation

一个自主 Agent 监控服务器状态——CPU 高了？自动认领"扩容"任务。磁盘满了？自动认领"清理日志"任务。证书快过期了？自动认领"续期"任务。这些任务来自 s07 的任务图，Agent 通过 s11 的自主循环发现并执行。

s12 是课程的最后一站：Worktree 隔离。
多个自主 Agent 同时修改代码时，怎么防止文件冲突？
答案：每个 Agent 在自己的 git worktree 里工作——隔离的文件系统，互不干扰。

s11 的自主 Agent 同时改代码会文件冲突。
s12：怎么用 Git Worktree 让每个 Agent 在独立环境里工作？

s12: Worktree Task Isolation →