Spring AI Alibaba 104｜Graph Core · 时光旅行与状态回溯

Spring AI Alibaba Graph 支持时光旅行功能，允许您查看和恢复 Graph 执行的历史状态。

要启用时光旅行，需要配置 Checkpointer：

import com.alibaba.cloud.ai.graph.CompileConfig;
import com.alibaba.cloud.ai.graph.CompiledGraph;
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;

/**
 * 配置 Checkpoint
 */
public static CompiledGraph configureCheckpoint(StateGraph stateGraph) throws GraphStateException {
// 创建 Checkpointer
var checkpointer = new MemorySaver();

// 配置持久化
var compileConfig = CompileConfig.builder()
            .saverConfig(SaverConfig.builder()
                    .register(checkpointer)
                    .build())
    .build();

    return stateGraph.compile(compileConfig);
}

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

import java.util.Map;

/**
 * 执行 Graph 并生成历史
 */
public static void executeGraphAndGenerateHistory(CompiledGraph graph) {
// 配置线程 ID
var config = RunnableConfig.builder()
    .threadId("conversation-1")
    .build();

// 执行 Graph
Map<String, Object> input = Map.of("query", "Hello");
    graph.invoke(input, config);

// 再次执行
graph.invoke(Map.of("query", "Follow-up question"), config);
}

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

import java.util.List;

/**
 * 查看状态历史
 */
public static void viewStateHistory(CompiledGraph graph) {
    var config = RunnableConfig.builder()
            .threadId("conversation-1")
            .build();

// 获取所有历史状态
    List<StateSnapshot> history = (List<StateSnapshot>) graph.getStateHistory(config);

System.out.println("State history:");
for (int i = 0; i < history.size(); i++) {
    StateSnapshot snapshot = history.get(i);
    System.out.printf("Step %d: %sn", i, snapshot.state());
        System.out.printf("  Checkpoint ID: %sn", snapshot.config().checkPointId().orElse("N/A"));
    System.out.printf("  Node: %sn", snapshot.node());
    }
}

输出示例:

State history:
Step 0: {query=Follow-up question, answer=Response to follow-up}
  Checkpoint ID: abc123
  Node: answer_node
Step 1: {query=Hello, answer=Initial response}
  Checkpoint ID: def456
  Node: answer_node
Step 2: {query=Hello}
  Checkpoint ID: ghi789
  Node: __START__

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

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

/**
 * 回溯到历史状态
 */
public static void travelBackToHistory(CompiledGraph graph) {
    var config = RunnableConfig.builder()
            .threadId("conversation-1")
            .build();

    // 获取所有历史状态
    List<StateSnapshot> history = (List<StateSnapshot>) graph.getStateHistory(config);

// 获取特定的历史状态
StateSnapshot historicalSnapshot = history.get(1);

// 使用历史状态的 checkpoint ID 创建新配置
var historicalConfig = RunnableConfig.builder()
    .threadId("conversation-1")
            .checkPointId(historicalSnapshot.config().checkPointId().orElse(null))
    .build();

// 从历史状态继续执行
graph.invoke(
    Map.of("query", "New question from historical state"),
    historicalConfig
);
}

基于历史状态创建新的执行分支：

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

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

/**
 * 分支创建
 */
public static void createBranch(CompiledGraph graph) {
    var config = RunnableConfig.builder()
            .threadId("conversation-1")
            .build();

    // 获取所有历史状态
    List<StateSnapshot> history = (List<StateSnapshot>) graph.getStateHistory(config);

    // 获取特定的历史状态
    StateSnapshot historicalSnapshot = history.get(1);

// 从历史状态创建新分支
var branchConfig = RunnableConfig.builder()
    .threadId("conversation-1-branch")  // 新的线程 ID
            .checkPointId(historicalSnapshot.config().checkPointId().orElse(null))
    .build();

// 在新分支上执行
    graph.invoke(
    Map.of("query", "Alternative path"),
    branchConfig
);
}

1. A/B 测试: 从相同起点测试不同路径
2. 错误恢复: 回溯到错误发生前的状态
3. 调试分析: 检查每个步骤的状态变化
4. 用户撤销: 允许用户撤销操作
5. 实验对比: 比较不同决策的结果

StateSnapshot 包含以下信息：

public interface StateSnapshot {
    // 状态数据
    Map<String, Object> state();

    // 检查点配置
    RunnableConfig config();

    // 执行的节点名称
    String node();

    // 下一个要执行的节点
    String getNext();
}

本文整理自 Spring AI Alibaba 官方文档《时光旅行与状态回溯》：https://java2ai.com/docs/frameworks/graph-core/examples/time-travel