graph-core 执行流程与 Flux 订阅机制

读 graph-core 源码时，一个反直觉的地方是：调用入口 app.invoke(inputs) 看起来是同步执行，但底层真实执行模型是流式的。invoke 本质是 stream(...).last().map(NodeOutput::state).block()——把同步调用包装在流式执行之上。

这意味着整个图的执行是用 Flux 把每个节点输出串成一条响应式流，而不是一次性递归算完再返回；invoke 只是订阅这条流并取最后状态。理解这条链路，是理解 graph-core 执行机制的关键。这里把从整体执行流程到 Flux 订阅机制的内容整理下来。

整体执行流程

主线流程：

graph LR
    A[定义图] --> B[编译图]
    B --> C[执行图]
    C --> D[节点返回局部状态]
    D --> E[框架合并到全局状态]
    E --> F[根据边选择下一节点]
    F --> G[最终返回 OverAllState]

精确拆开是八步：

1. 定义 StateGraph
   - 注册 node 节点
   - 注册 edge / conditional edge
   - 注册 KeyStrategy 合并策略

2. compile()
   - 校验图合法性
   - 固化节点、边、配置
   - 生成 CompiledGraph

3. invoke(initialState)
   - invoke 本质是 stream(...).last().map(NodeOutput::state).block()
   - 所以底层真实执行模型是流式执行

4. GraphRunner.run()
   - 创建 GraphRunnerContext
   - context 持有 compiledGraph、config、overallState、currentNodeId、nextNodeId

5. MainGraphExecutor.execute()
   - 处理 START / END / interrupt / resume / stop
   - 普通节点交给 NodeExecutor

6. NodeExecutor.executeNode()
   - 根据 nextNodeId 找到当前 node action
   - 执行 action.apply(overallState, config)
   - action 返回局部状态 Map<String, Object>

7. 状态合并与路由
   - mergeIntoCurrentState(updateState)
   - OverAllState.updateState(...) 按 KeyStrategy 合并
   - nextNodeId(...) 根据固定边或条件边选下一节点
   - 构造 NodeOutput
   - 回到 MainGraphExecutor 继续推进

8. 最终结果
   - 到 END 后流结束
   - invoke 取最后一个 NodeOutput.state
   - 返回 Optional<OverAllState>

各组件职责分工：

StateGraph：负责定义图
CompiledGraph：承载编译后的图和执行入口门面
GraphRunner / Executor：负责推进执行
OverAllState：保存全局状态
KeyStrategy：决定局部状态如何合并

CompiledGraph 不是执行器本体

一个容易混淆的点：CompiledGraph 看起来像执行引擎，因为调用入口是 app.invoke(inputs)，但它更像编译后的图模型加上执行入口门面。真正推进执行的是：

GraphRunner
-> MainGraphExecutor
-> NodeExecutor

CompiledGraph.invoke(...) 只是把同步调用包装在流式执行之上：

Optional.ofNullable(stream(inputs, config).last().map(NodeOutput::state).block());

拆开看：

invoke = stream + last + state + block

NodeExecutor 怎么流转回 MainGraphExecutor

执行链的核心在 NodeExecutor 执行完当前节点后，把控制权交回 MainGraphExecutor 的这段代码：

return Flux.just(GraphResponse.of(output))
    .concatWith(Flux.defer(() -> mainGraphExecutor.execute(context, resultValue)));

这段代码表达四件事：

先发出当前节点的 GraphResponse
当前节点输出发完后
再调用 mainGraphExecutor.execute(context, resultValue)
MainGraphExecutor 根据 context.nextNodeId 决定下一步

NodeExecutor 不自己用 while 直接跑完整张图，它只负责当前节点：

graph LR
    A[执行当前 node action] --> B[合并局部状态]
    B --> C[计算 nextNodeId]
    C --> D[构造当前 NodeOutput]
    D --> E[把控制权交回 MainGraphExecutor]

MainGraphExecutor 继续判断：是否 START、是否 END、是否中断、是否恢复、是否继续执行普通节点。两者共同构成执行循环。要理解这个循环怎么在响应式流里实现，需要先看几个 Reactor 概念。

concatWith：顺序拼接

简化例子：

Flux.just("A")
    .concatWith(Flux.just("B"));

concatWith 是顺序拼接：先订阅并消费左边的 Flux，左边发完并 onComplete 后，再订阅并消费右边的 Flux。输出顺序一定是 A 然后 B。

放回 graph-core：

Flux.just(GraphResponse.of(output))
    .concatWith(...下一轮执行...)

意思是先发当前节点 output，当前 output 的 Flux 完成后，再进入下一轮执行。

Flux.defer：延迟创建

Flux.defer 的作用是延迟创建 Flux：

Flux.defer(() -> Flux.just("B"))

现在先不创建 B 这个 Flux，等真正被订阅时，再执行 Supplier，创建 Flux.just("B")。

graph-core 这里需要 defer 的原因，是下一轮执行依赖当前节点执行后的 context：

currentNodeId 已更新
nextNodeId 已更新
overallState 已合并

不用 defer，下一轮 mainGraphExecutor.execute(...) 可能会在构造流时过早执行。用了 defer 后，语义变成：等当前节点 GraphResponse 发完，concatWith 开始订阅右侧 Flux，此时 defer 才调用 mainGraphExecutor.execute(...)，下一轮使用的是最新 context。

两条核心记忆：

concatWith 保证顺序：左边 onComplete 后才订阅右边
defer 保证下一轮晚点创建：被订阅时才执行 Supplier

订阅触发执行

响应式编程里一个核心概念：Flux 本身只是数据流水线的描述，数据并不会自动流动。单独写这段代码什么都不会输出：

