Spring AI Alibaba Graph Workflow 深度解析

2.2 Node（节点）

Node 是什么？

• 执行具体业务逻辑的单元
• 接收 State，处理后返回新的 State
• 可以是同步或异步

// Node 的本质
// State → Node → State
// 实现一个 Node
public class WeatherNode implements NodeAction {
    @Override
    public Map<String, Object> apply(OverAllState state) throws Exception {
        // 1. 读取 State
        String city = (String) state.value("city").orElse("北京");
        // 2. 执行业务逻辑
        String weather = getWeather(city);
        // 3. 返回新的 State
        return Map.of("weather", weather);
    }
}

2.3 Edge（边）

Edge 是什么？

• 连接节点的路径
• 定义执行顺序
• 支持条件路由

// 普通边：固定路由
workflow.addEdge("nodeA", "nodeB");

// 条件边：动态路由
workflow.addConditionalEdges("classifier", edge_async(state -> {
    String intent = (String) state.value("intent").orElse("default");
    return intent; // 返回下一个节点名称
}), Map.of(
    "weather", "weather_node",
    "news", "news_node",
    "default", "default_node"
));

三、构建你的第一个 Graph Workflow

3.1 完整流程

// 1. 定义 KeyStrategyFactory（状态更新策略）
KeyStrategyFactory keyStrategyFactory = () -> {
    Map<String, KeyStrategy> strategies = new HashMap<>();
    strategies.put("messages", new AppendStrategy()); // 追加
    strategies.put("result", new ReplaceStrategy()); // 替换
    return strategies;
};

// 2. 创建 StateGraph
StateGraph workflow = new StateGraph(keyStrategyFactory);

// 3. 添加节点
workflow.addNode("nodeA", node_async(new NodeA()));
workflow.addNode("nodeB", node_async(new NodeB()));

// 4. 添加边
workflow.addEdge(START, "nodeA");
workflow.addEdge("nodeA", "nodeB");
workflow.addEdge("nodeB", END);

// 5. 编译
CompiledGraph compiled = workflow.compile();

// 6. 执行
Map<String, Object> initialState = Map.of("input", "Hello");
OverAllState result = compiled.invoke(initialState, config).orElseThrow();

3.2 小 Demo 案例：智能客服路由

public class CustomerServiceGraph {
    // 节点 1: 意图分类
    public static class ClassifyNode implements NodeAction {
        private final ChatClient chatClient;
        
        @Override
        public Map<String, Object> apply(OverAllState state) {
            String query = (String) state.value("user_query").orElse("");
            String prompt = String.format("""
                分类用户意图（只返回类别）：
                - technical: 技术问题
                - billing: 账单问题
                - general: 一般咨询
                用户问题：%s
                """, query);
            String intent = chatClient.prompt()
                .user(prompt)
                .call()
                .content()
                .trim()
                .toLowerCase();
            return Map.of("intent", intent);
        }
    }

    // 节点 2: 技术支持
    public static class TechnicalSupportNode implements NodeAction {
        @Override
        public Map<String, Object> apply(OverAllState state) {
            String query = (String) state.value("user_query").orElse("");
            // 搜索技术文档
            String answer = searchTechDocs(query);
            return Map.of("answer", answer);
        }
    }

    // 节点 3: 账单处理
    public static class BillingNode implements NodeAction {
        @Override
        public Map<String, Object> apply(OverAllState state) {
            String query = (String) state.value("user_query").orElse("");
            // 查询账单系统
            String answer = queryBillingSystem(query);
            return Map.of("answer", answer);
        }
    }

    // 构建 Graph
    public static CompiledGraph createGraph(ChatModel chatModel) {
        ChatClient.Builder builder = ChatClient.builder(chatModel);
        KeyStrategyFactory factory = () -> {
            Map<String, KeyStrategy> strategies = new HashMap<>();
            strategies.put("user_query", new ReplaceStrategy());
            strategies.put("intent", new ReplaceStrategy());
            strategies.put("answer", new ReplaceStrategy());
            return strategies;
        };
        StateGraph workflow = new StateGraph(factory)
            .addNode("classify", node_async(new ClassifyNode(builder.build())))
            .addNode("technical", node_async(new TechnicalSupportNode()))
            .addNode("billing", node_async(new BillingNode()))
            .addNode("general", node_async(new GeneralNode()));

        // 添加边
        workflow.addEdge(START, "classify");
        // 条件路由
        workflow.addConditionalEdges("classify", edge_async(state -> (String) state.value("intent").orElse("general")), Map.of(
            "technical", "technical",
            "billing", "billing",
            "general", "general"
        ));
        workflow.addEdge("technical", END);
        workflow.addEdge("billing", END);
        workflow.addEdge("general", END);
        return workflow.compile();
    }

