AI回答结果流式返回给前台。

ai-chat/pom.xml

@@ -46,12 +46,19 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
         </dependency>
-
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+        </dependency>
         <dependency>
             <groupId>cn.hutool</groupId>
             <artifactId>hutool-all</artifactId>
             <version>5.8.22</version>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.ai</groupId>
+            <artifactId>spring-ai-pdf-document-reader</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

ai-chat/src/main/java/cn/szsd/ai/chat/controller/ChatController.java

@@ -2,13 +2,16 @@ package cn.szsd.ai.chat.controller;
 
 import cn.szsd.ai.chat.pojo.ElectromagneticResult;
 import cn.szsd.ai.chat.pojo.QueryDTO;
-import cn.szsd.ai.chat.pojo.UploadDTO;
 import cn.szsd.ai.chat.service.ChatService;
 import cn.szsd.ai.chat.service.ElectromagneticResultUtil;
 import cn.szsd.ai.chat.utils.ChatTaskThread;
+import cn.szsd.ai.chat.utils.ChatTaskThread1;
 import cn.szsd.ai.chat.utils.ThreadUtil;
 import jakarta.annotation.Resource;
+import org.springframework.http.MediaType;
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.multipart.MultipartFile;
+import reactor.core.publisher.Flux;
 
 import java.util.concurrent.*;
 import java.util.logging.Logger;
