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

Agent 任务分解:目标、子任务、前置条件、产物和验收

Agent 的“任务分解”不是把一句用户需求改写成若干条待办事项,而是把一个开放目标转换成可以执行、可以暂停、可以恢复、可以重规划、可以验证的工作结构。

一个合格的任务分解至少要回答五个问题:

  1. 目标是什么:最终要改变什么状态,或交付什么结果?
  2. 子任务是什么:为了达到目标,需要完成哪些相对独立的工作?
  3. 前置条件是什么:某个子任务开始前,必须已经具备哪些输入、权限、事实或环境状态?
  4. 产物是什么:子任务完成后,系统留下什么可传递、可验证的结果?
  5. 验收是什么:凭什么判定子任务或整体目标已经完成?

如果缺少其中任意一项,Agent 往往只能“看起来在工作”:它可以生成计划,却无法判断何时开始;可以调用工具,却无法判断结果是否足够;可以输出总结,却无法证明目标确实达成。

Anthropic 将 Agentic System 区分为两类:workflow 由代码预先编排路径,而 agent 由语言模型动态决定过程和工具使用。OpenAI 对 Agent 的描述也强调了规划、工具调用、专家协作以及维持足够状态来完成多步骤工作。(anthropic.com)

本文采用一个工程化定义:

任务分解是将目标表示为带有依赖关系、输入输出契约、状态和验收条件的任务图,并由执行器根据环境反馈推进或重构该任务图。


一、先区分目标、计划、任务和动作

任务分解失败,通常不是模型不会列清单,而是系统没有区分不同抽象层级。

1. 目标:期望达到的状态

目标描述的是最终状态,而不是执行动作。

例如:

为杭州团队准备一份 2026 年第三季度云成本分析报告,并给出可执行的降本建议。

这里的目标状态至少包括:

  • 分析对象是“杭州团队”;
  • 时间范围是“2026 年第三季度”;
  • 报告包含成本事实和趋势;
  • 报告包含建议;
  • 建议可以被审阅或执行。

“查询账单”“统计服务用量”“生成图表”都不是目标,而是可能的实现步骤。

可以把目标表示为一个期望状态:

G=(Starget,C,R)G = (S_{target}, C, R)

其中:

  • StargetS_{target}:期望达到的业务状态;
  • CC:约束,例如时间范围、权限、预算、格式;
  • RR:验收规则。

例如:

S_target:
  存在一份可供财务和技术负责人审阅的成本分析报告

C:
  仅使用 2026-07-01 至 2026-09-30 的账单数据
  只分析杭州团队所属账户
  不修改生产资源
  报告必须包含数据来源和异常说明

R:
  总成本可由明细汇总复算
  关键数字有来源
  建议包含预计影响、风险和实施前提

目标的核心特征是:完成与否不依赖 Agent 自己的主观描述,而依赖外部状态或可检查产物。

2. 计划:达到目标的假设路径

计划是 Agent 在当前信息下对执行路径的预测:

获取账单数据
→ 过滤团队和时间范围
→ 按服务聚合
→ 识别异常
→ 形成降本建议
→ 生成报告
→ 验收报告

计划不是事实,也不是永久承诺。它建立在若干假设上:

  • 账单 API 可访问;
  • 团队标签完整;
  • 成本明细可以按服务聚合;
  • 异常可以通过历史数据识别;
  • 建议不需要额外审批。

只要这些假设被工具结果推翻,计划就必须调整。

3. 任务:可以独立管理的工作单元

任务是计划中的一个可执行节点。它应当拥有明确的:

  • 输入;
  • 前置条件;
  • 执行动作;
  • 产物;
  • 验收条件;
  • 失败处理;
  • 重试或重规划策略。

“分析成本”通常太大,不能直接作为一个可靠任务。更合适的分解是:

T1:获取账单明细
T2:验证账单完整性
T3:按团队和服务聚合成本
T4:识别异常成本项
T5:生成降本候选方案
T6:验证候选方案的依据和风险
T7:生成报告
T8:验收报告

4. 动作:一次工具调用或一次确定性计算

动作是任务内部的执行步骤,例如:

  • 调用 get_billing_records
  • 读取对象存储文件;
  • 执行 SQL;
  • 运行单元测试;
  • 写入报告文件;
  • 请求人工审批。

任务不一定对应一次模型调用。一个任务可能包含多轮:

读取数据
→ 发现字段缺失
→ 查询字段定义
→ 再次读取
→ 校验通过

因此,任务是状态管理单位,动作是执行单位。把每个工具调用都当成一个任务,会造成任务图过细;把多个不可独立验证的工作合成一个任务,则会造成失败定位困难。


二、任务分解的核心数据结构

生产系统不应只保存一段自然语言计划,而应保存结构化任务图。

一个任务节点可以表示为:

