AI 工程基础体系 · 第 96/100 篇。内容覆盖机器学习、深度学习与生成式 AI；模型、数据、评测、权限和成本会作为同一生产系统处理。

Agent 规划与任务分解：ReAct、计划执行、反思和终止条件

Agent（智能体）不是“把问题发给大模型，再把回答返回”的固定函数，而是一个在环境中持续执行决策循环的系统。它需要根据目标选择动作，调用工具获得新信息，更新状态，并判断是否继续。

本文把 Agent 规划问题放在一个完整生产系统中讨论：

模型负责理解、推理和生成候选计划；
工具与数据提供可验证的外部事实；
执行器负责真正调用工具、处理超时和重试；
权限系统限制模型能做什么；
评测系统判断任务是否完成；
成本与延迟预算限制循环最多运行多久；
终止条件决定何时成功结束、失败退出或请求人工介入。

ReAct、计划执行和反思并不是互斥的三种 Agent，而是三种不同的控制策略：

ReAct 强调“边想边做、根据观察动态调整”；
计划执行强调“先形成任务结构，再按结构执行”；
反思强调“对中间结果或失败轨迹进行批评、修正和重试”。

实际系统通常会组合它们，但组合前必须明确每种机制解决什么问题，以及它们如何改变状态。

一、先建立 Agent 的形式化模型

1. Agent 不是模型，而是决策闭环

设一个任务为 $G$ ，例如：

找出本季度销售额下降超过 10% 的产品，分析原因，并生成一份带数据来源的报告。

Agent 在时刻 $t$ 的状态可以写成：

s_t = (G, H_t, E_t, P_t, B_t, R_t)

其中：

$G$ ：用户目标和约束；
$H_t$ ：历史轨迹，包括模型消息、工具调用和工具结果；
$E_t$ ：当前已获得的证据；
$P_t$ ：当前计划或未完成任务；
$B_t$ ：剩余预算，例如 token、金额、时间和最大步骤数；
$R_t$ ：权限与风险上下文，例如当前用户、可访问数据和是否允许写操作。

Agent 根据策略 $\pi$ 选择下一个动作：

a_t \sim \pi(a \mid s_t)

动作通常属于以下几类：

生成或修改计划；
调用工具；
向用户提问；
返回中间结果；
声明成功；
声明失败；
请求人工审批。

环境执行动作后返回观察值：

o_{t+1} = \mathcal{E}(s_t, a_t)

例如，SQL 工具返回查询结果，HTTP 工具返回状态码，用户回答一个澄清问题，权限系统拒绝一次写操作。Agent 再用观察值更新状态：

s_{t+1} = \mathcal{U}(s_t, a_t, o_{t+1})

因此，Agent 的核心不是“能不能生成一段合理文字”，而是能否在有限预算内找到一条满足目标、权限和证据要求的动作轨迹：

\tau = (s_0, a_0, o_1, a_1, o_2, \ldots, s_T)

其中 $T$ 是终止时刻。

2. 任务完成需要一个可判断的成功谓词

如果系统没有明确的成功条件，就无法判断“应该继续推理”还是“已经完成”。

定义一个成功谓词：

\text{Success}(s, G) \in \{0,1\}

例如，销售分析任务的成功条件可以是：

找到所有下降超过 10% 的产品；
每个产品都有可复算的销售数据；
每个原因都有对应证据，而不是只有语言模型猜测；
报告包含查询时间和数据范围；
没有访问超出权限的数据。

这比“生成一份看起来完整的报告”严格得多。后者是语言质量标准，不是任务完成标准。

3. 目标函数不仅是正确性

生产 Agent 通常优化的不只是答案正确率，还包括成本、延迟、风险和操作次数：

J(\tau)= \mathbb{E}[\text{quality}(\tau)] -\lambda_c \cdot \text{cost}(\tau) -\lambda_l \cdot \text{latency}(\tau) -\lambda_r \cdot \text{risk}(\tau)

其中：

$\text{quality}$ ：任务完成程度和答案质量；
$\text{cost}$ ：模型调用、工具调用和基础设施成本；
$\text{latency}$ ：端到端响应时间；
$\text{risk}$ ：错误写入、越权、泄露和不可逆操作的风险；
$\lambda_c,\lambda_l,\lambda_r$ ：系统对这些因素的权重。

这解释了一个常见现象：多思考几轮有时提高正确率，但并不意味着应该无限反思。额外循环只有在预期收益高于成本和风险时才值得执行。

二、任务分解：把目标转换成可执行结构

1. 任务分解解决什么问题

大模型直接处理长任务时，常见失败不是“不会生成文字”，而是：

忘记某个子目标；
在工具返回结果后没有更新原假设；
子任务之间的依赖关系错误；
重复执行已经完成的工作；
在证据不足时直接总结；
把“需要用户确认”的动作误当成“可以自动执行”。

任务分解把目标转换成一组具有输入、输出、依赖和完成条件的子任务。一个子任务至少应包含：

q_i = (id_i, input_i, output_i, deps_i, done_i, risk_i)

例如：

