《AI Agent 开发入门》11.3 重规划

执行中某步失败或结果不满足预期时,把「已完成的子任务 + 失败点 + 观察」喂回模型产出新计划:可复用已完成的、替换失败的、追加补救的。给出 replan 实现与 seed 复用机制,用假模型脚本化「原计划第 2 步失败→重规划出替代路径→完成」,真跑打印两版计划对比,并给出何时重规划、何时直接失败上报的决策清单。

11.3 重规划

11.2 的执行器遇到失败时,会把受影响的分支标成 skipped 然后收工——它诚实,但也无能:计划断了就是断了,它不会想办法绕过去。可现实里,「检索结果里缺方法 B 的资料」并不意味着任务失败,只意味着该换条路:扩大检索范围、换个数据源、或者干脆让模型基于常识补上。这些补救动作,11.2 的执行器一个都做不了,因为它只会照着固定的计划跑。

本节把 minagent 推进到 minagent/plan.py 的修订层:新增 replan,在某个任务失败或结果不达标时,把「已完成的子任务 + 失败点 + 观察」喂回模型,产出一份修订计划;已完成的节点通过 11.2 的 seed 机制复用,不重跑。

11.3.1 为什么计划会失效

计划是执行前对任务的一次快照,而真实世界在执行中才会暴露它错在哪。三类最常见的失效:

  • 前置假设错了。计划默认「检索能拿到方法 B 的资料」,但实际没拿到。后续依赖它的步骤全部悬空。
  • 工具不适用。计划让某个任务用 search,但那一步其实需要 read 一篇具体文档,用错了工具自然失败。
  • 任务拆得不对。某一步太大、模型一次做不完(撞 max_steps),或者拆得太碎、粒度不匹配。

这三种失效有个共同点:规划阶段无法预见,只有执行到那一步才知道。这跟 5.2 讲 ReAct「边想边做」的临场调整其实是同一类需求——只是 ReAct 的调整藏在每一轮的 Thought 里、程序抓不到,而 DAG 计划的调整必须是显式的、可审计的:停下来,产出一份新计划,再继续。

所以重规划要回答的核心问题是:已经做完的活儿,怎么不浪费。原计划里 retrieve 和 extract_a 都成功了,重规划后的新计划理应复用它们的结果,只重跑失败点及其下游。把「复用」做成默认,重规划才不会变成「全部推倒重来」——否则修一个失败节点要付出重跑整张图的代价。

11.3.2 replan 的输入:已完成 + 失败点 + 观察

重规划本质上还是让模型产出计划,和 11.1 的分解是同一个动作,区别只在输入。11.1 的输入是「目标」;重规划的输入是「目标 + 当前进度 + 卡在哪」。把这份输入渲染成一段提示词:

# minagent/plan.py(重规划提示词)
REPLAN_PROMPT = """原计划未能继续执行。请给出修订后的完整计划,仍只输出 JSON 数组。
已完成的任务可原样保留(其结果会在执行时复用);失败的任务需替换或追加补救步骤。"""


def render_replan(plan: Plan, done: dict[str, str], failure: str) -> str:
    lines = [REPLAN_PROMPT, "", "原计划:"]
    for t in plan:
        mark = "done" if t.id in done else "todo"
        lines.append(f"  [{mark}] {t.id}: {t.description}")
    lines.append("")
    lines.append(f"失败点:{failure}")
    lines.append("已完成任务的结果:")
    for k, v in done.items():
        lines.append(f"  {k} => {v}")
    return "\n".join(lines)

这段提示词里有三块信息,缺一不可:

信息块内容作用
原计划(带 done/todo 标记)每步的 id 与描述让模型知道整体结构,以及哪些还欠着
失败点哪个任务、为什么失败让模型知道该改哪
已完成的结果id => 产物让模型知道手上有什么可复用

把原计划里已完成的任务标成 [done]、未完成的标 [todo],是这份提示词最巧妙的一处。它让模型一眼看出进度的边界——[done] 之后的部分才是它能改的。如果只给失败点、不给进度,模型可能把已经成功的 retrieve 也重新规划一遍,白白浪费。

11.3.3 replan 的实现

