Agent 工程体系 · 第 16/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线;框架版本敏感能力会明确标注,不把实验行为写成通用保证。

Agent Plan-and-Execute:任务分解、依赖、重规划和进度状态

Plan-and-Execute 是一种把 Agent 运行过程显式拆成两个阶段的架构:

  1. Plan:根据目标生成任务计划,包括子任务、依赖关系、执行顺序、产物和验收条件。
  2. Execute:按照当前计划调度可执行任务,调用工具或子 Agent,记录结果,并在环境变化或任务失败时决定是否重规划。

它解决的不是“让模型多想几步”这么简单,而是把一个开放目标转换成一个可调度、可观测、可恢复、可修订的任务系统

OpenAI 将 Agent 概括为能够规划、调用工具、协作并维护足够状态以完成多步工作的应用;Anthropic 则区分了预定义代码路径的 workflow 与由模型动态决定过程和工具使用方式的 agent。Plan-and-Execute 位于两者之间:计划结构可以由模型生成,但执行、状态持久化、依赖检查和风险控制通常由程序负责。(developers.openai.com)


1. Plan-and-Execute 到底解决什么问题

考虑一个看似简单的目标:

为一个新产品做竞品调研,输出一份包含市场定位、价格、功能差异和结论的报告。

直接让 Agent 使用 ReAct 循环处理,可能得到这样的过程:

思考:我需要了解竞品。
行动:搜索产品 A。
观察:得到一篇产品介绍。
思考:还需要产品 B。
行动:搜索产品 B。
观察:得到价格页。
思考:应该写报告了。
行动:生成报告。

问题在于:

  • 没有明确哪些信息必须收集;
  • 没有记录“价格”是否已经覆盖所有竞品;
  • 没有定义什么结果算完成;
  • 某次搜索失败后,无法判断是重试、换查询,还是修改整体方案;
  • 报告生成过早,导致后续发现缺失信息;
  • 多个独立竞品无法安全并发;
  • 中断后只能依赖上下文恢复,不能从结构化状态继续。

Plan-and-Execute 的核心变化是先形成一个可检查的计划:

目标:输出竞品调研报告

T1:确定竞品集合
T2:收集竞品 A 的功能和价格
T3:收集竞品 B 的功能和价格
T4:收集竞品 C 的功能和价格
T5:统一整理比较维度
T6:生成报告
T7:检查报告是否满足验收条件

依赖:
T1 → T2、T3、T4
T2、T3、T4 → T5
T5 → T6
T6 → T7

此时 Agent 不再只是“连续地产生下一步动作”,而是在一个显式任务图上推进。


2. 计划、执行和 ReAct 的关系

2.1 ReAct 是局部循环,Plan-and-Execute 是全局控制

ReAct 通常描述如下循环:

观察当前上下文
  ↓
决定下一步思考或行动
  ↓
调用工具
  ↓
获得观察结果
  ↓
继续循环

它适合处理:

  • 下一步动作高度依赖刚刚获得的观察结果;
  • 路径难以提前确定;
  • 工具调用数量较少;
  • 任务更像实时探索。

Plan-and-Execute 则先建立较长范围的任务结构:

生成计划
  ↓
选择当前可执行任务
  ↓
执行任务并记录产物
  ↓
检查依赖、进度和验收条件
  ↓
继续执行或重规划

两者并不互斥。一个子任务内部完全可以使用 ReAct:

全局计划:
T2:收集竞品 A 的价格

T2 内部:
搜索官网
  ↓
观察页面结构
  ↓
发现价格由地区决定
  ↓
调用地区价格接口
  ↓
验证结果

因此,比较准确的关系是:

Plan-and-Execute 负责任务级编排,ReAct 负责步骤级探索。

如果把所有逻辑都交给 ReAct,系统容易出现“局部合理、全局失控”;如果把所有路径都写死为计划,又会失去 Agent 处理未知情况的能力。

2.2 与固定 Workflow 的边界

