LangChain4j Java AI 应用开发实战（五）：流式响应与对话记忆 - 提升用户体验的关键技术

xiaobing259

508人浏览 · 2026-05-25 08:00:00

xiaobing259 · 2026-05-25 08:00:00 发布

系列篇章💥

No.	文章
1	LangChain4j Java AI 应用开发实战（一）：LangChain4j 快速入门指南
2	LangChain4j Java AI 应用开发实战（二）：大模型参数调优实战：Temperature、TopP、MaxTokens 深度解析
3	LangChain4j Java AI 应用开发实战（三）：多模态 AI 开发 - 图片理解与图像生成实战
4	LangChain4j Java AI 应用开发实战（四）：提示词工程进阶 - 模板化与结构化 Prompt 设计
5	LangChain4j Java AI 应用开发实战（五）：流式响应与对话记忆 - 提升用户体验的关键技术

系列篇章💥
前言
一、为什么需要流式响应？
- 1.1 同步调用的痛点
- 1.2 流式响应的优势
二、流式响应：StreamingExamples 详解
三、对话记忆：ChatMemoryExamples 详解
四、持久化记忆：重启不丢失
五、常见问题与避坑指南
六、生产环境最佳实践
结语

前言

同步阻塞的 AI 调用让用户等待数秒甚至数十秒，体验极差；没有记忆的对话机器人每轮都是"初次见面"，无法进行连贯交流。流式响应和对话记忆是构建优质 AI 应用的两大核心技术。本文将深入讲解 LangChain4j 的 StreamingChatModel 实现逐字实时输出，以及 ChatMemory 管理多轮对话上下文。你将掌握 TokenWindowChatMemory 的滑动窗口机制、MessageWindowChatMemory 的消息数量控制，并通过实战案例构建支持多轮交互的智能客服，让 AI 应用具备实时反馈能力和长期记忆，为用户提供自然流畅的对话体验。

一、为什么需要流式响应？

1.1 同步调用的痛点

传统的 chat() 方法是同步阻塞的：

// ❌ 同步调用：用户必须等待完整回复
String answer = chatModel.chat("请写一篇 1000 字的文章");
System.out.println(answer);  // 10-30 秒后才看到结果

问题：

⏱️ 长时间等待：长文本生成可能需要 10-30 秒
😰 用户焦虑：界面卡住，用户不知道是否还在处理
📱 移动端体验差：网络不稳定时更容易超时
💸 资源浪费：服务器线程长时间阻塞

1.2 流式响应的优势

流式响应（Streaming）逐字返回生成的内容：

// ✅ 流式调用：立即开始显示
model.chat(prompt, new StreamingChatResponseHandler() {
    @Override
    public void onPartialResponse(String partialResponse) {
        System.out.print(partialResponse);  // 逐字打印
    }
});

优势：

⚡ 即时反馈：第一个字在 1-2 秒内显示
👀 打字机效果：用户看到文字逐个出现
📊 进度感知：用户知道 AI 正在工作
🔄 可中断：用户可以随时停止生成
💻 资源高效：非阻塞异步处理

二、流式响应：StreamingExamples 详解

2.1 完整代码解析

package com.langchain4j;

import dev.langchain4j.model.chat.response.ChatResponse;
import dev.langchain4j.model.chat.response.StreamingChatResponseHandler;
import dev.langchain4j.model.openai.OpenAiStreamingChatModel;

import java.util.concurrent.CompletableFuture;

public class T05_StreamingExamples {

    public static void main(String[] args) {

        // 1. 构建 OpenAI 流式聊天模型实例
        OpenAiStreamingChatModel model = OpenAiStreamingChatModel.builder()
                .baseUrl("http://langchain4j.dev/demo/openai/v1")
                .modelName("gpt-4o-mini")
                .apiKey("demo")
                .build();

        // 2. 定义发送给模型的提示词
        String prompt = "who are you?";

        // 3. 创建 CompletableFuture 用于异步等待流式响应完成
        CompletableFuture<ChatResponse> futureChatResponse = new CompletableFuture<>();

        // 4. 发送流式聊天请求，并传入自定义的流式响应处理器
        model.chat(prompt, new StreamingChatResponseHandler() {

            /**
             * 每当模型生成一段新文本时触发（即收到一个数据块/Chunk）
             */
            @Override
            public void onPartialResponse(String partialResponse) {
                System.out.print(partialResponse);  // 逐段打印，不换行
            }

            /**
             * 当模型完成全部文本生成后触发
             */
            @Override
            public void onCompleteResponse(ChatResponse completeResponse) {
                System.out.println("\n\nDone streaming");
                futureChatResponse.complete(completeResponse);
            }

            /**
             * 当流式传输过程中发生异常时触发
             */
            @Override
            public void onError(Throwable error) {
                futureChatResponse.completeExceptionally(error);
            }
        });

        // 5. 阻塞主线程，等待流式响应完全结束
        futureChatResponse.join();
    }
}