有了渲染好的提示词,replan 就是「调一次模型 + 解析」——和 11.1 的分解共用同一个 parse_plan:

# minagent/plan.py(重规划)
def replan(plan: Plan, done: dict[str, str], failure: str, model: Model) -> Plan:
    prompt = render_replan(plan, done, failure)
    resp = model.complete([Message(role="user", content=prompt)], [])
    return parse_plan(resp.message.content)

返回值仍是一个 Plan(list[Task])。这一点很关键:重规划的产物和原计划类型完全相同,所以 11.1 的 validate_plan、11.2 的 PlanExecutor 都能直接拿来用在新计划上,一行不用改。这就是「把计划做成普通数据结构」的回报——修订计划、执行计划、校验计划,处理的是同一种东西。

模型对新计划通常做三种操作,可以同时发生:

  • 复用:已完成的节点原样保留(id 不变),执行时靠 seed 跳过。
  • 替换:把失败节点换成一个新节点(如 extract_b → extract_b_v2),或换掉它的前置。
  • 追加:插入补救步骤(如新增 widen_search 去补资料),再把失败节点重新挂到补救步骤下游。

replan 不强制模型只做其中一种,怎么改是模型的判断;我们只负责把「原计划 + 进度 + 失败点」讲清楚,然后校验它给出的新计划(validate_plan)是否合法。改计划的智能交给模型,改完是否能用交给我们自己——这条分工贯穿整个规划章。

11.3.4 复用已完成的:seed 机制

新计划里那些「复用」的节点,不该重跑。11.2 的 PlanExecutor.run 已经留好了口子——seed 参数:

seed = {k: TaskResult(k, "done", v) for k, v in done.items()}
res = PlanExecutor(execute, max_workers=4).run(new_plan, seed=seed)

执行器在每一层先检查 if tid in results: continue——凡是在 seed 里出现过的任务 id,直接跳过,它的 TaskResult 作为下游任务的输入。于是新计划里 retrieve 和 extract_a 标着 [复用],实际一次都不跑,它们的产物直接从 seed 里取。

这里有个容易踩的坑:seed 的键必须和新计划里的 id 完全一致。如果模型在重规划时把 retrieve 改名叫 do_retrieve,seed 就匹配不上,do_retrieve 会被当成新任务重跑。所以 render_replan 里我们明确告诉模型「已完成的任务可原样保留」——id 稳定性是复用能生效的前提。更稳的做法是在 replan 返回后,程序化地把「同名 id 且已完成」的节点强制保留,而不是纯靠模型自觉;卷一先用提示词约束。

11.3.5 真跑:第 2 步失败后改出替代路径

构造一个场景:原计划四步,执行到 extract_b 时失败(原检索结果里缺方法 B 的资料)。把已完成的两步和失败点喂给模型,让它改计划。假模型的脚本里塞进修订后的计划 JSON:

# demo113.py
from minagent.fake import FakeModel
from minagent.plan import PlanExecutor, Task, TaskResult, replan

ORIGINAL = [
    Task("retrieve", "检索与主题相关的文档", []),
    Task("extract_a", "从检索结果抽取方法 A 的要点", ["retrieve"]),
    Task("extract_b", "从检索结果抽取方法 B 的要点", ["retrieve"]),
    Task("synthesize", "综合 A、B 要点写成对比结论", ["extract_a", "extract_b"]),
]

NEW_JSON = """[
  {"id": "retrieve",     "description": "检索与主题相关的文档", "depends_on": []},
  {"id": "extract_a",    "description": "从检索结果抽取方法 A 的要点", "depends_on": ["retrieve"]},
  {"id": "widen_search", "description": "扩大检索范围,补充方法 B 的资料", "depends_on": []},
  {"id": "extract_b_v2", "description": "基于补充资料重新抽取方法 B 的要点", "depends_on": ["widen_search"]},
  {"id": "synthesize",   "description": "综合 A、B 要点写成对比结论", "depends_on": ["extract_a", "extract_b_v2"]}
]"""

done = {"retrieve": "命中 3 篇文档", "extract_a": "extract_a 的要点"}
failure = "extract_b 失败:抽取器返回空(原检索结果里缺少方法 B 的资料)"

