|
|
@@ -0,0 +1,262 @@
|
|
|
+package com.njuzr.eaibackend.service.impl;
|
|
|
+
|
|
|
+import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
+import com.njuzr.eaibackend.config.DeepSeekConfig;
|
|
|
+import com.njuzr.eaibackend.dto.deepseek.DeepSeekMessage;
|
|
|
+import com.njuzr.eaibackend.dto.deepseek.DeepSeekRequest;
|
|
|
+import com.njuzr.eaibackend.dto.deepseek.DeepSeekResponse;
|
|
|
+import com.njuzr.eaibackend.dto.deepseek.DeepSeekStreamResponse;
|
|
|
+import com.njuzr.eaibackend.service.DeepSeekService;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+import org.springframework.http.HttpHeaders;
|
|
|
+import org.springframework.http.MediaType;
|
|
|
+import org.springframework.http.client.SimpleClientHttpRequestFactory;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.web.client.RestTemplate;
|
|
|
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
|
|
|
+
|
|
|
+import java.io.BufferedReader;
|
|
|
+import java.io.InputStreamReader;
|
|
|
+import java.net.HttpURLConnection;
|
|
|
+import java.net.URL;
|
|
|
+import java.nio.charset.StandardCharsets;
|
|
|
+import java.util.Collections;
|
|
|
+import java.util.List;
|
|
|
+import java.util.function.Consumer;
|
|
|
+
|
|
|
+/**
|
|
|
+ * DeepSeek 服务实现类
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+@Service
|
|
|
+public class DeepSeekServiceImpl implements DeepSeekService {
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private DeepSeekConfig deepSeekConfig;
|
|
|
+
|
|
|
+ private final ObjectMapper objectMapper = new ObjectMapper()
|
|
|
+ .configure(com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public DeepSeekResponse chatCompletion(String message) {
|
|
|
+ List<DeepSeekMessage> messages = Collections.singletonList(DeepSeekMessage.user(message));
|
|
|
+ return chatCompletion(messages);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public DeepSeekResponse chatCompletion(List<DeepSeekMessage> messages) {
|
|
|
+ try {
|
|
|
+ DeepSeekRequest request = DeepSeekRequest.builder()
|
|
|
+ .model(deepSeekConfig.getModel())
|
|
|
+ .messages(messages)
|
|
|
+ .stream(false)
|
|
|
+ .build();
|
|
|
+
|
|
|
+ RestTemplate restTemplate = createRestTemplate();
|
|
|
+ org.springframework.http.HttpHeaders headers = new org.springframework.http.HttpHeaders();
|
|
|
+ headers.setContentType(MediaType.APPLICATION_JSON);
|
|
|
+ headers.setBearerAuth(deepSeekConfig.getApiKey());
|
|
|
+
|
|
|
+ org.springframework.http.HttpEntity<DeepSeekRequest> entity =
|
|
|
+ new org.springframework.http.HttpEntity<>(request, headers);
|
|
|
+
|
|
|
+ String url = deepSeekConfig.getBaseUrl() + "/chat/completions";
|
|
|
+ log.info("DeepSeek 非流式请求: {}", url);
|
|
|
+
|
|
|
+ DeepSeekResponse response = restTemplate.postForObject(
|
|
|
+ url, entity, DeepSeekResponse.class);
|
|
|
+
|
|
|
+ log.info("DeepSeek 响应成功");
|
|
|
+ return response;
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("DeepSeek 请求失败: {}", e.getMessage(), e);
|
|
|
+ throw new RuntimeException("DeepSeek 请求失败: " + e.getMessage(), e);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public SseEmitter chatCompletionStream(String message) {
|
|
|
+ List<DeepSeekMessage> messages = Collections.singletonList(DeepSeekMessage.user(message));
|
|
|
+ return chatCompletionStream(messages);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public SseEmitter chatCompletionStream(List<DeepSeekMessage> messages) {
|
|
|
+ SseEmitter emitter = new SseEmitter(0L);
|
|
|
+
|
|
|
+ new Thread(() -> {
|
|
|
+ HttpURLConnection connection = null;
|
|
|
+ try {
|
|
|
+ DeepSeekRequest request = DeepSeekRequest.builder()
|
|
|
+ .model(deepSeekConfig.getModel())
|
|
|
+ .messages(messages)
|
|
|
+ .stream(true)
|
|
|
+ .build();
|
|
|
+
|
|
|
+ String url = deepSeekConfig.getBaseUrl() + "/chat/completions";
|
|
|
+ log.info("DeepSeek 流式请求: {}", url);
|
|
|
+
|
|
|
+ connection = (HttpURLConnection) new URL(url).openConnection();
|
|
|
+ connection.setRequestMethod("POST");
|
|
|
+ connection.setRequestProperty(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
|
|
|
+ connection.setRequestProperty(HttpHeaders.AUTHORIZATION, "Bearer " + deepSeekConfig.getApiKey());
|
|
|
+ connection.setDoOutput(true);
|
|
|
+ connection.setDoInput(true);
|
|
|
+ connection.setConnectTimeout(deepSeekConfig.getTimeout());
|
|
|
+ connection.setReadTimeout(deepSeekConfig.getTimeout());
|
|
|
+
|
|
|
+ String requestBody = objectMapper.writeValueAsString(request);
|
|
|
+ connection.getOutputStream().write(requestBody.getBytes(StandardCharsets.UTF_8));
|
|
|
+
|
|
|
+ int responseCode = connection.getResponseCode();
|
|
|
+ if (responseCode != 200) {
|
|
|
+ throw new RuntimeException("HTTP error code: " + responseCode);
|
|
|
+ }
|
|
|
+
|
|
|
+ try (BufferedReader reader = new BufferedReader(
|
|
|
+ new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8))) {
|
|
|
+ String line;
|
|
|
+ while ((line = reader.readLine()) != null) {
|
|
|
+ if (line.startsWith("data: ")) {
|
|
|
+ String data = line.substring(6);
|
|
|
+ if ("[DONE]".equals(data)) {
|
|
|
+ emitter.complete();
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ DeepSeekStreamResponse streamResponse =
|
|
|
+ objectMapper.readValue(data, DeepSeekStreamResponse.class);
|
|
|
+ String content = streamResponse.getContent();
|
|
|
+ if (content != null) {
|
|
|
+ emitter.send(content);
|
|
|
+ }
|
|
|
+ if (streamResponse.isDone()) {
|
|
|
+ emitter.complete();
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("解析流式响应失败: {}", data);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("DeepSeek 流式请求失败: {}", e.getMessage(), e);
|
|
|
+ emitter.completeWithError(e);
|
|
|
+ } finally {
|
|
|
+ if (connection != null) {
|
|
|
+ connection.disconnect();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }).start();
|
|
|
+
|
|
|
+ return emitter;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void chatCompletionStream(String message, Consumer<String> onContent, Runnable onComplete, Consumer<Throwable> onError) {
|
|
|
+ List<DeepSeekMessage> messages = Collections.singletonList(DeepSeekMessage.user(message));
|
|
|
+ chatCompletionStream(messages, onContent, onComplete, onError);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void chatCompletionStream(List<DeepSeekMessage> messages, Consumer<String> onContent, Runnable onComplete, Consumer<Throwable> onError) {
|
|
|
+ log.info("【DeepSeekService】chatCompletionStream被调用,消息数={}", messages.size());
|
|
|
+
|
|
|
+ new Thread(() -> {
|
|
|
+ HttpURLConnection connection = null;
|
|
|
+ try {
|
|
|
+ DeepSeekRequest request = DeepSeekRequest.builder()
|
|
|
+ .model(deepSeekConfig.getModel())
|
|
|
+ .messages(messages)
|
|
|
+ .stream(true)
|
|
|
+ .build();
|
|
|
+
|
|
|
+ String url = deepSeekConfig.getBaseUrl() + "/chat/completions";
|
|
|
+ log.info("【DeepSeekService】开始HTTP请求: {}", url);
|
|
|
+ log.info("【DeepSeekService】使用模型: {}", deepSeekConfig.getModel());
|
|
|
+
|
|
|
+ connection = (HttpURLConnection) new URL(url).openConnection();
|
|
|
+ connection.setRequestMethod("POST");
|
|
|
+ connection.setRequestProperty(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
|
|
|
+ connection.setRequestProperty(HttpHeaders.AUTHORIZATION, "Bearer " + deepSeekConfig.getApiKey());
|
|
|
+ connection.setDoOutput(true);
|
|
|
+ connection.setDoInput(true);
|
|
|
+ connection.setConnectTimeout(deepSeekConfig.getTimeout());
|
|
|
+ connection.setReadTimeout(deepSeekConfig.getTimeout());
|
|
|
+
|
|
|
+ String requestBody = objectMapper.writeValueAsString(request);
|
|
|
+ log.info("【DeepSeekService】请求体大小: {} bytes", requestBody.getBytes(StandardCharsets.UTF_8).length);
|
|
|
+
|
|
|
+ connection.getOutputStream().write(requestBody.getBytes(StandardCharsets.UTF_8));
|
|
|
+ connection.getOutputStream().flush();
|
|
|
+ connection.getOutputStream().close();
|
|
|
+
|
|
|
+ int responseCode = connection.getResponseCode();
|
|
|
+ log.info("【DeepSeekService】收到响应码: {}", responseCode);
|
|
|
+
|
|
|
+ if (responseCode != 200) {
|
|
|
+ log.error("【DeepSeekService】HTTP错误: {}", responseCode);
|
|
|
+ throw new RuntimeException("HTTP error code: " + responseCode);
|
|
|
+ }
|
|
|
+
|
|
|
+ log.info("【DeepSeekService】开始读取流式响应...");
|
|
|
+ int chunkCount = 0;
|
|
|
+
|
|
|
+ try (BufferedReader reader = new BufferedReader(
|
|
|
+ new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8))) {
|
|
|
+ String line;
|
|
|
+ while ((line = reader.readLine()) != null) {
|
|
|
+ if (line.startsWith("data: ")) {
|
|
|
+ String data = line.substring(6);
|
|
|
+ if ("[DONE]".equals(data)) {
|
|
|
+ log.info("【DeepSeekService】收到[DONE],流结束");
|
|
|
+ onComplete.run();
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ DeepSeekStreamResponse streamResponse =
|
|
|
+ objectMapper.readValue(data, DeepSeekStreamResponse.class);
|
|
|
+ String content = streamResponse.getContent();
|
|
|
+ if (content != null && !content.isEmpty()) {
|
|
|
+ onContent.accept(content);
|
|
|
+ chunkCount++;
|
|
|
+ if (chunkCount % 10 == 0) {
|
|
|
+ log.info("【DeepSeekService】已发送{}个内容块", chunkCount);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (streamResponse.isDone()) {
|
|
|
+ log.info("【DeepSeekService】响应标记完成");
|
|
|
+ onComplete.run();
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.warn("【DeepSeekService】解析流式响应失败: {}", data);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ log.info("【DeepSeekService】流式响应处理完成,共{}个内容块", chunkCount);
|
|
|
+
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("【DeepSeekService】流式请求失败: {}", e.getMessage(), e);
|
|
|
+ onError.accept(e);
|
|
|
+ } finally {
|
|
|
+ if (connection != null) {
|
|
|
+ connection.disconnect();
|
|
|
+ log.info("【DeepSeekService】连接已关闭");
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }).start();
|
|
|
+
|
|
|
+ log.info("【DeepSeekService】新线程已启动处理流式请求");
|
|
|
+ }
|
|
|
+
|
|
|
+ private RestTemplate createRestTemplate() {
|
|
|
+ SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
|
|
|
+ factory.setConnectTimeout(deepSeekConfig.getTimeout());
|
|
|
+ factory.setReadTimeout(deepSeekConfig.getTimeout());
|
|
|
+ return new RestTemplate(factory);
|
|
|
+ }
|
|
|
+}
|