<!-- Spring WebFlux 核心依賴 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<!-- Netty 依賴（WebFlux默認(rèn)集成，可顯式引入確保版本一致） -->
<dependency>
    <groupId>io.projectreactor.netty</groupId>
    <artifactId>reactor-netty</artifactId>
</dependency>
<!-- WebFlux 內(nèi)置HTTP客戶端（替代RestTemplate，用于調(diào)用DeepSeek API） -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux-client</artifactId>
</dependency>
<!-- JSON解析依賴（用于解析DeepSeek的響應(yīng)數(shù)據(jù)） -->
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>
<!-- 日志依賴（可選，用于調(diào)試流式數(shù)據(jù)） -->
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
</dependency>
<dependency>
    <groupId>ch.qos.logback</groupId>
    <artifactId>logback-classic</artifactId>
</dependency>

# DeepSeek模型服務(wù)配置（本地部署版）
model:
  name: deepseek-7b  # 模型名稱，根據(jù)實(shí)際部署的模型填寫
  stream: true        # 關(guān)鍵參數(shù)：?jiǎn)⒂昧魇巾憫?yīng)，必須設(shè)為true
  max_tokens: 2048    # 最大生成token數(shù)，根據(jù)業(yè)務(wù)需求調(diào)整
  temperature: 0.7    # 溫度參數(shù)，控制生成內(nèi)容的隨機(jī)性（0-1之間）
  top_p: 0.9          # 可選參數(shù)，控制采樣范圍
  api_key: your_api_key  # 本地部署可忽略，調(diào)用官方API需填寫

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.netty.http.client.HttpClient;

import java.time.Duration;

@Configuration
public class WebClientConfig {

    // 從配置文件中讀取DeepSeek API地址和API密鑰（推薦）
    private final String deepSeekBaseUrl = "https://api.deepseek.com/v1";
    private final String deepSeekApiKey = "your_deepseek_api_key"; // 替換為自己的API密鑰

    @Bean
    public WebClient deepSeekClient() {
        return WebClient.builder()
                .baseUrl(deepSeekBaseUrl) // DeepSeek API基礎(chǔ)地址
                .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .defaultHeader("Authorization", "Bearer " + deepSeekApiKey) // 官方API需攜帶密鑰
                .clientConnector(new ReactorClientHttpConnector(
                        // 配置HTTP客戶端，設(shè)置響應(yīng)超時(shí)時(shí)間（大模型推理耗時(shí)較長(zhǎng)，需適當(dāng)延長(zhǎng)）
                        HttpClient.create().responseTimeout(Duration.ofMinutes(5))
                ))
                .build();
    }
}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.io.IOException;

@Service
public class DeepSeekStreamService {

    private static final Logger log = LoggerFactory.getLogger(DeepSeekStreamService.class);

    private final WebClient webClient;

    // 構(gòu)造方法注入WebClient（全局單例）
    public DeepSeekStreamService(WebClient deepSeekClient) {
        this.webClient = deepSeekClient;
    }

    /**
     * 基礎(chǔ)流式推理方法
     * @param prompt 用戶輸入的提示詞
     * @return 流式響應(yīng)數(shù)據(jù)（每一個(gè)String是一個(gè)chunk）
     */
    public Flux<String> streamInference(String prompt) {
        // 構(gòu)建DeepSeek請(qǐng)求參數(shù)（符合DeepSeek API規(guī)范）
        InferenceRequest request = new InferenceRequest(
                "deepseek-7b-chat", // 模型名稱，根據(jù)實(shí)際使用的模型填寫
                prompt,
                true, // 啟用流式響應(yīng)
                2048, // 最大token數(shù)
                0.7   // 溫度參數(shù)
        );

        return webClient.post()
                .uri("/chat/completions") // DeepSeek聊天補(bǔ)全API路徑
                .bodyValue(request) // 發(fā)送請(qǐng)求體
                .accept(MediaType.TEXT_EVENT_STREAM) // 關(guān)鍵配置：接收SSE流式響應(yīng)
                .retrieve() // 發(fā)起請(qǐng)求并獲取響應(yīng)
                .bodyToFlux(String.class) // 將響應(yīng)體轉(zhuǎn)為Flux<String>（流式數(shù)據(jù)）
                .doOnNext(chunk -> log.debug("Received DeepSeek chunk: {}", chunk)) // 調(diào)試：打印每一個(gè)chunk
                .timeout(Duration.ofMinutes(10)) // 防止長(zhǎng)時(shí)間阻塞，超時(shí)拋出異常
                .onErrorResume(e -> {
                    log.error("Stream inference error", e);
                    return Flux.empty(); // 錯(cuò)誤處理：返回空流，避免影響整體服務(wù)
                });
    }