q1:
  id: fetch_sales
  input: 本季度和上季度销售数据
  output: 每个产品的销售额表
  deps: []
  done: 查询成功且包含两个完整季度
  risk: 只读

q2:
  id: detect_decline
  input: fetch_sales 的结果
  output: 下降超过 10% 的产品集合
  deps: [fetch_sales]
  done: 每个产品都能复算下降比例
  risk: 只读

q3:
  id: investigate_causes
  input: detect_decline 的产品集合、库存和营销数据
  output: 每个产品的候选原因及证据
  deps: [detect_decline]
  done: 每个原因关联至少一个数据来源
  risk: 只读

q4:
  id: write_report
  input: q1、q2、q3 的结果
  output: 报告草稿
  deps: [fetch_sales, detect_decline, investigate_causes]
  done: 报告包含结论、证据、时间范围和限制
  risk: 创建草稿

这里的 done 很重要。没有完成条件，分解出来的只是任务名称，不是可调度的工作单元。

2. 用依赖图表示任务

如果子任务之间存在依赖关系，可以表示成有向无环图（DAG）：

flowchart TD
    A[用户目标] --> B[获取销售数据]
    B --> C[计算下降产品]
    C --> D[调查库存与营销数据]
    B --> E[检查数据完整性]
    D --> F[生成带证据的报告]
    E --> F

图中的边表示数据或前置条件依赖：

q_i \rightarrow q_j

表示 $q_j$ 不能在 $q_i$ 的输出可用之前可靠执行。

如果两个节点没有路径关系，例如“检查数据完整性”和“准备产品分类映射”互不依赖，它们可以并发执行。但并发并不等于总是更好：

并发可以降低总延迟；
并发可能增加 API 限流概率；
并发写操作可能产生竞态；
多个任务可能重复读取或消耗相同预算；
一个任务失败后，依赖它的任务必须阻塞或改用降级路径。

3. 分解必须保持语义闭包

一个好的分解要满足：子任务的输出足以支持父任务的判断。可以把它理解为：

\bigcup_i output(q_i) \supseteq Evidence(G)

也就是说，所有子任务产出的证据集合，必须覆盖最终目标需要的证据。

反例是：

目标：解释销售下降原因。
分解：查询销售额、生成报告。

这个分解缺少“验证库存、价格、营销活动、产品下架状态”等原因证据。报告可以写出来，但无法证明原因。

另一个反例是把一个不可验证的主观任务当作完成条件：

子任务：判断市场情绪。
完成条件：模型认为结果合理。

更可靠的做法是明确数据来源、时间区间、样本量和置信度。例如：

从指定舆情数据集中提取产品相关文本，按固定分类器版本计算情绪比例，并报告样本量和分类置信度。

4. 动态分解与静态分解

静态分解在开始时一次性生成完整计划。它适合：

依赖关系清晰；
子任务输入稳定；
工具副作用较少；
需要提前审批或估算成本的流程。

动态分解只生成当前阶段需要的子任务，等工具结果回来后再继续拆分。它适合：

问题结构依赖外部发现；
工具结果可能改变后续路径；
搜索或诊断任务；
早期信息不足以生成可靠全局计划。

二者可以组合：先生成粗粒度 DAG，再在执行某个节点时细化其子任务。否则，静态计划容易在第一步工具调用后失效，动态计划则可能失去全局约束。

三、ReAct：在推理与行动之间交替

1. ReAct 的基本结构

ReAct 是 Reasoning and Acting 的缩写，核心是让 Agent 交替产生：

对当前状态的判断；
下一步动作；
环境观察；
基于观察的后续判断。

抽象轨迹如下：

r_t \rightarrow a_t \rightarrow o_{t+1} \rightarrow r_{t+1}

其中 $r_t$ 是推理或决策依据， $a_t$ 是动作， $o_{t+1}$ 是观察。

以“判断某产品销售下降原因”为例：

目标：解释产品 P 的销售额为什么下降超过 10%。

推理：先确认下降是否真实，不能直接依据用户描述。
动作：查询 P 本季度和上季度销售额。

观察：本季度 80 万，上季度 100 万，下降 20%。

推理：下降成立。需要检查库存是否不足。
动作：查询 P 的缺货天数和库存记录。

观察：本季度有 18 天缺货。

推理：缺货可能是原因，但需要检查价格和营销活动，避免把相关性当因果。
动作：查询价格变更与营销曝光。

观察：价格上涨 15%，营销曝光下降 40%。

结论：报告销售下降与缺货、涨价和曝光下降相关；不能仅凭这些数据断言单一因果。

ReAct 的价值在于：每次行动都可以吸收新的事实，而不是在没有外部信息时一次性猜完整答案。

2. ReAct 的状态转移

可以把 ReAct 写成以下循环：

s0 = 初始化目标、权限和预算

while not terminal(s):
    context = 构造当前上下文(s)
    decision = model(context)

    if decision 是工具调用:
        校验参数、权限和风险
        observation = execute_tool(decision)
        s = update(s, decision, observation)

    elif decision 是澄清问题:
        向用户提问
        等待用户输入
        s = update(s, user_answer)

    elif decision 是最终答案:
        s = mark_success 或 mark_failure

