基于多 Agent 的 Planning-Executor 架构设计

一、核心设计原则

1.1 统一的 Planning 能力

每个 Agent 都具备独立的 Planning 能力，这是多 Agent 系统的基础设计原则。无论 Agent 的角色如何（Leader 或 Executor），都需要能够：

理解任务需求

分解复杂任务为可执行的步骤

生成执行计划

设计理由：

支持将复杂任务递归分解为可执行的步骤：Agent 在执行任务时可能需要进一步分解子任务

提高系统的灵活性和可扩展性

避免单点故障，每个 Agent 都能独立工作

1.2 Agent 技能差异化

虽然所有 Agent 都具备 Planning 能力，但每个 Agent 的技能集（工具函数）是不同的。这是基于以下考虑：

资源限制：不可能让每个 Agent 都拥有所有技能

专业化：不同 Agent 专注于不同领域，提高执行效率

模块化设计：便于维护和扩展

LLM 上下文窗口限制：当前主流 LLM 的上下文窗口有限（如 GPT-4 的 128K tokens），如果将所有工具描述都放入上下文，会占用大量 token工具数量过多时，工具描述本身就会消耗大量上下文空间，留给实际对话和推理的空间会大幅减少通过技能差异化，每个 Agent 只需要了解自己领域的工具，大幅减少上下文消耗

工具选择准确性问题：当工具数量过多时，LLM 容易出现"工具选择困难症"，在相似功能的工具之间犹豫不决工具描述过多会导致 LLM 理解偏差，可能选择错误的工具或传递错误的参数通过专业化分工，每个 Agent 的工具集更小更精准，降低工具误用概率

推理能力限制：复杂任务需要多步骤推理，如果一次性提供所有工具和完整任务，LLM 的推理链容易断裂通过任务分解和专业化 Agent，每个 Agent 只需处理相对简单的子任务，降低推理复杂度

二、Planning 机制设计

2.1 Plan Stack 数据结构

每个 Agent 维护一个Plan Stack（计划栈），采用 LIFO（后进先出）的数据结构。

Stack 结构：

PlanStack:

Layer0: [Task1, Task2, Task3, ...]  <- 当前执行层

Layer1: [TaskA, TaskB, TaskC, ...]  <- 上一层计划

Layer2: [TaskX, TaskY, TaskZ, ...]  <- 更上层计划

...

为什么需要 Stack：

当 Agent 创建完一个计划任务列表后，其中某个任务可能需要 Agent 自己执行。在执行这个任务时，Agent 又会创建新的计划列表。如果不使用 Stack，新计划会"替换"当前计划，导致上层计划丢失。

典型场景：

Leader Agent 创建主计划：[搜索信息, 分析结果, 生成报告]

执行"搜索信息"任务时，需要创建子计划：[搜索关键词A, 搜索关键词B, 整合结果]

执行"搜索关键词A"时，可能需要更细粒度的计划：[调用搜索工具, 过滤结果, 提取关键信息]

当子任务完成后，需要回到上一层计划继续执行

Stack 操作：

Push：创建新计划时，将当前计划压入 Stack，新计划成为当前层

Pop：当前层计划完成后，弹出 Stack，恢复上一层计划

Peek：查看当前层计划，不修改 Stack

2.2 Plan 的数据结构

Plan 字段说明

Task Group 字段说明

Job 字段说明

Status 可选值（约定）

任务执行顺序说明：

Plan 采用任务组（Task Group）的结构，支持任务组级别的并行控制：

任务组顺序执行：任务组按照列表顺序依次执行

任务组内执行方式：由 parallel 字段控制parallel: false ：任务组内的任务按顺序执行（串行）parallel: true ：任务组内的任务可以同时执行（并行）

上下文传递：后续任务可以访问前面已完成任务的结果（通过 Task Result）

失败处理：如果某个任务失败，可以根据策略决定是否继续执行后续任务

执行流程示例：

示例1：串行任务组

Plan:

TaskGroup 1:"信息收集阶段"(parallel:false)

-Job 1:搜索相关资料

-Job 2:整理资料

TaskGroup 2:"内容创作阶段"(parallel:false)

-Job 3:撰写大纲

-Job 4:撰写文章

执行流程：

1.执行TaskGroup1（串行）：

-执行Job1完成

-执行Job2完成（可以使用Job1的结果）

2.执行TaskGroup2（串行）：

-执行Job3完成（可以使用TaskGroup1的结果）

-执行Job4完成（可以使用Job3的结果）

3.Plan完成

示例2：并行任务组

Plan:

TaskGroup 1:"多源信息收集"(parallel:true)

-Job 1:搜索主题A

-Job 2:搜索主题B

-Job 3:搜索主题C

TaskGroup 2:"信息整合"(parallel:false)

-Job 4:整合所有搜索结果

-Job 5:生成报告

执行流程：

1.执行TaskGroup1（并行）：

-并发执行Job1,Job2,Job3

-等待所有任务完成后继续

2.执行TaskGroup2（串行）：

-执行Job4完成（可以使用Job1,2,3的结果）

-执行Job5完成（可以使用Job4的结果）

3.Plan完成

任务组设计优势：

清晰的并行控制：通过 parallel 字段明确控制任务组内的执行方式

灵活的组织方式：可以将相关任务组织成任务组，提高计划的可读性

高效的执行：支持任务组级别的并行，提高执行效率

简单的依赖管理：任务组之间的依赖通过顺序自然体现，无需显式依赖关系

易于理解：任务组名称和并行标志使计划结构一目了然

Team Role 选择原则：

1. Team Role 应该是一个描述性的角色名称，能够清晰表达任务所需的专业能力

2. 根据任务的核心能力需求选择 Team Role，例如：搜索信息、收集资料：searcher, researcher创作内容、撰写文章：writer, content_creator图像设计、美术创作：artist, designer3D建模、技术实现：modeler, developer

