执行引擎:拓扑调度、异步并发与失败传播

30 秒导读: 一张 DAG(节点 + 连线)交给 WorkflowExecutor,它给每个节点起一个 asyncio.Task,每个 task 先 await 自己所有上游节点的 task,再执行自己。没有任何显式的"拓扑排序"代码——"谁等谁"这件事完全靠 await 依赖任务天然表达,依赖一就绪就触发,能并行的自动并行。这是 PySpur 最硬核、也最优雅的一处。

本章聚焦三件事:调度算法(节点怎么被并发跑起来)、数据流(一个节点的输入怎么从上游拼出来)、失败模型(某个节点崩了/跳过了,下游怎么办)。暂停/恢复只点到为止,完整机制见 human-in-loop。任务落库的细节见 llm-and-agent 之外的 task_recorder 章节(本章只提调用点)。

主角只有一个文件:backend/pyspur/execution/workflow_executor.py 里的 WorkflowExecutor 类。

1. 这是什么(先建直觉)

一个工作流就是一张有向无环图(DAG,Directed Acyclic Graph):节点是"要干的活"(调 LLM、路由、输出……),连线(link)是"数据从谁流到谁"。数据模型怎么描述这张图,见 workflow-model。

执行引擎要回答的核心问题只有一个:

给定这张图,按什么顺序、用多大并发把每个节点跑一遍,并把上游的输出正确喂给下游?

传统写法是先做拓扑排序(把节点排成"没有上游的排前面"的线性序),再一层层跑。PySpur 偏偏不这么做。它的思路更像"承诺(promise)链":

每个节点 = 一个异步任务(asyncio.Task)。
一个任务开跑第一件事,是 await 它所有上游任务。
于是"等谁先跑完"这个约束,被 await 隐式地编码进了任务之间——你不需要自己排序,asyncio 事件循环替你调度。

一句话直觉:把每个节点想成一个"我准备好就自动干活"的小机器人,它只盯着自己的上游;所有机器人同时启动,能动的先动,DAG 的执行顺序就自然浮现了。

2. 顶层全景(一次 run 怎么转)

2.1 三个阶段

从外部调用到拿回结果,WorkflowExecutor 走三步:

                    ┌─────────────────────────────────────────────┐
   构造期            │  __init__                                    │
  (同步、建索引)     │   _process_subworkflows  折叠子工作流          │
                    │   _build_node_dict       id → 节点            │
                    │   _build_dependencies    每个节点的上游集合     │
                    └───────────────────┬─────────────────────────┘
                                        │
                    ┌───────────────────▼─────────────────────────┐
   调度期            │  run / __call__ → _execute_workflow          │
  (异步、并发)       │   为每个节点 create_task(_execute_node)       │
                    │   gather 全部任务, return_exceptions=True     │
                    └───────────────────┬─────────────────────────┘
                                        │
                    ┌───────────────────▼─────────────────────────┐
   单节点            │  _execute_node(node_id)                      │
  (每节点一个)       │   await gather(上游任务)  ← DAG 就靠这行串起   │
                    │   组装 node_input → NodeFactory 造节点 → 执行  │
                    └─────────────────────────────────────────────┘

2.2 部件一句话职责

部件(符号)	干什么	位置
`__init__`	归一化输入、初始化状态、建两张索引表	`workflow_executor.py:39-69`
`_process_subworkflows`	按 `parent_id` 把子节点折叠进父节点 config	`workflow_executor.py:81-135`
`_build_node_dict`	`node.id → WorkflowNodeSchema` 的字典	`workflow_executor.py:137-138`
`_build_dependencies`	把 links 反向建成"每个节点的上游集合"	`workflow_executor.py:140-144`
`_execute_workflow`	调度核心:为每节点起 task、gather、收尾	`workflow_executor.py:729-869`
`_get_async_task_for_node_execution`	缓存每个节点的 task(保证只跑一次)	`workflow_executor.py:160-172`
`_execute_node`	单节点全流程:等上游 → 拼输入 → 执行	`workflow_executor.py:426-706`
`run` / `__call__`	对外入口;chatbot 消息历史注入/落库	`workflow_executor.py:871-932`
`run_batch`	批量跑多份输入	`workflow_executor.py:934-947`

