11.2 计划执行器
11.1 把计划从一条链升级成一张 DAG，但还没人执行它。有了 depends_on，执行器要回答一个新问题：哪些任务可以同时跑。答案藏在图的结构里——没有任何未完成前置的任务，此刻就能跑；它们构成一个「层」。一层跑完，下一层才解锁。这一节就写这个执行器。
本节把 minagent 推进到
minagent/plan.py的执行层：新增PlanExecutor，按分层拓扑序执行Plan，同一层的无依赖任务用ThreadPoolExecutor并行，层间串行，并把前置任务的产物作为后置任务的输入。
11.2.1 拓扑序不够，要分层
11.1 的 topo_order 给出一个线性序，但它把 extract_a 和 extract_b 也排成了一前一后——那是为了「读起来有序」，不是「必须如此」。执行器需要的是分层：把图切成若干「层」，同一层的任务彼此无依赖、可以并发，不同层之间有严格先后。
分层算法就是 11.1 环检测的变体——Kahn 算法每轮摘掉当前所有入度为 0 的节点，这一整轮摘掉的节点就是一层：
# minagent/plan.py（分层）
def layers(plan: Plan) -> list[list[str]]:
indeg = {t.id: 0 for t in plan}
adj: dict[str, list[str]] = {t.id: [] for t in plan}
for t in plan:
for d in t.depends_on:
adj[d].append(t.id)
indeg[t.id] += 1
cur = sorted(i for i, n in indeg.items() if n == 0)
out: list[list[str]] = []
while cur:
out.append(cur)
nxt: list[str] = []
for i in cur:
for j in adj[i]:
indeg[j] -= 1
if indeg[j] == 0:
nxt.append(j)
cur = sorted(nxt)
return out
对 11.1 那张图，layers 返回 [['retrieve'], ['extract_a', 'extract_b'], ['synthesize']]——三层，中间层两个任务可并行。sorted 只是为了让同一层的输出顺序稳定、便于断言，不影响并发语义。分层把「谁先谁后」和「谁能同时」两个问题一次答清，这正是执行器需要的全部信息。
11.2.2 PlanExecutor 的结构
执行结果需要比 11.1 的 TaskResult 多一个「状态」。定义三种状态：
| 状态 | 含义 | 何时产生 |
|---|---|---|
done | 成功，output 是产物 | 执行函数正常返回 |
failed | 抛了异常，output 是错误信息 | 执行函数抛出 |
skipped | 未执行，因前置未成功 | 前置任务非 done |
执行器本身很短。它逐层推进，每层里把「前置全部成功」的任务提交到线程池，其余标 skipped：
# minagent/plan.py（执行器）
import time
from concurrent.futures import ThreadPoolExecutor
@dataclass
class TaskResult:
task_id: str
status: str
output: str = ""
seconds: float = 0.0
class PlanExecutor:
def __init__(self, execute, max_workers: int = 4) -> None:
self.execute = execute
self.max_workers = max_workers
def _one(self, task: Task, results: dict[str, TaskResult]) -> TaskResult:
inputs = {d: results[d].output for d in task.depends_on}
t0 = time.perf_counter()
try:
out = self.execute(task, inputs)
return TaskResult(task.id, "done", out, round(time.perf_counter() - t0, 3))
except Exception as e:
msg = f"{type(e).__name__}: {e}"
return TaskResult(task.id, "failed", msg, round(time.perf_counter() - t0, 3))
def run(self, plan: Plan, seed: dict[str, TaskResult] | None = None) -> dict[str, TaskResult]:
by_id = {t.id: t for t in plan}
results: dict[str, TaskResult] = dict(seed or {})
for layer in layers(plan):
runnable: list[Task] = []
for tid in layer:
t = by_id[tid]
if tid in results: # 已完成的（seed 复用）直接跳过
continue
bad = [d for d in t.depends_on if results[d].status != "done"]
if bad:
results[tid] = TaskResult(tid, "skipped", f"依赖未完成: {bad}")
else:
runnable.append(t)
if runnable:
with ThreadPoolExecutor(max_workers=self.max_workers) as pool:
futs = {t.id: pool.submit(self._one, t, results) for t in runnable}
for t in runnable:
results[t.id] = futs[t.id].result()
return results
执行器把「怎么执行一个任务」抽象成一个回调 execute(task, inputs)，由调用方注入。这样同一个执行器既能驱动「调一次 Agent」的任务，也能驱动「直接调工具」的任务——调度逻辑与任务语义解耦。这正是 5.2 Executor「每步复用 run()」思路的一般化：执行器只管编排，不管每个节点内部是什么。
11.2.3 层内并行、层间串行
并行发生在 with ThreadPoolExecutor(...) 那一块。它只包裹当前层的 runnable 列表，把这一层所有就绪任务一次性提交。因为同层任务在定义上互不依赖，它们同时跑不会互相踩踏。
关键在于 pool 的生命周期：它在每层内部创建、层末关闭（with 块退出即 join 全部 future）。这保证了层与层之间的屏障——第 2 层一定要等第 1 层所有任务都结束后才开始。为什么不能跨层并发？因为第 2 层的任务依赖第 1 层的产物，前置还没出结果就开跑，只会拿到空输入。
for t in runnable: results[t.id] = futs[t.id].result() 这行按提交顺序取回结果。4.3 讲过，并发下完成顺序 ≠ 提交顺序，但这里我们最后统一取、且写进以 task_id 为键的字典，所以顺序天然对齐。取 .result() 会阻塞到该 future 完成，但因为同层任务已经并行跑着，整层耗时约等于最慢的那个任务，而不是它们的和。
max_workers 给默认值 4 而不是 len(runnable)，是 4.3 那条教训的延续：并发度不是越大越好。同层若有十几个任务、且都要打第三方接口，无上限并发会触发限流。4 是个保守的默认，真实场景按下游承受力调。
11.2.4 结果回灌：把前置产物喂给后置
_one 的第一行 inputs = {d: results[d].output for d in task.depends_on} 是 DAG 执行器的灵魂。它把该任务所有前置的产物收集成一个字典 {前置id: 产物}，交给 execute。这样 synthesize 拿到的 inputs 就是 {"extract_a": "...", "extract_b": "..."}——它能同时看到两个分支的结果，这正是 DAG 相对线性计划的价值：汇聚点（join）能拿到多条上游的产物。
对比 5.2：那里用一根 context 字符串把「所有历史结果」一路拼下去，越拼越长。DAG 的 inputs 是精确的——只给当前任务真正声明依赖的那些前置，不多给。这既省 token，又避免模型被无关的历史干扰。依赖关系在这里第二次发挥作用：第一次是决定执行顺序，第二次是决定「谁的结果该喂给谁」。
execute 拿到 task 和 inputs 后做什么，就是业务了。它可以是「用 description + inputs 拼一段提示词、调一次 run()」，也可以是「直接调某个工具」。把 inputs 拼成提示词的那一行，短得可以一眼看完：
# 一个「每个 Task 调一次 Agent」的 execute
def execute(task: Task, inputs: dict[str, str]) -> str:
ctx = "\n".join(f"[{k}] {v}" for k, v in inputs.items())
prompt = f"{task.description}\n参考前序结果：\n{ctx}" if ctx else task.description
return run(model, registry, prompt).final
注意 if ctx else 那个分支：retrieve 没有前置，inputs 是空字典，此时提示词里就不该出现「参考前序结果：」这个空标题。空输入要显式处理，否则模型会对着一个空列表发懵。本节的真跑用一个纯函数模拟 execute，把注意力集中在调度上；真实场景里换成上面这几行即可。
run 里的 seed 参数这里还没用上，它是给 11.3 重规划留的口子：重规划后，已经完成的任务不该重跑，把它们的 TaskResult 作为 seed 传进来，执行器就会在对应层跳过它们、直接复用产物。
11.2.5 失败传播：依赖失败的任务标 skipped
_one 把异常兜住，转成 failed 状态而不是往外抛——这样一个任务失败不会炸掉整层，同层其他任务照常跑完。这和 4.3「call() 永不抛异常」是同一个原则：调度器不该被单个节点的失败拖垮。
但失败必须向下传播。下一层里，任何依赖了这个失败任务的节点，bad 列表非空，直接标 skipped，连 execute 都不调。为什么不「跳过失败继续跑」？因为后置任务的输入正是那个失败任务的产物——拿一个空产物去跑，要么立刻报错，要么更糟：产出一个看似正常、实则基于空数据的错误结果。标 skipped 是诚实的选择：明确告诉上层「这条分支断了」，而不是伪造一个下游结果把问题掩盖掉。
失败的三种出路，对应三种策略：
| 策略 | 做法 | 适用 |
|---|---|---|
| 中止 | 任一失败即停全部 | 强一致流程 |
| 跳过 | 标记失败，能跑的分支继续 | 分支独立、部分结果有用 |
| 重规划 | 拿失败点改计划再继续 | 失败可补救 |
本节的执行器实现了跳过：独立分支（extract_a）照常完成，只把受影响的 synthesize 标 skipped。第三种「重规划」是下一节的主角。
一个细节：skipped 的 output 里我们写了具体是哪些依赖没完成（依赖未完成: ['extract_b']），而不是一个笼统的「失败」。这条信息对 11.3 的重规划至关重要——模型要知道是哪一步断了，才能针对性地改计划。错误信息的具体程度，直接决定重规划的质量：只写「失败了」的日志，到了重规划那一步等于什么都没说。
11.2.6 真跑：并行 vs 串行
用 time.sleep 模拟每个任务的真实耗时，真跑一遍看分层并行的收益：
# demo112.py
import time
from minagent.plan import PlanExecutor, Task, layers
PLAN = [
Task("retrieve", "检索与主题相关的文档", []),
Task("extract_a", "从检索结果抽取方法 A 的要点", ["retrieve"]),
Task("extract_b", "从检索结果抽取方法 B 的要点", ["retrieve"]),
Task("synthesize", "综合 A、B 要点写成对比结论", ["extract_a", "extract_b"]),
]
def make_execute(fail: set[str] | None = None):
fail = fail or set()
def execute(task: Task, inputs: dict[str, str]) -> str:
if task.id == "retrieve":
time.sleep(0.3)
return "命中 3 篇文档"
if task.id in ("extract_a", "extract_b"):
time.sleep(0.4)
if task.id in fail:
raise RuntimeError("抽取器返回空")
return f"{task.id} 的要点"
if task.id == "synthesize":
time.sleep(0.2)
return "结论：A 更快、B 更稳"
raise ValueError(task.id)
return execute
跑并行（max_workers=4）与串行（max_workers=1）两种配置，本机实测输出（Python 3.14.6）：
分层: [['retrieve'], ['extract_a', 'extract_b'], ['synthesize']]
并行耗时: 0.92s
retrieve done 0.30s 命中 3 篇文档
extract_a done 0.41s extract_a 的要点
extract_b done 0.41s extract_b 的要点
synthesize done 0.20s 结论：A 更快、B 更稳
串行耗时: 1.32s 加速比: 1.44x
逐层算一遍就对上了：retrieve 0.3s，中间层两个 0.4s 的任务并行后约 0.4s，synthesize 0.2s，合计约 0.9s，实测 0.92s。串行则是 0.3 + 0.4 + 0.4 + 0.2 = 1.3s，实测 1.32s。加速比 1.44x——正好等于「串行总时长 ÷ 并行总时长」。注意加速比不是 2，因为只有中间两个任务真并行，头尾两个是串行的。并行收益只来自「同层任务数」，DAG 的加速上限由最宽的那一层决定。
再让 extract_b 失败，看失败传播：
=== 一个任务失败 ===
retrieve done 命中 3 篇文档
extract_a done extract_a 的要点
extract_b failed RuntimeError: 抽取器返回空
synthesize skipped 依赖未完成: ['extract_b']
extract_b 失败，但它同层的 extract_a 不受影响、照常 done；只有依赖它的 synthesize 被标 skipped，output 里写明是 extract_b 没完成。失败被精确地限制在了受影响的分支里——这正是分层执行比线性执行更抗故障的地方。
11.2.7 真实模型下会怎样
本节的 execute 是一个写死的纯函数，sleep 模拟耗时、raise 模拟失败。真实场景里，每个 Task 的 execute 通常是一次完整的 run() 调用：
- 每个任务是一次真模型调用，耗时从数百毫秒到数十秒不等，并行带来的墙钟收益远比这里大——但也更不可预测，因为各任务的实际耗时差得很开。
- 失败原因五花八门：超时、限流、输出格式不对、工具报错。本节用统一的
failed兜住，真实系统要按 14 章的思路分类——可重试的（限流）和不可重试的（输入非法）要分开处理。 - 并发要受限流约束：真实 API 有速率限制，同层任务多时
max_workers=4可能仍偏高，需要按配额调，甚至给每个下游接口单独设并发池。
假模型（这里连模型都没用，是纯函数桩）证明的是「分层、并行、失败传播这套调度逻辑对」；真实场景暴露的是「每个节点的耗时与失败都不可控」。所以生产里 DAG 执行器通常还要配超时、重试、并发配额——卷一先立住调度骨架，加固留到卷二。
再补一个真实的量级感：本节每个任务睡 0.3~0.4 秒，是夸张地快。真实系统里一次 run() 少则一两秒、多则几十秒，中间两个抽取任务并行省下的可能是一分钟而不是零点几秒。任务越慢，分层并行的收益越可观；也正因为单节点慢，一个任务卡住对整张图墙钟时间的影响才越大，超时与失败传播才越重要。
阅读导航：上一节：11.1 任务分解：把大问题拆成子任务 DAG · 下一节：11.3 重规划 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。