model = FakeModel([NEW_JSON])
new_plan = replan(ORIGINAL, done, failure, model)

打印两版计划对比,本机实测输出(Python 3.14.6):

原计划:
  retrieve     <- (无)
  extract_a    <- retrieve
  extract_b    <- retrieve
  synthesize   <- extract_a, extract_b
修订计划:
  [复用] retrieve     <- (无)
  [复用] extract_a    <- retrieve
  [新] widen_search <- (无)
  [新] extract_b_v2 <- widen_search
  [新] synthesize   <- extract_a, extract_b_v2

对照两版能清楚看到三种操作:retrieve、extract_a 复用(id 不变、标记 [复用]);extract_b 被替换成 extract_b_v2,前置从 retrieve 换成新的 widen_search;widen_search 是追加的补救步骤,它不依赖任何前置,可以立刻开跑。最后 synthesize 被重新接线——它的第二条入边从 extract_b 改指 extract_b_v2。

把修订计划交给执行器,seed 带上已完成的两步,真跑:

执行修订计划:
  retrieve     done  命中 3 篇文档
  extract_a    done  extract_a 的要点
  widen_search done  补充命中 2 篇方法 B 的资料
  extract_b_v2 done  extract_b_v2 的要点
  synthesize   done  结论:A 更快、B 更稳

retrieve 和 extract_a 的 done 来自 seed,一次没重跑;只有 widen_search、extract_b_v2、synthesize 三个新节点真执行了。从「第 2 步失败、任务卡死」到「任务完成」,只多花了一次重规划调用 + 三个新节点的执行——原计划里已成功的部分一点没浪费。这就是重规划相对「直接失败上报」的价值:它把一次失败降级成一次绕行。

11.3.6 何时重规划,何时直接失败上报

重规划不是免费的——它多一次模型调用、且新计划同样可能错。所以不是「一失败就重规划」。判断标准其实很朴素:失败是不是可补救的。列成一张决策清单:

情况处理理由
失败可补救(换数据源、换工具、补步骤)重规划有明确替代路径,值得试
失败是暂时的(限流、超时)重试(14 章)同一计划再跑一次即可,不必改图
失败源于输入非法(参数错、文件不存在)失败上报改计划也救不了,要人介入
已重规划 N 次仍失败失败上报防止无限改计划,必须设上限
高风险动作失败(扣款、删除)失败上报绕行可能触发更糟的副作用

两条最重要的:重规划必须设次数上限(比如最多 2 次),否则「改计划→又失败→再改」会无限循环,烧光 token 也停不下来;有副作用的动作失败时宁可上报也不要自动绕行,因为模型想出的「替代路径」可能触发一个更危险的写操作。自动修复的边界,是「绕行的代价不超过失败本身」。

这条清单也解释了为什么 11.2 的执行器默认只做「跳过」而不自动重规划——跳过是保守的,重规划是激进的。把激进策略做成显式的一步(而不是藏在执行器里自动触发),是为了让它可被计数、可被设限、可被人工关掉。能自动做的事,不代表该自动做。

11.3.7 把重规划包成一个受控循环

把前面几块拼起来,就是一个「分解 → 执行 → 失败则重规划」的闭环。关键是把重规划放进一个有上限的循环里,并且每次都先校验计划:

# minagent/plan.py(受控循环)
def solve_with_replan(goal, model, tools, execute, max_replans: int = 2):
    plan = Planner(model).make_plan(goal, tools)
    done: dict[str, str] = {}
    res: dict[str, TaskResult] = {}
    for _ in range(max_replans + 1):
        errs = validate_plan(plan)
        if errs:
            plan = replan(plan, done, f"计划非法: {errs}", model)
            continue
        seed = {k: TaskResult(k, "done", v) for k, v in done.items()}
        res = PlanExecutor(execute).run(plan, seed=seed)
        failed = [r for r in res.values() if r.status == "failed"]
        if not failed:
            return res
        for r in res.values():
            if r.status == "done":
                done[r.task_id] = r.output
        plan = replan(plan, done, f"{failed[0].task_id}: {failed[0].output}", model)
    return res

