Agent 编排 — 从单体到群体智能

一个人做不完的事，交给一个团队——Agent 也是

📝 本章目标

读完本章，你将：

  1. 用三个 Prompt 让 AI 帮你实现子任务委托、后台执行和团队协作三种编排模式
  2. 理解委托模式的核心——上下文隔离与结果回传
  3. 掌握后台任务的线程管理和状态追踪
  4. 了解 Swarm 协作中角色分工与交接协议的设计思想

想象一个创业公司的技术总监。早期公司只有他一个人，写代码、改 Bug、跑测试、做代码审查，全是他一个人干。项目小的时候还行，项目一大就扛不住了——他一边写新功能，一边等测试跑完，一边审查昨天的代码，脑子里同时装着三件事，哪件都做不好。

后来公司招了人。他学会了一件事：派活。“小王你去写这个模块，做完了告诉我结果。""小李，测试跑起来要半小时，你先放后台跑着，别干等。""这个需求比较复杂，小王写完代码、小赵审查、小李写测试，三个人按顺序来。”

他自己呢？只需要拆任务、分配、收结果。效率翻了好几倍。

Agent 的编排就是同一件事——一个 AI 搞不定的复杂任务，拆成几个子任务，交给多个 AI 分头去做。别急着理解原理，先把它造出来。

动手：用三个 Prompt 让 AI 学会分工

确保你的项目已经有前几章构建的查询引擎和工具系统。如果没有，去 GitHub 仓库 git checkout ch04-permissions 获取起点。

打开 Claude Code，确认你在项目根目录，然后跟着走。

Prompt 1：让主 AI 能委托子任务

复制下面这段话，粘贴到 Claude Code 里：

有些任务太复杂，一个 AI 搞不定。帮我实现一种机制：
主 AI 可以把一个子任务交给另一个 AI 去做，
子 AI 独立完成后把结果交回来。
就像经理派活给员工，干完了汇报结果。

子 AI 要有自己独立的对话空间，不会搞乱主对话。

等 AI 跑完，它会帮你创建子 Agent 的委托机制——一个新工具，主 AI 调用它就能启动一个独立的子查询。试一下：

$ harness
You > 帮我分析 harness 目录下所有 Python 文件的代码质量，
      然后给出改进建议

Assistant > 我来把这个任务委托给一个专门的分析助手...

[delegate] 启动子任务：分析代码质量
[delegate] 子任务完成，返回结果

Assistant > 分析助手的报告如下：
1. engine.py — 整体结构良好，建议提取重复的错误处理逻辑...
2. tools.py — 工具注册可以用装饰器简化...
...

主 AI 把”分析代码质量”这个子任务委托出去了，子 AI 在自己的独立空间里完成分析，结果交回来后主 AI 继续处理。接下来让它能处理慢任务。

Prompt 2：把慢任务丢到后台

有些任务很慢——比如跑测试、分析大量代码。
不想让 AI 干等着。帮我实现后台机制：
把慢任务丢到后台，主 AI 继续处理别的。
后台完成后能查到结果。
$ harness
You > 后台跑一下所有测试，同时告诉我项目结构

Assistant > 好的，我把测试任务放到后台执行。

[background] 任务 #1 已提交：运行测试
[background] 状态：运行中

Assistant > 先来看项目结构，harness/ 目录下有这些模块...
（继续回答项目结构的问题，不用等测试跑完）

You > 测试跑完了吗？

Assistant > 让我查一下后台任务状态。

[background] 任务 #1：已完成（耗时 45s）
测试结果：12 passed, 0 failed

主 AI 不再干等着了——慢任务在后台跑，它继续处理别的请求。最后一步，让多个 AI 像团队一样协作。

Prompt 3：让多个 AI 像团队那样合作

我想让多个 AI 像团队那样合作。
比如一个写代码一个审代码一个写测试，
各有分工，做完自己那部分交给下一个。
帮我实现这种团队协作模式。
$ harness
You > 用团队模式帮我实现一个缓存模块：
      一个人写代码，一个人审查，一个人写测试

Assistant > 启动团队协作流程...

[swarm] 角色分配：
  - Coder：负责编写缓存模块
  - Reviewer：负责代码审查
  - Tester：负责编写测试

