把 Agent 计划树约束成可回滚任务图

“计划看着没问题，就是跑不起来”：这是 Agent 从 Demo 走向生产时最常见的一类故障。LLM 生成的多步计划在语义上自洽，落到执行器里却常常卡在第三步：某个输入引用找不到来源，某个写操作被执行了两次，某次失败之后没人知道该从哪一步重来。问题的根源在于计划与执行之间缺少一层确定性的约束，而不是模型能力不足。下面讨论怎么补上这一层，把模型输出的计划树压平成一张带依赖、可中断、可回滚的任务图。

计划断裂的五个位置

先把故障模式列清楚，否则约束设计会失焦。生产上遇到的执行失败，八成落在下面几类：

1. 粒度不一致。“查询订单详情”是一步，“完成退款并通知用户”也是一步，后者包含状态机流转、资金操作、外部消息三个副作用，却和前者共用同一个执行语义。
2. 隐式依赖。计划文本用“然后”“基于上述结果”表达先后，图结构里没有对应的边，调度器只能串行跑，或者更糟，并行跑。
3. 产物不可引用。“整理成结构化报告”没有定义产物形态，下游取不到字段，只能把上游的原始字符串整个塞进 prompt，一步错步步错。
4. 无幂等。超时重试是最基本的容错手段，但一个非幂等的写操作重试一次就是一次数据事故。
5. 无回滚语义。失败时只有两个选项：整体重来，或者人工介入。前者会放大副作用，后者无法规模化。

这五条里，第 2、3、4、5 条都是结构问题，可以在不改动模型的前提下用确定性代码消掉。第 1 条（粒度）无法彻底解决，只能通过“约束 + 带约束重规划”逐步逼近。所以设计重心应该放在执行器侧，而不是继续在 prompt 里堆“请确保每一步都可执行”这类祈祷式指令。prompt 是概率性的，校验必须是确定性的。

从计划树到任务图：只保留数据边

模型天然输出的是树：目标分解成子目标，子目标分解成步骤，父子之间是“分解关系”。执行器需要的是图：节点之间是“数据依赖”和“控制依赖”。这两者不是一回事，直接拿树当图跑是最常见的错误。

转换规则有三条：

• 后序遍历展开。叶子节点成为可执行任务；中间节点默认不产生执行体，只作为作用域。
• 隐式 join 显式化。父节点天然“依赖所有子节点完成”，但数据上往往只需要其中一两个的产物。把这层隐式依赖降级成显式 barrier 节点，否则并行度会被无谓地压掉。
• “验证/检查”类步骤转成断言节点。读类节点没有副作用，失败的正确反应是触发局部重规划，而不是回滚整张图。这类节点必须被识别出来，否则一次正常校验失败会拖着整个流程回滚。

转换之后，每个节点必须携带契约。没有契约的节点在图上就是一个黑盒，校验器无从下手。

节点契约：让每一步都可校验

契约的字段不多，但每一条都直接对应一种线上事故：

对应的数据结构大致是这样：

from dataclasses import dataclass
from enum import Enum

class Effect(str, Enum):
    READ = "read"                  # 无副作用，可任意重放
    WRITE = "write"                # 有副作用，需补偿
    IRREVERSIBLE = "irreversible"  # 不可撤销，必须收尾阶段执行

@dataclass
class TaskNode:
    id: str
    tool: str
    deps: list[str]
    inputs: dict[str, str]         # 参数名 -> "node_id.outputs.field"
    outputs_schema: dict[str, str]
    effect: Effect
    compensate: str | None = None
    idempotent: bool = False
    timeout_s: int = 60
    max_retry: int = 2

注意 inputs 用的是引用路径而不是字面值。这一步很关键：一旦参数以字面值写进节点，上游重新执行后下游拿到的就是脏数据，重放语义直接失效。模型很容易生成字面值（它会“顺手”把上一步的结果复述进来），所以这必须在校验阶段强制打回。

三道检查与带约束的重规划

校验分三层，顺序不能反，因为越靠前的检查越便宜、错误越好定位。

结构层：依赖是否存在、图是否无环、是否有不可达节点、是否有孤立的 barrier。环检测用拓扑排序顺带完成，不要单独写 DFS。

类型层：inputs 里每个引用路径是否能在上游的 outputs_schema 中找到。宽松模式允许字段名做一次归一化（大小写、下划线），严格模式直接报错，交给重规划处理。

副作用层：写操作是否声明了补偿工具；非幂等节点是否被允许自动重试（默认不允许，只能由人工或上层策略触发）；不可逆操作是否出现在图的收尾阶段。

def validate(nodes: list[TaskNode]) -> list[dict]:
    by_id = {n.id: n for n in nodes}
    errs = []
    for n in nodes:
        for d in n.deps:
            if d not in by_id:
                errs.append({"code": "MISSING_DEP", "node": n.id, "hint": f"依赖 {d} 不存在"})
        for arg, ref in n.inputs.items():
            src, _, path = ref.partition(".")
            if src not in n.deps:
                errs.append({"code": "UNDECLARED_EDGE", "node": n.id,
                             "hint": f"参数 {arg} 引用了非依赖节点 {src}"})
            elif path not in by_id[src].outputs_schema:
                errs.append({"code": "TYPE_MISMATCH", "node": n.id,
                             "hint": f"{src} 未声明产物字段 {path}"})
        if n.effect is Effect.WRITE and not n.compensate:
            errs.append({"code": "NO_COMPENSATION", "node": n.id, "hint": "写操作缺少补偿工具"})
        if n.effect is not Effect.READ and n.idempotent is False and n.max_retry > 0:
            errs.append({"code": "RETRY_WITHOUT_IDEMPOTENCY", "node": n.id,
                         "hint": "非幂等节点不应自动重试"})
    if cycle := find_cycle(nodes):
        errs.append({"code": "CYCLE", "node": cycle[0], "hint": " -> ".join(cycle)})
    return errs