Anthropic 将 prompt chaining、routing 和 parallelization 等模式归为由代码预先编排的 workflow,并指出固定子任务适合使用这类方式;Agent 更适合需要灵活性和模型驱动决策的任务。(anthropic.com)

Plan-and-Execute 的计划部分可以有不同来源:

计划来源 特征 适用情况
固定代码 结构稳定、可预测 合规流程、订单处理
模型生成 灵活,但需要验证 研究、分析、复杂操作
模板加模型填充 兼顾约束和适应性 有固定交付格式的开放任务
人工审批后执行 高风险动作受控 发布、付款、删除、外发

不要把“计划由模型生成”误认为“整个系统必须由模型控制”。在生产系统中,模型通常只负责提出计划,程序负责验证计划是否满足结构、权限和安全约束。


3. 任务分解:从目标到可执行子任务

3.1 目标不是任务

目标描述的是期望状态:

“输出一份竞品调研报告”

任务描述的是可以执行和验收的工作:

“从官方价格页提取竞品 A 的月付和年付价格,保存为结构化 JSON”

一个合格的子任务至少应包含:

{
  "id": "T2",
  "title": "收集竞品 A 的功能和价格",
  "objective": "获得可引用的功能列表和价格信息",
  "preconditions": [
    "竞品集合中存在产品 A",
    "允许访问公开网页"
  ],
  "inputs": [
    "产品 A 的官网地址"
  ],
  "expected_artifacts": [
    "feature_matrix_A.json",
    "source_records_A.json"
  ],
  "acceptance": [
    "至少包含核心功能、计费周期和币种",
    "每个价格字段都有来源记录",
    "无法确认的字段标记为 unknown"
  ],
  "risk": "medium"
}

这里的关键不是字段数量,而是将“完成”从自然语言愿望转成机器可检查的条件。

3.2 可执行性的形式化条件

设一个子任务为:

Ti=(Ii,Pi,Ai,Ci,Ri)T_i = (I_i, P_i, A_i, C_i, R_i)

其中:

  • IiI_i:输入集合;
  • PiP_i:前置条件;
  • AiA_i:预期产物;
  • CiC_i:验收条件;
  • RiR_i:风险和权限约束。

在状态 SS 下,任务可执行的基本条件为:

enabled(Ti,S)=available(Ii,S)satisfied(Pi,S)policy_allow(Ri,S)enabled(T_i, S) = available(I_i, S) \land satisfied(P_i, S) \land policy\_allow(R_i, S)

直觉是:

  • 没有输入,不能开始;
  • 前置任务没有产出,不能开始;
  • 即使技术上可以开始,权限或风险策略不允许,也不能开始。

例如:

T6:生成报告

如果 T5 尚未生成统一比较表,即使模型可以直接写报告,enabled(T6, S) 仍然应该为假。

3.3 分解的终止条件

分解不是越细越好。对子任务继续分解,直到满足以下条件之一:

  1. 可以由一个工具调用直接完成;
  2. 可以交给一个受限 Agent 在单次局部循环中完成;
  3. 输入、产物和验收条件已经足够明确;
  4. 继续拆分不会显著降低失败成本或增加可观测性。

例如:

“搜索竞品 A 的价格”

太粗,因为没有规定来源、价格类型和输出结构。

“打开竞品 A 官方价格页,提取个人版月付价格、年付折算月价、币种和页面更新时间;若页面未公开某字段则写 null,并保存页面标题和抓取时间”

已经接近可执行边界。

但下面的拆分通常过细:

打开浏览器
点击地址栏
输入 URL
按回车
等待页面
读取标题
读取第一段文字

这些是工具执行细节,不应成为任务规划层的业务子任务,否则计划图会被底层操作污染。


4. 依赖:为什么任务图比任务列表更重要

4.1 依赖的含义

任务依赖表示:

某个任务的执行或正确性依赖另一个任务产生的结果。

用有向图表示:

G=(V,E)G = (V, E)

其中:

  • VV 是任务节点集合;
  • EE 是依赖边集合;
  • (Tj,Ti)E(T_j, T_i) \in E,表示 TiT_i 依赖 TjT_j

例如:

T1 → T2

