用户在你的 AI 应用里问了一句话，然后盯着屏幕等了 3 秒，一个字都不出来——直到整个回答生成完，才"啪"一下全部弹出来。这种"憋大招"式回复体验极差，还让人以为程序卡死了。这篇文章用 LangChain4j 1.x 教你实现流式输出（Streaming）：大模型每吐出一个字，界面就立刻显示一个字，像打字机一样。首字延迟从"等完整答案"降到几百毫秒，代码完整可运行，读完就能在你项目里复现。

一、这个问题到底是什么

先说清楚"流式输出"到底解决什么痛点。

你调用大模型接口（比如 OpenAI 的 Chat Completions），默认是非流式的：你把问题发过去，服务端把整段回答一个 token 一个 token 地生成完，拼成完整字符串，再一次性返回给你。从你的角度看，就是"发请求 → 干等 → 突然收到一整段答案"。

问题就出在这个"干等"上。如果模型要生成 500 个 token 的答案，假设每个 token 花 30 毫秒，那总时长就是 15 秒。这 15 秒里你的界面是一片空白。用户会怎么想？“是不是卡了？”"是不是网络断了？"于是疯狂刷新，或者直接关掉。

流式输出（Streaming）要解决的就是这个"空白等待"问题。 它的思路是：模型每生成一个 token，服务端就立刻把这个 token 推给你；你收到一个，就显示一个。于是用户看到的是一行行字像打字机一样"蹦"出来，而不是干等 15 秒后收到一坨。

这里要引入一个关键指标：TTFT（Time to First Token，首字延迟），也就是"从发出请求，到收到第一个 token"的时间。它衡量的是"用户多久能看到第一反应"。

对比一下就清楚了：

• 非流式：用户感知延迟 ≈ 生成总时长（15 秒），期间一片空白。
• 流式：用户感知延迟 ≈ TTFT（通常几百毫秒），然后内容持续出现。

同一个模型、同一个答案，总生成时间其实没变（都是 15 秒）。但用户体验天差地别：前者像"卡死"，后者像"在认真打字"。对聊天机器人、AI 写作助手、代码补全这类产品，流式几乎是不加钱就能白拿的体验提升。

一句话总结这个问题：非流式接口让用户干等完整答案，流式接口让答案逐字出现，把感知延迟从"总时长"压缩到"首字延迟"。

二、底层原理到底怎么回事

要理解流式，得先知道大模型是怎么"说话"的。

大模型生成文字是"自回归"（autoregressive）的：一个字一个字往外蹦。 它不是一次性想好整段话，而是先预测第一个 token（可以粗略理解为一个字或一个词），然后把这个 token 拼回输入里，再预测下一个 token，如此反复，直到预测出"结束"标记为止。这个过程天然是串行、逐字的。

这就带来一个有意思的对比：

• 非流式：客户端发一个请求，服务端默默地把所有 token 都生成完，拼成完整答案，一次性返回。客户端只能干等。
• 流式：服务端每生成一个 token，就立刻通过 SSE（Server-Sent Events，服务器推送事件） 把这个 token 发出去，客户端收到一个显示一个。

SSE 你可以理解成一条"单向管道"：服务器主动、持续地把数据一段一段推给客户端，客户端不需要反复请求。它比 WebSocket 简单——只能服务器推给客户端，不能客户端反推，但对"逐字输出答案"这个场景刚刚好。

那 LangChain4j 里是怎么实现的？核心是两个接口把"非流式"和"流式"分开了：

• ChatModel 接口 = 非流式。它的 chat(...) 方法会阻塞，直到拿到完整 ChatResponse 才返回。
• StreamingChatModel 接口 = 流式。它的 chat(...) 方法不阻塞，而是接收一个回调处理器（handler），模型每吐一个 token，就回调你一次。

对应到 OpenAI 的模型类：

• OpenAiChatModel 实现了 ChatModel（非流式）。
• OpenAiStreamingChatModel 实现了 StreamingChatModel（流式）。

底层原理其实不玄乎：OpenAiStreamingChatModel 在调 OpenAI 的 /chat/completions 接口时，多带了一个参数 stream: true。OpenAI 收到这个参数后，就不会一次性返回完整答案，而是改走 SSE，一个 token 一个 token 地推。LangChain4j 在后台解析这个 SSE 流，每解析出一个 token，就调用一次你写的回调方法。

回调处理器是流式编程的核心，1.x 里它叫 StreamingChatResponseHandler，你需要实现几个方法：