这个循环把 11.1 的分解、11.2 的执行、本节的修订缝在了一起,四个要点:

  • for _ in range(max_replans + 1):循环次数硬上限,最多重规划 max_replans 次。这是防止无限改计划的唯一可靠手段——别指望模型自己收敛。
  • 先 validate_plan 再执行:非法计划不进入执行,直接回炉重规划,避免执行器在坏图上卡死。
  • done 逐轮累积:每轮把 done 的任务收进字典,下一轮的 seed 就带上它们,已完成的永不重跑。
  • 超限则原样返回:max_replans 用尽还失败,就带着最后一次结果返回给上层——此时该走「失败上报」,交给人处理。

真跑一遍:假模型脚本里放两版计划,执行函数故意让 extract_b 抛错。脚本第一次调用返回原计划,extract_b 失败后触发一次重规划,第二次调用返回修订计划:

# demo113b.py
model = FakeModel([INITIAL, NEW])   # 原计划、修订计划


def execute(task, inputs):
    if task.id == "extract_b":
        raise RuntimeError("抽取器返回空")
    # ... 其余任务正常返回各自的产物


res = solve_with_replan("对比方法 A 与方法 B", model, [], execute, max_replans=2)
print("模型调用次数:", len(model.calls))
for tid in ("retrieve", "extract_a", "widen_search", "extract_b_v2", "synthesize"):
    r = res[tid]
    print(f"  {r.task_id:<12} {r.status:<5} {r.output}")

本机实测输出(Python 3.14.6):

模型调用次数: 2
  retrieve     done  命中 3 篇文档
  extract_a    done  extract_a 的要点
  widen_search done  补充命中 2 篇方法 B 资料
  extract_b_v2 done  extract_b_v2 的要点
  synthesize   done  结论:A 更快、B 更稳

模型调用次数: 2 是整条链路的成本账:一次分解 + 一次重规划,两次调用就完成了「计划失败 → 绕行 → 成功」的全过程。执行层面,retrieve 和 extract_a 在第二轮靠 seed 复用、没重跑,真正新执行的只有 widen_search、extract_b_v2、synthesize 三步。这正是「受控重规划」的完整形态:把不可控的模型判断,关进一个可计数、有上限、能兜底的循环里。

11.3.8 真实模型下会怎样

本节的修订计划是写死在脚本里的,改得恰到好处。真实模型做重规划要糙得多:

  • 可能改错地方。模型有时会把已经成功的 retrieve 也重新规划,或把 synthesize 的依赖接错,产出一个非法计划——所以 replan 之后必须紧跟 validate_plan,不过就再问一次。
  • 可能陷入循环。如果新计划换汤不换药(把 extract_b 改成 extract_b2 却还是同样的前置),下一轮还会在同一处失败。次数上限是硬约束,不能靠模型自觉收敛。
  • 复用不一定可靠。模型未必老实保留 id,seed 匹配就可能落空、导致已完成任务被重跑。真实系统通常加一层程序化的「同名已完成节点强制保留」,不完全依赖模型的自觉。

假模型证明的是「重规划的输入契约、seed 复用、执行器对新计划的兼容」这套机制对;真实模型暴露的是「修订本身不可靠,需要校验 + 限次 + 兜底」。规划章的三节到此闭环:11.1 把目标拆成图,11.2 按图执行,11.3 在执行中改图。下一章 12.1 状态机式编排 会把「图」这件事做得更一般——不只是任务依赖,还有条件跳转、分支合并与人机在环。

最后再补一层成本意识:本节脚本里模型一次就改出了好计划,而真实 Claude 需要你把「哪些能复用、哪些要替换、补救步骤怎么挂」在提示词里讲清楚,且仍要准备好它改歪时的重试。每次重规划都是一次完整的模型调用,带上原计划、进度、失败点,token 不便宜。所以真实系统往往先在便宜的层面处理失败(重试、跳过),只有确认「不改计划就过不去」时才动用重规划。

阅读导航:上一节:11.2 计划执行器 · 下一节:12.1 状态机式编排 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「agent」更多文章

  1. 《AI Agent 高级编程》目录
  2. 《AI Agent 高级编程》11.3 全景回顾:从零到框架的完整路径与展望
  3. 《AI Agent 高级编程》11.2 真实 API 冒烟:假模型 vs 真模型量化差距