后面按"构造期 → 调度期 → 数据流 → 失败模型 → 对外入口"逐层展开。

3. 构造期:把图预处理成可调度的索引

__init__ 是同步的,只做"准备工作":归一化 workflow、初始化一堆状态容器,最后建两张索引。

3.1 状态容器

__init__ 里初始化的关键字段(workflow_executor.py:61-67):

字段	类型	作用
`_node_dict`	`Dict[str, WorkflowNodeSchema]`	id → 节点定义
`_dependencies`	`Dict[str, Set[str]]`	id → 上游 id 集合
`_node_tasks`	`Dict[str, asyncio.Task]`	id → 该节点的执行任务(去重缓存)
`_outputs`	`Dict[str, Optional[BaseNodeOutput]]`	id → 输出(也当"已算过"的备忘录)
`_failed_nodes`	`Set[str]`	已失败/被跳过的节点集合

3.2 `_build_dependencies`:把连线"反向"

link 记的是"从 source 流向 target"(正向)。但调度时每个节点关心的是"我要等谁"——也就是它的上游集合。所以这一步把 link 反向聚合:

# workflow_executor.py:140-144  _build_dependencies
dependencies = {node.id: set() for node in self.workflow.nodes}
for link in self.workflow.links:
    dependencies[link.target_id].add(link.source_id)  # target 依赖 source
self._dependencies = dependencies

结果:_dependencies[某节点] = 它所有上游节点的 id 集合。整个调度算法就靠这张表——_execute_node 里 await 的就是这些上游。

注意:这里没有做拓扑排序,也没有检测环。图无环是数据模型层的约定(见 workflow-model),执行引擎默认它已经是 DAG。

3.3 `_process_subworkflows`:把子节点折叠进父节点

PySpur 支持嵌套(某个节点内部本身是一张子工作流,如循环节点)。数据里用 node.parent_id 表达"我属于哪个父节点"。执行引擎不在顶层图里直接跑这些子节点,而是在构造期把它们折叠进父节点的 config:

折叠前(扁平存储)                     折叠后(交给调度器)
──────────────────                    ──────────────────
parent (parent_id=None)               parent
  ├ child_a (parent_id=parent)  ──►     └ config.subworkflow = {
  └ child_b (parent_id=parent)              nodes: [child_a, child_b],
child_a → child_b (link)                    links: [child_a→child_b]
                                          }

关键三步(workflow_executor.py:81-135):

按 parent_id 把节点分组(nodes_by_parent),parent_id=None 的是根节点。
对每个父节点,收集两端都在子节点集合内的 links(workflow_executor.py:106-110),打包成一个 WorkflowDefinitionSchema,塞进父节点 config["subworkflow"](workflow_executor.py:116-119)。
返回只含根节点的新 workflow,顶层 links 里也剔掉任何一端带 parent_id 的连线(workflow_executor.py:122-135)。

于是调度器眼里的图永远是"扁平的一层根节点";嵌套由父节点自己在执行时展开(父节点如何跑子工作流属于节点体系,见 node-system)。

只有传入的是 WorkflowDefinitionSchema 时才折叠;若传入的是 WorkflowModel,直接 model_validate 它的 definition,不折叠(workflow_executor.py:48-51)——因为存库的 definition 已是处理过的形态。

4. 调度核心:没有拓扑排序的隐式拓扑并发

这是本章的心脏。理解了这一节,就理解了 PySpur 的执行引擎。

4.1 思路:用 await 表达"等谁"

先看简化示意,把核心想法抽出来(去掉了失败/暂停/落库):

# 示意,非源码 —— 隐式拓扑调度的骨架
tasks = {}  # node_id -> asyncio.Task,做缓存

def task_for(node_id):
    if node_id not in tasks:                       # 已建过就复用,保证每节点只跑一次
        tasks[node_id] = asyncio.create_task(run_node(node_id))
    return tasks[node_id]

async def run_node(node_id):
    deps = dependencies[node_id]                   # 我的上游
    upstream = await asyncio.gather(*(task_for(d) for d in deps))  # 关键:等上游
    node_input = assemble(deps, upstream)          # 用上游输出拼我的输入
    return await create_node(node_id)(node_input)  # 执行我自己

# 启动:给每个节点都建 task,然后一起等
for nid in all_nodes:
    task_for(nid)