回调方法	什么时候被调用	用途
onPartialResponse(String)	每吐出一个 token（或一小段）	逐字显示、推给前端
onCompleteResponse(ChatResponse)	整段回答生成完	拿完整答案、token 用量
onError(Throwable)	生成过程中出错	报错、清理

注意一个关键点：chat() 方法调用后会立刻返回，真正的流式回调发生在后台线程里。 这就是为什么在普通的 main 方法里写完 chat(...) 之后，你得想办法让主线程别急着退出（后面示例会讲怎么处理）。

还有一个容易踩的坑要提前说：LangChain4j 0.x 和 1.x 的流式 API 名字不一样。 0.x 里叫 StreamingChatLanguageModel 和 StreamingResponseHandler（方法是 onNext/onComplete/onError）；1.x 里改成了 StreamingChatModel 和 StreamingChatResponseHandler（方法是 onPartialResponse/onCompleteResponse/onError）。你在网上搜到的老教程，大概率是 0.x 的写法，直接抄会编译不过。本文全部用 1.x 的新写法。

三、实战：手把手写代码

先上依赖。新建一个 Maven 项目，pom.xml 长这样：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>langchain4j-streaming</artifactId>
    <version>1.0.0</version>

    <properties>
        <maven.compiler.source>21</maven.compiler.source>
        <maven.compiler.target>21</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <!-- LangChain4j 核心（含 AiServices、模型接口） -->
        <dependency>
            <groupId>dev.langchain4j</groupId>
            <artifactId>langchain4j</artifactId>
            <version>1.19.0</version>
        </dependency>
        <!-- OpenAI 适配器（含 OpenAiChatModel、OpenAiStreamingChatModel） -->
        <dependency>
            <groupId>dev.langchain4j</groupId>
            <artifactId>langchain4j-open-ai</artifactId>
            <version>1.19.0</version>
        </dependency>
        <!-- 日志实现，方便你看到底层请求日志 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>2.0.18</version>
        </dependency>
    </dependencies>
</project>

这段配置在干嘛：langchain4j 是框架核心（模型接口 StreamingChatModel 在这里面），langchain4j-open-ai 是 OpenAI 适配器（OpenAiStreamingChatModel 在这里面），slf4j-simple 让程序能打印运行日志方便调试。Java 版本设成 21。三个版本号都是查过 Maven Central 的最新 GA 版。你要用 DeepSeek、通义这类 OpenAI 兼容接口，改 baseUrl 就能接，后面踩坑章节会讲。

运行前准备：你需要一个 OpenAI 的 API Key（或任何 OpenAI 兼容接口的 Key），把它放到环境变量 OPENAI_API_KEY 里。代码里用 System.getenv("OPENAI_API_KEY") 读取，避免把密钥硬编码进代码。

示例 1：最小可运行的"控制台打字机"

先写一个最简单的版本，让你看清流式输出的核心三步——建模型、写回调、发起请求：

package com.example.streaming;

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

public class MinimalStreaming {

    public static void main(String[] args) throws InterruptedException {
        // 第 1 步：建一个流式模型。gpt-4o-mini 便宜、快，演示够用。
        StreamingChatModel model = OpenAiStreamingChatModel.builder()
                .apiKey(System.getenv("OPENAI_API_KEY"))
                .modelName("gpt-4o-mini")
                .build();

        // 第 2 步：写回调处理器。模型每吐一个 token，onPartialResponse 就被调一次。
        StreamingChatResponseHandler handler = new StreamingChatResponseHandler() {
            @Override
            public void onPartialResponse(String partialResponse) {
                // 关键：用 print 而不是 println，不加换行，字才能连成一句话
                System.out.print(partialResponse);
            }

            @Override
            public void onCompleteResponse(ChatResponse completeResponse) {
                System.out.println("\n--- 生成完成 ---");
            }

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

        // 第 3 步：发起流式请求。注意：这个方法调用后立刻返回，不阻塞。
        model.chat("用一句话解释什么是流式输出", handler);

        // 流式回调跑在后台线程里，main 线程得等一会儿，否则程序会提前退出
        Thread.sleep(10_000);
    }
}

这段代码要重点看两处。

第一处是 onPartialResponse 里用的是 System.out.print（不带换行）。因为模型是一个字一个字调这个方法的，你如果写成 println，每个字后面都带换行，屏幕上就会变成竖着一列字，根本没法读。print 让每个字紧挨着拼成一句完整的话。

第二处是最后的 Thread.sleep(10_000)。前面原理里说过，model.chat(...) 会立刻返回，真正的 token 回调发生在后台线程。如果 main 方法里什么都不做直接结束，JVM 一退出，后台线程还没来得及跑完，你就什么都看不到。所以这里让 main 线程"睡" 10 秒，给后台流式线程留出运行时间。这只是演示用的笨办法，示例 2 会用更规范的 CountDownLatch 来精确等待。

运行后，你会看到回答不是一次性出现，而是一个字一个字地往外冒，就像打字机。

示例 2：拿到完整答案 + token 用量 + 规范等待

真实项目里，光把字打出来还不够，你通常还要知道：最终完整答案是什么、这次花了多少 token、为什么结束的。同时也不能用 Thread.sleep 这种靠猜的等待方式。这个示例把这些一次补齐：

package com.example.streaming;

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

import java.time.Duration;
import java.util.concurrent.CountDownLatch;

public class StreamingWithDetails {

