作者：逆境不可逃

技术永无止境

希望我的内容可以帮助到你！！！！！

---

大家吼 ! 我是 逆境不可逃 今天给大家带来文章《【与我学 ClaudeCode】协作篇 之 Autonomous Agents 深度解析：自组织任务认领与空闲治理》.

Learn-Claude-Code 官方地址 :shareAI-lab/learn-claude-code: Bash is all you need - A nano claude code–like 「agent harness」, built from 0 to 1

上一篇文章：

【与我学 ClaudeCode】协作篇 之 Team Protocols ：结构化请求 - 响应协作协议-CSDN博客

Autonomous Agents 是迭代的第 11 个版本（s11），核心解决 多 Agent 团队中任务分配的中心化瓶颈问题。它在 s10 协议框架的基础上，实现了「自治队友 + 任务看板轮询 + 自动认领」的自组织协作模式，让队友无需领导指派，自己扫描任务看板、认领未分配任务，真正实现 “模型自己找活干”。

学习路线： s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > s09 > s10 > s11 > s12

---

一、问题根源：为什么中心化任务分配撑不起大规模团队？

s09-s10 中，队友只在被明确指派时才行动：

• 领导需要给每个队友写 prompt 分配任务，任务看板上 10 个未认领的任务得手动分配，效率极低
• 当任务量和队友数量增加时，领导的分配工作会成为瓶颈，系统无法扩展
• 队友没有任务时只能闲置，无法主动寻找新工作，资源利用率低

真正的自治团队需要：

1. 队友主动扫描任务看板，认领未分配任务
2. 无需领导逐个指派，实现自组织协作
3. 空闲队友能自动退出，避免资源浪费

---

二、三大核心设计决策

Autonomous Agents 通过三个关键设计，构建了一个简单、可靠、可扩展的自组织多 Agent 系统。

1. 轮询未认领任务而非事件驱动通知

核心设计：自主队友每隔约 1 秒轮询共享任务看板以寻找未认领的任务，而非等待事件驱动的通知。轮询从根本上比发布 / 订阅更简单：没有订阅管理、没有事件路由、没有事件丢失的 bug。在基于文件的持久化下，轮询就是 “读取目录列表”—— 一个低成本操作，无论有多少 agent 在运行都能正常工作。1 秒的间隔平衡了响应性（新任务被快速发现）和文件系统开销（不会过度读取磁盘）。

替代方案的致命缺陷：

事件驱动通知（如 inotify/fsevents 的文件观察器，或发布 / 订阅通道）能将延迟从秒级降到毫秒级，但文件观察器是平台特定的，且在网络文件系统上不可靠；消息代理可以工作，但会增加基础设施依赖。对于任务需要数分钟才能完成的系统来说，1 秒发现新任务和 10 毫秒发现没有实际区别。

2. 空闲 60 秒后自动终止

核心设计：当自主队友没有任务可做且收件箱中没有消息时，它最多等待 60 秒后放弃并关闭。这防止了永远等待不会到来的工作的僵尸队友 —— 这在组长忘记发送关闭请求、或所有剩余任务都被外部事件阻塞时是真实存在的问题。60 秒窗口足够长，不会因为任务完成到新任务创建之间的短暂间隔而导致过早关闭；又足够短，不会让闲置队友浪费资源。

替代方案的致命缺陷：

没有超时（永远等待）会产生僵尸进程；过短的超时（5 秒）会在领导正在思考或打字时导致过早退出；心跳系统（领导定期 ping 队友以保持其存活）可以工作，但会增加协议复杂度。60 秒固定超时是平衡误退出和资源浪费的良好默认值。

3. 上下文压缩后重新注入队友身份

