SpringBoot 2.x 接入非标准SSE格式大模型流式响应的实战解决方案

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)
            // ⚠️ 必须设置为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) {
    // 请求体构建
    String requestBody = String.format("""
        {
            "model": "%s",
            "messages": [{"role": "user", "content": "%s"}],
            "stream": true
        }
        """, model, prompt);
    return webClient.post()
            // 请求配置
            .uri("/v1/chat/completions")
            .bodyValue(requestBody)
            .accept(MediaType.TEXT_EVENT_STREAM)
            .retrieve()
            .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
            .transform(DataBufferUtils::join)          // 字节流合并
            .map(buffer -> {                          // 字节转字符串
                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 -> {                           // 格式标准化处理
                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:")); // 二次过滤
}

// 包声明和导入...
@Service
@Slf4j
public class OpenAiService {
    // 配置项和初始化
    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) {
        // 构建请求体
        String requestBody = String.format("""
                {
                    "model": "gpt-4o-mini",
                    "messages": [{"role": "user", "content": "%s"}],
                    "stream": true
                }
                """, prompt);
        // 发送流式请求
        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
                // 使用字节流处理
                .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())
                // 规范 SSE 事件格式
                .map(event -> {
                    String trimmed = event.trim();
                    // 由于webflux设置了"produces = MediaType.TEXT_EVENT_STREAM_VALUE",
                    // 所以在返回数据时会自动添加“data:”，因此如果返回的格式带了“data:”需要手动去除
                    if (trimmed.startsWith("data:")) {
                        trimmed = trimmed.replaceFirst("data:","").trim();
                    }
                    return trimmed;
                })
                .filter(event -> !event.startsWith("data:"));
    }
}