@@ -18,12 +21,11 @@ import java.util.logging.Logger;
 public class ChatController {
 
     Logger log = Logger.getLogger(ChatController.class.getName());
-
     @Resource
     private ChatService chatService;
 
     @PostMapping("/chat")
-    public ElectromagneticResult<String> test(@RequestBody QueryDTO queryDTO) throws Exception {
+    public ElectromagneticResult<String> chat(@RequestBody QueryDTO queryDTO) throws Exception {
         log.info("question is --->" + queryDTO.getMsg());
         ChatTaskThread chatTaskThread = new ChatTaskThread(chatService, queryDTO);
         Future<String> future = ThreadUtil.getThreadPool().submit(chatTaskThread);
@@ -32,10 +34,16 @@ public class ChatController {
         return ElectromagneticResultUtil.success(res);
     }
 
-    @PostMapping("/upload")
-    public ElectromagneticResult<?> upload(@RequestBody UploadDTO uploadDTO) {
-        chatService.add(uploadDTO.getContent());
-        return ElectromagneticResultUtil.success("");
+    @PostMapping(path = "/chatStream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public Flux<String> chatStream(@RequestBody QueryDTO queryDTO) throws ExecutionException, InterruptedException {
+        ChatTaskThread1 chatTaskThread = new ChatTaskThread1(chatService, queryDTO);
+        Future<Flux<String>> future = ThreadUtil.getThreadPool().submit(chatTaskThread);
+        return future.get();
+    }
+
+    @RequestMapping("/upload")
+    public ElectromagneticResult<?> upload(@RequestParam("file") MultipartFile file) throws Exception {
+        return chatService.addFromUpload(file);
     }
 
 }

ai-chat/src/main/java/cn/szsd/ai/chat/service/ChatService.java

@@ -1,19 +1,27 @@
 package cn.szsd.ai.chat.service;
 
+import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.util.IdUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.crypto.digest.DigestUtil;
+import cn.szsd.ai.chat.pojo.ElectromagneticResult;
 import cn.szsd.ai.chat.pojo.QueryDTO;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.ai.chat.client.ChatClient;
 import org.springframework.ai.chat.client.advisor.*;
-import org.springframework.ai.chat.messages.UserMessage;
-import org.springframework.ai.chat.model.ChatResponse;
-import org.springframework.ai.chat.prompt.Prompt;
 import org.springframework.ai.document.Document;
 import org.springframework.ai.ollama.OllamaChatModel;
+import org.springframework.ai.reader.pdf.PagePdfDocumentReader;
 import org.springframework.ai.vectorstore.VectorStore;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.multipart.MultipartFile;
 import reactor.core.publisher.Flux;
 
+import java.io.File;
+import java.nio.charset.Charset;
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -34,11 +42,44 @@ public class ChatService {
     @Resource
     private QuestionAnswerAdvisor questionAnswerAdvisor;
 
+    @Value("file.md5RecordPath")
+    private String uploadFileMd5RecordPath;
+
+    @PostMapping
+    public void init() {
+        if (!FileUtil.exist(uploadFileMd5RecordPath)) {
+            FileUtil.touch(uploadFileMd5RecordPath);
+        }
+    }
+
     public void add(String content) {
         List<Document> documents = Stream.of(content).map(Document::new).collect(Collectors.toList());
         vectorStore.write(documents);
     }
 
+
+    public ElectromagneticResult<?> addFromUpload(MultipartFile file) throws Exception{
+        String fileType = FileUtil.extName(file.getOriginalFilename());
+        if (!StrUtil.equals(fileType, "pdf")) {
+            return ElectromagneticResultUtil.fail("当前仅支持pdf格式文件");
+        }
+        String fileMd5 = DigestUtil.md5Hex(file.getInputStream());
+        List<String> lines = FileUtil.readLines(uploadFileMd5RecordPath, Charset.defaultCharset());
+        if (lines.contains(fileMd5)) {
+            return ElectromagneticResultUtil.success(fileMd5);
+        }
+        String pdfTmpPath = FileUtil.getParent(uploadFileMd5RecordPath, 1) + File.separator + IdUtil.fastSimpleUUID() + "." + fileType;
+        FileUtil.writeFromStream(file.getInputStream(), pdfTmpPath);
+
+        PagePdfDocumentReader pagePdfDocumentReader = new PagePdfDocumentReader(pdfTmpPath);
+        List<Document> documents = pagePdfDocumentReader.read();
+        vectorStore.write(documents);
+        lines.add(fileMd5);
+        FileUtil.writeLines(lines, uploadFileMd5RecordPath, Charset.defaultCharset());
+        FileUtil.del(pdfTmpPath);
+        return ElectromagneticResultUtil.success(fileMd5);
+    }
+
     public String chat(QueryDTO queryDTO) {
 
         log.info("Start call model to answer");
@@ -52,8 +93,7 @@ public class ChatService {
                 .content();
     }
 
-    public Flux<ChatResponse> chat1(String msg, String userId) {
-        Prompt prompt = new Prompt(new UserMessage(msg));
-        return this.model.stream(prompt);
+    public Flux<String> chatStream(String msg) {
+        return ChatClient.builder(model).defaultAdvisors(messageChatMemoryAdvisor, questionAnswerAdvisor).build().prompt(msg).stream().content();
     }
 }

ai-chat/src/main/java/cn/szsd/ai/chat/utils/ChatTaskThread1.java

@@ -0,0 +1,23 @@
+package cn.szsd.ai.chat.utils;
+
+import cn.szsd.ai.chat.pojo.QueryDTO;
+import cn.szsd.ai.chat.service.ChatService;
+import lombok.AllArgsConstructor;
+import lombok.NoArgsConstructor;
+import reactor.core.publisher.Flux;
+
+import java.util.concurrent.Callable;
+
+@AllArgsConstructor
+@NoArgsConstructor
+public class ChatTaskThread1 implements Callable<Flux<String>> {
+
+    private ChatService chatService;
+    private QueryDTO queryDTO;
+
+
+    @Override
+    public Flux<String> call() throws Exception {
+        return chatService.chatStream(queryDTO.getMsg());
+    }
+}

ai-chat/src/main/resources/application.yml

@@ -1,8 +1,8 @@
 spring:
   elasticsearch:
-    password: LCFLr9iQx*mKtwfjJj40
+    password: _oO*of_l-b+4mrzXo6B0
     username: elastic
-    uris: http://127.0.0.1:9200
+    uris: http://139.196.179.195:9200
 
   ai:
     ollama:
@@ -29,3 +29,6 @@ spring:
 
 server:
   port: 8186
+
+file:
+  md5RecordPath: /workspace/app/ai/ele-ai/record.txt