AI项目博客
记录 AI 工程实践与大模型应用
        找回密码
当前位置：AI项目博客 AI框架 AI Alibaba 正文

Spring AI Alibaba 68｜Graph Core · 核心库概念指南 · 序列化器（Serializer）

2026-09-02 分类：AI Alibaba 阅读(15) 评论(0)

在图执行期间，状态需要被序列化（主要用于克隆目的），同时也为跨不同执行持久化状态提供能力。Spring AI Alibaba 目前提供了 Jackson、JDK 两种序列化策略实现。默认使用 Jackson 实现。

在序列化过程中，重点关注如下内容：

  1. 不依赖不安全的标准序列化框架
  2. 允许为第三方（非可序列化）类实现序列化
  3. 尽可能避免类加载问题
  4. 在序列化过程中管理可空值

自定义序列化器

默认情况下，Graph 使用 Jackson 作为序列化器，并且对几个主流模型厂商的 Message 类型做了兼容。但是对于用户自定义的消息类型，可能无法做到完全兼容，这里有三种常见做法：

1. 为自定义数据类增加 Jackson 注解

为自定义数据类添加 Jackson 注解，提升序列化兼容性。这是最简单直接的方式：

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;

// 使用 @JsonIgnoreProperties 忽略未知属性
@JsonIgnoreProperties(ignoreUnknown = true)
public class CustomMessage {
  private String content;
  private String type;

  // 使用 @JsonCreator 和 @JsonProperty 指定构造函数参数映射
  @JsonCreator
  public CustomMessage(
          @JsonProperty("content") String content,
          @JsonProperty("type") String type) {
      this.content = content;
      this.type = type;
  }

  // Getter 和 Setter 方法
  public String getContent() {
      return content;
  }

  public void setContent(String content) {
      this.content = content;
  }

  public String getType() {
      return type;
  }

  public void setType(String type) {
      this.type = type;
  }
}

2. 定制 StateGraph 中的默认 Serializer

通过 StateGraph.getStateSerializer() 获取序列化器，转换为 JacksonSerializer 类型，然后获取 ObjectMapper 进行定制：

import com.alibaba.cloud.ai.graph.StateGraph;
import com.alibaba.cloud.ai.graph.serializer.StateSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.SerializationFeature;

// 创建 StateGraph（使用默认的 JacksonSerializer）
StateGraph graph = new StateGraph(keyStrategyFactory);

// 获取序列化器并转换为 JacksonSerializer
StateSerializer stateSerializer = graph.getStateSerializer();
if (stateSerializer instanceof StateGraph.JacksonSerializer) {
  StateGraph.JacksonSerializer jacksonSerializer = 
          (StateGraph.JacksonSerializer) stateSerializer;

  // 获取 ObjectMapper
  ObjectMapper objectMapper = jacksonSerializer.getObjectMapper();

  // 定制 ObjectMapper，例如：
  // 1. 配置序列化选项
  objectMapper.configure(SerializationFeature.INDENT_OUTPUT, true);
  objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

  // 2. 注册自定义模块
  var module = new SimpleModule();
  // module.addSerializer(CustomMessage.class, new CustomMessageSerializer());
  // module.addDeserializer(CustomMessage.class, new CustomMessageDeserializer());
  objectMapper.registerModule(module);
}

说明：

  • StateGraph 默认使用 JacksonSerializer（继承自 SpringAIJacksonStateSerializer）
  • 通过 getStateSerializer() 获取序列化器后，可以转换为 JacksonSerializer 类型
  • 使用 getObjectMapper() 方法获取 ObjectMapper 进行定制
  • 这种方式适合在创建图之后需要动态定制序列化器的场景

3. 为 Graph 提供自定义序列化器（推荐）

可以为 StateGraph 指定任意序列化器，如框架已经提供的 JDK 原生类型 ObjectStateSerializer 或自定义的 Gson 序列化器实现等。

import com.alibaba.cloud.ai.graph.OverAllState;
import com.alibaba.cloud.ai.graph.serializer.plain_text.jackson.SpringAIJacksonStateSerializer;
import com.alibaba.cloud.ai.graph.state.AgentStateFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.JsonDeserializer;
import com.fasterxml.jackson.databind.JsonSerializer;
import com.fasterxml.jackson.databind.SerializerProvider;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.alibaba.cloud.ai.graph.StateGraph;

import java.io.IOException;

// 自定义消息类型
public class CustomMessage {
  private String content;
  private String type;

  // 构造函数、Getter、Setter...
  public CustomMessage(String content, String type) {
      this.content = content;
      this.type = type;
  }

  public String getContent() { return content; }
  public String getType() { return type; }
}

// 自定义序列化器
public class CustomMessageSerializer extends JsonSerializer<CustomMessage> {
  @Override
  public void serialize(CustomMessage value, JsonGenerator gen, SerializerProvider serializers) 
          throws IOException {
      gen.writeStartObject();
      gen.writeStringField("content", value.getContent());
      gen.writeStringField("type", value.getType());
      gen.writeEndObject();
  }
}