关键点是 execute_tool 不能直接相信模型输出。模型只能提出动作，真正的执行器还必须检查：

工具是否在允许列表中；
参数是否符合 schema；
当前用户是否有权限；
是否会产生写入或外部副作用；
是否超过调用预算；
是否需要人工审批。

3. ReAct 的优势与边界

ReAct 适合信息不完整、需要探索的任务，因为它可以根据观察改变路径。它的主要优势是：

能处理工具结果带来的分支；
不需要预先枚举所有子任务；
可以在早期失败后更换策略；
可以把事实获取和决策绑定起来。

但它也有结构性问题：

轨迹容易变长

每一步都可能包含模型调用和工具调用。设每轮平均耗时为 $L$ ，平均成本为 $C$ ，循环 $n$ 轮，则近似有：

\text{latency} \approx nL,\qquad \text{cost} \approx nC

如果没有明确预算，Agent 可能在“再查一个资料”“再验证一次”之间循环。

局部合理不等于全局完成

ReAct 可能在每一步都做了看似合理的动作，却忘记原始目标。例如它查完销售额后直接生成报告，遗漏了“每个原因都要有证据”的要求。

推理内容不能自动视为证据

模型在轨迹中的推理是决策依据，不是外部事实。生产系统应区分：

model_reasoning：模型的假设和计划；
tool_observation：工具返回的事实；
derived_result：由代码或查询计算出的结果；
final_claim：最终对外陈述。

如果把模型的猜测和数据库结果放在同一个“证据”字段中，评测和审计都会失真。

四、计划执行：先规划，再按计划调度

1. 计划执行的两阶段结构

计划执行（Plan-and-Execute）把 Agent 拆成两个主要阶段：

P = \text{Planner}(G, C)

A = \text{Executor}(P, E, B, R)

其中：

$G$ ：目标；
$C$ ：约束；
$P$ ：计划；
$E$ ：执行过程中获得的证据；
$B$ ：预算；
$R$ ：权限上下文。

规划器负责回答：

要完成哪些子任务；
子任务之间有什么依赖；
每个子任务需要什么工具；
什么结果才算完成；
哪些步骤需要用户确认。

执行器负责回答：

当前哪些节点已就绪；
如何调用工具；
失败后如何重试或降级；
哪些结果可以缓存；
何时进入下一个节点。

2. 计划不是自然语言清单，而是可验证对象

下面这种计划不适合直接执行：

1. 查找资料
2. 分析数据
3. 写报告

它缺少输入、输出、工具和终止条件。更可靠的计划应使用结构化数据：

{
  "goal": "找出销售额下降超过10%的产品并解释原因",
  "steps": [
    {
      "id": "fetch_sales",
      "tool": "sales_query",
      "depends_on": [],
      "input": {
        "periods": ["2024-Q3", "2024-Q4"]
      },
      "expected_output": "按产品分组的两个季度销售额",
      "completion": "结果包含两个季度且每行产品可计算下降比例"
    },
    {
      "id": "detect_decline",
      "tool": "local_python",
      "depends_on": ["fetch_sales"],
      "expected_output": "下降超过10%的产品",
      "completion": "每个产品都有可复算的下降比例"
    }
  ]
}

结构化计划便于：

schema 校验；
依赖调度；
断点恢复；
统计每个节点的耗时和成本；
对高风险节点单独审批；
让评测系统判断计划是否遗漏关键步骤。

3. 执行过程中的计划修订

计划执行不是“生成计划后完全不变”。当观察结果违反计划假设时，需要修订计划。

例如原计划假设：

销售表中存在 product_id、quarter、amount 三列。

实际工具返回：

错误：amount 字段不存在，实际字段为 net_revenue。

此时执行器不应让模型继续假设原 schema，而应：

将节点标记为 BLOCKED 或 FAILED；
保存真实错误；
判断是否有自动恢复路径，例如重新读取 schema；
更新后续节点输入；
重新执行受影响节点；
如果错误超出恢复范围，停止并报告。

一个实用规则是：

计划可以被修订，但已完成步骤和外部副作用不能被模型在语言层面“改写”。

如果已经发送邮件，就不能仅通过生成一条新计划把它当作“未发送”。

4. 计划执行的失败模式

计划过度具体

计划提前假定了未知事实，导致工具结果一变就整体失效。

计划过度抽象

计划只有几个宏观动词，执行器无法判断需要调用什么工具或何时完成。

依赖关系错误

把需要 q1 输出的任务标记为无依赖，可能造成空输入执行。

计划与权限不一致

计划包含“删除数据”或“发送通知”，但当前会话只有只读权限。权限检查必须在执行时再次进行，不能只在规划时检查。

五、反思：检查结果，而不是无条件再想一遍

1. 反思的定义

反思（Reflection）是对当前轨迹、计划或中间结果进行结构化评估，然后决定：

接受当前结果；
修正计划；
重做某个子任务；
更换工具或查询；
请求用户澄清；
终止并报告失败。