Flux.just("A")
    .concatWith(Flux.defer(() -> Flux.just("B")));

它只是描述了一条流水线——以后如果有人订阅我：先发 A，A 完成后再创建并发 B。真正触发执行的是终端操作：

.subscribe(System.out::println)
.block()
.collectList().block()

响应式流有三个基础信号：

onNext(value)：发出一个数据
onComplete()：正常结束
onError(error)：异常结束

concatWith 依赖的就是 onComplete()：左边 Flux 发出 onComplete 后，右边 Flux 才会被订阅。

用 A/B 例子串起来

Flux.just("A")
    .concatWith(Flux.defer(() -> Flux.just("B")))
    .subscribe(System.out::println);

执行过程拆成 14 步：

1.  subscribe 订阅整个 concat Flux
2.  concat Flux 先订阅左边 Flux.just("A")
3.  左边发出 onNext("A")
4.  println 收到 A
5.  左边发出 onComplete
6.  concat Flux 收到左边完成信号
7.  concat Flux 订阅右边 Flux.defer(...)
8.  defer 此时才调用 Supplier
9.  Supplier 返回 Flux.just("B")
10. concat Flux 订阅 B Flux
11. B Flux 发出 onNext("B")
12. println 收到 B
13. B Flux 发出 onComplete
14. 整条 concat Flux 完成

「等 A 被订阅完成」更准确的说法是：等左边 Flux 被订阅，并且发完 A 后发出 onComplete，右边 Flux 才会被订阅。

block 如何触发 graph-core 执行

回到 invoke：

Optional.ofNullable(
    stream(inputs, config)
        .last()
        .map(NodeOutput::state)
        .block()
);

这里的 block() 会订阅上游整个 Flux<NodeOutput>。上游链路：

graph TB
    S1[stream inputs, config] --> S2[streamFromInitialNode]
    S2 --> S3[runner.run overAllState]
    S3 --> S4[MainGraphExecutor.execute]
    S4 --> S5[NodeExecutor.execute]
    S5 --> S6[Flux.just GraphResponse.of output]
    S6 --> S7[concatWith defer 下一轮]
    S7 --> S8[flattenGraphResponsesPreservingOrder]
    S8 --> S9[Flux NodeOutput]
    S9 --> S10[last]
    S10 --> S11[map NodeOutput::state]
    S11 --> S12[block 返回结果]

一个类型转换细节：NodeExecutor 发出的是 Flux<GraphResponse<NodeOutput>>，而 CompiledGraph.streamFromInitialNode 会通过 flattenGraphResponsesPreservingOrder(...) 把它压平成 Flux<NodeOutput>。所以完整链路是：

NodeExecutor 发出 GraphResponse.of(output)
-> GraphRunner.run 返回 Flux<GraphResponse<NodeOutput>>
-> flatten 解出 NodeOutput
-> stream(...) 得到 Flux<NodeOutput>
-> last() 等最终节点
-> map(NodeOutput::state) 取最终 OverAllState
-> block() 返回结果

last 和 map 的关键细节

stream(inputs, config)
    .last()
    .map(NodeOutput::state)
    .block()

map(NodeOutput::state) 不会对每个节点都执行，因为前面有 last()。真实行为：

每个节点的 NodeOutput 都会往下游发
last() 持续接收，但只保留最后一个
等整个 Flux complete 后
last() 才把最后一个 NodeOutput 发给 map
map 只对最后一个 NodeOutput 执行
block 拿到最终 state

图是 START -> A -> B -> END 时，执行过程可能发出：

NodeOutput(START)
NodeOutput(A)
NodeOutput(B)
NodeOutput(END)

last() 的行为：

sequenceDiagram
    participant Flux as 上游 Flux
    participant Last as last()
    participant Map as map(NodeOutput::state)
    
    Flux->>Last: onNext(NodeOutput(START))
    Note over Last: 暂存为 last
    Flux->>Last: onNext(NodeOutput(A))
    Note over Last: 覆盖 last
    Flux->>Last: onNext(NodeOutput(B))
    Note over Last: 覆盖 last
    Flux->>Last: onNext(NodeOutput(END))
    Note over Last: 覆盖 last
    Flux->>Last: onComplete()
    Last->>Map: 发出 NodeOutput(END)
    Note over Map: 只对最后一个执行
    Map->>Map: 取出 OverAllState

然后 map(NodeOutput::state) 只对最后的 NodeOutput(END) 执行，取出里面的 OverAllState。

心智模型

把 graph-core 的执行和 Reactor 放在一起看：

block()：触发订阅整条执行链
MainGraphExecutor：负责判断当前阶段，是 START、END、中断，还是普通节点
NodeExecutor：负责执行当前节点 action，合并状态，计算 nextNodeId
Flux.just(output)：把当前节点输出发出去
concatWith(defer(…))：把下一轮执行按顺序挂在当前输出之后
last()：忽略中间节点，只保留最后一个 NodeOutput
map(NodeOutput::state)：取出最终全局状态 OverAllState

核心设计：graph-core 用 Flux 把每个节点输出串成一条响应式执行流，而不是一次性递归算完再返回；invoke 只是订阅这条流并取最后状态。这种设计让流式输出、背压处理、节点级中断恢复都建立在同一套响应式机制上——执行链本身就是流，中断和恢复只是流的暂停和继续。

整体执行流程#

CompiledGraph 不是执行器本体#

NodeExecutor 怎么流转回 MainGraphExecutor#

concatWith：顺序拼接#

Flux.defer：延迟创建#

订阅触发执行#

用 A/B 例子串起来#

block 如何触发 graph-core 执行#

last 和 map 的关键细节#

心智模型#