// 自定义反序列化器
public class CustomMessageDeserializer extends JsonDeserializer<CustomMessage> {
  @Override
  public CustomMessage deserialize(JsonParser p, DeserializationContext ctxt) 
          throws IOException {
      String content = null;
      String type = null;

      p.nextToken(); // 跳过 START_OBJECT
      while (p.nextToken() != com.fasterxml.jackson.core.JsonToken.END_OBJECT) {
          String fieldName = p.getCurrentName();
          p.nextToken();

          if ("content".equals(fieldName)) {
              content = p.getText();
          } else if ("type".equals(fieldName)) {
              type = p.getText();
          }
      }

      return new CustomMessage(content, type);
  }
}

// 创建自定义序列化器类
class CustomizedSerializer extends SpringAIJacksonStateSerializer {

  public CustomizedSerializer() {
      super(OverAllState::new);

      // 创建 SimpleModule 并注册自定义序列化器/反序列化器
      var module = new SimpleModule();
      module.addSerializer(CustomMessage.class, new CustomMessageSerializer());
      module.addDeserializer(CustomMessage.class, new CustomMessageDeserializer());

      // 注册模块到 ObjectMapper
      objectMapper.registerModule(module);

      // 可以继续定制 ObjectMapper，例如：
      // objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
      // objectMapper.configure(SerializationFeature.INDENT_OUTPUT, true);
  }
}

// 使用自定义序列化器创建 StateGraph
KeyStrategyFactory keyStrategyFactory = () -> new HashMap<>();
StateGraph graph = new StateGraph("demo", keyStrategyFactory, new CustomizedSerializer());

说明：

  • SpringAIJacksonStateSerializer 在构造函数中已经注册了 Spring AI 相关的消息类型序列化器
  • 继承后可以在构造函数中继续注册自定义类型的序列化器/反序列化器
  • 可以通过 objectMapper 字段（继承自父类）访问和定制 ObjectMapper
  • 这种方式可以完全控制序列化行为，适合复杂的自定义类型

在 Spring AI Alibaba 中，节点通常是

  • 一个函数式接口 ([AsyncNodeAction])，其入参是 OverAllState，您可以使用 [addNode] 方法将这些节点添加到图中。
  • 一个函数式接口 ([AsyncNodeActionWithConfig])，其入参是 OverAllState 和 RunnableConfig，您可以使用 [addNode] 方法将这些节点添加到图中。

由于 [AsyncNodeAction] 设计用于与 [CompletableFuture] 一起工作，您可以使用 node_async 静态方法将其适配为更简单的同步场景。

START 节点

// add a normal edge
START 节点是一个特殊节点，表示将用户输入发送到图的节点。引用此节点的主要目的是确定首先应该调用哪些节点。

import static com.alibaba.cloud.ai.graph.StateGraph.START;

END 节点

END 节点是一个特殊节点，表示终端节点。当您想要表示哪些边在完成后没有任何操作时，会引用此节点。

边定义了逻辑如何路由以及图如何决定停止。这是智能体工作方式和不同节点之间如何通信的重要部分。有几种关键类型的边：

  • 普通边（Normal Edges）：

    直接从一个节点到下一个节点。

  • 条件边（Conditional Edges）：

    调用函数来确定接下来要去哪个节点。

  • 入口点（Entry Point）：

    当用户输入到达时首先调用哪个节点。

  • 条件入口点（Conditional Entry Point）：

    调用函数来确定当用户输入到达时首先调用哪个节点。

普通边

如果您总是想从节点 A 到节点 B，可以直接使用 [addEdge] 方法。

import static com.alibaba.cloud.ai.graph.StateGraph.START;
import static com.alibaba.cloud.ai.graph.StateGraph.END;
import static com.alibaba.cloud.ai.graph.action.AsyncEdgeAction.edge_async;

// 添加条件边
graph.addConditionalEdges("nodeA", edge_async(state -> "nodeB"),
      Map.of("nodeB", "nodeB", "nodeC", "nodeC"));

您必须提供一个对象，将 routingFunction 的输出映射到下一个节点的名称。

Conditional Edges

若需可选地路由至一个或多个 Node 节点，可使用 addConditionalEdges 方法。该方法接收节点名称及一个函数式接口 AsyncEdgeAction 或 AsyncEdgeActionWithConfig，该接口将作为“路由函数”在节点执行后调用，用来决策下一个节点应该走向哪里。

为同一个节点设置出边

一个节点可以拥有多个出边，这样就形成了节点间的并行关系。如果一个节点有多个出边，那么所有这些目标节点将作为下一个并行执行。关于并行节点，可参考示例目录中的详细文档描述。

本文整理自 Spring AI Alibaba 官方文档《核心库概念指南》：https://java2ai.com/docs/frameworks/graph-core/core/core-library

赞(0)
未经允许不得转载：AI项目博客 » Spring AI Alibaba 68｜Graph Core · 核心库概念指南 · 序列化器（Serializer）
分享到

admin

相关推荐

评论 抢沙发

取消

AI项目博客

专注 AI 工程化落地 —— 覆盖智能体、大模型、Java AI 与 Python 的学习笔记，以及 AI 平台的项目实战记录。

关于本站文章归档

登录

找回密码

注册