一、Spring AI Alibaba Graph 是什么

Spring AI Alibaba Graph 是 Spring AI Alibaba 提供的工作流编排能力，用来把一个复杂 AI 任务拆成多个可执行节点，并通过共享状态、边、条件边、持久化检查点和中断恢复机制，把这些节点组织成一个可观测、可暂停、可恢复的执行图。它适合处理多步骤 Agent、人工审批、模型生成后处理、工具调用、异步回调等场景。

和普通的 service 串行调用相比，Graph 的核心价值在于：流程结构显式、节点职责清晰、状态可检查、执行可中断、失败后可基于检查点恢复。对于需要模型推理、外部系统调用、人工输入和后续动作组合在一起的系统，Graph 比单个方法链路更容易维护。

可以把 Spring AI Alibaba Graph 理解成一个面向 AI Agent 的状态机和工作流引擎。它不是替你完成模型调用，而是提供一套编排模型，让你把模型调用、数据查询、格式转换、外部动作、人工输入这些步骤组织成一张图。

这张图由几个核心概念组成：

概念	作用	常用 API
StateGraph	定义整张执行图，负责注册节点、边和条件边。	new StateGraph(...)
NodeAction	节点执行逻辑。读取当前状态，执行任务，返回状态更新。	implements NodeAction
EdgeAction	条件边路由逻辑。根据当前状态决定下一步进入哪个节点。	implements EdgeAction
OverAllState	Graph 的共享状态，节点通过它读取上下文。	state.value(...)
KeyStrategy	定义同一个状态 key 多次写入时如何合并。	ReplaceStrategy、KeyStrategyFactory
CompiledGraph	编译后的可执行图，提供 invoke、stream、getState 等执行能力。	stateGraph.compile(...)
CompileConfig	编译配置，常用于配置中断点和状态保存器。	CompileConfig.builder()
SaverConfig	注册 CheckpointSaver，用于保存 Graph 执行状态。	SaverConfig.builder().register(...)
RunnableConfig	运行配置，常用于指定 threadId 和 resume。	RunnableConfig.builder()

二、核心抽象：节点、状态和边

1. 节点：每个步骤只做一件事

Graph 的节点通常实现 NodeAction。节点方法接收 OverAllState，从状态中读取输入，执行自己的逻辑，然后返回一个 Map 作为状态更新。这个设计让节点天然具备函数式特征：输入是状态，输出也是状态的一部分。

public class ReviewNode implements NodeAction {
    @Override
    public Map<String, Object> apply(OverAllState state) throws Exception {
        String nextStep = StateGraph.END;
        boolean reviewApproved = state.value("reviewApproved", false);
        if (reviewApproved) {
            nextStep = "createNode";
        }
        return Map.of("reviewNextStep", nextStep);
    }
}

实现节点时，应避免把太多职责塞进一个 NodeAction。模型推理、JSON 提取、邮件发送、落库、人工判断这些动作如果混在一个节点里，后续调试和恢复都会变得困难。更合理的方式是按能力边界拆分节点，让每个节点的输入输出更稳定。

2. 状态：Graph 的共享记忆

Graph 中的状态不是普通局部变量，而是整个执行图的共享记忆。每个节点都可以读取状态，也可以返回新的状态片段。状态中应该保存业务事实和中间结果，而不是保存拼好的 Prompt 或临时展示文案。

通常可以通过 KeyStrategyFactory 注册状态 key，并给每个 key 指定合并策略。常见做法是对当前值采用 ReplaceStrategy，即后一次写入覆盖前一次写入。

KeyStrategyFactory keyStrategyFactory = () -> { HashMap<String, KeyStrategy> strategy = new HashMap<>(); strategy.put("bizId", new ReplaceStrategy()); strategy.put("inputData", new ReplaceStrategy()); strategy.put("currentData", new ReplaceStrategy()); strategy.put("contextData", new ReplaceStrategy()); strategy.put("generatedText", new ReplaceStrategy()); strategy.put("structuredResult", new ReplaceStrategy()); strategy.put("reviewNextStep", new ReplaceStrategy()); strategy.put("threadId", new ReplaceStrategy()); return strategy; };

状态设计的关键是稳定。只要状态 key 命名清晰、含义明确，节点之间就可以通过状态解耦。后续要替换某个节点的实现，只要它继续读取和写入同样的状态字段，整张图就不需要大改。

3. 边：表达固定流程和动态路由

