Spring AI Alibaba 项目源码学习（三）-Graph 执行流程分析

Graph 执行流程分析

概述

本文档分析 spring-ai-alibaba-graph-core 模块中 Graph 的执行流程，包括执行器（Executor）、调度机制、Checkpoint 机制和状态管理。
入口类说明

GraphRunner - 执行入口

GraphRunner 是基于 Project Reactor 的响应式图执行引擎，是执行 Graph 的主要入口。
核心职责：

• 封装编译后的图和执行配置
  
• 提供响应式执行接口（返回 Flux）
  
• 委托给 MainGraphExecutor 执行实际流程
  

关键代码：

1. public class GraphRunner {
3.         private final CompiledGraph compiledGraph;
5.         private final RunnableConfig config;
7.         private final AtomicReference<Object> resultValue = new AtomicReference<>();
9.         // Handler for main execution flow - demonstrates encapsulation
10.         private final MainGraphExecutor mainGraphExecutor;
12.         public GraphRunner(CompiledGraph compiledGraph, RunnableConfig config) {
13.                 this.compiledGraph = compiledGraph;
14.                 this.config = config;
15.                 // Initialize the main execution handler - demonstrates encapsulation
16.                 this.mainGraphExecutor = new MainGraphExecutor();
17.         }
19.         public Flux<GraphResponse<NodeOutput>> run(OverAllState initialState) {
20.                 return Flux.defer(() -> {
21.                         try {
22.                                 GraphRunnerContext context = new GraphRunnerContext(initialState, config, compiledGraph);
23.                                 // Delegate to the main execution handler - demonstrates polymorphism
24.                                 return mainGraphExecutor.execute(context, resultValue);
25.                         }
26.                         catch (Exception e) {
27.                                 return Flux.error(e);
28.                         }
29.                 });
30.         }

复制代码

MainGraphExecutor - 主执行器

MainGraphExecutor 是主执行流程处理器，继承自 BaseGraphExecutor，负责协调整个图的执行。
核心职责：

• 处理开始节点和结束节点
  
• 管理迭代次数和中断逻辑
  
• 协调节点执行和边路由
  
• 处理 Checkpoint 和恢复
  

关键代码：

1. public class MainGraphExecutor extends BaseGraphExecutor {
3.         private final NodeExecutor nodeExecutor;
5.         public MainGraphExecutor() {
6.                 this.nodeExecutor = new NodeExecutor(this);
7.         }
9.         /**
10.          * Implementation of the execute method. This demonstrates polymorphism as it provides
11.          * a specific implementation for main execution flow.
12.          * @param context the graph runner context
13.          * @param resultValue the atomic reference to store the result value
14.          * @return Flux of GraphResponse with execution result
15.          */
16.         @Override
17.         public Flux<GraphResponse<NodeOutput>> execute(GraphRunnerContext context, AtomicReference<Object> resultValue) {
18.                 try {
19.                         if (context.shouldStop() || context.isMaxIterationsReached()) {
20.                                 return handleCompletion(context, resultValue);
21.                         }
23.                         final var returnFromEmbed = context.getReturnFromEmbedAndReset();
24.                         if (returnFromEmbed.isPresent()) {
25.                                 var interruption = returnFromEmbed.get().value(new TypeRef<InterruptionMetadata>() {
26.                                 });
27.                                 if (interruption.isPresent()) {
28.                                         return Flux.just(GraphResponse.done(interruption.get()));
29.                                 }
30.                                 return Flux.just(GraphResponse.done(context.buildCurrentNodeOutput()));
31.                         }
33.                         if (context.getCurrentNodeId() != null && context.getConfig().isInterrupted(context.getCurrentNodeId())) {
34.                                 context.getConfig().withNodeResumed(context.getCurrentNodeId());
35.                                 return Flux.just(GraphResponse.done(GraphResponse.done(context.getCurrentStateData())));
36.                         }
38.                         if (context.isStartNode()) {
39.                                 return handleStartNode(context);
40.                         }
42.                         if (context.isEndNode()) {
43.                                 return handleEndNode(context, resultValue);
44.                         }
46.                         final var resumeFrom = context.getResumeFromAndReset();
47.                         if (resumeFrom.isPresent()) {
48.                                 if (context.getCompiledGraph().compileConfig.interruptBeforeEdge()
49.                                                 && java.util.Objects.equals(context.getNextNodeId(), INTERRUPT_AFTER)) {
50.                                         var nextNodeCommand = context.nextNodeId(resumeFrom.get(), context.getCurrentStateData());
51.                                         context.setNextNodeId(nextNodeCommand.gotoNode());
52.                                         context.setCurrentNodeId(null);
53.                                 }
54.                         }
56.                         if (context.shouldInterrupt()) {
57.                                 try {
58.                                         InterruptionMetadata metadata = InterruptionMetadata
59.                                                 .builder(context.getCurrentNodeId(), context.cloneState(context.getCurrentStateData()))
60.                                                 .build();
61.                                         return Flux.just(GraphResponse.done(metadata));

复制代码

NodeExecutor - 节点执行器

NodeExecutor 负责执行单个节点，处理节点动作的执行和结果处理。
核心职责：

• 执行节点动作（NodeAction）
  
• 处理中断逻辑（InterruptableAction）
  
• 处理流式输出（StreamingOutput）
  
• 更新状态并确定下一个节点
  

关键代码：

1. public class NodeExecutor extends BaseGraphExecutor {
3.         private final MainGraphExecutor mainGraphExecutor;
5.         public NodeExecutor(MainGraphExecutor mainGraphExecutor) {
6.                 this.mainGraphExecutor = mainGraphExecutor;
7.         }
9.         /**
10.          * Implementation of the execute method. This demonstrates polymorphism as it provides
11.          * a specific implementation for node execution.
12.          * @param context the graph runner context
13.          * @param resultValue the atomic reference to store the result value
14.          * @return Flux of GraphResponse with execution result
15.          */
16.         @Override
17.         public Flux<GraphResponse<NodeOutput>> execute(GraphRunnerContext context, AtomicReference<Object> resultValue) {
18.                 return executeNode(context, resultValue);
19.         }
21.         /**
22.          * Executes a node and handles its result.
23.          * @param context the graph runner context
24.          * @param resultValue the atomic reference to store the result value
25.          * @return Flux of GraphResponse with node execution result
26.          */
27.         private Flux<GraphResponse<NodeOutput>> executeNode(GraphRunnerContext context,
28.                         AtomicReference<Object> resultValue) {
29.                 try {
30.                         context.setCurrentNodeId(context.getNextNodeId());
31.                         String currentNodeId = context.getCurrentNodeId();
32.                         AsyncNodeActionWithConfig action = context.getNodeAction(currentNodeId);
34.                         if (action == null) {
35.                                 return Flux.just(GraphResponse.error(RunnableErrors.missingNode.exception(currentNodeId)));
36.                         }
38.                         if (action instanceof InterruptableAction) {
39.                                 context.getConfig().metadata(RunnableConfig.STATE_UPDATE_METADATA_KEY).ifPresent(updateFromFeedback -> {
40.                                         if (updateFromFeedback instanceof Map<?, ?>) {
41.                                                 context.mergeIntoCurrentState((Map<String, Object>) updateFromFeedback);
42.                                         } else {
43.                                                 throw new RuntimeException();
44.                                         }
45.                                 });
46.                                 Optional<InterruptionMetadata> interruptMetadata = ((InterruptableAction) action)
47.                                         .interrupt(currentNodeId, context.cloneState(context.getCurrentStateData()), context.getConfig());
48.                                 if (interruptMetadata.isPresent()) {

复制代码

GraphRunnerContext - 执行上下文

GraphRunnerContext 管理图执行过程中的状态和上下文信息。
核心职责：

• 管理当前节点和下一个节点
  
• 管理迭代次数
  
• 处理 Checkpoint 和恢复
  
• 管理状态更新
  

关键代码：

1. public class GraphRunnerContext {
3.         public static final String INTERRUPT_AFTER = "__INTERRUPTED__";
5.         private static final Logger log = LoggerFactory.getLogger(GraphRunner.class);
7.         final CompiledGraph compiledGraph;
9.         final AtomicInteger iteration = new AtomicInteger(0);
11.         OverAllState overallState;
13.         RunnableConfig config;
15.         String currentNodeId;
17.         String nextNodeId;
19.         String resumeFrom;
21.         ReturnFromEmbed returnFromEmbed;
23.         public GraphRunnerContext(OverAllState initialState, RunnableConfig config, CompiledGraph compiledGraph)
24.                         throws Exception {
25.                 this.compiledGraph = compiledGraph;
26.                 this.config = config;
28.                 if (config.metadata(RunnableConfig.HUMAN_FEEDBACK_METADATA_KEY).isPresent()) {
29.                         initializeFromResume(initialState, config);
30.                 } else {
31.                         initializeFromStart(initialState, config);
32.                 }
33.         }
35.         private void initializeFromResume(OverAllState initialState, RunnableConfig config) {
36.                 log.trace("RESUME REQUEST");
38.                 var saver = compiledGraph.compileConfig.checkpointSaver()
39.                         .orElseThrow(() -> new IllegalStateException("Resume request without a configured checkpoint saver!"));
40.                 var checkpoint = saver.get(config)
41.                         .orElseThrow(() -> new IllegalStateException("Resume request without a valid checkpoint!"));

复制代码

执行流程时序图

以下 PlantUML 时序图展示了 Graph 的完整执行流程：

Checkpoint 机制

BaseCheckpointSaver - Checkpoint 保存器接口

BaseCheckpointSaver 定义了 Checkpoint 保存和恢复的接口。
关键代码：

1. public interface BaseCheckpointSaver {
3.         String THREAD_ID_DEFAULT = "$default";
5.         record Tag(String threadId, Collection<Checkpoint> checkpoints) {
6.                 public Tag(String threadId, Collection<Checkpoint> checkpoints) {
7.                         this.threadId = threadId;
8.                         this.checkpoints = ofNullable(checkpoints).map(List::copyOf).orElseGet(List::of);
9.                 }
10.         }
12.         default Tag release(RunnableConfig config) throws Exception {
13.                 return null;
14.         }
16.         Collection<Checkpoint> list(RunnableConfig config);
18.         Optional<Checkpoint> get(RunnableConfig config);
20.         RunnableConfig put(RunnableConfig config, Checkpoint checkpoint) throws Exception;
22.         boolean clear(RunnableConfig config);
24.         default Optional<Checkpoint> getLast(LinkedList<Checkpoint> checkpoints, RunnableConfig config) {
25.                 return (checkpoints == null || checkpoints.isEmpty()) ? Optional.empty() : ofNullable(checkpoints.peek());
26.         }
28.         default LinkedList<Checkpoint> getLinkedList(List<Checkpoint> checkpoints) {
29.                 return Objects.nonNull(checkpoints) ? new LinkedList<>(checkpoints) : new LinkedList<>();
30.         }
32. }

复制代码

Checkpoint 流程

• 保存 Checkpoint：在执行节点后，通过 GraphRunnerContext.addCheckpoint() 保存当前状态
  
• 恢复 Checkpoint：通过 RunnableConfig 中的 checkpoint ID 恢复执行
  
• 释放 Checkpoint：执行完成后释放 Checkpoint 资源
  

状态管理

OverAllState 更新机制

状态更新通过 KeyStrategy 控制：

• ReplaceStrategy：替换策略，新值完全替换旧值
  
• AppendStrategy：追加策略，新值追加到列表
  
• Reducer：归约策略，使用自定义函数合并值
  

状态更新流程：

• 节点执行返回 Map 更新
  
• GraphRunnerContext 根据 KeyStrategy 合并更新
  
• 更新后的状态传递给下一个节点
  

实现关键点说明

1. 响应式编程

使用 Project Reactor 的 Flux 实现响应式执行：

• 支持流式输出
  
• 支持背压控制
  
• 支持异步执行
  

2. 模板方法模式

BaseGraphExecutor 定义执行框架，子类实现具体逻辑：

• MainGraphExecutor：主执行流程
  
• NodeExecutor：节点执行流程
  

3. 上下文模式

GraphRunnerContext 封装执行上下文：

• 管理当前执行状态
  
• 提供状态访问接口
  
• 处理 Checkpoint 和恢复
  

4. 中断和恢复机制

支持执行中断和恢复：

• InterruptableAction：可中断的动作
  
• InterruptionMetadata：中断元数据
  
• Checkpoint 保存和恢复
  

5. 迭代控制

通过 maxIterations 控制最大迭代次数，防止无限循环。
总结说明

核心执行流程

• 初始化：创建 GraphRunnerContext，初始化状态
  
• 开始执行：从 START 节点开始
  
• 节点执行：执行当前节点，更新状态
  
• 边路由：根据 EdgeAction 确定下一个节点
  
• Checkpoint：保存执行状态（可选）
  
• 迭代：重复步骤 3-5，直到到达 END 节点
  
• 完成：返回最终结果
  

关键设计特点

• 响应式：基于 Reactor 的响应式执行
  
• 可中断：支持执行中断和恢复
  
• 状态管理：灵活的状态更新策略
  
• 可观测：支持监控和追踪
  
• 容错性：错误处理和恢复机制
  

执行器层次结构

1. BaseGraphExecutor (抽象基类)
2.     ├── MainGraphExecutor (主执行器)
3.     └── NodeExecutor (节点执行器)

复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

感谢分享

yyds。多谢分享