校验失败后的处理方式决定了整套设计的成败。两种做法：

• 让模型自由重写整个计划：成本高，且经常修好一个错、引入两个新错，收敛性差。
• 带约束的局部修补：把错误结构化成 (code, node_id, hint) 三元组，要求模型只输出 patch，即新增、替换或删除指定节点，然后重新走一遍检查。

第二种明显更稳，因为错误被压成了模型能直接消费的最小上下文，重写范围被限制在出错节点附近。实践中把修补轮数上限定在 2 轮：两轮修不好，说明计划本身与可用工具集不匹配，此时正确动作是降级，即缩减目标范围、或者转人工确认，而不是继续让模型试。

执行期：状态机、检查点与幂等重放

执行器本质上是一个带持久化状态的调度器。节点状态至少要有：pending / ready / running / succeeded / failed / compensating / rolled_back / skipped。把 compensating 和 rolled_back 单独拆出来，是为了让“失败”和“已撤销”在可观测性上是两件事：前者要告警，后者是预期内的收敛。

三个必须做对的点：

检查点落在节点边界。 每次状态跃迁写一次持久存储（Redis 或数据库均可），记录 plan_id / node_id / status / 输入摘要 / 输出摘要。进程重启后从检查点恢复，恢复逻辑是“跳过 succeeded、重跑 failed、对 running 做补偿”，而不是从头再来。

幂等键由输入决定。 idem_key = hash(plan_id, node_id, normalized(inputs))。同一个键的结果直接复用，这样重放是安全的。工具侧的幂等由业务保证；如果保证不了，就在执行器侧加一层去重表。

中断是协作式的。 在节点边界中断，是干净且低成本的；中断一个 running 的节点，只能发取消信号然后假定它“可能已产生副作用”。这就是为什么补偿逻辑必须覆盖 running 状态：

for node in topo_order(graph):
    if store.status(node.id) == "succeeded":
        continue
    if node.effect is not Effect.READ and store.status(node.id) == "running":
        # 上次进程在 running 时崩溃，副作用可能已经发生
        run_compensation(node)
        store.set_status(node.id, "rolled_back")
    if node.effect is not Effect.READ and not node.idempotent:
        store.set_status(node.id, "running")  # 先落盘再执行，宁可多补偿不可漏补偿
    result = call_tool(node, timeout=node.timeout_s)
    store.save(node.id, result, status="succeeded")

注意“先落盘 running 再执行”这个顺序：反过来的话，进程在“已调用工具、未记录状态”的窗口里崩溃，恢复时就完全不知道副作用是否发生，只能全量回滚。这个窗口是很多线上事故的真实来源。

回滚：补偿事务与不可逆隔离

回滚不是“把状态改回去”，而是在业务语义上抵消已发生的副作用。按 Saga 的思路，沿拓扑逆序执行补偿，每个补偿本身也必须幂等、可重试，且失败时要进死信队列而不是阻塞整条回滚链路——回滚过程中卡死比不回滚更糟。

不同操作的回滚代价差异极大，这个差异必须体现在图的结构上：

由此得出一条结构性约束：把不可逆操作全部推到收尾阶段。前 90% 的步骤都设计成可撤销的“预备态”——创建草稿、锁定库存、生成待确认单——只有所有可逆步骤都成功、所有断言节点都通过之后，才进入 commit 阶段执行不可逆动作。这把回滚问题从“如何撤销”转成了“如何不撤销”，前者往往无解，后者是纯设计问题。

还有一类中间态需要单独处理：物理删除不掉的记录（审计日志、已推送的消息）。做法是语义回滚，即打 tombstone 标记并附上 rolled_back_by 字段，查询侧统一过滤。代价是每个读路径都要感知这个标记，所以 tombstone 应该在数据模型设计阶段就留好，而不是事后补。

落地建议

1. 先给工具集加契约，再改规划逻辑。 每个工具标注 effect / 是否幂等 / 补偿工具，这是所有约束的输入。工具没有补偿工具时，就把它归到收尾阶段，别急着优化 prompt。
2. 把校验器当作独立模块上线，先只告警不拦截。 跑一到两周，统计 UNDECLARED_EDGE、TYPE_MISMATCH、NO_COMPENSATION 的分布。这些数据会直接告诉你模型的计划主要坏在哪一层，比凭直觉改 prompt 有效得多。
3. 重规划限制在 patch 粒度，且硬性封顶 2 轮。 超过就降级：缩减目标、转人工、或返回“当前工具集无法完成该目标”。不要让模型在第 5 轮重规划里继续幻觉出不存在的工具。
4. 先做检查点和幂等键，再做回滚。 恢复能力解决的是“挂了能继续”，回滚解决的是“错了能撤销”，前者收益更高、实现更简单，也是后者的前置条件。
5. 在可观测性里把 plan_id 作为一等字段。 日志、链路追踪、指标都按 plan_id + node_id 打标，否则排查“第几步出的问题”这件事会消耗掉大部分故障处理时间。