输出效果（逐字显示）：

I am an AI assistant designed to help answer questions and provide information...
Done streaming

2.2 核心概念解析

（1）OpenAiStreamingChatModel vs OpenAiChatModel

特性	OpenAiChatModel	OpenAiStreamingChatModel
调用方式	同步阻塞	异步非阻塞
返回类型	`String`	无返回值（通过回调）
响应速度	等待完整回复	立即开始接收
适用场景	后台任务、批处理	实时交互、聊天界面
复杂度	简单	需处理回调

（2）StreamingChatResponseHandler 接口

这是流式响应的核心回调接口，包含三个方法：

public interface StreamingChatResponseHandler {
    
    // 1. 部分响应（多次调用）
    void onPartialResponse(String partialResponse);
    
    // 2. 完成响应（调用一次）
    void onCompleteResponse(ChatResponse completeResponse);
    
    // 3. 错误处理（调用一次，如果出错）
    void onError(Throwable error);
}

执行流程：

开始调用
   ↓
onPartialResponse("I")        ← 第 1 个 token
onPartialResponse(" am")      ← 第 2 个 token
onPartialResponse(" an")      ← 第 3 个 token
...                           ← 持续接收
onPartialResponse("assistant")← 最后几个 token
   ↓
onCompleteResponse(...)       ← 完成，包含元数据

（3）CompletableFuture 的作用

流式调用是异步非阻塞的，main 方法会立即继续执行。如果需要等待完成，使用 CompletableFuture：

// 创建 Future
CompletableFuture<ChatResponse> future = new CompletableFuture<>();

// 在 onCompleteResponse 中标记完成
@Override
public void onCompleteResponse(ChatResponse response) {
    future.complete(response);  // 标记成功
}

// 在 onError 中标记失败
@Override
public void onError(Throwable error) {
    future.completeExceptionally(error);  // 标记失败
}

// 等待完成（阻塞）
future.join();  // 或 future.get()

注意：在 Web 应用中，通常不需要 join()，由框架管理异步生命周期。

2.3 实战案例 1：Web 聊天界面（Spring Boot + SSE）

（1）需求

实现类似 ChatGPT 的网页聊天，AI 回复逐字显示。

（2）后端代码（Spring Boot Controller）

@RestController
@RequestMapping("/api/chat")
public class ChatController {

    private final OpenAiStreamingChatModel chatModel;

    public ChatController() {
        this.chatModel = OpenAiStreamingChatModel.builder()
            .apiKey(System.getenv("OPENAI_API_KEY"))
            .modelName("gpt-4o-mini")
            .build();
    }

    /**
     * 流式聊天接口（Server-Sent Events）
     */
    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public SseEmitter streamChat(@RequestParam String message) {
        
        // 创建 SSE 发射器，设置超时时间 60 秒
        SseEmitter emitter = new SseEmitter(60_000L);
        
        // 异步处理流式响应
        chatModel.chat(message, new StreamingChatResponseHandler() {
            
            @Override
            public void onPartialResponse(String partialResponse) {
                try {
                    // 发送每个 token 到前端
                    emitter.send(SseEmitter.event()
                        .name("message")
                        .data(partialResponse)
                    );
                } catch (IOException e) {
                    emitter.completeWithError(e);
                }
            }

            @Override
            public void onCompleteResponse(ChatResponse completeResponse) {
                // 发送完成信号
                try {
                    emitter.send(SseEmitter.event()
                        .name("done")
                        .data("[DONE]")
                    );
                    emitter.complete();
                } catch (IOException e) {
                    emitter.completeWithError(e);
                }
            }

            @Override
            public void onError(Throwable error) {
                emitter.completeWithError(error);
            }
        });
        
        return emitter;
    }
}

（3）前端代码（JavaScript）

function sendMessage(message) {
    const eventSource = new EventSource(`/api/chat/stream?message=${encodeURIComponent(message)}`);
    
    const chatBox = document.getElementById('chat-box');
    let aiMessageDiv = document.createElement('div');
    aiMessageDiv.className = 'ai-message';
    chatBox.appendChild(aiMessageDiv);
    
    eventSource.addEventListener('message', (event) => {
        // 逐字追加显示
        aiMessageDiv.textContent += event.data;
        
        // 自动滚动到底部
        chatBox.scrollTop = chatBox.scrollHeight;
    });
    
    eventSource.addEventListener('done', () => {
        eventSource.close();
        console.log('AI 回复完成');
    });
    
    eventSource.onerror = (error) => {
        console.error('SSE 错误:', error);
        eventSource.close();
    };
}

