Spring AI对话持久化与Token统计：基于ChatMemory与Advisor的落地方案

1. 项目背景与核心问题拆解

做LLM应用落地时，只要业务稍微有点规模，对话记忆和token统计就是两个绕不开的坎。尤其是Spring AI生态刚起步，很多团队在用的还是"用内存Map硬扛、token靠前端瞎猜、消息记录散落在业务代码里"这种原始状态。时间一长，数据乱成一团，想查一条历史会话要翻半天，想对账token成本更是无从下手。

我这次的项目标题看着简单，但背后要解决的问题其实很具体：

• 对话持久化不能依赖Spring AI默认的InMemoryChatMemory，业务数据要有自己的表结构和查询能力。
• token消耗不能只算一次请求的输入输出，要能对每次会话、每轮消息、甚至整个会话周期做聚合统计。
• 最关键的是，这些能力不能侵入现有业务代码。业务方不该关心"你这条消息是怎么存的、token怎么计的"，他们只需要调用对话接口，剩下的事由框架层自动完成。

我做的这套方案，核心思路是通过自定义ChatMemory实现、ChatMemoryAdvisor自动装配、以及一个独立的消息与元数据存储模块，把持久化和token统计从业务代码中彻底剥离出去。改造完成后，业务层看到的还是原来那个ChatClient调用，但底层已经自动完成了存储、统计、聚合。

这篇文章就把完整的设计思路、落地步骤、踩坑经验拆开讲清楚。

2. 方案选型：为什么不能直接硬编码业务逻辑

2.1 先搞清楚Spring AI的对话记忆机制

Spring AI的对话记忆核心是一个叫ChatMemory的接口，里面定义了add、get、clear这几个基本操作：

public interface ChatMemory { void add(String conversationId, List<Message> messages); List<Message> get(String conversationId); void clear(String conversationId); }

默认实现是InMemoryChatMemory，用ConcurrentHashMap存，应用重启就没了，也不支持按条件检索。真正要落地，必须自己实现一个持久化版本的ChatMemory。

但这里有个关键点：ChatMemory接口只管消息存取，它不负责"什么时候调用add、什么时候调用get"。这个调度逻辑在Advisor里。Spring AI提供了ChatMemoryAdvisor，它会在每次对话前自动把历史消息加载进Prompt，对话结束后把新消息追加进存储。

要自定义数据库结构，本质上要做两件事：

1. 自定义ChatMemory实现类，替换默认的内存实现。
2. 注册一个ChatMemoryAdvisor Bean，让Spring AI在每一轮对话时自动调用我们的持久化实现。

2.2 为什么自定义表结构而不是用现成的ConversationStore

我调研过现成的ConversationStore实现，它本质上是个偏向NoSQL的存储抽象，适合快速demo但不适合复杂业务。原因有几个：

• 现成方案通常是JSON序列化整段消息，查询历史消息列表时没法按角色、按时间、按token用量做条件过滤。
• 没法高效统计"这个用户这周消耗了多少token"，因为消息内容是包在一个大JSON里的，要统计就得全量反序列化。
• 业务方经常要展示对话记录，但不需要大段的原始Prompt。自定义表结构可以只存消息摘要和关键元数据。

所以最后我选择自建三张表：会话表、消息表、token统计表。会话表管会话元数据，消息表管具体每轮对话内容，token统计表管消耗明细和聚合维度。

2.3 "不侵入业务形式"的具体实现策略

这是整个方案里最容易被忽略、但也最重要的设计点。很多团队做类似功能，喜欢在Service层手动调用存储接口，像这样：

// 错误示范 String answer = chatClient.call(userMessage); messageStore.save(conversationId, userMessage, answer); tokenStatService.record(conversationId, tokenCount);

这样确实能跑，但业务代码被彻底污染了。每接一个新对话场景，都要重复写一遍存储和统计逻辑，一旦统计口径变了，所有调用方都得跟着改。

我的做法是把这些逻辑收敛到两个核心组件里：

• 自定义ChatMemory实现：负责把对话消息自动写入数据库。
• 自定义Advisor：负责在对话完成后读取token消耗并落库。

