<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.springframework.boot</groupId>
                    <artifactId>spring-boot-starter-tomcat</artifactId>
                </exclusion>
            </exclusions>

        </dependency>
         <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>

@Data
public class AiPromptDto {
    /**
     * 大模型id
     */
    private String serviceId;
    /**
     * 用戶輸入
     */
    private String userInput;
    /**
     * sessionId
     */
    private String sessionId;
    /**
     * 請(qǐng)求id
     */
    private String requestId;
    /**
     * 獲取token
     */
    private String token;
    }

@Data
@AllArgsConstructor
public class AnswerChunk {
    /**
     * 返回的內(nèi)容
     */
    private String content;
    private String sessionId;
}

reactor:
  netty:
    response:
      buffer-size: 0

public interface AiService {
    /**
     * 根據(jù)請(qǐng)求獲取流式返回的答案
     * @param request
     * @return
     */
    Flux<AnswerChunk>  processStream(AiPromptDto request);
    }

import org.springframework.web.reactive.function.client.WebClient;
@Service
public class AIServiceImpl implements AiService {
    private final WebClient webClient;
    //初始化webClient，并ssl校驗(yàn)，生產(chǎn)環(huán)境不要跳過
    public AIServiceImpl(WebClient.Builder webClientBuilder) {
        // 使用InsecureTrustManagerFactory來信任所有證書
        SslContextBuilder sslContextBuilder = SslContextBuilder.forClient()
                .trustManager(InsecureTrustManagerFactory.INSTANCE);
        HttpClient httpClient = HttpClient.create()
                .secure(sslContextSpec -> sslContextSpec.sslContext(sslContextBuilder))
                .responseTimeout(Duration.ofMinutes(timeout));
        this.webClient = webClientBuilder.clientConnector(
                new ReactorClientHttpConnector(httpClient)
        ).baseUrl(aiForceUrl).build();
    }
  @Override
    public Flux<AnswerChunk> processStream(AiPromptDto request) {
   String body = JSONUtil.toJsonStr(request);//參數(shù)都轉(zhuǎn)化為json字符串
   return webClient.post()
                    .uri(aiForceUrl + "/aiforceplatformapi/openapi/llm/debugSse")//大模型地址
                    .bodyValue(body)//body參數(shù)
                    .header("token", request.getToken())//設(shè)置請(qǐng)求頭
                    .header("Content-Type", "application/json")
                    .retrieve()//retrieve 方法會(huì)從服務(wù)器響應(yīng)中提取數(shù)據(jù)
                    .bodyToFlux(String.class)//響應(yīng)體解析為一個(gè)流式的 String 類型序列
                    .map(chunk -> {//解析數(shù)據(jù)以供存儲(chǔ)
                        //System.out.println("chunk = " + chunk);
                        String content = "";
                        // 解析大模型返回?cái)?shù)據(jù)
                        if (!chunk.contains("[DONE]")) {//結(jié)束標(biāo)志
                            if (chunk.contains("inquiryList")) {//處理返回的關(guān)聯(lián)查詢列表
                                content = parseChunk(chunk);
                                finalAnswer[0].setQueryList(content);
                            }else if (chunk.contains("messageId")&&chunk.contains("messageContent")) {//處理返回提示message
                                parseMessage(chunk, messageMap);
                            }  else if (chunk.contains("data")) {//處理返回的問題答案
                                content = parseChunk(chunk);
                                redisTemplate.opsForValue().append(request.getRequestId() + "_result", content);
                            } else if (chunk.contains("question")) {//處理返回question
                                //先刪除
                                questionService.deleteQuestionsByPreviousIdAndRequest(questionId, requestId);
                                //保存ai返回的question
                            } else if (chunk.contains("image")) {//處理圖片
                                parseImages(chunk, imagesUrl);
                            } else if (chunk.contains("referenceInfo")) {//處理參照信息
                                parseReference(chunk, aiAnswerReferenceList);
                            }
                        } else {
                            // 處理結(jié)束
                            end.set("[DONE]")；
                            finalAnswer[0].setState("DONE");
                        }
                        if (StringUtils.isEmpty(chunk)) {
                            chunk = "";
                        }
                        return new AnswerChunk(chunk, request.getRequestId());
                    })
                    .doOnComplete(() -> {//答案都完成后存儲(chǔ)對(duì)應(yīng)數(shù)據(jù)到數(shù)據(jù)庫中
                        String finalContent = redisTemplate.opsForValue().get(request.getRequestId() + "_result");
                        redisTemplate.delete(request.getRequestId());
                        //保存答案
                        String returnAnswer = "";
                        JSONObject answer1 = new JSONObject().putOnce("data", finalContent);
                        //具體實(shí)現(xiàn)
                    })
                    .onErrorResume(e -> {//錯(cuò)誤情況處理
                        finalAnswer[0].setState("FAILED");
                        answerService.saveOrUpdate(finalAnswer[0]);
                        return Flux.error(e);
                    });
}
}

@RestController
@RequestMapping("/aiAgent")
public class AiForceController {
   /**
     * 獲取內(nèi)容
     *
     * @param request MediaType.TEXT_EVENT_STREAM_VALUE 流式輸出，否則會(huì)一次返回
     *                charset=UTF-8 字符集，不設(shè)置會(huì)亂碼
     *                注意：使用get會(huì)中文亂碼
     * @return
     */
    @PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE + ";charset=UTF-8")
    public Flux<ServerSentEvent<String>> streamResponse(@RequestBody AiForcePromptDto request) {
        return aiService.processStream(request)
                .limitRate(100) // 限制每秒最大請(qǐng)求數(shù)
                .onBackpressureBuffer(100,//背壓策略：緩沖區(qū)大小為 100
                buffer -> {
                    logger.warn("Backpressure buffer overflow, dropping {} items", buffer);
                }).publishOn(Schedulers.boundedElastic(),1) // 單線程調(diào)度確保順序
                .flatMap(chunk -> { // 使用 flatMap 將一個(gè)異步流中的每個(gè)元素映射為另一個(gè)流，并將這些流合并為一個(gè)單一的流
                    String content = chunk.getContent();
                    if (StringUtils.isNotBlank(content)) {
                        String processedContent = content.replaceAll("`{3}", "\n```"); // 規(guī)范代碼塊格式
                        return Flux.just(ServerSentEvent.<String>builder()
                                .id(request.getRequestId())
                                .data(processedContent)
                                .build());
                    }
                    return Flux.empty();//如果內(nèi)容為空，就返回空的flux
                }, 1) // 設(shè)置并發(fā)度為 1，確保逐條發(fā)送
                .doOnNext(event -> logger.info("Streaming chunk: {}", event.data())); // 日志記錄每次發(fā)送的數(shù)據(jù)
    }
}
// Flux<ServerSentEvent<String>> 實(shí)現(xiàn) SSE（Server-Sent Events），以便客戶端可以實(shí)時(shí)接收服務(wù)器推送的消息