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 + 业务 + 人」编排进同一条流程。
