Claude Code 11：Autonomous Agents——自组织团队与动态抢单

在中心化多智能体团队中，所有任务的生命周期都依赖 Leader 进行人工调度：Leader 拆解任务、Leader 挨个点名分配、Leader 轮询等待每个 Worker 汇报。

这种“主管加员工”的中心化模式在面对几十个模块并行改造时，会迅速遇到Leader 吞吐瓶颈：

  1. Leader 上下文严重拥挤：Leader 需要记住每个 Worker 正在干什么，频繁派发指令导致 Leader 上下文迅速触顶；
  2. 算力严重空转：某个 Worker 花了 10 秒跑完任务，而 Leader 正在思考复杂架构，Worker 必须干等几分钟直到 Leader 回复并分发下一道指令；
  3. 容错单点故障：一旦负责分配任务的 Leader 进程挂死，整个多智能体集群全部停摆。

解决这一瓶颈的终极方向是自组织团队（Autonomous Self-Organizing Teams）：将任务全量推入共享的持久化任务池，每个 Worker 智能体自主轮询、竞争认领并推进任务。

本文深入剖析 Claude Code 中的 Coordinator Mode 设计，并从零实现一个带原子锁并发防冲突的动态抢单（Task Claiming）引擎。

生产源码探秘：Claude Code 的 Coordinator Mode

在 Claude Code 源码 src/coordinator/coordinatorMode.ts 中，系统支持将主 Agent 切换为高级编排者（Coordinator）：

1. 编译时与运行时双重门控

1
2
3
4
5
6
7
8
9
// coordinatorMode.ts 生产源码
export function isCoordinatorMode(): boolean {
  // 第一道门：Bun 打包阶段的宏定义特性开关
  if (feature('COORDINATOR_MODE')) {
    // 第二道门：运行时环境变量显式开启
    return isEnvTruthy(process.env.CLAUDE_CODE_COORDINATOR_MODE)
  }
  return false
}

这种双重门控在工程上确保了实验性多 Agent 特性既不会在通用生产包中泄漏，又能根据运行环境平滑激活。

2. 会话模式自适应恢复（Session Mode Recovery）

当用户恢复一个旧会话时，当前的外部环境变量可能与旧会话创建时不一致。源码中规定：会话元数据中的模式优先于外部系统环境。系统会自动检测并覆写 process.env，确保会话恢复后的上下文一致性。

3. Coordinator 的角色契约

在协调者模式下，主 Agent 的系统提示词被完全覆写为统筹人设：

“你是一个协调者（Coordinator）。你的职责不是亲自编写代码，而是将用户的宏观目标拆解，指挥 Workers 负责调研、实现和验证，汇聚最终成果并向用户汇报。”
同时，Coordinator 的本地代码写工具被冻结，只保留 Agent（拉起 Worker）与 SendMessage（下发总体约束）两个统筹工具。

动态抢单机制（Task Claiming）：从被动推到主动拉

自组织团队的核心变化是将任务分配从“Leader 推送（Push）”转变为“Worker 主动拉取（Pull）”：

并发冲突挑战：原子排他锁

当两个 Worker 同时空闲，并在同一毫秒扫描到“任务 A 已就绪”时，如果缺乏并发保护，两者都会判定自己可以认领任务 A，造成重复执行与代码覆盖。

必须通过物理文件锁（File Locking）或重命名原子操作实现 Compare-And-Swap（CAS），确保有且仅有一个 Worker 能够抢单成功。

极简 Python 实现：支持原子抢单的自组织引擎

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import os
import json
import time
from pathlib import Path
from typing import Optional