// 使用
sendMessage("你好，请介绍一下自己");

效果：用户输入后，AI 回复像打字机一样逐字显示，体验流畅。

2.4 实战案例 2：控制台交互式聊天机器人

（1）需求

在命令行中实现多轮对话，支持流式输出。

（2）完整代码

public class InteractiveChatBot {

    private final OpenAiStreamingChatModel model;

    public InteractiveChatBot() {
        this.model = OpenAiStreamingChatModel.builder()
            .apiKey(System.getenv("OPENAI_API_KEY"))
            .modelName("gpt-4o-mini")
            .build();
    }

    public void startChat() throws Exception {
        
        Scanner scanner = new Scanner(System.in);
        System.out.println("🤖 聊天机器人已启动！输入 'exit' 退出\n");

        while (true) {
            System.out.print("👤 你：");
            String userInput = scanner.nextLine();

            if ("exit".equalsIgnoreCase(userInput)) {
                System.out.println("🤖 再见！");
                break;
            }

            System.out.print("🤖 AI：");
            
            // 流式输出
            CompletableFuture<Void> future = new CompletableFuture<>();
            
            model.chat(userInput, new StreamingChatResponseHandler() {
                
                @Override
                public void onPartialResponse(String partialResponse) {
                    System.out.print(partialResponse);
                }

                @Override
                public void onCompleteResponse(ChatResponse completeResponse) {
                    System.out.println();  // 换行
                    future.complete(null);
                }

                @Override
                public void onError(Throwable error) {
                    System.err.println("\n❌ 错误：" + error.getMessage());
                    future.completeExceptionally(error);
                }
            });
            
            // 等待本轮对话完成
            future.join();
            System.out.println();  // 空行分隔
        }

        scanner.close();
    }

    public static void main(String[] args) throws Exception {
        new InteractiveChatBot().startChat();
    }
}

运行效果：

🤖 聊天机器人已启动！输入 'exit' 退出

👤 你：你好
🤖 AI：你好！很高兴与你交流。有什么我可以帮助你的吗？

👤 你：你是谁
🤖 AI：我是一个人工智能助手，旨在回答问题和提供信息...

👤 你：exit
🤖 再见！

2.5 性能优化技巧

（1）缓冲输出

频繁的系统调用会影响性能，使用缓冲：

@Override
public void onPartialResponse(String partialResponse) {
    // ❌ 每次调用都写入系统输出（慢）
    System.out.print(partialResponse);
    
    // ✅ 使用 StringBuilder 缓冲（快）
    buffer.append(partialResponse);
    if (buffer.length() > 50) {  // 每 50 个字符刷新一次
        System.out.print(buffer.toString());
        buffer.setLength(0);
    }
}

（2）背压控制

如果前端处理速度慢，需要控制发送速率：

private long lastSendTime = 0;
private static final long MIN_INTERVAL_MS = 50;  // 最小间隔 50ms

@Override
public void onPartialResponse(String partialResponse) {
    long now = System.currentTimeMillis();
    if (now - lastSendTime >= MIN_INTERVAL_MS) {
        emitter.send(partialResponse);
        lastSendTime = now;
    } else {
        // 累积到缓冲区，稍后发送
        buffer.append(partialResponse);
    }
}

（3）超时处理

防止长时间无响应：

SseEmitter emitter = new SseEmitter(30_000L);  // 30 秒超时

// 设置超时回调
emitter.onTimeout(() -> {
    logger.warn("SSE 连接超时");
    emitter.complete();
});

三、对话记忆：ChatMemoryExamples 详解

3.1 为什么需要对话记忆？

（1）没有记忆的问题

用户：我叫张三
AI：你好，张三！

用户：我叫什么名字？
AI：抱歉，我不知道你的名字。  ← ❌ 忘记了！

原因：每次调用都是独立的，AI 不记得之前的对话。

（2）有记忆的效果

用户：我叫张三
AI：你好，张三！

用户：我叫什么名字？
AI：你叫张三。  ← ✅ 记住了！

3.2 完整代码解析

package com.langchain4j;

import dev.langchain4j.data.message.AiMessage;
import dev.langchain4j.data.message.SystemMessage;
import dev.langchain4j.data.message.UserMessage;
import dev.langchain4j.memory.ChatMemory;
import dev.langchain4j.memory.chat.TokenWindowChatMemory;
import dev.langchain4j.model.chat.response.StreamingChatResponseHandler;
import dev.langchain4j.model.openai.OpenAiStreamingChatModel;
import dev.langchain4j.model.openai.OpenAiTokenCountEstimator;

import java.util.concurrent.CompletableFuture;