    // 使用
    public static void main(String[] args) {
        CompiledGraph graph = createGraph(chatModel);
        Map<String, Object> initialState = Map.of(
            "user_query", "我的软件无法启动"
        );
        var config = RunnableConfig.builder()
            .threadId("customer_001")
            .build();
        // 流式执行
        graph.stream(initialState, config)
            .doOnNext(output -> {
                System.out.println("节点：" + output.node());
                System.out.println("状态：" + output.state().data());
            })
            .blockLast();
    }
}

四、状态管理：KeyStrategy

4.1 什么是 KeyStrategy？

KeyStrategy（键策略） 是 Graph Workflow 中用于控制状态更新行为的核心机制。

// KeyStrategy 的本质是一个函数
public interface KeyStrategy {
    Object apply(Object oldValue, Object newValue);
}

作用：当节点返回新的状态数据时，KeyStrategy 决定如何将新值与旧值合并。

4.2 什么是 KeyStrategyFactory？

KeyStrategyFactory 是一个工厂接口，用于为 State 中的每个键指定更新策略。

public interface KeyStrategyFactory {
    Map<String, KeyStrategy> getStrategies();
}

// 使用示例
KeyStrategyFactory factory = () -> {
    Map<String, KeyStrategy> strategies = new HashMap<>();
    strategies.put("messages", new AppendStrategy()); // messages 使用追加策略
    strategies.put("result", new ReplaceStrategy()); // result 使用替换策略
    strategies.put("counter", new CustomStrategy()); // counter 使用自定义策略
    return strategies;
};

4.3 为什么需要 KeyStrategy？

问题场景：当多个节点修改同一个 State 键时，如何合并？

KeyStrategy 就是用来解决这个问题的！

4.4 工作原理

4.5 内置策略详解

1️⃣ ReplaceStrategy（替换策略）

用途：直接用新值替换旧值（最常用）

strategies.put("result", new ReplaceStrategy());
// 行为示例：
// oldValue: "old result"
// newValue: "new result"
// 返回："new result" ← 直接替换
// 适用场景：
// - 单一结果值（result、answer、status）
// - 配置项（config、settings）
// - 标识符（id、type、intent）

2️⃣ AppendStrategy（追加策略）

用途：将新值追加到列表中

strategies.put("messages", new AppendStrategy());
// 行为示例：
// oldValue: ["msg1", "msg2"]
// newValue: "msg3"
// 返回：["msg1", "msg2", "msg3"] ← 追加到列表
// 如果 oldValue 不是列表：
// oldValue: "msg1"
// newValue: "msg2"
// 返回：["msg1", "msg2"] ← 自动转换为列表
// 适用场景：
// - 消息历史（messages、history）
// - 日志记录（logs、events）
// - 收集结果（results、findings）

3️⃣ 自定义策略

用途：实现自己的合并逻辑

// 示例 1: 累加策略（数值累加）
strategies.put("counter", (oldValue, newValue) -> {
    int old = oldValue != null ? (int) oldValue : 0;
    int add = (int) newValue;
    return old + add;
});
// 使用：
// oldValue: 5
// newValue: 3
// 返回：8

// 示例 2: Map 合并策略
strategies.put("metadata", (oldValue, newValue) -> {
    Map<String, Object> old = (Map<String, Object>) oldValue;
    Map<String, Object> newMap = (Map<String, Object>) newValue;
    Map<String, Object> merged = new HashMap<>(old);
    merged.putAll(newMap); // 新值覆盖旧值
    return merged;
});
// 使用：
// oldValue: {"key1": "value1", "key2": "value2"}
// newValue: {"key2": "new_value2", "key3": "value3"}
// 返回：{"key1": "value1", "key2": "new_value2", "key3": "value3"}

// 示例 3: 去重追加策略
strategies.put("tags", (oldValue, newValue) -> {
    Set<String> tags = new HashSet<>((List<String>) oldValue);
    tags.add((String) newValue);
    return new ArrayList<>(tags);
});

4.6 完整的实现 KeyStrategyFactory 示例

public class MyKeyStrategyFactory implements KeyStrategyFactory {
    @Override
    public Map<String, KeyStrategy> getStrategies() {
        Map<String, KeyStrategy> strategies = new HashMap<>();
        // 1. 单一值使用替换策略
        strategies.put("user_query", new ReplaceStrategy());
        strategies.put("intent", new ReplaceStrategy());
        strategies.put("result", new ReplaceStrategy());
        strategies.put("status", new ReplaceStrategy());
        // 2. 列表使用追加策略
        strategies.put("messages", new AppendStrategy());
        strategies.put("history", new AppendStrategy());
        strategies.put("logs", new AppendStrategy());
        // 3. 数值使用累加策略
        strategies.put("token_count", (oldValue, newValue) -> {
            int old = oldValue != null ? (int) oldValue : 0;
            return old + (int) newValue;
        });
        // 4. Map 使用合并策略
        strategies.put("metadata", (oldValue, newValue) -> {
            if (oldValue == null) return newValue;
            Map<String, Object> merged = new HashMap<>((Map) oldValue);
            merged.putAll((Map) newValue);
            return merged;
        });
        return strategies;
    }
}