{
  "id": "T4",
  "title": "识别异常成本项",
  "objective": "找出需要进一步调查的高成本或异常增长项",
  "inputs": [
    "artifact:aggregated_costs",
    "artifact:validated_billing"
  ],
  "preconditions": [
    "billing_period == '2026-Q3'",
    "aggregation_check == 'passed'",
    "currency_normalized == true"
  ],
  "actions": [
    "compare_with_previous_period",
    "detect_outliers",
    "classify_explanations"
  ],
  "outputs": [
    "artifact:cost_anomalies"
  ],
  "acceptance": [
    "每个异常项都有比较基线",
    "每个异常项都有数值变化",
    "无法解释的异常必须标记为 unknown"
  ],
  "failure_policy": {
    "transient": "retry",
    "missing_data": "replan",
    "permission_denied": "request_approval",
    "invalid_output": "repair"
  },
  "status": "blocked"
}

这里有一个重要区别:

  • objective 说明这个任务要解决什么问题;
  • actions 说明可能执行哪些动作;
  • outputs 说明执行后留下什么;
  • acceptance 说明什么叫完成;
  • failure_policy 说明失败后改变什么。

如果只有 titleactions,它只是待办事项,不是可恢复任务。

任务图而非任务列表

任务之间通常存在依赖关系。用有向图表示:

D=(V,E)D = (V, E)

其中:

  • VV:任务节点集合;
  • EE:依赖边集合;
  • (Ti,Tj)E(T_i, T_j) \in E,表示 TjT_j 需要 TiT_i 的结果或状态。

例如:

flowchart TD
    T1[获取账单明细] --> T2[验证账单完整性]
    T2 --> T3[按团队和服务聚合]
    T3 --> T4[识别异常成本项]
    T3 --> T5[生成降本候选方案]
    T4 --> T6[验证建议依据]
    T5 --> T6
    T6 --> T7[生成报告]
    T7 --> T8[验收报告]

图中 T4T5 都依赖 T3,但彼此不依赖,因此在资源允许时可以并发执行。T6 必须等待两者完成,因为它需要同时检查异常和建议。

任务图必须是有向无环图,还是允许循环,取决于系统设计:

  • 固定流程通常使用 DAG;
  • 需要评审、修复、重试的 Agent 流程,本质上是带循环的状态机;
  • 循环不应直接修改原计划,而应产生新的计划版本或新的任务实例。

三、目标必须可验收:从愿望到判定函数

“写一份高质量报告”不是可执行目标,因为“高质量”没有判定方法。

目标验收可以形式化为一个判定函数:

Accept(G,S,A){true,false,unknown}Accept(G, S, A) \rightarrow \{true, false, unknown\}

其中:

  • GG:目标和验收规则;
  • SS:执行后的环境状态;
  • AA:产生的产物;
  • true:满足验收;
  • false:明确不满足;
  • unknown:证据不足,不能安全判定。

引入 unknown 很重要。没有证据不等于失败,也不等于成功。

例如,报告中的总成本是 100 万元,但没有保存账单明细和查询条件:

  • 不能直接判定数字错误;
  • 也不能因为报告格式正确就判定成功;
  • 正确状态应是 unknown,触发补证据或人工审查。

验收条件的四种类型

1. 存在性条件

判断某个产物是否存在:

报告文件存在
查询结果已保存
代码补丁已提交
审批记录已生成

存在性条件只能证明“有东西”,不能证明“东西正确”。

2. 结构条件

判断产物是否满足格式和字段要求:

报告包含摘要、数据范围、明细、异常和建议五个章节
每条建议包含依据、预期收益和风险
JSON 符合指定结构
SQL 查询结果包含 team、service、cost 三列

结构条件适合用程序校验。

3. 语义条件

判断内容是否满足业务含义:

每个异常项都有可追溯的比较基线
建议不能把一次性成本误判为持续性浪费
结论不得超出数据支持范围

语义条件可以由规则、模型评审或人工共同判断,但不能只依赖生成 Agent 自评。

4. 状态条件

判断外部系统是否真正发生变化:

工单状态变为 resolved
代码测试全部通过
资源标签已更新
退款审批已完成

状态条件必须从真实系统查询,而不是相信工具调用返回的自然语言描述。

验收条件必须尽量满足三个性质

设验收条件为 CC,理想情况下应满足:

  1. 可观察:能从日志、数据库、文件或工具结果中获取证据;
  2. 可重复:相同输入和相同状态下,判定结果不会因措辞变化而大幅波动;
  3. 与目标相关:通过验收确实意味着目标更接近完成。

例如:

“报告写得很好”

不可观察、不可重复,也难以证明目标达成。

改成:

“报告中的总成本等于明细成本之和,误差不超过 0.01 元;
每个超过上季度 20% 的服务项都有异常说明;
每条降本建议都引用至少一个数据证据。”

就可以分别用数值校验、规则校验和引用完整性校验完成验收。


四、子任务不是平均切分,而是按依赖和可验证性切分

任务分解的质量,不取决于任务数量,而取决于每个任务是否具有合适的边界。

一个好的子任务应当满足什么条件

设任务 TiT_i 的输入为 IiI_i,输出为 OiO_i,前置条件为 PiP_i,验收条件为 CiC_i

