langchain4j 响应流式支持

本页介绍如何使用底层 LLM API 进行响应流式传输。如需高层级 LLM API，请参阅 AI Services。

为什么需要流式传输？

LLM 逐 token 生成文本，因此许多 LLM 提供商提供了按 token 流式传输响应的方式，而无需等待整个文本生成完毕。这显著改善了用户体验，因为用户无需等待未知时长的处理时间，几乎可以立即开始阅读响应内容。

核心接口

ChatModel 和 LanguageModel 接口分别对应 StreamingChatModel 和 StreamingLanguageModel 接口。它们具有相似的 API支持流式传输响应。这些接口接受一个 StreamingChatResponseHandler 接口的实现作为参数。

StreamingChatResponseHandler

通过实现 StreamingChatResponseHandler，你可以为以下事件定义操作：

方法	说明
onPartialResponse(String token)	每当收到一个新的 token（或部分响应）时调用
onCompleteResponse(ChatResponse response)	当整个响应生成完毕时调用
onError(Throwable error)	当发生错误时调用

使用 StreamingChatModel 的示例

以下示例展示了如何使用 StreamingChatModel 实现流式传输：

StreamingChatModel model = OpenAiChatModel.builder()
    .apiKey("your-api-key")
    .modelName("gpt-4o-mini")
    .build();

model.chat(
    UserMessage.from("讲一个笑话"),
    new StreamingChatResponseHandler() {
        @Override
        public void onPartialResponse(String token) {
            System.out.print(token); // 逐 token 输出
        }

        @Override
        public void onCompleteResponse(ChatResponse response) {
            System.out.println("\n\n[响应完成]");
            System.out.println("Token 用量: " + response.tokenUsage());
        }

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

使用 LambdaStreamingResponseHandler

一种更简洁的流式传输方式是使用 LambdaStreamingResponseHandler 工具类。该类提供了使用 lambda 表达式创建 StreamingChatResponseHandler 的静态方法。

onPartialResponse()

使用 lambda 流式传输响应的方式非常简单，只需调用 onPartialResponse() 静态方法，传入一个定义如何处理部分响应的 lambda 表达式：

model.chat(
    UserMessage.from("讲一个笑话"),
    LambdaStreamingResponseHandler.onPartialResponse(
        token -> System.out.print(token)
    )
);

onPartialResponseAndError()

onPartialResponseAndError() 方法允许你同时为 onPartialResponse() 和 onError() 事件定义操作：

model.chat(
    UserMessage.from("讲一个笑话"),
    LambdaStreamingResponseHandler.onPartialResponseAndError(
        token -> System.out.print(token),
        error -> error.printStackTrace()
    )
);

取消流式传输

如果你希望取消流式传输，可以从以下方法中实现：

model.chat(
    UserMessage.from("写一篇长文章"),
    new StreamingChatResponseHandler() {
        @Override
        public void onPartialResponse(String token) {
            System.out.print(token);
            // 在某个条件下取消
            if (token.contains("不想继续了")) {
                // 取消逻辑见下方
            }
        }

        @Override
        public void onCompleteResponse(ChatResponse response) {
            System.out.println("\n[响应完成]");
        }

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

使用 StreamingHandle 取消

上下文对象中包含 StreamingHandle，可用于取消流式传输：

AtomicReference<StreamingHandle> handleRef = new AtomicReference<>();

model.chat(
    UserMessage.from("写一篇长文章"),
    new StreamingChatResponseHandler() {
        @Override
        public void onPartialResponse(String token) {
            System.out.print(token);
            // 获取 handle 并保存引用
            // 实际使用中可通过其他方式获取 handle
        }

        @Override
        public void onCompleteResponse(ChatResponse response) {
            System.out.println("\n[响应完成]");
        }

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

// 调用 handle.cancel() 取消流式传输
// handle.cancel();

当调用 StreamingHandle.cancel() 时，LangChain4j 将关闭连接并停止流式传输。一旦调用了 StreamingHandle.cancel()，StreamingChatResponseHandler 将不再接收任何后续回调。

在 AI Service 中使用流式传输

如果你使用的是高层级的 AI Service API，流式传输的用法更加简洁。详见 AI Services 教程中的流式响应部分。

interface MyAssistant {
    TokenStream chat(String message);
}

MyAssistant assistant = AiServices.create(MyAssistant.class, model);

assistant.chat("讲一个笑话")
    .onPartialResponse(token -> System.out.print(token))
    .onComplete(() -> System.out.println("\n[完成]"))
    .onError(error -> error.printStackTrace())
    .start();

小结

特性	说明
逐 token 输出	用户无需等待完整响应，体验更好
底层 API	StreamingChatModel / StreamingLanguageModel
回调处理	onPartialResponse、onCompleteResponse、onError
简洁写法	LambdaStreamingResponseHandler 工具类
取消机制	StreamingHandle.cancel() 关闭连接并停止回调
高层 API	AI Service 的 TokenStream 提供更优雅的流式体验