// 使用
StateGraph workflow = new StateGraph(new MyKeyStrategyFactory());

4.7 KeyStrategy 最佳实践

✅ 好的实践

KeyStrategyFactory factory = () -> {
    Map<String, KeyStrategy> strategies = new HashMap<>();
    // 1. 明确每个键的用途
    strategies.put("user_input", new ReplaceStrategy()); // 用户输入
    strategies.put("conversation", new AppendStrategy()); // 对话历史
    strategies.put("final_answer", new ReplaceStrategy()); // 最终答案
    // 2. 为所有可能的键都定义策略
    strategies.put("error", new ReplaceStrategy());
    strategies.put("metadata", new ReplaceStrategy());
    return strategies;
};

❌ 不好的实践

KeyStrategyFactory factory = () -> {
    Map<String, KeyStrategy> strategies = new HashMap<>();
    // ❌ 只定义部分键（其他键会使用默认策略，可能导致意外行为）
    strategies.put("messages", new AppendStrategy());
    // ❌ 策略选择不当
    strategies.put("result", new AppendStrategy()); // result 应该用 ReplaceStrategy
    return strategies;
};

4.8 默认行为

如果某个键没有定义 KeyStrategy，会怎样？

// 默认行为：使用 ReplaceStrategy
// 即：新值直接替换旧值

建议：为所有可能出现的键都显式定义策略，避免依赖默认行为。

五、边的类型详解

5.1 普通边（固定路由）

// 单个边
workflow.addEdge("nodeA", "nodeB");
// 从 START 到节点
workflow.addEdge(START, "nodeA");
// 从节点到 END
workflow.addEdge("nodeZ", END);

5.2 条件边（相当于动态路由）

workflow.addConditionalEdges(
    "classifier", // 源节点
    edge_async(state -> {
        // 条件函数
        String type = (String) state.value("type").orElse("default");
        // 根据 State 决定下一个节点
        if ("urgent".equals(type)) {
            return "urgent_handler";
        } else if ("normal".equals(type)) {
            return "normal_handler";
        } else {
            return "default_handler";
        }
    }),
    Map.of(
        // 路由映射
        "urgent_handler", "urgent_handler",
        "normal_handler", "normal_handler",
        "default_handler", "default_handler"
    )
);

5.3 并行边

// A 完成后，B1/B2/B3 并行执行
workflow.addEdge("A", List.of("B1", "B2", "B3"));
// B1/B2/B3 都完成后，执行 C
workflow.addEdge(List.of("B1", "B2", "B3"), "C");

六、可视化导出

虽然 Graph Workflow 是代码定义的，但可以导出为可视化图表！

// 导出为 Mermaid 图
GraphRepresentation mermaid = workflow.getGraph(
    GraphRepresentation.Type.MERMAID,
    "Customer Service Workflow"
);
System.out.println(mermaid.getContent());
// 输出：
// graph TD
// __START__ --> classify
// classify --> technical
// classify --> billing
// classify --> general
// technical --> __END__
// billing --> __END__
// general --> __END__

将输出的 Mermaid 代码粘贴到支持 Mermaid 的平台（如 GitHub、Notion），即可看到可视化图表！

七、Graph vs ReactAgent

7.1 ReactAgent 的本质

ReactAgent 本质上就是一个预定义的 StateGraph！

// ReactAgent 内部实际上是这样的 Graph：
StateGraph reactGraph = new StateGraph()
    .addNode("llm_node", llmNode) // 调用 LLM
    .addNode("tool_node", toolNode) // 执行工具
    .addEdge(START, "llm_node")
    .addConditionalEdges("llm_node", edge_async(state -> {
        // 如果需要调用工具，路由到 tool_node
        // 否则结束
        return needsTools(state) ? "tool_node" : END;
    }), Map.of("tool_node", "tool_node", END, END))
    .addEdge("tool_node", "llm_node"); // 工具执行后回到 LLM

7.2 何时使用 Graph？何时使用 ReactAgent？

八、总结

8.1 核心要点回顾

1. Graph Workflow 是底层机制
   • ReactAgent 基于 Graph 实现
   • 提供完全的控制能力
2. 三大核心组件
   • State：数据流转
   • Node：业务逻辑
   • Edge：执行顺序
3. 与可视化平台的对比
   • 本质相同（都是 DAG）
   • 形式不同（代码 vs UI）
   • 各有优势
4. 灵活的编排能力
   • 条件路由
   • 并行执行
   • 状态管理

8.2 Graph Workflow vs ReactAgent

参考资源

• 官方文档：https://java2ai.com
• 官方 GitHub：https://github.com/alibaba/spring-ai-alibaba