[swarm] Coder 开始工作...
[swarm] Coder 完成，交给 Reviewer...
[swarm] Reviewer 完成审查，发现 2 个建议
[swarm] Coder 根据建议修改...
[swarm] 交给 Tester...
[swarm] Tester 完成，全部测试通过

Assistant > 团队协作完成！
- 缓存模块已创建
- 代码审查通过（2 个建议已采纳）
- 5 个单元测试全部通过

💡 三个 Prompt 做了什么

  • Prompt 1 实现了委托——主 AI 把子任务交给独立的子 AI
  • Prompt 2 实现了后台——慢任务不阻塞，异步执行
  • Prompt 3 实现了团队——多个 AI 角色分工，按流程协作

现在你手里有了一个支持多 Agent 编排的系统。完整代码在 GitHub 仓库，对应 tag ch05-orchestration。

接下来我们回过头，理解你刚刚构建的东西。

深入理解

三种编排模式

你刚刚用三个 Prompt 构建了三种不同的 Agent 编排模式。它们解决的问题不同，适用场景也不同：

表 5-1：三种编排模式对比

模式委托（Delegation）后台（Background）团队（Swarm）
核心思想主 AI 派活给子 AI慢任务异步执行多角色按流程协作
类比经理派活给员工把任务扔进后台队列流水线生产
上下文子 AI 有独立上下文共享主上下文每个角色有专属上下文
适用场景复杂子任务需要专注I/O 密集型慢操作需要多视角的复杂任务
典型例子代码分析、文档生成跑测试、大文件扫描写代码→审查→测试

三种模式不是互斥的——在真实系统中，它们经常组合使用。比如团队模式里的每个角色，都可以把自己的子任务委托出去；慢操作则自动放到后台执行。

理解委托模式

委托是最基础的编排模式。它的核心问题只有一个：怎么让子 AI 干活的时候不把主 AI 的对话搞乱？

答案是：给子 AI 一个独立的对话空间——独立的消息历史、独立的系统提示、独立的工具集。

图 5-1：委托模式的执行流程

  1. 主 Agent — 发现需要委托
  2. 创建子查询 — 独立 QueryState
  3. 子 Agent 执行 — 独立对话空间
  4. 返回结果 — 只传回文本摘要
  5. 主 Agent 继续 — 上下文不受影响

关键：上下文隔离

💡 核心概念：上下文隔离

委托模式的灵魂是上下文隔离。

子 AI 拥有一个全新的对话空间——它看不到主 AI 的聊天记录，主 AI 也看不到子 AI 的中间过程。两者之间唯一的桥梁是任务描述（主 AI 告诉子 AI 该做什么）和结果摘要（子 AI 把结论交回来）。

为什么要隔离？两个原因：

  1. 防止污染——子任务可能调用大量工具、产生几十轮对话，如果全部灌进主对话，主 AI 的上下文会被撑爆
  2. 聚焦专注——子 AI 只看到跟自己任务相关的信息，不会被主对话里的其他内容分散注意力

委托的实现骨架

打开你生成的代码，找到委托工具的实现。去掉辅助代码后，核心逻辑只有十几行：

# harness/orchestration.py — 委托核心
def delegate_task(task_description, tools=None):
    # 1. 创建独立的查询状态
    sub_state = QueryState(
        messages=[{"role": "user",
                   "content": task_description}],
        system="你是一个专注执行子任务的助手。",
        tools=tools or default_tools,
    )

    # 2. 用同一个查询引擎跑子任务
    result = query_loop(sub_state)

    # 3. 只把结果文本返回给主 Agent
    return result

注意第一步——QueryState 是全新的。子 AI 看到的 messages 里只有任务描述，没有主对话的任何历史。这就是隔离。

委托的成本

委托不是免费的。每次委托都是一次新的 API 调用链——子 AI 从零开始理解任务，可能需要多轮工具调用才能完成。这意味着额外的 token 消耗和时间开销。

什么时候值得委托？一个简单判断：如果子任务需要超过 3 轮工具调用，或者会产生大量中间结果，就值得独立出去。

后台任务的实现

后台任务解决的是阻塞问题。有些操作天然就慢——跑测试套件、扫描大型代码库、等待外部 API 响应。如果主 AI 干等着，用户体验极差。

后台任务的核心思想很简单：把慢操作扔到另一个线程，主 AI 继续响应用户。

任务的生命周期

每个后台任务都有明确的状态：

