Java 玩转 AI 聊天？LangChain4J 三种流式响应方式全解析！

引言

随着人工智能技术的飞速发展，尤其是大型语言模型（LLMs）的崛起，构建智能聊天应用变得越来越普及。LangChain4j 是一个强大的 Java 库，旨在简化与 LLMs 的集成，帮助开发者快速构建各种 AI 应用，包括聊天机器人。在用户与 AI 聊天时，实时的、流式的响应能够显著提升用户体验，让对话感觉更自然和流畅。LangChain4j 提供了多种方式来实现流式响应，本文将深入探讨这三种主要的流式响应方式，并通过代码示例演示如何在 Java 中使用 LangChain4j 构建具有流式聊天能力的 AI 应用。

技术背景

1. 大型语言模型 (LLMs): 预训练的深度学习模型，能够理解和生成人类语言，例如 OpenAI 的 GPT 系列、Google 的 Gemini 系列等。这些模型是构建智能聊天应用的核心。

2. LangChain: 一个用于构建基于 LLMs 的应用的框架，提供了各种模块和工具，包括模型集成、提示管理、记忆、链式调用等。LangChain4j 是 LangChain 的 Java 移植版本。

3. 流式响应 (Streaming Response): 在传统的请求-响应模式中，服务器通常在处理完所有数据后才发送完整的响应。流式响应则允许服务器在生成数据的过程中逐步发送响应片段，客户端可以立即开始处理和展示这些片段，从而实现更快的首屏渲染和更流畅的用户体验。

4. Java Reactive Streams: Java 生态中用于处理异步数据流的标准规范，通过 java.util.concurrent.Flow API 提供支持。Reactive Streams 定义了发布者（Publisher）、订阅者（Subscriber）、处理器（Processor）等概念，用于构建响应式应用。

应用使用场景

流式响应在 AI 聊天应用中尤为重要，适用于以下场景：

• 长时间的 AI 回复: 当 LLM 需要生成较长的回复时，流式响应可以让用户更快地看到部分结果，而不是长时间等待空白屏幕。
• 实时互动: 用户可以根据 AI 已经生成的部分回复来调整自己的问题，实现更实时的互动。
• 提升用户体验: 逐步呈现 AI 的回答，营造更自然、更像真人对话的感觉。
• 减少感知延迟: 即使用户需要等待完整的回复，但看到内容逐步出现会感觉更快。
• 复杂的链式调用: 在 LangChain4j 中，一个请求可能涉及多个 LLM 调用或其他操作。流式响应可以逐步呈现每个步骤的结果。

LangChain4j 三种流式响应方式全解析

LangChain4j 主要通过以下三种方式提供流式响应：

1. ChatLanguageModel.generateStream(prompt): 直接从聊天语言模型获取响应的 Publisher<AiMessage>。每个 AiMessage 通常包含一个文本片段。

2. ChatLanguageModel.generateStream(prompt, responseHandler): 提供一个 ResponseHandler 接口，允许你在每个响应块到达时执行自定义逻辑。

3. StreamingChatLanguageModel 接口: 这是一个专门为流式聊天设计的接口，继承自 ChatLanguageModel，并定义了返回 Publisher<String> 的方法，直接提供文本片段流。

下面将详细介绍每种方式并提供代码示例。

不同场景下详细代码实现

为了演示方便，假设我们已经配置好了 LangChain4j 并拥有一个可用的 ChatLanguageModel 实例（例如通过 OpenAI 或其他 LLM 提供商）。

场景 1：使用 generateStream(prompt) 获取 Publisher<AiMessage>

import dev.langchain4j.model.chat.ChatLanguageModel;
import dev.langchain4j.model.chat.ChatMessage;
import dev.langchain4j.model.output.AiMessage;
import reactor.core.publisher.Flux;

import java.util.List;

public class SimpleStreamingChat {

    private final ChatLanguageModel chatLanguageModel;

    public SimpleStreamingChat(ChatLanguageModel chatLanguageModel) {
        this.chatLanguageModel = chatLanguageModel;
    }

    public void chat(String userMessage) {
        List<ChatMessage> messages = List.of(ChatMessage.user(userMessage));
        Flux<AiMessage> responsePublisher = chatLanguageModel.generateStream(messages);

        responsePublisher.subscribe(
                aiMessage -> System.out.print(aiMessage.text()), // 打印每个文本片段
                error -> System.err.println("Error during streaming: " + error),
                () -> System.out.println("\n-- Stream completed --")
        );

        // 注意：这里需要阻塞主线程，否则流可能在打印完成前结束
        try {
            Thread.sleep(5000); // 假设流在 5 秒内完成
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        // 替换为你的 ChatLanguageModel 实例
        ChatLanguageModel openAiChatModel = /* ... your OpenAI ChatLanguageModel instance ... */;
        SimpleStreamingChat chatApp = new SimpleStreamingChat(openAiChatModel);
        chatApp.chat("Tell me a short story about a robot learning to love.");
    }
}

场景 2：使用 generateStream(prompt, responseHandler) 处理每个响应块

import dev.langchain4j.model.chat.ChatLanguageModel;
import dev.langchain4j.model.chat.ChatMessage;
import dev.langchain4j.model.output.ResponseHandler;
import dev.langchain4j.model.output.StreamingResponse;

import java.util.List;

public class ChatWithResponseHandler {