await asyncio.gather(*tasks.values())

重点看两行:

await asyncio.gather(*(task_for(d) ...))——一个节点开跑先等上游,这就是"依赖就绪即触发"。
task_for 的缓存——多个下游共享同一个上游时,上游只跑一次,大家 await 同一个 task。

没有排序、没有队列、没有"哪个节点该轮到"的调度循环。asyncio 事件循环负责真正的调度;DAG 的顺序是 await 依赖关系涌现出来的。

4.2 缓存:`_get_async_task_for_node_execution`

去重缓存是这套机制成立的关键——否则同一个上游被多个下游各起一次 task 会重复执行:

# workflow_executor.py:160-172
def _get_async_task_for_node_execution(self, node_id):
    if node_id in self._node_tasks:
        return self._node_tasks[node_id]        # 命中缓存,复用同一个 Task
    task = asyncio.create_task(self._execute_node(node_id))
    self._node_tasks[node_id] = task
    if self.task_recorder:
        self.task_recorder.create_task(node_id, {})  # 落库:记一条 task(细节见 05)
    return task

4.3 启动:`_execute_workflow`

_execute_workflow 是调度的外层驱动(workflow_executor.py:729-869),核心几步:

# workflow_executor.py:797-802(节选)
for node_id in nodes_to_run:
    self._get_async_task_for_node_execution(node_id)   # 给每个节点都起 task

# 等全部,但不让异常冒泡(下面自己处理)
results = await asyncio.gather(*self._node_tasks.values(), return_exceptions=True)

要点:

给"所有"节点都起 task(workflow_executor.py:798-799),不是只起源头。因为每个 task 内部会自己 await 上游,所以"全起"不会乱序——上游没好,下游 task 就卡在 await 上。
return_exceptions=True(workflow_executor.py:802):某个节点抛异常不中断整个 gather,异常被当成结果收集起来,由后续逻辑逐个判定(失败?暂停?见 §6)。
InputNode 特殊处理:先把外部 input 存进 _initial_inputs,并同步先算出 InputNode 的输出(workflow_executor.py:767-782),让下游有起点。
带 parent_id 的节点从 nodes_to_run 里剔除(workflow_executor.py:789-791),它们由父节点执行。

4.4 一个例子:菱形 DAG 怎么跑

        ┌──► B ──┐
   A ───┤        ├──► D
        └──► C ──┘

_dependencies:B={A}, C={A}, D={B,C}, A={}。

执行时:所有节点的 task 一起 create。B 和 C 都 await task_for(A)——它们共享同一个 A 的 task(缓存),A 只跑一次;A 一完成,B、C 并发开跑;D 在 await gather(task_B, task_C),两者都好了才动。整张图的并发结构,不写一行调度逻辑就自动成型。

5. 输入组装:上游输出怎么变成本节点的输入

节点跑之前,_execute_node 要把上游输出拼成一个字典 node_input 喂进去。键用上游节点的 title,值是上游输出。不同类型的上游走不同分支。

5.1 通用规则:按 `predecessor.title` 建键

遍历每个上游,默认直接用上游的 title 当键(workflow_executor.py:536-575):

# workflow_executor.py:574-575(默认分支)
else:
    node_input[predecessor_node.title] = output

这就是为什么工作流里的引用长成 {{某节点title.字段名}}——title 是数据流的命名空间。

5.2 三个特殊上游

上游类型	特殊处理	位置
`RouterNode`	按 link 的 `source_handle` 取某个分支的输出	`workflow_executor.py:540-559`
`HumanInterventionNode`	`model_dump()` 展平成 dict 再塞入	`workflow_executor.py:560-573`
`InputNode`(本节点是它)	忽略上游,直接用 `_initial_inputs`	`workflow_executor.py:578-579`

RouterNode(路由分支): 路由节点有多个输出分支,link 上的 source_handle 指明"这条连线走哪个分支"。引擎先建一张 (source_id, target_id) → source_handle 的表(_get_source_handles,workflow_executor.py:146-158,只对 RouterNode 建),然后 getattr(output, source_handle) 取出对应分支输出:

# workflow_executor.py:542-559(节选)
source_handle = source_handles.get((dep_id, node_id))
if not source_handle:
    raise ValueError(...)                       # 路由连线必须带 handle