它不是简单地让模型多生成一段“我再检查一下”。反思必须产生可执行的状态变化。

可以定义反思函数：

c_t = \text{Critic}(s_t, G)

其中 $c_t$ 至少应包含：

判断：是否满足成功条件
缺口：还缺少哪些证据或步骤
错误：哪些结论与观察冲突
修复动作：下一步具体执行什么
严重性：是否必须阻断最终输出

2. 反思的三种常见位置

任务级反思

检查最终结果是否满足用户目标。

例如：

目标要求三个产品都有原因分析。
当前报告只包含两个产品。
动作：阻止提交，返回 detect_decline 的结果，补做第三个产品。

步骤级反思

检查某个工具结果是否有效。

例如：

查询返回 0 行。
可能原因：确实没有数据，也可能是日期过滤器错误。
动作：先查询可用日期范围，不得直接得出“没有销售”。

轨迹级反思

检查 Agent 是否陷入重复、偏题或循环。

例如：

最近四轮都在调用同一个搜索工具，查询词只改变了标点。
动作：停止重复搜索，改为请求用户指定来源或进入失败终止。

3. 反思必须连接到修复动作

无效的反思通常是：

结果可能不够准确，我会更加仔细。

它没有改变状态，也没有增加证据。有效的反思应当类似：

问题：结论把相关性写成了因果性。
缺少：没有控制价格变化和营销曝光的影响。
修复：将结论改为“相关因素”，并查询价格变更记录；如果没有因果实验数据，不得使用“导致”。

反思后的状态变化是：

s_{t+1} = \text{update}(s_t,\text{critic},\text{repair\_action})

如果没有新的观察、计划变化或输出修订，反思循环通常只是额外成本。

4. 反思的反例：自我确认循环

设 Agent 生成结论：

库存不足是销售下降的原因。

然后让同一个模型检查：

这个结论合理吗？

模型可能回答：

是的，库存不足通常会导致销售下降。

这并没有新增证据，只是同一先验对自己的复述。更可靠的反思应尽量依赖：

独立计算；
schema 和约束校验；
第二个模型或不同提示；
不同数据源；
人工审批；
业务规则；
反事实或对照数据。

这里的“独立”不是绝对保证。使用两个相同模型并不等于错误独立，只能作为经验上的风险降低。

六、四种机制如何组合

一个常见的组合流程如下：

flowchart LR
    A[用户目标与约束] --> B[初始规划]
    B --> C[选择可执行节点]
    C --> D[调用工具或模型]
    D --> E[记录观察与证据]
    E --> F{步骤检查}
    F -- 不通过 --> G[修复或重试]
    G --> C
    F -- 通过 --> H{全局成功条件满足?}
    H -- 否 --> C
    H -- 是 --> I[最终反思]
    I --> J{风险和证据足够?}
    J -- 否 --> G
    J -- 是 --> K[终止并返回]

其中：

计划执行决定节点和依赖；
ReAct用于单个节点内部的动态探索；
反思检查工具结果、局部输出和最终报告；
终止条件控制整个状态机退出。

例如，“调查销售下降原因”这个计划节点本身可以使用 ReAct：

计划节点：调查原因
  ├─ 查询库存
  ├─ 如果库存正常，查询价格变化
  ├─ 如果价格未变，查询营销曝光
  └─ 根据证据生成候选原因

外层仍然按照 DAG 调度，内层则根据观察动态选择工具。

七、终止条件：Agent 必须知道何时停止

1. 终止不是“模型输出 final”

把模型输出 final 作为唯一终止信号是不安全的。模型可能：

漏掉一个子任务；
工具失败后假装成功；
在证据不完整时过早结束；
因提示词歧义无限输出；
把权限拒绝解释成没有数据。

终止应由外部控制器依据状态判断，而不是完全交给模型。

定义终止函数：

\text{Terminal}(s) = \text{Success}(s) \lor \text{Failure}(s) \lor \text{BudgetExhausted}(s) \lor \text{HumanRequired}(s)

2. 成功终止

成功终止必须同时满足目标谓词和必要约束：

\text{Success}(s,G)= \text{GoalSatisfied}(s,G) \land \text{EvidenceSufficient}(s) \land \text{PermissionValid}(s) \land \text{OutputValid}(s)

例如：

所有必需 DAG 节点完成；
工具结果没有未处理错误；
最终字段通过 schema 校验；
证据引用都存在；
没有违反数据权限；
高风险动作已经审批。

3. 失败终止

失败不是“什么都没返回”，而是一个带原因的终态。常见失败原因包括：

必要工具不可用；
权限不足；
数据不存在或不完整；
输入歧义且用户没有回答；
重试次数耗尽；
发现不可接受的安全风险；
依赖节点永久失败。

失败结果应区分：

FAILED_DATA:
  数据源不包含目标时间范围。

FAILED_PERMISSION:
  当前身份无权访问成本明细。

FAILED_TOOL:
  外部 API 连续超时。

FAILED_INPUT:
  用户未指定“本季度”的具体日期范围。

