Spring AI Alibaba 80｜Graph Core · 人类反馈 · 模式二：interruptBefore 模式

interruptBefore 模式需要在编译 Graph 时提前指定中断点，在指定节点执行前自动中断。这种方式适合已知的中断点，配置简单直接。

优势

• 配置简单：只需在编译配置中指定中断点
• 无需修改节点：普通节点即可，不需要实现特殊接口
• 明确的中断点：中断位置在编译时确定，易于理解和维护

定义带中断的 Graph

import com.alibaba.cloud.ai.graph.CompileConfig;
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.checkpoint.config.SaverConfig;
import com.alibaba.cloud.ai.graph.checkpoint.savers.MemorySaver;
import com.alibaba.cloud.ai.graph.exception.GraphStateException;
import com.alibaba.cloud.ai.graph.state.strategy.AppendStrategy;
import com.alibaba.cloud.ai.graph.state.strategy.ReplaceStrategy;

import java.util.HashMap;
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.AsyncEdgeAction.edge_async;
import static com.alibaba.cloud.ai.graph.action.AsyncNodeAction.node_async;

/**
* 定义带中断的 Graph
* 使用 interruptBefore 配置在指定节点前中断
*/
public static CompiledGraph createGraphWithInterrupt() throws GraphStateException {
  // 定义节点
  var step1 = node_async(state -> {
      return Map.of("messages", "Step 1");
  });

  var humanFeedback = node_async(state -> {
      return Map.of(); // 等待用户输入，不修改状态
  });

  var step3 = node_async(state -> {
      return Map.of("messages", "Step 3");
  });

  // 定义条件边
  var evalHumanFeedback = edge_async(state -> {
      var feedback = (String) state.value("human_feedback").orElse("unknown");
      return (feedback.equals("next") || feedback.equals("back")) ? feedback : "unknown";
  });

  // 配置 KeyStrategyFactory
  KeyStrategyFactory keyStrategyFactory = () -> {
      HashMap<String, KeyStrategy> keyStrategyHashMap = new HashMap<>();
      keyStrategyHashMap.put("messages", new AppendStrategy());
      keyStrategyHashMap.put("human_feedback", new ReplaceStrategy());
      return keyStrategyHashMap;
  };

  // 构建 Graph
  StateGraph builder = new StateGraph(keyStrategyFactory)
          .addNode("step_1", step1)
          .addNode("human_feedback", humanFeedback)
          .addNode("step_3", step3)
          .addEdge(START, "step_1")
          .addEdge("step_1", "human_feedback")
          .addConditionalEdges("human_feedback", evalHumanFeedback,
                  Map.of("back", "step_1", "next", "step_3", "unknown", "human_feedback"))
          .addEdge("step_3", END);

  // 配置内存保存器和中断点
  var saver = new MemorySaver();

  var compileConfig = CompileConfig.builder()
          .saverConfig(SaverConfig.builder()
                  .register(saver)
                  .build())
          .interruptBefore("human_feedback") // 在 human_feedback 节点前中断
          .build();

  return builder.compile(compileConfig);
}

执行 Graph 直到中断

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

import java.util.Map;

import reactor.core.publisher.Flux;

/**
* 执行 Graph 直到中断
*/
public static void executeUntilInterrupt(CompiledGraph graph) {
  // 初始输入
  Map<String, Object> initialInput = Map.of("messages", "Step 0");

  // 配置线程 ID
  var invokeConfig = RunnableConfig.builder()
          .threadId("Thread1")
          .build();

  // 运行 Graph 直到第一个中断点
  graph.stream(initialInput, invokeConfig)
          .doOnNext(event -> System.out.println(event))
          .doOnError(error -> System.err.println("流错误: " + error.getMessage()))
          .doOnComplete(() -> System.out.println("流完成"))
          .blockLast();
}

输出:

NodeOutput{node=__START__, state={messages=[Step 0]}}
NodeOutput{node=step_1, state={messages=[Step 0, Step 1]}}
流完成

等待用户输入并更新状态

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

import java.util.Map;

/**
* 等待用户输入并更新状态
*/
public static RunnableConfig waitUserInputAndUpdateState(CompiledGraph graph) throws Exception {
  var invokeConfig = RunnableConfig.builder()
          .threadId("Thread1")
          .build();

  // 检查当前状态
  System.out.printf("--State before update--
%s
", graph.getState(invokeConfig));

  // 模拟用户输入
  var userInput = "back"; // "back" 表示返回上一个节点
  System.out.printf("
--User Input--
用户选择: '%s'

", userInput);

  // 更新状态（模拟 human_feedback 节点的输出）
  // 注意：interruptBefore 模式下，传入 null 作为节点 ID
  var updateConfig = graph.updateState(invokeConfig, Map.of("human_feedback", userInput), null);

  // 检查更新后的状态
  System.out.printf("--State after update--
%s
", graph.getState(updateConfig));

  return updateConfig;
}

输出:

--State before update--
StateSnapshot{node=step_1, state={messages=[Step 0, Step 1]}, config=RunnableConfig{ threadId=Thread1, nextNode=human_feedback }}

--User Input--
用户选择: 'back'

--State after update--
StateSnapshot{node=step_1, state={messages=[Step 0, Step 1], human_feedback=back}, config=RunnableConfig{ threadId=Thread1, nextNode=human_feedback }}

继续执行 Graph

/**
* 继续执行 Graph
*/
public static void continueExecution(CompiledGraph graph, RunnableConfig updateConfig) {
  // 继续执行 Graph（input 为 null，使用之前的状态）
  graph.stream(null, updateConfig)
          .doOnNext(event -> System.out.println(event))
          .doOnError(error -> System.err.println("流错误: " + error.getMessage()))
          .doOnComplete(() -> System.out.println("流完成"))
          .blockLast();
}

输出:

NodeOutput{node=human_feedback, state={messages=[Step 0, Step 1], human_feedback=back}}
NodeOutput{node=step_1, state={messages=[Step 0, Step 1], human_feedback=back}}
流完成

第二次等待用户输入

var userInput = "next"; // "next" 表示继续下一个节点
System.out.printf("
--User Input--
用户选择: '%s'
", userInput);

// 更新状态
var updateConfig = graph.updateState(invokeConfig, Map.of("human_feedback", userInput), null);

System.out.printf("
getNext()
	with invokeConfig:[%s]
	with updateConfig:[%s]
",
  graph.getState(invokeConfig).getNext(),
  graph.getState(updateConfig).getNext());

继续执行直到完成

// 继续执行 Graph
graph.stream(null, updateConfig)
      .doOnNext(event -> System.out.println(event))
      .doOnError(error -> System.err.println("流错误: " + error.getMessage()))
      .doOnComplete(() -> System.out.println("流完成"))
      .blockLast();

输出:

NodeOutput{node=human_feedback, state={messages=[Step 0, Step 1], human_feedback=next}}
NodeOutput{node=step_3, state={messages=[Step 0, Step 1, Step 3], human_feedback=next}}
NodeOutput{node=__END__, state={messages=[Step 0, Step 1, Step 3], human_feedback=next}}
流完成

• 需要人工审核的审批流程：在关键节点等待人工审核
• 需要用户确认的关键操作：在执行重要操作前等待用户确认
• 交互式对话系统：在对话过程中等待用户输入
• 多步骤表单填写：在表单的每个步骤之间等待用户输入
• 动态工作流：根据用户反馈动态调整工作流路径

本文整理自 Spring AI Alibaba 官方文档《人类反馈》：https://java2ai.com/docs/frameworks/graph-core/examples/human-in-the-loop