Graph 中有两类边。普通边表示固定流转，例如 START 到第一个节点、A 节点到 B 节点。条件边表示动态路由，例如根据审批结果、模型分类结果、风险判断结果决定下一步。

stateGraph.addEdge(StateGraph.START, "inputDataNode"); stateGraph.addEdge("generateNode", "parseNode"); stateGraph.addConditionalEdges( "reviewNode", AsyncEdgeAction.edge_async(new ReviewEdge()), Map.of("createNode", "createNode", StateGraph.END, StateGraph.END) );

条件边通常配合 EdgeAction 使用。EdgeAction 读取状态并返回下一步节点名称，Graph 再根据 addConditionalEdges 中配置的映射找到真实目标节点。

public class ReviewEdge implements EdgeAction { @Override public String apply(OverAllState state) throws Exception { String next = state.value("reviewNextStep", ""); if (next.equals("createNode")) { return "createNode"; } return StateGraph.END; } }

三、如何实现一张可执行 Graph

实现 Spring AI Alibaba Graph 一般分四步：定义状态 key，创建 StateGraph，注册节点和边，编译成 CompiledGraph。

1. 创建 StateGraph

StateGraph 构造时传入图名称和 KeyStrategyFactory。图名称主要用于标识和可视化，KeyStrategyFactory 决定状态字段的合并策略。

StateGraph stateGraph = new StateGraph("aiWorkflowGraph", keyStrategyFactory);

2. 注册节点

节点注册时，需要把 NodeAction 包装成异步节点动作。通常使用 AsyncNodeAction.node_async(...) 注册节点。

stateGraph.addNode("generateNode", AsyncNodeAction.node_async(new PredictNode(chatClient))); stateGraph.addNode("parseNode", AsyncNodeAction.node_async(new ExtractNode(chatClient))); stateGraph.addNode("reviewNode", AsyncNodeAction.node_async(new ReviewNode()));

模型调用节点可以注入 ChatClient，数据节点可以注入 mapper 或 service，动作节点可以注入邮件、消息、数据库等外部依赖。Graph 不限制节点内部做什么，但建议每个节点保持单一职责。

3. 注册边和条件边

边决定执行顺序。普通边用于固定步骤，条件边用于分支。条件边的路由结果最好由前一个节点写入状态，这样路由逻辑可观察、可测试。

stateGraph.addEdge(StateGraph.START, "inputDataNode"); stateGraph.addEdge(StateGraph.START, "contextDataNode"); stateGraph.addEdge("inputDataNode", "generateNode"); stateGraph.addEdge("contextDataNode", "generateNode"); stateGraph.addEdge("generateNode", "parseNode"); stateGraph.addEdge("parseNode", "notifyNode"); stateGraph.addEdge("notifyNode", "reviewNode");

4. 编译 Graph

最简单的 Graph 可以直接 compile。只要涉及人工输入、跨请求恢复、应用重启恢复，就应该在编译时配置 SaverConfig 和 CompileConfig。

CompiledGraph compiledGraph = stateGraph.compile();

四、状态持久化：CheckpointSaver 的作用

Graph 的持久化通过 CheckpointSaver 完成，并在编译 Graph 时注册到 SaverConfig。状态持久化解决的是“流程执行到一半必须停下来，稍后再继续”的问题。

本地调试可以使用 MemorySaver，但它只适合进程内调试，应用重启后状态会丢失。生产环境一般应使用 Redis、MySQL 等外部存储。以数据库型 Saver 为例，通常会保存 Graph 线程和检查点信息，用于在跨请求、跨实例或应用重启后恢复执行状态。

RedisSaver redisSaver = RedisSaver.builder() .redisson(redissonClient) .build(); SaverConfig saverConfig = SaverConfig.builder() .register(redisSaver) .build();

CheckpointSaver 的意义不是简单缓存，而是保存 Graph 在某个 threadId 下的执行状态。后续恢复时，Graph 可以根据同一个 threadId 找回状态和执行位置。

五、中断与恢复：interruptBefore、threadId、resume

1. interruptBefore：在指定节点前暂停

当流程需要人工输入或外部异步回调时，可以使用 interruptBefore 配置中断点。Graph 执行到该节点前会暂停，并把当前状态写入 Saver。

CompileConfig compileConfig = CompileConfig.builder() .interruptBefore("reviewNode") .saverConfig(saverConfig) .build(); CompiledGraph compiledGraph = stateGraph.compile(compileConfig);