ABORTED_BUDGET:
  达到最大步骤或金额预算。

这些状态的恢复方法不同，不能统一显示为“Agent 执行失败”。

4. 预算终止

至少需要设置以下预算：

最大模型调用次数；
最大工具调用次数；
最大总步骤数；
最大墙钟时间；
最大 token 或金额；
单个工具的超时和重试次数；
单个节点的最大循环次数。

预算不是只为节省钱，也是在防止：

工具故障导致无限重试；
反思失控；
恶意输入诱导大量调用；
搜索结果不断扩展；
并发任务耗尽连接池。

5. 循环和停滞检测

如果最近 $k$ 个状态没有实质变化，可以判定停滞：

\text{stalled}(s_{t-k:t}) = \mathbf{1} \left[ \text{progress}(s_{t-k}) = \text{progress}(s_t) \right]

“实质变化”不能只比较字符串。应比较：

已完成节点集合；
新增证据数量；
计划版本；
未满足条件数量；
工具参数和返回结果；
错误类型是否改变。

例如，连续三次调用同一个工具、使用等价参数、没有新增证据，就应停止或切换策略。

6. 人工介入终止

以下动作通常不应只由模型决定：

删除或覆盖数据；
向外部客户发送消息；
执行资金、权限或生产配置变更；
发布未经审核的内容；
处理高敏感个人数据。

此时终止状态可以是：

WAITING_FOR_APPROVAL

它不是成功，也不是失败。系统应保存可恢复的状态，等待审批后从原节点继续，而不是重新从头生成计划。

八、一个可运行的最小计划执行器

下面的 Python 示例不依赖外部库，用一个确定性的销售分析任务展示：

结构化计划；
DAG 依赖；
工具执行；
观察记录；
步骤完成条件；
全局成功终止；
工具错误和预算终止。

它不是完整的大模型 Agent，而是 Agent 控制器的最小骨架。真实系统可以把 make_plan 替换成模型生成并经过 schema 校验的计划，把 TOOLS 替换成数据库、HTTP 或 MCP 工具。

from dataclasses import dataclass, field
from typing import Any, Callable


@dataclass
class Step:
    step_id: str
    depends_on: list[str]
    tool: str
    args: dict[str, Any]
    status: str = "PENDING"
    output: Any = None
    error: str | None = None


@dataclass
class State:
    goal: str
    steps: dict[str, Step]
    evidence: list[dict[str, Any]] = field(default_factory=list)
    calls: int = 0
    max_calls: int = 10


def sales_query(periods: list[str]) -> list[dict[str, Any]]:
    """模拟只读数据工具。"""
    if set(periods) != {"2024-Q3", "2024-Q4"}:
        raise ValueError("只提供 2024-Q3 和 2024-Q4 的演示数据")

    return [
        {"product": "A", "period": "2024-Q3", "sales": 100},
        {"product": "A", "period": "2024-Q4", "sales": 80},
        {"product": "B", "period": "2024-Q3", "sales": 100},
        {"product": "B", "period": "2024-Q4", "sales": 95},
        {"product": "C", "period": "2024-Q3", "sales": 50},
        {"product": "C", "period": "2024-Q4", "sales": 60},
    ]