表示 T2 只有在 T1 完成并产出满足要求的结果后,才具备执行条件。

一个有效的计划图通常应满足:

G 是有向无环图(DAG)G \text{ 是有向无环图(DAG)}

如果存在环:

T1 → T2 → T3 → T1

就没有任何任务可以合法地作为起点,调度器会陷入永久等待。环并不一定意味着业务逻辑错误,也可能意味着分解过程把“相互校验”错误地建模为硬依赖。此时应改为:

  • 先执行一个初始版本;
  • 再用校验任务反馈;
  • 必要时创建修订任务。

4.2 数据依赖和控制依赖

两类依赖必须区分。

数据依赖

T2 需要 T1 生成的 competitor_ids

没有数据就无法执行。

控制依赖

只有 T3 验收通过,T4 才能发布

T4 可能技术上可以运行,但业务上不能运行。

这一区分影响重规划和并发:

  • 数据依赖失败时,通常需要修改输入或任务;
  • 控制依赖失败时,可能只需修复质量问题;
  • 允许忽略控制依赖,会产生“任务完成但流程非法”的结果。

4.3 依赖不是顺序列表

下面两种表示的含义不同:

顺序列表:
T1, T2, T3, T4
任务图:
T1 → T2
T1 → T3
T2、T3 → T4

在顺序列表中,T2 和 T3 被迫串行;在任务图中,T2 和 T3 可以并行。

若任务耗时分别为:

T1 = 2 分钟
T2 = 5 分钟
T3 = 4 分钟
T4 = 3 分钟

顺序执行耗时:

2+5+4+3=142 + 5 + 4 + 3 = 14

按依赖并行执行耗时:

2+max(5,4)+3=102 + \max(5,4) + 3 = 10

但只有在以下条件同时成立时才能并发:

  1. 任务之间没有数据依赖;
  2. 工具调用不会产生冲突;
  3. 资源配额允许;
  4. 失败后可以独立重试;
  5. 结果合并规则已定义。

因此,“独立”不能只由模型口头判断,应该由计划中的输入和写集描述支持。


5. 计划数据结构:让计划成为可持久化对象

一个最小可用的计划可以设计为:

from dataclasses import dataclass, field
from enum import Enum
from typing import Any


class TaskStatus(str, Enum):
    PENDING = "pending"
    READY = "ready"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"
    BLOCKED = "blocked"
    CANCELLED = "cancelled"
    NEEDS_REPLAN = "needs_replan"


@dataclass
class Artifact:
    name: str
    value: Any
    version: int = 1