图 5-2：后台任务的状态流转

  • PENDING（等待执行） → 线程启动 → RUNNING（正在执行）
  • RUNNING → 正常结束 → COMPLETED（执行成功）
  • RUNNING → 异常/超时 → FAILED（执行失败）

核心代码

后台任务管理器的骨架非常简洁：

# harness/background.py — 后台任务核心
class BackgroundManager:
    def __init__(self):
        self.tasks = {}
        self.next_id = 1

    def submit(self, func, description):
        task_id = self.next_id
        self.next_id += 1
        self.tasks[task_id] = {
            "status": "PENDING",
            "description": description,
            "result": None,
        }

        def run():
            self.tasks[task_id]["status"] = "RUNNING"
            try:
                result = func()
                self.tasks[task_id]["status"] = "COMPLETED"
                self.tasks[task_id]["result"] = result
            except Exception as e:
                self.tasks[task_id]["status"] = "FAILED"
                self.tasks[task_id]["result"] = str(e)

        Thread(target=run, daemon=True).start()
        return task_id

    def check(self, task_id):
        return self.tasks.get(task_id)

submit() 把任务扔进线程，返回一个 ID；check() 用 ID 查状态。主 AI 只需要两个操作：提交和查询。

后台任务的注意事项

⚠️ 后台任务的陷阱

  1. 线程安全——后台任务和主线程共享进程内存。如果后台任务修改了共享数据（比如文件系统），可能产生竞争条件。实践中用锁或者让后台任务只读
  2. 资源上限——不能无限开线程。设一个上限（比如最多 5 个并发后台任务），超出就排队等待
  3. 超时机制——后台任务不能跑到天荒地老。设一个合理的超时（比如 5 分钟），超时自动标记为 FAILED

Swarm 团队协作

Swarm 是最复杂也最强大的编排模式。它让多个 AI 扮演不同角色，按照预设的流程协作完成一个复杂任务。

角色与交接

Swarm 的两个核心概念：角色和交接。

每个角色有三样东西：

  • 身份——系统提示告诉它”你是谁、你擅长什么”
  • 工具集——它能用哪些工具（写代码的角色有文件写入工具，审查的角色只有文件读取工具）
  • 交接规则——做完自己的部分后，把结果交给谁

图 5-3：三角色协作流程

  • Coder（编写代码） → 代码完成 → Reviewer（审查代码）
  • Reviewer → 需要修改 → Coder（回退修改）
  • Reviewer → 审查通过 → Tester（编写测试）

注意 Reviewer 到 Coder 之间有一条回退箭头——如果审查发现问题，代码要打回去改。这种循环是 Swarm 模式的常见特征。

交接协议

角色之间怎么”交接”？不是简单地把全部对话扔给下一个人。交接协议定义了传递的内容：

  1. 当前角色完成自己的任务，产出一个工作成果（比如写好的代码文件）
  2. 当前角色生成一份交接摘要——做了什么、产出在哪、有什么需要注意的
  3. 下一个角色收到交接摘要和必要的文件路径，在自己的上下文中开始工作
  4. 如果下一个角色发现问题需要打回，生成反馈摘要交给上一个角色

每个角色都有独立的上下文——它不需要知道其他角色的全部对话细节，只需要看到交接摘要和相关文件。这跟现实中的团队协作完全一样：你不需要看同事的所有聊天记录，只需要看他交给你的文档和备注。

Swarm 的核心数据结构

理解了概念，来看代码。角色和交接可以用很简洁的数据结构表达：

# harness/orchestration.py — Swarm 核心
@dataclass
class SwarmRole:
    name: str                    # "Coder" / "Reviewer" / "Tester"
    system: str                  # 角色专属的系统提示
    tools: list[str]             # 角色能用的工具
    handoff_to: list[str]        # 做完后交给谁

@dataclass
class Handoff:
    from_role: str               # 谁交出来的
    to_role: str                 # 交给谁
    summary: str                 # 做了什么、产出在哪
    artifacts: list[str]         # 产出的文件路径
    feedback: str | None = None  # 打回时的修改意见

