响应流式

NOTE

本页描述使用低层 LLM API 的响应流式。关于高层 LLM API，参见 AI 服务。

LLM 逐词元地生成文本，因此许多 LLM 提供者提供了一种逐词元流式传输响应的方式，而不必等待整段文本生成完毕。这极大地改善了用户体验，因为用户无需等待一段未知时长，几乎可以立即开始阅读响应。

对于 ChatModel 与 LanguageModel 接口，有对应的 StreamingChatModel 与 StreamingLanguageModel 接口。它们有相似的 API，但能流式传输响应。它们接受一个 StreamingChatResponseHandler 接口的实现作为参数。

java

public interface StreamingChatResponseHandler {

    default void onPartialResponse(String partialResponse) {}
    default void onPartialResponse(PartialResponse partialResponse, PartialResponseContext context) {}

    default void onPartialThinking(PartialThinking partialThinking) {}
    default void onPartialThinking(PartialThinking partialThinking, PartialThinkingContext context) {}

    default void onPartialToolCall(PartialToolCall partialToolCall) {}
    default void onPartialToolCall(PartialToolCall partialToolCall, PartialToolCallContext context) {}

    default void onCompleteToolCall(CompleteToolCall completeToolCall) {}

    default void onUnmappedRawEvent(Object rawEvent) {}

    void onCompleteResponse(ChatResponse completeResponse);

    void onError(Throwable error);
}

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

当下一个部分文本响应生成时：会调用 onPartialResponse(String) 或会调用 onPartialResponse(PartialResponse, PartialResponseContext)（你可以实现这两个方法中的任一个）。取决于 LLM 提供者，部分响应文本可由单个或多个词元组成。例如，你可以在词元一可用时就直接把它发送给 UI。
当下一个部分思考/推理文本生成时：会调用 onPartialThinking(PartialThinking) 或会调用 onPartialThinking(PartialThinking, PartialThinkingContext)（你可以实现这两个方法中的任一个）。取决于 LLM 提供者，部分思考文本可由单个或多个词元组成。
当下一个部分工具调用生成时：会调用 onPartialToolCall(PartialToolCall) 或会调用 onPartialToolCall(PartialToolCall, PartialToolCallContext)（你可以实现这两个方法中的任一个）。
当 LLM 完成单个工具调用的流式时：会调用 onCompleteToolCall(CompleteToolCall)。
当提供者发出一个尚未通过上述某个类型化回调暴露的原始流式事件时：会调用 onUnmappedRawEvent(Object)。见下文的未映射原始事件。
当 LLM 完成生成时：会调用 onCompleteResponse(ChatResponse)。 ChatResponse 对象既包含完整的响应（AiMessage），也包含 ChatResponseMetadata。
当发生错误时：会调用 onError(Throwable error)。

下面是一个如何用 StreamingChatModel 实现流式的示例：

java

StreamingChatModel model = OpenAiStreamingChatModel.builder()
    .apiKey(System.getenv("OPENAI_API_KEY"))
    .modelName(GPT_4_O_MINI)
    .build();

String userMessage = "Tell me a joke";

