架构与构建多智能体系统

本章内容

多智能体系统架构设计
在 Agent 与 Agentic Flow 之间取得平衡
理解 Agent Flow 中的任务交接
使用护栏验证 Agent 工作流

随着单智能体系统出现不久，开发者和研究人员便开始尝试引入更多 Agent。其想法十分简单：更多的 Agent 可以协同解决更复杂的问题。2023 年，以 AutoGPT 和 BabyAGI 为代表的第一波多智能体系统迅速走红。但短短几周后，它们便暴露出了同样的问题——无限循环。一个 Agent 不断地产生新的子目标，虽然演示效果令人惊艳，但实际运行却十分脆弱，难以控制。随后，多智能体的发展开始走向结构化。例如：

MetaGPT 提出了基于角色分工的团队架构，模拟一家软件公司的组织形式，包括 CEO、产品经理、工程师、测试等角色。
CrewAI 与 AutoGen 则进一步规范了协作团队模式，让每个 Agent 都拥有明确的职责和通信机制。

到了 2024 年，人们的关注点已经从：

「Agent 越多越好。」

转变为：

合理组织 Agent，比单纯增加 Agent 数量更重要。

因此，多种成熟的多智能体架构模式逐渐形成。与此同时，越来越复杂的单智能体系统也开始出现。它们的理念同样十分直接：

一个足够强大的 Agent，可以避免复杂的多智能体协调与通信。

然而实践证明，单 Agent 在以下几个方面仍然存在明显局限：

工具使用能力有限
推理与规划复杂任务能力不足
难以在多个任务之间灵活切换注意力

本章将重点介绍构建多智能体系统时最常见、最实用的架构模式。我们将讨论：

为什么需要从单 Agent 演进到多 Agent；
Agent 之间如何通信；
Agent 如何进行任务交接；
如何监控并保护 Agent 与 Agent 之间的交互过程。

4.1 多智能体系统架构设计

下面介绍三种最基础、也是最常见的多智能体架构。它们能够扩展单 Agent 的能力，使系统能够处理更加复杂、更细粒度、更多步骤的任务。可以把这些架构理解成几种可以自由组合的设计模式。没有任何硬性规定：

某一种业务必须采用某一种架构。

实际项目中，这几种模式往往可以混合使用。

多智能体系统通常比单智能体系统能力更强。但如果设计不当，它们也会带来一些新的问题，例如：

成本更高（更多 LLM 调用）
延迟更大（Agent 间通信）
行为更加不可预测

只要遵循良好的设计原则，多智能体系统依然可以做到既易于开发，也易于维护。经过近两年的实践，目前已经形成三种经典架构。这些架构我们在第一章已经简单介绍过，这里再次回顾。

图 4.1 展示了三种典型模式：

Flow（流水线）
Orchestrator（中心-辐射 / Hub-and-Spoke）
Collaboration（协作网络 / Peer-to-Peer）

本书插图

在介绍这些模式之前，我们需要先理解三个决定 Agent 架构的重要概念，决策、控制和通信。

决策：假设一个研究 Agent 搜集了十篇论文。现在它需要决定哪三篇值得进一步总结？负责做出这个决定的 Agent，就是决策者。因此决策指的是根据已有上下文，决定下一步应该做什么。设计多 Agent 时，需要回答两个问题：哪个 Agent 拥有决策权？它拥有哪些上下文？上下文的重要性与决策权同样重要。上下文太少，Agent 会盲目决策。上下文太多，真正重要的信息反而容易被淹没。不同架构对于上下文的管理方式完全不同。
控制：继续刚才的例子，Agent 已经决定总结三篇论文。负责获取这些资料、执行摘要生成并写出最终结果的那个 Agent，就是拥有控制权的 Agent。因此控制表示真正执行任务的能力。需要注意：决策权与执行权是两件不同的事情。例如：规划 Agent只负责决定下一步做什么，但它自己并不会调用工具。而 Worker Agent 负责真正执行工具，却不会决定下一步做什么。因此，一个 Agent 可以：
- 有决策权，没有执行权；
- 有执行权，没有决策权；
- 或者两者兼具。
在 Agent 系统中，控制权通常会按照顺序和并行两种方式在多个 Agent 之间流转。不同架构最大的区别之一，就在于控制权如何流动。
通信：假设现在有四个 Agent 正在共同撰写一份研究报告。问题来了，它们应该所有人共享全部工作内容，还是每个人只看到协调者分配给自己的部分？这就是通信的问题。它关注的是上下文如何在 Agent 之间传播。例如，在Collaborative（协作）模式中，所有 Agent 通常共享同一个上下文。任何 Agent 都可以基于别人已经完成的工作继续推进。而在Hierarchical（层级式）模式中，上下文不会直接共享。所有信息都必须经过协调 Agent（Orchestrator）进行筛选和转发。开放式通信能够让多个 Agent 更容易进行协同合作，但代价是上下文不断膨胀，导致上下文变长、信息冗余，增加推理成本。封闭式通信则让每个 Agent 始终专注于自己的职责和局部上下文，但代价是不同 Agent 之间可能会重复完成相同的工作，造成一定的资源浪费。这两种通信方式并没有绝对的优劣之分，哪一种更合适取决于具体的系统需求。

在 Flow（流水线）模式中，决策、控制和通信都按照流水线的方式执行，任务会依次从一个 Agent 流转到下一个 Agent。在 Orchestrator（编排者）模式中，决策集中由一个中心 Agent 负责，通信也由该中心 Agent 发起和协调；而控制权则根据需要，从中心 Agent 分配给各个工作 Agent，待其完成任务后再收回。最后，Collaboration（协作）模式允许决策权和控制权在多个 Agent 之间动态传递，同时所有 Agent 都通过一个集中式通信通道进行信息交换和协作。表 4.1 对这三种模式的特点进行了总结。

表 4.1 Agent 三个核心概念

概念	Agent 中的体现
Decision-making（决策）	Agent 决定调用哪些 Tool，以及如何调用。进一步扩展后，Orchestrator Agent 可以决定其他 Agent 应该执行哪些任务。
Control（控制）	Agent 真正执行工具、完成任务并输出结果的能力。拥有控制权的 Agent 就是真正干活的人。
Communication（通信）	Agent 之间交换信息的方式。在 Flow 中，消息通常依次传递；在 Orchestrator 中，由协调 Agent 转发；不同模式拥有不同的信息流结构。

这三个概念决策、控制和通信构成了所有 Agent 架构设计的三个基本维度。本章后续介绍的 Flow、Orchestration和Collaboration三种模式，本质上就是在这三个维度上采用了不同的设计方案。实际项目中，选择哪一种模式，更多取决于经验。通常建议：先从最简单的 Flow 开始。当系统逐渐复杂之后，再逐步演进到Orchestrator、Collaboration或其他更复杂的协调机制。在真正开始设计 Agent 架构之前，最值得先思考的问题并不是：

我要用哪种 Pattern？

而是：

谁负责做决定？
谁负责执行？
信息如何流动？

只有先想清楚这三个问题，后续选择哪种架构，才是经过设计后的结果，而不是默认选择。

4.1.1 决策与控制模式

决策和控制是任何 Agent 系统的核心，因此我们来看几个更复杂的示例，展示三种不同的策略：

Agent Flow：没有中央决策者，也没有中央控制器，仅通过 Agent 之间传递信息。
Agent Orchestration：由一个编排 Agent负责决策，其他 Agent 提供信息，最终由编排 Agent 执行整体计划。
Agent Hierarchy：一个管理 Agent类似项目经理，将任务分配给多个 Worker Agent。

图 4.2 展示了这些常见的 Agent 决策与控制模式。

本书插图

图 4.2 多 Agent 架构中的决策（Command）与控制模式：Flow、Orchestration 和 Manager 架构。

从图中可以看到：

顶部是 Flow 模式，没有中心化的决策者或控制者。
中间是 Orchestration 模式，由 Orchestrator 统一负责决策，并根据需要将任务分配给 Worker Agent。
底部是 Manager-Worker 模式，它是 Orchestrator 模式的进一步扩展。在这种模式中，多个 Manager 形成层级关系，逐层向下分配任务，而底层 Worker Agent 保留实际执行任务的控制权。表 4.2 总结了这三种模式的优缺点。

总体来说，架构越复杂，可获得的控制能力越强，前提是你能够正确管理决策、控制和通信三个要素。例如，如果你还不熟悉任务委派模式（如 Orchestrator 或 Manager-Worker），建议先使用 Flow 模式。Agent 架构本质上只是构建多 Agent 系统的一种设计模式。任何模式都可以应用于任何场景，但正如后面将看到的，不同模式更适合不同的问题。

表 4.2 Agent 架构对比

架构模式	优点	缺点
Flow（流程）	将复杂单 Agent 拆分为多个简单 Agent；大型目标易于拆解；无需复杂决策逻辑；易于测试和调试。	无法按需执行部分 Agent，通常只能整体执行；决策能力较弱；某个 Agent 出错可能导致整个流程失败。
Orchestrator（编排）	擅长复杂决策；可根据需要选择执行部分或全部 Agent；适合直接与用户交互；Worker Agent 出错时容易恢复。	构建、调试和评估都更复杂；Orchestrator 需要完善的评估机制、护栏和反馈机制。
Collaboration（协作）	适合目标模糊、开放性强、存在多种可能结果的复杂任务。	成本高、Token 消耗大、延迟高；评估和反馈机制设计困难。

4.1.2 使用共享内存、消息传递和 MCP 进行通信

通信是控制 Agent 能获取多少信息的关键机制。我们限制 Agent 之间通信，主要有两个原因，而且这两个原因会相互叠加。即使是看起来规模不大的系统，它们同样成立。

第一个原因是成本。共享上下文中的每一个 Token，都需要在每一次 LLM 调用时重新付费。在多 Agent 系统中，这种成本会随着 Agent 数量迅速放大。例如，一个 Worker Agent 如果每次都读取Orchestrator 的完整历史和另外三个 Worker Agent 的全部对话，那么它实际上为大量自己根本不需要的信息支付了 Token 成本。

第二个原因是信息选择准确率。现代 LLM 虽然能够处理几十万 Token 的上下文，但随着无关内容不断增加，它们定位真正重要信息的能力会逐渐下降。这一现象通常被称为上下文稀释或中间遗忘，它不是一种比喻，而是已经被实验验证的真实性能问题。

来看一个具体例子。用户要求一个研究 Orchestrator完成三家公司的竞争分析。Orchestrator 将任务拆解为：

三个并行运行的 Research Agent，每个负责研究一家公司；
一个 Synthesis Agent（综合 Agent），负责整合三份研究结果。

此时，合理的通信方式应该是，对于每个 Research Agent，它只需要知道：

公司名称；
研究标准；
输出格式。

如果把下面这些内容也发送给它：

用户完整请求；
Orchestrator 的整体计划；
另外两家公司的研究结果；

不仅会增加每一次调用的 Token 成本，还会迫使它不断思考：

为什么我要研究这家公司，而不是另外两家？

而 Synthesis Agent 的需求又不同。它应该收到三份完整的研究报告，但不需要看到 Orchestrator 内部是如何规划整个任务的。因此，每一次 Agent 之间的信息边界，其实都是一个经过设计的决策：只有真正对当前步骤有价值的信息，才值得进入上下文。

目前 Agent 系统常见的通信方式包括：

按需传递消息（Message Passing）
共享整个对话上下文（Shared Conversation）
使用 MCP 等协议进行结构化通信

图 4.3 展示了这些通信方式。

本书插图

图 4.3 Flow 架构下不同 Agent 通信模式的比较。

图中，顶部展示的是最常见的 Message Passing（消息传递）模式。所有消息和对象都从一个 Agent 顺序传递到下一个 Agent。有时候，我们并不希望所有信息都被共享，而是只让中间的 Worker Agent 获取完成任务所需的信息。

