Spring AI Alibaba Graph 入门实践：理解黑板、节点与边

在智能体或复杂 AI 工作流中，业务步骤往往不是简单的顺序调用，而是需要共享状态、条件分支、并行处理和中断恢复。Spring AI Alibaba Graph 将这些过程组织成可编译、可执行的状态图。下面从一个最小示例出发，逐步理解黑板、节点和边如何配合完成流程编排。

Spring AI Alibaba Graph 快速上手：黑板、节点、边

版本：基于 spring-ai-alibaba 1.0.0.4（配合 Spring AI 1.0.3）

环境准备

<dependency>
    <groupId>com.alibaba.cloud.ai</groupId>
    <artifactId>spring-ai-alibaba-graph-core</artifactId>
    <version>1.0.0.4</version>
</dependency>

---

第 1 层 · 最小骨架

先不谈业务。下面这 40 行是一张完整可运行的图。

输入一段文本，超过 10 个字符就截断，否则原样返回。

import com.alibaba.cloud.ai.graph.*;
import com.alibaba.cloud.ai.graph.action.AsyncEdgeAction;
import com.alibaba.cloud.ai.graph.action.AsyncNodeAction;
import com.alibaba.cloud.ai.graph.state.strategy.ReplaceStrategy;

import java.util.Map;

public class MinimalDemo {

    public static void main(String[] args) throws Exception {

        // ① 黑板：声明有哪些 key，以及多个节点写同一个 key 时怎么合并
        KeyStrategyFactory keys = () -> Map.of(
                "text",   new ReplaceStrategy(),
                "size",   new ReplaceStrategy(),
                "result", new ReplaceStrategy());

        // ② 节点：只读写黑板
        AsyncNodeAction measure = AsyncNodeAction.node_async(state -> {
            String text = state.value("text", "");
            return Map.of("size", text.length() > 10 ? "LONG" : "SHORT");
        });

        AsyncNodeAction keep = AsyncNodeAction.node_async(state -> {
            String text = state.value("text", "");
            return Map.of("result", text);
        });

        AsyncNodeAction cut = AsyncNodeAction.node_async(state -> {
            String text = state.value("text", "");
            return Map.of("result", text.substring(0, 10) + "...");
        });

        // ③ 路由函数：读黑板，返回一个「标签」
        AsyncEdgeAction route = AsyncEdgeAction.edge_async(
                state -> state.value("size", "SHORT"));

        // ④ 组装成图
        CompiledGraph graph = new StateGraph("demo", keys)
                .addNode("measure", measure)
                .addNode("keep",    keep)
                .addNode("cut",     cut)
                .addEdge(StateGraph.START, "measure")
                .addConditionalEdges("measure", route,
                        Map.of("LONG",  "cut",
                               "SHORT", "keep"))
                .addEdge("keep", StateGraph.END)
                .addEdge("cut",  StateGraph.END)
                .compile();

        // ⑤ 跑两次，观察分支
        System.out.println(graph.call(Map.of("text", "hi")).get().data());
        System.out.println(graph.call(Map.of("text", "Hello, Spring AI Alibaba Graph!")).get().data());
    }
}

跑起来会看到两次输出，第二次走了 cut 分支：

{text=hi, size=SHORT, result=hi}
{text=Hello, Spring AI Alibaba Graph!, size=LONG, result=Hello, Spr...}

这张图长什么样

START ──► measure ──┬── "SHORT" ──► keep ──► END
                    └── "LONG"  ──► cut  ──► END

核心概念

API	在上面代码里的位置	一句话说明
OverAllState	state.value(...) 读、return Map.of(...) 写	节点之间共享的黑板
addNode	.addNode("measure", measure)	注册一个「干活的单元」
addEdge	.addEdge(StateGraph.START, "measure")	注册一条固定跳转
addConditionalEdges	.addConditionalEdges("measure", route, Map.of(...))	按黑板内容决定走哪条路