public class T06_ChatMemoryExamples {

    public static void main(String[] args) throws Exception {

        OpenAiStreamingChatModel model = OpenAiStreamingChatModel.builder()
                .baseUrl("http://langchain4j.dev/demo/openai/v1")
                .modelName("gpt-4o-mini")
                .apiKey("demo")
                .build();

        // 创建对话记忆（最多保留 1000 个 Tokens）
        ChatMemory chatMemory = TokenWindowChatMemory.withMaxTokens(
            1000, 
            new OpenAiTokenCountEstimator(GPT_4_O_MINI)
        );

        // 添加系统消息（角色设定）
        SystemMessage systemMessage = SystemMessage.from(
            "You are a senior developer explaining to another senior developer, " +
            "the project you are working on is an e-commerce platform with Java back-end, " +
            "Oracle database, and Spring Data JPA"
        );
        chatMemory.add(systemMessage);

        // 第一轮对话
        UserMessage userMessage1 = UserMessage.userMessage(
            "How do I optimize database queries for a large-scale e-commerce platform? " +
            "Answer short in three to five lines maximum."
        );
        chatMemory.add(userMessage1);

        System.out.println("[User]: " + userMessage1.singleText());
        System.out.print("[LLM]: ");

        AiMessage aiMessage1 = streamChat(model, chatMemory);
        chatMemory.add(aiMessage1);  // 保存 AI 回复到记忆

        // 第二轮对话（依赖第一轮上下文）
        UserMessage userMessage2 = UserMessage.userMessage(
            "Give a concrete example implementation of the first point? " +
            "Be short, 10 lines of code maximum."
        );
        chatMemory.add(userMessage2);

        System.out.println("\n\n[User]: " + userMessage2.singleText());
        System.out.print("[LLM]: ");

        AiMessage aiMessage2 = streamChat(model, chatMemory);
        chatMemory.add(aiMessage2);
    }

    private static AiMessage streamChat(OpenAiStreamingChatModel model, ChatMemory chatMemory)
            throws Exception {

        CompletableFuture<AiMessage> futureAiMessage = new CompletableFuture<>();

        StreamingChatResponseHandler handler = new StreamingChatResponseHandler() {

            @Override
            public void onPartialResponse(String partialResponse) {
                System.out.print(partialResponse);
            }

            @Override
            public void onCompleteResponse(ChatResponse completeResponse) {
                futureAiMessage.complete(completeResponse.aiMessage());
            }

            @Override
            public void onError(Throwable throwable) {
                futureAiMessage.completeExceptionally(throwable);
            }
        };

        // 关键：发送完整的对话历史给模型
        model.chat(chatMemory.messages(), handler);
        return futureAiMessage.get();
    }
}

输出示例：

[User]: How do I optimize database queries for a large-scale e-commerce platform? Answer short in three to five lines maximum.
[LLM]: To optimize database queries for a large-scale e-commerce platform, focus on proper indexing of frequently accessed columns, utilize pagination for large result sets, and implement caching strategies to minimize database hits. Additionally, analyze query execution plans to identify slow queries and optimize them, and make use of batch processing for bulk operations to reduce overhead.

[User]: Give a concrete example implementation of the first point? Be short, 10 lines of code maximum.
[LLM]: Certainly! Here's an example of how to create an index on the `product_name` column in an `products` table using an SQL statement:

