多智能体协作实践：从单打独斗到团队作战（构建自己的AI Agent)

"任务太大一个人干不完，要能分给队友" —— 持久化队友 + JSONL 邮箱。

前面的子智能体是一次性的：生成、干活、返回摘要、消亡。没有身份，没有跨调用的记忆。

真正的团队协作需要三样东西：

1. 能跨多轮对话存活的持久智能体
2. 身份和生命周期管理
3. 智能体之间的通信通道

一、团队架构

Teammate lifecycle:
  spawn -> WORKING -> IDLE -> WORKING -> ... -> SHUTDOWN

Communication:
  .team/
    config.json           <- 团队名册 + 状态
    inbox/
      alice.jsonl         <- 追加式，读取即清空
      bob.jsonl
      lead.jsonl

              +--------+    send("alice","bob","...")    +--------+
              | alice  | -----------------------------> |  bob   |
              | loop   |    bob.jsonl << {json_line}    |  loop  |
              +--------+                                +--------+
                   ^                                         |
                   |        BUS.read_inbox("alice")          |
                   +---- alice.jsonl -> read + drain ---------+

二、核心组件

1. TeammateManager

通过 config.json 维护团队名册：

class TeammateManager:
    def __init__(self, team_dir: Path):
        self.dir = team_dir
        self.dir.mkdir(exist_ok=True)
        self.config_path = self.dir / "config.json"
        self.config = self._load_config()
        self.threads = {}

2. 生成队友

spawn() 创建队友并在线程中启动 Agent 循环：

def spawn(self, name: str, role: str, prompt: str) -> str:
    member = {
        "name": name,
        "role": role,
        "status": "working"
    }
    self.config["members"].append(member)
    self._save_config()
    
    thread = threading.Thread(
        target=self._teammate_loop,
        args=(name, role, prompt),
        daemon=True
    )
    thread.start()
    return f"Spawned teammate '{name}' (role: {role})"

3. MessageBus

追加式 JSONL 收件箱：

• send() 追加一行
• read_inbox() 读取全部并清空

class MessageBus:
    def send(self, sender, to, content, 
             msg_type="message", extra=None):
        msg = {
            "type": msg_type,
            "from": sender,
            "content": content,
            "timestamp": time.time()
        }
        if extra:
            msg.update(extra)
        with open(self.dir / f"{to}.jsonl", "a") as f:
            f.write(json.dumps(msg) + "\n")

    def read_inbox(self, name):
        path = self.dir / f"{name}.jsonl"
        if not path.exists():
            return "[]"
        msgs = [json.loads(l) for l in 
                path.read_text().strip().splitlines() if l]
        path.write_text("")  # drain
        return json.dumps(msgs, indent=2)

4. 队友循环

每个队友在每次 LLM 调用前检查收件箱：

def _teammate_loop(self, name, role, prompt):
    messages = [{"role": "user", "content": prompt}]
    for _ in range(50):
        # 检查收件箱
        inbox = BUS.read_inbox(name)
        if inbox != "[]":
            messages.append({
                "role": "user",
                "content": f"<inbox>{inbox}</inbox>"
            })
            messages.append({
                "role": "assistant",
                "content": "Noted inbox messages."
            })
        
        # 调用 LLM
        response = client.messages.create(...)
        
        if response.stop_reason != "tool_use":
            break
        # 执行工具，追加结果...
    
    self._find_member(name)["status"] = "idle"

三、试一试

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

试试这些 prompt：

1. Spawn alice (coder) and bob (tester). Have alice send bob a message.
2. Broadcast "status update: phase 1 complete" to all teammates
3. Check the lead inbox for any messages

也可以使用快捷命令：

• /team 查看团队名册和状态
• /inbox 手动检查领导的收件箱

四、团队协作模式

模式 1：领导指派

Leader: Spawn alice (coder) and bob (reviewer)
Leader: Send alice "Implement user authentication"
Alice:  (works on task)
Alice:  Send leader "Task complete, needs review"
Leader: Send bob "Review alice's auth implementation"
Bob:    (reviews code)
Bob:    Send leader "Review complete, 2 issues found"

模式 2：广播通知

Leader: Broadcast "Phase 1 complete, starting Phase 2"
Alice:  (receives broadcast)
Bob:    (receives broadcast)

模式 3：点对点协作

Alice: Send bob "Can you check the API spec?"
Bob:   (checks spec)
Bob:   Send alice "Spec says POST /api/v1/users"

五、进阶：自治智能体

s09 的队友只在被明确指派时才动。领导得给每个队友写 prompt。

在 s11 中，我们将升级到自治智能体：

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

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

空闲阶段轮询

def _idle_poll(self, name, messages):
    for _ in range(IDLE_TIMEOUT // POLL_INTERVAL):
        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']}</auto-claimed>"
            })
            return True
    
    return False  # timeout -> shutdown

自组织工作流

1. 创建任务看板
2. 生成多个队友
3. 队友自动扫描看板
4. 认领可执行任务
5. 完成后继续扫描
6. 60秒空闲后自动关闭

不需要领导逐个分配，真正的自组织。

六、试一试（自治版）

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

试试这些 prompt：

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.

快捷命令：

• /tasks 查看带 owner 的任务看板
• /team 监控谁在工作、谁在空闲

七、设计哲学

多模型，通过文件协调。

这是 Harness 工程的精髓：

• 不共享内存，避免竞态条件
• 通过文件系统通信，天然持久化
• JSONL 格式简单、可扩展
• 每个 Agent 独立运行，松耦合

这种模式不仅适用于编程，也适用于：

• 客服团队：不同专长的 Agent 处理不同类型的工单
• 研究团队：文献检索、实验设计、数据分析分工协作
• 创意团队：头脑风暴、方案评估、细节执行分层处理

八、总结

从单 Agent → 子 Agent → 多 Agent 团队 → 自治 Agent，Harness 的复杂度逐步提升，但核心原则不变：

Harness 提供机制，模型决定策略。

• Harness 提供 spawn、send、inbox 工具
• 模型决定什么时候生成队友、给谁发消息
• 模型决定认领哪个任务、如何协作

优秀的 Harness 工程师是环境设计师——创造能让智能高效协作的基础设施。

---

Bash 就够了。真正的 Agent 是宇宙所需要的全部。