在上一篇《异步 Tool Executor：并发、超时熔断与幂等重试》里，我们通过 asyncio 协程池、信号量背压、硬超时熔断和全抖动指数退避，把单轮工具调用的执行速度压榨到了极致。
现在，如果大模型在一轮思考中同时发出了 5 个没有相互依赖的工具调用，我们的 AsyncToolExecutor 可以在一个 RTT（往返时间）内把它们全部并发跑完。
但当我们把视角从“单轮动作怎么跑得快”拉高到“一个包含十几个步骤的复杂任务怎么全局组织”时，你会发现我们在第 2 篇和第 3 篇里依赖的 纯 ReAct 模式（Reasoning + Acting） 存在一个根深蒂固的结构性短板：它是一个局部贪心的“走一步看一步”状态机。
我们来看一个非常典型的复杂工程任务：
“请帮我调研当前代码库中
scheduler.py和block_manager.py的接口设计，同时去读取benchmark_report.json里的性能瓶颈数据；结合源码分析为什么长 Prompt 下吞吐会下降，撰写一份优化方案写入optimization_plan.md，最后运行一次语法检查脚本确认文档格式无误。”
如果让一个纯 ReAct Agent 来做这件事，它的执行轨迹往往是这样的：
视野狭窄导致跨步串行（Missed Global Parallelism）：尽管模型支持 Parallel Tool Calling，但在没有显式全局规划（Plan）的情况下，模型在 Step 1 往往先想：“我先看看
scheduler.py吧”，于是只发出了 1 个工具调用；等 Step 2 拿到结果后，再想：“接下来看看block_manager.py”；等 Step 3 再去读benchmark_report.json。原本完全互不依赖、可以在第一波全部并发执行的 3 个探测任务，被硬生生拆成了 3 轮串行的 LLM 推理！长程目标漂移（Goal Drift & Context Myopia）：当任务执行到第 7、第 8 步时，
messages列表里已经塞满了几千行的源码内容和工具报错。模型在预测下一步动作 aₖ ~ π_θ(· | τₖ₋₁) 时，注意力会被最近两步的局部细节（比如某一个正则表达式没匹配上）完全吸走，甚至忘了最初用户还要求“最后运行一次语法检查脚本”，直接提前输出Final Answer交差了。缺乏显式的进度与依赖契约：在纯 ReAct 循环里，没有任何数据结构能直接回答“这个任务一共分几步、现在完成了百分之多少、哪些步骤被阻塞了”。
回想一下，在传统计算机系统或编译器设计里，面对复杂的计算图或构建任务（比如 Make、Bazel 或 PyTorch 的计算图执行引擎），我们是怎么做的？
我们绝不会让执行器闭着眼睛走一步看一步，而是先将目标分解为一张 有向无环图（Directed Acyclic Graph, DAG），明确每个节点（Task Node）的输入前置依赖（Dependencies），然后由调度器基于 拓扑排序（Topological Sort） 与 就绪队列（Ready Queue），把每一个波次（Wavefront）中所有入度为 0 的就绪节点全部并发发射出去！
同时，Agent 任务和静态编译脚本有一个最大的不同：外部环境是不确定的。万一 DAG 里的第 3 个节点执行失败了，或者第 2 个节点查出来的结果彻底推翻了最初的假设，静态的死计划就会撞墙。因此，我们的调度器还必须支持 动态重规划（Dynamic Re-planning / Adaptive Graph Mutation）。
这一篇，也是第二阶段（结构化输出与工具调度引擎）的收官之作，我们就来从零手写一个集成了 拓扑分层并发调度、Kahn 环路检测、节点间上下文引用注入，以及运行时动态重规划 的 DAGPlanAndSolveEngine。
1. 这一篇要解决什么问题
我们先对比一下 纯 ReAct 线性循环 与 DAG Plan-and-Solve 架构 在系统控制流上的本质差异。
在纯 ReAct 中，规划（Planning）和执行（Execution）是揉碎在每一个 Step 里耦合进行的，状态转移是一条单链：
Step₁ ⟶ Step₂ ⟶ Step₃ ⟶ … ⟶ Step_N
而在 DAGPlanAndSolveEngine 中，我们将系统职责解耦为三个相互配合的角色：
规划器（
Planner）：在任务开始时，根据用户目标 u 生成一张显式的任务依赖图 G = (V, E)。其中每个节点 vᵢ ∈ V 代表一个原子子任务（可直接绑定工具调用或子推理目标），每条有向边 (vᵢ, vⱼ) ∈ E 表示“子任务 vⱼ 必须等待 vᵢ 成功产出结果后才能启动”。拓扑波次调度器（
DAGScheduler）：维护每个节点的实时状态（PENDING、READY、RUNNING、COMPLETED、FAILED、SKIPPED）。在每一轮调度循环中，自动找出所有前置依赖均已COMPLETED的节点集合（即当前波次 Wavefront），将它们打包交给我们在第 5 篇实现的异步并发引擎同时执行！重规划器（
Replanner）：当任意关键节点执行失败（FAILED），或者某个探测节点的输出触发了计划修正条件时，暂停后续依赖分支，将“已完成节点的产物 + 失败节点的错误现场 + 剩余未完成子图”交由Replanner局部修剪并生成新的后继子图 G'，实现闭环自愈。
用架构图表示，整条调度链路如下：
User Query
│
▼
┌───────────────────┐
│ Planner │ ──► 生成初始任务图 G = (V, E) 并通过 Kahn 算法校验无环性
└─────────┬─────────┘
│
▼
┌───────────────────────────────────────────────────────────────────┐
│ DAG Topological Scheduler │
│ 1. 计算当前就绪集 ReadySet = { v ∈ PENDING | deps(v) ⊆ COMPLETED } │
│ 2. 注入前置依赖节点的输出产物 Context(deps(v)) │
└─────────┬─────────────────────────────────────────────────────────┘
│
▼ asyncio.gather (Wavefront 并发执行全部就绪节点)
┌───────────────────────────────────────────────────────────────────┐
│ Node Workers (Tool Execution / Sub-Step) │
└─────────┬───────────────────────────────┬─────────────────────────┘
│ (全部成功) │ (出现节点失败或需修正计划)
▼ ▼
更新节点状态为 COMPLETED ┌───────────────────┐
解锁下一波次后继节点 │ Replanner │ ──► 保留已完成节点，重写剩余子图 G'
│ └─────────┬─────────┘
└───────────────┬───────────────────┘
▼ (当所有节点 COMPLETED)
Synthesizer 汇总全局最终答案
2. DAG 调度的形式化建模：关键路径、Kahn 环路检测与加速比定理
在动手写调度器之前，我们先用图论和调度理论把 DAG 执行的数学性质推导清楚。
2.1 任务依赖图与前置条件约束
设规划器将复杂任务分解为包含 N 个子任务的有向图 G = (V, E)，其中节点集 V = {v₁, v₂, …, v_N}，有向边集 E ⊆ V × V。
对于任意节点 vᵢ ∈ V，定义其**直接前驱集合（Direct Predecessors / Dependencies）**为 P(vᵢ)，**直接后继集合（Direct Successors）**为 S(vᵢ)：
P(vᵢ) = { u ∈ V | (u, vᵢ) ∈ E }，S(vᵢ) = { w ∈ V | (vᵢ, w) ∈ E }
节点 vᵢ 的动态未完成入度（Remaining In-Degree） d_rem(vᵢ, t) 定义为在时刻 t 尚未进入 COMPLETED 状态的前驱节点数量：
d_rem(vᵢ, t) = Σ(u ∈ P(vᵢ)) 𝕀[ status(u, t) ≠ COMPLETED ]
调度器的核心不变式（Scheduling Invariant）非常简单：一个处于 PENDING 状态的节点 vᵢ 在时刻 t 能够转移为 READY 并被发射执行，当且仅当其动态未完成入度归零：
vᵢ ∈ R(t) ⟺ status(vᵢ, t) = PENDING ∧ d_rem(vᵢ, t) = 0
当节点 vᵢ 启动执行时，它不仅能看到自己的任务描述，调度器还会自动将它的所有直接前驱节点 u ∈ P(vᵢ) 的输出产物 { O(u) | u ∈ P(vᵢ) } 作为结构化上下文注入给 vᵢ。这意味着：不需要把整张图所有历史节点的输出都塞进每个节点的 Prompt 里，节点 vᵢ 只需要看它显式依赖的 P(vᵢ) 的输出，这在多分支任务中能大幅节省 Token 消耗！
2.2 为什么必须用 Kahn 算法做环路检测（Cycle Detection）？
因为图 G = (V, E) 是由大模型生成的，模型偶尔会犯逻辑糊涂，生成出带有循环依赖的死锁图（例如任务 A 依赖任务 B，任务 B 依赖任务 C，而任务 C 又反过来依赖任务 A：v_A → v_B → v_C → v_A）。
如果图里存在环路 C ⊆ V，那么环上每一个节点的动态入度永远满足 d_rem(v, t) ≥ 1，调度器会在跑完环外节点后发现 R(t) = ∅，但仍有未完成节点，直接陷入死锁（Deadlock）！
为了在规划阶段第一时间拦截非法环路并把图划分成清晰的并发波次（Topological Generations / Wavefronts），我们使用经典的 分层 Kahn 算法：
设初始未访问节点集 U₀ = V。在第 l 层（l = 1, 2, …），取出当前导出子图 G[Uₗ₋₁] 中所有入度为 0 的节点构成第 l 个并发波次 Wₗ：
Wₗ = { v ∈ Uₗ₋₁ | P(v) ∩ Uₗ₋₁ = ∅ }
随后更新剩余节点集 Uₗ = Uₗ₋₁ \ Wₗ。当迭代终止于 Wₗ₊₁ = ∅ 时：
无环充要条件：图 G 是合法的 DAG，当且仅当剩余未剥离节点集为空，即： U_L = ∅ ⟺ Σ(l = 1…L) |Wₗ| = |V|
环路精确定位：若 U_L ≠ ∅，则 U_L 中包含的节点即为导致循环依赖的死锁子图节点集合，我们可以立即将其格式化为报错信息要求 Planner 修正！
2.3 Work-Span（工作量-关键路径）加速比定理
在并发计算理论（Work-Span Model / Brent's Theorem）中，一张 DAG 任务图有两个决定性能的本征指标：
总工作量（Total Work, T_1）：图中所有节点各自执行耗时 c(v) 的总和，也就是纯串行 ReAct 必须付出的时间： T_1 = Σ(v ∈ V) c(v)
关键路径长度（Critical Path / Span, T_∞）：从入度为 0 的源节点到出度为 0 的汇节点的所有有向路径中，耗时累加最长的那条路径的长度。可以通过在 DAG 上做一次动态规划（DP）在 O(|V| + |E|) 时间内精确算出： EF(v) = c(v) + max(u ∈ P(v)) EF(u)，T_∞ = max(v ∈ V) EF(v) 其中 EF(v) 表示节点 v 的最早完成时间（Earliest Finish Time，若 P(v) = ∅ 则记 max(u ∈ ∅) EF(u) = 0）。
在并发上限为 P 的异步执行器下，DAG 调度的总耗时 T_P 严格满足 Graham-Brent 上下界不等式：
max( T_1/P, T_∞ ) ≤ T_P ≤ (T_1 − T_∞) / P + T_∞
当并发槽位充足（P ≥ max_l |Wₗ|）时，T_P 直接收敛到关键路径下界 T_∞！
举个例子：刚才我们提到的工程任务包含 6 个节点——前 3 个并行的代码/报告读取节点各耗时 2 s，第 4 个分析节点依赖前 3 个节点耗时 3 s，第 5 个写文档节点耗时 2 s，第 6 个语法检查节点耗时 1 s。
串行总工作量 T_1 = (2 + 2 + 2) + 3 + 2 + 1 = 12 s，且需要 6 轮串行调度；
DAG 关键路径 T_∞ = max(2, 2, 2) + 3 + 2 + 1 = 8 s，前 3 个节点在第 1 波次（W₁）同时完成，不仅端到端耗时缩短了 33%，而且整个任务的拓扑脉络一目了然！
3. 核心数据结构：定义 TaskNode 与 TaskDAG（dag_schema.py）
我们来动手实现这套 DAG 调度引擎。首先，在 dag_schema.py 中定义任务节点 TaskNode 和支持 Kahn 算法拓扑分层、环路检测、关键路径计算与失败级联传播的图容器 TaskDAG。
特别注意我们在 TaskNode 上的一个实用设计：参数模板占位符替换（Output Interpolation）。
比如节点 task_1 调用工具找到了配置文件的路径 "/etc/app/prod.yaml"，依赖它的后继节点 task_2 在规划阶段还不知道这个路径具体是什么，Planner 可以在 task_2 的工具参数里写上占位符 {"file_path": "{{task_1.output}}"}。当 task_1 执行完毕、调度器准备发射 task_2 时，会自动把 {{task_1.output}} 替换为 task_1 的真实输出！
from __future__ import annotations
import json
import re
from dataclasses import asdict, dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Set, Tuple
class NodeStatus(str, Enum):
"""DAG 节点的生命周期状态"""
PENDING = "pending" # 尚有前置依赖未完成
READY = "ready" # 所有前置依赖均已 COMPLETED，等待调度器发射
RUNNING = "running" # 正在异步执行中
COMPLETED = "completed" # 执行成功并已产出 result
FAILED = "failed" # 执行失败（将触发重规划或级联跳过后继节点）
SKIPPED = "skipped" # 因前置依赖失败或重规划被剪枝跳过
@dataclass
class TaskNode:
"""DAG 中的单个原子任务节点"""
id: str # 唯一节点标识，如 "step_1"
description: str # 子任务目标描述
tool_name: Optional[str] = None # 绑定的工具名；若为 None 则代表纯 LLM 推理汇总节点
tool_args: Dict[str, Any] = field(default_factory=dict)
dependencies: List[str] = field(default_factory=list) # 前驱节点 ID 列表
status: NodeStatus = NodeStatus.PENDING
result: Optional[str] = None
error: Optional[str] = None
execution_time_ms: float = 0.0
estimated_cost_ms: float = 1000.0 # 用于关键路径 (Critical Path) 预估
def resolve_interpolated_args(self, completed_outputs: Dict[str, str]) -> Dict[str, Any]:
"""
将 tool_args 中的 {{node_id.output}} 占位符替换为前驱节点的真实输出结果。
"""
raw_json = json.dumps(self.tool_args, ensure_ascii=False)
def _replacer(match: re.Match) -> str:
dep_id = match.group(1).strip()
if dep_id in completed_outputs:
# 转义字符串内部的特殊字符以保持合法 JSON 结构
val = completed_outputs[dep_id].strip()
escaped = json.dumps(val, ensure_ascii=False)[1:-1]
return escaped
return match.group(0)
interpolated = re.sub(r"\{\{\s*([a-zA-Z0-9_-]+)\.output\s*\}\}", _replacer, raw_json)
try:
return json.loads(interpolated)
except json.JSONDecodeError:
return dict(self.tool_args)
class DAGValidationError(ValueError):
"""当任务图存在未知依赖引用或循环依赖 (Cycle) 时抛出"""
pass
class TaskDAG:
"""
有向无环图 (DAG) 任务容器：
内置分层 Kahn 拓扑排序、环路检测、关键路径 (Span) 分析与状态传播。
"""
def __init__(self, nodes: Optional[List[TaskNode]] = None) -> None:
self.nodes: Dict[str, TaskNode] = {}
if nodes:
for node in nodes:
self.add_node(node)
self.validate()
def add_node(self, node: TaskNode) -> None:
self.nodes[node.id] = node
def validate(self) -> List[List[str]]:
"""
执行完整性检查与分层 Kahn 算法：
- 检查是否存在指向未知节点 ID 的依赖边或自环；
- 若存在环路，精确指出构成死锁的节点集合并抛出 DAGValidationError；
- 若合法，返回按并发波次划分的拓扑分层列表 [[wave_1_ids], [wave_2_ids], ...]。
"""
# 1. 检查未知前驱引用与自环
for nid, node in self.nodes.items():
for dep in node.dependencies:
if dep == nid:
raise DAGValidationError(f"Node '{nid}' has a self-dependency cycle.")
if dep not in self.nodes:
raise DAGValidationError(
f"Node '{nid}' depends on non-existent node '{dep}'. "
f"Valid node IDs: {list(self.nodes.keys())}"
)
# 2. 分层 Kahn 算法剥离入度为 0 的波次 (Wavefronts)
remaining: Set[str] = set(self.nodes.keys())
waves: List[List[str]] = []
while remaining:
current_wave = [
nid
for nid in remaining
if all(dep not in remaining for dep in self.nodes[nid].dependencies)
]
if not current_wave:
# 剩余节点中没有任何入度为 0 的节点 -> 必然存在环路！
cycle_nodes = sorted(list(remaining))
raise DAGValidationError(
f"Cyclic dependency detected in TaskDAG! Deadlocked nodes: {cycle_nodes}"
)
waves.append(sorted(current_wave))
remaining -= set(current_wave)
return waves
def compute_work_and_span(self, use_actual_time: bool = False) -> Tuple[float, float, List[str]]:
"""
基于动态规划计算 DAG 的总工作量 (Total Work T_1)、关键路径耗时 (Span T_inf) 与关键路径节点序列。
"""
waves = self.validate()
earliest_finish: Dict[str, float] = {}
predecessor_on_cp: Dict[str, Optional[str]] = {}
total_work = 0.0
for wave in waves:
for nid in wave:
node = self.nodes[nid]
cost = node.execution_time_ms if use_actual_time else node.estimated_cost_ms
total_work += cost
if not node.dependencies:
earliest_finish[nid] = cost
predecessor_on_cp[nid] = None
else:
best_dep = max(node.dependencies, key=lambda d: earliest_finish[d])
earliest_finish[nid] = earliest_finish[best_dep] + cost
predecessor_on_cp[nid] = best_dep
if not earliest_finish:
return 0.0, 0.0, []
end_node = max(earliest_finish, key=lambda k: earliest_finish[k])
span = earliest_finish[end_node]
# 回溯重构关键路径节点序列
critical_path: List[str] = []
curr: Optional[str] = end_node
while curr is not None:
critical_path.append(curr)
curr = predecessor_on_cp[curr]
critical_path.reverse()
return total_work, span, critical_path
def get_ready_nodes(self) -> List[TaskNode]:
"""
扫描并返回当前所有满足启动条件（自身为 PENDING 且所有前驱均已 COMPLETED）的节点，
同时将其状态推进为 READY。
"""
ready_list: List[TaskNode] = []
for node in self.nodes.values():
if node.status in (NodeStatus.PENDING, NodeStatus.READY):
if all(self.nodes[dep].status == NodeStatus.COMPLETED for dep in node.dependencies):
node.status = NodeStatus.READY
ready_list.append(node)
return ready_list
def mark_failed_and_cascade(self, failed_node_id: str, error_msg: str) -> List[str]:
"""
当某个节点失败且不进行重规划时，将其标记为 FAILED，
并通过 BFS 将其所有下游传递后继节点标记为 SKIPPED。
"""
failed_node = self.nodes[failed_node_id]
failed_node.status = NodeStatus.FAILED
failed_node.error = error_msg
skipped_ids: List[str] = []
queue: List[str] = [failed_node_id]
visited: Set[str] = {failed_node_id}
while queue:
curr_id = queue.pop(0)
for nid, node in self.nodes.items():
if nid not in visited and curr_id in node.dependencies:
if node.status in (NodeStatus.PENDING, NodeStatus.READY):
node.status = NodeStatus.SKIPPED
node.error = f"Skipped due to upstream failure in '{failed_node_id}'."
skipped_ids.append(nid)
visited.add(nid)
queue.append(nid)
return skipped_ids
def get_completed_outputs(self) -> Dict[str, str]:
return {
nid: (node.result or "")
for nid, node in self.nodes.items()
if node.status == NodeStatus.COMPLETED
}
def is_all_done(self) -> bool:
return all(
node.status in (NodeStatus.COMPLETED, NodeStatus.FAILED, NodeStatus.SKIPPED)
for node in self.nodes.values()
)
4. 实现动态重规划（Adaptive Re-planning）与子图热替换
静态 DAG 跑顺风局非常快，但一旦遇到逆风局——比如某个探测节点 step_2 在读取 config.yaml 时报错说“文件不存在，新版本已迁移至 settings.toml”，如果直接把下游依赖 step_2 的分析节点全部 SKIPPED 掉，整个任务就半途而废了。
这时候，动态重规划（Dynamic Re-planning） 就派上用场了。
当第 l 波次中出现节点执行失败时，我们的调度引擎不会丢弃已经成功跑完的节点（比如 step_1 和 step_3 已经花时间和 Token 拿到了宝贵的输出），而是执行一次 原位子图热替换（In-Place Subgraph Hot-Swap）：
锁定已完成集合 V_done：保留所有状态为
COMPLETED的节点及其输出产物；调用
Replanner：将原始目标 u、已完成节点的输出摘要 { (v, O(v)) | v ∈ V_done }、以及刚刚失败节点的错误原因 (v_fail, E(v_fail)) 传给模型，要求模型生成一张修正后的剩余任务子图 G_rem；合并与无环验证：将 V_done 与新生成的 G_rem 合并，调用
dag.validate()确认新图不存在悬空依赖或环路，然后无缝恢复下一波次的拓扑调度！
我们来编写包含 PlannerProtocol、ReplannerProtocol 以及核心调度循环的 dag_engine.py：
from __future__ import annotations
import asyncio
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
from async_executor import AsyncToolExecutor
from dag_schema import DAGValidationError, NodeStatus, TaskDAG, TaskNode
from schema import Observation, ToolCall
@dataclass
class WaveExecutionLog:
"""记录单个拓扑波次 (Wavefront) 的并发执行详情"""
wave_index: int
node_ids: List[str]
wave_wall_time_ms: float
sum_node_time_ms: float
triggered_replan: bool = False
@dataclass
class DAGRunReport:
"""整张 DAG 任务图的完整运行报告"""
user_query: str
dag: TaskDAG
waves_log: List[WaveExecutionLog] = field(default_factory=list)
replan_count: int = 0
total_wall_time_ms: float = 0.0
final_answer: str = ""
class PlannerProtocol:
"""负责根据用户初始请求生成 DAG 计划，并在节点失败时执行动态重规划"""
async def create_initial_plan(
self, user_query: str, tool_schemas: List[Dict[str, Any]]
) -> TaskDAG:
raise NotImplementedError
async def replan_on_failure(
self,
user_query: str,
completed_outputs: Dict[str, str],
failed_node: TaskNode,
tool_schemas: List[Dict[str, Any]],
) -> List[TaskNode]:
"""根据已完成节点产物与当前失败节点的报错，返回替换剩余未完成部分的新节点列表"""
raise NotImplementedError
async def synthesize_node_or_final(
self,
user_query: str,
node: Optional[TaskNode],
dependency_outputs: Dict[str, str],
) -> str:
"""执行不绑定具体工具的纯 LLM 分析汇总节点，或生成最终全局总结"""
raise NotImplementedError
class DAGPlanAndSolveEngine:
"""
基于拓扑波次并发调度与动态重规划的 Plan-and-Solve 引擎。
"""
def __init__(
self,
planner: PlannerProtocol,
tool_executor: AsyncToolExecutor,
max_replans: int = 2,
) -> None:
self.planner = planner
self.tool_executor = tool_executor
self.max_replans = max_replans
async def _execute_single_node(
self, user_query: str, node: TaskNode, completed_outputs: Dict[str, str]
) -> None:
"""执行单个就绪节点（支持绑定工具调用或纯 LLM 推理汇总节点）"""
node.status = NodeStatus.RUNNING
start = time.perf_counter()
# 收集当前节点直接依赖的前驱输出
dep_context = {dep: completed_outputs[dep] for dep in node.dependencies if dep in completed_outputs}
if node.tool_name:
# 1. 解析并替换参数中的 {{dep_id.output}} 占位符
resolved_args = node.resolve_interpolated_args(completed_outputs)
tc = ToolCall(
id=f"dag_{node.id}_{uuid.uuid4().hex[:6]}",
name=node.tool_name,
arguments=resolved_args,
)
obs: Observation = await self.tool_executor.execute_one(tc)
node.execution_time_ms = (time.perf_counter() - start) * 1000.0
if obs.is_error:
node.status = NodeStatus.FAILED
node.error = obs.output
else:
node.status = NodeStatus.COMPLETED
node.result = obs.output
else:
# 2. 纯 LLM 推理汇总节点（例如结合前两个工具的输出做对比分析）
try:
synthesis = await self.planner.synthesize_node_or_final(
user_query=user_query,
node=node,
dependency_outputs=dep_context,
)
node.execution_time_ms = (time.perf_counter() - start) * 1000.0
node.status = NodeStatus.COMPLETED
node.result = synthesis
except Exception as exc:
node.execution_time_ms = (time.perf_counter() - start) * 1000.0
node.status = NodeStatus.FAILED
node.error = f"{type(exc).__name__}: {exc}"
def _apply_subgraph_replan(self, dag: TaskDAG, new_nodes: List[TaskNode]) -> None:
"""
原位热替换未完成子图：
保留所有 COMPLETED 节点，移除 PENDING / READY / FAILED / SKIPPED 节点，
挂载 Replanner 新生成的后继节点并做 Kahn 无环校验。
"""
completed_ids = {
nid for nid, n in dag.nodes.items() if n.status == NodeStatus.COMPLETED
}
# 清理未完成或已失败的旧节点
for nid in list(dag.nodes.keys()):
if nid not in completed_ids:
del dag.nodes[nid]
# 注入新规划的节点
for new_node in new_nodes:
if new_node.id in completed_ids:
continue
dag.add_node(new_node)
# 验证合并后的图依然是合法的 DAG
dag.validate()
async def run(self, user_query: str) -> DAGRunReport:
overall_start = time.perf_counter()
schemas = self.tool_executor.get_schemas()
# 1. 生成初始 DAG 计划并执行 Kahn 无环验证
dag = await self.planner.create_initial_plan(user_query, schemas)
dag.validate()
report = DAGRunReport(user_query=user_query, dag=dag)
wave_idx = 0
# 2. 按拓扑波次 (Wavefront) 循环推进，直至整张图所有节点处理完毕
while not dag.is_all_done():
ready_nodes = dag.get_ready_nodes()
if not ready_nodes:
# 防御性检查：如果还有未完成节点但没有任何节点就绪，立刻终止避免死循环
break
wave_idx += 1
wave_start = time.perf_counter()
completed_snapshot = dag.get_completed_outputs()
# 并发发射当前波次的所有就绪节点！
await asyncio.gather(
*(self._execute_single_node(user_query, n, completed_snapshot) for n in ready_nodes)
)
wave_wall_ms = (time.perf_counter() - wave_start) * 1000.0
sum_node_ms = sum(n.execution_time_ms for n in ready_nodes)
# 检查当前波次是否有节点执行失败
failed_nodes = [n for n in ready_nodes if n.status == NodeStatus.FAILED]
triggered_replan = False
if failed_nodes:
primary_failed = failed_nodes[0]
if report.replan_count < self.max_replans:
# 触发动态重规划：保留已完成节点成果，局部重写剩余子图
report.replan_count += 1
triggered_replan = True
new_sub_nodes = await self.planner.replan_on_failure(
user_query=user_query,
completed_outputs=dag.get_completed_outputs(),
failed_node=primary_failed,
tool_schemas=schemas,
)
self._apply_subgraph_replan(dag, new_sub_nodes)
else:
# 超过最大重规划次数：级联跳过下游依赖分支
for fn in failed_nodes:
dag.mark_failed_and_cascade(fn.id, fn.error or "Execution failed")
report.waves_log.append(
WaveExecutionLog(
wave_index=wave_idx,
node_ids=[n.id for n in ready_nodes],
wave_wall_time_ms=wave_wall_ms,
sum_node_time_ms=sum_node_ms,
triggered_replan=triggered_replan,
)
)
# 3. 汇总所有已完成节点的输出，生成最终答复
report.final_answer = await self.planner.synthesize_node_or_final(
user_query=user_query,
node=None,
dependency_outputs=dag.get_completed_outputs(),
)
report.total_wall_time_ms = (time.perf_counter() - overall_start) * 1000.0
return report
5. 实战演练：测试 Kahn 环路拦截、拓扑并发加速与故障动态重规划
我们来写一个完整的验证脚本 demo_dag_planner.py，一次性验证 DAGPlanAndSolveEngine 的三项核心能力：
实验一（Kahn 算法环路死锁拦截）：故意构造一张存在三元环
step_2 -> step_3 -> step_4 -> step_2的错误计划图，验证dag.validate()如何在毫秒级精准揪出死锁环路节点['step_2', 'step_3', 'step_4']。实验二（多分支拓扑波次并发 + 参数占位符注入 + 故障动态重规划）：
在 Wave 1 中，调度器自动识别出
step_1（读取scheduler.py）、step_2（读取block_manager.py）和step_3（读取旧路径legacy_bench.json）没有任何前置依赖，将它们 3 路并发发射！其中
step_1和step_2在200ms内成功完成；而step_3报错提示"File 'legacy_bench.json' not found. Benchmarks moved to 'v2_bench.json'"。引擎立刻触发 动态重规划（Replan）：完整保留
step_1和step_2已经读出来的源码结果（绝不重跑），将失败的step_3替换为读取正确路径'v2_bench.json'的step_3_fixed，并在下游step_4（汇总分析）和step_5（使用{{step_4.output}}占位符写入报告）中无缝衔接完成任务！
import asyncio
from typing import Any, Dict, List, Optional
from async_executor import AsyncToolExecutor
from dag_engine import DAGPlanAndSolveEngine, PlannerProtocol
from dag_schema import DAGValidationError, TaskDAG, TaskNode
# 1. 注册模拟工具到第 5 篇实现的 AsyncToolExecutor
executor = AsyncToolExecutor(max_concurrency=8)
@executor.register(timeout_sec=5.0, idempotent=True)
async def read_source_module(module_name: str) -> str:
"""异步读取指定源码模块的核心摘要。"""
await asyncio.sleep(0.20)
if module_name == "scheduler.py":
return "Scheduler: supports continuous batching and PD disaggregation."
if module_name == "block_manager.py":
return "BlockManager: manages prefix caching via hash-indexed KV blocks."
return f"Module {module_name} content."
@executor.register(timeout_sec=5.0, idempotent=True)
async def load_benchmark_json(file_path: str) -> str:
"""加载性能压测 JSON 报告。"""
await asyncio.sleep(0.20)
if file_path == "legacy_bench.json":
raise FileNotFoundError(
"File 'legacy_bench.json' no longer exists. Benchmark data was migrated to 'v2_bench.json'."
)
return '{"ttft_ms": 38.2, "prefix_cache_hit_rate": 0.91}'
@executor.register(timeout_sec=5.0, idempotent=False)
async def write_markdown_report(target_file: str, summary_content: str) -> str:
"""将最终分析报告写入目标 Markdown 文件。"""
await asyncio.sleep(0.10)
return f"Saved {len(summary_content)} chars to '{target_file}' -> Content: {summary_content}"
# 2. 实现模拟 Planner：展示初始 DAG 生成与遇到 FileNotFoundError 时的局部重规划
class SimulatedDAGPlanner(PlannerProtocol):
async def create_initial_plan(self, user_query: str, tool_schemas: List[Dict[str, Any]]) -> TaskDAG:
# 初始计划：step_1, step_2, step_3 互不依赖（可在 Wave 1 完全并发执行！）
# step_4 依赖前三个节点；step_5 依赖 step_4 并通过 {{step_4.output}} 动态注入参数
nodes = [
TaskNode(
id="step_1",
description="读取 scheduler.py 核心设计",
tool_name="read_source_module",
tool_args={"module_name": "scheduler.py"},
dependencies=[],
estimated_cost_ms=200.0,
),
TaskNode(
id="step_2",
description="读取 block_manager.py 核心设计",
tool_name="read_source_module",
tool_args={"module_name": "block_manager.py"},
dependencies=[],
estimated_cost_ms=200.0,
),
TaskNode(
id="step_3",
description="加载压测报告 (初始误用了旧文件名 legacy_bench.json)",
tool_name="load_benchmark_json",
tool_args={"file_path": "legacy_bench.json"},
dependencies=[],
estimated_cost_ms=200.0,
),
TaskNode(
id="step_4",
description="综合源码与压测数据生成优化结论",
tool_name=None, # 纯 LLM 推理汇总节点
dependencies=["step_1", "step_2", "step_3"],
estimated_cost_ms=150.0,
),
TaskNode(
id="step_5",
description="将分析结论写入 report.md",
tool_name="write_markdown_report",
tool_args={
"target_file": "optimization_report.md",
"summary_content": "{{step_4.output}}",
},
dependencies=["step_4"],
estimated_cost_ms=100.0,
),
]
return TaskDAG(nodes)
async def replan_on_failure(
self,
user_query: str,
completed_outputs: Dict[str, str],
failed_node: TaskNode,
tool_schemas: List[Dict[str, Any]],
) -> List[TaskNode]:
# 看到 step_3 报错说文件迁移到了 v2_bench.json，保留已完成的 step_1 和 step_2，
# 生成修复后的 step_3_fixed 以及更新依赖后的 step_4, step_5
print(f"\n [Replanner 触发] 检测到节点 '{failed_node.id}' 失败: {failed_node.error}")
print(f" [Replanner 决策] 保留已完成节点 {list(completed_outputs.keys())}，重写剩余子图 -> 改读 'v2_bench.json'")
return [
TaskNode(
id="step_3_fixed",
description="从迁移后的正确路径 v2_bench.json 加载压测报告",
tool_name="load_benchmark_json",
tool_args={"file_path": "v2_bench.json"},
dependencies=[],
estimated_cost_ms=200.0,
),
TaskNode(
id="step_4",
description="综合源码与压测数据生成优化结论",
tool_name=None,
dependencies=["step_1", "step_2", "step_3_fixed"],
estimated_cost_ms=150.0,
),
TaskNode(
id="step_5",
description="将分析结论写入 report.md",
tool_name="write_markdown_report",
tool_args={
"target_file": "optimization_report.md",
"summary_content": "{{step_4.output}}",
},
dependencies=["step_4"],
estimated_cost_ms=100.0,
),
]
async def synthesize_node_or_final(
self,
user_query: str,
node: Optional[TaskNode],
dependency_outputs: Dict[str, str],
) -> str:
await asyncio.sleep(0.10)
if node is not None and node.id == "step_4":
return (
f"Analysis: ({dependency_outputs.get('step_1')}) + "
f"({dependency_outputs.get('step_2')}) + "
f"Metrics={dependency_outputs.get('step_3_fixed')}"
)
return f"全部 DAG 节点执行完毕！最终写入结果：{dependency_outputs.get('step_5')}"
def test_kahn_cycle_detection() -> None:
print("=== 实验一：测试 Kahn 拓扑算法的环路死锁拦截 ===")
cyclic_nodes = [
TaskNode(id="step_1", description="初始化", dependencies=[]),
TaskNode(id="step_2", description="任务 A", dependencies=["step_1", "step_4"]),
TaskNode(id="step_3", description="任务 B", dependencies=["step_2"]),
TaskNode(id="step_4", description="任务 C (反向依赖 B 形成环路!)", dependencies=["step_3"]),
]
try:
TaskDAG(cyclic_nodes)
except DAGValidationError as exc:
print(f"成功拦截非法环路: {exc}\n")
async def test_dag_execution_and_replan() -> None:
print("=== 实验二：测试 DAG 波次并发、参数占位符注入与动态重规划 ===")
engine = DAGPlanAndSolveEngine(planner=SimulatedDAGPlanner(), tool_executor=executor)
report = await engine.run("分析 scheduler 与 block_manager 源码及压测报告，输出优化文档。")
print("\n--- 拓扑波次 (Wavefront) 推进日志 ---")
for w in report.waves_log:
print(
f"Wave {w.wave_index}: 并发节点={w.node_ids} | "
f"波次墙钟耗时={w.wave_wall_time_ms:.1f}ms (节点累加耗时={w.sum_node_time_ms:.1f}ms) | "
f"触发重规划={w.triggered_replan}"
)
total_work, span, cp = report.dag.compute_work_and_span(use_actual_time=True)
print("\n--- DAG Work-Span 性能剖析 ---")
print(f"最终图节点状态 : { {nid: n.status.value for nid, n in report.dag.nodes.items()} }")
print(f"串行总工作量 T1: {total_work:.1f} ms")
print(f"关键路径 Span : {span:.1f} ms (关键路径序列: {' -> '.join(cp)})")
print(f"实际总墙钟耗时 : {report.total_wall_time_ms:.1f} ms (含 1 次故障重规划)")
print(f"最终汇总输出 : {report.final_answer}")
if __name__ == "__main__":
test_kahn_cycle_detection()
asyncio.run(test_dag_execution_and_replan())
我们来跑一下这段脚本，看看控制台打印出的完整调度过程：
=== 实验一：测试 Kahn 拓扑算法的环路死锁拦截 ===
成功拦截非法环路: Cyclic dependency detected in TaskDAG! Deadlocked nodes: ['step_2', 'step_3', 'step_4']
=== 实验二：测试 DAG 波次并发、参数占位符注入与动态重规划 ===
[Replanner 触发] 检测到节点 'step_3' 失败: [TOOL EXECUTION ERROR] Tool 'load_benchmark_json' failed after 1 attempt(s) with FileNotFoundError: File 'legacy_bench.json' no longer exists. Benchmark data was migrated to 'v2_bench.json'.
[Replanner 决策] 保留已完成节点 ['step_1', 'step_2']，重写剩余子图 -> 改读 'v2_bench.json'
--- 拓扑波次 (Wavefront) 推进日志 ---
Wave 1: 并发节点=['step_1', 'step_2', 'step_3'] | 波次墙钟耗时=201.4ms (节点累加耗时=602.1ms) | 触发重规划=True
Wave 2: 并发节点=['step_3_fixed'] | 波次墙钟耗时=200.8ms (节点累加耗时=200.6ms) | 触发重规划=False
Wave 3: 并发节点=['step_4'] | 波次墙钟耗时=100.5ms (节点累加耗时=100.4ms) | 触发重规划=False
Wave 4: 并发节点=['step_5'] | 波次墙钟耗时=100.6ms (节点累加耗时=100.5ms) | 触发重规划=False
--- DAG Work-Span 性能剖析 ---
最终图节点状态 : {'step_1': 'completed', 'step_2': 'completed', 'step_3_fixed': 'completed', 'step_4': 'completed', 'step_5': 'completed'}
串行总工作量 T1: 802.2 ms
关键路径 Span : 401.7 ms (关键路径序列: step_1 -> step_4 -> step_5)
实际总墙钟耗时 : 704.1 ms (含 1 次故障重规划)
最终汇总输出 : 全部 DAG 节点执行完毕！最终写入结果：Saved 153 chars to 'optimization_report.md' -> Content: Analysis: (Scheduler: supports continuous batching and PD disaggregation.) + (BlockManager: manages prefix caching via hash-indexed KV blocks.) + Metrics={"ttft_ms": 38.2, "prefix_cache_hit_rate": 0.91}
仔细看上面 Wave 1 和 Wave 2 的衔接，这就是图调度架构相比线性 ReAct 最迷人的地方：
在 Wave 1 中，调度器一眼看出
step_1、step_2、step_3入度全为 0，直接在一个波次内三路并发启动，累加耗时602.1ms的工作量只花了201.4ms就跑完了；当
step_3遭遇FileNotFoundError时，由于step_1和step_2已经成功完成并把源码摘要存进了节点状态里，Replanner介入后执行了_apply_subgraph_replan——step_1和step_2的成果被原封不动地保留了下来，只有坏掉的step_3被替换成了step_3_fixed；到了 Wave 3 和 Wave 4，
step_4自动拿到了step_1、step_2和新跑出来的step_3_fixed的输出完成汇总，随后step_5通过{{step_4.output}}模板插值，分毫不差地把汇总文本写进了目标文件！
6. 什么时候用 ReAct，什么时候用 DAG Plan-and-Solve？
写到这里，有的同学可能会问：“既然 DAG 调度这么强，那我们是不是应该把所有任务都用 DAGPlanAndSolveEngine 来跑，彻底抛弃 ReAct？”
在系统工程里，没有银弹，只有权衡（Trade-off）。我们可以用一个非常清晰的二维矩阵来决定两者的适用边界：
实际上，在最先进的生产级 Agent 架构中，这两者并不是二选一的对立关系，而是 分层嵌套关系：
宏观层（Macro-Level） 使用本篇的 DAG 规划调度器，将复杂任务拆解为几个阶段清晰、可并行的里程碑节点；
微观层（Micro-Level） 当某个复杂的 DAG 节点无法靠单次固定工具调用完成（例如该节点是“修复
scheduler.py中的单元测试报错”）时，该节点内部启动一个独立的 ReAct 子会话（Sub-Session） 去完成局部的探索闭环！
小结
到这里，我们圆满完成了第二阶段（结构化输出与工具调度引擎）的全部三篇内容。
回顾一下我们在这一篇里构建的核心模块：
建立了 DAG 任务调度的数学模型：定义了前驱集 P(v) 与动态未完成入度 d_rem(v, t)，用分层 Kahn 算法解决了大模型规划可能产生的循环依赖死锁问题，并通过 Work-Span（T_1 与 T_∞）动态规划量化了图调度的理论加速比。
实现了
TaskNode与TaskDAG容器（dag_schema.py）：支持{{node_id.output}}跨节点参数占位符自动插值、Kahn 拓扑波次划分、关键路径回溯以及失败级联跳过（Cascade Skip）。实现了支持原位子图热替换的
DAGPlanAndSolveEngine（dag_engine.py）：按拓扑波次并发发射所有就绪节点，并在节点失败时保留已完成节点的计算成果，仅对未完成子图执行局部重规划与热替换。
现在，我们的 Agent 已经有了强健的状态机、100% 可靠的结构化输出、高并发的异步执行器和全局 DAG 调度大脑。
但到目前为止，我们注册给 Agent 的工具还都只是温顺的 Python 小函数。从下一篇（第 7 篇）开始，我们将正式进入极具挑战性的第三阶段：《执行环境、安全沙箱与生态协议》。
当我们要赋予 Agent 最强大的终极武器——直接执行任意 Bash 命令与 Python 代码（Code Interpreter） 时，怎么防止它一句 rm -rf、一个 fork 炸弹或者一段死循环把宿主机搞垮？
在第 7 篇 《安全沙箱（Sandbox）：让 Agent 敢于执行代码与 Bash》 中，我们将发挥 Linux 系统编程的看家本领，从零实现一个支持 独立进程组信号斩杀（Process Group SIGKILL）、resource / ulimit 内存与 CPU 硬限制、只读挂载 / 路径逃逸拦截，以及标准输出/错误流防 OOM 环形截断捕获 的安全执行沙箱！
