Admin：创建页面，内容为“
"任务太大一个人干不完，要能分给队友" —— 持久化队友 + JSONL 邮箱。
前面的子智能体是一次性的：生成、干活、返回摘要、消亡。没有身份，没有跨调用的记忆。真正的团队协作需要三样东西： # '''能跨多轮对话存活的持久智能体''' # '''身份和生命周期管理''' # '''智能体之间的通信通道''' == 一、团队架构 == `Teammate lifecycle…”`


2026-03-23T05:04:36Z
创建页面，内容为“<blockquote>"任务太大一个人干不完，要能分给队友" —— 持久化队友 + JSONL 邮箱。</blockquote>前面的子智能体是一次性的：生成、干活、返回摘要、消亡。没有身份，没有跨调用的记忆。  真正的团队协作需要三样东西：  # '''能跨多轮对话存活的持久智能体''' # '''身份和生命周期管理''' # '''智能体之间的通信通道'''  == 一、团队架构 ==  <code>Teammate lifecycle…”
新页面
<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 是宇宙所需要的全部。''
多智能体协作实践：从单打独斗到团队作战（构建自己的AI Agent) - 版本历史
