implementation 'org.springframework.boot:spring-boot-starter-web'
implementation 'org.springframework.boot:spring-boot-starter-webflux'

@Bean
public WebClient init() {
    return WebClient.builder()
            .baseUrl(baseUrl)
            .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + openAi)
            // ?? 必須設(shè)置為JSON格式
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .build();
}

@GetMapping(value = "/stream/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> streamChatEnhanced(@RequestParam("prompt") String prompt) {
    // 請(qǐng)求體構(gòu)建
    String requestBody = String.format("""
        {
            "model": "%s",
            "messages": [{"role": "user", "content": "%s"}],
            "stream": true
        }
        """, model, prompt);
    return webClient.post()
            // 請(qǐng)求配置
            .uri("/v1/chat/completions")
            .bodyValue(requestBody)
            .accept(MediaType.TEXT_EVENT_STREAM)
            .retrieve()
            .bodyToFlux(DataBuffer.class)  // ?? 關(guān)鍵配置點(diǎn)
            .transform(this::processStream)
            // 重試和超時(shí)配置
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
            .timeout(Duration.ofSeconds(180));
            // 錯(cuò)誤處理
            .doOnError(e -> log.error("Stream error", e))
            .doFinally(signal -> log.info("Stream completed: {}", signal));
}

private Flux<String> processStream(Flux<DataBuffer> dataBufferFlux) {
    return dataBufferFlux
            .transform(DataBufferUtils::join)          // 字節(jié)流合并
            .map(buffer -> {                          // 字節(jié)轉(zhuǎn)字符串
                String content = buffer.toString(StandardCharsets.UTF_8);
                DataBufferUtils.release(buffer);
                return content;
            })
            .flatMap(content ->                       // 處理粘包問題
                Flux.fromArray(content.split("\\r?\\n\\r?\\n")))
            .filter(event -> !event.trim().isEmpty()) // 過濾空事件
            .map(event -> {                           // 格式標(biāo)準(zhǔn)化處理
                String trimmed = event.trim();
                if (trimmed.startsWith("data:")) {
                    String substring = trimmed.substring(5);
                    return substring.startsWith(" ") ? substring.substring(1) : substring;
                }
                return trimmed;
            })
            .filter(event -> !event.startsWith("data:")); // 二次過濾
}

// 包聲明和導(dǎo)入...
@Service
@Slf4j
public class OpenAiService {
    // 配置項(xiàng)和初始化
    private String openAiApiKey = "sk-xxxxxx";
    private String baseUrl = "https://openai.com/xxxx";
    private String model = "gpt-4o";
    private WebClient webClient;
    @PostConstruct
    public void init() {
        webClient = WebClient.builder()
                .baseUrl(baseUrl)
                .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + openAiApiKey)
                .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .build();
    }
    @GetMapping(value = "/stream/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamChatEnhanced(@RequestParam("prompt") String prompt) {
        // 構(gòu)建請(qǐng)求體
        String requestBody = String.format("""
                {
                    "model": "gpt-4o-mini",
                    "messages": [{"role": "user", "content": "%s"}],
                    "stream": true
                }
                """, prompt);
        // 發(fā)送流式請(qǐng)求
        return webClient.post()
            .uri("/v1/chat/completions")
            .bodyValue(requestBody)
            .retrieve()
            .onStatus(HttpStatusCode::isError, response ->
                    response.bodyToMono(String.class)
                            .flatMap(error -> Mono.error(new RuntimeException("API Error: " + error)))
            )
            .bodyToFlux(DataBuffer.class)
            .transform(this::processStream)
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
            .timeout(Duration.ofSeconds(180))
            .doOnError(e -> log.error("Stream error", e))
            .doFinally(signal -> log.info("Stream completed: {}", signal));
    }
    private Flux<String> processStream(Flux<DataBuffer> dataBufferFlux) {
        return dataBufferFlux
                // 使用字節(jié)流處理
                .transform(DataBufferUtils::join)
                .map(buffer -> {
                    String content = buffer.toString(StandardCharsets.UTF_8);
                    DataBufferUtils.release(buffer);
                    return content;
                })
                // 按 SSE 事件邊界，防止粘包的問題
                .flatMap(content -> Flux.fromArray(content.split("\\r?\\n\\r?\\n")))
                // 過濾空事件
                .filter(event -> !event.trim().isEmpty())
                // 規(guī)范 SSE 事件格式
                .map(event -> {
                    String trimmed = event.trim();
                    // 由于webflux設(shè)置了"produces = MediaType.TEXT_EVENT_STREAM_VALUE",
                    // 所以在返回?cái)?shù)據(jù)時(shí)會(huì)自動(dòng)添加“data:”，因此如果返回的格式帶了“data:”需要手動(dòng)去除
                    if (trimmed.startsWith("data:")) {
                        trimmed = trimmed.replaceFirst("data:","").trim();
                    }
                    return trimmed;
                })
                .filter(event -> !event.startsWith("data:"));
    }
}