@dataclass
class Task:
    id: str
    title: str
    depends_on: list[str] = field(default_factory=list)
    expected_artifacts: list[str] = field(default_factory=list)
    acceptance: list[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    attempts: int = 0
    artifacts: list[Artifact] = field(default_factory=list)
    error: str | None = None


@dataclass
class Plan:
    goal: str
    version: int
    tasks: dict[str, Task]

这里的 status 不是日志文本,而是调度器的控制输入。artifacts 也不是普通对话消息,而是后继任务可以引用的结构化事实。

例如,T2 的产物可以是:

{
  "task_id": "T2",
  "artifacts": {
    "feature_matrix_A": {
      "storage": "s3://agent-run-123/feature_matrix_A.json",
      "schema_version": "1.0"
    },
    "source_records_A": {
      "storage": "s3://agent-run-123/source_records_A.json",
      "schema_version": "1.0"
    }
  },
  "acceptance": {
    "passed": true,
    "warnings": []
  }
}

大文本不应全部嵌入每次模型上下文。计划状态中只保存引用、摘要、版本和校验信息,执行器按需加载产物。


6. 进度状态:区分“正在做”和“已经成立”

6.1 状态机

推荐至少区分以下状态:

stateDiagram-v2
    [*] --> PENDING
    PENDING --> READY: 依赖满足
    READY --> RUNNING: 调度
    RUNNING --> SUCCEEDED: 产物通过验收
    RUNNING --> FAILED: 可恢复或不可恢复错误
    FAILED --> READY: 允许重试
    FAILED --> NEEDS_REPLAN: 当前方案失效
    RUNNING --> BLOCKED: 等待外部条件
    BLOCKED --> READY: 条件满足
    RUNNING --> CANCELLED: 取消或安全终止
    SUCCEEDED --> [*]

几个状态不能混用:

  • PENDING:任务存在,但依赖尚未满足;
  • READY:依赖和策略检查通过,可以调度;
  • RUNNING:已有执行尝试;
  • SUCCEEDED:产物存在且验收通过;
  • FAILED:本次执行失败,但还未决定下一步;
  • BLOCKED:等待人工审批、外部系统或资源;
  • NEEDS_REPLAN:继续执行原计划已经不合理;
  • CANCELLED:明确终止,不应自动重试。

6.2 状态转换必须有事件依据

错误做法是:

task.status = "succeeded"

正确做法是先记录执行事件:

{
  "event": "task.accepted",
  "task_id": "T2",
  "attempt": 2,
  "artifact_ids": ["feature_matrix_A:v3"],
  "acceptance_result": {
    "passed": True,
    "checks": {
      "has_price": True,
      "has_currency": True,
      "has_source": True
    }
  }
}

然后由状态归约逻辑计算当前状态:

def reduce_task_status(task: Task, acceptance_passed: bool, error: str | None):
    if acceptance_passed:
        return TaskStatus.SUCCEEDED
    if error:
        return TaskStatus.FAILED
    return TaskStatus.RUNNING

这样可以避免“工具返回成功”与“业务任务成功”被混淆。

例如,网页抓取工具返回 HTTP 200,只说明请求成功,不说明价格字段已经抽取成功。任务验收应进一步检查:

请求成功
≠ 页面内容符合预期
≠ 价格字段存在
≠ 价格字段有来源
≠ 整个子任务完成

6.3 进度百分比不能简单按任务数量计算

若有四个任务:

T1:确定竞品集合
T2:收集 A
T3:收集 B
T4:生成报告

完成 T1、T2、T3 后,按数量计算进度为 75%。但如果 T4 是最终交付,系统仍然没有可用报告。

可以采用加权进度:

progress=iwiqiiwiprogress = \frac{\sum_i w_i \cdot q_i}{\sum_i w_i}

其中:

  • wiw_i:任务重要性权重;
  • qiq_i:任务质量进度,范围通常为 [0,1][0,1]

但加权进度仍不能代替完成判定。最终完成必须满足:

done(S)=goal_artifact_exists(S)acceptance_passed(S)no_required_task_unresolved(S)done(S) = goal\_artifact\_exists(S) \land acceptance\_passed(S) \land no\_required\_task\_unresolved(S)

“完成了 90%”只是用户界面上的估计,不是业务事实。


7. 执行器:从任务图中选择下一批任务

执行器的基本流程如下:

flowchart TD
    A[读取持久化计划] --> B[检查任务依赖]
    B --> C{是否存在 READY 任务}
    C -- 否 --> D{是否所有任务完成}
    D -- 是 --> E[结束并返回最终产物]
    D -- 否 --> F{是否存在失败或阻塞}
    F -- 是 --> G[重试、等待或重规划]
    F -- 否 --> H[诊断计划图或状态不一致]
    C -- 是 --> I[按资源和风险选择任务]
    I --> J[并发执行]
    J --> K[保存工具结果和产物]
    K --> L[执行验收]
    L --> M[更新状态与事件]
    M --> B

一个简化的调度器如下:

def is_ready(task: Task, tasks: dict[str, Task]) -> bool:
    return (
        task.status == TaskStatus.PENDING
        and all(tasks[parent].status == TaskStatus.SUCCEEDED
                for parent in task.depends_on)
    )


def find_ready_tasks(plan: Plan) -> list[Task]:
    ready = []
    for task in plan.tasks.values():
        if is_ready(task, plan.tasks):
            task.status = TaskStatus.READY
            ready.append(task)
    return ready


def choose_batch(tasks: list[Task], limit: int = 3) -> list[Task]:
    # 实际系统还应检查工具配额、租约、写冲突和风险等级
    return tasks[:limit]

输入是当前计划和任务状态;输出是本轮可以执行的任务集合。

但这个实现有一个重要前提:所有状态更新必须具备并发控制。否则两个 Worker 可能同时看到同一个 READY 任务,然后重复执行。

常见做法包括:

UPDATE tasks
SET status = 'running',
    worker_id = 'worker-7',
    lease_until = CURRENT_TIMESTAMP + INTERVAL '60 seconds'
WHERE task_id = 'T2'
  AND status = 'ready';

只有受影响行数为 1 的 Worker 获得执行权。受影响行数为 0 表示任务已被其他 Worker 抢占。

这里的租约用于处理 Worker 崩溃:

RUNNING + lease_until 已过期
→ 标记为可恢复
→ 检查工具是否具有幂等键
→ 决定恢复或重试

不能看到租约过期就无条件重试。例如“发送邮件”已经成功,但 Worker 在写回状态前崩溃;无幂等控制的重试可能造成重复发送。


8. 并发执行:依赖独立还不够

假设 T2、T3 都依赖 T1,因此可以并行:

T1 → T2
T1 → T3

仍需检查它们是否操作同一资源。

定义任务的读集和写集:

T2:
  read = {competitor_A_url}
  write = {feature_matrix_A}

T3:
  read = {competitor_B_url}
  write = {feature_matrix_B}

若:

write(T2)write(T3)=write(T_2) \cap write(T_3) = \varnothing

且不存在:

write(T2)read(T3)write(T3)read(T2)write(T_2) \cap read(T_3) \quad\text{或}\quad write(T_3) \cap read(T_2)

则两者通常可以并行。

反例:

T2:更新共享文件 report.md
T3:更新共享文件 report.md

即使两个任务业务上独立,也可能发生写覆盖。可选方案是:

  • 每个任务写独立产物,最后由 T5 合并;
  • 使用版本控制或事务写入;
  • 给共享资源加锁;
  • 让一个任务负责顺序合并。

并发的收益来自减少关键路径长度,代价是增加:

  • 资源竞争;
  • 结果合并;
  • 部分失败;
  • 重试协调;
  • 顺序一致性问题。

因此,并发应由依赖图和资源模型共同决定,而不是简单地“把所有 READY 任务一起跑”。


9. 重规划:什么时候应该改变计划

9.1 重规划不是失败重试

重试的含义是:

计划仍然有效,只是这次执行暂时失败。

例如:

网页请求超时
→ 更换连接或等待后重试

重规划的含义是:

当前计划依赖的假设、输入或路径已经不成立,需要生成新计划或修改现有计划。

例如:

原计划:从官网公开价格页提取价格
实际:官网没有公开价格,必须提交销售询价

继续重试同一个 URL 没有意义,应转为:

T2a:确认官网是否存在公开价格
T2b:寻找官方文档或报价说明
T2c:若仍不可得,标记为“需销售询价”
T2d:修改报告中的价格比较范围

9.2 重规划触发条件

常见触发条件包括:

  1. 关键前置条件失效
用户要求的文件不存在
  1. 外部世界发生变化
API 参数已废弃
库存状态发生变化
  1. 产物无法满足后继任务
收集到的价格没有币种,不能进行比较
  1. 预算、权限或时间约束变化
搜索预算已用尽
工具权限被撤销
  1. 执行路径不可恢复
登录需要人工验证,自动化无法继续
  1. 发现原目标存在歧义
“最低价格”没有说明是月付、年付还是促销价格

9.3 局部重规划和全局重规划

不应每次错误都废弃整个计划。

设任务图被分成:

  • 已完成区域 CC
  • 当前执行区域 XX
  • 未开始区域 UU

如果 T2 失败只影响 T2 和依赖 T2 的任务,那么可以保留:

已完成:T1
受影响:T2、T5、T6、T7
仍有效:T3、T4

只重规划受影响的后继子图:

Affected(T2)={T2}Descendants(T2)Affected(T_2) = \{T_2\} \cup Descendants(T_2)

这比全局重规划更稳定,因为已经验收的事实不需要重新推断。

但以下情况需要考虑全局重规划:

  • 目标约束改变;
  • 关键资源不可用;
  • 计划整体成本超过预算;
  • 已完成产物被证明存在系统性错误;
  • 外部环境变化使多个分支同时失效。

9.4 重规划必须保留历史版本

不要直接覆盖旧计划:

plan.version += 1
plan.tasks = new_tasks

更安全的结构是:

{
  "plan_id": "plan-123",
  "version": 2,
  "parent_version": 1,
  "reason": "官方价格页不存在",
  "preserved_tasks": ["T1", "T3", "T4"],
  "superseded_tasks": ["T2", "T5", "T6"],
  "new_tasks": ["T2a", "T2b", "T2c"]
}

这样可以回答:

  • 为什么计划改变;
  • 哪些任务被保留;
  • 哪些任务被废弃;
  • 新计划依据了什么事实;
  • 是否重复执行了有副作用的动作。

10. 一个完整算例:生成带来源的竞品报告

10.1 初始目标

目标:
比较三个 SaaS 产品的功能、价格和适用场景,
生成一份带来源记录的 Markdown 报告。

约束:
- 优先使用官方来源;
- 价格必须标明币种和计费周期;
- 不确定的信息不能猜测;
- 最终报告必须包含比较表、结论和限制说明。

10.2 初始计划

T1:确认三个竞品及其官方入口
T2:收集产品 A 资料
T3:收集产品 B 资料
T4:收集产品 C 资料
T5:统一字段和单位
T6:生成比较表与分析
T7:检查引用完整性
T8:输出最终报告

依赖关系:

T1 → T2、T3、T4
T2、T3、T4 → T5
T5 → T6
T6 → T7
T7 → T8

10.3 第一次执行

初始状态:

T1 = READY
T2 = PENDING
T3 = PENDING
T4 = PENDING
T5 = PENDING
T6 = PENDING
T7 = PENDING
T8 = PENDING

T1 成功后产出:

{
  "competitors": [
    {"id": "A", "official_url": "..."},
    {"id": "B", "official_url": "..."},
    {"id": "C", "official_url": "..."}
  ]
}

状态变为:

T1 = SUCCEEDED
T2 = READY
T3 = READY
T4 = READY

此时 T2、T3、T4 可以并发执行。

10.4 部分成功

执行结果:

T2 = SUCCEEDED
T3 = SUCCEEDED
T4 = FAILED

T4 失败原因:

产品 C 的官方价格页要求登录,当前 Agent 没有账户权限。

此时不能直接执行 T5,因为 T5 的条件是三个产品资料都满足最低字段要求。

系统可以有三种策略:

策略一:重试

适用于临时错误:

DNS 失败
429 限流
页面暂时超时

策略二:阻塞等待

适用于需要外部输入:

T4 = BLOCKED
原因:等待用户提供登录授权或公开报价文件

策略三:局部重规划

适用于原路径已不可行:

T4a:检查产品 C 的官方文档和帮助中心
T4b:寻找官方销售页面或公开套餐说明
T4c:将不可获得的字段标记为 unknown
T5:在比较字段中支持 unknown

如果业务允许“不完整比较”,则 T5 的验收条件也必须随计划版本明确改变:

原条件:
三个产品都必须有价格。

新条件:
价格缺失必须显式标记 unknown,并在限制说明中解释原因。

这不是偷偷降低质量,而是一次有记录的目标解释变化。

10.5 最终验收

T7 不应只检查 Markdown 文件存在,而应执行结构化检查:

def validate_report(report: dict) -> list[str]:
    errors = []

    if not report.get("comparison_table"):
        errors.append("缺少比较表")

    for row in report.get("comparison_table", []):
        if row.get("price") not in (None, "unknown"):
            if not row.get("currency"):
                errors.append(f"{row['name']} 缺少币种")
            if not row.get("billing_period"):
                errors.append(f"{row['name']} 缺少计费周期")

        if not row.get("source_ids"):
            errors.append(f"{row['name']} 缺少来源")

    if not report.get("limitations"):
        errors.append("缺少限制说明")

    return errors

输入是结构化报告对象,输出是验收错误集合。

若返回空列表:

T7 = SUCCEEDED
T8 = READY

如果报告文件存在但 T7 返回:

产品 C 的价格字段为空,且没有 unknown 标记

则 T7 失败,T8 不得执行。


11. 计划生成:让模型输出“可验证计划”,而不是散文

模型生成计划时,最重要的不是让它写得详细,而是限制输出为可验证结构。

示例 JSON Schema 的核心部分:

{
  "type": "object",
  "required": ["goal", "tasks"],
  "properties": {
    "goal": {"type": "string", "minLength": 1},
    "tasks": {
      "type": "array",
      "items": {
        "type": "object",
        "required": [
          "id",
          "title",
          "depends_on",
          "expected_artifacts",
          "acceptance"
        ],
        "properties": {
          "id": {"type": "string", "pattern": "^T[0-9]+$"},
          "title": {"type": "string"},
          "depends_on": {
            "type": "array",
            "items": {"type": "string"}
          },
          "expected_artifacts": {
            "type": "array",
            "items": {"type": "string"}
          },
          "acceptance": {
            "type": "array",
            "items": {"type": "string"}
          }
        }
      }
    }
  }
}

生成后还需要进行程序验证:

def validate_plan(plan: dict):
    task_ids = {task["id"] for task in plan["tasks"]}

    for task in plan["tasks"]:
        unknown = set(task["depends_on"]) - task_ids
        if unknown:
            raise ValueError(
                f"{task['id']} 引用了不存在的任务: {unknown}"
            )

        if task["id"] in task["depends_on"]:
            raise ValueError(f"{task['id']} 不能依赖自身")

    if has_cycle(plan["tasks"]):
        raise ValueError("计划包含循环依赖")

还应验证:

  • 任务 ID 唯一;
  • 依赖任务存在;
  • 依赖图无环;
  • 最终目标有对应产物;
  • 每个任务都有验收条件;
  • 高风险任务包含审批或权限要求;
  • 任务没有未定义的工具;
  • 计划没有明显重复任务。

模型输出结构化 JSON 只解决“格式正确”,不能证明“计划正确”。计划验证至少包含两层:

结构验证:字段、类型、引用、DAG
语义验证:目标覆盖、产物可用、权限允许、验收可判定

12. 执行失败的分类与处理

失败处理的关键是先判断失败属于哪一层。

12.1 工具层失败

HTTP 429
网络超时
数据库连接断开

通常可以:

  • 按退避策略重试;
  • 更换备用工具;
  • 延长租约;
  • 暂停任务等待资源恢复。

12.2 任务层失败

成功打开页面,但没有找到价格字段

工具没有失败,任务却没有完成。此时可以:

  • 修改查询;
  • 使用另一页面;
  • 请求更适合的提取方式;
  • 将字段标记为未知;
  • 重规划后继任务。

12.3 计划层失败

所有产品资料都依赖一个不可访问的站点

这是计划假设错误,继续执行单个任务没有意义,应修改计划结构。

12.4 目标层失败

用户要求“保证报告绝对准确”

这个目标本身不可验证或不可满足。Agent 应将其转换为可操作约束,例如:

所有外部事实必须带来源;
无法确认的事实必须标记 unknown;
结论必须区分事实和推断。

失败记录应包含:

{
  "task_id": "T4",
  "attempt": 1,
  "failure_class": "task",
  "error_code": "MISSING_REQUIRED_FIELD",
  "message": "官方页面未提供币种",
  "recoverability": "replan",
  "observed_at": "2026-09-01T10:30:00+08:00"
}

recoverability 很重要,因为它直接决定调度器动作:

retry  → 重试
wait   → 阻塞等待
replan → 修改计划
abort  → 终止运行

13. 终止条件:什么时候 Agent 应该停止

Plan-and-Execute 至少需要四类终止条件。

13.1 成功终止

最终产物存在
所有必要任务已通过验收
没有未解决的阻塞

13.2 失败终止

关键任务不可恢复
没有可接受的替代计划
超过重规划次数或预算

13.3 外部等待终止

需要用户授权
需要人工确认
等待异步任务或外部系统

这不是失败。运行应持久化为可恢复状态,而不是返回模糊的“处理中”。

13.4 安全终止

即将执行不可逆操作
工具权限与目标不匹配
检测到越权或不可信输入

高风险操作可以把任务转换为:

READY → BLOCKED

等待审批,而不是让模型自行决定继续。

OpenAI 的 Agents SDK 文档也将 guardrails、人类审核、可恢复运行状态和暂停审批列为 Agent 工作流的重要能力;其 SDK runner 负责工具循环、handoff,并在运行完成或等待审批时停止。(developers.openai.com)


14. 常见误解和反例

误解一:有计划就不需要重新思考

错误。计划只是当前假设下的路线图,不是对未来环境的保证。

初始计划:调用 API 获取订单
现实:API 返回权限错误

执行器必须重新评估,而不是盲目按照旧计划继续。

误解二:任务完成等于工具返回成功

错误。工具返回成功只说明调用层成功,任务还需要业务验收。

搜索成功
≠ 找到了目标信息
≠ 信息可信
≠ 信息满足后继任务要求

误解三:所有任务都应该由模型动态拆分

错误。稳定的业务流程应优先使用代码约束。Anthropic 建议从最简单的实现开始,并提醒框架的抽象层可能遮蔽底层提示和响应,增加调试难度。(anthropic.com)

例如,支付确认、库存扣减和发票开具之间的顺序,不应每次交给模型自由决定。

误解四:把全部历史消息当作进度状态

错误。聊天历史记录了过程,但不等于规范化状态。

聊天记录可能包含:

“应该已经完成了”
“我好像找到了价格”

任务状态需要的是:

{
  "status": "succeeded",
  "artifact_id": "price_table:v2",
  "acceptance_passed": true
}

前者是语言,后者是系统事实。

误解五:重规划越频繁越智能

错误。频繁重规划会导致:

  • 已完成任务被反复执行;
  • 计划版本快速膨胀;
  • 工具调用成本增加;
  • Agent 在局部方案之间振荡;
  • 难以解释最终路径。

可以设置重规划门槛:

只有当失败不可由重试解决,
或后继任务无法满足前置条件,
或成本/权限约束发生变化时,
才触发重规划。

15. 生产系统中的最小闭环

一个可用的 Plan-and-Execute 系统,最少需要以下闭环:

目标解析
  ↓
计划生成
  ↓
计划验证
  ↓
任务调度
  ↓
工具或子 Agent 执行
  ↓
产物写入
  ↓
验收
  ↓
状态更新
  ↓
继续执行 / 重试 / 阻塞 / 重规划 / 终止

可以把一次运行抽象为:

St+1=transition(St,actiont,observationt)S_{t+1} = transition(S_t, action_t, observation_t)

其中:

  • StS_t:第 tt 时刻的计划、任务、产物和运行约束;
  • actiontaction_t:执行器选择的工具调用或状态操作;
  • observationtobservation_t:工具结果、验收结果或外部事件;
  • St+1S_{t+1}:状态转换后的新状态。

模型负责提出候选行动,状态机负责判定行动是否合法,验收器负责判断结果是否成立,重规划器负责在原有策略失效时产生新的任务图。

如果只实现了模型调用,没有持久化计划、依赖、产物和状态,那么得到的通常只是“带长提示词的 ReAct”;如果只实现了固定任务图,又不允许根据观察结果修改计划,那么得到的通常是 workflow,而不是具有适应性的 Agent。

真正的 Plan-and-Execute,核心不在于计划文本有多长,而在于计划是否成为一个能够被验证、调度、部分执行、局部修订并从中断处恢复的系统对象。


系列导航与关联阅读

官方资料

本文依据 Agent、模型、协议与框架官方资料重新梳理;正文、示例与生产清单由 WR BLOG 编写。