另一种方式是 Shared Memory（共享内存）或 Shared Conversation（共享会话）。所有 Agent 都可以访问同一份上下文，因此拥有完全一致的信息。

图底部展示的是 Tool Exchange（工具交换）。Agent 并不是直接彼此通信，而是将另一个 Agent 当作 MCP 工具或通过函数接口调用另一个 Agent。

表 4.3 总结了这些通信模式的优缺点。

表 4.3 Flow 模式中的通信方式比较

通信方式	优点	缺点
Message Passing（消息传递）	简单高效；只传递必要信息；能够过滤噪声。	通信质量完全依赖上一位 Agent 的输出质量。
Shared Message Thread（共享消息线程）	所有 Agent 都能了解完整历史；出现错误时更容易恢复。	上下文不断增长；Token 消耗增加；Agent 更容易失焦或执行无关任务。
Tool Exchange（MCP）	仅传递必要信息；通信格式结构化、标准化。	每次通信都需要一次工具调用，存在一定开销。

与决策和控制模式一样，通信模式也可以自由组合。在复杂工作流中，你可能故意混合多种通信方式，以发挥各自优势。不过，一般来说，更推荐在同一个代码库中的多个 Agent 保持一致的通信模式，这样系统更容易维护，也更容易理解。

4.1.3 多智能体协调策略（Coordination）

命令（Command）、控制（Control）和通信（Communication）是构建智能体系统的前三个 C。除此之外，还有第四个 C——协调（Coordination）。协调定义了多个智能体如何沿着单一路径、并行路径或多条执行路径协同完成任务。图 4.4 展示了 Flow（流水线）和 Orchestration（编排）模式中四种典型的协调执行方式。

本书插图

图 4.4 智能体按顺序或并行方式协调执行任务的几种常见模式

图中展示了四种最常见的多智能体协调策略。

最上方是顺序流水线（Sequential Flow），也就是智能体之间依次执行，一个智能体完成后将结果传递给下一个智能体。

第二种是并行流程（Parallel Flow），其中中间的两个智能体可以同时执行任务，以提高整体执行效率。

第三种是层级协调（Hierarchical Coordination），它可以看作是一种经过扩展的编排（Orchestration）模式，其中部分工作智能体并行执行，而另一些则顺序执行。

最下方是迭代式评审（Iterative Critique）模式。在这种模式下，一个评审（Critic）智能体负责持续审核一个工作（Worker）智能体的输出，并不断循环，直到结果满足目标要求为止。需要注意的是，这种模式与辩论（Debate）模式不同。Debate 通常由两个或多个地位平等的智能体互相讨论、争论，并逐步达成共识；而 Iterative Critique 则是一个评审不断修改一个工作智能体的输出。这两个概念很容易混淆，因为它们都涉及对结果进行评价，但二者在参与智能体数量以及彼此关系上完全不同：

Critique（评审）：一对一、层级式结构，Critic 对 Worker 的输出拥有审核权和否决权。
Debate（辩论）：多对多、平等协作，没有任何一个智能体拥有最终控制权。

在实际项目中，我们可以根据具体情况组合不同的协调策略，也可以将一种策略嵌套到另一种策略之中，从而构建出更加复杂的多智能体工作流。不过，一个良好的实践原则始终是：先采用最简单、能够满足需求的模式，之后再根据实际情况逐步优化和演进。

智能体最大的优势在于拥有自主决策能力，但在生产环境中，无限制的自主性通常并不是最佳选择。构建智能体系统最大的挑战，就是在以下两者之间取得平衡：

灵活性：智能体能够自主决定下一步该做什么，这也是它成为智能体的原因。
控制权：系统决定智能体允许做什么，从而保证整个系统可靠、可控。

许多实践者认为，这是构建 Agent 系统过程中最难把握的问题，而且不存在适用于所有场景的统一答案。

更准确地说，自主性与约束并不是二选一，而是一条连续的光谱（Spectrum）。你选择落在哪个位置，主要取决于任务的重要程度：

高风险任务：（生产环境、监管行业、不可逆操作、面向客户的输出）需要更多限制；
低风险任务：（内部研究、探索性分析、草稿生成）则可以给予智能体更大的自主空间。

实践中，有几条经验值得遵循：

从满足需求所需的最低自主性开始设计；
当智能体表现稳定可靠后，再逐步放宽权限；
将约束重点放在最关键的位置，例如：
- 工具访问权限
- 输出 Schema
- 行动预算
- 人工审批（Human-in-the-loop），尤其是涉及不可逆操作时
在其他影响较小的地方，则尽可能保持灵活。

可以把智能体的自主能力理解成一种需要通过可靠表现逐步获得的权限，而不是系统默认赋予的能力。

基于这一理念，将单智能体系统首先演化成 Flow（流水线）模式通常是一个不错的起点。Flow 模式能够提供足够的结构化约束，使系统保持稳定可靠，同时又允许每个步骤内部保留一定程度的自主决策能力。在此基础上，再根据实际运行效果逐步增加约束或放宽限制，通常比一开始就设计复杂架构更加稳妥。

下面，我们将详细介绍图 4.4 中展示的几种协调执行模式，以及另外一些在实践中也非常常见的策略。

顺序流水线（Sequential Pipeline，Agent-to-Agent）

顺序流水线（Sequential Pipeline）是最简单的一种 Agent-to-Agent 协作模式，也是前面章节已经介绍过的基本模式。在这种模式中，每个智能体负责流水线中的一个阶段，并将自己的输出作为下一阶段智能体的输入。整个流程具有以下特点：

线性
确定性
最容易理解和调试

不过，看似简单的流水线背后，其实隐藏着几个重要的设计决策。

第一，是阶段之间的数据结构

每个智能体输出的数据格式，都会成为下一个智能体输入的数据格式。因此，只要某个阶段修改了输出 Schema，后续所有阶段都可能受到影响。也就是说，流水线中的数据接口是一种天然的契约。

第二，是失败处理

假设整个流水线共有四个阶段，如果第三个阶段执行失败，那么整个流程都会终止。

因此，生产环境中的流水线通常都会为每个阶段设计：

Retry（自动重试）
Exponential Backoff（指数退避）
Fallback（降级处理）

如果重试次数耗尽，则返回明确的错误信息，而不是让整个系统无声失败。

第三，是这种模式适合什么场景。

顺序流水线适用于每一步都依赖前一步结果的任务，例如：

必须先完成数据分析，才能进行总结；
必须先生成数据，才能进行分析。

如果各个阶段彼此独立，则顺序执行就不是最佳选择。例如：

可以同时完成多个任务 → 更适合 Parallel Flow（并行流程）
需要不断修改和优化结果 → 更适合 Iterative Critique（迭代评审）

因此，顺序流水线更适合那些天然具有固定步骤的工作流程。例如：

数据清洗 → 数据分析 → 自动摘要

每一步都依赖前一步结果，因此采用流水线最自然，也最容易追踪整个流程。

虽然 Sequential Pipeline 是一种经典模式，但它也存在几个不可忽视的成本。需要强调的是，这些成本并不是这种模式的缺点，而只是帮助我们判断什么时候应该考虑采用其他协调方式。

第一，是延迟。

由于所有阶段必须依次执行，因此整个执行时间等于所有阶段耗时之和。即使其中某些步骤完全可以并行执行，也必须等待前一步结束。对于彼此独立的任务来说，并行执行（Parallel Flow）能够显著缩短整体耗时。

第二，是脆弱性。

任何一个阶段发生故障，都会导致整个流水线停止。虽然可以通过 Retry、Fallback 等机制缓解这一问题，但随着流水线越来越长，失败风险也会不断增加。

第三，是缺乏修订能力。

顺序流水线本质上是一条单向流程。如果某个阶段生成了较差的结果，通常无法只重新执行这一阶段，而需要重新启动整个流水线，或者额外设计专门的修订逻辑。因此，对于以下类型任务：

写作
设计
复杂分析

通常更适合采用 Iterative Critique（迭代评审），因为它天然支持不断修改和优化结果。

并行委派（Parallel Delegation）

多个智能体也可以同时处理一个任务中的不同子任务，待所有任务完成后，再统一合并结果。在这种模式下，系统会将彼此独立的任务分配给不同的智能体并行执行，随后由下游智能体或后续步骤负责整合、汇总这些输出。

并行流程最大的优势，是能够同时处理多个互不依赖的工作，从而显著提升整体执行效率。例如，一个 AI 系统需要同时分析文本、图片和音频。三个模块可以分别交由三个智能体同时处理，最后再统一融合分析结果。这种模式同样适用于需要同时探索多种可能性的场景，例如一次性生成多个方案，再从中选择最佳结果。

不过，并行流程并不适合所有任务。如果多个子任务之间存在较强的依赖关系，或者必须严格按照固定顺序执行，那么并行化反而会增加复杂度。此外，如果最终需要进行大量同步、协调或复杂的数据融合，那么并行执行带来的额外管理开销可能会抵消它所带来的性能收益。对于本身规模较小、天然就是串行执行的任务而言，引入并行化通常只会增加系统复杂度，而不会带来明显收益。

层级协调（Hierarchical Coordination）

在层级协调（Hierarchical Coordination）模式中，一个顶层的管理者（Manager）或编排者（Orchestrator）智能体负责将复杂目标拆解为多个子任务，再根据每个子任务的特点分配给不同的专业 Worker 智能体，并向它们下达具体指令、监督执行过程，最后整合所有结果。可以把这种模式理解为：一个“老板”管理多个“专家”。这里的老板不仅仅负责把任务拆开再合并结果，而是真正承担了规划、决策和控制的职责。虽然这种模式底层同样会利用并行执行来提高效率，但整个系统的大部分复杂性其实都集中在管理层上。

它与 Parallel Flow 在图示上看起来非常相似，因此有必要仔细区分二者。

Parallel Flow 的工作方式是：

将任务拆分成若干彼此独立的部分；
同时执行这些任务；
最后通过简单的聚合得到最终结果。

在这种模式中，Orchestrator 只是负责拆分任务和合并结果，几乎不参与真正的规划。而 Hierarchical Coordination 则不同。它把并行执行当作一种工具，而不是整个架构本身。真正的核心工作由 Orchestrator 完成，包括：

决定每个 Worker 负责什么任务；
为不同 Worker 提供不同的角色和指令；
持续监控各 Worker 的执行情况；
对结果进行分析、协调和整合，而不仅仅是简单合并。

因此，即使三个 Worker 都是在并行执行，同一套结构也可能属于两种不同模式。关键区别在于：

如果 Orchestrator 只是把一个统一任务平均拆分，再把结果拼接起来，那么它属于 Parallel Flow。
如果 Orchestrator 根据整体目标进行规划，为不同 Worker 分配不同职责，并负责监督和协调，那么它属于 Hierarchical Coordination。

一个简单的判断标准就是：

观察 Orchestrator 到底在做什么。

如果它完全可以被一个简单的 Fan-out / Fan-in（任务分发 / 结果汇总）机制取代，那么这就是 Parallel Flow。如果不能，因为它真正承担了：

任务规划
角色分配
执行监督
结果协调

那么这就是 Hierarchical Coordination。层级协调特别适用于那些复杂、多步骤、能够通过任务拆解和专业分工提升效率的问题。例如，一个内容生成系统可以由 Orchestrator：

动态规划研究流程；
决定需要哪些子任务；
为每个子任务选择最合适的专家 Agent；
最后统一整合研究成果。

这种模式最大的优势是：

具有清晰的整体控制能力；
能够协调多个子任务之间的依赖关系；
更容易保证整体目标一致。

不过，它并不是所有场景都适合。对于那些：

较简单的问题；
边界明确的任务；
单个 Agent 足以完成的工作；

采用层级协调往往属于过度设计。这种模式不仅管理成本高，而且系统复杂。如果 Manager 本身规划错误，或者执行失败，那么整个系统都会受到影响。此外，当任务本身无法被清晰拆分，或者系统对延迟要求非常高时（因为规划过程本身需要时间），层级协调通常也不是最佳选择。

迭代式辩论（Iterative Debate）是目前生产级 Agent 系统中最重要的多智能体协作模式之一，因此值得重点理解。这种模式下，两个或多个智能体通过不断经历：

提出方案
相互评审
修改优化

多个循环之后，共同收敛到一个高质量的最终结果，而不是第一次得到一个看似合理的答案就停止。这种模式主要有两种不同的实现方式。

对称式:在对称模式下，每个智能体既负责提出方案，也负责评价别人的方案。所有智能体地位平等。大家轮流：
- 提出自己的观点；
- 挑战其他人的观点；
- 最终逐渐形成共识。
整个过程中没有固定分工。
非对称式:非对称模式则进行了角色划分。通常包括：
- Proposer / Worker：负责生成候选方案；
- Critic / Judge：负责审核方案，并指出问题。
Critic 不负责提出解决方案，而是：
- 给出具体修改意见；
- 判断当前结果是否达到质量要求；
- 决定是否继续下一轮优化。
实际上，这就是前面介绍过的 Iterative Critique 模式，只不过进一步扩展成：
- 多个 Proposer；
- 多个 Critic；
- 或两者同时存在。

两种模式分别适用于不同类型的问题。

对称模式适合：没有唯一正确答案的问题，例如：
- 学术研究综合
- 战略分析
- 伦理推理
这些任务最大的价值来自不同观点之间的真正碰撞。
非对称模式适合：具有明确质量标准的问题，例如：
- 能否通过测试的代码
- 是否符合评分标准的摘要
- 是否满足约束条件的执行计划
这里 Critic 的职责就是保证最终结果达到预设标准。

无论是哪一种模式，都必须设计停止条件。否则，多个智能体可能会一直讨论下去。常见停止条件包括：

最大讨论轮数
收敛检测
- 大家已经达成一致；
- 或连续几轮没有新的修改。
Critic 明确给出 Pass。

生产环境通常会把这些方式组合起来：

设置最大轮数，防止无限循环；
同时允许 Critic 提前结束流程，只要结果已经足够好。

这种不断优化的过程，非常适合：

高风险决策；
模糊问题；
需要深入分析；
创造性任务。

由于多个 Agent 从不同角度思考，它们能够互相发现错误，不断提升整体推理质量。例如，多个 Agent 对某个科研问题展开讨论，最终得到一个更可靠、更充分论证的结论。当然，它也有明显缺点。对于那些：

简单问题；
快速响应场景；
明确答案的问题；

这种多轮讨论往往得不偿失。此外，它需要大量 LLM 调用，因此成本较高、延迟较高。如果所有 Agent 都拥有相同的知识盲区或偏见，那么 Debate 也无法真正提升质量。它们甚至可能不断强化彼此的错误观点。另外，在达成共识的过程中，有时也可能把真正富有创造性的方案平均掉。

投票 / Best-of-N（集成方法，Ensemble）

这种模式下，会让多个 Agent（或者同一个 Agent 多次运行）分别独立解决同一个问题，然后再通过投票或选择机制，从多个结果中选出最佳答案。投票方式可以包括：

简单多数投票
加权投票
Judge Agent 负责最终裁决

投票通过汇聚多个观点来提升系统可靠性和准确性的方法。它通常用于诊断系统、评估任务等场景：由多个模型或多个 Agent 对同一个输入进行分析，然后由系统选择多数意见，或者选择最一致的答案。这种机制能够降低单个 Agent 出错或产生偏差所带来的影响，因为异常或偏离主流的结果会被其他 Agent 的意见所“压制”或淘汰。

不过，这种模式也存在明显局限。如果所有 Agent 本质上完全相同，那么它们往往会犯相同的错误。此时，再多的投票也没有意义。此外，如果每次生成答案成本很高或推理速度较慢，那么同时运行多个 Agent 也会造成较大的资源浪费。对于那些单个 Agent 就能轻松解决的问题，Best-of-N 往往属于过度设计。

另外，一个简单的多数投票也可能失败。例如，多数 Agent 都因为相同偏见或相同错误信息而给出了错误答案。这种情况下，多数意见反而会变成错误答案。因此，这种模式真正重要的设计问题其实是：

如何让多个 Agent 彼此真正具有差异性。

提升多样性最有效的方法，是采用不同的大模型。例如Claude、GPT、Gemini三个模型同时参与。由于它们训练数据不同、对齐方式不同、推理风格不同，因此得到的答案通常更加多样化。这也是生产系统在预算允许情况下最常采用的方法。

其次，可以通过 Prompt 来制造差异。例如一个 Agent 扮演乐观主义者、一个扮演怀疑主义者、一个扮演领域专家，不同 Prompt 会引导同一个模型走向不同的推理路径。不过，这种方式带来的差异仍然受到模型本身偏见的限制，因此效果通常不如混合不同模型。

总体而言，多样性才是 Voting 模式真正需要重点设计的问题。不同模型能够提供最大的多样性，但成本最高。不同 Prompt 成本较低，也能提供一定程度的多样性，只是效果有限。

角色扮演协作（Role-playing Collaboration）

角色扮演模式中，不同 Agent 被赋予不同角色，并通过协作式对话共同完成任务。一个经典例子就是 CAMEL Framework。其中包括AI User 和 AI Assistant。AI User 负责提出问题和需求；AI Assistant 负责回答问题并执行任务。双方不断交流，一步一步推进整个任务。通过模拟“指导者—助手”、“客户—开发者”或“教师—学生”等互动关系，智能体能够在交流过程中不断澄清需求，并通过多轮协作迭代地提升最终输出的质量。

当任务能够从交互式交流或多视角头脑风暴中获益时，可以采用这种模式。例如，在代码生成场景中，一个智能体可以扮演“客户”，负责描述需求并测试代码；另一个智能体则扮演“开发者”，负责编写代码。这种模式同样适用于生成对话数据或问答数据集：由一个智能体负责提出问题，另一个智能体负责作答。当问题需要通过反复澄清需求，或需要创造性的来回讨论时，不同角色之间的互动能够显著提升任务完成效果。

角色扮演并非总是有帮助。如果不同角色无法提供新的信息或额外价值，那么对于简单任务而言，来回的对话只会增加系统延迟。此外，如果角色定义不清晰，还存在陷入循环讨论或低效对话的风险。还有一种情况需要注意：如果缺乏外部校验机制，两个智能体可能会相互放大彼此的错误。例如，“用户”智能体基于一个错误的假设提出需求，就可能误导“助手”智能体，最终导致整个协作过程偏离正确方向。

条件路由（Conditional Routing / Branching）

在这种模式下，系统包含一个路由 Agent或路由逻辑，它会根据输入内容的特点，将请求分发给不同的专业 Agent 或工作流处理。换句话说，整个流程的第一步就是一个分类器，它负责回答“这个任务应该交给谁处理？”，然后再将任务交接给最合适的专家 Agent。例如，当用户提出一个问题时，系统会先判断它属于哪一类。如果是数学问题，则交给数学求解 Agent；如果涉及法律推理，则交给法律专家 Agent。

条件路由非常适合处理异构任务，即不同类型的问题需要不同能力的 Agent，而不是让所有 Agent 都尝试处理所有请求。这种模式广泛应用于客服机器人和通用 AI 助手框架。例如，技术问题可以被路由到技术支持 Agent，账单问题则发送给财务 Agent。这样能够确保每个任务都由最擅长的 Agent 或工作流负责，提高整体效率和准确率。

不过，如果所有输入都非常相似，或者一个通用 Agent 就能够胜任所有任务，那么这种模式就没有必要了。此外，如果路由器发生误判，把任务交给了错误的 Agent，就可能导致最终回答质量下降。如果一个问题同时涉及多个领域（例如既涉及技术又涉及财务），一个简单的分类器往往难以正确处理。因此，条件路由最适用于能够清晰划分领域边界的任务。另外，路由本身也需要消耗时间和计算资源，因此对于简单、统一的任务而言，引入路由机制反而会增加不必要的开销。

点对点网络（Peer-to-Peer Network）

在点对点（P2P）架构中，没有任何中心协调者（Coordinator）。所有 Agent 都以去中心化的方式进行通信与协作，彼此共享知识或阶段性成果，并共同收敛到最终解决方案。这种协调方式通常依赖于信息广播、协商或协作，而不是由某个上层 Agent 制定统一计划。

P2P 模式特别适合分布式决策以及对系统鲁棒性要求较高的场景。例如，在网络安全监控系统中，多个检测 Agent 可以相互交换告警信息和分析结果，共同识别潜在威胁，而无需依赖某一个中心节点。这种去中心化架构通常具有更好的容错能力：即使某个 Agent 失效，其他 Agent 仍然能够继续协同工作，并能够根据 Agent 的加入或退出动态调整整个系统。

不过，由于没有统一的协调者，这种模式的设计和调试难度都较高。因此，对于需要严格全局控制或固定执行顺序的任务，并不推荐采用 P2P 架构。自由协作的网络容易出现竞争条件以及各 Agent 对系统状态理解不一致的问题。

另外，随着 Agent 数量增加，通信开销会迅速增长，因为每个 Agent 都可能需要与多个其他 Agent 保持联系。如果缺乏统一的协作策略或冲突解决机制，整个系统甚至可能出现状态漂移或死锁。与其他复杂架构一样，对于规模较小的团队或简单直接的问题，采用点对点网络通常得不偿失，一个简单的工作流往往更加高效。

表 4.4 总结了上述各种 Agent 架构模式分别适用于哪些场景，以及哪些情况下不建议使用。

表 4.4 Agent 架构模式概览

架构模式	适用场景	不适用场景
顺序流水线（Sequential Pipeline）	适用于具有明确顺序、需要按步骤执行的任务。	并非每次都需要执行所有 Agent。
并行委派（Parallel Delegation）	多个任务可以并行执行，从而提升整体性能。	—
分层协调（Hierarchical Coordination）	用户接口 Agent 代表用户协调多个专业 Agent 完成任务。	不需要复杂决策或 Agent 指挥时。
迭代辩论与优化（Iterative Debate & Refinement）	希望对结果进行双重或多重校验，提高准确性。	问题非常简单，可直接完成。
投票 / Best-of-N（Ensemble）	问题存在多种可能解释或答案，需要综合多个结果。	问题只有唯一明确答案。
角色协作（Role-playing Collaboration）	需要高创造力、创新性输出，与主要目标高度契合。	实时系统或创造性要求较低的场景。
条件路由（Conditional Routing）	多领域客服系统，或需要人工审批、人工介入的工作流。	—
点对点网络（Peer-to-Peer Network）	去中心化控制可提升系统鲁棒性，降低单点故障风险。	—

4.2 在 Agent 与 Agent Flow 之间取得平衡

下面我们将重点介绍 Agent Flow 模式，并看看它如何将一个庞大、单体的 Agent，演化为一个更加实用且高效的 Agent 系统。Agent Flow 模式并不是凭空设计出来的，而是在大量生产级 Agent 系统实践过程中逐渐形成的一种架构模式。虽然在很多情况下，我们仍然倾向于优先使用单个 Agent 来解决问题，但当任务变得复杂时，将任务拆分为多个 Agent 组成的工作流，往往能够更好地理解整个系统的行为，也更容易分析、调试和优化每个 Agent 所执行的动作。

4.2.1 将单个 Agent 演化为 Agent Flow

如果你之前没有构建 Agent 系统的实践经验，那么你的第一个 Agent 系统大概率会采用单 Agent 架构。对于许多简单的应用场景来说，一个 Agent 已经完全足够。然而，随着不断为 Agent 增加功能，它需要调用越来越多的工具，系统提示词也会越来越长。当 Agent 承担的职责越来越多时，它最终会变得“过载”，此时通常就需要将一个 Agent 拆分成多个 Agent。

从单 Agent 演化为 Agent-to-Agent Flow，通常意味着将原来的 Agent 拆分为多个具有明确职责的专业 Agent，每个 Agent 负责工作流中的某一个环节，并承担与自身角色对应的任务。将单个 Agent 拆分为多个 Agent，通常有以下几个原因：

过载——Agent 拥有过多的工具、过长的 Prompt，或者两者兼而有之，导致开始无法正确执行任务。一种十分常见的问题是注意力过载。例如，Agent 同时拥有 search_documents 和 search_files 两个功能相近的工具，由于工具描述存在重叠，它可能无法判断到底该调用哪一个工具。解决办法通常不是继续增强模型，而是让每个工具的描述更加清晰、互不重叠，使模型仅根据工具描述就能够准确区分它们。
专业化——当 Agent 专注于某一个明确的任务或目标时，通常能够获得更好的表现。
成本与延迟——相比一个拥有大量工具、能够访问全部上下文信息的大型 Agent，只拥有少量工具且只接收相关上下文的专业 Agent，其运行成本通常更低、速度也更快。需要记住，无论是工具定义还是上下文通信，本质上都会转化为 Token。而这些额外 Token 在很多 LLM 调用过程中，并不会真正带来价值，却会持续增加成本。

图 4.5 展示了一个单体 Agent如何被拆分为由多个角色组成的 Agent Flow。

本书插图

图 4.5 单个 Agent 被拆分为一个多 Agent 工作流。原来的 Agent 被划分为三个职责明确、边界清晰的角色（Agent），每个 Agent 仅封装与自己职责相关的一组工具。原来的单 Agent 最终演化为一个由研究 Agent、规划 Agent以及文件系统 Agent组成的工作流。

列表 4.1 展示了图 4.5 上半部分所对应的“拆分前”版本。在这个示例中，一个 Agent 同时连接多个 MCP Server，并且需要了解每一个 Server 所提供的工具以及如何使用它们。值得注意的是，代码中所有 MCP Server 都是在同一个上下文中创建并管理的。这样，当 Agent 完成任务后，这三个 MCP Server 都会被统一关闭和清理。

列表 4.1 01_single_agent_multiple_mcp.py

servers = [
   MCPServerStdio(
       name="Research Tools",    #1
       params=MCPServerStdioParams(
           command="mcp",
           args=["run", str(SCRIPT)],
       ),
   ),
   MCPServerStdio(
       name="sequential-thinking",    #2
       params={
           "command": "npx",
           "args": [
               "-y",
               "@modelcontextprotocol/server-sequential-thinking"
           ],
       },
   ),
   MCPServerStdio(
       name="filesystem",    #3
       params={
           "command": "npx",
           "args": [
               "-y",
               "@modelcontextprotocol/server-filesystem", SANDBOX
           ],
       },
   ),
]

instructions = """    #4
You are a research assistant who can use tools to perform and plan research.
Given a research goal, use the research tools to find research sources.
Then, use the sequential thinking tool to plan the research.
Finally, use the filesystem tool to write the research plan as a text file.
"""

async with (
   servers[0] as research_srv,
   servers[1] as thinking_srv,
   servers[2] as fs_srv,
):    #5
   agent = Agent(
       name="Assistant",
       instructions=instructions,
       mcp_servers=[research_srv, thinking_srv, fs_srv],
   )
   goal = """
Produce a research plan to find the book:
'The Hitchhiker's Guide to the Galaxy'
"""
   print("Running...", goal)
   result = await Runner.run(agent, goal)    #6
   print(result.final_output)

代码说明

1. 创建一个由 Python 文件实现的 MCP Server，用于提供前面章节实现的研究工具。
1. 引用官方提供的 Sequential Thinking MCP Server，用于帮助 Agent 对复杂任务进行规划。
1. 引用官方 Filesystem MCP Server，用于与本地文件系统交互。
1. 为 Agent 提供系统提示词。
1. 将三个 MCP Server 放入同一个上下文中统一管理。
1. 使用指定目标运行 Agent。

运行列表 4.1 的代码后，Agent 会完成整个研究流程，并最终返回保存研究计划的文件名。虽然此时系统能够正常工作，但这种架构的可扩展性很快就会遇到瓶颈。实践表明，一个拥有 30 个工具、4,000 Token 系统提示词的 Agent，其工具选择准确率通常会明显低于另一个仅拥有 8 个工具、1,500 Token Prompt 的 Agent。

原因在于，每次调用 LLM 时，每个工具的 JSON Schema 和工具描述都会一同发送给模型。如果一个 Agent 拥有大约 30 个工具，那么仅工具定义本身，每次调用模型就可能额外增加 6,000～12,000 个 Token，甚至在用户真正的 Prompt 被处理之前，就已经消耗掉如此多的上下文。以 Claude Sonnet 4.6 的输入价格计算，仅这些工具描述，每次调用模型就可能额外增加约 0.02～0.04 美元的成本。而在 Agent 的循环执行过程中，这部分开销还会随着每一步调用不断累积。

4.2.2 构建 Agent-to-Agent Flow

Agent-to-Agent Flow 与 Prompt Chaining 十分相似，它们都遵循“前一步的输出作为下一步输入”的思路。不同之处在于，Prompt 本身只是文本，而 Agent 可以借助工具进行决策，因此每一个节点不仅能够传递信息，还能够自主完成推理和执行任务。在本节中，我们将把图 4.4 中的单 Agent 架构，转换为图中下半部分所展示的多 Agent Flow 模式。

通常，在拆分一个 Agent 时，首先应该思考它承担了哪些角色，然后按照角色进行拆分。按角色划分不仅能够更清晰地界定每个 Agent 的职责范围和所有权，还能够在系统提示词中进一步强化这种角色定位，同时限制每个 Agent 只能访问完成自身职责所需的工具。

列表 4.2 展示了如何将一个单 Agent 转换为 Agent-to-Agent Flow。在这个示例中，我们针对每一个 MCP Server 创建了一个对应的 Agent，并为其定义了专属角色。实际项目中，Agent 的拆分方式并没有固定答案，它取决于所使用的工具、希望赋予 Agent 的角色，以及最终希望获得的输出结果。一个比较实用的原则是：先考虑如何让 Agent 更加专业化，再根据这些专业能力设计相应的角色。

列表 4.2 02_agent_to_agent_flow.py

research_agent = Agent(
   name="Research Agent",    #1
   instructions="""
You are a research assistant.
Your role is to find research sources.
""",
)

thinking_agent = Agent(
   name="Thinking Agent",    #1
   instructions="""
You are a research assistant.
Your role is to plan the research.
""",
)

filesystem_agent = Agent(
   name="Filesystem Agent",    #1
   instructions="""
You are a research assistant.
Your role is to write the research plan as a text file.
""",
)

# MCP server setup remains the same
async with (
   servers[0] as research_srv,
   servers[1] as thinking_srv,
   servers[2] as fs_srv,
):
   goal = """
Produce a research plan to find the book 'The Hitchhiker's Guide to the Galaxy'
"""

   print("Running...", goal)

   research_agent.mcp_servers = [research_srv]    #2
   result = await Runner.run(research_agent, goal)

   thinking_agent.mcp_servers = [thinking_srv]    #2
   result = await Runner.run(thinking_agent, result.final_output)

   filesystem_agent.mcp_servers = [fs_srv]    #3
   result = await Runner.run(filesystem_agent, result.final_output)    #4

   print(result.final_output)    #4

代码说明

1. 每个 Agent 都拥有独立的角色、对应的 MCP 工具以及各自负责的任务。
1. 为当前 Agent 指定其需要使用的 MCP Server。
1. 为文件系统 Agent 指定对应的 MCP Server，并接收前一个 Agent 的输出作为输入。
1. 前一个 Agent 的输出会直接作为下一个 Agent 的输入，最终输出整个工作流的结果。

将单个 Agent 拆分为 Agent Flow 后，最大的优势在于系统变得更加容易扩展。你可以在工作流中自由增加新的 Agent 节点、接入更多 MCP 工具，或者加入其他类型的处理步骤，而无需让一个 Agent 承担所有职责。与此同时，这种架构还能让你更精细地控制整个工作流中的决策过程，使 Agent 的行为更加确定、更加可预测。正如下一节将要介绍的那样，这也是 Agent Flow 相较于单 Agent 架构的重要优势之一。

4.2.3 Agent Flow 中的自主性与决策

赋予 Agent 决策能力和自主性能够极大增强系统的能力。然而，由于 Agent 本质上由大语言模型驱动，因此它们并不是确定性的。这意味着，Agent 所做出的决策可能会受到许多因素的影响，例如工具返回的结果、输入内容以及模型本身的随机性。因此，同样的输入，在不同时间运行，Agent 可能会做出不同的选择。如果希望 Agent Flow 或多 Agent 系统具有更强的确定性，一个常见的方法就是将关键决策点从 Agent 内部移到程序代码中，让代码而不是 Agent 来控制这些决策。通常，我们会把这些关键判断抽离出来，由程序逻辑统一控制，从而保证系统每次执行都能够得到一致、可重复的结果。

图 4.6 展示了一个带有外部决策点的 Agent Flow。在该示例中，这个决策逻辑由程序代码实现，而不是交给 Agent 自己判断。这样就能够确保来自 Research Agent 的搜索结果数量始终大于 0，相当于建立了一道护栏，防止系统出现不可预测的行为。如果把这一判断完全交给 Agent，那么这种强制性的业务规则就有可能被绕过。

本书插图

图 4.6 在 Agent-to-Agent Flow 中加入了一个由代码实现的确定性决策点。将决策权从 Agent 手中移交给程序逻辑，可以让整个工作流更加稳定、一致，也更容易预测其行为。

列表 4.3 展示了为了支持这种确定性决策而修改后的代码。作为本次修改的一部分，我们还更新了 Research Tools MCP Server，使它每次随机返回不同数量的搜索来源，以模拟真实 Agent 或工具调用过程中存在的随机性。

列表 4.3 03_variable_research_tools.py

# variable research tools – updates
@mcp.tool()
def get_research_sources() -> list[str]:
   """Provides 0 to 3 random research sources."""
   search_sources = [
       "Wikipedia",
       "Google",
       "YouTube",
   ]
   num_sources = random.randint(0, 3)    #1
   if num_sources == 0:
       return []    #2
   return random.sample(search_sources, num_sources)

代码说明

1. 随机决定返回多少个搜索来源。
1. 如果数量为 0，则返回一个空列表。

这里故意让搜索来源数量随机变化，是为了模拟由 LLM 驱动的 Agent 所具有的随机性。需要牢记的是，除非采取措施提高确定性，否则任何由 LLM 生成的输出都可能发生变化。常见的方法包括使用结构化输出、类型约束以及其他约束模式。

列表 4.4 展示了为了支持这一决策流程，对 Agent 及整个 Agent Flow 所做的修改。

列表 4.4 03_agent_to_agent_decisions.py

# updates to the agent roles, prompts and output data
class ResearchSourcesModel(BaseModel):    #1
   research_sources: List[str]
   """A list of research sources to use for research."""

research_agent = Agent(
   name="Research Agent",
   output_type=ResearchSourcesModel,    #1
   instructions="""
You are a research assistant.
Your role is to find research sources.
Do not make up or invent any research sources.    #2
""",
)

