Spring AI Alibaba 76｜Graph Core · 图执行取消 · 取消图流

Spring AI Alibaba Graph 提供了强大的图执行取消机制，这对于长时间运行的工作流程特别有用。此功能基于 java-async-generator 库的取消能力构建。

当您使用 stream 方法执行图时，会得到一个 AsyncGenerator 实例。此生成器可以被取消，允许您优雅地或立即停止图的执行。

要取消流，需要在生成器上调用 cancel(boolean mayInterruptIfRunning) 方法。

在下面的示例中，我们从与主线程不同的线程发起取消请求来测试取消功能。

使用 forEachAsync 消费流

当您使用 forEachAsync 消费流时，图执行在单独的线程中运行。

• cancel(true) (立即取消): 这将中断执行线程，导致 forEachAsync 返回的 CompletableFuture 以 InterruptedException 异常完成。
• cancel(false) (优雅取消): 这将让当前正在执行的节点完成，然后在启动下一个节点之前停止执行。

以下是一个演示立即取消的示例：

import com.alibaba.cloud.ai.graph.CompiledGraph;
import com.alibaba.cloud.ai.graph.NodeOutput;
import com.alibaba.cloud.ai.graph.RunnableConfig;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

import reactor.core.Disposable;
import reactor.core.publisher.Flux;

// 编译并获取图实例
CompiledGraph compiledGraph = stateGraph.compile();

// 创建运行配置
RunnableConfig runnableConfig = RunnableConfig.builder()
      .threadId("test-thread")
      .build();

// 准备输入数据
Map<String, Object> inputData = new HashMap<>();
// ... 添加输入数据

// 执行图并获取流
Flux<NodeOutput> stream = compiledGraph.stream(inputData, runnableConfig);

// 从新线程在 500 毫秒后请求取消
CompletableFuture.runAsync(() -> {
  try {
      Thread.sleep(500);
      // Flux 使用 dispose() 来取消
      System.out.println("请求取消执行");
  }
  catch (InterruptedException e) {
      throw new RuntimeException(e);
  }
});

// 异步处理每个输出
var disposable = stream.subscribe(
      output -> System.out.println("当前迭代节点: " + output),
      error -> System.out.println("流错误: " + error.getMessage()),
      () -> System.out.println("流完成")
);

// 等待流完成或取消
try {
  stream.blockLast();
}
catch (Exception e) {
  System.err.println("执行异常: " + e.getMessage());
}

// 验证是否已取消（Flux 使用 isDisposed 检查）
System.out.println("是否已取消: " + disposable.isDisposed());

使用迭代器消费流

当您使用 for-each 循环遍历流时，执行在当前线程上运行。

• cancel(true) 或 cancel(false): 两者都会导致迭代器的 hasNext() 方法返回 false，从而有效地停止循环。

以下是一个示例：

import com.alibaba.cloud.ai.graph.CompiledGraph;
import com.alibaba.cloud.ai.graph.NodeOutput;
import com.alibaba.cloud.ai.graph.RunnableConfig;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

import reactor.core.Disposable;
import reactor.core.publisher.Flux;

// 编译并获取图实例
CompiledGraph compiledGraph = stateGraph.compile();

// 创建运行配置
RunnableConfig runnableConfig = RunnableConfig.builder()
      .threadId("test-thread")
      .build();

// 准备输入数据
Map<String, Object> inputData = new HashMap<>();
// ... 添加输入数据

// 执行图并获取流
Flux<NodeOutput> stream = compiledGraph.stream(inputData, runnableConfig);

// 从新线程在 500 毫秒后请求取消
var disposable = stream.subscribe(
      output -> {
          System.out.println("当前迭代节点: " + output);
      },
      error -> {
          System.out.println("流错误: " + error.getMessage());
      },
      () -> {
          System.out.println("流完成");
      }
);

CompletableFuture.runAsync(() -> {
  try {
      Thread.sleep(500);
      disposable.dispose(); // 取消流
      System.out.println("已请求取消执行");
  }
  catch (InterruptedException e) {
      throw new RuntimeException(e);
  }
});

// 等待流完成或取消
try {
  stream.blockLast();
}
catch (Exception e) {
  System.out.println("流被中断: " + e.getMessage());
}

// 验证取消状态
System.out.println("是否已取消: " + disposable.isDisposed());

您可以通过在生成器上调用 isCancelled() 方法来检查流是否已被取消。

import reactor.core.Disposable;

public static void checkCancellationStatus(Disposable disposable) {
  if (disposable.isDisposed()) {
      System.out.println("流已被取消");
  }
  else {
      System.out.println("流仍在运行");
  }
}

取消功能同样适用于嵌套图（子图）。如果您取消父图的执行，取消操作会传播到当前正在执行的任何子图。

取消机制由 java-async-generator 库提供。有关底层取消机制的更深入说明，请参阅 java-async-generator 仓库中的 CANCELLATION.md 文档。

本文整理自 Spring AI Alibaba 官方文档《图执行取消》：https://java2ai.com/docs/frameworks/graph-core/examples/cancellation