def detect_decline(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """由确定性代码计算下降比例，避免让模型直接计算关键指标。"""
    grouped: dict[str, dict[str, float]] = {}

    for row in rows:
        grouped.setdefault(row["product"], {})[row["period"]] = row["sales"]

    result = []
    for product, values in grouped.items():
        if "2024-Q3" not in values or "2024-Q4" not in values:
            raise ValueError(f"{product} 缺少完整季度数据")

        previous = values["2024-Q3"]
        current = values["2024-Q4"]
        decline = (previous - current) / previous

        if decline > 0.10:
            result.append({
                "product": product,
                "decline_rate": decline,
            })

    return result


TOOLS: dict[str, Callable[..., Any]] = {
    "sales_query": sales_query,
    "detect_decline": detect_decline,
}


def dependencies_done(step: Step, steps: dict[str, Step]) -> bool:
    return all(steps[d].status == "DONE" for d in step.depends_on)


def run_agent(state: State) -> dict[str, Any]:
    while True:
        if state.calls >= state.max_calls:
            return {
                "status": "ABORTED_BUDGET",
                "reason": f"工具调用次数达到上限 {state.max_calls}",
                "evidence": state.evidence,
            }

        pending = [
            step for step in state.steps.values()
            if step.status == "PENDING"
        ]

        # 没有未完成步骤时，检查全局成功条件
        if not pending:
            declined = state.steps["detect_decline"].output
            if declined is None:
                return {
                    "status": "FAILED",
                    "reason": "缺少下降产品结果",
                    "evidence": state.evidence,
                }

            return {
                "status": "SUCCEEDED",
                "declined_products": declined,
                "evidence": state.evidence,
            }

        # 选择一个依赖已完成的节点
        ready = [
            step for step in pending
            if dependencies_done(step, state.steps)
        ]

        if not ready:
            return {
                "status": "FAILED_PLAN",
                "reason": "存在循环依赖或不可满足的依赖",
                "evidence": state.evidence,
            }

        step = ready[0]
        step.status = "RUNNING"
        state.calls += 1

        # 将上游输出注入当前步骤
        args = dict(step.args)
        if step.step_id == "detect_decline":
            args["rows"] = state.steps["fetch_sales"].output

        try:
            tool = TOOLS[step.tool]
            output = tool(**args)
            step.output = output
            step.status = "DONE"

            state.evidence.append({
                "step_id": step.step_id,
                "tool": step.tool,
                "output": output,
            })

        except Exception as exc:
            step.status = "FAILED"
            step.error = str(exc)
            return {
                "status": "FAILED_TOOL",
                "step_id": step.step_id,
                "reason": step.error,
                "evidence": state.evidence,
            }


def make_plan() -> dict[str, Step]:
    return {
        "fetch_sales": Step(
            step_id="fetch_sales",
            depends_on=[],
            tool="sales_query",
            args={"periods": ["2024-Q3", "2024-Q4"]},
        ),
        "detect_decline": Step(
            step_id="detect_decline",
            depends_on=["fetch_sales"],
            tool="detect_decline",
            args={},
        ),
    }


if __name__ == "__main__":
    state = State(
        goal="找出销售额下降超过10%的产品",
        steps=make_plan(),
        max_calls=10,
    )

    result = run_agent(state)
    print(result)

在正常执行时，状态变化如下：

初始：
  fetch_sales=PENDING
  detect_decline=PENDING
  calls=0

第一次循环：
  fetch_sales 无依赖，可以执行
  调用 sales_query
  fetch_sales=DONE
  calls=1

第二次循环：
  detect_decline 的依赖已完成
  调用 detect_decline
  detect_decline=DONE
  calls=2

第三次循环：
  没有 PENDING 步骤
  检查全局成功条件
  返回 SUCCEEDED

预期关键输出为：

declined_products:
  [{"product": "A", "decline_rate": 0.2}]

这里选择用确定性函数计算下降比例，而不是让模型从自然语言中计算，原因是计算规则可以测试、复算和审计。模型可以负责解释结果，但关键数值应尽量交给可验证程序。

这个示例仍然缺少生产系统中的重试、并发、持久化和审批。它也没有动态反思，因为演示数据不会产生异常。若加入反思，应在每个节点完成后检查：

def reflect_after_step(step: Step) -> dict[str, Any]:
    if step.step_id == "fetch_sales":
        rows = step.output
        products = {r["product"] for r in rows}
        periods = {r["period"] for r in rows}
        if periods != {"2024-Q3", "2024-Q4"}:
            return {
                "ok": False,
                "repair": "重新查询，确保两个季度都存在",
            }
        if not products:
            return {
                "ok": False,
                "repair": "检查过滤条件或请求用户确认产品范围",
            }

    return {"ok": True}

反思结果只有在改变状态时才有意义，例如重新执行节点、修改参数、阻止下游节点或进入人工审批。

九、并发、重试与故障路径

1. 哪些任务可以并发

在 DAG 中，所有依赖已满足的节点都可以进入就绪集合：

Ready(s)=\{q_i \mid status(q_i)=PENDING \land \forall d\in deps_i,\ status(q_d)=DONE\}

例如，销售数据获取完成后：

查询库存；
查询价格；
查询营销曝光；

如果它们彼此只读且没有共享写状态，可以并发执行。

但调度器仍需考虑：

每个工具的并发上限；
API 速率限制；
同一资源的锁；
每个节点的独立超时；
失败是否阻塞整个父任务；
结果是否需要保持顺序。

2. 重试必须区分错误类型

不是所有错误都应该重试：

错误类型	是否通常重试	原因
网络超时	可以	可能是瞬时故障
HTTP 429 限流	可以退避	应遵守服务端限制
HTTP 401/403	通常不重试	身份或权限没有改变
参数 schema 错误	不应盲目重试	需要修正参数
查询结果为空	不一定	可能是真实空结果
模型输出格式错误	可有限重试	需要重新约束输出
写操作超时	高风险	可能已成功，不能简单重复

对于写操作，必须考虑幂等性。若一个“创建订单”请求超时，客户端不知道服务端是否已经创建成功，直接重试可能产生两个订单。更安全的方式是使用幂等键，或先查询操作状态。

3. 失败应沿依赖图传播

如果 fetch_sales 失败，则依赖它的 detect_decline 不能继续执行。状态可以这样传播：

fetch_sales: FAILED_TOOL
detect_decline: BLOCKED
write_report: BLOCKED

BLOCKED 与 FAILED 不同：

FAILED 表示该节点已经执行但失败；
BLOCKED 表示它没有执行，因为前置条件不可用。

这一区分有助于恢复：修复数据查询后，可以重新运行 fetch_sales，然后解除后续节点的阻塞。

十、与 MCP 和 Agent 框架的关系

1. MCP 解决工具和上下文互操作，不负责规划

Model Context Protocol（MCP）是一种让 AI 应用以统一方式连接外部数据源、工具和提示模板的协议。按照 MCP 的架构，通常存在：

Host：承载 AI 应用的宿主；
Client：Host 中与某个 MCP Server 建立连接的协议客户端；
Server：提供工具、资源或提示的服务端。

MCP 关注的是协议层面的发现、调用、消息交互和生命周期。常见能力包括：

发现服务器提供的工具；
获取工具名称、描述和输入 schema；
调用工具并接收结构化结果；
访问资源或使用提示模板；
处理初始化、能力协商和连接生命周期。

MCP 本身不规定：

Agent 必须使用 ReAct；
Agent 必须先生成完整计划；
反思应运行几次；
什么条件表示业务任务成功；
业务权限是否足够；
工具返回的数据是否真实；
最终答案是否正确。

因此，MCP 工具可以作为 ReAct 的动作，也可以作为计划执行器的节点，但规划器、调度器、终止控制器和业务评测器仍属于 Agent 应用。

2. MCP 工具调用的正确控制边界

一个安全的数据流应当是：

sequenceDiagram
    participant U as 用户
    participant H as Agent Host
    participant M as 模型
    participant P as 策略与权限层
    participant C as MCP Client
    participant S as MCP Server
    participant D as 外部数据系统

    U->>H: 提交目标
    H->>M: 目标、计划状态、可用工具
    M->>H: 提议工具调用
    H->>P: 校验工具、参数、权限、风险
    P-->>H: 允许或拒绝
    H->>C: 发送 MCP 调用
    C->>S: 工具调用请求
    S->>D: 访问数据或执行操作
    D-->>S: 结果
    S-->>C: 结构化工具结果
    C-->>H: 观察值
    H->>H: 更新状态、反思、检查终止条件

关键边界是：模型提议调用，策略层批准调用，MCP Client 负责协议通信，MCP Server 执行工具。

不能因为 MCP 工具声明了输入 schema，就认为调用天然安全。schema 只能约束数据形状，不能替代：

身份认证；
资源级授权；
参数范围检查；
敏感数据脱敏；
写操作审批；
速率限制；
审计日志。

MCP 规范保证的是协议参与者之间的交互约定；具体工具是否正确、安全、幂等，取决于工具实现和部署策略。

3. Agent 框架通常提供什么

Agent 框架一般会提供部分或全部能力：

Agent 和工具定义；
模型调用循环；
工具参数解析；
handoff，即把任务转交给其他 Agent；
guardrail，即输入、输出或工具调用检查；
tracing，即记录轨迹；
并发和会话管理。

OpenAI Agents Guide 介绍的 Agent 应用通常也围绕这些概念组织，但具体 SDK、接口名称和参数会随版本变化。工程实现不能只依据概念名称猜测 API；应以当前 SDK 的官方文档、类型定义和变更记录为准。

无论使用哪种框架，都应确认以下生命周期是否真实存在：

创建会话
  -> 获取可用工具
  -> 模型产生动作
  -> 框架或应用执行工具
  -> 写入观察结果
  -> 继续循环或终止
  -> 持久化轨迹与评测结果

如果框架只负责模型调用，而工具执行、审批和终止仍由应用实现，就不要把这些能力误认为框架自动保证。

十一、上下文管理：规划质量受状态质量限制

1. 完整轨迹不等于有效上下文

把所有历史消息原样放进上下文会导致：

token 成本增长；
旧计划和新计划同时存在；
工具错误被大量重复；
模型难以区分事实和假设；
敏感数据在不必要的步骤中持续传播。

更可靠的上下文应分层：

目标层：
  用户目标、硬约束、输出格式、截止时间

计划层：
  当前计划版本、节点状态、依赖关系、待办事项

证据层：
  工具结果、来源、时间、查询参数、数据质量

决策层：
  当前假设、待验证问题、失败原因、反思结果

安全层：
  用户身份、权限、审批状态、预算

压缩层：
  对已完成轨迹的结构化摘要

2. 事实、推断和未验证假设必须分开

例如：

{
  "fact": {
    "source": "sales_query",
    "product": "A",
    "q3_sales": 100,
    "q4_sales": 80
  },
  "derived": {
    "decline_rate": 0.2,
    "method": "(100 - 80) / 100"
  },
  "hypothesis": {
    "text": "缺货可能是下降原因",
    "status": "UNVERIFIED"
  }
}

如果把 hypothesis 直接复制到最终报告，就会把推测伪装成事实。反思器应检查所有最终结论是否有对应的 fact 或可复算的 derived 结果。

十二、常见误解与诊断方法

误解一：思考越多，结果越可靠

增加循环可能增加信息，也可能只增加重复。诊断时应查看每轮是否新增：

工具调用结果；
有效证据；
已完成节点；
计划修订；
解决的错误。

如果连续多轮只生成不同措辞，没有新增观察，则问题不是模型“思考不够”，而是缺少停滞检测或搜索策略。

误解二：有计划就不会遗漏步骤

计划只是候选结构。要防止遗漏，必须把计划与成功谓词绑定。例如最终报告要求“每个下降产品都有证据”，就需要程序检查产品集合和证据集合是否一致。

可以定义覆盖率：

\text{coverage} = \frac{\text{已有证据覆盖的必需项目数}} {\text{必需项目总数}}

当 coverage < 1 时，不能进入成功终止，即使文本看起来完整。

误解三：反思可以替代评测

反思器也是模型或规则系统，也可能出错。离线评测仍然需要真实任务集，检查：

成功率；
子任务遗漏率；
工具参数错误率；
无效调用率；
错误终止率；
平均和尾部成本；
越权尝试拦截率；
需要人工介入的比例。

在线系统还应记录完整轨迹，但日志必须脱敏，并避免把访问令牌、个人数据和模型上下文无约束地写入日志。

误解四：工具返回空结果就是事实

空结果可能表示：

数据确实不存在；
查询条件写错；
时区或日期范围错误；
权限过滤后为空；
服务端返回了不完整结果。

因此，空结果通常是一个需要解释的观察，而不是直接的业务结论。可以先执行 schema 检查、可用日期范围查询或权限诊断。

误解五：模型说“完成”就代表完成

“完成”是自然语言事件，“成功”是系统状态。只有当外部检查器确认成功谓词成立时，状态才应转换为 SUCCEEDED。

十三、如何评测规划与任务分解

单独评测最终答案不够，因为两个 Agent 可能得到相同答案，但一个调用了越权工具，另一个使用了合法证据。

评测至少应覆盖四个层面。

1. 计划质量

检查：

是否覆盖所有必需子目标；
依赖关系是否正确；
是否存在循环；
输入输出是否符合工具 schema；
是否为高风险动作设置审批；
每个节点是否有可验证完成条件。

2. 执行质量

检查：

工具参数正确率；
对错误码的处理是否正确；
重试是否遵守上限；
是否能从中间状态恢复；
是否发生重复写入；
并发是否超过服务约束。

3. 结果质量

检查：

数值是否可复算；
结论是否得到证据支持；
是否区分相关性和因果性；
是否完整回答用户目标；
是否明确数据缺口和不确定性。

4. 控制质量

检查：

是否及时终止；
是否在预算耗尽后停止；
是否检测循环；
是否在权限拒绝后避免绕过；
是否在高风险动作前请求审批；
失败状态是否准确分类。

一个简单的轨迹评测记录可以是：

{
  "task_id": "sales_root_cause_001",
  "success": true,
  "steps_planned": 4,
  "steps_completed": 4,
  "tool_calls": 6,
  "retries": 1,
  "budget_exhausted": false,
  "evidence_coverage": 1.0,
  "unauthorized_attempts": 0,
  "human_approval_required": false
}

这里的 success 不能仅由模型自评生成，而应由任务评测器根据预先定义的检查器计算。

十四、选择哪种策略

可以按照任务的不确定性和风险做选择：

任务特征	更适合的策略
单步查询、结果明确	简单工具调用
需要根据观察选择下一步	ReAct
依赖关系稳定、步骤较多	计划执行
最终结果容易遗漏或产生事实错误	计划执行 + 反思
高风险写操作	计划执行 + 权限检查 + 人工审批
外部环境变化频繁	粗粒度计划 + 动态 ReAct
工具故障代价高	显式状态机 + 有限重试 + 失败终止

一个实用的架构不是让大模型控制所有事情，而是把不同职责分开：

模型：
  生成计划、提出工具调用、解释候选结果

确定性代码：
  schema 校验、计算、依赖调度、预算、终止判断

工具：
  访问数据库、文件、HTTP 服务或业务系统

策略层：
  权限、审批、敏感数据和风险控制

评测层：
  完成条件、证据覆盖和质量检查

这样做的因果关系是：模型适合处理语义不确定性，代码适合处理必须稳定一致的控制逻辑。把终止、权限和关键计算全部交给模型，会让最重要的系统性质变成不可预测的文本行为。

结语

ReAct、计划执行和反思分别处理三个问题：

ReAct 解决“根据环境观察动态决定下一步”；
计划执行解决“如何组织多步骤依赖并进行调度”；
反思解决“如何发现结果缺口并触发修复”。

它们最终都要落到同一个状态机上：目标、计划、证据、权限、预算和终止状态必须显式存在。没有成功谓词，Agent 不知道何时完成；没有失败状态，它只能用自然语言掩盖错误；没有外部权限和预算控制，增加推理轮次反而可能扩大风险。

在生产系统中，最可靠的 Agent 通常不是“最会思考”的 Agent，而是能够把模型的候选决策放入可验证执行闭环，并在证据不足、权限不允许、预算耗尽或状态停滞时明确停止的 Agent。

系列导航与关联阅读

系列入口：AI 工程完整学习路线：从机器学习与 Transformer 到 RAG、Agent 和生产治理
上一篇：Graph RAG：实体关系、图构建、社区摘要、检索与适用边界
下一篇：Agent 工具沙箱：文件、命令、网络、凭证、审批与审计

官方资料

本文依据研究论文、标准组织与主流框架官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。