Spring AI源码分析流式回答(最新推荐)

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

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

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

private Flux<ChatResponse> doGetFluxChatResponse2(DefaultChatClientRequestSpec inputRequest) {
//此处省略重复代码
    var fluxChatResponse = this.chatModel.stream(prompt);
//此处省略重复代码
    return advisedResponse;
}

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

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))
//此处省略一堆代码