thinking_agent = Agent(…   #3

filesystem_agent = Agent(
   name="Filesystem Agent",
   instructions="""    #4
You are a filesystem assistant.
Your role is to write the output as a text file.
Never make up or invent any output.
""",
)

# updates to the agent flow with the new decision point
research_agent.mcp_servers = [research_srv]
result = await Runner.run(research_agent, goal)

research_sources = result.final_output.research_sources    #5

if research_sources and len(research_sources) > 0:     #5
   thinking_agent.mcp_servers = [thinking_srv]

   agent_input = dict(    #6
       research_sources=research_sources,
       goal=goal,
   )

   result = await Runner.run(thinking_agent, str(agent_input))
   research_plan = result.final_output
else:
   research_plan = "No research sources found and no plan was created."

filesystem_agent.mcp_servers = [fs_srv]

agent_input = dict(    #6
   output=research_plan,
   goal=goal,
)

result = await Runner.run(filesystem_agent, str(agent_input))
print(result.final_output)

代码说明

1. 使用强类型输出，将 Agent 的输出限定为字符串列表。
1. 修改 Prompt，明确要求 Agent 不得编造任何研究来源。
1. 该 Agent 无需修改。
1. 进一步明确 Filesystem Agent 的职责，使其仅负责保存输出文件。
1. 判断是否获得了搜索来源，这就是整个流程中的决策点。
1. 当需要传递多个输入时，将它们包装成一个字典，并使用字段名称进行标记。

这一段代码包含了不少变化，因此下面逐一进行说明。首先，我们修改了 Research Agent，让它返回一个名为 ResearchSourcesModel 的结构化输出模型，而不是普通字符串。这个模型规定 Agent 的输出必须包含一个字符串列表 research_sources。结构化输出会在 SDK 层面自动校验返回结果，因此即使底层 LLM 想返回格式错误的数据，也无法突破这一约束。

这样一来，我们就可以非常容易地判断是否存在搜索来源。如果列表为空，就表示没有找到任何来源；如果列表中存在字符串，则说明搜索成功。假如没有采用这种结构化输出，而是让 Agent 返回普通文本，那么程序就必须编写额外的解析逻辑，从一大段自然语言中提取搜索来源，还需要处理各种格式错误以及不同 LLM 调用过程中格式发生漂移的问题。

接下来，我们定义了整个流程中的决策点。程序首先检查返回的搜索来源数量。如果没有任何来源，则直接使用一个固定字符串作为研究计划。如果成功获得搜索来源，则继续调用 Thinking Agent 来生成研究计划。在调用 Thinking Agent 之前，我们先构建了一个字典，并为每项输入添加字段名称。当这个字典被转换为字符串发送给 Agent 时，它实际上会变成一份带有标签的 JSON 数据，这能够帮助 Agent 更准确地理解各个字段的含义。

另一种做法是把整个对话历史直接传递给下一位 Agent。虽然这样也能工作，但往往会导致 Agent 接收到大量无关信息，增加理解负担，因此通常更推荐只提供当前任务真正需要的数据。这种方式不仅能够降低 Agent 被上下文干扰的概率，还可以显著减少 Token 消耗。当然，它仍然存在一些局限性，而这些问题将在本章后续内容中进一步讨论。

完成决策之后，系统继续将研究计划（无论是真实生成的还是默认字符串）传递给 Filesystem Agent。Filesystem Agent 唯一的职责就是把结果写入文件。对于 Agent Flow 而言，一个最佳实践是：无论内部流程如何变化，对外始终保持一致的数据格式和输出类型。

这样能够让整个系统更加稳定，也更容易维护和扩展。为了保持示例简洁，本例中省略了一些工程上的细节。下一节将介绍如何对 Agent 的输入和输出进行强类型设计，从而构建更加健壮、更加可靠的 Agent Flow。

4.3 理解 Agent Flow 中的 Handoff（交接）

正如上一节代码示例所展示的，在 Agent 与外部代码之间管理数据流时，必须严格遵循类型和数据格式的规范。至于如何实现这些类型约束和格式约束，则取决于你希望 Agent 采用哪种通信方式。图 4.7 展示了 Agent 系统中最常见的三种通信模式。

本书插图

图 4.7 三种 Agent Flow 通信模式，展示了 Agent 如何相互传递消息，以及如何在 Agent 之间传递决策权（Command）与执行控制权（Control）

从图的左侧开始，可以看到共享会话（Conversational Flow）模式。这是一种已经非常成熟的通信方式，所有 Agent 都在同一条对话线程中进行交流，包括 Agent 的输入输出、工具调用，以及各种中间信息。这种模式最大的优点是所有 Agent 都共享同一份上下文，因此任何 Agent 都能够了解之前发生的一切。但它也存在明显缺点：随着流程推进，越来越多的上下文会被发送给每一个 Agent，从而增加 Token 消耗和推理负担。

图中间展示的是Pass-off（交接）通信模式。在这种模式下，负责调用下一个 Agent 的代码完全控制每次 Agent 调用的输入与输出。上一节中我们已经实践过这种方式。它的优势在于，开发者可以完全掌控每一次 Agent 调用，并能够在流程中插入各种确定性的业务逻辑或决策点。但代价也很明显：所有流程控制都需要手工编写代码，随着 Agent 数量增加，代码复杂度也会迅速上升。

最后，图右侧展示的是 OpenAI Agents SDK 提供的带 Guardrail(护栏) 的增强型 Agent Handoff 模式。这种模式的优势在于，所有 Agent 的通信仍然保留在主会话线程中，因此既能够保持上下文连续性，又能够通过过滤机制减少不必要的 Token 开销。同时，还可以对整个 Agent Flow 进行统一监控。不过，它也有一个缺点：如果某一次 Handoff（交接）过程中发生异常，要恢复整个会话状态并继续执行并不是一件简单的事情，需要一定的技巧。

实际上，许多 Agent Framework 都实现了图 4.7 中某种形式的通信模式，包括我们前面介绍过的各种层级式（Hierarchical）和编排式（Orchestration）架构。接下来，我们重点介绍 OpenAI Agents SDK 所实现的 Conversation Handoff（会话交接）模式。

4.3.1 使用 Handoff 构建 Agent-to-Agent Flow

Handoff（交接）模式允许控制权在 Agent 之间自动转移，而无需像上一节那样由开发者手动编写每一次 Agent 调用逻辑。使用 Handoff 后，大量原本需要编写的编排（Orchestration）代码都可以省略。不过，它也带来了新的问题：Agent 之间的依赖关系被隐藏到了 Prompt 中，因此整个流程不像显式代码那样容易理解和排查。另外，Agent 自身必须知道应该把任务交接给哪一个 Agent。因此，我们不仅需要修改 Agent 的 Instructions（提示词），还需要重新建立 Agent 之间的连接关系，如下面的代码所示。

Listing 4.5 04_agent_to_agent_handoffs.py

# updates to the agents
research_agent = Agent(
   name="Research Agent",
   output_type=ResearchSourcesModel,
   instructions="""
You are a research assistant.
Your role is to find research sources.
Do not make up or invent any research sources.
Always hand off to the thinking agent.    #1
""",
)

thinking_agent = Agent( #2
   name="Thinking Agent",
   instructions="""
You are a research planning assistant.
Your role is to plan the research.
You will receive a list of research sources from the research agent.
Use the sequentialThinking tool to create a research plan based on the sources.
Always hand off to the filesystem agent.    #1
""",
)

filesystem_agent = Agent(
   name="Filesystem Agent",
   instructions="""
You are a filesystem assistant.
Your role is to write the output as a text file.
Never make up or invent any output.
""",
)

# updates to agent connections
research_agent.mcp_servers = [research_srv]
research_agent.handoffs = [thinking_agent]    #2

thinking_agent.mcp_servers = [thinking_srv]
thinking_agent.handoffs = [filesystem_agent]    #2

filesystem_agent.mcp_servers = [fs_srv]

print("Running...", goal)

result = await Runner.run(
   research_agent,    #3
   goal,
   max_turns=25,    #4
)

print(result.final_output)

代码说明

#1 修改 Agent 的提示词，明确告诉 Agent 下一步必须交接（handoff）给哪个 Agent。
#2 为 Agent 设置对应的 handoff 目标，建立 Agent Flow。
#3 现在只需要启动第一个 Agent，后续 Agent 会自动接管。
#4 提高 max_turns，允许整个 Agent Flow 调用更多次 LLM。

可以看到，相比上一节需要手动调用每一个 Agent，这里的代码已经简洁了许多，Agent 之间的切换过程也变得更加自然。不过，这种便利也是有代价的。首先，每个 Agent 的 Prompt 都必须明确写出应该交接给哪个 Agent，而且必须使用目标 Agent 的名称。也就是说，Agent 不再能够完全独立存在，它必须知道整个流程中的下一位参与者。其次，在一个复杂的多 Agent 系统中，由于 Agent 之间的连接关系隐藏在 Prompt 与 Handoff 配置中，整个系统的执行流程可能会变得不容易理解，也更难排查问题。幸运的是，我们还有一些可视化工具，可以帮助我们展示和分析整个 Agent Flow 的结构与执行过程。

4.3.2 可视化 Agent Flow

使用 OpenAI 内置的 Traces 与 Handoff 模式有一个非常实用的优势，那就是能够自动可视化整个 Agent Flow。其中一种方式就是使用 Listing 4.6 中提供的 draw_graph() 方法生成 Agent 流程图。目前，这个可视化还不会显示每个 Agent 所连接的 MCP Server，但 Agent 与 Agent 之间的交接（Handoff）关系会自动绘制出来。

Listing 4.6 05_visualizing_agent_flows.py

research_agent.mcp_servers = [research_srv]
research_agent.handoffs = [thinking_agent]

thinking_agent.mcp_servers = [thinking_srv]
thinking_agent.handoffs = [filesystem_agent]

filesystem_agent.mcp_servers = [fs_srv]

from agents.extensions.visualization import draw_graph

draw_graph(research_agent).view()    #1

input("Press Enter to continue...")    #2

代码说明

#1 显示 Agent Flow 的可视化图。
#2 暂停程序，以便查看生成的流程图。

关于 Agent 的命名，还有几个值得注意的实践经验。Agent 的名称不仅仅是一个普通字符串，它还是 Trace、可视化以及 Handoff 路由所使用的唯一标识符。因此，一旦修改了 Agent 的名称，之前保存的 Trace、Dashboard、监控数据，以及代码中所有引用旧名称的地方都会失效。当系统规模扩大到二十个甚至更多 Agent 时，随意命名很快就会成为团队协作的问题。因此，大多数团队都会制定统一的命名规范。一种比较常见的命名方式是：domain.role.version（领域.角色.版本），例如research.planner.v2、finance.analyst.v1。

当 Agent 数量继续增加，仅靠命名规范已经不足以管理整个系统，这时通常会引入 Agent Registry（Agent 注册中心）。Agent Registry 可以理解为整个 Agent 系统的单一事实来源。它负责维护每个 Agent 的名称、定义、依赖关系以及当前版本。有了 Registry 之后，系统便不再依赖硬编码的 Agent 导入，而是通过 Registry 动态查找 Agent。这样不仅修改 Agent 名称更加安全，也能够非常方便地查看整个系统中有哪些 Agent，以及它们之间的关系。本章不会实现一个完整的 Agent Registry，但随着 Agent 数量不断增长，这种设计模式会变得越来越重要。

除了 draw_graph() 之外，你还可以登录 OpenAI Dashboard，进入 Dashboard → Traces 查看整个 Agent Flow。

图 4.8 展示了 draw_graph() 生成的流程图与 OpenAI Dashboard 中 Traces 页面所展示的 Agent Flow。

本书插图

图 4.8 两种查看 Agent Flow 的方式：draw_graph() 可视化，以及 OpenAI Dashboard（Logs → Traces）中的流程视图

Agent Flow 图是一种非常好的文档化工具，它能够帮助开发者快速理解整个 Agent 系统的执行流程。与此同时，Traces 页面则提供了更加深入的调试能力，你可以查看整个流程中的每一次 Agent 调用、LLM 请求以及 Handoff 过程，从而分析整个 Agent Flow 的实际执行情况。虽然本例中的 Agent Flow 非常简单，但已经足以演示如何可视化多个 Agent 之间的 Handoff 流程。

4.3.3 监控 Handoff（交接过程）

默认情况下，OpenAI Agents SDK 在执行 Handoff 时，并不会告诉开发者究竟有哪些数据被传递给了下一个 Agent。然而，在调试复杂 Agent 系统时，了解 Agent 之间到底传递了什么数据，以及为什么会发生这次 Handoff，往往是定位问题的关键。为了暴露这些信息，我们可以使用 handoff 包装器（handoff wrapper），在 Agent 交接时执行一个回调函数，从而监控每一次 Handoff 所传递的数据。下面的代码展示了如何使用这一机制。

Listing 4.7 06_agent_to_agent_monitoring_handoffs.py

async def research_handoff(    #1
   ctx: RunContextWrapper[None],
   sources: ResearchSourcesModel,    #2
):
   print(f"Thinking agent called with sources: {sources.research_sources}")

research_agent.mcp_servers = [research_srv]

agent_handoff = handoff(    #3
   agent=thinking_agent,
   on_handoff=research_handoff,    #1
   input_type=ResearchSourcesModel,    #2
)

research_agent.handoffs = [agent_handoff]    #3

thinking_agent.mcp_servers = [thinking_srv]
thinking_agent.handoffs = [filesystem_agent]

filesystem_agent.mcp_servers = [fs_srv]

print("Running...", goal)

result = await Runner.run(
   research_agent,
   goal,
   max_turns=25,
)

print(result.final_output)

代码说明

#1 当发生 Handoff 时，将自动调用这个回调函数。
#2 指定 Handoff 输入的数据类型，这里与前一个 Agent 的输出类型一致。当然，也可以传递其他信息，例如触发 Handoff 的原因。
#3 使用 handoff() 将目标 Agent 包装成一个 Handoff 对象，然后交给当前 Agent 使用。

借助这个回调函数，我们就能够实时观察数据是如何在 Agent 之间流动的。除此之外，我们还可以在回调函数中进一步处理这些数据，甚至在 Handoff 发生时触发其他 Agent Workflow，实现更加复杂的业务逻辑。乍一看，这种方式似乎也可以用于验证 Agent 之间传递的数据是否正确。不过，作者指出，对于数据校验，其实还有一种更合适的设计模式——Guardrail（护栏）。一个比较好的实践原则是：

Callback（回调） 用于观察、记录和监控 Agent 之间的数据流，例如日志记录、性能监控或调试。
Guardrail（护栏） 则用于真正控制数据质量，例如阻止非法数据继续流转、修正错误数据，或拒绝执行不符合规则的操作。

换句话说，Callback 负责“看见发生了什么”，而 Guardrail 负责“决定是否允许继续发生”。

4.4 使用 Guardrail（护栏）验证 Agent Flow

要成功构建复杂的 Agent Flow，不仅需要验证进入整个流程的数据，还需要验证流程输出的数据，以及 Agent 与 Agent 之间传递的数据。Guardrail（护栏）正是承担这一职责的重要机制。不过，它并不仅仅是一个内容过滤器。更准确地说，Guardrail 是一种语义层面的断路器。它能够在某个高风险操作真正执行之前就终止整个流程，而不是等到输出已经生成之后再进行过滤。这与传统意义上的输出过滤属于完全不同的设计模式。

当然，这种能力并不是免费的，而且代价值得认真考虑。基于 LLM 的 Guardrail 每执行一次验证，通常都需要额外调用一次（甚至多次）模型。因此，在 Agent Flow 的每一个步骤上都会增加额外的 Token 消耗和响应延迟。对于高并发、大规模部署的 Agent 系统来说，仅 Guardrail 层产生的成本，就可能与 Agent 本身相当，甚至超过 Agent 的成本。

不过，也存在一些成本更低的替代方案，在适用的场景下，它们应该成为首选。例如，正则表达式和 Schema 校验可以以完全确定性的方式检测格式错误或非法输出。不过，如果这些规则设计得过于复杂，同样可能增加一定的执行延迟。另一种常见方案是使用分类模型。这类模型通常规模较小，并经过专门微调，用于识别诸如有害内容、个人隐私信息、Prompt 注入等结构化风险。相比直接调用大型 LLM，它们能够以更低的成本完成风险检测。此外，还可以采用基于代码的验证，例如断言、类型检查以及各种业务规则判断。这些方式能够捕获那些 Agent 本来就不应该产生的错误，因此既高效又可靠。

那么，什么时候才值得使用基于 LLM 的 Guardrail 呢？答案是：**只有当验证本身需要语义理解，而无法通过确定性规则表达时。**例如：

判断一个 Agent 制定的计划是否真正达到了某种模糊的质量标准；
判断回复内容是否符合用户当前表现出的情绪状态；
判断一次 Tool Call 的真实意图是否与用户的原始需求一致。

这些任务都无法依靠简单的规则匹配完成，因为它们要求系统真正理解内容的含义，而不仅仅是比较字符串。

因此，一个更加实用、也更加稳健的安全策略，是采用多层防御。通常可以按照下面的顺序进行验证：

首先执行成本最低、速度最快的确定性检查，例如 Schema、Regex、类型验证等；
然后使用分类模型检测各种结构化风险；
最后，再由 LLM Guardrail 负责处理那些只有语义理解才能完成的验证任务。

需要注意的是，用 LLM 去验证另一个 LLM 本身也存在风险。因为验证模型与执行模型可能拥有相同的偏差或犯下相同类型的错误，因此验证模型并不能保证一定正确。这也是为什么作者强调：LLM Guardrail 不应该成为唯一的防线，而应该作为整个安全体系中的最后一道语义验证层。

接下来，我们将介绍 OpenAI Agents SDK 提供的 Guardrail 实现方式。需要强调的是，这只是生产环境中多层防御体系的一部分，而不是全部。

4.4.1 实现输入与输出 Guardrail

在传统软件开发中，Guardrail 被广泛用于限制程序能够执行的行为。例如，在确定性的程序中，我们会使用 Guardrail 防止诸如除零错误、非法输入、越界访问以及各种异常情况，从而提高整个系统的稳定性。

对于 AI Agent 来说，由于它们天生具有非确定性，同样的输入可能产生不同的输出；与此同时，它们接收到的用户输入往往也具有高度的不确定性，因此整个系统出现异常的概率会更高。因此，在 Agent 系统中，Guardrail 不仅负责限制系统的输入和输出，还能够控制 Agent 与 Agent 之间的数据流，确保整个 Agent Flow 的行为始终符合预期。

这个示例涉及的内容比较多，因此我们先来看图 4.9，从整体上理解 Guardrail 在 Agent Flow 中所处的位置。

本书插图

图 4.9 Guardrail 可以用于验证和控制 Agent 或 Agent Flow 的输入与输出。

从图中可以看到，Guardrail 被放置在 Agent 的外围，对进入 Agent 的输入以及 Agent 输出的结果进行验证，必要时还可以进行修正。这些验证既可以使用传统代码完成，也可以交由其他 Agent 来完成。尤其是基于 LLM 的 Agent，在处理自然语言输入输出时具有非常强的解析和理解能力，因此几乎可以验证任意形式的文本内容。下一节我们将进一步介绍这种方式。

下面的 Listing 4.8 展示了 OpenAI Agents SDK 提供的 Guardrail 模式，以及如何实现输入和输出 Guardrail。

Listing 4.8 07_input_output_guardrails.py

class ResearchOutputModel(BaseModel):    #1
   """Output model for the research agent."""
   research_plan: str
   """The final research plan as text."""
   research_plan_file: str
   """The path to the research plan file."""

@input_guardrail    #2
async def research_guardrail(
   ctx: RunContextWrapper[None],
   agent: Agent,
   input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:

   forbidden_research = "The Hitchhiker's Guide to the Galaxy"

   if forbidden_research in input:
       research_forbidden = True
   else:
       research_forbidden = False

   return GuardrailFunctionOutput(
       output_info=f"user asked: {input}",
       tripwire_triggered=research_forbidden,    #3
   )

@output_guardrail    #4
async def research_output_guardrail(
   ctx: RunContextWrapper,
   agent: Agent,
   output: ResearchOutputModel
) -> GuardrailFunctionOutput:

   if len(output.research_plan) < 100:
       insufficient_research = True
   else:
       insufficient_research = False

   return GuardrailFunctionOutput(
       output_info="research plan length: {len(output.research_plan)}",
       tripwire_triggered=insufficient_research,    #3
   )

# code omitted…

agent = Agent(
   name="Assistant",
   instructions=instructions,
   mcp_servers=[research_srv, thinking_srv, fs_srv],
   output_type=ResearchOutputModel,    #1
   input_guardrails=[research_guardrail],    #2
   output_guardrails=[research_output_guardrail],    #4
)

goal = """
Produce a research plan to find the book 'The Hitchhiker's Guide to the Galaxy'
"""

try:
   print("Running...", goal)
   result = await Runner.run(agent, goal)
   print(result.final_output)

except InputGuardrailTripwireTriggered as input_tripped:
   print(f"""
Input guardrail tripwire triggered:
{input_tripped.guardrail_result.output.output_info}
""")

except OutputGuardrailTripwireTriggered as output_tripped:
   print(f"""
Output guardrail tripwire triggered:
{output_tripped.guardrail_result.output.output_info}
""")

代码说明

#1 为 Agent 指定结构化输出类型，方便后续进行类型检查与输出验证。
#2 输入 Guardrail 会在 Agent 执行之前验证输入内容，如果验证失败，则抛出输入异常。
#3 Guardrail 通过返回布尔值 tripwire_triggered 来决定是否触发 Tripwire，一旦为 True，SDK 会根据 Guardrail 类型抛出对应异常（输入或输出异常）。
#4 输出 Guardrail 会在 Agent 完成后验证输出结果，如果验证失败，则抛出输出异常。

这段代码展示了如何为单个 Agent 同时添加输入 Guardrail 与输出 Guardrail。在两个 Guardrail 内部，仅使用了最简单的字符串比较和长度判断来验证输入与输出是否合法。如果验证失败，就将 tripwire_triggered 设置为 True，从而终止整个 Agent Flow。当然，在真实项目中，Guardrail 内部完全可以实现更加复杂的验证逻辑，例如业务规则检查、多阶段验证、模型调用，甚至调用其他 Agent 来辅助判断，只要最终能够判断输入或输出是否符合要求即可。

4.4.2 使用 Agent 作为 Guardrail

使用 Agent 来验证和约束其他 Agent，会使整个 Agent Flow 变得更加复杂。但与此同时，仅仅通过编写 Prompt，就能够让一个 Agent 完成复杂的验证、标注以及数据处理工作，这种能力带来的价值也不可低估。代码清单 4.9 展示了前面那个带有 Guardrail 的单 Agent 示例，不过这一次，Guardrail 本身也是由一个 Guardrail Agent 驱动的。

该清单仅展示了更新后的代码，主要变化在于：原先使用字符串匹配和长度判断的验证逻辑，被 Agent 所取代。如果后续需要修改输入或输出的验证规则，只需调整 Guardrail Agent 的 Prompt 即可，无需修改大量程序逻辑。这种方式非常灵活，也极具扩展性，但同时也可能引入灾难性的失败风险。幸运的是，第 5 章将介绍如何通过一系列模式，让这种 Agent 与 Agent Flow 架构变得更加健壮。

代码清单 4.9 08_agent_guardrails.py（更新后的代码）

class ResearchOutputModel(BaseModel):
   """Output model for the research agent."""
   research_plan: str
   """The final research plan as text."""
   research_plan_file: str
   """The path to the research plan file."""
   is_sufficiently_detailed: bool    #1
   """Flag to indicate if the research plan is sufficiently detailed."""

class ResearchInputModel(BaseModel):    #2
   """Input model for the research agent."""
   research_validation: str
   """The research goal to achieve."""
   is_research_forbidden: bool
   """Flag to indicate if the research is forbidden."""

input_guardrail_agent = Agent(
   name="Input Guardrail Agent",
   instructions="""
You are an input guardrail agent.
Make sure the research is not about the following topics:
"The Hitchhiker's Guide to the Galaxy"
""",
   output_type=ResearchInputModel,    #2
)

@input_guardrail
async def research_guardrail(
   ctx: RunContextWrapper[None], agent: Agent, input: str | list[TResponseInputItem]
) -> GuardrailFunctionOutput:
   result = await Runner.run(    #3
       input_guardrail_agent,
       input,
       context=ctx.context)
   return GuardrailFunctionOutput(
       output_info=result.final_output.research_validation,
       tripwire_triggered=result.final_output.is_research_forbidden,
   )

output_guardrail_agent = Agent(
   name="Output Guardrail Agent",
   instructions="""
You are an output guardrail agent.
Make sure the research plan is sufficiently detailed.
""",
   output_type=ResearchOutputModel,    #1
)

@output_guardrail
async def research_output_guardrail(
   ctx: RunContextWrapper, agent: Agent, output: ResearchOutputModel
) -> GuardrailFunctionOutput:
   result = await Runner.run(    #3
       output_guardrail_agent,
       str(output),
       context=ctx.context)
   return GuardrailFunctionOutput(
       output_info=result.final_output.research_plan,
       tripwire_triggered=result.final_output.is_sufficiently_detailed is False,
   )

代码说明

1. 更新输出模型，使其既能作为业务 Agent 的输出类型，也能作为输出 Guardrail Agent 的输入和输出模型。
1. 新建一个输入模型，专门供输入 Guardrail Agent 使用，以便更规范地组织和验证输入数据。
1. 在输入和输出 Guardrail 函数内部，不再使用手写代码进行验证，而是调用专门的 Guardrail Agent 来完成验证与检查。

正如 Agents SDK 所实现的那样，Guardrail 能够帮助我们控制 Agent 或 Agent Flow 的输入与输出。不过需要注意的是，在 Agents SDK 的设计中，Guardrail 并不是为了与 Handoff 配合使用而设计的。这是因为 SDK 将 Handoff 视为 Agent 所调用的一种 Tool（工具），而不是独立的数据流控制节点。当然，这并不意味着 Guardrail 无法应用于 Agent 之间的通信。在实际项目中，我们仍然可以在 Agent 与 Agent 之间传递数据时加入 Guardrail，对交互的数据进行验证、过滤或修正，只是需要开发者自行设计相应的实现方式。

4.4.3 为 Pass-off Agent Flow 添加 Guardrail

会话式 Flow 更容易编写和维护，但如果希望精确控制 Agent 之间的数据传递，这种模式就存在一定的局限性。尤其是在 Handoff（交接）过程中，下游 Agent 本身并没有内置机制来防止格式错误的输入、上下文泄露，或者来自上游 Agent 的恶意指令。

在实际应用中，最常见的失败模式主要有三种。

第一种是数据结构错误，即发送方传递的数据结构与接收方预期的不一致。
第二种是上下文泄露**，即发送方把过多的历史记录、规划信息或中间状态一起传递给下游 Agent，导致后者受到无关信息干扰，影响推理质量。
第三种是指令注入。例如，工具返回结果或上游 Agent 的输出中夹带了看似“系统指令”的文本，而接收 Agent 将这些内容误认为可信指令并执行。相比前两种情况，指令注入的风险最高，因为它可能完全劫持接收 Agent 的行为。

Guardrail 正是用于在 Agent 之间的数据边界上堵住这些漏洞。实践中，并不需要为每一次 Handoff 都添加 Guardrail，而应该优先部署在那些高风险交接点，例如写入数据、发送消息、执行不可逆操作等关键节点。

代码清单 4.10 展示了如何利用 Guardrail 控制传递给下一个 Agent 的信息。该示例是在前面的 Pass-off Agent Flow 基础上进行修改，通过 Guardrail 确保研究计划足够详细之后，才允许其继续传递给负责写文件的 Filesystem Agent。如果 Guardrail 被触发，则说明当前流程发生了失败，需要进行恢复或输出备用结果。虽然这种机制看起来与前面介绍的“决策节点”类似，但两者的设计目的并不相同——Decision Point 用于控制流程，而 Guardrail 则用于检测失败并阻止风险继续传播。

代码清单 4.10 09_agent_passoff_guardrails.py（仅展示更新部分）

class ResearchPlanModel(BaseModel):    #1
   """Output model for the research plan."""
   research_plan: str
   """The final research plan as text."""
   is_sufficiently_detailed: bool
   """Flag to indicate if the research plan is sufficiently detailed."""

research_plan_guardrail_agent = Agent(    #2
   name="Research Plan Guardrail Agent",
   instructions="""
You are an output guardrail agent.
Confirm the research plan is sufficiently detailed, atleast 1000 characters in length.
If it is not sufficiently detailed, flag it.
""",
   output_type=ResearchPlanModel,
)

@output_guardrail
async def research_plan_guardrail(    #3
   ctx: RunContextWrapper, agent: Agent, output: ResearchPlanModel
) -> GuardrailFunctionOutput:
   result = await Runner.run(
       research_plan_guardrail_agent, output.research_plan, context=ctx.context
   )
   return GuardrailFunctionOutput(
       output_info=result.final_output,
       tripwire_triggered=result.final_output.is_sufficiently_detailed,
   )

thinking_agent = Agent(
   name="Thinking Agent",
   instructions="""
You are a research planning assistant.
Your role is to plan the research.
You will receive a list of research sources from the research agent.
Use the sequentialThinking tool to create a research plan based on the sources.
Always hand off to the filesystem agent.
""",
   output_type=ResearchPlanModel,
   output_guardrails=[research_plan_guardrail],
)

research_agent.mcp_servers = [research_srv]
result = await Runner.run(research_agent, goal)

thinking_agent.mcp_servers = [thinking_srv]
try:
   result = await Runner.run(thinking_agent, result.final_output)
   final_output = result.final_output.research_plan
except OutputGuardrailTripwireTriggered:
   final_output = "A research plan was not generated. Please try again with a different goal."

filesystem_agent.mcp_servers = [fs_srv]
result = await Runner.run(filesystem_agent, final_output)
print(result.final_output)

代码说明

1. 更新研究计划的数据模型，增加一个布尔字段，用于标记研究计划是否足够详细。
1. 创建一个专门用于验证研究计划质量的 Guardrail Agent。
1. 定义输出 Guardrail，在其中调用 Guardrail Agent 对研究计划进行验证。
1. 更新 Thinking Agent 的输出类型，使其使用新的 ResearchPlanModel。
1. 为 Thinking Agent 挂载输出 Guardrail，使其生成的结果必须先通过验证。
1. 将 Thinking Agent 的输出保存为字符串，以便后续传递给 Filesystem Agent。
1. 如果触发 Output Guardrail，则捕获异常，并生成备用输出。

在这个示例中，如果研究计划不够详细，我们完全可以不直接结束流程，而是重新调用 Thinking Agent，让它再次生成更完善的研究计划。下一段代码便展示了这种自动重试机制。

代码清单 4.11 10_agent_guardrails_retry.py

# only relevant code shown
print("Running...", goal)

research_agent.mcp_servers = [research_srv]
result = await Runner.run(research_agent, goal)

thinking_agent.mcp_servers = [thinking_srv]

final_output = result.final_output
max_retries = 3

for attempt in range(max_retries):
   try:
       result = await Runner.run(thinking_agent, final_output)
       final_output = result.final_output.research_plan
       break
   except OutputGuardrailTripwireTriggered as output_tripped:
       final_output = output_tripped.guardrail_result.output.output_info

   if attempt == max_retries - 1:
       final_output = "A research plan was not generated. Please try again with a different goal."

filesystem_agent.mcp_servers = [fs_srv]
result = await Runner.run(filesystem_agent, final_output)
print(result.final_output)

代码说明

1. 将上一阶段 Agent 的输出保存到变量中，方便后续不断替换最新的输入。
1. 使用 for 循环实现有限次数的自动重试。
1. 当 Output Guardrail 被触发时，将 Guardrail 返回的反馈信息作为下一轮 Thinking Agent 的输入，引导 Agent 改进输出。
1. 如果所有重试均失败，则生成固定的错误信息，并继续后续工作流。

需要注意的是，在这个实现中，我们会根据当前是否属于首次执行还是重试，动态替换传递给 Thinking Agent 的输入内容。同时，Python 能够方便地将各种简单数据类型自动转换为字符串，因此可以直接作为 Agent 的输入。

恢复机制和重试机制是提升 Agent 输出质量的重要手段，它们能够让 Agent 根据反馈不断迭代，逐步生成更好的结果。不过，重试次数绝不能无限制，否则很容易形成死循环。因此，生产环境通常都会设置最大重试次数，当达到上限后便返回明确的错误信息。此外，每一次重试都会额外消耗 Token、增加响应延迟，而且并不能保证一定能够修正错误。因此，在设计 Agent Flow 时，应谨慎决定哪些环节值得重试，哪些环节则应直接失败并交由上层流程处理。

生产环境实践建议

当重试是由于限流、网络故障等临时性错误触发时，不应立即再次重试，而应该采用指数退避+ 抖动策略。也就是说，每次失败后等待的时间逐渐增加，并在等待时间中加入一定的随机偏移。这样做可以有效避免惊群效应。如果大量失败的 Agent 在同一时刻立即重试，它们会同时向服务发起新的请求，不仅无法解决问题，反而会进一步加重系统负载，使原本的故障更加严重。

设计 Agent Flow 时，你需要决定希望在什么层级控制 Agent 之间的通信。如果希望快速构建 Agent 系统，可以在会话式 Flow 之上使用 Handoff 模式，它能够减少大量编排代码，让多个 Agent 自然地接力完成任务。不过，正如下一节将介绍的，对于需要更复杂控制逻辑、更精细通信方式以及需要与外部 Agent 集成的场景，Pass-off 模式通常是更好的选择，因为它能够让开发者完全掌控 Agent 之间传递的数据和执行流程。

当然，这些模式之间并不是互斥的，我们完全可以根据系统的发展不断调整和组合。例如，可以将原本 Flow 中的某一个节点升级为一个 Orchestrator（编排 Agent），从而让整个 Agent Flow 演化成 Orchestration 架构。代码清单 4.12 就展示了这种实现方式。

在这个示例中，Orchestration Agent 不再直接完成所有工作，而是将其他 Agent 当作自己的工具来调用。这样，Orchestrator 就能够统一规划整个流程，并根据需要将任务委派给不同的 Agent。同时，如果观察代码结构，还会发现各个 Agent 的职责变得更加独立，模块之间的边界也更加清晰。

代码清单 4.12 11_agent_orchestration_tools.py

@function_tool    #1
async def research_agent(instructions: str) -> ResearchSourcesModel:
   """
   Use the research agent to find research sources.
   """
   agent = Agent(
       name="Research Agent",
       instructions="""
You are a research assistant.
Your role is to find research sources.
Never make up or invent any research sources.
""",
       output_type=ResearchSourcesModel,
       mcp_servers=[research_srv],
   )
   async with research_srv:
       result = await Runner.run(agent, instructions)
       return result.final_output


orchestration_agent = Agent(    #2
   name="Orchestration Agent",
   instructions="""
You are a research planning and orchestration assistant.
Your role is to plan the research, find existing research already done and update it.
Use the research agent to find research sources.
Use the sequentialThinking tool to create a research plan based on the sources.
Use the filesystem agent to help find existing research and update it.
Use the filesystem agent to write the output as a text file.
""",
   tools=[research_agent, filesystem_agent],    #1
)

orchestration_agent.mcp_servers = [thinking_srv]    #3

print("Running...", goal)

result = await Runner.run(
   orchestration_agent,
   goal,
   max_turns=25,
)

print(result.final_output)

代码说明

1. 将 Research Agent 封装成一个 Function Tool，使其他 Agent 可以像调用普通工具一样调用它。
1. 创建 Orchestration Agent，它负责整个流程的规划、任务拆分以及任务委派，并通过 Prompt 明确说明何时调用哪些工具。
1. Research Agent 和 Filesystem Agent 不再直接参与 Flow，而是作为 Tool 被 Orchestrator 调用；Orchestrator 自身仍然使用 sequentialThinking MCP Server 完成规划工作。
1. 整个程序只需要运行 Orchestration Agent，它会自动决定何时调用各个 Tool Agent 来完成整个任务。

图 4.10 展示了执行该 Agent 系统后，在 Traces 页面中看到的运行结果。

本书插图

图 4.10 执行 Orchestration Agent 后的 Traces 页面

在 Traces 页面中，可以清楚地看到所有被 Orchestration Agent 调用的 Agent，它们都会以 Tool 或任务委派（Delegation）的形式出现。同时，整个执行过程中使用的 MCP Server 以及其他工具调用，也都会被完整记录下来。因此，如果希望获得更加清晰的 Trace 信息，可以考虑将 sequentialThinking 工具进一步拆分到一个独立的 Agent 中，而不是直接作为 Orchestrator 的 MCP Tool 使用。总体来说，Orchestration 模式最大的优势就在于能够集中控制整个 Agent 系统的执行过程。

不过，如果实际运行这一版本的 Agent 系统，并将其输出结果与前面的 Agent Flow 示例进行比较，就会发现最终生成的研究计划中，缺少了之前明确列出的研究工具（Research Sources）。这个问题当然可以通过进一步优化 Prompt 来解决，但这种解决方案往往不够稳定，更容易受到模型随机性的影响，导致输出质量波动甚至失败。

因此，一个经验法则是：**在真正构建出几个稳定可靠的 Agent Flow 之前，不要急于采用 Orchestration 模式。**很多时候，提高 Agent Flow 效果的最佳方式，并不是设计更加复杂的架构，而是做好以下几件事情：

持续优化 Agent 的 Instructions（系统提示）。
尽可能使用结构化的数据类型作为输入和输出。
在关键节点合理部署 Guardrail，确保数据和流程的可靠性。

保持架构简单，避免过度设计，通常比堆砌复杂的多 Agent 架构更容易构建出稳定、可靠的 Agentic System。

下一章将进一步深入探讨 Agent 如何进行决策以及任务规划。

4.5 练习

练习 1：将单个 Agent 重构为两步 Flow

目标： 体验最简单的 Agent 到 Agent 转换过程。

任务：

打开 01_single_agent_multiple_mcp.py。
将其复制为 exercise1_two_step_flow.py。
将单个 Agent 拆分为 ResearchAgent 和 PlanningAgent（复用已声明的 MCP Server）。
将第一个 Agent 的输出直接作为第二个 Agent 的输入（连续调用两次 Runner.run）。
运行新的脚本，确认最终控制台输出仍然包含研究计划文件的文件名。

预计耗时： 10 分钟

练习 2：加入确定性的决策节点

目标： 根据一个简单的类型化条件添加流程分支。

任务：

以练习 1 的代码为基础继续开发。
创建一个 ResearchSourcesModel（继承 pydantic.BaseModel），内部包含 List[str]。
将 ResearchAgent 的输出限制为该模型（使用 output_type=）。
在第一次 Runner.run 之后，统计返回列表的长度。
如果 len(sources) == 0，输出 "No sources—flow aborted" 并退出程序。
否则继续执行 PlanningAgent。
连续运行程序三次，你应该偶尔能够看到流程被终止（aborted）的分支。

预计耗时： 12 分钟

练习 3：将流程改造成 SDK Handoff

目标： 使用 SDK 内置的 Handoff 机制替代手动串联。

任务：

将 02_agent_to_agent_flow.py 复制为 exercise3_handoffs.py。
在每个 Agent 的提示词中增加一句说明，将控制权交给下一个 Agent（例如："Always hand off to Planning Agent."）。
设置 agent_a.handoffs = [agent_b]。
只调用第一个 Agent（Runner.run），并设置 max_turns=20。
在 OpenAI Traces 页面（或通过 draw_graph）确认整个流程自动完成了三次交接：Research → Planning → Filesystem。

预计耗时： 15 分钟

练习 4：可视化并分析 Agent 图

目标： 学会生成并阅读 Agent Flow 图。

任务：

在 exercise3_handoffs.py 中导入 agents.extensions.visualization 下的 draw_graph。
在执行 Runner.run 之前调用 draw_graph(research_agent).render(view=True)。
浏览器打开图形后，截图或记录所有节点（Node）和边（Edge）。
总结一条观察结果，例如："ResearchAgent 并不会直接调用 Filesystem Tool。" 并将其写成脚本末尾的注释。

预计耗时： 5 分钟

练习 5：添加带自动重试的输出 Guardrail

目标： 防止生成质量较差的研究计划，并能够自动恢复。

任务：

将 03_agent_to_agent_decisions.py 复制为 exercise5_guardrail_retry.py。
创建一个 ResearchPlanModel，包含两个字段：
- research_plan: str
- is_detailed: bool
创建一个输出 Guardrail Agent，当 research_plan.__len__() >= 1000 时，将 is_detailed 设置为 True。
使用 @output_guardrail 包装 PlanningAgent，并通过该 Guardrail 调用 Agent。
实现一个最多重试 2 次的 Retry Loop（参考清单 4.10）。如果两次重试仍失败，则写入一个名为 PLAN_TOO_SHORT.txt 的文件，其中仅包含一行文字，说明生成失败。
运行程序，并临时将长度阈值改为 100，分别验证成功路径和失败路径都能够正常工作。

预计耗时： 15 分钟

本章总结

单 Agent 架构在功能不断增加后通常会遇到扩展性瓶颈，将其拆分为由多个专业 Agent 组成的 Agent Flow，可以重新获得更好的可维护性、可扩展性和执行性能。
Agent-to-Agent Flow 可以看作是 Prompt Chaining 的升级版。每个节点不仅能够独立推理，还可以调用工具，并将精简、类型化的结果传递给下游 Agent，从而减少上下文开销。
对于必须保证一致性的流程，应在关键位置加入确定性的决策节点（例如代码判断或 Schema 校验），不要把通过/失败这样的关键决策完全交给具有随机性的 LLM。
OpenAI Agents SDK 提供两种 Handoff 模式：共享会话（Conversational）和显式传递（Pass-off）。前者开发效率更高，后者能够提供更细粒度的流程控制。
通过配置 Agent 的 handoffs 字段，并在提示词中明确说明交接对象，即可实现 Agent 之间的自动切换，而无需编写额外的编排代码。
应尽早使用 draw_graph() 对 Agent Flow 进行可视化，并结合 Dashboard 的 Traces 页面分析隐藏的循环、工具调用链以及性能瓶颈。
对于高风险的数据传递，应使用 Guardrail 对输入和输出进行验证，在数据进入下一阶段之前进行拒绝、重试或修正。Guardrail 被触发时会抛出明确的异常，方便实现自动恢复流程。
Guardrail 本身也可以由 LLM Agent 实现，这样可以通过自然语言描述验证规则，而不必依赖脆弱的正则表达式或简单长度判断。不过，这类 Guardrail 同样需要结构化 Schema 和充分测试。
即使采用多 Agent 架构，也要严格控制每个 Agent 可访问的工具数量。每增加一个工具，都会增加每次调用时的 Token 开销，因此建议每个 Agent 的工具数量控制在 10 个以内，并且仅保留与自身职责相关的工具。
Flow、Orchestration 以及 Manager-Worker 是三种最常见的 Agent 编排模式。建议优先采用简单的 Flow，当确实需要集中调度和统一决策时，再引入 Orchestrator。
一个项目最好只选择一种 Agent 通信方式（例如共享上下文、消息传递、MCP 或新兴的 A2A 协议）。混合多种通信机制会显著增加系统调试和维护成本。
一个可投入生产环境的 Agent Flow 通常具备以下几个关键特征：结构化输入输出、确定性的检查点、流程可视化、精简的工具集以及完善的 Guardrail。这些能力共同保证了系统具有良好的扩展性、恢复能力和稳定性。
Agent 可以采用多种协作策略，包括：顺序流水线（Sequential Flow）、并行委派（Parallel Delegation）、层级协调（Hierarchical Coordination）、迭代辩论与优化（Iterative Debate and Refinement）、投票/Best-of-N 集成（Voting / Best-of-N Ensemble）、角色协作（Role-playing Collaboration）、条件路由（Conditional Routing）以及点对点网络（Peer-to-peer Network）。

本章内容

4.1 多智能体系统架构设计

表 4.1 Agent 三个核心概念

4.1.1 决策与控制模式

表 4.2 Agent 架构对比

4.1.2 使用共享内存、消息传递和 MCP 进行通信

表 4.3 Flow 模式中的通信方式比较

4.1.3 多智能体协调策略（Coordination）

顺序流水线（Sequential Pipeline，Agent-to-Agent）

并行委派（Parallel Delegation）

层级协调（Hierarchical Coordination）

迭代式辩论与优化（Iterative Debate and Refinement）

投票 / Best-of-N（集成方法，Ensemble）

角色扮演协作（Role-playing Collaboration）

条件路由（Conditional Routing / Branching）

点对点网络（Peer-to-Peer Network）

表 4.4 Agent 架构模式概览

4.2 在 Agent 与 Agent Flow 之间取得平衡

4.2.1 将单个 Agent 演化为 Agent Flow

列表 4.1 01_single_agent_multiple_mcp.py

4.2.2 构建 Agent-to-Agent Flow

列表 4.2 02_agent_to_agent_flow.py

4.2.3 Agent Flow 中的自主性与决策

列表 4.3 03_variable_research_tools.py

列表 4.4 03_agent_to_agent_decisions.py

4.3 理解 Agent Flow 中的 Handoff（交接）

4.3.1 使用 Handoff 构建 Agent-to-Agent Flow

Listing 4.5 04_agent_to_agent_handoffs.py

4.3.2 可视化 Agent Flow

Listing 4.6 05_visualizing_agent_flows.py

4.3.3 监控 Handoff（交接过程）

Listing 4.7 06_agent_to_agent_monitoring_handoffs.py

4.4 使用 Guardrail（护栏）验证 Agent Flow

4.4.1 实现输入与输出 Guardrail

Listing 4.8 07_input_output_guardrails.py

4.4.2 使用 Agent 作为 Guardrail

代码清单 4.9 08_agent_guardrails.py（更新后的代码）

4.4.3 为 Pass-off Agent Flow 添加 Guardrail

代码清单 4.10 09_agent_passoff_guardrails.py（仅展示更新部分）

代码清单 4.11 10_agent_guardrails_retry.py

生产环境实践建议

代码清单 4.12 11_agent_orchestration_tools.py

4.5 练习

练习 1：将单个 Agent 重构为两步 Flow

练习 2：加入确定性的决策节点

练习 3：将流程改造成 SDK Handoff

练习 4：可视化并分析 Agent 图

练习 5：添加带自动重试的输出 Guardrail

本章总结