---

第 2 层 · 逐个拆解

2.1 OverAllState：黑板

一个 Map<String, Object>，外加「每个 key 怎么合并」的规则。

节点 A 不直接调用节点 B。A 把结果写到黑板上，B 从黑板上读。节点之间互不认识

声明黑板上有哪些格子

KeyStrategyFactory keys = () -> Map.of(
        "text",   new ReplaceStrategy(),
        "size",   new ReplaceStrategy(),
        "result", new ReplaceStrategy());

KeyStrategyFactory 是个 @FunctionalInterface，返回 Map<String, KeyStrategy>。写成匿名内部类等价：

KeyStrategyFactory keys = new KeyStrategyFactory() {
    @Override
    public Map<String, KeyStrategy> apply() {
        return Map.of("text", new ReplaceStrategy());
    }
};

框架内置三种策略：

策略	合并语义	典型场景
ReplaceStrategy	新值覆盖旧值	本文两个例子全用它
AppendStrategy	追加到 List 尾部	对话历史、消息流水
MergeStrategy	Map 深度合并	多个节点往同一个 Map 上写字段

ReplaceStrategy：发生在每个节点执行完、框架把该节点返回的 Map 合并进黑板的那一刻——apply(旧值, 新值) 的实现就是 return 新值。被换掉的是这个 key 上整个 value 对象。

选型口诀：中间结果用 Replace，对话历史/日志用 Append，多节点共写一个 Map 用 Merge。

没声明的 key 会怎样？ 默认走 ReplaceStrategy

纯覆盖场景理论上可以不声明。但建议写全：显式声明等于给黑板列了一份 schema，哪个节点读哪些 key、写哪些 key，一眼就能看明白。

读：三个重载

String text = state.value("text", "");                       // 带默认值，返回原始类型
Optional<Object> v = state.value("text");                    // 单参版本，返回 Optional
Optional<String> s = state.value("text", String.class);      // 指定类型，返回 Optional

日常用得最多的是第一个

写：不是 set，是 return

节点写黑板，不调用任何 set 方法，而是返回一个 Map。框架拿到这个 Map，按每个 key 的 KeyStrategy 合并进黑板。

看个例子——这个节点只干一件事，把 count 加一：

AsyncNodeAction increment = AsyncNodeAction.node_async(state -> {
    int count = state.value("count", 0);        // ← 读
    return Map.of("count", count + 1);           // ← 写：返回 Map 就是写
});

返回的 Map 里只需要放这次要改的 key，不用返回整个黑板。

三个必须记住的规则：

写法	含义
return Map.of()	不修改黑板（常用于「条件不满足，直接跳过」）
return Map.of("k", v)	把 k 更新为 v（按该 key 的策略更新）
Map 里的 value 为 null	会触发 Map.of() 的 NPE —— 别塞 null，用空串 / 空集合代替

注意：Map.of 本身不允许 null 值，而节点里从外部拿到 null 是很常见的（比如查库返回空）。

为什么设计成「返回 Map」而不是「调用 set」？

因为节点可能并行执行。两个节点同时写黑板时，「谁覆盖谁」需要一个明确的规则——这就是 KeyStrategy 存在的理由。如果改成 set，就得在框架内部加锁，还要额外定义合并语义，反而更乱。

初始化黑板

黑板不是凭空来的，初始值由调用方传入：

OverAllState state = graph.call(Map.of("text", "hi")).get();

graph.call(Map) 是免 RunnableConfig 的便捷重载；需要会话隔离时用 graph.call(Map, config)。

进去的只有 text，但跑完后黑板上有三个 key——size 和 result 都是节点写出来的。节点之间零耦合，全靠黑板串联。

2.2 addNode：注册一个干活的节点

.addNode("measure", measure)

两个参数：

• 节点名（"measure"）：后面 addEdge / addConditionalEdges 引用它时用的字符串 ID，全局唯一。
• 节点动作：一个 AsyncNodeAction 实例。