    public static void main(String[] args) throws InterruptedException {
        StreamingChatModel model = OpenAiStreamingChatModel.builder()
                .apiKey(System.getenv("OPENAI_API_KEY"))
                .modelName("gpt-4o-mini")
                .temperature(0.7)                        // 生成随机度，越低越保守
                .maxTokens(300)                          // 限制最大输出 token 数
                .timeout(Duration.ofSeconds(60))         // 整体超时
                .logRequests(true)                       // 打印底层请求日志，方便调试
                .logResponses(true)
                .build();

        // 用 StringBuilder 把逐字的 token 重新拼成完整答案
        StringBuilder fullText = new StringBuilder();
        // 用一个"闸门"让 main 线程精确等待流式结束
        CountDownLatch latch = new CountDownLatch(1);

        model.chat("用三句话介绍杭州", new StreamingChatResponseHandler() {
            @Override
            public void onPartialResponse(String partialResponse) {
                System.out.print(partialResponse);
                fullText.append(partialResponse);   // 边显示边收集
            }

            @Override
            public void onCompleteResponse(ChatResponse completeResponse) {
                System.out.println("\n========== 流式结束 ==========");
                // 结束原因：STOP 表示自然结束，LENGTH 表示撞到 maxTokens 被截断
                System.out.println("结束原因 finishReason = " + completeResponse.finishReason());
                System.out.println("模型名 modelName       = " + completeResponse.modelName());
                System.out.println("输入 token 数          = " + completeResponse.tokenUsage().inputTokenCount());
                System.out.println("输出 token 数          = " + completeResponse.tokenUsage().outputTokenCount());
                System.out.println("总 token 数            = " + completeResponse.tokenUsage().totalTokenCount());
                latch.countDown();   // 打开闸门，放行 main 线程
            }

            @Override
            public void onError(Throwable error) {
                error.printStackTrace();
                latch.countDown();   // 出错也要打开闸门，否则 main 线程永远卡住
            }
        });

        // 阻塞等待，直到流式结束（或出错）才继续往下走
        latch.await();
    }
}

这里引入两个新东西，解释一下为什么这么写。

第一个是 CountDownLatch（可以理解成一个"闸门"）。在流式回调里，onCompleteResponse 和 onError 里都调用了 latch.countDown()，把闸门打开；main 线程调用 latch.await() 就会一直阻塞，直到闸门被打开。这样 main 线程就能精确等流式结束，而不是靠 Thread.sleep 猜时间——答案生成 5 秒就等 5 秒，生成 30 秒就等 30 秒。注意 onError 里也必须 countDown()，否则一旦出错，闸门永远关着，程序就死等在那里。

第二个是 ChatResponse 里的元数据。completeResponse.finishReason() 告诉你为什么结束：STOP 是自然生成完，LENGTH 是被 maxTokens 截断了——这个在生产里很有用，比如你发现大量回答都是 LENGTH，就说明 maxTokens 设小了，答案老被腰斩。tokenUsage() 则告诉你这次花了多少钱（按 token 计费时尤其重要）。

示例 3：把流式输出推给前端（SSE 接口）

前面两个示例都是打印到控制台。真实场景是给浏览器用。这个示例用 JDK 自带的 HttpServer 起一个 SSE 接口，浏览器（或 curl）访问时，就能实时看到逐字输出：

package com.example.streaming;

import com.sun.net.httpserver.HttpServer;
import dev.langchain4j.model.chat.StreamingChatModel;
import dev.langchain4j.model.chat.response.ChatResponse;
import dev.langchain4j.model.chat.response.StreamingChatResponseHandler;
import dev.langchain4j.model.openai.OpenAiStreamingChatModel;

import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.URLDecoder;
import java.nio.charset.StandardCharsets;

public class SseStreamingServer {

