Spring Boot中接入DeepSeek的流式输出

第一步，添加依赖：

 <!-- WebFlux 响应式支持 -->
<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

第二步，配置WebClient。这里需要设置WebClient实例，用于向DeepSeek的API发送请求。需要配置baseUrl，可能还需要添加认证头，比如Authorization Bearer token

@Configuration
public class DeepSeekConfig {
    @Value("${deepseek.api.key}")
    private String apiKey;

    @Value("${deepseek.api.url}")
    private String apiUrl;


    @Bean
    public WebClient deepseekWebClient() {
        return WebClient.builder()
                .baseUrl(apiUrl)
                .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + apiKey)
                .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .codecs(configurer ->
                        configurer.defaultCodecs().maxInMemorySize(16 * 1024 * 1024)) //处理大响应
                .build();
    }
}

第三步，创建服务类，使用WebClient发送请求并处理流式响应。这里可能需要将响应转换为Flux，然后处理每个数据块。比如，读取每个chunk的数据，提取需要的内容，可能还要处理不同的数据格式，比如JSON对象或者特定格式的文本。

@Service
@RequiredArgsConstructor
public class DeepSeekService {

    private final WebClient deepSeekClient;

    public Flux<String> streamCompletion(String prompt) {
        Map<String, Object> requestBody = new HashMap<>();
        requestBody.put("model", "deepseek-reasoner");
        requestBody.put("messages", List.of(Map.of("role", "user", "content", prompt)));
        // 启用流式
        requestBody.put("stream", true);
        requestBody.put("temperature", 0.7);

        return deepSeekClient.post()
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(requestBody)
                .retrieve()
                .bodyToFlux(String.class)
                .filter(data -> !"[DONE]".equals(data)) // 过滤结束标记
                .map(this::extractContent)
                .onErrorResume(e -> {
                    // 错误处理
                    return Flux.error(new RuntimeException("API调用失败: " + e.getMessage()));
                });
    }
    // 解析响应数据
    private String extractContent(String data) {
        try {
            JsonNode node = new ObjectMapper().readTree(data);
            return (Strings.isBlank(node.at("/choices/0/delta/reasoning_content").asText()) || "null".equalsIgnoreCase(node.at("/choices/0/delta/reasoning_content").asText())) ? node.at("/choices/0/delta/content").asText() : node.at("/choices/0/delta/reasoning_content").asText();
        } catch (Exception e) {
            return "";
        }
    }


}

第四步，创建控制器

@RestController
@RequiredArgsConstructor
@Slf4j
public class StreamController {
    // 添加速率限制
    private final RateLimiter rateLimiter = RateLimiter.create(10); // 10次/秒

    private final DeepSeekService deepSeekService;

    @GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamChat(@RequestParam String message) {
        if (!rateLimiter.tryAcquire()) {
            return Flux.error(new RuntimeException("请求过于频繁"));
        }
        return deepSeekService.streamCompletion(message)
                .map(content -> "data: " + content + "\n\n") // SSE格式
                .doOnComplete(() -> log.info("流式传输完成"))
                .doOnError(e -> log.error("流式错误: {}", e.getMessage()))
                .log("deepseek-stream", Level.FINE) // 详细日志
                .metrics();// 集成Micrometer监控
    }
}

配置

spring.application.name=deepseek_demo
server.port=8080

deepseek.api.key=sk-?????????????
deepseek.api.url=https://api.deepseek.com/v1/chat/completions

测试：

@SpringBootTest
class DeepseekDemoApplicationTests {
    @Autowired
    private DeepSeekService deepSeekService;
    @Test
    void contextLoads() {
        Flux<String> flux = deepSeekService.streamCompletion("springboot如何接入deepseek的流式输出，给出详细的步骤及代码实现");
        StepVerifier.create(flux)
                .thenConsumeWhile(content -> {
                    System.out.print(content);
                    return true;
                })
                .verifyComplete();
    }
}