route_output = getattr(output, source_handle, None)
if route_output is not None:
    node_input[predecessor_node.title] = route_output
else:
    self._outputs[node_id] = None               # 这条分支没走 → 本节点作废
    ...
    return None

关键语义:路由没选中的那条下游,拿到的分支输出是 None,于是本节点直接返回 None(不执行)。 这就是分支"择一而走"的实现方式——没被选中的支路整条静默取消。

HumanInterventionNode(人工介入): 把输出 model_dump() 展平成 dict 再放进 node_input(workflow_executor.py:562-573),让下游能用 {{HumanInterventionNode_1.input_node.input_1}} 这种路径访问原始结构。

InputNode: 本节点若是 InputNode,直接用外部初始输入覆盖(workflow_executor.py:578-579):node_input = self._initial_inputs.get(node_id, {})。

5.3 组装完的清理

拼完后过滤掉 None 值:node_input = {k: v for k,v in node_input.items() if v is not None}(workflow_executor.py:590)。
若 node_input 最终为空 → 抛 UnconnectedNodeError(workflow_executor.py:605-607),表示这节点没接上任何有效输入。

组装完成后才 NodeFactory.create_node(...) 造出真正的节点实例并 await node_instance(node_input) 执行(workflow_executor.py:609-628)。节点实例化与工厂机制见 node-system。

6. 失败与跳过语义:一个节点崩了,下游怎么办

这是引擎的第二个硬核处。核心是两个自定义异常 + _failed_nodes 集合的传播。

6.1 两个异常

异常	含义	定义
`UpstreamFailureError`	上游失败,本节点被跳过	`workflow_executor.py:28-29`
`UnconnectedNodeError`	本节点没有任何有效输入	`workflow_executor.py:32-33`

6.2 失败传播:靠 `await` 天然扩散

_execute_node 里 await 上游时,上游若抛异常,asyncio.gather 会把异常冒上来,本节点立刻转成 UpstreamFailureError:

# workflow_executor.py:466-476
try:
    predecessor_outputs = await asyncio.gather(
        *(self._get_async_task_for_node_execution(d) for d in dependency_ids)
    )
except Exception as e:
    raise UpstreamFailureError(f"Node {node_id} skipped due to upstream failure") from e

紧接着还有一道显式检查——即便上游没抛异常但被记进了 _failed_nodes,本节点也跳过并把自己也加进 _failed_nodes(workflow_executor.py:478-481):

if any(dep_id in self._failed_nodes for dep_id in dependency_ids):
    self._failed_nodes.add(node_id)             # 失败沿 DAG 顺流而下
    raise UpstreamFailureError(...)

于是"失败"沿着依赖链一路顺流传播:一个节点炸了,它的整条下游全部被标记跳过。捕获处按情况把 task 记为 FAILED(真炸)或 CANCELED(被上游拖累),见 workflow_executor.py:687-706 和 649-686。

6.3 `None` 输入即取消

PySpur 用 None 表示"这路数据没了/这节点没跑"。默认语义:任何一个上游输出是 None,本节点就取消(返回 None,记 CANCELED):

# workflow_executor.py:504-507
if node.node_type != "CoalesceNode" and any(
    output is None for output in predecessor_outputs
):
    self._outputs[node_id] = None
    ...   # 记 CANCELED(除非上游是"暂停"导致的,则记 PENDING)
    return None

组装后还有一道:非 CoalesceNode 时,node_input 里任一值为 None 也直接作废(workflow_executor.py:582-584)。这套 None 语义正是 §5.2 里"路由没选中的支路整条取消"能生效的底层原因。

6.4 `CoalesceNode` 的特殊 `None` 处理

上面每处判断都写着 if node.node_type != "CoalesceNode"。Coalesce(合流)节点存在的意义就是打破默认语义——它要在多条互斥支路里挑出"那条真正跑通的",所以规则反过来:

	默认节点	CoalesceNode
触发取消的条件	任一输入为 `None`(`any`)	全部输入为 `None`(`all`)
代码	`workflow_executor.py:582-584`	`workflow_executor.py:585-587`

# workflow_executor.py:585-587
elif node.node_type == "CoalesceNode" and all(v is None for v in node_input.values()):
    self._outputs[node_id] = None
    return None

