graph-core 中断与检查点机制的设计与用法

CompileConfig.builder()
    .interruptBefore("dangerousNode")      // 执行前中断
    .interruptAfter("reviewNode")          // 执行后中断
    .interruptBeforeEdge(true)             // 边路由前中断
    .build();

.addNode("conditionalNode", (state, config) -> {
    if (isHighRisk(state)) {
        config.markNodeAsInterrupted("conditionalNode");  // 运行时决定中断
    }
    return Map.of("result", "...");
})

// NodeExecutor 执行流程
1. 执行节点 action
2. 合并状态 mergeIntoCurrentState(updateState)
3. 计算下一节点 nextNodeId
4. 保存 Checkpoint（自动）              // 关键：自动保存
5. 返回 NodeOutput

public class Checkpoint {
    private final String id;              // UUID（每次保存生成新 ID）
    private Map<String, Object> state;    // 完整状态快照
    private String nodeId;                // 当前节点 ID
    private String nextNodeId;            // 下一个节点 ID
}

public interface BaseCheckpointSaver {
    Collection<Checkpoint> list(RunnableConfig config);        // 列出所有 Checkpoint
    Optional<Checkpoint> get(RunnableConfig config);          // 获取当前 Checkpoint
    RunnableConfig put(RunnableConfig config, Checkpoint cp); // 保存 Checkpoint
    Tag release(RunnableConfig config);                       // 释放资源
}

RunnableConfig.builder()
    .metadata(CHECKPOINTS_NUM_RETAINED, 10)  // 只保留最近 10 个
    .build();

// 框架按配置清理最旧的
default void retainLatestCheckpoints(LinkedList<Checkpoint> checkpoints, RunnableConfig config) {
    checkpointsNumRetained(config).ifPresent(numRetained -> {
        while (checkpoints.size() > numRetained) {
            checkpoints.removeLast();
        }
    });
}

// 第一阶段：正常执行，遇到中断点暂停
workflow.stream(initialState, config).collectList().block();
// 返回 InterruptionMetadata，流暂停

// 用户审查、修改状态后，手动恢复
RunnableConfig resumeConfig = RunnableConfig.builder().resume().build();
workflow.stream(null, resumeConfig)  // 手动触发恢复
    .collectList().block();
// 从中断点继续执行

// CompiledGraph.java:309-332
public RunnableConfig updateState(RunnableConfig config, 
                                  Map<String, Object> values, 
                                  String asNode) throws Exception {
    // 1. 获取当前 Checkpoint
    Checkpoint checkpoint = saver.get(config).map(Checkpoint::copyOf).orElseThrow();
    
    // 2. 合并用户修改的值（使用 KeyStrategy）
    checkpoint = checkpoint.updateState(values, keyStrategyMap);
    
    // 3. 可选：指定下一个节点（覆盖原有路由）
    if (asNode != null) {
        Command nextCommand = nextNodeId(asNode, checkpoint.getState(), config);
        checkpoint = checkpoint.updateState(nextCommand.update(), keyStrategyMap);
    }
    
    // 4. 保存修改后的 Checkpoint
    return saver.put(config, checkpoint);
}

// 中断时的状态
// state: {"sql": "DELETE FROM users WHERE 1=1"}  // 错误

// 用户修正
config = workflow.updateState(config, Map.of(
    "sql", "DELETE FROM users WHERE id = 123"
));

// Resume 继续执行（基于修正后的状态）
workflow.stream(null, config.withResume());

public final class InterruptionMetadata extends NodeOutput {
    private final Map<String, Object> metadata;          // 自定义元数据
    private List<ToolCall> toolsAutomaticallyApproved;   // 工具调用列表
    private List<ToolFeedback> toolFeedbacks;            // 工具反馈
}

.interruptBeforeEdge(true)  // 单独使用无意义

.interruptAfter("B")         // 必须配合使用
.interruptBeforeEdge(true)   // 修饰 interruptAfter 行为

配置 interruptAfter("B")（默认）：
  B 执行 → 合并状态(messages="B") → 评估边 → nextNodeId="D" → 中断
  Resume → 直接执行 D
  updateState(messages="C")  无法改变路由（D 已固化）

配置 interruptAfter("B") + interruptBeforeEdge(true)：
  B 执行 → 合并状态(messages="B") → 中断（边未评估）
  Resume → 重新评估边 → 从 messages 读取
  updateState(messages="C")  可以改变路由（走 C）

checkpoint = saver.get(config)
currentNodeId = null                              // 重置
nextNodeId = checkpoint.getNextNodeId()           // 从 checkpoint 恢复
overallState = initialState.input(checkpoint.getState())  // 恢复状态

StateGraph codeReviewWorkflow = new StateGraph(factory)
    .addNode("generateCode", aiGenerateCode)
    .addNode("runTests", executeTests)
    .addNode("deploy", deployToProduction)
    .addEdge(START, "generateCode")
    .addEdge("generateCode", "runTests")
    .addEdge("runTests", "deploy")
    .addEdge("deploy", END)
    .compile(CompileConfig.builder()
        .interruptAfter("generateCode")  // 代码生成后人工审查
        .interruptBefore("deploy")       // 部署前二次确认
        .build());

StateGraph riskWorkflow = new StateGraph(factory)
    .addNode("calculateRisk", aiCalculateRisk)
    .addNode("approveOrReject", (state, config) -> {
        double riskScore = state.value("riskScore");
        if (riskScore > 80) {
            config.markNodeAsInterrupted("approveOrReject");  // 高风险需人工
        }
        return Map.of("autoDecision", riskScore < 30 ? "approve" : "pending");
    })
    .addNode("executeTransaction", executeTransaction)
    .compile(CompileConfig.builder()
        .interruptBefore("executeTransaction")  // 交易前最后确认
        .build());

class CheckpointedWorkflow {
    CheckpointSaver saver;
    Set<String> interruptPoints;
    
    void execute(State initialState, String sessionId) {
        State current = initialState;
        String nodeId = START;
        
        while (nodeId != END) {
            State newState = executeNode(nodeId, current);
            saver.save(sessionId, nodeId, newState);  // 每个节点自动保存
            
            if (interruptPoints.contains(nodeId)) {
                throw new InterruptException(nodeId, newState);  // 暂停并返回
            }
            
            nodeId = routeNext(nodeId, newState);
            current = newState;
        }
    }
    
    void resume(String sessionId, State modifications) {
        Checkpoint cp = saver.load(sessionId);
        State modifiedState = cp.state.merge(modifications);  // 应用用户修改
        execute(modifiedState, cp.nextNodeId, sessionId);
    }
}