查看“︁多智能体协作实践：从单打独斗到团队作战（构建自己的AI Agent)”︁的源代码

<blockquote>"任务太大一个人干不完，要能分给队友" —— 持久化队友 + JSONL 邮箱。</blockquote>前面的子智能体是一次性的：生成、干活、返回摘要、消亡。没有身份，没有跨调用的记忆。

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

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

== 一、团队架构 ==
 <code>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 ---------+</code>

== 二、核心组件 ==

=== 1. TeammateManager ===
通过 <code>config.json</code> 维护团队名册：
 <code>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 = {}</code>

=== 2. 生成队友 ===
<code>spawn()</code> 创建队友并在线程中启动 Agent 循环：
 <code>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})"</code>

=== 3. MessageBus ===
追加式 JSONL 收件箱：

* <code>send()</code> 追加一行
* <code>read_inbox()</code> 读取全部并清空

 <code>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)</code>

=== 4. 队友循环 ===
每个队友在每次 LLM 调用前检查收件箱：
 <code>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"</code>

== 三、试一试 ==
 <code>cd learn-claude-code
 python agents/s09_agent_teams.py</code>
试试这些 prompt：

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

也可以使用快捷命令：

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

== 四、团队协作模式 ==

=== 模式 1：领导指派 ===
 <code>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"</code>

=== 模式 2：广播通知 ===
 <code>Leader: Broadcast "Phase 1 complete, starting Phase 2"
 Alice:  (receives broadcast)
 Bob:    (receives broadcast)</code>

=== 模式 3：点对点协作 ===
 <code>Alice: Send bob "Can you check the API spec?"
 Bob:   (checks spec)
 Bob:   Send alice "Spec says POST /api/v1/users"</code>

== 五、进阶：自治智能体 ==
s09 的队友只在被明确指派时才动。领导得给每个队友写 prompt。

在 s11 中，我们将升级到'''自治智能体'''：<blockquote>"队友自己看看板，有活就认领"</blockquote>队友自己扫描任务看板，认领没人做的任务，做完再找下一个。

=== 空闲阶段轮询 ===
 <code>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''</code>

=== 自组织工作流 ===
 <code>1. 创建任务看板
 2. 生成多个队友
 3. 队友自动扫描看板
 4. 认领可执行任务
 5. 完成后继续扫描
 6. 60秒空闲后自动关闭</code>
'''不需要领导逐个分配，真正的自组织。'''

== 六、试一试（自治版） ==
 <code>cd learn-claude-code
 python agents/s11_autonomous_agents.py</code>
试试这些 prompt：

# <code>Create 3 tasks on the board, then spawn alice and bob. Watch them auto-claim.</code>
# <code>Spawn a coder teammate and let it find work from the task board itself</code>
# <code>Create tasks with dependencies. Watch teammates respect the blocked order.</code>

快捷命令：

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

== 七、设计哲学 ==
'''多模型，通过文件协调。'''

这是 Harness 工程的精髓：

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

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

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

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

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

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

'''优秀的 Harness 工程师是环境设计师'''——创造能让智能高效协作的基础设施。
----''Bash 就够了。真正的 Agent 是宇宙所需要的全部。''