node_async 是什么

框架要的是异步接口 AsyncNodeAction，它继承自 Function<OverAllState, CompletableFuture<Map<String,Object>>>。

而写业务逻辑时通常是同步的——NodeAction 就是那个同步接口，只有一个方法：

public interface NodeAction {
    Map<String, Object> apply(OverAllState state) throws Exception;
}

AsyncNodeAction.node_async(nodeAction) 是个适配器，把同步实现包成异步。NodeAction 是函数式接口（@FunctionalInterface，只有一个抽象方法 apply）,支持lambda：

AsyncNodeAction node = AsyncNodeAction.node_async(state -> Map.of("k", "v"));

两种写法：逻辑简单就 lambda，逻辑复杂就 implements NodeAction 写个类。

2.3 addEdge：固定跳转

.addEdge(StateGraph.START, "measure")     // 入口
.addEdge("keep", StateGraph.END)          // 出口

两个特殊节点：

• StateGraph.START —— 图的入口，从它出发的边定义「谁先跑」。
• StateGraph.END —— 图的终点，指向它的边表示「跑完就收工」。

并行：多条出边就是并行

这是 addEdge 最有价值的能力：

// START 有两条出边 → 下面两个节点并行执行
graph.addEdge(StateGraph.START, "fetchA");
graph.addEdge(StateGraph.START, "fetchB");

// 两条边都汇入 merge → merge 会等两个都跑完才执行
graph.addEdge("fetchA", "merge");
graph.addEdge("fetchB", "merge");

START ─┬─► fetchA ──┐
       │            ├──► merge
       └─► fetchB ──┘

两条规则：

• 一个节点有多条出边 = 这些目标并行跑。
• 多条边指向同一个目标 = 该目标等所有上游完成才跑（天然的 join / 栅栏）。

2.4 addConditionalEdges：按黑板内容分支

固定跳转不够用时用它。三个参数必须一起理解：

graph.addConditionalEdges(
        "measure",                                  // ① 挂在哪个节点后面
        AsyncEdgeAction.edge_async(state ->         // ② 路由函数
                state.value("size", "SHORT")),
        Map.of("LONG",  "cut",                      // ③ 映射表
               "SHORT", "keep"));

参数	作用
① 源节点	这个节点的动作执行完之后才调用路由函数
② 路由函数	入参 OverAllState，返回字符串标签
③ 映射表	把标签翻译成真正的节点名

路由函数只读、不写

EdgeAction 也是函数式接口（@FunctionalInterface），支持 lambda：

public interface EdgeAction {
    String apply(OverAllState state) throws Exception;
}

它只读黑板，不写黑板。 这个分工很重要：

NodeAction 读黑板 + 干活 + 写黑板；EdgeAction 只读黑板、不写黑板，返回一个路由值。

映射表：两套独立的命名空间

映射表的语义是 路由函数返回值 → 目标节点名——key 和 value 是两套命名空间。

写成 Map.of("LONG", "cut", ...)，因为标签恰好和节点名长得像。开发中完全可以（也应该）用更有业务含义的标签：

graph.addConditionalEdges("checkNode",
        AsyncEdgeAction.edge_async(state -> {
            Boolean ok = (Boolean) state.value("approved").orElse(false);
            return ok ? "ADOPT" : "REJECT";          // ← 标签
        }),
        Map.of("ADOPT",  "publishNode",              // ← 标签 → 节点名
               "REJECT", StateGraph.END));

这层间接性最大的价值是支持循环。 比如「起草 → 自检 → 不合格就重写」：

Map.of("retry", "draftNode",     // 回炉，形成环
       "done",  StateGraph.END)

路由函数只管说「重写」还是「通过」，具体跳到哪个节点，映射表说了算。 。

2.5 中断与恢复：把「人」接进控制流

