✅Spring AI 核心特性：流式输出
前面讲过不用Sping AI，在Java中做流式输出的方案，主要是基于SSE，之前讲Spring AI的例子的时候，其实也提到过。
比如ChatModel中通过stream方法可以做流式输出，是因为他实现了StreamingChatModel接口中。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| @RestController
@RequestMapping("/ai/output")
public class StreamOutputController implements InitializingBean {
@Autowired
private DashScopeChatModel chatModel;
private ChatClient chatClient;
@RequestMapping("/stream/chat")
public Flux<String> streamChat(
@RequestParam(value = "message") String message, HttpServletResponse response) {
response.setContentType("text/event-stream");
response.setCharacterEncoding("UTF-8");
Prompt prompt = new Prompt(message, DashScopeChatOptions.builder().withModel("qwen-plus").build());
Flux<ChatResponse> chatResponseFlux = chatModel.stream(prompt);
return chatResponseFlux.map(resp -> resp.getResult().getOutput().getText());
}
}
|
这是一个controller中的方法，返回值是Flux类型的，返回给前端的就是个流式的内容输出。
在ChatClient中也一样的，也有一个stream方法，返回值也是一个Flux类型。
1
2
3
4
5
6
7
| @RequestMapping("/stream/chat")
public Flux<String> streamChat(
@RequestParam(value = "message") String message, HttpServletResponse response) {
response.setContentType("text/event-stream");
response.setCharacterEncoding("UTF-8");
return chatClient.prompt(message).stream().content();
}
|
通过源码可以看到，不管是ChatModel还是ChatClient，最终都是依赖reactor.core.publisher.Flux#deferContextual来实现的。
这里面用到了一个Flux，他是啥呢？
实现原理
大模型的原理的本质就是预测下一个词做输出的，所以，我们就可以不用等他全部都预测完再输出，可以他预测一个我们返回一个。所以，这就是流式输出。
在Java中，有一个响应式编程的库——Reactor，他出现很多年了，一直都不温不火，主要是因为代码写起来太抽象了，但是确实有高性能、异步、非阻塞等优势。尤其在JDK 21之后虚拟线程出来之后，更一度要退出历史舞台了。但是突然大模型火了，而Reactor提供的支持流式输出天然契合大模型的输出。
响应式编程是一种典型的观察者模式，当有新的可用的数据到来时，Publisher 会对Subscriber进行通知，这种推动是响应式的关键。
Reactor项目的主要组件为 reactor-core，一个专注于响应式流规范并基于Java8的响应式库。Reactor引入了可组合的响应式类型，这些类型既实现了 Publisher 又提供了丰富的操作符：Flux 和 Mono。
Flux 是 Reactor 库中的一个发布者（Publisher），遵循 Reactive Streams 规范。
它代表一个异步的、非阻塞的序列，可以发射：
0 到多个数据项（onNext）
一个可选的错误（onError）
或一个完成信号（onComplete）
想要定义一个Flux比较简单：
1
2
3
4
5
| Flux<String> seq1 = Flux.just("foo", "bar", "foobar");
List<String> iterable = Arrays.asList("foo", "bar", "foobar");
Flux<String> seq2 = Flux.fromIterable(iterable);
|
Flux的订阅也比较简单，主要用他提供的subscribe()方法。
1
2
3
4
5
6
7
8
9
10
11
|
seq1.subscribe(i -> System.out.println(i));
seq1.subscribe(i -> System.out.println(i),
error -> System.err.println("Error: " + error));
seq1.subscribe(i -> System.out.println(i),
error -> System.err.println("Error " + error),
() -> System.out.println("Done"));
|
更多关于Reactor的介绍：https://easywheelsoft.github.io/reactor-core-zh/index.html#about-doc
