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 状态机式编排 。
继续阅读
探索更多技术文章
浏览归档,发现更多关于系统设计、工具链和工程实践的内容。