    private final ChatLanguageModel chatLanguageModel;

    public ChatWithResponseHandler(ChatLanguageModel chatLanguageModel) {
        this.chatLanguageModel = chatLanguageModel;
    }

    public void chat(String userMessage) {
        List<ChatMessage> messages = List.of(ChatMessage.user(userMessage));

        chatLanguageModel.generateStream(messages, new ResponseHandler<StreamingResponse<String>>() {
            @Override
            public void onNext(StreamingResponse<String> response) {
                System.out.print(response.content()); // 处理每个文本片段
            }

            @Override
            public void onError(Throwable error) {
                System.err.println("Error during streaming: " + error);
            }

            @Override
            public void onComplete() {
                System.out.println("\n-- Stream completed --");
            }
        });

        // 注意：这里可能需要阻塞主线程，具体取决于你的应用模型
        try {
            Thread.sleep(5000); // 假设流在 5 秒内完成
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        // 替换为你的 ChatLanguageModel 实例
        ChatLanguageModel openAiChatModel = /* ... your OpenAI ChatLanguageModel instance ... */;
        ChatWithResponseHandler chatApp = new ChatWithResponseHandler(openAiChatModel);
        chatApp.chat("Explain the theory of relativity in a few sentences.");
    }
}

场景 3：使用 StreamingChatLanguageModel 获取 Publisher<String>

首先，你需要确保你使用的 LLM 集成提供了 StreamingChatLanguageModel 的实现。例如，OpenAI 集成中可能有一个 OpenAiStreamingChatModel。

import dev.langchain4j.model.chat.ChatMessage;
import dev.langchain4j.model.chat.StreamingChatLanguageModel;
import reactor.core.publisher.Flux;

import java.util.List;

public class StreamingChatModelExample {

    private final StreamingChatLanguageModel streamingChatLanguageModel;

    public StreamingChatModelExample(StreamingChatLanguageModel streamingChatLanguageModel) {
        this.streamingChatLanguageModel = streamingChatLanguageModel;
    }

