深入探索Spring AI：源碼分析流式回答
今天，我們將重點(diǎn)講解流式響應(yīng)的概念與實(shí)現(xiàn)。畢竟，AI的流式回答功能與其交互體驗(yàn)密切相關(guān)，是提升用戶滿意度的重要組成部分。
基本用法
基本用法非常簡單，只需增加一個 stream 方法即可實(shí)現(xiàn)所需功能。接下來，我們將通過代碼示例來展示這一過程，幫助您更清晰地理解如何在實(shí)際應(yīng)用中進(jìn)行操作。請看以下代碼：
@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，而是轉(zhuǎn)換為非阻塞的 StreamResponseSpec。與此同時，返回的數(shù)據(jù)類型也由之前的 String 變更為 Flux。
在深入探討其具體應(yīng)用之前，首先讓我來介紹一下 Flux 的概念與特性。
Spring WebFlux的處理器實(shí)現(xiàn)
首先，在 WebFlux 中，處理器已經(jīng)實(shí)現(xiàn)了非阻塞式的功能。這意味著，只要我們的代碼返回一個 Flux 對象，就能輕松實(shí)現(xiàn)響應(yīng)功能。通過這種方式，應(yīng)用程序能夠高效地處理并發(fā)請求，而不會因阻塞操作而影響整體性能。
@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，雖然這不是我們的重點(diǎn)，但了解其基本概念還是很有幫助的。Spring WebFlux 是 Spring 框架的一部分，專為構(gòu)建反應(yīng)式應(yīng)用而設(shè)計(jì)。它支持異步和非阻塞的編程模型，使得處理高并發(fā)請求變得更加高效。以下是 WebFlux 的幾個關(guān)鍵特性：
- 反應(yīng)式編程：WebFlux 基于反應(yīng)式編程模型，使用 Mono 和 Flux 類型來處理數(shù)據(jù)流。Mono 表示零或一個元素，而 Flux 則表示零個或多個元素。這種模型使得我們可以輕松處理異步數(shù)據(jù)流，從而提高代碼的可讀性和可維護(hù)性。
- 非阻塞 I/O：WebFlux 通過非阻塞的 I/O 操作（如 Netty 或 Servlet 3.1+ 容器）來實(shí)現(xiàn)高效的資源利用。與傳統(tǒng)的阻塞 I/O 不同，WebFlux 在等待響應(yīng)時能夠釋放線程，這樣一來，就可以顯著提高應(yīng)用的并發(fā)能力，支持更多的同時請求而不增加線程開銷。
了解這些特性將為后續(xù)的非阻塞式響應(yīng)設(shè)計(jì)奠定基礎(chǔ)，幫助我們更好地利用 WebFlux 的能力來提升應(yīng)用性能。
源碼分析
現(xiàn)在我們來詳細(xì)看看我們的 content 是如何操作的。接下來的代碼示例將展示具體的實(shí)現(xiàn)方式，幫助我們理解在 WebFlux 中如何處理數(shù)據(jù)流和響應(yīng):
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);
}這里的實(shí)現(xiàn)相對簡單，主要是傳入了一個函數(shù)。接下來，我們將深入分析 doGetFluxChatResponse 的代碼實(shí)現(xiàn)，以便更好地理解其具體邏輯和運(yùn)作方式：
private Flux<ChatResponse> doGetFluxChatResponse2(DefaultChatClientRequestSpec inputRequest) {
//此處省略重復(fù)代碼
var fluxChatResponse = this.chatModel.stream(prompt);
//此處省略重復(fù)代碼
return advisedResponse;
}這里的代碼邏輯與阻塞回答基本相同，唯一的不同之處在于它調(diào)用了 chatModel.stream(prompt) 方法。接下來，我們將深入探討 chatModel.stream(prompt) 方法的具體實(shí)現(xiàn)和其背后的設(shè)計(jì)思路：
public Flux<ChatResponse> stream(Prompt prompt) {
return Flux.deferContextual(contextView -> {
//此處省略重復(fù)代碼
Flux<OpenAiApi.ChatCompletionChunk> completionChunks = this.openAiApi.chatCompletionStream(request,
getAdditionalHttpHeaders(prompt));
//此處省略重復(fù)代碼
Flux<ChatResponse> chatResponse = completionChunks.map(this::chunkToChatCompletion)
.switchMap(chatCompletion -> Mono.just(chatCompletion).map(chatCompletion2 -> {
//此處省略重復(fù)代碼
return new ChatResponse(generations, from(chatCompletion2, null));
}
}));
//此處省略重復(fù)代碼
return new MessageAggregator().aggregate(flux, observationContext::setResponse);
});
}同樣的邏輯在這里就不再贅述，我們將重點(diǎn)關(guān)注其中的區(qū)別。在這一部分，我們使用了 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 向指定路徑發(fā)起一個 POST 請求，同時設(shè)置合適的請求頭和請求體。在獲取響應(yīng)數(shù)據(jù)時，使用了事件流的方式（通過 bodyToFlux 方法）來接收響應(yīng)內(nèi)容，并對數(shù)據(jù)進(jìn)行過濾和轉(zhuǎn)換，最終將其轉(zhuǎn)化為 ChatCompletionChunk 對象。
盡管其余的業(yè)務(wù)邏輯與之前相似，但有一點(diǎn)顯著的區(qū)別，即整個流程的返回類型以及與 OpenAI API 的調(diào)用方式都是非阻塞式的。
總結(jié)
在當(dāng)今的數(shù)字時代，流式響應(yīng)機(jī)制不僅提升了系統(tǒng)的性能，還在用戶體驗(yàn)上扮演了關(guān)鍵角色。通過引入 Flux 類型，Spring WebFlux 的設(shè)計(jì)理念使得應(yīng)用能夠以非阻塞的方式處理并發(fā)請求，從而有效利用資源并減少響應(yīng)延遲。
我們終于全面講解了Spring AI的基本操作，包括阻塞式回答、流式回答以及記憶增強(qiáng)功能。這些內(nèi)容為我們深入理解其工作機(jī)制奠定了基礎(chǔ)。接下來，我們將繼續(xù)深入探索源碼，重點(diǎn)分析回調(diào)函數(shù)、實(shí)體類映射等重要功能。
這將幫助我們更好地理解Spring AI的內(nèi)部運(yùn)作原理，并為進(jìn)一步的優(yōu)化和定制化提供指導(dǎo)。