核心设计：自动压缩对话时，生成的摘要会丢失关键元数据：队友的名称、所属团队和 agent_id。没有这些信息，队友无法认领任务（任务文件以 agent_id 为键）、无法检查收件箱，也无法在消息中表明身份。因此每次自动压缩后，系统会向对话中重新注入一个结构化的身份块："你是 [team] 团队的 [name]，你的 agent_id 是 [id]，你的收件箱在 [path]"。这是队友在记忆丢失后保持功能所需的最小上下文。

替代方案的致命缺陷：

将身份信息放在系统提示中（可在压缩中存活）可以避免这个问题，但违反了 s05 中缓存友好的静态系统提示设计；在摘要提示中嵌入身份（“总结时，始终包含你的名字和团队”）是不可靠的 ——LLM 可能会省略它。显式的压缩后注入是确定性的，保证能工作。

---

三、系统整体架构与工作原理

1. 队友生命周期：WORK 与 IDLE 双阶段

+-------+
| spawn |
+---+---+
    |
    v
+-------+   tool_use     +-------+
| WORK  | <------------- |  LLM  |
+---+---+                +-------+
    |
    | stop_reason != tool_use (或 idle tool called)
    v
+--------+
|  IDLE  |  poll every 5s for up to 60s
+---+----+
    |
    +---> check inbox --> message? ----------> WORK
    |
    +---> scan .tasks/ --> unclaimed? -------> claim -> WORK
    |
    +---> 60s timeout ----------------------> SHUTDOWN

• WORK 阶段：队友执行任务，处理工具调用和消息，状态为 working
• IDLE 阶段：队友每 5 秒轮询一次，检查收件箱和任务看板，状态为 idle
  • 若收到消息或找到未认领任务，返回 WORK 阶段
  • 若 60 秒内无任务无消息，自动关闭线程，状态变为 shutdown

2. 关键组件与实现细节

(1) 任务看板扫描与认领

def scan_unclaimed_tasks() -> list:
    """扫描任务看板，寻找可认领的任务"""
    TASKS_DIR.mkdir(exist_ok=True)
    unclaimed = []
    for f in sorted(TASKS_DIR.glob("task_*.json")):
        task = json.loads(f.read_text())
        # 筛选条件：pending状态、无owner、无未完成依赖
        if (task.get("status") == "pending"
                and not task.get("owner")
                and not task.get("blockedBy")):
            unclaimed.append(task)
    return unclaimed

def claim_task(task_id: int, owner: str) -> str:
    """认领任务，更新状态和owner"""
    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())
        # 校验任务状态和owner，避免重复认领
        if task.get("owner"):
            existing_owner = task.get("owner") or "someone else"
            return f"Error: Task {task_id} has already been claimed by {existing_owner}"
        if task.get("status") != "pending":
            status = task.get("status")
            return f"Error: Task {task_id} cannot be claimed because its status is '{status}'"
        if task.get("blockedBy"):
            return f"Error: Task {task_id} is blocked by other task(s) and cannot be claimed yet"
        # 更新任务信息
        task["owner"] = owner
        task["status"] = "in_progress"
        path.write_text(json.dumps(task, indent=2))
    return f"Claimed task #{task_id} for {owner}"

(2) 空闲阶段轮询逻辑

def _idle_poll(self, name, messages):
    """IDLE阶段轮询：检查收件箱和任务看板"""
    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  # 收到消息，返回WORK阶段
        # 检查任务看板
        unclaimed = scan_unclaimed_tasks()
        if unclaimed:
            task = unclaimed[0]
            claim_result = claim_task(task["id"], name)
            if claim_result.startswith("Error:"):
                continue  # 认领失败（如被其他队友抢先），继续轮询
            # 认领成功，将任务信息加入对话
            task_prompt = (
                f"<auto-claimed>Task #{task['id']}: {task['subject']}\n"
                f"{task.get('description', '')}</auto-claimed>"
            )
            # 若对话过短（压缩后），注入身份块
            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."})
            messages.append({"role": "user", "content": task_prompt})
            messages.append({"role": "assistant", "content": f"Claimed task #{task['id']}. Working on it."})
            return True  # 认领任务，返回WORK阶段
    return False  # 超时无任务无消息，关闭线程