    public void chat(String userMessage) {
        List<ChatMessage> messages = List.of(ChatMessage.user(userMessage));
        Flux<String> responsePublisher = streamingChatLanguageModel.generateStream(messages);

        responsePublisher.subscribe(
                System.out::print, // 直接打印每个文本片段
                error -> System.err.println("Error during streaming: " + error),
                () -> System.out.println("\n-- Stream completed --")
        );

        // 注意：这里需要阻塞主线程
        try {
            Thread.sleep(5000); // 假设流在 5 秒内完成
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        // 替换为你的 StreamingChatLanguageModel 实例
        StreamingChatLanguageModel openAiStreamingChatModel = /* ... your OpenAI StreamingChatLanguageModel instance ... */;
        StreamingChatModelExample chatApp = new StreamingChatModelExample(openAiStreamingChatModel);
        chatApp.chat("Write a short poem about a cat.");
    }
}

原理解释

1. ChatLanguageModel.generateStream(prompt):

• 这个方法返回一个 reactor.core.publisher.Flux，它是 Reactive Streams 规范在 Reactor 库中的实现，扮演着 Publisher 的角色。
• LLM 在生成响应时，会将内容分割成多个块（tokens 或文本片段），然后通过这个 Flux 依次发布出来。
• 订阅这个 Flux 的 Subscriber（在我们的例子中是 System.out::print 对应的 Lambda 表达式）会接收到每个 AiMessage 对象。
• AiMessage 对象通常包含 text() 方法，用于获取当前的文本片段。
• 整个响应完成后，Flux 会发送一个完成信号 (onComplete)。如果发生错误，会发送错误信号 (onError)。

2. ChatLanguageModel.generateStream(prompt, responseHandler):

• 这个方法不直接返回 Publisher，而是接受一个 ResponseHandler 接口的实现。
• LangChain4j 内部会处理流的订阅和块的接收。
• 对于每个接收到的 StreamingResponse<String> 对象，responseHandler 的 onNext() 方法会被调用，你可以在这里处理每个文本片段。
• StreamingResponse<String> 包装了实际的文本内容。
• responseHandler 接口还提供了 onError() 和 onComplete() 方法，用于处理流的错误和完成事件。
• 这种方式提供了更细粒度的控制，你可以在每个块到达时执行更复杂的逻辑，而不仅仅是打印。

3. StreamingChatLanguageModel 接口:

• 这个接口是为流式聊天场景专门设计的，它继承了 ChatLanguageModel 的基本功能，并定义了 generateStream 方法直接返回 Publisher<String>。
• 这简化了流式处理的流程，因为你直接接收到文本片段流，而不需要从 AiMessage 中提取文本。
• 使用这个接口的 LLM 集成通常会在底层处理好将 LLM 的原始流转换为文本片段流的逻辑。

核心特性:

• 异步和非阻塞: 流式响应是异步的，不会阻塞调用线程，允许应用在等待 LLM 响应的同时处理其他任务。
• 响应式编程: 使用 Reactive Streams (Reactor Flux) 进行流的处理，提供了丰富的操作符来转换、过滤和组合数据流。
• 事件驱动: 通过订阅流，你的代码可以对 LLM 生成的每个响应块做出反应。
• 可定制的处理: ResponseHandler 提供了高度的自定义能力来处理每个响应片段。
• 简化 API (for StreamingChatLanguageModel): 直接提供文本流，降低了使用门槛。

原理流程图以及原理解释:

由于流式响应涉及到异步数据流，传统的同步流程图不太适用。我们可以用一个更概念性的图来表示：

概念性流式响应流程:

解释:

1. 用户提供输入给聊天模型。
2. 调用 generateStream 方法启动流式响应。
3. LLM 开始处理输入，并逐步生成响应的多个片段（chunks）。
4. 这些片段通过一个异步数据流（Publisher）被依次发送出来。
5. 订阅者（通过 subscribe 或 ResponseHandler）接收到每个片段，并进行处理（例如打印或显示在 UI 上）。
6. 这个过程持续进行，直到 LLM 完成生成完整的响应。

环境准备

要在 Java 中使用 LangChain4j 实现流式聊天，你需要：

1. Java Development Kit (JDK) 11 或更高版本。
2. Maven 或 Gradle: 作为项目构建工具。
3. LangChain4j 依赖: 在你的 pom.xml (Maven) 或 build.gradle (Gradle) 文件中添加 LangChain4j 的相关依赖。这通常包括核心库以及你选择使用的 LLM 提供商的集成库（例如 langchain4j-openai).
4. LLM 提供商 API 密钥: 你需要注册一个 LLM 提供商（如 OpenAI）并获取 API 密钥，以便 LangChain4j 可以与 LLM 服务进行通信。
5. Reactor Core 依赖: LangChain4j 的流式响应基于 Reactor 库，所以你需要包含 reactor-core 依赖。

Maven 依赖示例 (pom.xml):

<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j-core</artifactId>
    <version>最新版本</version>
</dependency>
<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j-openai</artifactId>
    <version>最新版本</version>
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>最新版本</version>
</dependency>

将 <最新版本> 替换为 LangChain4j 和 Reactor 的最新稳定版本。

代码示例实现

前面的“不同场景下详细代码实现”部分已经提供了具体的代码示例。你需要将占位符的 ChatLanguageModel 或 StreamingChatLanguageModel 实例替换为你的实际配置。这通常涉及到使用你的 API 密钥和选择合适的模型。

例如，使用 OpenAI 的 OpenAiChatModel 和 OpenAiStreamingChatModel 的配置可能如下所示：

import dev.langchain4j.model.chat.ChatLanguageModel;
import dev.langchain4j.model.chat.StreamingChatLanguageModel;
import dev.langchain4j.model.openai.OpenAiChatModel;
import dev.langchain4j.model.openai.OpenAiStreamingChatModel;

public class ModelFactory {

    public static ChatLanguageModel createOpenAiChatModel(String apiKey) {
        return OpenAiChatModel.builder()
                .apiKey(apiKey)
                .modelName("gpt-3.5-turbo") // 选择合适的模型
                .build();
    }

    public static StreamingChatLanguageModel createOpenAiStreamingChatModel(String apiKey) {
        return OpenAiStreamingChatModel.builder()
                .apiKey(apiKey)
                .modelName("gpt-3.5-turbo") // 选择合适的模型
                .build();
    }
}

然后在你的主程序中使用这些工厂方法创建模型实例。

运行结果

当你运行这些示例代码时，你会看到 AI 的回复不是一次性输出，而是逐步地打印出来，每个 AiMessage 的文本内容或 StreamingResponse 的内容会随着 LLM 的生成而实时显示在控制台上。整个回复结束后，会打印出流完成的提示。

测试步骤以及详细代码

1. 配置 LangChain4j: 确保你的项目中正确引入了 LangChain4j 和 LLM 提供商的依赖，并且你已经设置好了 API 密钥。
2. 选择流式响应方式: 根据你的需求选择 generateStream(prompt), generateStream(prompt, responseHandler), 或使用 StreamingChatLanguageModel。
3. 编写测试代码: 编写一个简单的 Java程序，使用你选择的流式响应方式与 LLM 进行交互，发送一个用户消息。
4. 运行程序: 运行你的 Java 程序，观察控制台的输出。你应该看到 AI 的回复是逐步呈现的。
5. 验证流的完整性: 确保在整个回复完成后，流的 onComplete() 方法被调用。
6. 测试错误处理: 可以尝试发送一些可能导致错误的输入，观察 onError() 方法是否被正确调用。