条件边管分支，但「停下来等人」需要另一个机制：编译期的中断点 + 存档。

CompiledGraph graph = stateGraph.compile(CompileConfig.builder()
        .interruptBefore("approvalNode")             // ← 执行到该节点【之前】暂停
        .saverConfig(SaverConfig.builder()
                .register(SaverEnum.MEMORY.getValue(), new MemorySaver())   // ← 存档
                .build())
        .build());

• interruptBefore("approvalNode")：图跑到 approvalNode 前就停下来，把当前黑板快照存档，graph.call(...) 直接返回。
• saverConfig：没有持久化就没有恢复。中断了但没存档，那张图就再也叫不醒了。

框架内置四种存档实现：

实现	说明
MemorySaver	存内存，零依赖，适合本地开发和学习
FileSystemSaver	存文件，单机重启不丢
RedisSaver	存 Redis，生产多实例部署用它（需要 Redisson）
MongoSaver	存 MongoDB

生产换 Redis 只改一行：.register(SaverEnum.REDIS.getValue(), new RedisSaver(redissonClient))——redissonClient 就是注入的 RedissonClient Bean。

恢复：四步，一步都不能少

RunnableConfig config = RunnableConfig.builder()
        .threadId(threadId)                  // ← 依赖线程ID(UUID生成即可)找回暂停的图
        .build();

StateSnapshot snapshot = graph.getState(config);        // ① 取回存档
OverAllState state = snapshot.state();

state.withResume();                                     // ② 标记「这次是恢复，不是重新开始」
state.withHumanFeedback(new OverAllState.HumanFeedback( // ③ 塞入人的决策
        Map.of("approved", true), ""));
graph.call(state, config);                              // ④ 继续跑

条件边 + 中断是绝配：中断让流程停下来等人，条件边按人的决策选路。而且两者是两层解耦——节点只负责把决策翻译成黑板上的标签，路由函数只负责按标签选路。

---

三分钟速查表

// 黑板
new StateGraph("名字", keyStrategyFactory)          // 声明 key + 合并策略
state.value("key", defaultValue)                    // 读（带默认值）
return Map.of("key", value)                         // 写（返回 Map）
return Map.of()                                     // 不写

// 节点
graph.addNode("nodeName", AsyncNodeAction.node_async(nodeAction))
// nodeAction: NodeAction 或 lambda，返回 Map<String,Object>

// 边
graph.addEdge(StateGraph.START, "firstNode")        // 入口
graph.addEdge("a", "b")                             // 普通边；a 多条出边 = 并行
graph.addEdge("lastNode", StateGraph.END)           // 终点

graph.addConditionalEdges("source", AsyncEdgeAction.edge_async(edgeAction), mapping)
// edgeAction: EdgeAction 或 lambda，返回「标签」字符串
// mapping: Map.of("标签", "真实节点名", ..., StateGraph.END, StateGraph.END)

// 编译与运行
graph.compile(CompileConfig.builder()
        .interruptBefore("approvalNode")            // 人工中断点
        .saverConfig(SaverConfig.builder()
                .register(SaverEnum.MEMORY.getValue(), new MemorySaver())  
                .build())
        .build())

graph.call(Map.of("text", "hi")).get()                                  // 免 config
graph.call(Map.of("ticketId", "T-1001"), RunnableConfig.builder().threadId(id).build())

Map<String, Object> blackboard = graph.call(Map.of("text", "hi")).get().data();  // 拿整块黑板

// 人工恢复四步
StateSnapshot snap = graph.getState(cfg);
snap.state().withResume();
snap.state().withHumanFeedback(new OverAllState.HumanFeedback(Map.of("approved", true), ""));
graph.call(snap.state(), cfg);

重点：

黑板（OverAllState）负责传数据，节点（addNode）负责干活，边（addEdge / addConditionalEdges）负责决定下一步——三者解耦，就能把「LLM + 业务 + 人」编排进同一条流程。