news 2026/7/18 3:57:09

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

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
ai-agent框架spring ai/alibaba 原理源码分析(五)graph III 图执行

图执行

上文分析了图编译,编译后图具备高效执行的数据结构,本文分析图执行,图执行可分两部分,主图执行,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那节

完成节点处理

​构建NodeOutput和记录检查点,flux流连接

图返回处理

最后,本节分析图执行返回结果,图执行有同步和流式两种

CompiledGraph

上图,上两个是同步调用,第三个是流式调用

同步调用也是使用流式调用,last方法只要最后一个结果,block方法订阅数据流触发flux发射事件;流式调用返回flux,client需要自行订阅,自行处理flux流

附件

非InterruptableAction的action实现中断,替换spring-ai-alibaba-graph-example的interruptable-action-example同名类即可测试

SensitiveOperationNode.java

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/7/18 3:56:04

你的合同管理在第几级?——企业合同管理成熟度五级模型

每次年中汇报&#xff0c;法务负责人大概率会被老板问一个问题&#xff1a;“我们的合同管理现在在什么水平&#xff1f;”这个问题不好答。说“还行”太模糊&#xff0c;说“有系统”不够准确&#xff0c;说“正在搞AI”又怕被追问到底搞出了什么效果。不好答的根本原因&#…

作者头像 李华
网站建设 2026/7/18 3:56:01

椰子水从“天然爆款”到“信任重构”,行业拐点下的破局之路

在全民追捧“0糖0脂”的健康消费时代&#xff0c;椰子水曾凭借“天然、低糖、补充电解质”的纯净标签&#xff0c;成为中产饮品里的“白月光”——从小众高端功能饮&#xff0c;一跃成为健康饮料赛道的现象级爆款。但2026年315前夕&#xff0c;“椰子水集体造假”“100%纯椰水掺…

作者头像 李华
网站建设 2026/7/18 3:55:03

生产级PDF文档嵌入流水线:从Azure PDF解析到向量搜索实战

1. 项目概述&#xff1a;这不是又一个“聊天机器人”&#xff0c;而是一套能真正读懂你PDF的文档智能中枢“Unlocking Document Intelligence”——这句话在2024年已经不是PPT里的空洞口号&#xff0c;而是每天被法务、合规、HR、技术支持团队真实喊出来的刚需。我上个月帮一家…

作者头像 李华
网站建设 2026/7/18 3:54:52

语音AI全链路协同:从识别到行动的实时工作伙伴

1. 语音AI不再只是“能说”&#xff0c;而是“会听、会想、会做事”的实时工作伙伴 你有没有过这样的体验&#xff1a;在嘈杂的超市里给客服打电话&#xff0c;报一串带字母的订单号&#xff0c;对方反复确认三遍才听清&#xff1b;或者开跨国视频会议时&#xff0c;一边要听翻…

作者头像 李华
网站建设 2026/7/18 3:54:24

Claude Code 必须运行在 Docker 沙盒中的技术原理与实践

1. 项目概述&#xff1a;为什么你必须把 Claude Code 关进 Docker 这个“玻璃房”我第一次被 Claude Code “温柔地”删掉.env文件时&#xff0c;正端着咖啡站在厨房。它在终端里输出一行轻快的提示&#xff1a;“Detected unused environment file — removed for clarity.” …

作者头像 李华
网站建设 2026/7/18 3:54:22

Unity 2022安卓打包报错全解析:从Gradle原理到实战解决

1. 项目概述&#xff1a;当Unity 2022遇上安卓打包的“玄学”报错作为一名在游戏开发一线摸爬滚打了十多年的老鸟&#xff0c;我几乎见证了Unity引擎的每一次重大迭代。从早期的Unity 4到现在的Unity 2022 LTS&#xff0c;引擎的每一次升级都带来了更强大的功能和更便捷的工作流…

作者头像 李华