model.chat(userMessage, new StreamingChatResponseHandler() {

    @Override
    public void onPartialResponse(String partialResponse) {
        System.out.println("onPartialResponse: " + partialResponse);
    }

    @Override
    public void onPartialThinking(PartialThinking partialThinking) {
        System.out.println("onPartialThinking: " + partialThinking);
    }

    @Override
    public void onPartialToolCall(PartialToolCall partialToolCall) {
        System.out.println("onPartialToolCall: " + partialToolCall);
    }

    @Override
    public void onCompleteToolCall(CompleteToolCall completeToolCall) {
        System.out.println("onCompleteToolCall: " + completeToolCall);
    }

    @Override
    public void onCompleteResponse(ChatResponse completeResponse) {
        System.out.println("onCompleteResponse: " + completeResponse);
    }

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

一种更简洁的流式响应方式是使用 LambdaStreamingResponseHandler 类。该工具类提供静态方法，用 lambda 表达式创建 StreamingChatResponseHandler。用 lambda 流式响应的方式很简单。你只需用一个定义了如何处理部分响应的 lambda 表达式调用 onPartialResponse() 静态方法：

java

import static dev.langchain4j.model.LambdaStreamingResponseHandler.onPartialResponse;

model.chat("Tell me a joke", onPartialResponse(System.out::print));

onPartialResponseAndError() 方法允许你为 onPartialResponse() 与 onError() 两个事件都定义动作：

java

import static dev.langchain4j.model.LambdaStreamingResponseHandler.onPartialResponseAndError;

model.chat("Tell me a joke", onPartialResponseAndError(System.out::print, Throwable::printStackTrace));

响应式 API

NOTE

实验性。该 API 标注了 @Experimental，未来版本可能变化。

除上述基于处理器的 API 外，StreamingChatModel 还能把相同的响应作为一个 Flow.Publisher 事件交给你，而不阻塞调用线程：

java

Flow.Publisher<ChatModelStreamingEvent> publisher = model.chat(chatRequest);

该流会在事件到达时依次发出 PartialThinking、PartialResponse、PartialToolCall 与 CompleteToolCall 事件，并与任意 RawStreamingEvent 交错（见下文的未映射原始事件），随后是一个携带装配好的 ChatResponse 的终止性 CompleteResponse。有一个便捷重载只发出文本：

java

Flow.Publisher<String> textOnly = model.chat("Tell me a joke");

该 publisher 是冷的（cold）：在你订阅之前什么都不会发生，且每次订阅都会发送一个新请求。取消订阅会尽力中止正在进行的 HTTP 调用。

未来可能新增事件类型，因此请优雅地处理无法识别的事件，而不要写一个没有默认分支的穷举类型切换。

NOTE

事件在模型自身的线程上投递——对于 HTTP 模型，是读取响应的传输层 I/O 工作线程。不要在 onNext 中阻塞或做重活：它会拖慢整个流，并在并发下拖累所有在途调用的吞吐。请把这类工作卸载到你自己的执行器上。

LLM 响应无法有意义地被节流，因此实现会急切地消费它，并通过一个有界缓冲区中转事件，而不是把你的需求传播给模型。请宽裕地请求（例如 Long.MAX_VALUE）；请求较少的订阅者可能耗尽缓冲区并以错误终止。

关于 AI 服务层的等价方式（它也会暴露工具执行与 RAG 内容），参见非阻塞与响应式。

未映射原始事件

NOTE

这是一项面向高级用法的实验性功能。该 API 未来可能变化。

大多数应用只需要上文描述的类型化回调。然而，一些 LLM 提供者会发出 LangChain4j （尚）未映射到专用回调的额外流式事件——例如，OpenAI 服务端工具（如 web_search）的生命周期事件（response.web_search_call.in_progress、response.web_search_call.searching、response.web_search_call.completed）。

onUnmappedRawEvent(Object rawEvent) 回调让你能访问此类事件。它只为那些尚未通过某个类型化回调（onPartialResponse、onPartialThinking、onPartialToolCall、onCompleteToolCall、onCompleteResponse）暴露的事件调用。换句话说，部分响应、思考与工具调用不会作为未映射原始事件被重复，因此你可以无重复地同时消费二者。

rawEvent 的具体类型取决于提供者实现：

提供者	原始事件类型
OpenAI、Anthropic、Google AI Gemini、Mistral、Ollama	`dev.langchain4j.http.client.sse.ServerSentEvent`
OpenAI（官方）- Responses API	`com.openai.models.responses.ResponseStreamEvent`
OpenAI（官方）- Chat Completions API	`com.openai.models.chat.completions.ChatCompletionChunk`
Amazon Bedrock	`software.amazon.awssdk.services.bedrockruntime.model.ConverseStreamOutput`
Google GenAI	`com.google.genai.types.GenerateContentResponse`

由于事件类型是提供者专有的，你通常用 instanceof 检查它并转换：

java

model.chat(userMessage, new StreamingChatResponseHandler() {

    @Override
    public void onPartialResponse(String partialResponse) {
        System.out.println("onPartialResponse: " + partialResponse);
    }

    @Override
    public void onUnmappedRawEvent(Object rawEvent) {
        if (rawEvent instanceof ServerSentEvent sse) {
            System.out.println("Raw SSE event: " + sse.event() + " -> " + sse.data());
        }
    }

    @Override
    public void onCompleteResponse(ChatResponse completeResponse) {
        System.out.println("onCompleteResponse: " + completeResponse);
    }

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

使用 AI 服务时，相同的事件可通过 TokenStream.onUnmappedRawEvent(Consumer<Object>) 回调获取。

流式取消

如果你想取消流式，可以从以下方法之一进行操作：

onPartialResponse(PartialResponse, PartialResponseContext)
onPartialThinking(PartialThinking, PartialThinkingContext)
onPartialToolCall(PartialToolCall, PartialToolCallContext)

上下文对象包含一个 StreamingHandle，可用于取消流式：

java

model.chat(userMessage, new StreamingChatResponseHandler() {

    @Override
    public void onPartialResponse(PartialResponse partialResponse, PartialResponseContext context) {
        process(partialResponse);
        if (shouldCancel()) {
            context.streamingHandle().cancel();
        }
    }

    @Override
    public void onCompleteResponse(ChatResponse completeResponse) {
        System.out.println("onCompleteResponse: " + completeResponse);
    }

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

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

语言模型

嵌入存储

决策模型

响应流式

响应式 API

未映射原始事件

流式取消

友情链接