\```sql
CREATE INDEX idx_product_name ON products(product_name);
\```
In Spring Data JPA, you can also use annotations to define indexes in your entity class:

\```java
@Entity
@Table(name = "products", indexes = @Index(name = "idx_product_name", columnList = "product_name"))
public class Product {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(nullable = false)
    private String productName;
}

关键点：第二轮对话中，AI 理解 “the first point” 指的是第一轮提到的 “proper indexing”，这就是记忆的作用。

3.3 核心概念解析

（1）ChatMemory 接口

ChatMemory 是对话记忆的核心抽象：

public interface ChatMemory {
    
    // 添加消息
    void add(ChatMessage message);
    
    // 获取所有消息
    List<ChatMessage> messages();
    
    // 清除记忆
    void clear();
}

消息类型：

类型	说明	示例
SystemMessage	系统指令，设定 AI 角色	“你是客服助手”
UserMessage	用户输入	“你好”
AiMessage	AI 回复	“你好！有什么帮助？”

（2）TokenWindowChatMemory

这是最常用的实现，基于 Token 数量限制记忆大小：

ChatMemory chatMemory = TokenWindowChatMemory.withMaxTokens(
    1000,                              // 最大 1000 Tokens
    new OpenAiTokenCountEstimator(model) // Token 估算器
);

工作原理：

添加消息 → 计算总 Tokens → 超过限制？→ 删除最早的消息

滑动窗口机制：

初始状态：
[SystemMessage] [User1] [AI1]                          = 300 Tokens

添加 User2：
[SystemMessage] [User1] [AI1] [User2]                  = 400 Tokens

添加 AI2：
[SystemMessage] [User1] [AI1] [User2] [AI2]            = 500 Tokens

...持续添加...

超过 1000 Tokens 时：
[User3] [AI3] [User4] [AI4] [User5] [AI5]              = 950 Tokens
 ↑ 删除最早的消息，保留最近的对话

优势：

✅ 自动控制成本（Token 数决定 API 费用）
✅ 避免超出模型上下文限制
✅ 保留最近的对话（最相关）

（3）MessageWindowChatMemory

另一种实现，基于消息数量限制：

ChatMemory chatMemory = MessageWindowChatMemory.withMaxMessages(10);

特点：

保留最近 10 条消息
不考虑 Token 数量
适合短消息场景

对比：

特性	TokenWindowChatMemory	MessageWindowChatMemory
限制依据	Token 数量	消息数量
成本控制	✅ 精确	❌ 不确定
适用场景	长文本、生产环境	短消息、测试环境
复杂度	需要 Token 估算器	简单

推荐：生产环境优先使用 TokenWindowChatMemory。

3.4 实战案例：多轮对话客服系统

（1）需求

构建支持多轮交互的客服机器人，能记住用户订单信息。

（2）完整代码

public class CustomerServiceWithMemory {

    private final OpenAiStreamingChatModel model;
    private final ChatMemory chatMemory;

    public CustomerServiceWithMemory() {
        this.model = OpenAiStreamingChatModel.builder()
            .apiKey(System.getenv("OPENAI_API_KEY"))
            .modelName("gpt-4o-mini")
            .temperature(0.3)
            .build();

        // 初始化对话记忆（最多 2000 Tokens）
        this.chatMemory = TokenWindowChatMemory.withMaxTokens(
            2000,
            new OpenAiTokenCountEstimator(GPT_4_O_MINI)
        );

        // 添加系统指令
        SystemMessage systemMessage = SystemMessage.from("""
            你是某电商平台的智能客服助手。
            
            职责：
            1. 解答用户关于订单、物流、退款等问题
            2. 语气友好、专业、简洁
            3. 如果无法回答，引导用户联系人工客服
            
            重要：请记住用户的订单号和相关信息，以便在多轮对话中引用。
            """);
        
        chatMemory.add(systemMessage);
    }

    public String chat(String userInput) throws Exception {
        
        // 添加用户消息
        chatMemory.add(UserMessage.from(userInput));

        // 流式输出
        StringBuilder response = new StringBuilder();
        CompletableFuture<Void> future = new CompletableFuture<>();

        model.chat(chatMemory.messages(), new StreamingChatResponseHandler() {
            
            @Override
            public void onPartialResponse(String partialResponse) {
                response.append(partialResponse);
                System.out.print(partialResponse);
            }

            @Override
            public void onCompleteResponse(ChatResponse completeResponse) {
                System.out.println();
                
                // 保存 AI 回复到记忆
                chatMemory.add(completeResponse.aiMessage());
                
                future.complete(null);
            }

            @Override
            public void onError(Throwable error) {
                future.completeExceptionally(error);
            }
        });

        future.join();
        return response.toString();
    }

    public static void main(String[] args) throws Exception {
        
        CustomerServiceWithMemory bot = new CustomerServiceWithMemory();
        Scanner scanner = new Scanner(System.in);

        System.out.println("🤖 客服机器人已启动！\n");

        while (true) {
            System.out.print("👤 用户：");
            String input = scanner.nextLine();

            if ("exit".equalsIgnoreCase(input)) {
                break;
            }

            System.out.print("🤖 客服：");
            bot.chat(input);
            System.out.println();
        }

        scanner.close();
    }
}

对话示例：

👤 用户：我的订单号是 ORD-2024-001234，想查询物流状态
🤖 客服：您好！我已记录您的订单号 ORD-2024-001234。让我为您查询物流状态...
         您的订单目前已发货，预计 2024-03-20 送达。快递公司为顺丰速运，
         运单号 SF1234567890。您可以通过顺丰官网追踪详细物流信息。

👤 用户：如果我没收到怎么办？
🤖 客服：如果您在预计送达时间（2024-03-20）后仍未收到订单 ORD-2024-001234，
         建议您：
         1. 检查快递追踪信息，确认是否有异常
         2. 联系快递员或当地顺丰网点
         3. 如仍无法解决，我们可以为您申请补发或退款
         
         请问还有其他可以帮助您的吗？

关键点：第二轮对话中，AI 记得订单号 ORD-2024-001234 和预计送达时间，无需用户重复提供。

四、持久化记忆：重启不丢失

4.1 问题

内存中的 ChatMemory 在程序重启后会丢失所有对话历史。

4.2 解决方案：持久化存储

LangChain4j 提供多种持久化方案：

存储类型	适用场景	优点	缺点
内存	开发调试	快速、简单	重启丢失
SQLite	小规模应用	零配置、文件存储	并发性能一般
MySQL	中大规模应用	高性能、可扩展	需要数据库运维
Redis	高并发场景	极速、支持 TTL	成本高、需运维

4.3 实战：MySQL 持久化

（1）数据库表设计

CREATE TABLE chat_memory (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    memory_id VARCHAR(255) NOT NULL,  -- 用户 ID 或会话 ID
    message_index INT NOT NULL,        -- 消息索引
    message_type VARCHAR(50) NOT NULL, -- SYSTEM/USER/AI
    content TEXT NOT NULL,             -- 消息内容
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_memory_id (memory_id),
    UNIQUE KEY uk_memory_index (memory_id, message_index)
);

（2）JPA Entity

@Entity
@Table(name = "chat_memory")
@Data
public class ChatMemoryEntity {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    @Column(name = "memory_id", nullable = false)
    private String memoryId;

    @Column(name = "message_index", nullable = false)
    private Integer messageIndex;

    @Column(name = "message_type", nullable = false)
    private String messageType;

    @Column(name = "content", columnDefinition = "LONGTEXT")
    private String content;

    @Column(name = "created_at")
    private LocalDateTime createdAt;
}

（3）Repository

@Repository
public interface ChatMemoryRepository extends JpaRepository<ChatMemoryEntity, Long> {
    
    List<ChatMemoryEntity> findByMemoryIdOrderByMessageIndexAsc(String memoryId);
    
    void deleteByMemoryId(String memoryId);
    
    int countByMemoryId(String memoryId);
}

（4）自定义 ChatMemoryStore

@Component
public class MySqlChatMemoryStore implements ChatMemoryStore {

    @Autowired
    private ChatMemoryRepository repository;

    @Override
    public List<ChatMessage> getMessages(Object memoryId) {
        String memId = (String) memoryId;
        
        return repository.findByMemoryIdOrderByMessageIndexAsc(memId)
            .stream()
            .map(this::convertToChatMessage)
            .collect(Collectors.toList());
    }

    @Override
    public void updateMessages(Object memoryId, List<ChatMessage> messages) {
        String memId = (String) memoryId;
        
        // 删除旧消息
        repository.deleteByMemoryId(memId);
        
        // 保存新消息
        for (int i = 0; i < messages.size(); i++) {
            ChatMessage message = messages.get(i);
            
            ChatMemoryEntity entity = new ChatMemoryEntity();
            entity.setMemoryId(memId);
            entity.setMessageIndex(i);
            entity.setMessageType(getMessageType(message));
            entity.setContent(extractContent(message));
            entity.setCreatedAt(LocalDateTime.now());
            
            repository.save(entity);
        }
    }

    @Override
    public void deleteMessages(Object memoryId) {
        repository.deleteByMemoryId((String) memoryId);
    }

    private ChatMessage convertToChatMessage(ChatMemoryEntity entity) {
        switch (entity.getMessageType()) {
            case "SYSTEM":
                return SystemMessage.from(entity.getContent());
            case "USER":
                return UserMessage.from(entity.getContent());
            case "AI":
                return AiMessage.from(entity.getContent());
            default:
                throw new IllegalArgumentException("未知消息类型：" + entity.getMessageType());
        }
    }

    private String getMessageType(ChatMessage message) {
        if (message instanceof SystemMessage) return "SYSTEM";
        if (message instanceof UserMessage) return "USER";
        if (message instanceof AiMessage) return "AI";
        throw new IllegalArgumentException("未知消息类型");
    }

    private String extractContent(ChatMessage message) {
        return message.text();
    }
}

（5）使用持久化记忆

@Service
public class PersistentChatService {

    @Autowired
    private MySqlChatMemoryStore memoryStore;

    @Autowired
    private OpenAiStreamingChatModel chatModel;

    public String chat(String userId, String userInput) throws Exception {
        
        // 创建持久化记忆
        ChatMemory chatMemory = ChatMemory.builder()
            .id(userId)  // 用户 ID 作为记忆 ID
            .chatMemoryStore(memoryStore)
            .maxTokens(2000)
            .tokenCountEstimator(new OpenAiTokenCountEstimator(GPT_4_O_MINI))
            .build();

        // 添加用户消息
        chatMemory.add(UserMessage.from(userInput));

        // 调用模型
        StringBuilder response = new StringBuilder();
        CompletableFuture<Void> future = new CompletableFuture<>();

        chatModel.chat(chatMemory.messages(), new StreamingChatResponseHandler() {
            
            @Override
            public void onPartialResponse(String partialResponse) {
                response.append(partialResponse);
            }

            @Override
            public void onCompleteResponse(ChatResponse completeResponse) {
                chatMemory.add(completeResponse.aiMessage());  // 自动保存到 MySQL
                future.complete(null);
            }

            @Override
            public void onError(Throwable error) {
                future.completeExceptionally(error);
            }
        });

        future.join();
        return response.toString();
    }
}

效果：即使用户关闭浏览器、服务器重启，下次登录时对话历史依然存在。

五、常见问题与避坑指南

5.1 记忆溢出导致上下文超长

❌ 现象：

Error: This model's maximum context length is 8192 tokens. 
However, your messages resulted in 9500 tokens.

原因：未限制 ChatMemory 大小，对话历史累积超过模型限制。

✅ 解决方案：

// ✅ 正确：设置合理的 Token 限制
ChatMemory chatMemory = TokenWindowChatMemory.withMaxTokens(
    6000,  // 预留 2000 Tokens 给新消息（假设模型上限 8192）
    new OpenAiTokenCountEstimator(model)
);

建议：

GPT-4o-mini（128K 上下文）：设置 10000-20000 Tokens
GPT-3.5-turbo（16K 上下文）：设置 10000-12000 Tokens
始终预留 20% 空间给新消息

5.2 流式响应乱码

❌ 现象：

控制台输出中文乱码：浣犲ソ

原因：Windows 控制台默认编码不是 UTF-8。

✅ 解决方案：

方式 1：IDEA 配置

Run → Edit Configurations → VM options:
-Dfile.encoding=UTF-8

方式 2：代码中指定编码

@Override
public void onPartialResponse(String partialResponse) {
    try {
        OutputStreamWriter writer = new OutputStreamWriter(
            System.out, StandardCharsets.UTF_8
        );
        writer.write(partialResponse);
        writer.flush();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

5.3 SSE 连接中断

❌ 现象：

前端收到部分数据后连接断开，报错 EventSource connection closed。

原因：

网络不稳定
服务器超时
代理服务器缓冲

✅ 解决方案：

方式 1：增加超时时间

SseEmitter emitter = new SseEmitter(120_000L);  // 120 秒

方式 2：心跳保活

// 每 15 秒发送一次心跳
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.scheduleAtFixedRate(() -> {
    try {
        emitter.send(SseEmitter.event()
            .name("heartbeat")
            .data(":ping")  // SSE 注释，不会显示在前端
        );
    } catch (IOException e) {
        scheduler.shutdown();
    }
}, 0, 15, TimeUnit.SECONDS);

方式 3：前端自动重连

function connectWithRetry(url, maxRetries = 3) {
    let retries = 0;
    
    function connect() {
        const eventSource = new EventSource(url);
        
        eventSource.onopen = () => {
            console.log('SSE 连接成功');
            retries = 0;  // 重置重试计数
        };
        
        eventSource.onerror = (error) => {
            console.error('SSE 错误:', error);
            eventSource.close();
            
            if (retries < maxRetries) {
                retries++;
                console.log(`第 ${retries} 次重连...`);
                setTimeout(connect, 2000 * retries);  // 指数退避
            } else {
                alert('连接失败，请刷新页面重试');
            }
        };
        
        return eventSource;
    }
    
    return connect();
}

// 使用
const es = connectWithRetry('/api/chat/stream?message=你好');

5.4 多用户记忆隔离

❌ 现象：

用户 A 的对话历史泄露给用户 B。

原因：所有用户共享同一个 ChatMemory 实例。

✅ 解决方案：

方式 1：按用户 ID 隔离（推荐）

@Service
public class IsolatedChatService {

    private final ConcurrentHashMap<String, ChatMemory> userMemories = new ConcurrentHashMap<>();

    public String chat(String userId, String userInput) {
        
        // 为每个用户创建独立的记忆
        ChatMemory chatMemory = userMemories.computeIfAbsent(userId, id -> 
            TokenWindowChatMemory.withMaxTokens(2000, tokenEstimator)
        );

        chatMemory.add(UserMessage.from(userInput));
        
        // ... 调用模型
        
        return response;
    }

    // 清除用户记忆（退出登录时）
    public void clearUserMemory(String userId) {
        ChatMemory memory = userMemories.remove(userId);
        if (memory != null) {
            memory.clear();
        }
    }
}

方式 2：使用 @MemoryId 注解（Spring Boot + AI Service）

@AiService
public interface ChatAssistant {

    @SystemMessage("你是智能客服助手")
    String chat(@MemoryId String userId, @UserMessage String message);
}

// Spring 自动为每个 userId 创建独立记忆
@Autowired
private ChatAssistant assistant;

public String chat(String userId, String message) {
    return assistant.chat(userId, message);  // 自动隔离
}

5.5 流式响应与记忆结合的陷阱

❌ 问题：

在 onPartialResponse 中无法获取完整的 AI 消息，导致记忆保存不完整。

@Override
public void onPartialResponse(String partialResponse) {
    // ❌ 错误：这里只有部分文本
    chatMemory.add(AiMessage.from(partialResponse));
}

✅ 正确做法：

@Override
public void onCompleteResponse(ChatResponse completeResponse) {
    // ✅ 正确：在完成后保存完整消息
    AiMessage aiMessage = completeResponse.aiMessage();
    chatMemory.add(aiMessage);
}

六、生产环境最佳实践

6.1 监控与日志

（1）记录关键指标

@Component
public class ChatMetricsRecorder {

    private final MeterRegistry meterRegistry;

    public void recordStreamingDuration(String userId, long durationMs) {
        meterRegistry.timer("chat.streaming.duration", "userId", userId)
            .record(durationMs, TimeUnit.MILLISECONDS);
    }

    public void recordTokenUsage(String userId, int inputTokens, int outputTokens) {
        meterRegistry.counter("chat.tokens.input", "userId", userId)
            .increment(inputTokens);
        meterRegistry.counter("chat.tokens.output", "userId", userId)
            .increment(outputTokens);
    }

    public void recordMemorySize(String userId, int messageCount, int tokenCount) {
        meterRegistry.gauge("chat.memory.messages", 
            Tags.of("userId", userId), messageCount);
        meterRegistry.gauge("chat.memory.tokens", 
            Tags.of("userId", userId), tokenCount);
    }
}

（2）Grafana 看板

监控以下指标：

平均响应时间（P50/P95/P99）
Token 用量趋势
记忆大小分布
错误率

6.2 性能优化

（1）异步处理

@Service
public class AsyncChatService {

    @Async("chatExecutor")
    public CompletableFuture<String> chatAsync(String userId, String message) {
        
        // 异步处理，不阻塞主线程
        String response = chat(userId, message);
        return CompletableFuture.completedFuture(response);
    }

    @Bean("chatExecutor")
    public Executor chatExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("chat-");
        return executor;
    }
}

（2）缓存常用回复

@Service
public class CachedChatService {

    private final Cache<String, String> responseCache = Caffeine.newBuilder()
        .maximumSize(1000)
        .expireAfterWrite(1, TimeUnit.HOURS)
        .build();

    public String chat(String userId, String message) {
        
        // 计算消息哈希
        String cacheKey = calculateHash(message);
        
        // 检查缓存
        String cachedResponse = responseCache.getIfPresent(cacheKey);
        if (cachedResponse != null) {
            return cachedResponse;  // 直接返回缓存
        }
        
        // 调用 AI
        String response = callAI(message);
        
        // 存入缓存
        responseCache.put(cacheKey, response);
        
        return response;
    }
}

6.3 安全加固

（1）输入过滤

public String sanitizeInput(String input) {
    // 移除潜在的危险内容
    return input
        .replaceAll("<script.*?>.*?</script>", "")  // XSS
        .replaceAll("(?i)ignore.*instruction", "")   // Prompt 注入
        .trim();
}

（2）速率限制

@Service
public class RateLimitedChatService {

    private final Map<String, RateLimiter> userLimiters = new ConcurrentHashMap<>();

    public String chat(String userId, String message) {
        
        RateLimiter limiter = userLimiters.computeIfAbsent(userId, id -> 
            RateLimiter.create(10.0)  // 每秒 10 次请求
        );

        if (!limiter.tryAcquire()) {
            throw new BusinessException("请求过于频繁，请稍后再试");
        }

        return callAI(message);
    }
}

结语

通过本文的学习，你已经掌握了 LangChain4j 流式响应和对话记忆的核心技术。从 OpenAiStreamingChatModel 实现逐字实时输出，到 TokenWindowChatMemory 管理多轮对话上下文，再到 MySQL 持久化存储和用户隔离，这些技术能让你的 AI 应用具备流畅的交互体验和长期记忆能力。记住，流式响应提升的是用户体验，对话记忆保障的是对话连贯性，两者结合才能构建真正智能的聊天机器人。下一篇我们将进入声明式 AI Service 的世界，学习如何通过接口定义即可获得 AI 能力，彻底告别繁琐的 HTTP 调用和响应解析，让代码更加简洁优雅！

在这里插入图片描述