    /**
     * 帶重試機(jī)制的流式推理方法（生產(chǎn)環(huán)境推薦）
     * 針對(duì)模型服務(wù)臨時(shí)不可用、網(wǎng)絡(luò)波動(dòng)等場(chǎng)景，實(shí)現(xiàn)自動(dòng)重試
     */
    public Flux<String> resilientStreamInference(String prompt) {
        return streamInference(prompt)
                // 重試機(jī)制：最多重試3次，每次間隔1秒，僅對(duì)IO異常重試
                .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
                        .filter(ex -> ex instanceof IOException)
                        .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> 
                                new RuntimeException("Stream retry exhausted", retrySignal.failure())));
    }

    // 內(nèi)部靜態(tài)類：DeepSeek請(qǐng)求參數(shù)封裝（符合API規(guī)范）
    private static class InferenceRequest {
        private String model;
        private String prompt;
        private boolean stream;
        private int max_tokens;
        private double temperature;

        // 構(gòu)造方法
        public InferenceRequest(String model, String prompt, boolean stream, int max_tokens, double temperature) {
            this.model = model;
            this.prompt = prompt;
            this.stream = stream;
            this.max_tokens = max_tokens;
            this.temperature = temperature;
        }

        // getter/setter（省略，可自動(dòng)生成）
        public String getModel() { return model; }
        public void setModel(String model) { this.model = model; }
        public String getPrompt() { return prompt; }
        public void setPrompt(String prompt) { this.prompt = prompt; }
        public boolean isStream() { return stream; }
        public void setStream(boolean stream) { this.stream = stream; }
        public int getMax_tokens() { return max_tokens; }
        public void setMax_tokens(int max_tokens) { this.max_tokens = max_tokens; }
        public double getTemperature() { return temperature; }
        public void setTemperature(double temperature) { this.temperature = temperature; }
    }
}

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

@RestController
@RequestMapping("/api/ai")
public class DeepSeekStreamController {

    private static final Logger log = LoggerFactory.getLogger(DeepSeekStreamController.class);

    private final DeepSeekStreamService deepSeekStreamService;
    private final ObjectMapper objectMapper; // JSON解析工具

    // 構(gòu)造方法注入依賴
    public DeepSeekStreamController(DeepSeekStreamService deepSeekStreamService, ObjectMapper objectMapper) {
        this.deepSeekStreamService = deepSeekStreamService;
        this.objectMapper = objectMapper;
    }

    /**
     * 流式聊天接口（SSE）
     * @param prompt 用戶輸入的提示詞
     * @return 流式響應(yīng)數(shù)據(jù)（解析后的純文本內(nèi)容）
     */
    @GetMapping(value = "/stream-chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamChat(@RequestParam String prompt) {
        // 調(diào)用帶重試的流式方法
        return deepSeekStreamService.resilientStreamInference(prompt)
                // 解析每一個(gè)chunk：提取模型輸出的文本內(nèi)容
                .map(this::parseChunk)
                // 客戶端斷開連接時(shí)觸發(fā)（如用戶關(guān)閉頁(yè)面）
                .doOnCancel(() -> log.info("Client disconnected, stream stopped"))
                // 流式處理異常時(shí)觸發(fā)
                .doOnError(e -> log.error("Stream chat error", e));
    }

    /**
     * 解析DeepSeek的流式響應(yīng)chunk
     * DeepSeek的流式響應(yīng)格式：data: {"id":"xxx","choices":[{"delta":{"content":"xxx"}}]}
     * 需提取choices[0].delta.content中的內(nèi)容
     */
    private String parseChunk(String chunk) {
        try {
            // 去除chunk中的"data: "前綴（SSE格式要求）
            String jsonStr = chunk.replace("data: ", "").trim();
            // 忽略結(jié)束標(biāo)識(shí)（DeepSeek流式結(jié)束時(shí)會(huì)返回data: [DONE]）
            if ("[DONE]".equals(jsonStr)) {
                return "";
            }
            // 解析JSON
            JsonNode node = objectMapper.readTree(jsonStr);
            // 提取文本內(nèi)容，避免空指針
            return node.path("choices").get(0).path("delta").path("content").asText();
        } catch (JsonProcessingException e) {
            log.error("Failed to parse DeepSeek chunk", e);
            return ""; // 解析失敗時(shí)返回空字符串，不影響后續(xù)流式輸出
        }
    }
}

/**
 * 完善的錯(cuò)誤處理+重試機(jī)制
 */