这个机制适合人工审批、表单补充、二次确认、外部系统回调等场景。关键是中断点不是写在节点内部的阻塞等待，而是在 Graph 编译配置中声明。

2. threadId：一次流程实例的唯一锚点

RunnableConfig.threadId 用来标识一次 Graph 执行实例。只要流程需要恢复，就必须使用稳定 threadId。它最好来自业务单号、审批单号、任务实例 ID 或流程实例 ID，不要使用每次都变化的随机值。

RunnableConfig config = RunnableConfig.builder() .threadId(threadId) .build(); compiledGraph.invoke(initialState, config);

没有稳定 threadId，后续回调就无法定位之前保存的 checkpoint。很多恢复失败问题，本质上都是首次执行和恢复执行使用了不同 threadId。

3. getState 和 resume：恢复中断流程

外部输入回来后，可以先通过 getState 获取当前检查点，再更新状态，然后使用 resume 继续执行。

RunnableConfig config = RunnableConfig.builder() .threadId(threadId) .build(); StateSnapshot stateSnapshot = compiledGraph.getState(config); OverAllState state = stateSnapshot.state(); state.updateState(Map.of("reviewApproved", reviewApproved)); RunnableConfig resumeConfig = RunnableConfig.builder() .threadId(threadId) .resume() .build(); compiledGraph.invoke(state, resumeConfig);

另一种常见写法是使用 updateState 把人工输入写回 Graph 状态，再用 stream 或 invoke 基于更新后的配置继续执行。无论使用哪种方式，核心都是三件事：同一个 threadId、写回必要状态、从 checkpoint 继续执行。

六、stream 与可观测性

CompiledGraph 不只支持 invoke，也支持 stream。stream 可以在执行过程中返回节点输出，适合把 Agent 执行进度推给前端，或者用于调试和日志观察。

compiledGraph.stream(initialState, config) .doOnNext(output -> log.info("node output: {}", output)) .blockLast();

在涉及中断点的流程中，stream 的另一个价值是可以明确看到流程执行到哪个节点后暂停。结合 getState，就能检查当前状态里已经有哪些中间结果。

七、Graph 可视化

也可以使用 GraphRepresentation 输出 PlantUML，用来查看 Graph 结构。对复杂流程来说，图形化结构非常有用：可以确认 START、END、普通边、条件边是否符合预期，也方便排查节点之间是否存在意外连接。

GraphRepresentation graphRepresentation = compiledGraph.getGraph( GraphRepresentation.Type.PLANTUML, "workflowGraph", true ); log.info("Graph PlantUML code: {}", graphRepresentation.content());

八、版本差异：humanFeedback 的迁移

在早期版本中，OverAllState 曾经提供 humanFeedback 方法。从 v1.1.0.0-M4 开始，该方法已经被移除。新版推荐使用 interruptBefore、CheckpointSaver、RunnableConfig、updateState 或 resume 这套机制处理人工反馈。

这意味着人工输入不再是状态对象上的一个特殊方法，而是 Graph 执行协议的一部分：先在指定节点前中断，保存状态；人工输入回来后，按 threadId 找到状态；写入新状态；继续执行后续节点。

九、实现时容易踩的点

1. 状态 key 没有注册策略，导致节点写入状态后被过滤或合并行为不符合预期。

2. threadId 使用随机值，恢复时找不到之前的 checkpoint。

3. 把人工等待写成节点内阻塞，而不是用 interruptBefore 声明中断点。

4. 节点粒度太粗，导致失败后需要重复执行大量逻辑，也不利于观察中间状态。

5. 条件边返回值和 addConditionalEdges 的映射 key 不一致，导致路由失败。

6. 只用 MemorySaver 做验证，却误以为应用重启后状态也能恢复。

7. 回调接口没有幂等控制，重复回调可能导致流程被重复推进。

十、总结

Spring AI Alibaba Graph 核心是用于把复杂 AI 流程的执行结构显式化。节点负责单步能力，状态负责跨节点传递上下文，边负责流程走向，CheckpointSaver 负责持久化，interruptBefore 和 resume 负责人工介入后的恢复。

当一个 AI 功能开始涉及多次模型调用、外部系统动作、条件分支、人工确认或异步回调时，就应该考虑用 Graph 来组织流程。这样可以避免把所有逻辑堆在一个 service 方法里，也能让流程更容易观察、恢复和演进。