3. 系统会根据 Team Role 自动找到对应的团队成员来执行任务

三、Agent 团队架构

3.1 角色划分

Leader Agent（领导者）

职责：

分析用户输入的领域问题

理解用户意图和需求

创建初始执行计划

协调团队成员执行任务

汇总和整合执行结果

特点：

具备全局视野和问题分解能力

不专注于具体技能执行，而是任务规划

可以调用自己（递归规划），例如：总结任务执行结果重新理解用户需求调整执行计划

Executor Agent（执行者）

职责：

执行具体的技能任务

专注于特定领域的工具使用

返回任务执行结果

特点：

具备特定技能集（如搜索、写作、数据分析等）

更倾向于执行而非分析

可以相互调用，形成协作链，例如：搜索 Agent 调用写作 Agent 完成搜索结果的总结数据分析 Agent 调用可视化 Agent 生成图表

3.2 调用关系规则

Leader 调用规则

Leader 不能被 Executor 调用

原因：Leader 负责理解和分解最初的用户问题，不是具体执行者

例外情况：当用户问题发生变化或需要重新理解时，Executor 可以请求 Leader 重新分析

实现方式：通过特殊的"重新规划"请求机制

Leader 可以调用自己场景：

(1). 总结多个任务的执行结果

(2). 根据执行结果调整计划

(3). 重新评估用户需求

Executor 调用规则

Executor 可以相互调用

场景：

搜索 Agent 写作 Agent：搜索完成后需要总结

数据分析 Agent 可视化 Agent：分析完成后需要可视化

多个 Executor 协作完成复杂任务

Executor 可以调用 Leader（特殊情况）

场景：

用户需求发生变化

执行过程中发现需要重新理解问题

实现方式：通过"重新规划请求"机制，而非直接调用

3.3 团队协作流程

四、Agent 实例管理

4.1 Agent ID vs Agent Instance ID

Agent ID（模板标识）：

表示 Agent 的类型/模板

例如：searcher, writer, analyzer

一个 Agent ID 可以对应多个运行实例

Agent Instance ID（实例标识）：

表示 Agent 的运行实例

唯一标识符，例如：searcher_001, searcher_002

用于区分同一类型 Agent 的不同实例

为什么需要 Instance ID：

在运行时，可能会同时启动多个相同 Agent ID 的实例，例如：

多个 searcher 并发搜索不同主题

多个 writer 并行处理不同的写作任务

多个 analyzer 同时分析不同的数据集

实例管理：

每个 Agent Instance 维护独立的对话上下文

每个 Agent Instance 维护独立的 Plan Stack

通过 Instance ID 进行任务路由和结果关联

4.2 消息隔离机制

LLM 交互的消息隔离

在跟 LLM 进行交互时，必须区分不同 Agent Instance 的消息，确保每个 Agent Instance 的对话消息是分开独立的。

原因：

LLM 对消息顺序有严格要求

Tool Call 的消息后面必须紧跟 Tool 的返回消息

如果多个 Agent 的消息混合在一起，会导致：LLM 无法正确理解上下文Tool Call 和 Tool Response 不匹配对话状态混乱

实现方式：

每个 Agent Instance 维护独立的 Message History

消息存储时关联 Agent Instance ID

LLM 调用时只传入当前 Agent Instance 的消息

五、Agent 协作与信息共享

5.1 任务执行上下文共享

虽然每个 Agent Instance 的消息是隔离的，但 Agent 之间需要协作，因此需要共享任务执行上下文。

共享的信息：

任务输入：谁收到了什么任务（消息）

任务输出：输出了什么任务结果（消息）

任务状态：任务的执行状态（pending, executing, completed, failed）

任务顺序：任务在 Plan 中的执行顺序（通过列表顺序自然体现）

共享方式：

内部 Agent（同一系统内）

内存共享：适用于单机部署，速度快

Redis 共享：适用于分布式部署，支持跨进程

数据库共享：适用于需要持久化的场景

外部 Agent（通过 A2A 接入）

封装层：不能直接继承调用

安全隔离：避免涉密信息泄露

接口标准化：通过标准化的 API 接口交互

信息过滤：只共享必要的任务上下文，过滤敏感信息

5.2 上下文共享的数据结构

5.3 外部 Agent 的安全封装

封装层设计：

代理模式：外部 Agent 通过代理层接入

权限控制：限制外部 Agent 的访问范围

数据脱敏：共享前进行数据脱敏处理

审计日志：记录所有外部 Agent 的交互

实现示例：

classExternalAgentWrapper:

def__init__(self, agent_id: str, endpoint: str, credentials: dict):

self.agent_id = agent_id

self.endpoint = endpoint

self.credentials = credentials

self.sanitizer = DataSanitizer()

defexecute_task(self, task: TaskExecutionContext) -> TaskExecutionContext:

# 数据脱敏

      sanitized_task = self.sanitizer.sanitize(task)

# 调用外部 Agent

      result = self._call_external_agent(sanitized_task)

# 记录审计日志

self._log_interaction(task, result)

return result

六、实现要点总结

6.1 核心数据结构

Plan Stack：每个 Agent Instance 一个 Plan Stack

Message History：每个 Agent Instance 一个 Message History，隔离存储

Task Execution Context：共享存储，所有 Agent 可访问

6.2 关键设计决策

6.3 待完善的设计点

Plan 的版本管理：当计划需要调整时，如何管理版本

失败重试机制：任务执行失败时的重试策略

任务执行策略：任务失败时是否继续执行后续任务，还是停止整个 Plan

资源限制：如何限制 Agent 的资源使用（如并发数、内存等）

监控和调试：如何监控和调试多 Agent 系统的运行状态