Spring AI Alibaba 项目源码学习（二）-Graph 定义与描述分析

Graph 定义与描述分析

请关注公众号：阿呆-bot
概述

本文档分析 spring-ai-alibaba-graph-core 模块中 Graph 的定义和描述机制，包括入口类、关键类关系、核心实现代码和设计模式。
入口类说明

StateGraph - Graph 定义入口

StateGraph 是定义工作流图的主要入口类，用于构建包含节点（Node）和边（Edge）的状态图。
核心职责：

• 管理图中的节点集合（Nodes）
  
• 管理图中的边集合（Edges）
  
• 提供节点和边的添加方法
  
• 支持条件路由和子图
  
• 编译为可执行的 CompiledGraph
  

关键代码：

1. public class StateGraph {
3.         /**
4.          * Constant representing the END of the graph.
5.          */
6.         public static final String END = "__END__";
8.         /**
9.          * Constant representing the START of the graph.
10.          */
11.         public static final String START = "__START__";
13.         /**
14.          * Constant representing the ERROR of the graph.
15.          */
16.         public static final String ERROR = "__ERROR__";
18.         /**
19.          * Constant representing the NODE_BEFORE of the graph.
20.          */
21.         public static final String NODE_BEFORE = "__NODE_BEFORE__";
23.         /**
24.          * Constant representing the NODE_AFTER of the graph.
25.          */
26.         public static final String NODE_AFTER = "__NODE_AFTER__";
28.         /**
29.          * Collection of nodes in the graph.
30.          */
31.         final Nodes nodes = new Nodes();
33.         /**
34.          * Collection of edges in the graph.
35.          */
36.         final Edges edges = new Edges();
38.         /**
39.          * Factory for providing key strategies.
40.          */
41.         private KeyStrategyFactory keyStrategyFactory;
43.         /**
44.          * Name of the graph.
45.          */
46.         private String name;
48.         /**
49.          * Serializer for the state.
50.          */
51.         private final StateSerializer stateSerializer;

复制代码

构造函数：

1. public StateGraph(String name, KeyStrategyFactory keyStrategyFactory, PlainTextStateSerializer stateSerializer) {
2.                 this.name = name;
3.                 this.keyStrategyFactory = keyStrategyFactory;
4.                 this.stateSerializer = stateSerializer;
5.         }
7.         public StateGraph(KeyStrategyFactory keyStrategyFactory, PlainTextStateSerializer stateSerializer) {
8.                 this.keyStrategyFactory = keyStrategyFactory;
9.                 this.stateSerializer = stateSerializer;
10.         }
12.         /**
13.          * Constructs a StateGraph with the specified name, key strategy factory, and SpringAI
14.          * state serializer.
15.          * @param name the name of the graph
16.          * @param keyStrategyFactory the factory for providing key strategies
17.          * @param stateSerializer the SpringAI state serializer to use
18.          */
19.         public StateGraph(String name, KeyStrategyFactory keyStrategyFactory, SpringAIStateSerializer stateSerializer) {
20.                 this.name = name;
21.                 this.keyStrategyFactory = keyStrategyFactory;
22.                 this.stateSerializer = stateSerializer;
23.         }
25.         /**
26.          * Constructs a StateGraph with the specified key strategy factory and SpringAI state
27.          * serializer.
28.          * @param keyStrategyFactory the factory for providing key strategies
29.          * @param stateSerializer the SpringAI state serializer to use
30.          */
31.         public StateGraph(KeyStrategyFactory keyStrategyFactory, SpringAIStateSerializer stateSerializer) {
32.                 this.keyStrategyFactory = keyStrategyFactory;
33.                 this.stateSerializer = stateSerializer;
34.         }
36.         public StateGraph(String name, KeyStrategyFactory keyStrategyFactory) {
37.                 this.name = name;
38.                 this.keyStrategyFactory = keyStrategyFactory;
39.                 this.stateSerializer = new JacksonSerializer();
40.         }
42.         /**
43.          * Constructs a StateGraph with the provided key strategy factory.
44.          * @param keyStrategyFactory the factory for providing key strategies
45.          */
46.         public StateGraph(KeyStrategyFactory keyStrategyFactory) {
47.                 this.keyStrategyFactory = keyStrategyFactory;
48.                 this.stateSerializer = new JacksonSerializer();
49.         }
51.         /**
52.          * Default constructor that initializes a StateGraph with a Gson-based state
53.          * serializer.
54.          */
55.         public StateGraph() {
56.                 this.stateSerializer = new JacksonSerializer();
57.                 this.keyStrategyFactory = HashMap::new;
58.         }

复制代码

CompiledGraph - 编译后的可执行图

CompiledGraph 是 StateGraph 编译后的可执行形式，包含了优化后的节点工厂和边映射。
核心职责：

• 存储编译后的节点工厂映射
  
• 管理边映射关系
  
• 提供节点执行入口
  
• 支持中断和恢复
  

关键代码：

1. public class CompiledGraph {
3.         private static final Logger log = LoggerFactory.getLogger(CompiledGraph.class);
5.         private static String INTERRUPT_AFTER = "__INTERRUPTED__";
7.         /**
8.          * The State graph.
9.          */
10.         public final StateGraph stateGraph;
12.         /**
13.          * The Compile config.
14.          */
15.         public final CompileConfig compileConfig;
17.         /**
18.          * The Node Factories - stores factory functions instead of instances to ensure thread safety.
19.          */
20.         final Map<String, Node.ActionFactory> nodeFactories = new LinkedHashMap<>();
22.         /**
23.          * The Edges.
24.          */
25.         final Map<String, EdgeValue> edges = new LinkedHashMap<>();
27.         private final Map<String, KeyStrategy> keyStrategyMap;
29.         private final ProcessedNodesEdgesAndConfig processedData;
31.         private int maxIterations = 25;
33.         /**
34.          * Constructs a CompiledGraph with the given StateGraph.
35.          * @param stateGraph the StateGraph to be used in this CompiledGraph
36.          * @param compileConfig the compile config
37.          * @throws GraphStateException the graph state exception
38.          */
39.         protected CompiledGraph(StateGraph stateGraph, CompileConfig compileConfig) throws GraphStateException {
40.                 maxIterations = compileConfig.recursionLimit();

复制代码

关键类关系

Node - 节点抽象

Node 表示图中的节点，包含唯一标识符和动作工厂。
关键代码：

1. public class Node {
3.         public static final String PRIVATE_PREFIX = "__";
5.         public interface ActionFactory {
7.                 AsyncNodeActionWithConfig apply(CompileConfig config) throws GraphStateException;
9.         }
11.         private final String id;
13.         private final ActionFactory actionFactory;
15.         public Node(String id, ActionFactory actionFactory) {
16.                 this.id = id;
17.                 this.actionFactory = actionFactory;
18.         }
20.         /**
21.          * Constructor that accepts only the `id` and sets `actionFactory` to null.
22.          * @param id the unique identifier for the node
23.          */
24.         public Node(String id) {
25.                 this(id, null);
26.         }
28.         public void validate() throws GraphStateException {
29.                 if (Objects.equals(id, StateGraph.END) || Objects.equals(id, StateGraph.START)) {
30.                         return;
31.                 }
33.                 if (id.isBlank()) {
34.                         throw Errors.invalidNodeIdentifier.exception("blank node id");
35.                 }
37.                 if (id.startsWith(PRIVATE_PREFIX)) {
38.                         throw Errors.invalidNodeIdentifier.exception("id that start with %s", PRIVATE_PREFIX);
39.                 }
40.         }
42.         /**
43.          * id
44.          * @return the unique identifier for the node.
45.          */
46.         public String id() {
47.                 return id;
48.         }
50.         /**
51.          * actionFactory
52.          * @return a factory function that takes a {@link CompileConfig} and returns an
53.          * {@link AsyncNodeActionWithConfig} instance for the specified {@code State}.
54.          */
55.         public ActionFactory actionFactory() {
56.                 return actionFactory;
57.         }

复制代码

Edge - 边抽象

Edge 表示图中的边，连接源节点和目标节点，支持条件路由。
关键代码：

1. public record Edge(String sourceId, List<EdgeValue> targets) {
3.         public Edge(String sourceId, EdgeValue target) {
4.                 this(sourceId, List.of(target));
5.         }
7.         public Edge(String id) {
8.                 this(id, List.of());
9.         }
11.         public boolean isParallel() {
12.                 return targets.size() > 1;
13.         }
15.         public EdgeValue target() {
16.                 if (isParallel()) {
17.                         throw new IllegalStateException(format("Edge '%s' is parallel", sourceId));
18.                 }
19.                 return targets.get(0);
20.         }
22.         public boolean anyMatchByTargetId(String targetId) {
23.                 return targets().stream()
24.                         .anyMatch(v -> (v.id() != null) ? Objects.equals(v.id(), targetId)
25.                                         : v.value().mappings().containsValue(targetId)
27.                         );
28.         }
30.         public Edge withSourceAndTargetIdsUpdated(Node node, Function<String, String> newSourceId,
31.                         Function<String, EdgeValue> newTarget) {
33.                 var newTargets = targets().stream().map(t -> t.withTargetIdsUpdated(newTarget)).toList();
34.                 return new Edge(newSourceId.apply(sourceId), newTargets);
36.         }

复制代码

OverAllState - 全局状态

OverAllState 是贯穿整个图执行过程的全局状态对象，用于在节点间传递数据。
关键代码：

1. public final class OverAllState implements Serializable {
2.         public static final Object MARK_FOR_REMOVAL = new Object();
4.         /**
5.          * Internal map storing the actual state data. All get/set operations on state values
6.          * go through this map.
7.          */
8.         private final Map<String, Object> data;
10.         /**
11.          * Mapping of keys to their respective update strategies. Determines how values for
12.          * each key should be merged or updated.
13.          */
14.         private final Map<String, KeyStrategy> keyStrategies;
16.         /**
17.          * Store instance for long-term memory storage across different executions.
18.          */
19.         private Store store;
21.         /**
22.          * The default key used for standard input injection into the state. Typically used
23.          * when initializing the state with user or external input.
24.          */

复制代码

关键类关系图

以下 PlantUML 类图展示了 Graph 定义相关的关键类及其关系：

1. @startuml
2. !theme plain
3. skinparam classAttributeIconSize 0
5. package "Graph Definition" {
6.     class StateGraph {
7.         -Nodes nodes
8.         -Edges edges
9.         -KeyStrategyFactory keyStrategyFactory
10.         -String name
11.         -StateSerializer stateSerializer
12.         +addNode(String, NodeAction)
13.         +addEdge(String, String)
14.         +addConditionalEdges(...)
15.         +compile(CompileConfig): CompiledGraph
16.     }
18.     class CompiledGraph {
19.         +StateGraph stateGraph
20.         +CompileConfig compileConfig
21.         -Map<String, Node.ActionFactory> nodeFactories
22.         -Map<String, EdgeValue> edges
23.         -Map<String, KeyStrategy> keyStrategyMap
24.     }
26.     class Node {
27.         -String id
28.         -ActionFactory actionFactory
29.         +id(): String
30.         +actionFactory(): ActionFactory
31.         +validate(): void
32.     }
34.     class Edge {
35.         -String sourceId
36.         -List<EdgeValue> targets
37.         +isParallel(): boolean
38.         +target(): EdgeValue
39.         +validate(Nodes): void
40.     }
42.     class OverAllState {
43.         -Map<String, Object> data
44.         -Map<String, KeyStrategy> keyStrategies
45.         -Store store
46.         +get(String): Object
47.         +put(String, Object): void
48.         +registerKeyAndStrategy(String, KeyStrategy): void
49.     }
51.     interface NodeAction {
52.         +apply(OverAllState): Map<String, Object>
53.     }
55.     interface EdgeAction {
56.         +apply(OverAllState): String
57.     }
59.     class KeyStrategy {
60.         +merge(Object, Object): Object
61.     }
63.     class StateSerializer {
64.         +serialize(OverAllState): String
65.         +deserialize(String): OverAllState
66.     }
67. }
69. StateGraph "1" --> "1" CompiledGraph : compiles to
70. StateGraph "1" --> "*" Node : contains
71. StateGraph "1" --> "*" Edge : contains
72. StateGraph --> KeyStrategyFactory : uses
73. StateGraph --> StateSerializer : uses
74. Node --> NodeAction : creates via ActionFactory
75. Edge --> EdgeAction : uses for conditional routing
76. CompiledGraph --> Node : stores factories
77. CompiledGraph --> Edge : stores mappings
78. OverAllState --> KeyStrategy : uses
79. OverAllState --> Store : uses
81. note right of StateGraph
82.   入口类：用于定义工作流图
83.   支持添加节点和边
84.   支持条件路由和子图
85. end note
87. note right of CompiledGraph
88.   编译后的可执行图
89.   包含优化的节点工厂
90.   线程安全的工厂函数
91. end note
93. note right of OverAllState
94.   全局状态对象
95.   在节点间传递数据
96.   支持键策略管理
97. end note
99. @enduml

复制代码

实现关键点说明

1. Builder 模式

StateGraph 使用链式调用模式构建图，支持流畅的 API：

1. StateGraph graph = new StateGraph("MyGraph", keyStrategyFactory)
2.     .addNode("node1", nodeAction1)
3.     .addNode("node2", nodeAction2)
4.     .addEdge(START, "node1")
5.     .addConditionalEdges("node1", edgeAction, Map.of("yes", "node2", "no", END))
6.     .addEdge("node2", END);

复制代码

2. 工厂模式

Node 使用 ActionFactory 接口延迟创建节点动作，确保线程安全：

1. public interface ActionFactory {
2.     AsyncNodeActionWithConfig apply(CompileConfig config) throws GraphStateException;
3. }

复制代码

这种设计允许在编译时创建工厂函数，在执行时根据配置创建实际的动作实例。
3. 策略模式

KeyStrategy 用于控制状态键的更新策略，支持不同的合并逻辑（Replace、Append、Reduce 等）。
4. 序列化支持

StateGraph 支持多种状态序列化器：

• PlainTextStateSerializer：纯文本序列化
  
• SpringAIStateSerializer：Spring AI 标准序列化
  
• JacksonSerializer：Jackson JSON 序列化（默认）
  

5. 验证机制

Node 和 Edge 都实现了 validate() 方法，确保图的完整性：

• 节点 ID 不能为空或使用保留前缀
  
• 边引用的节点必须存在
  
• 并行边不能有重复目标
  

总结说明

核心设计理念

• 声明式 API：通过 StateGraph 提供声明式的图定义方式，隐藏底层复杂性
  
• 编译时优化：StateGraph 编译为 CompiledGraph，将定义转换为可执行形式
  
• 状态管理：OverAllState 作为全局状态容器，支持键策略和序列化
  
• 类型安全：使用泛型和接口确保类型安全
  
• 可扩展性：通过接口和工厂模式支持自定义节点和边动作
  

关键优势

• 灵活性：支持同步和异步节点、条件路由、并行执行
  
• 可维护性：清晰的类层次结构和职责分离
  
• 可测试性：接口抽象便于单元测试
  
• 性能：编译时优化和工厂模式减少运行时开销
  

使用流程

• 定义图：使用 StateGraph 添加节点和边
  
• 编译图：调用 compile() 方法生成 CompiledGraph
  
• 执行图：通过 GraphRunner 执行编译后的图
  
• 状态传递：使用 OverAllState 在节点间传递数据
  

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

新版吗？好像是停更了吧。