流式输出

约 665 字大约 2 分钟

2025-12-14

基础知识

langchain4j-reactor

langchain4j-reactor 是 LangChain4j 生态系统中一个专门为 响应式编程 (Reactive Programming) 提供的适配器模块。简单来说，它是连接 LangChain4j（基于回调机制）与 Spring WebFlux（基于 Project Reactor 机制）的“官方桥梁”。

核心作用：消除样板代码，让你能够直接使用 Flux<String> 或 TokenStream 等响应式类型，而不需要每次都手动写 Sinks 转换逻辑
原理：langchain4j-reactor 内部其实就是封装了 Sinks 模式。
- 当我们在 AiServices 中调用返回 Flux 的方法时，该模块的拦截器会捕获调用。
- 它内部创建一个 Sinks.Many<String>，并注册一个 StreamingResponseHandler 到底层的 StreamingChatLanguageModel，它将回调中的 onNext 桥接到 Sink 的 tryEmitNext。
- 最后将 Sink 转换为 Flux 返回给你。

Sinks 模式

在 Java 的响应式编程中，Sinks 是连接“命令式编程（Imperative）”与“响应式编程（Reactive）”的桥梁。类比一个**“漏斗”**：我们手动把数据（像水一样）倒入漏斗，漏斗的另一端就会自动流出标准的水流（Flux 或 Mono），供下游系统消费。

Sinks 包含两个视角：

输入端 (Sink)： 允许你的代码手动触发 next (数据), error (异常), complete (结束)。
输出端 (Flux/Mono)： 将输入端的数据转换为标准的响应式流，供下游订阅。

Sinks 和 Flux<> 配合

过程如下：

建立连接：我们需要一个“汇流排”（Sink），它的作用是一端接收 AI 的回调数据，另一端把数据变成 Flux 流吐出去
数据转换：当大模型（如 GPT-4）生成 Token 时，通过 Reactor 的机制将其推入管道。
前端消费：Spring Boot Controller 将 Flux<String> 以 text/event-stream (SSE) 的格式返回给浏览器。

动手实操

引入依赖

<!-- 流式输出 -->
<dependency>
    <groupId>dev.langchain4j</groupId>
    <artifactId>langchain4j-reactor</artifactId>
    <version>${langchain4j.version}</version>
</dependency>

yml 配置

server:
  servlet:
    # 设置响应的字符编码，避免流式返回输出乱码
    encoding:
      charset: utf-8
      enabled: true
      force: true
      
langchain4j:
  open-ai:
    # 流式模型
    streaming-chat-model:
      model-name: # 模型名称
      api-key: # 模型 key
      base-url: # 模型 url

业务层

@AiService(wiringMode = EXPLICIT,
        chatModel = "openAiChatModel",
        streamingChatModel = "openAiStreamingChatModel",
        contentRetriever = "contentRetriever"
)
public interface OpenAiService {

    String chat(String message);

    @SystemMessage("你是一名智能客服")
    String chatWithRag(String message);

    @SystemMessage("你是一名AI购物助手，根据用户的提问帮助用户搜索相关的商品信息")
    Flux<String> chatWhitStream(String message);
}

controller 层

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

    @Autowired
    private OpenAiService openAiService;
    
    @RequestMapping(value="/stream", produces = TEXT_EVENT_STREAM_VALUE)  // 响应数据以data开头
    public Flux<String> chatOllamSearch(@RequestParam("message") String message) {
        return openAiService.chatWhitStream(message);
    }

}

调用接口

http://localhost:8080/api/chat/stream?message=购买手机，推荐哪个平台

效果：

版权所有

版权归属：haipeng-lin

许可证：署名 4.0 国际 (CC-BY-4.0)