class AutonomousTaskEngine:
    def __init__(self, task_dir: str = "./.shared_tasks"):
        self.task_dir = Path(task_dir)
        self.task_dir.mkdir(parents=True, exist_ok=True)

    def atomic_claim_task(self, worker_id: str) -> Optional[dict]:
        """
        原子抢单核心算法：
        扫描所有处于 pending 且无未完成依赖的任务，尝试将其锁定为 running 并打上 worker_id 烙印
        """
        for task_file in sorted(self.task_dir.glob("*.json")):
            lock_file = task_file.with_suffix(".lock")
            try:
                # 1. 尝试以排他方式创建锁文件 (O_CREAT | O_EXCL 原生原子操作)
                fd = os.open(str(lock_file), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
                os.close(fd)
            except FileExistsError:
                # 另一个 Worker 正在检查该任务，跳过
                continue

            try:
                # 2. 持有锁后读取任务内容
                with open(task_file, "r", encoding="utf-8") as f:
                    task = json.load(f)

                # 检查是否仍处于待处理状态
                if task["status"] == "pending":
                    # 检查前置依赖是否全部完成
                    deps_met = self._check_dependencies(task.get("blocked_by", []))
                    if deps_met:
                        # 3. 抢单成功，写入 Worker 归属标记并翻转状态
                        task["status"] = "running"
                        task["owner"] = worker_id
                        task["claimed_at"] = time.time()
                        
                        with open(task_file, "w", encoding="utf-8") as f:
                            json.dump(task, f, ensure_ascii=False, indent=2)
                        
                        print(f"[{worker_id}] 成功原子认领任务: [{task['id']}] {task['description']}")
                        return task
            finally:
                # 4. 无论成功与否，释放临时锁文件
                if lock_file.exists():
                    lock_file.unlink()

        return None

    def _check_dependencies(self, blocked_by: list) -> bool:
        for dep_id in blocked_by:
            dep_path = self.task_dir / f"{dep_id}.json"
            if not dep_path.exists():
                return False
            with open(dep_path, "r", encoding="utf-8") as f:
                d = json.load(f)
            if d.get("status") != "completed":
                return False
        return True

    def mark_completed(self, task_id: str):
        """将任务标记为完成，自然解锁依赖该任务的下游节点"""
        path = self.task_dir / f"{task_id}.json"
        with open(path, "r", encoding="utf-8") as f:
            task = json.load(f)
        task["status"] = "completed"
        task["completed_at"] = time.time()
        with open(path, "w", encoding="utf-8") as f:
            json.dump(task, f, ensure_ascii=False, indent=2)

自治 Worker 的常驻循环

每个 Worker 智能体不再等待他人指示，而是运行一个极简的自主生命周期：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def worker_autonomous_loop(worker_id: str, engine: AutonomousTaskEngine):
    print(f"Worker [{worker_id}] 已进入自组织抢单模式...")
    while True:
        # 1. 尝试从公共池中认领可执行任务
        task = engine.atomic_claim_task(worker_id)
        if not task:
            # 暂无可执行任务，退避休眠 2 秒后继续探测
            time.sleep(2)
            continue

        # 2. 调动本 Worker 的私有 Agent Loop 执行具体编码
        print(f"[{worker_id}] 开始推进: {task['description']}")
        # 模拟执行代码修改与测试验证...
        time.sleep(3) 

        # 3. 任务顺利达成，回写终态并自动解锁下游拓扑
        engine.mark_completed(task["id"])
        print(f"[{worker_id}] 任务 [{task['id']}] 已闭环，继续寻找下一个目标！")

在这套架构下，整个系统实现了高度弹性：

  • 系统需要加速时，只需直接在新的终端或容器中多启动 3 个 Worker 进程，它们会自动加入抢单，整个过程零协调成本；
  • 某个 Worker 意外崩溃，其他 Worker 探测到锁超时后可以自动回收重试，具备极强的容灾韧性。

总结

自组织团队代表了多智能体系统从“命令控制”向“市场驱动”的演进：

  1. 去中心化解耦：Leader 不再做微观任务指派，仅负责顶层目标拆解与最终验收；
  2. 基于原子锁的动态抢单：利用排他文件锁解决并发竞争，确保每个任务有且仅被一个 Worker 执行；
  3. 弹性水平伸缩：新加入的 Worker 随开随跑，无需修改任何集群拓扑。

然而，当多个自治的 Worker 同时开始跑起来、并行修改本地代码时，一个终极的物理冲突随之而来：如果两个 Worker 同时对 src/index.ts 进行写操作，文件系统立刻会出现毁灭性的写踩踏！

下一篇也是本系列的终局篇：Worktree 隔离——用独立的 Git 工作树彻底消除并发冲突。