快捷導(dǎo)航

Spring AI源碼分析流式回答(最新推薦)

更新時間：2024年11月14日 11:53:36 作者：努力的小雨

本文我們將重點講解流式響應(yīng)的概念與實現(xiàn),畢竟,AI的流式回答功能與其交互體驗密切相關(guān),是提升用戶滿意度的重要組成部分,我們將通過代碼示例來展示這一過程,幫助您更清晰地理解如何在實際應(yīng)用中進行操作,感興趣的朋友一起看看吧

在上一章節(jié)中，我們深入分析了Spring AI的阻塞式請求與響應(yīng)機制，并探討了如何增強其記憶能力。今天，我們將重點講解流式響應(yīng)的概念與實現(xiàn)。畢竟，AI的流式回答功能與其交互體驗密切相關(guān)，是提升用戶滿意度的重要組成部分。

基本用法

基本用法非常簡單，只需增加一個 stream 方法即可實現(xiàn)所需功能。接下來，我們將通過代碼示例來展示這一過程，幫助您更清晰地理解如何在實際應(yīng)用中進行操作。請看以下代碼：

@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的處理器實現(xiàn)

首先，在 WebFlux 中，處理器已經(jīng)實現(xiàn)了非阻塞式的功能。這意味著，只要我們的代碼返回一個 Flux 對象，就能輕松實現(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，雖然這不是我們的重點，但了解其基本概念還是很有幫助的。Spring WebFlux 是 Spring 框架的一部分，專為構(gòu)建反應(yīng)式應(yīng)用而設(shè)計。它支持異步和非阻塞的編程模型，使得處理高并發(fā)請求變得更加高效。以下是 WebFlux 的幾個關(guān)鍵特性：

反應(yīng)式編程：WebFlux 基于反應(yīng)式編程模型，使用 Mono 和 Flux 類型來處理數(shù)據(jù)流。Mono 表示零或一個元素，而 Flux 則表示零個或多個元素。這種模型使得我們可以輕松處理異步數(shù)據(jù)流，從而提高代碼的可讀性和可維護性。
非阻塞 I/O：WebFlux 通過非阻塞的 I/O 操作（如 Netty 或 Servlet 3.1+ 容器）來實現(xiàn)高效的資源利用。與傳統(tǒng)的阻塞 I/O 不同，WebFlux 在等待響應(yīng)時能夠釋放線程，這樣一來，就可以顯著提高應(yīng)用的并發(fā)能力，支持更多的同時請求而不增加線程開銷。

了解這些特性將為后續(xù)的非阻塞式響應(yīng)設(shè)計奠定基礎(chǔ)，幫助我們更好地利用 WebFlux 的能力來提升應(yīng)用性能。

源碼分析

現(xiàn)在我們來詳細(xì)看看我們的 content 是如何操作的。接下來的代碼示例將展示具體的實現(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);
}

這里的實現(xiàn)相對簡單，主要是傳入了一個函數(shù)。接下來，我們將深入分析 doGetFluxChatResponse 的代碼實現(xià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) 方法的具體實現(xiàn)和其背后的設(shè)計思路：

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);
        });
    }

同樣的邏輯在這里就不再贅述，我們將重點關(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ù)進行過濾和轉(zhuǎn)換，最終將其轉(zhuǎn)化為 ChatCompletionChunk 對象。

盡管其余的業(yè)務(wù)邏輯與之前相似，但有一點顯著的區(qū)別，即整個流程的返回類型以及與 OpenAI API 的調(diào)用方式都是非阻塞式的。

總結(jié)

在當(dāng)今的數(shù)字時代，流式響應(yīng)機制不僅提升了系統(tǒng)的性能，還在用戶體驗上扮演了關(guān)鍵角色。通過引入 Flux 類型，Spring WebFlux 的設(shè)計理念使得應(yīng)用能夠以非阻塞的方式處理并發(fā)請求，從而有效利用資源并減少響應(yīng)延遲。

我們終于全面講解了Spring AI的基本操作，包括阻塞式回答、流式回答以及記憶增強功能。這些內(nèi)容為我們深入理解其工作機制奠定了基礎(chǔ)。接下來，我們將繼續(xù)深入探索源碼，重點分析回調(diào)函數(shù)、實體類映射等重要功能。

這將幫助我們更好地理解Spring AI的內(nèi)部運作原理，并為進一步的優(yōu)化和定制化提供指導(dǎo)。

到此這篇關(guān)于深入探索Spring AI：源碼分析流式回答的文章就介紹到這了,更多相關(guān)Spring AI流式回答內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: