# 第九章 Agent Teams (智能体团队)


`s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > [ s09 ] s10 > s11 > s12`

>"本专栏基于开源项目 `learn-claude-code` 的官方文档。原文档非常硬核，为了方便像我一样的新手小白理解，我对文档进行了**逐行精读**，并加入了很多**中文注释、大白话解释和踩坑记录**。希望这套'咀嚼版'教程能帮你推开 AI Agent 开发的大门。"


项目地址：[shareAI-lab/learn-claude-code: Bash is all you need - A nano Claude Code–like agent, built from 0 to 1](https://github.com/shareAI-lab/learn-claude-code)

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


## 一、问题：子智能体是一次性的、后台任务是无脑的


### s04 的问题：子智能体是"一次性"的


子智能体的生命周期是：

```
生成 → 干活 → 返回摘要 → 被销毁
```
它没有身份、没有名字、不能跨轮对话存活、不能互相通信。父智能体说"去读五个文件总结一下"，子智能体做完就没了，下次需要再重新生成一个全新的。

对应的代码在 `s04` 里可以看到：

```
def run_subagent(prompt: str) -> str:
    sub_messages = [{"role": "user", "content": prompt}]  # 全新的空上下文
    for _ in range(30):
        response = client.messages.create(...)
        if response.stop_reason != "tool_use":
            break
        # execute tools...
    return "".join(b.text for b in response.content...)  # 返回摘要，函数结束
# 函数返回后，sub_messages 被丢弃，这个子智能体就"死了"
```
**没有 `client.messages.create()` 的循环调用，就没有"脑子"。** s08 的后台线程就更直接了：

### s08 的问题：后台线程是"哑执行器"


```
def _execute(self, task_id, command):
    r = subprocess.run(command, shell=True, ...)  # 直接跑 shell 命令
    output = (r.stdout + r.stderr).strip()[:50000]
    self._notification_queue.append({"task_id": task_id, "result": output[:500]})
    # 函数结束，没有任何 LLM 调用
```
整个函数里**没有 `client.messages.create()`**，所以它不会思考、不会做决策、不会根据输出调整下一步。命令是什么就跑什么，跑完就完了。

### 到底缺什么？


|需求|s04子智能体|s08后台任务|
|:-:|:-:|:-:|
|有 LLM（能思考）|✅|❌|
|有持久身份|❌|❌|
|能跨轮存活|❌|❌|
|能互相通信|❌|❌|
|有完整 agent loop|✅|❌|

s09 要做的就是：**既有 s04 的"脑子"（LLM 循环），又有 s08 的"持久性"（线程不销毁），再加上智能体之间的通信能力。**

## 二、解决方案


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

Communication:
  .team/
    config.json           <- 团队名册 + 状态
    inbox/
      alice.jsonl         <- append-only, drain-on-read
      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** — 管理队友的创建、生命周期、线程

2. **MessageBus** — 基于 JSONL 文件的消息收发系统

3. **_teammate_loop** — 每个队友的"大脑循环"（包含 LLM 调用）


## 三、工作原理


### 3.1 TeammateManager：团队管理器


`TeammateManager` 通过 `config.json` 维护团队名册，记录每个队友的 name、role、status。

```
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 = {}                   # 内存中记录每个队友的线程对象

    def _load_config(self) -> dict:
        if self.config_path.exists():
            return json.loads(self.config_path.read_text())
        return {"team_name": "default", "members": []}  # 空团队

    def _save_config(self):
        self.config_path.write_text(json.dumps(self.config, indent=2))  # 持久化到磁盘
```
`config.json` 长这样：

```
{
  "team_name": "default",
  "members": [
    {"name": "alice", "role": "coder", "status": "working"},
    {"name": "bob", "role": "tester", "status": "idle"}
  ]
}
```
>**为什么用文件而不是内存？** 因为配置文件持久化到磁盘，进程重启后团队名册还在。


### 3.2 spawn()：创建队友


`spawn()` 做三件事：注册到 config → 标记为 working → 启动线程。

```
def spawn(self, name: str, role: str, prompt: str) -> str:
    member = self._find_member(name)
    if member:
        # 已存在的队友，只有 idle（空闲）/shutdown（关闭） 状态才能重新 spawn
        if member["status"] not in ("idle", "shutdown"):
            return f"Error: '{name}' is currently {member['status']}"
        member["status"] = "working"
        member["role"] = role
    else:
        # 新队友，追加到成员列表
        member = {"name": name, "role": role, "status": "working"}
        self.config["members"].append(member)
    self._save_config()  # 写入 config.json

    # 关键：在新线程中启动 teammate_loop
    thread = threading.Thread(
        target=self._teammate_loop,
        args=(name, role, prompt),
        daemon=True,  # 守护线程，主进程退出时自动结束
    )
    self.threads[name] = thread
    thread.start()
    return f"Spawned '{name}' (role: {role})"
```
>**对比 s04：** s04 的 `run_subagent()` 是一个函数调用，阻塞等待完成。s09 的 `spawn()` 启动线程后**立即返回**，队友在后台独立运行。


### 3.3 MessageBus：消息收件箱


所有数据都存在磁盘上的 JSONL 文件中：

```
.team/inbox/
  alice.jsonl    <- 每行一条 JSON 消息
  bob.jsonl
  lead.jsonl
```
```
class MessageBus:
    def __init__(self, inbox_dir: Path):
        self.dir = inbox_dir
        self.dir.mkdir(parents=True, exist_ok=True)

    def send(self, sender: str, to: str, content: str,
             msg_type: str = "message", extra: dict = None) -> str:
        if msg_type not in VALID_MSG_TYPES:
            return f"Error: Invalid type '{msg_type}'. Valid: {VALID_MSG_TYPES}"
        msg = {
            "type": msg_type,
            "from": sender,
            "content": content,
            "timestamp": time.time(),
        }
        if extra:
            msg.update(extra)
        # 往收件人的 .jsonl 文件末尾追加一行
        inbox_path = self.dir / f"{to}.jsonl"
        with open(inbox_path, "a") as f:  # "a" = append 模式
            f.write(json.dumps(msg) + "\n")
        return f"Sent {msg_type} to {to}"

    def read_inbox(self, name: str) -> list:
        inbox_path = self.dir / f"{name}.jsonl"
        if not inbox_path.exists():
            return []
        messages = []
        for line in inbox_path.read_text().strip().splitlines():
            if line:
                messages.append(json.loads(line))
        inbox_path.write_text("")  # drain：读完就清空
        return messages

    def broadcast(self, sender: str, content: str, teammates: list) -> str:
        count = 0
        for name in teammates:
            if name != sender:
                self.send(sender, name, content, "broadcast")
                count += 1
        return f"Broadcast to {count} teammates"
```
>**为什么选 JSONL 而不是 JSON 数组？**
- JSONL 的 `append` 操作文件系统级别保证原子性，不需要加锁
- 无需先读出整个数组再写回，直接追加一行即可
- 每个队友有独立文件，互相之间不会冲突


### 3.4 _teammate_loop：队友的"大脑"（这是核心！）


这是 s09 和 s04/s08 最关键的区别。**每个队友跑的是完整的 agent loop，里面有 LLM 调用。**

```
def _teammate_loop(self, name: str, role: str, prompt: str):
    # 设置系统提示，告诉 LLM "你是谁、你是什么角色"
    sys_prompt = (
        f"You are '{name}', role: {role}, at {WORKDIR}. "
        f"Use send_message to communicate. Complete your task."
    )
    messages = [{"role": "user", "content": prompt}]
    tools = self._teammate_tools()  # 这个队友能用的所有工具

    for _ in range(50):  # 最多 50 轮，防止死循环
        # 第一步：检查收件箱，看看有没有别人发来的消息
        inbox = BUS.read_inbox(name)
        for msg in inbox:
            messages.append({"role": "user", "content": json.dumps(msg)})

        # 第二步：调用 LLM —— 这就是"脑子"所在
        try:
            response = client.messages.create(
                model=MODEL,
                system=sys_prompt,
                messages=messages,
                tools=tools,
                max_tokens=8000,
            )
        except Exception:
            break

        messages.append({"role": "assistant", "content": response.content})

        # LLM 没有调用工具 = 任务完成，退出循环
        if response.stop_reason != "tool_use":
            break

        # 第三步：执行 LLM 请求的工具调用
        results = []
        for block in response.content:
            if block.type == "tool_use":
                output = self._exec(name, block.name, block.input)
                print(f"  [{name}] {block.name}: {str(output)[:120]}")
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(output),
                })
        # 把工具执行结果喂回给 LLM，让它决定下一步
        messages.append({"role": "user", "content": results})

    # 循环结束，标记为 idle
    member = self._find_member(name)
    if member and member["status"] != "shutdown":
        member["status"] = "idle"
        self._save_config()
```
**这个循环和 s04 的 `run_subagent()` 几乎一样**，区别在于：


- s04 的子智能体执行完返回摘要就销毁了

- s09 的队友执行完后只是变成 `idle` 状态，线程结束，但 `config.json` 里保留了它的记录，下次可以 `spawn()` 唤醒它


### 3.5 队友能用哪些工具？


```
def _teammate_tools(self) -> list:
    return [
        # 基础工具（继承自 s02）
        {"name": "bash", ...},          # 跑 shell 命令
        {"name": "read_file", ...},     # 读文件
        {"name": "write_file", ...},    # 写文件
        {"name": "edit_file", ...},     # 编辑文件
        # s09 新增的通信工具
        {"name": "send_message", ...},  # 给其他队友发消息
        {"name": "read_inbox", ...},    # 读取自己的收件箱
    ]
```
>**注意：** 队友没有 `spawn_teammate` 工具，只有 leader 才能创建新队友。这避免了无限递归生成。


### 3.6 领导的 9 个工具


领导（lead）比队友多了 3 个管理工具：队友的工具都是定义在定义在 TeammateManager 类里，而领导的工具是在TOOL_HANDLERS中。

```
TOOL_HANDLERS = {
    # 基础工具（4 个）
    "bash":            lambda **kw: _run_bash(kw["command"]),
    "read_file":       lambda **kw: _run_read(kw["path"], kw.get("limit")),
    "write_file":      lambda **kw: _run_write(kw["path"], kw["content"]),
    "edit_file":       lambda **kw: _run_edit(kw["path"], kw["old_text"], kw["new_text"]),
    # 团队管理工具（3 个）
    "spawn_teammate":  lambda **kw: TEAM.spawn(kw["name"], kw["role"], kw["prompt"]),
    "list_teammates":  lambda **kw: TEAM.list_all(),
    "broadcast":       lambda **kw: BUS.broadcast("lead", kw["content"], TEAM.member_names()),
    # 通信工具（2 个）
    "send_message":    lambda **kw: BUS.send("lead", kw["to"], kw["content"], kw.get("msg_type", "message")),
    "read_inbox":      lambda **kw: json.dumps(BUS.read_inbox("lead"), indent=2),
}
```
### 3.7 数据流完整示例


假设用户说 "Spawn alice and bob, have alice write a hello function, bob test it"：

```
用户 prompt
    │
    ▼
Lead agent loop
    │
    ├─ 调用 spawn_teammate("alice", "coder", "write a hello function")
    │     → 新线程启动 alice 的 teammate_loop
    │     → config.json: alice.status = "working"
    │
    ├─ 调用 spawn_teammate("bob", "tester", "test the hello function")
    │     → 新线程启动 bob 的 teammate_loop
    │     → config.json: bob.status = "working"
    │
    ▼
alice 线程（独立运行）              bob 线程（独立运行）
    │                                  │
    ├─ LLM 决定: 写 hello.py          ├─ LLM 决定: 等 alice 完成
    ├─ 调用 write_file                ├─ 调用 read_inbox（空的）
    ├─ 调用 send_message("bob",       ├─ LLM 决定: 等消息
    │   "hello function written")      │
    │     → bob.jsonl 追加一行 JSON    │
    ├─ LLM 决定: 完成                 │
    ├─ status = "idle"                │
                                      ├─ read_inbox 读到消息
                                      ├─ LLM 决定: 读 hello.py 并写测试
                                      ├─ 调用 read_file("hello.py")
                                      ├─ 调用 write_file("test_hello.py")
                                      ├─ 调用 bash("pytest test_hello.py")
                                      ├─ LLM 决定: 完成
                                      ├─ status = "idle"
```
## 四、相对 s08 的变更


|组件|之前(s08)|之后(s09)|
|:-:|:-:|:-:|
|Tools|6|9 (+spawn/send/read_inbox)|
|智能体数量|单一|领导 + N 个队友|
|持久化|无|config.json + JSONL 收件箱|
|线程|后台命令|每线程完整 agent loop|
|生命周期|一次性|idle -> working -> idle|
|通信|无|message + broadcast|

## 五、关键代码对比


### s08 后台线程：没有 LLM，不能思考


```
def _execute(self, task_id, command):
    r = subprocess.run(command, shell=True, ...)
    # ↑ 只有 shell 调用，没有 client.messages.create()
    # ↓ 执行完就完了，不会根据结果做决策
    output = (r.stdout + r.stderr).strip()
    self._notification_queue.append({"task_id": task_id, "result": output})
```
### s09 队友线程：有 LLM，能思考


```
def _teammate_loop(self, name, role, prompt):
    messages = [{"role": "user", "content": prompt}]
    for _ in range(50):
        inbox = BUS.read_inbox(name)  # 检查收件箱
        for msg in inbox:
            messages.append({"role": "user", "content": json.dumps(msg)})

        response = client.messages.create(...)  # ← 这就是"脑子"
        # LLM 看到上下文后自主决定：
        # - 要不要跑 shell？
        # - 跑完结果不对要不要换个方式？
        # - 要不要给别的队友发消息？
        # - 要不要读个文件看看？
        if response.stop_reason != "tool_use":
            break
        # 执行工具，把结果喂回 LLM，继续循环...
    self._find_member(name)["status"] = "idle"
```
**一句话总结：s08 是"执行一段命令"，s09 是"让一个有脑子的 agent 去完成一个任务"。**

## 六、5 种消息类型


代码中定义了 5 种消息类型（s09 中实现了 2 种，其余为 s10 预留）：

```
VALID_MSG_TYPES = {
    "message",               # 普通文本消息（s09 实现）
    "broadcast",             # 群发给所有队友（s09 实现）
    "shutdown_request",      # 请求优雅关闭（s10 预留）
    "shutdown_response",     # 批准/拒绝关闭请求（s10 预留）
    "plan_approval_response", # 批准/拒绝计划（s10 预留）
}
```
## 七、试一试


```
cd learn-claude-code
python agents/s09_agent_teams.py
```
试试这些 prompt（英文 prompt 对 LLM 效果更好，也可以用中文）：


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`

4. 输入 `/team` 查看团队名册和状态

5. 输入 `/inbox` 手动检查领导的收件箱

s01 > s02 > s03 > s04 > s05 > s06 | s07 > s08 > [ s09 ] s10 > s11 > s12

【从零手写 ClaudeCode：learn-claude-code 项目实战笔记】（9）Agent Teams (智能体团队)

本章构建持久化智能体团队，解决一次性子智能体和哑后台任务的问题。每个队友拥有独立线程与 LLM 循环，通过 JSONL 文件邮箱实现跨轮通信；TeammateManager 维护团队名册与生命周期，MessageBus 提供线程安全的消息收发，支持并行协作、任务分派与状态同步，为大型项目多 Agent 协同奠定基础。

人工智能

编程语言

后端

本专栏详解AI智能体团队开发技术，基于开源项目learn-claude-code官方文档，提供中文注释和实战解析。重点讲解智能体团队管理、消息通信机制和持久化设计，包含config.json团队配置、JSONL消息收件箱实现、完整agentloop工作流程等核心技术。适合AI开发者学习智能体协作系统架构与实现原理。

文件系统

数据流

智能体

JSON

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【从零手写 ClaudeCode：learn-claude-code 项目实战笔记】（9）Agent Teams (智能体团队)-腾讯云开发者社区-腾讯云

【从零手写 ClaudeCode：learn-claude-code 项目实战笔记】（9）Agent Teams (智能体团队)

【从零手写 ClaudeCode：learn-claude-code 项目实战笔记】（9）Agent Teams (智能体团队)

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