当前位置：首页 > news >正文

ai-agent框架spring ai/alibaba 原理源码分析(五)graph III 图执行

news 2026/7/18 4:19:25

图执行

上文分析了图编译，编译后图具备高效执行的数据结构，本文分析图执行，图执行可分两部分，主图执行，MainGraphExecutor负责，主图执行主要是图执行控制，包括START， END，中断；点执行，NodeExecutor负责节点处理。

图执行使用响应式流，如下图

可以认为，图遍历的过程是Flux(数据源)连接成响应式管道并加上处理的过程，Flux是节点Action执行的返回。做个类比，Flux链好比流水线，输送数据(材料)，数据处理类比员工，材料到工位加工处理，流水线的开关类比Flux的订阅

这里看出响应式编程的特点，先构建好数据流动管道，加上相应的处理，开关打开(调用block方法)，数据发射触发处理，实现异步无等待的处理

检查点

本节首先介绍一下检查点，检查点是图的执行期重要的组件，负责保存执行环境，用作中断/恢复，回溯等关键功能，也是图执行容错机制的支撑，也是agent的短期记忆组件。

上图检查点的类图

CheckPoint检查点保存着图执行的上下文，用于中断恢复，容错，回放等

id：checkpoint id，

state：全局状态，OverAllState

nodeId/nextNodeId 当前执行的nodeId，下一个执行nodeId

检查点整体存储类似文件夹，文件夹id就是threadId(会话id)，通常外部传入，checkpoints按时间倒序list存储

BaseCheckPointSaver检查点存储接口，有多种实现，redis，memory，mysql等

执行类图

下面分析参与执行的类图

上图是参与执行的类图，关于图的数据结构，参看系列-图I和系列-图II分析了图结构，编译后的数据结构

GraphRunner执行入口，负责GraphRunnerContext构建和初始，然后调用MainGraphExecutor执行图

GraphRunnerContext运行的上下文，包括全局的状态OverAllState，运行配置RunnableConfig，运行标的CompiledGraph；负责识别是新执行还是恢复执行；还有2个重要的职责，nextNodeId，计算遍历的下一个节点，addCheckpoint记录检查点

MainGraphExecutor/NodeExecutor执行器，前者负责主图执行，后者负责节点执行，主图执行重点是START节点，END节点，中断等，节点执行交给NodeExecutor，节点执行完，返回MainGraphExecutor，这样交替推进图的执行

RunnableConfig运行时配置，包含执行所需的参数设置，其中关键属性：threadId, checkPointId, nextNode

nodeFactories/Node.ActionFactory/AsyncNodeActionWithConfig nodeFactories是图的节点集，在系列-编译图介绍过。

Node.ActionFactory构建节点Action，类型是AsyncNodeActionWithConfig

GraphResponse图执行返回结构，包括NodeOutput，returnValue属性，前者包含全局状态，后者可以承载多种类型，主要用于中断metadata，也有其他用途，有一点值得注意：中断metadata是继承NodeOutput

总结：图执行使用响应式编程，构建flux流，给流数据添加处理逻辑，订阅后发射事件，触发处理，可以认为，MainGraphExecutor/NodeExecutor做的不是执行图，而是构建action执行返回的flux流，添加数据处理的过程，好比部署好流水线，工位，待流水启动，物料流动，工位加工物料，最终输出成品

MainGraphExecutor 负责执行的控制，包括构建起始/结束事件，中断行为控制

NodeExecutor 负责执行action，连接action返回的flux，添加处理，包括返回的转换, 构建发射事件(GraphResponse<NodeOutput>类型)，添加处理其中职责合并action新增的或修改的最新全局状态，也就是图执行最终返回的结果

触发执行是 stream(inputs, config).last().map(NodeOutput::state).block() 的block方法， saa Graph的Flux不同于一般的响应式流，内置检查点（Checkpoint）组件，赋予了Flux流“记忆”和“暂停”，支持中断和恢复

下面分析图执行，代码比较多，分4部分，执行入口，主图执行，节点执行，执行返回处理

执行入口

GraphRunner.run是图的执行入口

