图执行上文分析了图编译编译后图具备高效执行的数据结构本文分析图执行图执行可分两部分主图执行MainGraphExecutor负责主图执行主要是图执行控制 包括START END中断点执行NodeExecutor负责节点处理。图执行使用响应式流如下图可以认为图遍历的过程是Flux(数据源)连接成响应式管道并加上处理的过程Flux是节点Action执行的返回。做个类比Flux链好比流水线输送数据(材料)数据处理类比员工材料到工位加工处理流水线的开关类比Flux的订阅这里看出响应式编程的特点先构建好数据流动管道加上相应的处理开关打开(调用block方法)数据发射触发处理实现异步无等待的处理检查点本节首先介绍一下检查点检查点是图的执行期重要的组件负责保存执行环境用作中断/恢复回溯等关键功能也是图执行容错机制的支撑也是agent的短期记忆组件。上图检查点的类图CheckPoint检查点保存着图执行的上下文用于中断恢复容错回放等idcheckpoint idstate全局状态OverAllStatenodeId/nextNodeId 当前执行的nodeId下一个执行nodeId检查点整体存储类似文件夹文件夹id就是threadId(会话id)通常外部传入checkpoints按时间倒序list存储BaseCheckPointSaver检查点存储接口有多种实现redismemorymysql等执行类图下面分析参与执行的类图上图是参与执行的类图关于图的数据结构参看系列-图I和系列-图II分析了图结构编译后的数据结构GraphRunner执行入口负责GraphRunnerContext构建和初始然后调用MainGraphExecutor执行图GraphRunnerContext运行的上下文包括全局的状态OverAllState运行配置RunnableConfig运行标的CompiledGraph负责识别是新执行还是恢复执行还有2个重要的职责nextNodeId计算遍历的下一个节点addCheckpoint记录检查点MainGraphExecutor/NodeExecutor执行器前者负责主图执行后者负责节点执行主图执行重点是START节点END节点中断等节点执行交给NodeExecutor节点执行完返回MainGraphExecutor这样交替推进图的执行RunnableConfig运行时配置包含执行所需的参数设置其中关键属性threadId, checkPointId, nextNodenodeFactories/Node.ActionFactory/AsyncNodeActionWithConfig nodeFactories是图的节点集在系列-编译图介绍过。Node.ActionFactory构建节点Action类型是AsyncNodeActionWithConfigGraphResponse图执行返回结构包括NodeOutputreturnValue属性前者包含全局状态后者可以承载多种类型主要用于中断metadata也有其他用途有一点值得注意中断metadata是继承NodeOutput总结图执行使用响应式编程构建flux流给流数据添加处理逻辑订阅后发射事件触发处理可以认为MainGraphExecutor/NodeExecutor做的不是执行图而是构建action执行返回的flux流添加数据处理的过程好比部署好流水线工位待流水启动物料流动工位加工物料最终输出成品MainGraphExecutor 负责执行的控制包括构建起始/结束事件中断行为控制NodeExecutor 负责执行action连接action返回的flux添加处理包括返回的转换, 构建发射事件(GraphResponseNodeOutput类型)添加处理其中职责合并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节点中断处理中断-returnFromEmbedsaa graph支持中断和恢复中断方式有4种本节分析其中一种设置returnFromEmbedMainGraphExecutor的execute上图GraphRunnerContext的returnFromEmbed属性类型是中断metadata设置该属性有两种情况一是interruptAfter中断action有3个方法interrupt-apply-interruptAfterreturnFromEmbed是interruptAfter设置二是action没有实现InterruptableAction但返回中断metadata这种情况在源码中发现并作了测试验证附件的SensitiveOperationNode.java修改自interruptable-action-example示例的SensitiveOperationNode.java直接替换可进行测试。上述两种情况在action apply后处理action结果时不能直接返回因此采取设置上下文returnFromEmbed的方式回到主图执行处理中断-markNodeAsInterrupted中断的第二种方式RunnableConfig的markNodeAsInterrupted方法设置MainGraphExecutor的execute上图RunnableConfig的markNodeAsInterrupted方法设置中断目前没发现有使用看上去可用来在图执行时动态设置中断节点例如出现工具不在线异常动态设置工具调用节点中断要求人工确认工具是否正常恢复执行撤销中断中断-resumeFrom中断的第三种方式实现InterruptableActionMainGraphExecutor的execute上图是InterruptableAction的interrupt执行后用户介入恢复执行再次进入MainGraphExecutor的execute上下文识别恢复执行设置了resumeFrom值是中断的nodeId。InterruptableAction的3个方法interrupt-apply-interruptAfterinterrupt是action执行前的中断interruptAfter是action执行后InterruptableAction执行在后面节点执行分析。INTERRUPT_AFTER是action结果处理时加上去的逻辑节点如下图的箭头2计算了nextNodeIdsetNextNodeId就是覆盖了INTERRUPT_AFTER后续这个计算了nextNodeIdsetNextNodeId当前节点移到下一节点中断不会执行我理解后面执行action的apply-interruptAfter重复了去掉INTERRUPT_AFTER。中断-compileConfig设置这是中断的第四种方式compileConfig配置上图compileConfig配置中断这种方式相对简单指定节点支持interruptsBefore和interruptsAfter。shouldInterrupt判断包括上节INTERRUPT_AFTER的判断结合上节分析排除重复的中断配置可以看到主图好大的篇幅在处理中断saa很重视中断4种实现中断的方式1 实现InterruptableAction2 RunnableConfig的markNodeAsInterrupted设置中断3 action返回InterruptionMetadata4 compileConfig配置中断节点执行最后 交给NodeExecutor执行节点节点NodeExecutor负责节点的执行分为Action执行和执行结果处理两阶段图执行是响应式节点执行是构建flux链设定数据流处理真正的处理在flux订阅后NodeExecutor构建flux链添加处理设定处理节点获取actionNodeExecutor的executNode设定执行节点获取节点的action系统-图II一文分析过编译图的数据结构action通过Node.ActionFactory构建后者在图的节点集nodeFactories获取中断执行NodeExecutor的executNode上图是中断action处理这里可以是第一次进入也可以是外部确认后的再(n)次恢复进入RunnableConfig.STATE_UPDATE_METADATA_KEY有值说明是恢复进入带来外部信息需合并到全局状态再执行中断这样正确判断是否(继续)中断。回顾主图3)中断-resumeFrom, 这里执行interrupt方法flux中断返回中断metadata给用户用户介入后重新进入上下文设置resumeFrom。执行actionNodeExecutor的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类型为GraphResponseNodeOutput, 构建发射事件(GraphResponseNodeOutput类型)添加事件处理下面先分析中断action的interruptAfter第6节再分析flux类型处理中断action处理上面3种返回如果有flux类型返回到这里表示没有flux只有静态的状态NodeExecutor的handleActionResult方法上图是interruptAfter方法执行InterruptableAction的3个方法interrupt-apply-interruptAfter在上两节执行了前两个方法这里执行interruptAfter这里执行带了updateStateaction apply的执行结果若返回中断metadata先合并新状态然后计算下一个node再记录checkpoint返回。有两点注意一这里检查点计算了netxNodeId恢复回来直接执行下一节点这点与interrupt不同意味着interruptAfter只是给用户反馈action的执行结果没有介入功能二如果前面有flux处理这里不执行那么interruptAfter什么时候执行下面以嵌入flux处理分析嵌入flux本节插入flux分析下一节继续分析NodeExecutor的handleActionResult6.1) 获取嵌入fluxNodeExecutor的getEmbedFlux过滤Flux类型而且只要第一个也就是action返回多个flux也只有一个有用transformFluxToGraphResponse转换flux结果包括ChatResponseNodeOutput的返回包装成GraphResponse重新构造事件放入flux流6.2) 处理flux上节在action返回找到flux转换类型本节分析flux处理NodeExecutor的processGraphResponseFlux上图主要是将flux返回放到临时变量给下面更新上下文使用上图更新上下文副作用操作不涉及fluxsetReturnFromEmbedWithValue在主图执行分析过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并且interruptBeforeEdgetrue即中断点在箭头2位置边之前这里就是INTERRUPT_AFTER逻辑节点插入的地方可以回顾到中断-resumeFrom那节完成节点处理构建NodeOutput和记录检查点flux流连接图返回处理最后本节分析图执行返回结果图执行有同步和流式两种CompiledGraph上图上两个是同步调用第三个是流式调用同步调用也是使用流式调用last方法只要最后一个结果block方法订阅数据流触发flux发射事件流式调用返回fluxclient需要自行订阅自行处理flux流附件非InterruptableAction的action实现中断替换spring-ai-alibaba-graph-example的interruptable-action-example同名类即可测试SensitiveOperationNode.java