接上节继续，本篇将学习如何实现并行工作流。
上面这张图，用代码很容易绘制，参考以下代码。
public static StateGraph<AgentState> getParallelGraph() throws GraphStateException {
return new StateGraph<>(AgentState::new)
.addNode("node-1", node_async(new Node1Action()))
.addNode("node-2", node_async(new Node2Action()))
.addNode("node-3", node_async(new Node3Action()))
.addEdge(START, "node-1")
.addEdge("node-1", "node-2")
.addEdge("node-1", "node-3")
.addEdge("node-2", GraphDefinition.END)
.addEdge("node-3", GraphDefinition.END);
}
虽然图上看着貌似node-2,node-3并行在跑，但真的如此吗？我们把node-2和node-3的apply()里加点sleep
public class Node2Action implements NodeAction<AgentState> {
@Override
public Map<String, Object> apply(AgentState state) throws Exception {
System.out.println("current Node: node-2");
Thread.sleep(1000);
return Map.of("myData", "node2-my-value",
"node2Key", "node2-value");
}
}
public class Node3Action implements NodeAction<AgentState> {
@Override
public Map<String, Object> apply(AgentState state) throws Exception {
System.out.println("current Node: node-3");
Thread.sleep(1000);
return Map.of("myData", "node3-my-value",
"node3Key", "node3-value");
}
}
然后在node-1里，记录下start时间戳
public class Node1Action implements NodeAction<AgentState> {
@Override
public Map<String, Object> apply(AgentState state) throws Exception {
System.out.println("current Node: node-1");
Thread.sleep(1000);
return Map.of(
"myData", "node1-my-value",
"node1Key", "node1-value",
//记录开始时间
"start", System.currentTimeMillis());
}
}
getParallelGraph().compile()
.invoke(Map.of("test", "test-init-value"))
.ifPresent(c -> {
long start = (long) c.data().getOrDefault("start", 0L);
System.out.println(c.data());
long end = System.currentTimeMillis();
System.out.println((end - start) + "ms");
});
current Node: node-1
current Node: node-2
current Node: node-3
{node1Key=node1-value, start=1770719927373, test=test-init-value, node2Key=node2-value, node3Key=node3-value, myData=node3-my-value}
2017msLangGraph4可以手动指定线程池实现真正的并发处理。
StateGraph<AgentState> graphNoThreadPool = getParallelGraph();
ExecutorService executorService = Executors.newFixedThreadPool(2);
RunnableConfig rc = RunnableConfig.builder()
//从node-1开始并行执行node-2和node-3（使用线程池）
.addParallelNodeExecutor("node-1", executorService)
.build();
graphNoThreadPool.compile()
.invoke(Map.of("test", "test-init-value"), rc) //调用时，使用特定的RunnableConfig
.ifPresent(c -> {
long start = (long) c.data().getOrDefault("start", 0L);
System.out.println(c.data());
long end = System.currentTimeMillis();
System.out.println((end - start) + "ms");
//记得关闭线程池
executorService.shutdown();
});
运行结果
current Node: node-1
current Node: node-2
current Node: node-3
{node1Key=node1-value, start=1770722528938, test=test-init-value, node2Key=node2-value, node3Key=node3-value, myData=node3-my-value}
1015ms
明显快多了。如果是jdk 25版本，也可以使用虚拟线程：
ExecutorService virtualThreadPerTaskExecutor = Executors.newVirtualThreadPerTaskExecutor();
RunnableConfig rc2 = RunnableConfig.builder()
//从node-1开始并行执行node-2和node-3（使用线程池）
.addParallelNodeExecutor("node-1", virtualThreadPerTaskExecutor)
.build();
graphNoThreadPool2.compile()
.invoke(Map.of("test", "test-init-value"), rc2)
.ifPresent(c -> {
long start = (long) c.data().getOrDefault("start", 0L);
System.out.println(c.data());
long end = System.currentTimeMillis();
System.out.println((end - start) + "ms");
});
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。