首先处理图执行的两种情况：新开始和恢复

GraphRunnerContext

上图识别图执行是新开始还是恢复，判别是否恢复，依据是是否有用户反馈或者运行配置设置了checkPointId，后者通常是用户获取checkpoint列表，指定checkpoint恢复点

上两图是新开始和恢复执行的上下文初始化，初始化当前NodeId，下一个NodeId，全局状态，恢复执行从checkpoint获取，恢复还设置resumeFrom，这个标志在主图执行分析

主图

本节分析主图执行，负责图执行整体的控制，处理START/END节点，中断处理

中断-returnFromEmbed

saa graph支持中断和恢复，中断方式有4种，本节分析其中一种，设置returnFromEmbed

MainGraphExecutor的execute

上图，GraphRunnerContext的returnFromEmbed属性，类型是中断metadata，设置该属性有两种情况，

一是interruptAfter，中断action有3个方法，interrupt->apply->interruptAfter，returnFromEmbed是interruptAfter设置；

二是action没有实现InterruptableAction，但返回中断metadata，这种情况在源码中发现，并作了测试验证，附件的SensitiveOperationNode.java，修改自interruptable-action-example示例的SensitiveOperationNode.java，直接替换可进行测试。

上述两种情况在action apply后，处理action结果时，不能直接返回，因此采取设置上下文returnFromEmbed的方式，回到主图执行处理

中断-markNodeAsInterrupted

中断的第二种方式，RunnableConfig的markNodeAsInterrupted方法设置

MainGraphExecutor的execute

上图，RunnableConfig的markNodeAsInterrupted方法设置中断，目前没发现有使用，看上去可用来在图执行时动态设置中断节点，例如，出现工具不在线异常，动态设置工具调用节点中断，要求人工确认工具是否正常，恢复执行，撤销中断

中断-resumeFrom

中断的第三种方式，实现InterruptableAction

MainGraphExecutor的execute

上图是InterruptableAction的interrupt执行后，用户介入恢复执行，再次进入MainGraphExecutor的execute，上下文识别恢复执行设置了resumeFrom，值是中断的nodeId。

InterruptableAction的3个方法，interrupt->apply->interruptAfter，interrupt是action执行前的中断，interruptAfter是action执行后，InterruptableAction执行在后面节点执行分析。

INTERRUPT_AFTER是action结果处理时加上去的逻辑节点，如下图的箭头2

计算了nextNodeId，setNextNodeId就是覆盖了INTERRUPT_AFTER，后续这个计算了nextNodeId，setNextNodeId，当前节点移到下一节点，中断不会执行，我理解，后面执行action的apply->interruptAfter，重复了，去掉INTERRUPT_AFTER。

中断-compileConfig设置

这是中断的第四种方式，compileConfig配置

上图，compileConfig配置中断，这种方式相对简单，指定节点，支持interruptsBefore和interruptsAfter。

shouldInterrupt判断包括上节INTERRUPT_AFTER的判断，结合上节分析，排除重复的中断配置

可以看到主图好大的篇幅在处理中断，saa很重视中断，4种实现中断的方式

1 实现InterruptableAction

2 RunnableConfig的markNodeAsInterrupted设置中断

3 action返回InterruptionMetadata

4 compileConfig配置中断

节点执行

最后，交给NodeExecutor执行节点

节点

NodeExecutor负责节点的执行，分为Action执行和执行结果处理两阶段，图执行是响应式，节点执行是构建flux链，设定数据流处理，真正的处理在flux订阅后，NodeExecutor构建flux链，添加处理

设定处理节点，获取action

NodeExecutor的executNode

设定执行节点，获取节点的action，系统-图II一文分析过编译图的数据结构，action通过Node.ActionFactory构建，后者在图的节点集nodeFactories获取

中断执行

NodeExecutor的executNode

上图是中断action处理，这里可以是第一次进入，也可以是外部确认后的再(n)次恢复进入，RunnableConfig.STATE_UPDATE_METADATA_KEY有值说明是恢复进入，带来外部信息，需合并到全局状态，再执行中断，这样正确判断是否(继续)中断。回顾主图3)中断-resumeFrom, 这里执行interrupt方法，flux中断，返回中断metadata给用户，用户介入后重新进入，上下文设置resumeFrom。