def run_swarm(roles, task, max_rounds=10):
    current = roles[0]           # 从第一个角色开始
    handoff = Handoff(
        from_role="user", to_role=current.name,
        summary=task, artifacts=[],
    )

    for round_num in range(max_rounds):
        # 用委托模式执行当前角色
        result = delegate_task(
            task_description=handoff.summary,
            system=current.system,
            tools=current.tools,
        )
        # 解析结果，决定交给谁
        next_name = parse_handoff_target(result)
        if next_name is None:     # 没有下一个 → 流程结束
            return result
        handoff = build_handoff(current.name, next_name, result)
        current = find_role(roles, next_name)

    return "达到最大轮次限制，流程终止"

注意 max_rounds 参数——这是防止 Reviewer 和 Coder 之间无限打回的关键。延伸思考里提到的”死循环”问题，就靠这个上限兜底。实践中 10 轮足够完成大多数任务；如果 10 轮还没收敛，通常说明任务拆得不对，而不是需要更多轮。

💡 什么时候用 Swarm

Swarm 模式适合需要多视角的任务。典型场景：

  • 写代码 + 审查 + 测试——三种不同的思维方式
  • 需求分析 + 架构设计 + 实现——从抽象到具体的渐进过程
  • 翻译 + 校对 + 本地化——每一步需要不同专业知识

如果任务用一个 AI 就能做好（比如简单的文件读写），不要用 Swarm——它的编排开销会拖慢速度。

Claude Code 的 Agent 实现

我们刚才从零实现了三种编排模式。Claude Code 本身是怎么做 Agent 编排的？它主要用的是委托模式——通过一个叫 TaskTool 的内置工具实现。

❗ Claude Code 的 TaskTool

当 Claude Code 的主 Agent 遇到复杂任务时，它不是自己硬扛，而是调用 TaskTool 启动一个子进程。子进程里运行着一个独立的 Agent 实例，有自己的查询循环、自己的上下文、自己的工具集。

这跟我们的 delegate_task() 原理完全一样，但 Claude Code 做了几个关键优化：

  • 进程隔离而非线程隔离——子 Agent 跑在独立进程中，崩溃不会影响主 Agent
  • 工具集继承与裁剪——子 Agent 默认继承主 Agent 的工具，但可以根据任务类型裁剪掉不需要的
  • 结果截断——子 Agent 的返回结果如果太长，会自动截断到合理长度，避免撑爆主上下文

从编排到通信

我们目前实现的三种模式有一个共同限制：所有 Agent 都在同一台机器上运行，通过内存或线程通信。在更大规模的系统中——比如多台机器上的 Agent 集群——通信方式会从函数调用变成消息队列、从共享内存变成网络协议。

但核心思想不会变：拆任务、分配、收结果。无论底层通信方式怎么变，编排的逻辑结构是一样的。

图 5-4：编排架构的演进

  1. 函数调用 — 同进程，最简单
  2. 线程/进程 — 同机器，隔离性好
  3. 消息队列 — 跨机器，异步解耦
  4. 分布式协议 — 跨集群，容错恢复

（规模从小到大递增）

对于我们的 harness 项目来说，线程和进程级别的编排已经足够。但理解这个演进方向，有助于你在未来面对更大规模的系统时做出正确的架构选择。

延伸思考

在进入下一章之前，回顾你构建的编排系统，思考几个问题：

  1. 如果子 Agent 在执行过程中又需要委托子任务（嵌套委托），会发生什么？需要设深度限制吗？
  2. 后台任务完成后，主 AI 怎么知道该通知用户？是用户主动问，还是有办法主动推送？
  3. Swarm 模式中，如果 Reviewer 和 Coder 之间反复打回修改（陷入死循环），你会怎么处理？

章节小结

  • 三个 Prompt 构建了三种 Agent 编排模式：委托（子任务分派）→ 后台（异步执行）→ 团队（多角色协作）
  • 委托模式的灵魂是上下文隔离——子 AI 有独立的对话空间，结果通过摘要回传
  • 后台任务通过线程实现异步执行，有 PENDING → RUNNING → COMPLETED/FAILED 四个状态
  • Swarm 团队协作靠角色（身份 + 工具集）和交接协议（摘要 + 文件路径）驱动
  • Claude Code 用 TaskTool 实现委托，通过进程隔离保证主 Agent 的稳定性
  • 三种模式不互斥，可以组合使用——团队里的角色可以委托子任务，慢操作可以放后台
  • 下一章我们给 Agent 装上”记忆”——让它跨会话保持对项目的了解
English EN 简体中文 ZH