直觉:路由把数据分到 A/B/C 三条互斥支路,只有一条有真值、另两条是 None。默认节点见 None 就死;Coalesce 却能容忍——只要还有一条非 None 就照跑,从而把分叉重新合并成一股。它是分支择一之后的"汇流阀"。

6.5 失败/跳过的全景

上游 task 抛异常 ──► gather 冒泡 ──► 本节点 raise UpstreamFailureError
                                          │
上游 ∈ _failed_nodes ─────────────────────┤ (显式再查一道)
                                          ▼
                            本节点加入 _failed_nodes,继续向下游传播
                                          │
上游输出 == None ──► any(None) 命中 ──► 本节点返回 None,记 CANCELED
   (CoalesceNode 例外:只有 all(None) 才取消)

暂停(PauseError)走的是另一条平行的通道:它不进 _failed_nodes,下游被标 PENDING 而非 CANCELED,等待恢复。这条通道贯穿了 _execute_node(workflow_executor.py:644-647)和 _execute_workflow(workflow_executor.py:805-861)的大量分支,但完整状态机留给 human-in-loop。

7. 对外入口:run / run_batch / call

7.1 `run`:普通工作流 + chatbot 的消息历史

run 包在 _execute_workflow 外面(workflow_executor.py:871-917),对普通工作流它几乎是透传;对 chatbot 类工作流(spur_type == SpurType.CHATBOT)额外做消息历史的注入与落库:

run(input)
  │  spur_type == CHATBOT ?
  ├─ 是:注入历史 ── 若 input 没带 message_history,
  │                _get_message_history(session_id) 从库里查出注入 input
  │
  ├──────► _execute_workflow(...)  ← 真正执行
  │
  └─ 是:落库 ── 从 OutputNode 输出取 assistant_message,
                _store_message_history(session_id, user_message, assistant_message)

注入:_get_message_history 查 SessionModel 的 messages,把历史塞进 input["message_history"](workflow_executor.py:249-272、878-889)。
落库:执行完从 OutputNode 输出里取 assistant_message,连同用户消息一起 _store_message_history 写两条 MessageModel(user + assistant)并 commit(workflow_executor.py:274-296、895-915)。

普通(WORKFLOW 类型)工作流完全不碰这两段——if self.workflow.spur_type == SpurType.CHATBOT 直接跳过。

7.2 `call` 与 `run_batch`

__call__(workflow_executor.py:919-932):就是 await self.run(...) 的糖,让 executor 可当函数直接调用(见文件末尾 __main__ 的 asyncio.run(executor(input)) 用例,workflow_executor.py:1025)。
run_batch(workflow_executor.py:934-947):对一批输入并发跑 run,按 batch_size(默认 100)分批 gather。注意它对每份输入复用同一个 executor 实例——_outputs 等状态在 batch 间是共享的,_execute_workflow 开头会 pop 掉待跑节点的旧输出来复位(workflow_executor.py:793-795)。

8. 巧妙之处(可带走的精华)

用 await 代替拓扑排序。 整个调度没有排序、没有 ready 队列;"谁等谁"直接由 await asyncio.gather(上游 task) 表达,asyncio 事件循环当调度器。代码量极小,并发天然最大化(workflow_executor.py:466-472)。
task 缓存 = 天然去重执行。 _get_async_task_for_node_execution 保证每个节点只有一个 Task,菱形/扇出结构里共享上游只跑一次(workflow_executor.py:160-172)。
None 作为一等公民的"没数据"信号。 分支取消、路由择一、上游跳过,全部统一编码成"输出 None",下游 any(None) 即取消——一个约定串起整套控制流(workflow_executor.py:504-507、582-584)。
CoalesceNode 反转 None 规则实现汇流。 把 any 改成 all,让"多条互斥支路合一"这件事只需一个节点类型 + 引擎里几处 != "CoalesceNode" 判断(workflow_executor.py:585-587)。
构造期把嵌套折叠成扁平。 调度器永远只看一层根节点,子工作流被藏进父节点 config,大幅简化调度逻辑(workflow_executor.py:81-135)。

9. 边界与局限(诚实说)