执行action

NodeExecutor的executNode

上图中断处理完，首先看到通知，NODE_BEFORE，即action执行前，框架自身目前观测组件监听，对应Observation的onStart，详情参看系列-观测I和系列-观测II。

执行action的apply方法，后面处理action的返回

action返回结果处理

本节分析action执行结果处理，首先了解下action返回

上图典型的异步action的返回，map类型，上图的action简单设置一个返回，实际可以返回多个元素的map，这也是导致处理action结果的复杂性

NodeExecutor的handleActionResult方法

！可能是注释有问题，并行有并行分支和多Command的条件并行，对应的都是第三种，第二种现在代码并没有用到，所以backward compatibility应该是第二项的注释

上图3类action返回结果的处理，分普通acton，并行的action两种情况，并行分支和多Command条件，

处理逻辑是转换action返回的flux类型为GraphResponse<NodeOutput>, 构建发射事件(GraphResponse<NodeOutput>类型)，添加事件处理，下面先分析中断action的interruptAfter，第6节再分析flux类型处理

中断action处理

上面3种返回如果有flux类型返回，到这里，表示没有flux，只有静态的状态

NodeExecutor的handleActionResult方法

上图是interruptAfter方法执行，InterruptableAction的3个方法，interrupt->apply->interruptAfter，在上两节执行了前两个方法，这里执行interruptAfter，这里执行带了updateState，action apply的执行结果，若返回中断metadata，先合并新状态，然后计算下一个node，再记录checkpoint，返回。

有两点注意：

一，这里检查点计算了netxNodeId，恢复回来直接执行下一节点，这点与interrupt不同，意味着，interruptAfter只是给用户反馈action的执行结果，没有介入功能

二，如果前面有flux处理，这里不执行，那么interruptAfter什么时候执行，下面以嵌入flux处理分析

嵌入flux

本节插入flux分析，下一节继续分析NodeExecutor的handleActionResult

6.1) 获取嵌入flux

NodeExecutor的getEmbedFlux

过滤Flux类型，而且只要第一个，也就是action返回多个flux也只有一个有用，

transformFluxToGraphResponse转换flux结果，包括ChatResponse，NodeOutput的返回包装成GraphResponse，重新构造事件放入flux流

6.2) 处理flux

上节在action返回找到flux，转换类型，本节分析flux处理

NodeExecutor的processGraphResponseFlux

上图，主要是将flux返回放到临时变量，给下面更新上下文使用

上图更新上下文，副作用操作，不涉及flux

setReturnFromEmbedWithValue在主图执行分析过，action没有实现InterruptableAction，但返回InterruptionMetadata，在这里设置returnFromEmbed，注意，saa图限制并行的分支不支持中断，因此3类action flux处理只有这里处理中断，并行分支，多command并行即使返回中断metadata也忽略。

上图，partialState是action返回的状态，过滤调flux，留下静态的返回，更新到上下文，有两点注意：

一，combinedUpdateState还是上下文更新，都是updateState在后，覆盖partialStateWithoutFlux

二，红框部分是在中断action处理的问题答案，interruptAfter在action返回flux的情况的调用

计算下一个node，再记录checkpoint，设置returnFromEmbedWithValue，待回到主图执行器处理

6.x) flux流串联

最后，处理完的 Flux数据流，等待节点状态更新，后续流串联，Mono.thenMany mono转flux的方法

！强调这里的处理是挂在Flux的处理，是构建flux流的过程

继续action返回处理

前面5) 分析中断action的interruptAfter后，插入action返回flux处理，本节继续分析action返回处理

上图另一种中断，compileConfig设定的nodeId，以下图配合说明

箭头1指向的是当前节点，interruptsAfter().contains(context.getCurrentNodeId()) 当前节点设定为interruptsAfter，并且interruptBeforeEdge=true，即中断点在箭头2位置，边之前，这里就是INTERRUPT_AFTER逻辑节点插入的地方，可以回顾到中断-resumeFrom那节