一个任务可执行的基本条件是:

Pi(S)=trueP_i(S) = true

执行后,若得到状态 SS' 和产物 OiO_i,任务完成需要满足:

Ci(S,Oi)=trueC_i(S', O_i) = true

同时,产物必须能够满足后继任务的输入要求:

OiIjO_i \models I_j

这表示:前一个任务的输出不仅要“存在”,还必须符合下一个任务能消费的契约。

例如,T3:按团队和服务聚合成本 的产物不能只是:

“华东区成本较高”

因为后续任务无法复算,也无法知道“华东区”如何映射到杭州团队。

更适合的产物是:

{
  "period": "2026-Q3",
  "currency": "CNY",
  "group_by": ["team", "service"],
  "rows": [
    {
      "team": "hangzhou-platform",
      "service": "object-storage",
      "cost": 182340.15,
      "source_record_count": 1482
    }
  ],
  "query_hash": "sha256:...",
  "aggregation_check": {
    "detail_total": 982341.20,
    "grouped_total": 982341.20,
    "difference": 0.00
  }
}

过粗的任务:失败无法定位

T1:完成云成本分析报告

这个任务同时包含:

  • 数据获取;
  • 数据清洗;
  • 分组统计;
  • 异常检测;
  • 建议生成;
  • 报告写作;
  • 质量审查。

如果最终结果错误,系统无法判断是数据不全、计算错误、异常判断错误,还是报告表达错误。重试整个任务还会重复执行所有高成本步骤。

过细的任务:协调成本超过收益

另一种极端是:

T1:读取第一行账单
T2:读取第二行账单
T3:读取第三行账单
...
T10000:读取第一万行账单

这会导致:

  • 调度开销增加;
  • 状态记录膨胀;
  • 依赖关系难以维护;
  • 单个任务产物没有业务意义;
  • Agent 需要在大量无意义节点之间协调。

更合理的边界通常是“一批数据读取”“一个确定性变换”或“一个可验收业务结论”。

按不确定性切分

如果一个任务的后续路径高度依赖中间结果,就应该在该处建立任务边界。

例如:

读取数据
→ 判断是否存在缺失
→ 若完整:继续聚合
→ 若不完整:补拉数据或请求人工确认

这里不能把“读取、判断、补拉、聚合”硬编码成一条固定链路,因为数据完整性决定后续分支。

任务边界应放在“获取数据”之后,让系统先获得事实,再决定下一步。


五、前置条件:任务何时可以开始

前置条件不是任务描述中的背景信息,而是执行器用来判断“现在能不能运行”的机器可判定条件。

前置条件的四个层次

1. 数据前置条件

要求输入数据存在、完整且版本正确:

账单数据已获取
账单时间范围覆盖 2026-07-01 至 2026-09-30
所有记录包含 team_id、service、amount

2. 环境前置条件

要求运行环境可用:

分析容器已创建
Python 依赖已安装
对象存储路径可读
数据库连接可用

3. 权限前置条件

要求主体有权执行动作:

Agent 具有 billing:read 权限
Agent 不具有 production:write 权限
发送报告前已经获得收件人确认

权限不是提示词约束,而应由工具层和执行环境强制执行。

4. 业务前置条件

要求业务状态满足规则:

报告时间范围已经由用户确认
成本口径已确定为含税还是未税
候选降本方案未涉及冻结中的资源

前置条件与输入参数的区别

下面两项看起来相似,实际不同:

period = "2026-Q3"
billing_data_exists = true

period 是输入参数;billing_data_exists 是前置条件。

输入参数通常由任务创建时写入。前置条件则需要在执行前根据当前状态重新判断,因为它可能在任务创建后发生变化。

例如:

  • 任务创建时数据库可用;
  • 执行时数据库已经不可用;
  • 因此前置条件必须实时检查,不能只复制创建时的环境快照。

前置条件不足时不应盲目执行

如果 T4 需要 aggregated_costs,但 T3 只产生了一个没有货币单位的数值表,那么 T4 不是“可以先试试”,而是应保持 blocked

常见错误是让 Agent 自己猜测:

“虽然没有货币字段,但默认都是人民币。”

这会把缺失事实伪装成确定事实。更安全的处理是:

blocked:
  reason: currency_not_specified
  required_action:
    - query billing schema
    - or ask user to confirm currency

六、产物:让任务之间传递事实,而不是传递散文

产物是任务执行后的持久化结果,既供后续任务消费,也供恢复、审计和验收使用。

产物的四个要求

1. 可定位

产物需要有稳定标识:

{
  "artifact_id": "art_01J...",
  "type": "aggregated_costs",
  "uri": "s3://agent-runs/run-123/aggregated-costs.parquet",
  "created_by": "T3",
  "created_at": "2026-10-02T09:30:00Z"
}

后续任务不应只依赖上下文中的一段文本,而应通过 artifact_id 或 URI 读取确定版本。

2. 可解释

产物应携带来源、处理步骤和关键假设:

{
  "source": ["art_raw_billing_v7"],
  "transform": "group_by(team_id, service), sum(amount)",
  "assumptions": [
    "amount is tax-inclusive",
    "refund records are negative values"
  ]
}

没有来源和假设,后续 Agent 无法区分事实、推断和默认值。

3. 可验证

产物中应包含验证信息:

{
  "row_count": 42,
  "input_row_count": 18342,
  "checksum": "sha256:...",
  "validation": {
    "schema": "passed",
    "sum_reconciliation": "passed",
    "duplicate_check": "passed"
  }
}

4. 可消费

产物格式要与后续任务的输入契约匹配。对机器处理而言,结构化 JSON、CSV、Parquet、数据库表通常比自然语言更可靠;自然语言适合承载解释和面向人的最终交付。

产物不是最终答案

例如,T4 的产物可以是异常清单:

{
  "anomalies": [
    {
      "team": "hangzhou-platform",
      "service": "object-storage",
      "current_cost": 182340.15,
      "previous_cost": 113201.40,
      "change_rate": 0.6122,
      "baseline": "2026-Q2",
      "explanation": "unknown",
      "confidence": 0.86
    }
  ]
}

其中:

  • change_rate 是计算结果;
  • baseline 是比较基线;
  • explanation 是待验证解释;
  • confidence 是模型或规则给出的置信信息,不是事实证明。

后续任务可以针对 explanation == "unknown" 的记录继续查询,而不必重新分析全部账单。


七、完整算例:把“生成报告”分解为可执行任务图

下面以一个较完整的请求为例:

分析 2026 年第三季度杭州平台团队的云成本,找出主要增长来源,并提出不修改生产资源的降本建议,最后生成一份可审阅报告。

第一步:提取目标和约束

结构化后得到:

{
  "goal": "deliver_cloud_cost_report",
  "scope": {
    "team": "hangzhou-platform",
    "period": {
      "start": "2026-07-01",
      "end": "2026-09-30"
    }
  },
  "constraints": [
    "read_only",
    "no_production_mutation",
    "recommendations_must_have_evidence"
  ],
  "deliverable": "reviewable_markdown_report"
}

这里的 read_only 不是一句提示,而应映射为工具权限策略:分析 Agent 可以读取账单和监控数据,但不能调用资源修改接口。

第二步:识别事实依赖

报告至少依赖:

账单明细
团队归属关系
服务维度聚合
上季度比较基线
异常解释证据
建议与证据的映射

于是可以建立以下任务:

T1 获取本季度账单明细
T2 获取团队和资源归属关系
T3 获取上季度聚合成本
T4 校验账单和归属关系
T5 聚合杭州平台团队本季度成本
T6 识别成本增长和异常
T7 查询异常项的解释证据
T8 生成降本候选方案
T9 验证每条建议的证据、收益和风险
T10 生成报告
T11 验收报告

依赖关系如下:

flowchart LR
    T1[本季度账单] --> T4[数据校验]
    T2[团队归属关系] --> T4
    T3[上季度聚合成本] --> T6[增长与异常分析]
    T4 --> T5[本季度成本聚合]
    T5 --> T6
    T6 --> T7[查询异常解释]
    T6 --> T8[生成降本候选]
    T7 --> T9[验证建议]
    T8 --> T9
    T9 --> T10[生成报告]
    T10 --> T11[报告验收]

T1T2T3 在互不依赖时可以并发;T6 必须等待本季度聚合和上季度基线;T9 必须同时等待解释证据和候选建议。

第三步:为关键任务定义契约

T5:本季度成本聚合 为例:

id: T5
objective: 按团队和服务计算 2026-Q3 成本
inputs:
  - validated_billing
  - team_membership
preconditions:
  - validated_billing.status == passed
  - team_membership.status == passed
  - period.start == 2026-07-01
  - period.end == 2026-09-30
  - currency == CNY
outputs:
  - aggregated_costs
acceptance:
  - 每条记录包含 team、service、cost
  - 明细总额与聚合总额差异 <= 0.01
  - 查询条件和输入产物版本已记录
on_failure:
  missing_currency: block_and_ask
  reconciliation_failed: investigate
  transient_database_error: retry

注意 reconciliation_failed 不能简单重试。重试只能解决临时读取失败,不能解决确定性计算错误。若总额对不上,应检查:

  • 是否过滤掉了退款;
  • 是否存在重复记录;
  • 是否混用了含税和未税金额;
  • 是否存在跨月结算;
  • 是否发生分页遗漏。

这就是错误分类的作用:不同失败原因对应不同控制流。

第四步:逐步推进状态

一个任务至少需要区分以下状态:

pending      已创建但尚未检查
blocked      前置条件不满足
ready        前置条件满足,可以执行
running      正在执行
succeeded    执行成功且验收通过
failed       执行失败
needs_review 结果存在但需要人工或额外证据
cancelled    被取消

任务状态转换可表示为:

stateDiagram-v2
    [*] --> pending
    pending --> blocked: 前置条件不满足
    pending --> ready: 前置条件满足
    blocked --> ready: 条件满足
    ready --> running: 调度执行
    running --> succeeded: 产物验收通过
    running --> failed: 不可恢复错误
    running --> needs_review: 证据不足或高风险
    failed --> ready: 可重试且未超限
    failed --> pending: 需要重规划
    needs_review --> succeeded: 审核通过
    needs_review --> pending: 补充证据
    pending --> cancelled: 取消
    ready --> cancelled: 取消

执行器不能依据模型回复中的“已完成”来推进状态,而应依据事件:

ToolCallStarted
ToolCallSucceeded
ArtifactWritten
ValidationPassed
ValidationFailed
ApprovalRequested
ApprovalGranted
ApprovalRejected
Timeout

例如,工具返回:

{
  "message": "查询成功,共获取 18342 条记录"
}

并不等于 T1.succeeded。执行器还需要确认:

  • 文件是否确实写入;
  • 记录数是否符合分页统计;
  • 时间范围是否完整;
  • schema 校验是否通过;
  • 校验摘要是否持久化。

八、执行器如何决定下一步

一个简单的调度规则是:

def is_ready(task, artifacts, state):
    for condition in task["preconditions"]:
        if not evaluate(condition, artifacts, state):
            return False
    return task["status"] in {"pending", "blocked"}

def dependencies_satisfied(task, tasks):
    return all(
        dep["status"] == "succeeded"
        for dep in task["dependencies"]
    )

在实际系统中,evaluate 不能只做字符串匹配,而要读取带版本的状态和产物元数据。

下面给出一个不依赖第三方库的最小可运行示例,展示如何根据依赖关系推进任务:

from dataclasses import dataclass, field
from typing import Dict, List, Set


@dataclass
class Task:
    id: str
    dependencies: Set[str] = field(default_factory=set)
    status: str = "pending"


def runnable_tasks(tasks: Dict[str, Task]) -> List[str]:
    result = []

    for task in tasks.values():
        if task.status != "pending":
            continue

        if all(tasks[dep].status == "succeeded"
               for dep in task.dependencies):
            result.append(task.id)

    return result


tasks = {
    "T1": Task("T1"),
    "T2": Task("T2"),
    "T3": Task("T3"),
    "T4": Task("T4", {"T1", "T2"}),
    "T5": Task("T5", {"T4"}),
    "T6": Task("T6", {"T3", "T4"}),
}

print(runnable_tasks(tasks))
# ['T1', 'T2', 'T3']

tasks["T1"].status = "succeeded"
tasks["T2"].status = "succeeded"
tasks["T3"].status = "succeeded"

print(runnable_tasks(tasks))
# ['T4']

tasks["T4"].status = "succeeded"

print(runnable_tasks(tasks))
# ['T5', 'T6']

这个例子只实现了依赖调度,没有实现真实执行。它说明三个关键事实:

  1. 初始时只有没有依赖的任务可运行;
  2. T4 必须等待 T1T2 同时成功;
  3. T5T6T4 成功后可以并发。

生产执行器还需要加入:

  • 幂等键;
  • 超时;
  • 并发上限;
  • 租约或锁;
  • 任务版本;
  • 失败次数;
  • 取消传播;
  • 产物提交事务;
  • 重启恢复。

九、并发:能并发不等于应该并发

任务图提供了并发机会,但是否并发执行还要看资源、冲突和成本。

1. 数据依赖决定正确性

若任务 TaT_a 的输出是 TbT_b 的输入,则:

TaTbT_a \rightarrow T_b

不能并发。

例如:

清洗账单 → 聚合账单

聚合不能在清洗尚未提交时读取不稳定结果。

2. 写冲突决定安全性

两个任务没有显式数据依赖,但可能写入同一资源:

T1:更新工单标签
T2:关闭工单

如果两者修改同一个工单状态,仍然需要串行化、乐观锁或事务。

3. 外部副作用决定恢复方式

读取任务通常可以安全重试;发送邮件、退款、创建资源等动作可能产生重复副作用。

因此,具有副作用的工具至少应支持一种机制:

idempotency_key = run_id + task_id + attempt_group

重试时使用相同幂等键,工具端将重复请求识别为同一个业务操作。

4. 并发带来成本和上下文合并问题

并发执行多个 Agent 后,需要合并它们的产物:

研究 Agent A:发现服务成本上涨
研究 Agent B:发现监控指标异常
研究 Agent C:发现资源标签缺失

合并器必须处理:

  • 结论冲突;
  • 证据重复;
  • 事实与推断混淆;
  • 不同时间点的数据;
  • 不同置信度;
  • 同名资源的 ID 不一致。

不能简单把三段文本拼接后交给最终 Agent,因为文本合并可能丢失来源和冲突信息。


十、重规划:不是重新生成一份漂亮计划

重规划是根据新事实修改后续任务图。

什么时候需要重规划

以下情况通常意味着原计划假设失效:

  1. 任务输入不存在
账单 API 没有返回 2026-Q3 数据
  1. 数据形状与预期不同
返回数据没有 team_id,只有 account_id
  1. 资源或权限不可用
可以读取账单,但不能读取资源标签
  1. 产物验收失败
聚合总额与明细总额不一致
  1. 环境产生新状态
用户补充说明:对象存储成本中包含一次性迁移费用
  1. 执行成本超过约束
继续逐资源调查需要读取 50 万条监控记录,超过预算

重规划的输入

重规划器不应只接收原始用户请求,还应接收:

{
  "original_goal": "...",
  "current_plan_version": 3,
  "completed_tasks": ["T1", "T2", "T3"],
  "failed_task": "T4",
  "failure": {
    "type": "missing_field",
    "field": "team_id"
  },
  "artifacts": [
    "art_billing_v7",
    "art_account_mapping_v2"
  ],
  "constraints": {
    "read_only": true,
    "remaining_budget": 120
  }
}

这样重规划才能回答:

哪些已完成产物仍然有效?哪些任务需要废弃?需要新增什么任务?原目标是否仍然可达?

重规划的三种结果

1. 局部修复

原计划整体仍有效,只需新增一个补充任务:

T4 发现缺少 team_id
→ 新增 T4a:通过 account_id 查询团队映射
→ T4 继续执行

2. 路径替换

原数据源不可用,改用备用数据源:

账单 API 不可用
→ 使用已归档的 Parquet 数据
→ 增加 freshness 验收

3. 目标降级或请求人工决策

如果无法获得必要证据,系统不应编造结论:

无法确认资源归属
→ 报告只能按账户维度分析
→ 请求用户确认是否接受范围降级

重规划应保留计划版本:

plan_v1:按 team_id 分析
plan_v2:增加 account_id → team_id 映射
plan_v3:因映射缺失,降级为账户维度

版本记录的价值在于:出现错误时,可以解释 Agent 为什么在当时选择了某条路径,而不是只看到最后一份计划。


十一、失败路径:让错误改变状态,而不是污染上下文

Agent 系统常见的错误处理是把工具错误原样塞回上下文,然后让模型“继续想办法”。这在简单场景中可能有效,但在长流程中会导致:

  • 已失败任务被误认为已完成;
  • 错误数据被后续任务继续消费;
  • Agent 反复调用同一个失败工具;
  • 重试和重规划边界不清;
  • 失败成本不断累积。

更可靠的方式是把错误分类为状态事件。

错误类型 典型表现 处理方式
瞬时错误 超时、限流、临时网络错误 有上限的重试
输入错误 参数缺失、格式错误 修复参数或回退到规划
权限错误 访问被拒绝 请求授权或替代路径
数据错误 字段缺失、校验不一致 补充数据、调查或重规划
业务拒绝 审批不通过、状态不允许 等待人工决策或终止
工具语义错误 工具返回成功但结果不符合契约 标记产物无效,禁止下游消费
不可恢复错误 目标资源不存在且无替代方案 失败终止或目标降级

重试不是万能修复

若查询因为网络超时失败:

retry_count = 1

合理。

若聚合校验每次都出现:

detail_total = 982341.20
grouped_total = 981204.73

继续重试没有意义,因为相同输入和相同算法会产生相同结果。系统应该进入:

needs_investigation

并创建调查任务:

检查分页是否遗漏
检查重复记录
检查退款符号
检查时间边界
检查税费口径

十二、验收不是最后一步,而是每个任务的局部闭环

如果只在最后验收,错误会在任务图中传播。

设任务链为:

T1T2T3T4T_1 \rightarrow T_2 \rightarrow T_3 \rightarrow T_4

T1T_1 的错误输出以概率 pp 进入后续任务,并且后续任务没有独立校验,那么最终错误概率会随着链路增长。虽然实际概率不一定简单相乘,但方向是确定的:缺少中间验收会增加错误传播范围。

因此,每个任务都应形成:

读取输入
→ 执行动作
→ 生成产物
→ 验收产物
→ 提交状态

只有验收通过,产物才可标记为:

available_for_downstream = true

否则产物即使已经写入存储,也只能是:

quarantined = true

三层验收结构

层一:工具级验收

确认工具调用本身成功:

HTTP 状态码为 200
数据库游标正常关闭
文件写入成功

层二:产物级验收

确认结果符合数据契约:

字段齐全
类型正确
记录数合理
校验和匹配
总额可复算

层三:目标级验收

确认所有子目标都支持最终交付:

报告覆盖用户要求的团队和时间范围
所有核心结论均有证据
建议没有越过只读约束
未解释异常被明确标记

工具级成功不能替代产物级成功,产物级成功也不能替代目标级成功。


十三、Agent 如何参与任务分解

任务分解不是“全部交给模型”或“全部写死在代码”两种极端之间的选择。

固定分解

适用于结构稳定、风险高、验收明确的流程:

接收申请
→ 校验字段
→ 查询账户
→ 计算金额
→ 请求审批
→ 执行操作
→ 写入审计记录

代码决定任务和依赖,模型只负责有限的分类、提取或解释。

这种方式可预测性高,适合交易、审批和安全敏感动作。

动态分解

适用于子任务数量和内容依赖具体输入的工作:

审查一个大型代码变更
→ 具体需要检查哪些文件、接口和测试,取决于变更内容

此时由规划 Agent 根据目标和上下文生成任务图,再由执行器校验:

  • 是否存在循环依赖;
  • 是否缺少验收;
  • 是否违反权限;
  • 是否产生未授权副作用;
  • 是否超过任务、时间或成本上限。

Anthropic 将 orchestrator-workers 描述为由中心模型动态拆分任务、交给 worker,再综合结果;它与预定义的 parallelization 的关键差异在于,子任务不是预先固定,而是根据具体输入动态决定。(anthropic.com)

规划 Agent 不应拥有最终控制权

模型可以提出:

{
  "tasks": [...],
  "dependencies": [...],
  "acceptance": [...]
}

但运行时应由确定性控制器检查:

计划结构是否合法?
是否引用了不存在的工具?
是否缺少资源权限?
是否包含未经批准的副作用?
是否有最大迭代次数?
是否能从失败状态恢复?

规划模型负责提出候选路径,执行器负责接受、拒绝和推进路径。


十四、如何设计 Agent 的任务输出格式

任务计划适合使用结构化输出,而不是依赖 Markdown 标题解析。

一个简化的任务计划结构如下:

{
  "goal": {
    "id": "deliver_cloud_cost_report",
    "description": "交付杭州平台团队 2026-Q3 云成本分析报告",
    "acceptance": [
      {
        "id": "A1",
        "type": "coverage",
        "rule": "覆盖指定团队和日期范围"
      },
      {
        "id": "A2",
        "type": "reconciliation",
        "rule": "明细总额与聚合总额差异不超过 0.01"
      },
      {
        "id": "A3",
        "type": "traceability",
        "rule": "每条核心结论都有来源产物"
      }
    ]
  },
  "tasks": [
    {
      "id": "T1",
      "objective": "获取本季度账单",
      "depends_on": [],
      "outputs": ["raw_billing"],
      "acceptance": ["数据覆盖完整日期范围"]
    },
    {
      "id": "T2",
      "objective": "校验账单",
      "depends_on": ["T1"],
      "outputs": ["validated_billing"],
      "acceptance": ["schema passed", "duplicate check passed"]
    }
  ]
}

这里的关键不是 JSON 本身,而是字段语义稳定:

  • goal.acceptance 验收整体目标;
  • tasks[].acceptance 验收局部任务;
  • depends_on 表示调度依赖;
  • outputs 表示产物契约;
  • objective 表示任务目的,而不是工具调用清单。

如果系统使用模型生成该结构,应对输出进行二次验证:

def validate_plan(plan):
    assert "goal" in plan
    assert plan["tasks"]

    task_ids = {task["id"] for task in plan["tasks"]}

    for task in plan["tasks"]:
        assert task["objective"]
        assert task["outputs"]
        assert task["acceptance"]

        for dep in task.get("depends_on", []):
            assert dep in task_ids
            assert dep != task["id"]

    assert is_acyclic(plan["tasks"])

实际系统还应检查:

  • 依赖图是否有环;
  • 任务 ID 是否唯一;
  • 产物名称是否重复且语义冲突;
  • 验收条件是否为空;
  • 工具是否存在;
  • 任务是否违反权限策略;
  • 计划是否超出最大深度和最大节点数。

十五、状态、产物和事件必须分开保存

一个容易被忽视的问题是:任务状态、任务产物和执行事件不是同一种数据。

任务状态

表示当前控制面状态:

{
  "task_id": "T5",
  "status": "succeeded",
  "attempt": 1,
  "plan_version": 2,
  "updated_at": "2026-10-02T09:45:12Z"
}

任务产物

表示任务产生的业务结果:

{
  "artifact_id": "art_agg_001",
  "task_id": "T5",
  "type": "aggregated_costs",
  "version": 1,
  "uri": "...",
  "validation_status": "passed"
}

执行事件

表示发生过什么:

[
  {
    "type": "TaskStarted",
    "task_id": "T5",
    "at": "2026-10-02T09:40:00Z"
  },
  {
    "type": "ArtifactWritten",
    "artifact_id": "art_agg_001",
    "at": "2026-10-02T09:44:31Z"
  },
  {
    "type": "ValidationPassed",
    "task_id": "T5",
    "at": "2026-10-02T09:45:12Z"
  }
]

分开保存的原因是:

  • 状态用于快速调度;
  • 产物用于后续消费;
  • 事件用于审计、诊断和恢复。

如果进程在 ArtifactWritten 之后、ValidationPassed 之前崩溃,恢复器可以发现“产物存在但任务未提交成功”,然后重新执行验收,而不是盲目重跑整个任务。


十六、常见误解和失败表现

误解一:计划越详细越可靠

计划越详细,越可能包含未经验证的假设。

例如:

1. 调用账单接口
2. 获取 18342 条记录
3. 使用 pandas 清洗
4. 删除 amount 为空的记录
5. 按 service 分组

第 2 步的记录数可能是模型猜的,第 4 步删除空金额记录可能造成数据丢失。真正可靠的计划应把事实发现交给执行阶段:

获取账单
→ 验证分页总数和字段
→ 根据校验结果选择清洗策略

误解二:子任务完成等于目标完成

所有子任务都显示 succeeded,不代表目标一定完成。

可能出现:

  • 子任务各自成功,但覆盖范围不完整;
  • 建议生成成功,但没有证据;
  • 报告生成成功,但引用了旧版本产物;
  • 工具调用成功,但外部状态没有改变。

因此必须有独立的整体验收任务。

误解三:模型说“已完成”就可以更新状态

模型输出是非可信声明,工具和验证器才是事实来源。

错误流程:

模型:我已验证账单完整性。
执行器:将 T2 标记为 succeeded。

正确流程:

模型提出验证动作
→ 工具返回验证结果
→ 程序执行规则校验
→ 持久化验证产物
→ 更新 T2 状态

误解四:失败就重新生成整个计划

全量重规划会丢失已经获得的有效事实,并造成重复成本。

正确做法是保留:

  • 已成功任务;
  • 已验证产物;
  • 失败原因;
  • 当前约束;
  • 原计划版本。

然后只修改受影响的后续子图。

误解五:并行 Agent 越多越快

如果后续必须等待所有 Agent,且每个 Agent 都需要加载大量上下文,增加并发数可能只会增加:

  • 模型调用成本;
  • 结果合并成本;
  • 冲突处理成本;
  • 限流概率;
  • 失败面。

并行化应建立在真正独立的子任务之上,而不是把同一个问题机械地复制给多个 Agent。Anthropic 也将并行化分为独立子任务的 sectioning 和多次尝试后投票的 voting,两者的目的分别是降低延迟或提高信心。(anthropic.com)


十七、生产系统中的最小闭环

一个可以长期运行的任务分解系统,至少需要下面这条闭环:

flowchart TD
    A[用户目标] --> B[目标结构化]
    B --> C[生成或选择任务图]
    C --> D[计划校验]
    D --> E[选择可运行任务]
    E --> F[执行工具或 Agent]
    F --> G[写入产物]
    G --> H[局部验收]
    H -->|通过| I[提交任务成功]
    H -->|失败| J[分类错误]
    J -->|瞬时错误| K[有限重试]
    J -->|数据或路径变化| L[局部重规划]
    J -->|需要判断| M[人工审查]
    K --> E
    L --> C
    M --> E
    I --> N{整体目标满足?}
    N -->|否| E
    N -->|是| O[整体验收与交付]

这条闭环中有三个不可省略的控制点:

  1. 计划校验:防止模型生成非法或越权任务;
  2. 局部验收:防止坏产物向下游传播;
  3. 整体验收:防止“所有任务完成”被误认为“目标完成”。

如果任务是长时间运行的,还需要持久化计划版本、任务状态和产物索引。OpenAI 的 Agents SDK 文档将运行循环、状态、编排、守卫、可恢复状态和追踪作为独立能力;其 SDK 适合由应用方控制部署、工具、状态存储和审批,而 SDK 管理 Agent 循环。(developers.openai.com)


十八、何时不应该做任务分解

任务分解本身也有成本。对于以下请求,直接使用一次模型调用可能更合适:

把这段文字改写得更简洁
解释一个简单的技术概念
从一份短文本中提取几个字段
生成一个不需要外部事实的草稿

如果没有:

  • 多步工具调用;
  • 外部状态变化;
  • 中间结果复用;
  • 可独立验收的阶段;
  • 失败后恢复需求;

那么引入任务图、状态机和重规划器,可能只是增加延迟和维护成本。

Anthropic 建议从最简单的方案开始,只有当多步 Agentic System 能够带来可度量收益时再增加复杂度;其经验也指出,Agent 通常以更高的成本和延迟换取开放任务中的灵活性,同时存在错误累积风险。(anthropic.com)


结语:任务分解的本质是建立可验证的中间世界

任务分解真正解决的不是“如何让模型列出更多步骤”,而是把开放目标转换为一个有边界的执行系统:

  • 目标定义最终状态;
  • 子任务划分可独立管理的工作;
  • 前置条件决定任务何时可运行;
  • 产物在任务之间传递结构化事实;
  • 验收决定结果是否可以被信任;
  • 依赖关系决定执行顺序和并发机会;
  • 状态机管理运行、暂停、失败和恢复;
  • 重规划根据新事实修改未来路径;
  • 事件和证据使执行过程可以诊断和审计。

最重要的工程原则是:

不要让 Agent 只产生“下一步要做什么”,而要让它产生“为什么现在可以做、做完留下什么、凭什么算完成、失败后如何改变路径”。

当这些信息都进入结构化任务图之后,Agent 才不再只是一个会调用工具的语言模型,而成为一个能够在不确定环境中持续推进、验证和恢复的执行系统。


系列导航与关联阅读

官方资料

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