(3) 身份重注入实现

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>",
    }

# 在认领任务时注入
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."})

(4) 队友工具集扩展

def _teammate_tools(self) -> list:
    return [
        # 基础工具（bash、read_file等）
        {"name": "bash", ...},
        {"name": "read_file", ...},
        {"name": "write_file", ...},
        {"name": "edit_file", ...},
        # 通信工具
        {"name": "send_message", ...},
        {"name": "read_inbox", ...},
        # 协议工具
        {"name": "shutdown_response", ...},
        {"name": "plan_approval", ...},
        # 新增自治工具
        {"name": "idle", 
         "description": "Signal that you have no more work. Enters idle polling phase.",
         "input_schema": {"type": "object", "properties": {}}},
        {"name": "claim_task", 
         "description": "Claim a task from the task board by ID.",
         "input_schema": {"type": "object", "properties": {"task_id": {"type": "integer"}}, "required": ["task_id"]}},
    ]

(5) 执行流程

---

四、与 Team Protocols（s10）的关键变更对比

组件	之前（s10 Team Protocols）	之后（s11 Autonomous Agents）
工具集	12 个工具	14 个工具（新增 idle、claim_task）
任务分配	领导手动指派	队友自动认领未分配任务
空闲阶段	无	每 5 秒轮询一次，持续 60 秒
自治性	被动接收任务	主动寻找任务，自组织协作
身份处理	系统提示中固定	压缩后自动重注入身份块
资源治理	无超时控制	60 秒空闲自动关闭线程

---

五、核心优势与创新点

1. 去中心化自组织协作：无需领导手动分配任务，队友主动扫描看板认领任务，系统扩展性大幅提升
2. 简单可靠的轮询机制：基于文件系统的轮询实现简单、跨平台、无依赖，避免了事件驱动的复杂性和可靠性问题
3. 空闲自动治理，资源高效利用：60 秒超时自动关闭闲置队友，避免僵尸进程浪费资源，同时不会过早关闭
4. 压缩后身份恢复，功能不中断：通过身份重注入解决了上下文压缩导致的身份丢失问题，队友始终知道自己是谁、属于哪个团队
5. 线程安全的任务认领：通过锁机制防止多队友同时认领同一任务，保证任务分配的一致性

---

六、运行示例

场景：自组织任务认领流程

1. 领导创建 3 个未分配任务：
   • #1 实现用户登录接口（状态：pending，无 owner，无 blockedBy）
   • #2 编写登录接口单元测试（状态：pending，无 owner，无 blockedBy）
   • #3 重构数据库查询逻辑（状态：pending，无 owner，无 blockedBy）
2. 两个队友 coder 和 tester 处于 IDLE 阶段，每 5 秒轮询一次任务看板
3. coder 先扫描到任务 #1，调用 claim_task(1, "coder") 认领成功，任务状态变为 in_progress，owner 设为 coder
4. tester 下一次轮询时，扫描到任务 #2，调用 claim_task(2, "tester") 认领成功，开始编写测试
5. coder 完成任务 #1 后，再次进入 IDLE 阶段，轮询时扫描到任务 #3，认领并开始工作
6. 所有任务完成后，两个队友在 60 秒超时后自动关闭，无闲置资源浪费

---

七、可扩展方向

• 任务优先级认领：为任务添加优先级字段，队友优先认领高优先级任务
• 基于角色的任务过滤：队友根据自己的角色（如后端 / 测试）认领匹配的任务，避免跨角色认领
• 任务负载均衡：队友根据当前任务负载调整认领频率，避免单个队友认领过多任务
• 超时任务自动回收：长时间未完成的任务自动回收为未认领状态，供其他队友认领
• 任务推荐机制：领导可在任务文件中添加推荐队友字段，队友优先认领推荐给自己的任务