public Flux<String> perfectResilientStream(String prompt) {
    return webClient.post()
            .uri("/chat/completions")
            .bodyValue(new InferenceRequest(prompt))
            .accept(MediaType.TEXT_EVENT_STREAM)
            .retrieve()
            // 處理HTTP錯(cuò)誤狀態(tài)碼（如5xx服務(wù)器錯(cuò)誤、4xx客戶端錯(cuò)誤）
            .onStatus(HttpStatus::is4xxClientError, response -> {
                log.error("Client error: {}", response.statusCode());
                return Mono.error(new RuntimeException("Invalid request, status: " + response.statusCode()));
            })
            .onStatus(HttpStatus::is5xxServerError, response -> {
                log.error("Model service error: {}", response.statusCode());
                return Mono.error(new RuntimeException("Model service unavailable, status: " + response.statusCode()));
            })
            .bodyToFlux(String.class)
            // 重試機(jī)制：指數(shù)退避重試，最多3次，間隔1s、2s、4s
            .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
                    .filter(ex -> ex instanceof IOException || ex.getMessage().contains("Model service unavailable"))
                    .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) -> 
                            new RuntimeException("Stream retry failed after 3 times", retrySignal.failure())))
            // 異常降級(jí)：重試失敗后，返回友好提示
            .onErrorResume(e -> {
                log.error("Final stream error", e);
                return Flux.just("服務(wù)臨時(shí)不可用，請(qǐng)稍后再試~");
            });
}

// 控制消費(fèi)速度：每秒最多處理10個(gè)chunk，避免數(shù)據(jù)堆積
public Flux<String> streamWithBackpressure(String prompt) {
    return deepSeekStreamService.streamInference(prompt)
            .limitRate(10) // 核心配置：控制消費(fèi)速率
            .map(this::parseChunk)
            .subscribe(
                    content -> {
                        // 消費(fèi)邏輯（如返回給前端）
                        System.out.print(content);
                    },
                    error -> log.error("Consume error", error),
                    () -> log.info("Stream consume completed")
            );
}

// 配置背壓緩沖區(qū)，大小為50，緩沖區(qū)滿時(shí)丟棄新數(shù)據(jù)
streamInference(prompt)
        .onBackpressureBuffer(50, 
                () -> log.warn("Backpressure buffer full, discard new chunk"),
                BackpressureOverflowStrategy.DROP_OLDEST)
        .limitRate(10);

reactor:
  netty:
    http:
      pool:
        max-connections: 100  # 最大連接數(shù)，根據(jù)服務(wù)器性能調(diào)整（如8核16G可設(shè)為100-200）
        acquire-timeout: 5s   # 連接獲取超時(shí)時(shí)間，超時(shí)則拋出異常
        max-idle-time: 30s    # 連接最大空閑時(shí)間，空閑超過該時(shí)間則關(guān)閉連接
        pending-acquire-limit: 50 # 等待連接的最大隊(duì)列長(zhǎng)度，隊(duì)列滿時(shí)拒絕請(qǐng)求

import { useState, useEffect } from 'react';

function DeepSeekStreamChat() {
    const [prompt, setPrompt] = useState('');
    const [output, setOutput] = useState('');
    const [loading, setLoading] = useState(false);

    // 發(fā)送流式請(qǐng)求，接收響應(yīng)
    const sendStreamRequest = () => {
        if (!prompt.trim()) {
            alert('請(qǐng)輸入提示詞');
            return;
        }
        // 重置輸出和加載狀態(tài)
        setOutput('');
        setLoading(true);

        // 創(chuàng)建EventSource，連接后端SSE接口
        const eventSource = new EventSource(`/api/ai/stream-chat?prompt=${encodeURIComponent(prompt)}`);

        // 接收流式數(shù)據(jù)
        eventSource.onmessage = (e) => {
            setOutput(prev => prev + e.data);
        };

        // 處理錯(cuò)誤
        eventSource.onerror = (error) => {
            console.error('Stream error:', error);
            setLoading(false);
            eventSource.close(); // 關(guān)閉連接
        };

        // 流式結(jié)束（后端返回[DONE]時(shí)觸發(fā)）
        eventSource.onclose = () => {
            setLoading(false);
            console.log('Stream completed');
        };

        // 組件卸載時(shí)關(guān)閉連接
        return () => {
            eventSource.close();
        };
    };

    return (
        <div style={0 auto', padding: '20px' }}>
            DeepSeek流式聊天<textarea
                value={ => setPrompt(e.target.value)}
                placeholder="請(qǐng)輸入提示詞（如：解釋量子計(jì)算）"
                style={{ width: '100%', height: '100px', marginBottom: '10px' }}
            />
            <button onClick={
                {loading ? '正在生成...' : '發(fā)送請(qǐng)求'}
            <div style={: '20px', padding: '10px', border: '1px solid #eee' }}>
                響應(yīng)結(jié)果：{output}
    );
}

export default DeepSeekStreamChat;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import com.fasterxml.jackson.databind.ObjectMapper;

@SpringBootApplication
public class DeepSeekStreamApplication {

    public static void main(String[] args) {
        SpringApplication.run(DeepSeekStreamApplication.class, args);
    }

    // 注入ObjectMapper（JSON解析工具）
    @Bean
    public ObjectMapper objectMapper() {
        return new ObjectMapper();
    }
}