深入探索Spring AI：源碼分析流式回答
今天，我們將重點講解流式響應的概念與實現。畢竟，AI的流式回答功能與其交互體驗密切相關，是提升用戶滿意度的重要組成部分。
基本用法
基本用法非常簡單，只需增加一個 stream 方法即可實現所需功能。接下來，我們將通過代碼示例來展示這一過程，幫助您更清晰地理解如何在實際應用中進行操作。請看以下代碼：
@GetMapping(value = "/ai-stream",produces = MediaType.APPLICATION_OCTET_STREAM_VALUE + ";charset=UTF-8")
Flux<String> generationByStream(@RequestParam("userInput") String userInput) {
Flux<String> output = chatClient.prompt()
.user(userInput)
.stream()
.content();
return output;
}在我們增加 stream 方法之后，返回的對象類型將不再是原來的阻塞式 CallResponseSpec，而是轉換為非阻塞的 StreamResponseSpec。與此同時，返回的數據類型也由之前的 String 變更為 Flux。
在深入探討其具體應用之前，首先讓我來介紹一下 Flux 的概念與特性。
Spring WebFlux的處理器實現
首先，在 WebFlux 中，處理器已經實現了非阻塞式的功能。這意味著，只要我們的代碼返回一個 Flux 對象，就能輕松實現響應功能。通過這種方式，應用程序能夠高效地處理并發請求，而不會因阻塞操作而影響整體性能。
@Override
public Mono<Void> handle(ServerWebExchange exchange) {
if (this.handlerMappings == null) {
return createNotFoundError();
}
if (CorsUtils.isPreFlightRequest(exchange.getRequest())) {
return handlePreFlight(exchange);
}
return Flux.fromIterable(this.handlerMappings)
.concatMap(mapping -> mapping.getHandler(exchange))
.next()
.switchIfEmpty(createNotFoundError())
.onErrorResume(ex -> handleResultMono(exchange, Mono.error(ex)))
.flatMap(handler -> handleRequestWith(exchange, handler));
}這里簡單介紹一下 Spring WebFlux，雖然這不是我們的重點，但了解其基本概念還是很有幫助的。Spring WebFlux 是 Spring 框架的一部分，專為構建反應式應用而設計。它支持異步和非阻塞的編程模型，使得處理高并發請求變得更加高效。以下是 WebFlux 的幾個關鍵特性：
- 反應式編程：WebFlux 基于反應式編程模型，使用 Mono 和 Flux 類型來處理數據流。Mono 表示零或一個元素，而 Flux 則表示零個或多個元素。這種模型使得我們可以輕松處理異步數據流，從而提高代碼的可讀性和可維護性。
- 非阻塞 I/O：WebFlux 通過非阻塞的 I/O 操作（如 Netty 或 Servlet 3.1+ 容器）來實現高效的資源利用。與傳統的阻塞 I/O 不同，WebFlux 在等待響應時能夠釋放線程，這樣一來，就可以顯著提高應用的并發能力，支持更多的同時請求而不增加線程開銷。
了解這些特性將為后續的非阻塞式響應設計奠定基礎，幫助我們更好地利用 WebFlux 的能力來提升應用性能。
源碼分析
現在我們來詳細看看我們的 content 是如何操作的。接下來的代碼示例將展示具體的實現方式，幫助我們理解在 WebFlux 中如何處理數據流和響應:
public Flux<String> content() {
return doGetFluxChatResponse(this.request).map(r -> {
if (r.getResult() == null || r.getResult().getOutput() == null
|| r.getResult().getOutput().getContent() == null) {
return "";
}
return r.getResult().getOutput().getContent();
}).filter(StringUtils::hasLength);
}這里的實現相對簡單，主要是傳入了一個函數。接下來，我們將深入分析 doGetFluxChatResponse 的代碼實現，以便更好地理解其具體邏輯和運作方式：
private Flux<ChatResponse> doGetFluxChatResponse2(DefaultChatClientRequestSpec inputRequest) {
//此處省略重復代碼
var fluxChatResponse = this.chatModel.stream(prompt);
//此處省略重復代碼
return advisedResponse;
}這里的代碼邏輯與阻塞回答基本相同，唯一的不同之處在于它調用了 chatModel.stream(prompt) 方法。接下來，我們將深入探討 chatModel.stream(prompt) 方法的具體實現和其背后的設計思路：
public Flux<ChatResponse> stream(Prompt prompt) {
return Flux.deferContextual(contextView -> {
//此處省略重復代碼
Flux<OpenAiApi.ChatCompletionChunk> completionChunks = this.openAiApi.chatCompletionStream(request,
getAdditionalHttpHeaders(prompt));
//此處省略重復代碼
Flux<ChatResponse> chatResponse = completionChunks.map(this::chunkToChatCompletion)
.switchMap(chatCompletion -> Mono.just(chatCompletion).map(chatCompletion2 -> {
//此處省略重復代碼
return new ChatResponse(generations, from(chatCompletion2, null));
}
}));
//此處省略重復代碼
return new MessageAggregator().aggregate(flux, observationContext::setResponse);
});
}同樣的邏輯在這里就不再贅述，我們將重點關注其中的區別。在這一部分，我們使用了 chatCompletionStream，而且與之前不同的是，這里不再使用 retryTemplate，而是引入了 webClient，這是一個能夠接收事件流的工具類。
public Flux<ChatCompletionChunk> chatCompletionStream(ChatCompletionRequest chatRequest,
MultiValueMap<String, String> additionalHttpHeader) {
Assert.notNull(chatRequest, "The request body can not be null.");
Assert.isTrue(chatRequest.stream(), "Request must set the stream property to true.");
AtomicBoolean isInsideTool = new AtomicBoolean(false);
return this.webClient.post()
.uri(this.completionsPath)
.headers(headers -> headers.addAll(additionalHttpHeader))
.body(Mono.just(chatRequest), ChatCompletionRequest.class)
.retrieve()
.bodyToFlux(String.class)
// cancels the flux stream after the "[DONE]" is received.
.takeUntil(SSE_DONE_PREDICATE)
// filters out the "[DONE]" message.
.filter(SSE_DONE_PREDICATE.negate())
.map(content -> ModelOptionsUtils.jsonToObject(content, ChatCompletionChunk.class))
//此處省略一堆代碼這段代碼的主要目的是通過 webClient 向指定路徑發起一個 POST 請求，同時設置合適的請求頭和請求體。在獲取響應數據時，使用了事件流的方式（通過 bodyToFlux 方法）來接收響應內容，并對數據進行過濾和轉換，最終將其轉化為 ChatCompletionChunk 對象。
盡管其余的業務邏輯與之前相似，但有一點顯著的區別，即整個流程的返回類型以及與 OpenAI API 的調用方式都是非阻塞式的。
總結
在當今的數字時代，流式響應機制不僅提升了系統的性能，還在用戶體驗上扮演了關鍵角色。通過引入 Flux 類型，Spring WebFlux 的設計理念使得應用能夠以非阻塞的方式處理并發請求，從而有效利用資源并減少響應延遲。
我們終于全面講解了Spring AI的基本操作，包括阻塞式回答、流式回答以及記憶增強功能。這些內容為我們深入理解其工作機制奠定了基礎。接下來，我們將繼續深入探索源碼，重點分析回調函數、實體類映射等重要功能。
這將幫助我們更好地理解Spring AI的內部運作原理，并為進一步的優化和定制化提供指導。