    public static void main(String[] args) throws IOException {
        StreamingChatModel model = OpenAiStreamingChatModel.builder()
                .apiKey(System.getenv("OPENAI_API_KEY"))
                .modelName("gpt-4o-mini")
                .build();

        HttpServer server = HttpServer.create(new InetSocketAddress(8080), 0);

        server.createContext("/chat", exchange -> {
            // 从 URL 参数 q 里取问题，做 URL 解码，支持中文
            String question = "用一句话介绍流式输出";
            String rawQuery = exchange.getRequestURI().getRawQuery();
            if (rawQuery != null && rawQuery.startsWith("q=")) {
                question = URLDecoder.decode(rawQuery.substring(2), StandardCharsets.UTF_8);
            }

            // 设置 SSE 响应头：text/event-stream 是关键
            exchange.getResponseHeaders().set("Content-Type", "text/event-stream; charset=utf-8");
            exchange.getResponseHeaders().set("Cache-Control", "no-cache");
            exchange.getResponseHeaders().set("Connection", "keep-alive");
            exchange.sendResponseHeaders(200, 0); // 0 表示 chunked 分块传输

            OutputStream out = exchange.getResponseBody();

            model.chat(question, new StreamingChatResponseHandler() {
                @Override
                public void onPartialResponse(String partialResponse) {
                    // 每个 token 包装成一条 SSE 消息发给浏览器
                    String sse = "data: " + partialResponse.replace("\n", "\\n") + "\n\n";
                    try {
                        out.write(sse.getBytes(StandardCharsets.UTF_8));
                        out.flush();   // 必须 flush，否则浏览器收不到实时数据
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }

                @Override
                public void onCompleteResponse(ChatResponse completeResponse) {
                    try {
                        out.write("data: [DONE]\n\n".getBytes(StandardCharsets.UTF_8));
                        out.flush();
                        out.close();
                        exchange.close();
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }

                @Override
                public void onError(Throwable error) {
                    try {
                        out.write(("data: [ERROR] " + error.getMessage() + "\n\n")
                                .getBytes(StandardCharsets.UTF_8));
                        out.flush();
                        out.close();
                        exchange.close();
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                }
            });
        });

        server.start();
        System.out.println("SSE 服务已启动：http://localhost:8080/chat?q=你的问题");
    }
}

这段代码的关键在两个地方。

第一是响应头里的 Content-Type: text/event-stream。这是 SSE 协议的规定，浏览器一看这个头就知道"这是一条服务器推送流，别当成普通网页"。SSE 的数据格式是 data: 内容\n\n，两条消息之间用空行分隔，所以每个 token 要拼成 "data: " + 文字 + "\n\n" 再写出去。

第二是每次 out.write 之后必须 out.flush()。因为 HttpServer 的 OutputStream 有缓冲，你不 flush，数据就憋在缓冲区里，浏览器那边迟迟收不到，流式效果就没了。这是 SSE 编程里最常见的坑。

启动后，在终端里执行：

curl -N "http://localhost:8080/chat?q=%E7%94%A8%E4%B8%80%E5%8F%A5%E8%AF%9D%E4%BB%8B%E7%BB%8D%E6%B5%81%E5%BC%8F%E8%BE%93%E5%87%BA"

（%E7... 是"用一句话介绍流式输出"的 URL 编码，-N 参数让 curl 不做缓冲、实时打印。）你会看到回答一个字一个字地蹦出来，最后以 data: [DONE] 结束。把这个 SSE 接口接上前端的 EventSource，就是一个能"打字机式"回复的 AI 应用了。

四、踩坑经验和最佳实践

这一节全是实战里真会踩的坑，提前避开能省不少时间。

坑 1：版本 API 名字变了，老教程会坑你。 LangChain4j 0.x 里流式接口叫 StreamingChatLanguageModel，回调叫 StreamingResponseHandler（方法 onNext/onComplete/onError）；1.x 改成了 StreamingChatModel 和 StreamingChatResponseHandler（方法 onPartialResponse/onCompleteResponse/onError）。你搜到的老文章多半是 0.x 写法，直接抄会编译不过。认准本文用的 1.19.0 版本和 1.x 类名：OpenAiStreamingChatModel、StreamingChatModel、StreamingChatResponseHandler。

坑 2：chat() 是非阻塞的，main 线程会提前退出。 前面反复强调，model.chat(...) 调用后立刻返回，token 回调发生在后台线程。如果你在 main 里写完 chat 就直接结束，程序秒退，什么都看不到。规范做法是用 CountDownLatch（如示例 2），或部署在 Web 容器里（Spring Boot 这类框架本身不会退出）。

坑 3：onPartialResponse 里千万别加换行。 每个 token 都会触发一次回调，你如果在这里 println，每个字一行，答案会竖着排。要用 print 或直接 append 到 StringBuilder，让字连成句。

坑 4：onError 里忘了释放等待。 如果你用 CountDownLatch 等结束，onCompleteResponse 里 countDown() 了，但 onError 里没 countDown()，一旦请求出错，main 线程就永远卡死。两个分支都要释放。

坑 5：SSE 输出不 flush 就收不到实时数据。 示例 3 里讲过，OutputStream 有缓冲，不 flush 浏览器就看不到逐字效果，只有等连接关闭才"一次性"收到，流式就白做了。

坑 6：API Key 别硬编码进代码。 用 System.getenv("OPENAI_API_KEY") 读环境变量，或放到配置文件/密钥管理里。硬编码进代码，一旦提交到 Git，密钥就泄露了。

最佳实践（几条建议）：

• 接 OpenAI 兼容接口：DeepSeek、通义、Kimi 等都兼容 OpenAI 协议，只要把 baseUrl 改成对应地址、modelName 改成对方模型名即可，代码其余部分不用动。
• 必设 maxTokens 和 timeout：防止模型跑飞无限烧钱，也防止请求无限挂起。
• 关注 finishReason：如果大量回答都是 LENGTH，说明 maxTokens 设小了，答案老被截断，需要调大。
• 统计 token 用量：onCompleteResponse 里的 tokenUsage() 是算成本的关键，接入计费系统时一定要用。

五、性能对比和技术选型

先说结论：流式输出不改变总生成时间，改变的是"用户感知到的等待时间"。

以"生成 500 个 token 的回答、每个 token 30 毫秒"为例，做个直观对比：

指标	非流式	流式
总生成时间	15 秒	15 秒（不变）
用户看到第一个字的时间（TTFT）	15 秒	约 0.3～0.5 秒
用户感知	“卡死了”	“在打字”
后端实现	简单	稍复杂（回调 + SSE）

可以看到，流式的收益是纯"体验层面"的，它把"干等 15 秒"变成了"0.3 秒就有反应"。对聊天、写作、代码补全这类"边生成边看"的场景，这个收益是决定性的。

但流式不是万能的，有些场景该用非流式：

• 需要严格结构化输出（比如让模型返回一个 JSON 给程序解析）：非流式拿完整结果一次解析更省心，流式还得自己在回调里拼完整 JSON 再解析。
• 批量、离线处理（比如夜里跑一批摘要任务）：没人盯着看，流式徒增复杂度，用非流式批量跑更划算。

技术选型建议：面向用户的对话/生成类产品，直接上流式；后台批处理、结构化提取类任务，用非流式。LangChain4j 两个接口（StreamingChatModel 和 ChatModel）都给你备好了，按场景选即可，不必二选一。

六、总结

这篇文章解决的是"AI 应用回复慢、用户干等"的问题。核心就一句话：大模型本来就是逐字生成的，非流式接口偏要攒齐了才给你，流式接口则每吐一个字就推给你一次。

实现上记住三步：用 OpenAiStreamingChatModel 建一个流式模型；写一个 StreamingChatResponseHandler 回调，在 onPartialResponse 里逐字显示；调 chat() 发起请求，并用 CountDownLatch（或 Web 框架）管住线程生命周期。要落地到浏览器，就用 SSE（text/event-stream + 每条 data: ...\n\n），记得每次 flush。

踩坑清单再复习一遍：认准 1.x 的新类名（别抄 0.x 老教程）、chat() 非阻塞要管住 main 线程、回调里别乱加换行、onError 别忘释放等待、SSE 必须 flush、密钥别硬编码。

流式输出的价值不在于"更快生成"，而在于"让用户感觉更快"。把这个体验优化做上，你的 AI 应用离"好用"就近了一大步。