不检测环。 引擎假定输入已是 DAG;若存在环,两个节点会互相 await 对方的 task 而死锁(代码里看不到任何环检测)。无环由数据模型层保证(见 workflow-model)。
_execute_workflow 假定有且仅有一个根 InputNode。 next(...) 取第一个无 parent_id 的 InputNode(workflow_executor.py:768-774),没有则会抛 StopIteration。
循环节点的 precomputed 输出不支持。 若某节点的预计算输出是 list(loop 节点),直接跳过不校验(workflow_executor.py:745-749)。
run_batch 复用实例、状态共享。 同一 executor 跨输入共享 _outputs/_node_tasks,靠开头 pop 复位;并非每份输入一个干净实例。
失败即传播,无重试。 一个节点异常,整条下游全 CANCELED,引擎本身不做重试或降级(重试若有,属节点层)。
暂停/恢复的完整逻辑不在本章。 _handle_pause_exception、get_blocked_nodes、is_downstream_of_pause、_fix_canceled_tasks_after_pause 等只在此列出调用点,状态机见 human-in-loop。

10. 代码地图(导航索引)

主题	文件	符号
执行引擎主类	`backend/pyspur/execution/workflow_executor.py`	`WorkflowExecutor`
上游集合(反向建依赖)	`backend/pyspur/execution/workflow_executor.py`	`_build_dependencies`
id → 节点索引	`backend/pyspur/execution/workflow_executor.py`	`_build_node_dict`
子工作流折叠	`backend/pyspur/execution/workflow_executor.py`	`_process_subworkflows`
调度核心(起 task + gather)	`backend/pyspur/execution/workflow_executor.py`	`_execute_workflow`
task 去重缓存	`backend/pyspur/execution/workflow_executor.py`	`_get_async_task_for_node_execution`
单节点全流程(等上游+拼输入+执行)	`backend/pyspur/execution/workflow_executor.py`	`_execute_node`
路由分支 handle 映射	`backend/pyspur/execution/workflow_executor.py`	`_get_source_handles`
上游失败异常	`backend/pyspur/execution/workflow_executor.py`	`UpstreamFailureError`
无输入异常	`backend/pyspur/execution/workflow_executor.py`	`UnconnectedNodeError`
对外入口 + chatbot 历史	`backend/pyspur/execution/workflow_executor.py`	`run`
消息历史读/写	`backend/pyspur/execution/workflow_executor.py`	`_get_message_history` / `_store_message_history`
批量执行	`backend/pyspur/execution/workflow_executor.py`	`run_batch`
link/node 数据结构	`backend/pyspur/schemas/workflow_schemas.py`	`WorkflowLinkSchema` / `WorkflowNodeSchema`
暂停异常(来自节点)	`backend/pyspur/nodes/logic/human_intervention.py`	`PauseError`
任务落库	`backend/pyspur/execution/task_recorder.py`	`TaskRecorder`

相关章节: index · workflow-model · node-system · human-in-loop · llm-and-agent

1. 这是什么(先建直觉)​

2. 顶层全景(一次 run 怎么转)​

2.1 三个阶段​

2.2 部件一句话职责​

3. 构造期:把图预处理成可调度的索引​

3.1 状态容器​

3.2 _build_dependencies:把连线"反向"​

3.3 _process_subworkflows:把子节点折叠进父节点​

4. 调度核心:没有拓扑排序的隐式拓扑并发​

4.1 思路:用 await 表达"等谁"​

4.2 缓存:_get_async_task_for_node_execution​

4.3 启动:_execute_workflow​

4.4 一个例子:菱形 DAG 怎么跑​

5. 输入组装:上游输出怎么变成本节点的输入​

5.1 通用规则:按 predecessor.title 建键​

5.2 三个特殊上游​

5.3 组装完的清理​

6. 失败与跳过语义:一个节点崩了,下游怎么办​

6.1 两个异常​

6.2 失败传播:靠 await 天然扩散​

6.3 None 输入即取消​

6.4 CoalesceNode 的特殊 None 处理​

6.5 失败/跳过的全景​

7. 对外入口:run / run_batch / call​

7.1 run:普通工作流 + chatbot 的消息历史​

7.2 __call__ 与 run_batch​

8. 巧妙之处(可带走的精华)​

9. 边界与局限(诚实说)​

10. 代码地图(导航索引)​