Spring AI项目中VectorStoreChatMemoryAdvisor流式响应问题解析

问题背景

在Spring AI项目中，VectorStoreChatMemoryAdvisor作为聊天记忆实现的一个重要组件，负责将对话内容持久化到向量存储中。然而，当开发者尝试在流式响应模式下使用该组件时，系统会抛出异常，导致功能无法正常工作。

问题现象

当开发者配置了VectorStoreChatMemoryAdvisor并尝试以流式方式获取对话响应时，系统会抛出IllegalStateException异常，错误信息表明"exactly one of text or media must be specified"。这意味着在创建Document对象时，既没有提供文本内容也没有提供媒体内容，违反了Document类的构造约束。

技术分析

深入分析问题根源，我们可以发现几个关键点：

1. 流式响应处理机制：在流式模式下，响应内容是以分块形式逐步返回的，而不是一次性返回完整内容。VectorStoreChatMemoryAdvisor默认没有针对这种模式进行特殊处理。

2. 消息聚合缺失：原始实现中，advisor直接处理每个流式分块，而不是等待完整响应聚合后再处理。这导致它处理的是不完整的消息片段，其中可能缺少必要的文本或媒体内容。

3. 继承结构问题：VectorStoreChatMemoryAdvisor继承自BaseAdvisor，而BaseAdvisor提供的默认流式处理逻辑并不适合聊天记忆这种需要完整消息的场景。

解决方案

针对这个问题，正确的解决方法是重写adviseStream方法，确保在持久化到向量存储之前，先完整聚合流式响应。具体实现思路如下：

1. 引入消息聚合器：使用ChatClientMessageAggregator来收集所有流式分块，直到获得完整响应。

2. 延迟持久化：只有在获得完整响应后，才调用after方法将消息持久化到向量存储。

3. 线程调度优化：使用适当的调度器来确保处理过程不会阻塞主线程。

以下是关键代码示例（Kotlin）：

override fun adviseStream(
    chatClientRequest: ChatClientRequest,
    streamAdvisorChain: StreamAdvisorChain
): Flux<ChatClientResponse> {
    val scheduler: Scheduler = this.scheduler
    return Mono.just(chatClientRequest)
        .publishOn(scheduler)
        .map { request -> this.before(request, streamAdvisorChain) }
        .flatMapMany { streamAdvisorChain.nextStream(it) }
        .transform { flux ->
            ChatClientMessageAggregator().aggregateChatClientResponse(
                flux
            ) { response -> this.after(response, streamAdvisorChain) }
        }
}

最佳实践建议

1. 全面测试覆盖：对于涉及流式处理的组件，应该同时测试非流式和流式两种场景。

2. 明确处理边界：在设计advisor时，需要明确区分哪些操作适合在流式分块上执行，哪些需要等待完整响应。

3. 文档说明：对于需要特殊处理流式响应的advisor，应该在文档中明确说明其行为特点和使用限制。

4. 性能考量：在实现消息聚合时，需要注意内存使用情况，避免在处理大量或长时间运行的流时出现内存问题。

总结

这个问题揭示了在流式处理场景下实现聊天记忆功能时需要考虑的特殊情况。通过引入消息聚合机制，我们确保了只有在获得完整响应后才执行持久化操作，从而解决了原始实现中的问题。这个案例也提醒我们，在设计类似功能时，必须充分考虑各种响应模式下的行为差异，确保组件在所有场景下都能正常工作。

对于Spring AI项目的使用者来说，理解这个问题的本质有助于更好地使用和扩展聊天记忆功能，同时也为处理其他类似的流式处理场景提供了参考模式。