业务方拿到的是一个干干净净的ChatClient：

String answer = chatClient.call(conversationId, userMessage);

就这么一行。没有存储代码，没有token统计代码，没有额外的状态管理。所有横切关注点都被Spring/AI的Advisor机制拦截并自动处理了。这就叫不侵入业务形式。

3. 自定义数据库结构设计详解

3.1 消息表设计：一条消息一行记录

消息表是整个方案的基石。我设计的时候重点考虑了查询场景，尽量做到一条记录能独立还原对话上下文。

表结构如下：

CREATE TABLE ai_chat_message ( id BIGINT AUTO_INCREMENT PRIMARY KEY, conversation_id VARCHAR(64) NOT NULL, message_type VARCHAR(16) NOT NULL COMMENT 'USER或ASSISTANT', content MEDIUMTEXT NOT NULL, prompt_tokens INT DEFAULT 0, completion_tokens INT DEFAULT 0, total_tokens INT DEFAULT 0, create_time DATETIME NOT NULL, INDEX idx_conversation_time (conversation_id, create_time) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

• conversation_id是会话唯一标识，可以是UUID，也可以是业务自己传的订单号或用户ID拼接值。
• message_type区分用户消息和AI回复，方便后续渲染对话界面。
• content字段存原始文本内容，MEDIUMTEXT足够覆盖99%的场景。
• prompt_tokens和completion_tokens分别记录输入和输出token数，这是token统计的原子数据。

这里加了一个联合索引(conversation_id, create_time)，目的是让"查询某个会话的所有消息并且按时间排序"这条高频SQL能走索引，避免全表扫描。实测在千万级消息量下依然可以毫秒级返回。

3.2 会话表设计：一个会话一行主体记录

会话表本身不长，但作用很关键。它是消息表和token统计表的"主键"关联点，也承担了部分业务扩展字段。

CREATE TABLE ai_conversation ( conversation_id VARCHAR(64) PRIMARY KEY, title VARCHAR(200), user_id VARCHAR(64) NOT NULL, app_id VARCHAR(64) COMMENT '业务应用标识，用于区分多场景', model_name VARCHAR(64), first_message_time DATETIME, last_message_time DATETIME, message_count INT DEFAULT 0, CONSTRAINT idx_user_app UNIQUE (user_id, app_id, conversation_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

user_id和app_id的组合索引非常有用。比如你要查"用户A在应用B下的所有会话列表"，这个索引直接命中。title字段可以存会话的第一条消息摘要，做列表页展示时就不用JOIN消息表了。

3.3 token统计表设计：维度优先

token统计表的设计经历了两次重构。最初我打算只在消息表里记录每次对话的token数，然后查询时用SUM聚合。但后来发现业务方经常要按"天""周""月"维度看消耗趋势，全表SUM性能很差，而且还要排除掉一些测试会话、白名单会话。

所以单独拆了一张统计表，专门做预聚合：

CREATE TABLE ai_token_stat ( id BIGINT AUTO_INCREMENT PRIMARY KEY, stat_date DATE NOT NULL, user_id VARCHAR(64), app_id VARCHAR(64), model_name VARCHAR(64), prompt_tokens_total BIGINT DEFAULT 0, completion_tokens_total BIGINT DEFAULT 0, total_tokens_total BIGINT DEFAULT 0, message_count INT DEFAULT 0, update_time DATETIME, UNIQUE KEY uk_stat_dimension (stat_date, user_id, app_id, model_name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

表名的设计意图很明确：按统计周期+业务维度做唯一约束，一条记录代表一个聚合桶。当天某个用户某个应用某款模型的所有token消耗，都累加到对应的那条记录里。查询"本周总消耗"直接SUM几行数据就出来了，不用去碰庞大的消息明细。

3.4 为什么把会话和消息拆成两张表

很多人觉得会话和消息是一对多关系，存一张表加个会话ID字段就够了。但实际业务里不是这样。

会话列表页需要高频查询，它关心的是会话标题、最后活跃时间、消息数量。如果这些字段和消息内容混在同一张表里，查列表时要把几千上万条消息记录都扫一遍，还要DISTINCT去重，性能非常差。拆表之后，会话列表页只查ai_conversation这张几十万行的表，SQL简单直接。

消息明细页则是按conversation_id精准定位，走联合索引，也很快。

这种"宽表变窄表"的设计思路，在AI应用这种"读多写也多"的场景下是必要的。会话的生命周期数据量可能不大，但消息的数据量是指数增长的。

4. 核心实现：自定义ChatMemory与Advisor装配

4.1 自定义ChatMemory实现类

这一步是整个方案的"地基"。Spring AI在调用ChatMemoryAdvisor时会自动注入ChatMemory接口的实现，我们只需要提供一个自定义的Bean即可覆盖默认行为。

先实现消息的增删查：

@Component public class DatabaseChatMemory implements ChatMemory { @Autowired private ChatMessageRepository messageRepository; @Autowired private ConversationRepository conversationRepository; @Override public void add(String conversationId, List<Message> messages) { if (messages == null || messages.isEmpty()) { return; } for (Message message : messages) { if (!isPersisted(message)) { saveMessage(conversationId, message); } } updateConversationInfo(conversationId, messages); } @Override public List<Message> get(String conversationId) { return messageRepository.findByConversationIdOrderByCreateTimeAsc(conversationId) .stream() .map(this::toSpringAiMessage) .collect(Collectors.toList()); } @Override public void clear(String conversationId) { messageRepository.deleteByConversationId(conversationId); conversationRepository.deleteById(conversationId); } }

写这个add方法时踩过几个坑，重点说两件事。

第一，消息去重。Spring AI在对话时可能会把同一批历史消息反复传入add方法，如果每次都直接insert，数据库里会出现大量重复记录，导致上下文越来越长。我的做法是给消息表加一个业务唯一键，用conversation_id + 消息内容hash + create_time来做去重标识，或者重写isPersisted方法检查库中是否已存在相同消息。这一点不做的话，用不了几天库就废了。

第二，get方法的顺序。对话历史的顺序必须严格按时间升序排列，否则LLM拿到的上下文是乱的，模型回答质量直接下降。

4.2 注册ChatMemoryAdvisor并开启自动装配

光有自定义ChatMemory还不够，Spring AI默认不会主动管理对话历史。要让它生效，需要注册ChatMemoryAdvisor：

@Bean public ChatMemoryAdvisor chatMemoryAdvisor(ChatMemory chatMemory) { return ChatMemoryAdvisor.builder(chatMemory) .build(); }

如果你的项目里已经手动构造了ChatClient，比如通过ChatClient.builder(chatModel)，那么需要把Advisor加到builder里：

ChatClient.builder(chatModel) .defaultAdvisors(chatMemoryAdvisor) .build();

装配好之后，业务代码里只需要：

String response = chatClient.prompt() .user("你好，请介绍一下你自己") .advisors(advisorSpec -> advisorSpec .param(ChatMemoryAdvisor.CHAT_MEMORY_CONVERSATION_ID_KEY, "user-123-session-abc")) .call() .content();

整个过程中，业务方唯一需要关心的是传一个conversationId，其余全部由框架接管。这就是我们最初定下的"不侵入"目标。

4.3 token统计的实现机制

token统计不能直接写在业务代码里，也不能靠前端传值。我采用的是在Advisor中拦截Response对象，从中提取usage字段。

Spring AI的ChatResponse里带了usage信息，里面包含promptTokens、completionTokens、totalTokens。在Advisor实现中，先让链路正常走完，拿到最终响应后统一记录。

核心代码如下：

@Component public class TokenUsageAdvisor implements OperationAroundAdvisor { private final TokenStatService tokenStatService; public TokenUsageAdvisor(TokenStatService tokenStatService) { this.tokenStatService = tokenStatService; } @Override public Object around(OperationAroundAdvisorCall call) { String conversationId = call.getAdvisorParams().getChatMemoryConversationId(); String messageContent = call.getAdvisorParams().getUserText(); // 执行真正的对话调用 Object response = call.call(); if (response instanceof ChatResponse chatResponse) { TokenUsage usage = chatResponse.getMetadata().getUsage(); if (usage != null) { tokenStatService.record( conversationId, messageContent, chatResponse.getResult().getOutput().getText(), usage.getPromptTokens(), usage.getCompletionTokens(), usage.getTotalTokens() ); } } else if (response instanceof Flux<?> flux) { // 流式响应场景需要在订阅后统一回收token统计 return flux.doOnComplete(() -> { TokenUsage lastUsage = getLastUsageFromFlux(flux); // 异步记录token }); } return response; } @Override public String getName() { return "token-usage-advisor"; } @Override public int getOrder() { return 1; } }

这个方案有个隐含的好处：token统计不再依赖业务方主动埋点，每一次对话的消耗在框架层就自动沉淀到库里。不管业务方是在哪个Service方法里调用的ChatClient，只要走Advisor链路，统计就跑不掉。

4.4 流式响应场景的处理

流式对话的token统计是个容易踩坑的细节。普通call()方法是一次性拿到完整ChatResponse，而stream()返回的是Flux ，每个chunk里都可能带usage信息，但只有最后一个chunk的usage才是完整的整轮消耗。

我的处理方式是：先构建一个Flux，对每个ChatResponse chunk做遍历取出usage字段暂存，在Flux完成时把最后一次拿到的usage作为整轮对话的消耗记录落库。

private Flux<ChatResponse> handleStreaming(TokenUsageAdvisorChain chain, String conversationId) { Flux<ChatResponse> stream = chain.call(); AtomicReference<TokenUsage> lastUsageRef = new AtomicReference<>(); return stream.doOnNext(response -> { TokenUsage usage = response.getMetadata().getUsage(); if (usage != null) { lastUsageRef.set(usage); } }).doOnComplete(() -> { TokenUsage lastUsage = lastUsageRef.get(); if (lastUsage != null && lastUsage.getTotalTokens() > 0) { tokenStatService.record(conversationId, lastUsage); } }); }

这里有个细节：有些模型供应商在流式返回时，每个chunk都会带usage，有些只在最后带。doOnNext里直接覆盖赋值，最后取到的就是最完整的那份。

4.5 Advisor的order排序问题

多个Advisor同时生效时，执行顺序非常关键。我一开始把TokenUsageAdvisor的order设置成默认值，结果发现对话历史还没有通过ChatMemoryAdvisor注入进去，token统计就先执行了，导致统计到的input token数偏低。

原因在于，ChatMemoryAdvisor负责在对话前把历史消息拼到Prompt里，如果TokenUsageAdvisor先执行了，它看到的userText是不含历史消息的原始输入，统计出来的promptTokens自然不准。

正确顺序是：TokenUsageAdvisor要放在ChatMemoryAdvisor之后执行，确保统计的是完整上下文。

@Bean public TokenUsageAdvisor tokenUsageAdvisor(TokenStatService tokenStatService) { TokenUsageAdvisor advisor = new TokenUsageAdvisor(tokenStatService); advisor.setOrder(2); // 大于ChatMemoryAdvisor的order值 return advisor; }

这里再补充说明一下Advisor的order语义：order值越小越先执行。ChatMemoryAdvisor默认order是0，TokenUsageAdvisor设成1或更大，就能保证"记忆加载在先、token统计在后"。这个顺序问题在文档里根本没写，纯靠踩坑试出来，新手遇到会非常困惑。

5. 完整实操过程与代码落地

5.1 工程结构一览

我建议按模块拆分，不要把所有代码塞到一个包：

src/main/java/com/example/ai/ ├── advisor/ │ ├── TokenUsageAdvisor.java │ └── AdvisorConfig.java ├── chatmemory/ │ ├── DatabaseChatMemory.java │ └── CustomChatMemoryConfig.java ├── entity/ │ ├── ChatMessageEntity.java │ ├── ConversationEntity.java │ └── TokenStatEntity.java ├── repository/ │ ├── ChatMessageRepository.java │ ├── ConversationRepository.java │ └── TokenStatRepository.java ├── service/ │ ├── TokenStatService.java │ └── ChatSessionService.java └── controller/ └── ChatController.java

5.2 消息实体的映射细节

用JPA还是MyBatis都可以，但有几个字段映射要特别注意：

@Entity @Table(name = "ai_chat_message") public class ChatMessageEntity { @Id @GeneratedValue(strategy = GenerationType.IDENTITY) private Long id; @Column(name = "conversation_id", length = 64, nullable = false) private String conversationId; @Column(name = "message_type", length = 16, nullable = false) private String messageType; @Column(name = "content", columnDefinition = "MEDIUMTEXT") private String content; @Column(name = "prompt_tokens") private Integer promptTokens; @Column(name = "completion_tokens") private Integer completionTokens; @Column(name = "total_tokens") private Integer totalTokens; @Column(name = "create_time", nullable = false) private LocalDateTime createTime; }

content字段建议用columnDefinition = "MEDIUMTEXT"强制指定，否则Hibernate默认可能生成varchar(255)，长文本直接截断。

5.3 消息去重的核心实现

这是我花时间最多的一块。为什么要单独拿出来讲？因为直接决定系统能不能长期稳定运行。

Spring AI在流式对话和普通对话中调用ChatMemory.add的时机不一样，同一个session可能被重复调用add多次。我的去重方案是在插入前做一次存在性检查：

public void add(String conversationId, List<Message> messages) { for (Message message : messages) { boolean exists = messageRepository.existsByConversationIdAndContentAndCreateTime( conversationId, message.getText(), message.getTimestamp() ); if (!exists) { saveMessage(conversationId, message); } } }

这么写完发现还有个隐患：如果两个用户在同一毫秒问了完全一样的问题，就会误判为重复。于是我把createTime精确到纳秒，或者在内容字段中拼上一个随机因子。更稳妥的做法是维护一个"已处理消息ID集合"，但因为消息在add时还没生成数据库自增ID，所以退而求其次用内容+时间戳组合判断，目前实测下来误判率极低。

5.4 TokenStatService的原子更新

token累计的写入必须用数据库原子操作，否则并发情况下数据会错。我写的更新逻辑是这样的：

@Transactional public void record(String conversationId, TokenUsage usage) { statRepository.upsert( LocalDate.now(), getCurrentUserId(conversationId), applicationName, modelName, usage.getPromptTokens(), usage.getCompletionTokens(), usage.getTotalTokens() ); }

对应的SQL用了MySQL的ON DUPLICATE KEY UPDATE：

INSERT INTO ai_token_stat ( stat_date, user_id, app_id, model_name, prompt_tokens_total, completion_tokens_total, total_tokens_total, message_count, update_time ) VALUES ( #{statDate}, #{userId}, #{appId}, #{modelName}, #{promptTokens}, #{completionTokens}, #{totalTokens}, 1, NOW() ) ON DUPLICATE KEY UPDATE prompt_tokens_total = prompt_tokens_total + VALUES(prompt_tokens_total), completion_tokens_total = completion_tokens_total + VALUES(completion_tokens_total), total_tokens_total = total_tokens_total + VALUES(total_tokens_total), message_count = message_count + 1, update_time = NOW();

这个SQL是整个统计模块的核心。它保证即使同一个会话有多轮并发对话，累计数字也不会串。

6. 常见问题排查与避坑指南

6.1 历史消息重复加载导致上下文爆炸

现象：对话进行到第五轮，发送的prompt里历史消息变成了十五轮的量，token消耗飙升。

排查思路：打印ChatMemory.get返回的消息数量，发现重复。原因是add被调用多次，同样的消息插入了多遍。

解决：按我上面说的方法加去重。另外一个隐蔽的坑是Spring AI从某个版本开始，ChatMemoryAdvisor默认会包含系统消息在内的全部历史记录，如果你的系统消息是动态拼接的，会反复累积。可以在Advisor中通过param排除：

.param(ChatMemoryAdvisor.CHAT_MEMORY_RETRIEVE_SIZE_KEY, 20)

限制只取最近20条消息，可以兜底控制上下文长度。

6.2 流式对话token统计为0

现象：普通对话token统计正常，流式对话全部为0或只有最后一轮的数值。

原因分析：流式场景下ChatResponse的usage字段是分片的，如果我在doOnNext里取到的是null，说明模型供应商没有在流式协议中返回完整的usage统计。

解决：建议不要只依赖流式协议的usage。可以改为根据实际发送的文本内容，用tokenizer自己估算。或者更简单：在流式结束后，用最终返回结果重新构造一次TokenUsage：

TokenUsage lastUsage = new TokenUsage( estimatedInputTokens, estimatedOutputTokens, estimatedInputTokens + estimatedOutputTokens );

同时把estimatedInputTokens的计算方式定为：历史消息总字符数/4 + 当前输入的字符数/4，这个估算虽然不精确，但误差控制在10%以内，完全可以满足统计报表的需求。

6.3 Advisor执行顺序错乱导致统计缺失

现象：对话能正常返回，但token记录里经常缺数据，或者记录的输入token数偏少。

排查过程：我先在TokenUsageAdvisor里加日志，打印进来的userText，发现没有包含任何历史消息。这说明ChatMemoryAdvisor还没有执行，历史prompt没有拼接上。

解决：给TokenUsageAdvisor设置更高的order值（比如10），确保在ChatMemoryAdvisor（order=0）之后执行。改完再测，数据就正常了。

6.4 懒加载与事务边界问题

现象：调用ChatMemory.get时，从数据库查出来的消息是实体对象，但Spring AI内部要求返回List ，我转换时报LazyInitializationException。

原因：Entity在事务提交后变成了detached状态，再访问懒加载字段就爆异常。

解决：在Repository层直接用DTO投影或原生SQL查出来映射成Spring AI的Message对象，不要传Entity在外面用。

@Query("SELECT new com.example.ai.entity.ChatMessageDTO(m.conversationId, m.messageType, m.content, m.createTime) " + "FROM ChatMessageEntity m WHERE m.conversationId = :conversationId ORDER BY m.createTime ASC") List<ChatMessageDTO> findMessagesByConversationId(String conversationId);

这样查询返回的就是轻量DTO，后续组装成Spring AI的Message没有任何外部依赖。

6.5 常见问题速查表

7. 经验总结与后续扩展思路

做这个项目最大的体会就是：Spring AI的Advisor机制是处理横切需求的最佳位置。持久化、token统计、甚至后面的敏感词过滤、知识库召回，全部可以塞进Advisor里。业务代码始终只需要关心"问什么、拿什么答案"，其他的一切交给框架。

关于扩展，我建议下一步可以这样推进：

• 把自定义ChatMemory、TokenUsageAdvisor、表结构脚本打包成一个独立starter，供团队内多个项目复用。目前已经做到了，新项目引入这个starter依赖，配置好数据源，直接获得稳定的对话持久化能力。
• 对话导出功能。有了结构化的消息表和会话表，按日期范围导出某个用户或某个应用的会话记录是很自然的扩展。可以做成异步导出Excel，为运营分析提供数据。
• 多模型场景下的token成本对比。现在统计表里已经有model_name维度，按模型分组查询总消耗，可以直观看到哪款模型成本最高、哪款回复质量最好，帮团队做模型选型决策。
• 基于消息内容的向量化存储。目前的表结构是纯关系型，后续要接语义检索，可以考虑把消息内容同步到向量库。但因为元数据和向量分开存储，不受影响。

我个人在实际操作中最深的感受是：对话数据相比其他业务数据更敏感，涉及用户输入和模型输出，设计存储结构时一定要考虑权限隔离和生命周期管理。建议按user_id做数据隔离，定期清理过期的会话数据，避免数据膨胀拖垮查询性能。

这个方案的完整复现路径就在上面：自定义表结构建三张表、自定义ChatMemory替换默认实现、自定义Advisor做token拦截、注意Advisor顺序和流式响应处理。核心代码加起来也就一两百行，但解决的是项目从demo走向生产的关键一步。如果大家在自己的项目里按这个思路落地，大概率不会再被对话记忆和token统计这两个问题反复折磨。