Spring AI Alibaba 88｜Graph Core · 并行节点定义 · 定义带并行节点的图

Spring AI Alibaba Graph 允许您定义并行节点以加速总图执行。

要实现并发执行，您必须使用 RunnableConfig 为特定的并行节点提供一个 Executor：

import com.alibaba.cloud.ai.graph.RunnableConfig;
import java.util.concurrent.ForkJoinPool;

RunnableConfig runnableConfig = RunnableConfig.builder()
  .addParallelNodeExecutor("<parallel node id>", ForkJoinPool.commonPool())
  .build();

注意：

如果未指定 Executor，并行节点将被顺序调度，要并发运行它们，您必须依赖 CompletableFuture 的异步功能，使用 AsyncNodeAction

当前并行节点实现执行存在一些整体限制：

• 仅支持 Fork-Join 模型

┌─┐
          │A│
          └─┘
           |
     ┌-----------┐
     |     |     |
   ┌──┐  ┌──┐  ┌──┐
   │A1│  │A2│  │A3│
   └──┘  └──┘  └──┘
     |     |     |
     └-----------┘
           |
          ┌─┐
          │B│
          └─┘

• 仅允许一个并行步骤 ⚠️

┌─┐
          │A│
          └─┘
           |
     ┌-----------┐
     |     |     |
   ┌──┐  ┌──┐  ┌──┐
   │A1│  │A2│  │A3│
   └──┘  └──┘  └──┘
     |     |     |
   ┌──┐    |     |
   │A4│ ❌ 不允许
   └──┘    |     |
     |     |     |
     └-----------┘
           |
          ┌─┐
          │B│
          └─┘

• 不允许使用条件边 ⚠️

以下是一些示例，展示如何创建分支数据流。

只需将多个节点关联到同一条边即可。

示例 – 定义并行节点

import com.alibaba.cloud.ai.graph.CompiledGraph;
import com.alibaba.cloud.ai.graph.KeyStrategy;
import com.alibaba.cloud.ai.graph.KeyStrategyFactory;
import com.alibaba.cloud.ai.graph.StateGraph;
import com.alibaba.cloud.ai.graph.action.AsyncNodeAction;
import com.alibaba.cloud.ai.graph.exception.GraphStateException;
import com.alibaba.cloud.ai.graph.state.strategy.AppendStrategy;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.alibaba.cloud.ai.graph.StateGraph.END;
import static com.alibaba.cloud.ai.graph.StateGraph.START;
import static com.alibaba.cloud.ai.graph.action.AsyncNodeAction.node_async;

// 定义状态策略
KeyStrategyFactory keyStrategyFactory = () -> {
  HashMap<String, KeyStrategy> keyStrategyHashMap = new HashMap<>();
  keyStrategyHashMap.put("messages", new AppendStrategy());
  return keyStrategyHashMap;
};

// 创建节点的辅助方法
private static AsyncNodeAction makeNode(String message) {
  return node_async(state -> Map.of("messages", List.of(message)));
}

// 构建并行图
StateGraph workflow = new StateGraph(keyStrategyFactory)
      .addNode("A", makeNode("A"))
      .addNode("A1", makeNode("A1"))
      .addNode("A2", makeNode("A2"))
      .addNode("A3", makeNode("A3"))
      .addNode("B", makeNode("B"))
      .addNode("C", makeNode("C"))
      .addEdge("A", "A1")   // A 到 A1
      .addEdge("A", "A2")   // A 到 A2 (并行)
      .addEdge("A", "A3")   // A 到 A3 (并行)
      .addEdge("A1", "B")   // A1 到 B
      .addEdge("A2", "B")   // A2 到 B (汇聚)
      .addEdge("A3", "B")   // A3 到 B (汇聚)
      .addEdge("B", "C")
      .addEdge(START, "A")
      .addEdge("C", END);

CompiledGraph graph = workflow.compile();

执行并行 Graph

// 执行 Graph
compiledGraph.stream(Map.of())
      .doOnNext(step -> System.out.println(step))
      .doOnError(error -> System.err.println("流错误: " + error.getMessage()))
      .doOnComplete(() -> System.out.println("流完成"))
      .blockLast();

图示：

START
      |
      A
    / | 
   A1 A2 A3  (并行执行)
     | /
      B
      |
      C
      |
     END

条件返回到并行节点

您也可以在所有并行执行结束后有条件地返回到特定并行节点：

import com.alibaba.cloud.ai.graph.CompiledGraph;
import com.alibaba.cloud.ai.graph.StateGraph;
import static com.alibaba.cloud.ai.graph.StateGraph.END;
import static com.alibaba.cloud.ai.graph.StateGraph.START;
import static com.alibaba.cloud.ai.graph.action.AsyncEdgeAction.edge_async;

StateGraph workflow = new StateGraph(keyStrategyFactory)
      .addNode("A", makeNode("A"))
      .addNode("A1", makeNode("A1"))
      .addNode("A2", makeNode("A2"))
      .addNode("A3", makeNode("A3"))
      .addNode("B", makeNode("B"))
      .addNode("C", makeNode("C"))
      .addEdge("A", "A1")
      .addEdge("A", "A2")
      .addEdge("A", "A3")
      .addEdge("A1", "B")
      .addEdge("A2", "B")
      .addEdge("A3", "B")
      // 条件边：根据状态决定是继续还是返回
      .addConditionalEdges("B",
              edge_async(state -> {
                  // 检查上一个节点
                  List<String> messages = (List<String>) state.value("messages").orElse(List.of());
                  String lastMessage = messages.isEmpty() ? "" : messages.get(messages.size() - 1);
                  return lastMessage.equals("A3") ? "continue" : "back";
              }),
              Map.of(
                      "back", "A1",
                      "continue", "C"
              )
      )
      .addEdge(START, "A")
      .addEdge("C", END);

CompiledGraph graph = workflow.compile();

本文整理自 Spring AI Alibaba 官方文档《并行节点定义》：https://java2ai.com/docs/frameworks/graph-core/examples/parallel-branch