完成AI相关接口

electrmangnetic/pom.xml

@@ -111,6 +111,27 @@
             <artifactId>jaxb-impl</artifactId>
             <version>2.3.3</version>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.ai</groupId>
+            <artifactId>spring-ai-elasticsearch-store-spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.ai</groupId>
+            <artifactId>spring-ai-ollama-spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>co.elastic.clients</groupId>
+            <artifactId>elasticsearch-java</artifactId>
+            <version>8.13.4</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.ai</groupId>
+            <artifactId>spring-ai-pdf-document-reader</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/aop/ServiceAspect.java

@@ -13,6 +13,8 @@ import org.aspectj.lang.reflect.MethodSignature;
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Component;
 import org.springframework.util.StopWatch;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import java.util.Objects;
 
@@ -26,7 +28,7 @@ public class ServiceAspect {
      */
     @Around("execution(* com.electromagnetic.industry.software.manage.controller..*.*(..)))")
     public Object process(ProceedingJoinPoint jp) throws Throwable {
-        String methodInfo = jp.getTarget().getClass().getSimpleName() + "."
+         String methodInfo = jp.getTarget().getClass().getSimpleName() + "."
                 + jp.getSignature().getName();
         Object[] args = jp.getArgs();
         String[] argNames = ((MethodSignature) jp.getSignature()).getParameterNames();
@@ -41,12 +43,13 @@ public class ServiceAspect {
         }
         UserThreadLocal.setReqArgs(paramInfo);
         Object rvt = jp.proceed();
-        if (rvt instanceof ResponseEntity) {
+        String returnInfo = "";
+        if (rvt instanceof ResponseEntity || rvt instanceof Flux) {
             UserThreadLocal.setRes(ElectromagneticResultUtil.success(""));
-            return rvt;
+        } else {
+            UserThreadLocal.setRes((ElectromagneticResult) rvt);
+            returnInfo = JSONUtil.toJsonStr(rvt);
         }
-        UserThreadLocal.setRes((ElectromagneticResult) rvt);
-        String returnInfo = JSONUtil.toJsonStr(rvt);
         log.info("请求接口结束:{},返回参数:{},接口耗时:{}", methodInfo, returnInfo, (System.currentTimeMillis() - startTime) + "毫秒");
         stopwatch.stop();
         log.debug(stopwatch.prettyPrint());

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/config/AiAdvisor.java

@@ -0,0 +1,27 @@
+package com.electromagnetic.industry.software.manage.config;
+
+import jakarta.annotation.Resource;
+import org.springframework.ai.chat.client.advisor.MessageChatMemoryAdvisor;
+import org.springframework.ai.chat.client.advisor.QuestionAnswerAdvisor;
+import org.springframework.ai.chat.memory.InMemoryChatMemory;
+import org.springframework.ai.vectorstore.VectorStore;
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Component;
+
+@Component
+public class AiAdvisor {
+
+    @Resource
+    private VectorStore vectorStore;
+
+    @Bean
+    public QuestionAnswerAdvisor questionAnswerAdvisor() {
+        return new QuestionAnswerAdvisor(vectorStore);
+    }
+
+    @Bean
+    public MessageChatMemoryAdvisor messageChatMemoryAdvisor() {
+        return new MessageChatMemoryAdvisor(new InMemoryChatMemory());
+    }
+
+}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/config/ElePropertyConfig.java

@@ -62,14 +62,6 @@ public class ElePropertyConfig {
     @Value("${backup.mysql.path}")
     private String sqlDirs;
 
-    @Getter
-    @Value("${ai.remote.host}")
-    private String aiRemoteHost;
-
-    @Getter
-    @Value("${ai.remote.port}")
-    private String aiRemotePort;
-
     public String getEleTmpPath() {
         if (EleCommonUtil.isWinOs()) {
             return winPrefix + eleTmpPath;

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/controller/AiChatController.java

@@ -1,39 +0,0 @@
-package com.electromagnetic.industry.software.manage.controller;
-
-import cn.hutool.core.io.FileUtil;
-import cn.hutool.core.util.StrUtil;
-import com.electromagnetic.industry.software.common.resp.ElectromagneticResult;
-import com.electromagnetic.industry.software.common.util.ElectromagneticResultUtil;
-import com.electromagnetic.industry.software.manage.pojo.req.ChatQueryDTO;
-import com.electromagnetic.industry.software.manage.service.serviceimpl.AiChatService;
-import jakarta.annotation.Resource;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestParam;
-import org.springframework.web.bind.annotation.RestController;
-import org.springframework.web.multipart.MultipartFile;
-
-import java.io.IOException;
-
-@RestController
-@RequestMapping("/data/ed/ai/")
-public class AiChatController {
-
-    @Resource
-    private AiChatService aiChatService;
-
-    @RequestMapping(value = "chat")
-    public ElectromagneticResult<?> chat(@RequestBody ChatQueryDTO chatQueryDTO) {
-        return aiChatService.chat(chatQueryDTO.getQuestion());
-    }
-
-    @RequestMapping(value = "addKnowledge")
-    public ElectromagneticResult<?> addKnowledge(@RequestParam("file") MultipartFile file) throws IOException {
-        String fileType = FileUtil.extName(file.getOriginalFilename());
-        if (!StrUtil.equals(fileType, "pdf")) {
-            return ElectromagneticResultUtil.fail("-1", "当前仅支持pdf格式文件");
-        }
-        return aiChatService.addKnowledge(file);
-    }
-
-}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/controller/AiController.java

@@ -0,0 +1,59 @@
+package com.electromagnetic.industry.software.manage.controller;
+
+import com.electromagnetic.industry.software.common.resp.ElectromagneticResult;
+import com.electromagnetic.industry.software.common.util.ElectromagneticResultUtil;
+import com.electromagnetic.industry.software.manage.pojo.req.QueryDTO;
+import com.electromagnetic.industry.software.manage.service.serviceimpl.ChatService;
+import com.electromagnetic.industry.software.manage.tasks.ChatTaskThread1;
+import com.electromagnetic.industry.software.manage.tasks.ChatTaskThread2;
+import com.electromagnetic.industry.software.manage.tasks.ChatTaskThread3;
+import com.electromagnetic.industry.software.manage.tasks.ThreadUtil;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.ai.chat.model.ChatResponse;
+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.ExecutionException;
+import java.util.concurrent.Future;
+
+@Slf4j
+@RestController
+@RequestMapping("/data/ed/ai")
+public class AiController {
+
+    @Resource
+    private ChatService chatService;
+
+    @RequestMapping("/upload")
+    public ElectromagneticResult<?> upload(@RequestParam("file") MultipartFile file) throws Exception {
+        return chatService.addFromUpload(file);
+    }
+
+    @PostMapping("/chat")
+    public ElectromagneticResult<String> chat(@RequestBody QueryDTO queryDTO) throws Exception {
+        log.info("question is --->" + queryDTO.getMsg());
+        ChatTaskThread1 chatTaskThread = new ChatTaskThread1(chatService, queryDTO);
+        Future<String> future = ThreadUtil.getThreadPool().submit(chatTaskThread);
+        String res = future.get();
+        log.info("answer is --->" + res);
+        return ElectromagneticResultUtil.success(res);
+    }
+
+    @PostMapping(path = "/chatStreamStr", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public Flux<String> chatStreamStr(@RequestBody QueryDTO queryDTO) throws ExecutionException, InterruptedException {
+        ChatTaskThread2 chatTaskThread = new ChatTaskThread2(chatService, queryDTO);
+        Future<Flux<String>> future = ThreadUtil.getThreadPool().submit(chatTaskThread);
+        return future.get();
+    }
+
+    @PostMapping(path = "/chatStreamResp", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
+    public Flux<ChatResponse> chatStreamResp(@RequestBody QueryDTO queryDTO) throws ExecutionException, InterruptedException {
+        ChatTaskThread3 chatTaskThread = new ChatTaskThread3(chatService, queryDTO);
+        Future<Flux<ChatResponse>> future = ThreadUtil.getThreadPool().submit(chatTaskThread);
+        return future.get();
+    }
+
+}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/pojo/req/QueryDTO.java

@@ -0,0 +1,11 @@
+package com.electromagnetic.industry.software.manage.pojo.req;
+
+import lombok.Data;
+
+@Data
+public class QueryDTO {
+
+    private String msg;
+
+//    private String userId;
+}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/pojo/req/UploadDTO.java

@@ -0,0 +1,10 @@
+package com.electromagnetic.industry.software.manage.pojo.req;
+
+import lombok.Data;
+
+@Data
+public class UploadDTO {
+
+    private String content;
+
+}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/service/serviceimpl/AiChatService.java

@@ -1,41 +0,0 @@
-package com.electromagnetic.industry.software.manage.service.serviceimpl;
-
-import cn.hutool.core.text.StrFormatter;
-import cn.hutool.http.HttpUtil;
-import cn.hutool.json.JSONUtil;
-import com.electromagnetic.industry.software.common.parse.PdfParse;
-import com.electromagnetic.industry.software.common.resp.ElectromagneticResult;
-import com.electromagnetic.industry.software.manage.config.ElePropertyConfig;
-import jakarta.annotation.Resource;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Service;
-import org.springframework.web.multipart.MultipartFile;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-@Slf4j
-@Service
-public class AiChatService {
-
-    @Resource
-    private ElePropertyConfig elePropertyConfig;
-
-    public ElectromagneticResult<?> chat(String message) {
-        Map<String, Object> map = new HashMap<>();
-        String url = StrFormatter.format("http://{}:{}/data/ed/ai/chat", elePropertyConfig.getAiRemoteHost(), elePropertyConfig.getAiRemotePort());
-        String res = HttpUtil.post(url, JSONUtil.toJsonStr(map), 5 * 60 * 1000);
-        ElectromagneticResult<?> result = JSONUtil.toBean(res, ElectromagneticResult.class);
-        return result;
-    }
-
-    public ElectromagneticResult<?> addKnowledge(MultipartFile file) throws IOException {
-        String content = new PdfParse().parseAllText(file.getInputStream(), "pdf");
-        Map<String, Object> map = new HashMap<>();
-        map.put("content", content);
-        String url = StrFormatter.format("http://{}:{}/data/ed/ai/upload", elePropertyConfig.getAiRemoteHost(), elePropertyConfig.getAiRemotePort());
-        String res = HttpUtil.post(url, JSONUtil.toJsonStr(map));
-        return JSONUtil.toBean(res, ElectromagneticResult.class);
-    }
-}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/service/serviceimpl/ChatService.java

@@ -0,0 +1,109 @@
+package com.electromagnetic.industry.software.manage.service.serviceimpl;
+
+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 com.electromagnetic.industry.software.common.resp.ElectromagneticResult;
+import com.electromagnetic.industry.software.common.util.ElectromagneticResultUtil;
+import com.electromagnetic.industry.software.manage.pojo.req.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.http.codec.ServerSentEvent;
+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;
+
+@Slf4j
+@Service
+public class ChatService {
+
+    @Resource
+    private VectorStore vectorStore;
+
+    @Resource
+    private OllamaChatModel model;
+
+    @Resource
+    private MessageChatMemoryAdvisor messageChatMemoryAdvisor;
+
+    @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("-1", "当前仅支持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");
+
+        return ChatClient.builder(model).defaultAdvisors(messageChatMemoryAdvisor, questionAnswerAdvisor).build().prompt()
+                .user(queryDTO.getMsg())
+                .advisors(advisorSpec -> advisorSpec
+//                        .param(CHAT_MEMORY_CONVERSATION_ID_KEY, queryDTO.getUserId())
+                        .param(AbstractChatMemoryAdvisor.CHAT_MEMORY_RETRIEVE_SIZE_KEY, 100))
+                .call()
+                .content();
+    }
+
+    public Flux<String> chatStreamStr(String msg) {
+        return ChatClient.builder(model).defaultAdvisors(messageChatMemoryAdvisor, questionAnswerAdvisor).build().prompt(msg).stream().content();
+    }
+
+    public Flux<ChatResponse> chatStreamResponse(String msg) {
+        ChatClient.StreamResponseSpec stream = ChatClient.builder(model).defaultAdvisors(messageChatMemoryAdvisor, questionAnswerAdvisor).build().prompt(new Prompt(new UserMessage(msg))).stream();
+        return stream.chatResponse();
+    }
+}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/tasks/ChatTaskThread1.java

@@ -0,0 +1,21 @@
+package com.electromagnetic.industry.software.manage.tasks;
+
+import com.electromagnetic.industry.software.manage.pojo.req.QueryDTO;
+import com.electromagnetic.industry.software.manage.service.serviceimpl.ChatService;
+import lombok.AllArgsConstructor;
+import lombok.NoArgsConstructor;
+
+import java.util.concurrent.Callable;
+
+@NoArgsConstructor
+@AllArgsConstructor
+public class ChatTaskThread1 implements Callable<String> {
+
+    private ChatService chatService;
+    private QueryDTO queryDTO;
+
+    @Override
+    public String call() throws Exception {
+        return chatService.chat(queryDTO);
+    }
+}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/tasks/ChatTaskThread2.java

@@ -0,0 +1,21 @@
+package com.electromagnetic.industry.software.manage.tasks;
+
+import com.electromagnetic.industry.software.manage.pojo.req.QueryDTO;
+import com.electromagnetic.industry.software.manage.service.serviceimpl.ChatService;
+import lombok.AllArgsConstructor;
+import lombok.NoArgsConstructor;
+import reactor.core.publisher.Flux;
+
+import java.util.concurrent.Callable;
+
+@AllArgsConstructor
+@NoArgsConstructor
+public class ChatTaskThread2 implements Callable<Flux<String>> {
+    private ChatService chatService;
+    private QueryDTO queryDTO;
+
+    @Override
+    public Flux<String> call() throws Exception {
+        return chatService.chatStreamStr(queryDTO.getMsg());
+    }
+}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/tasks/ChatTaskThread3.java

@@ -0,0 +1,23 @@
+package com.electromagnetic.industry.software.manage.tasks;
+
+import com.electromagnetic.industry.software.manage.pojo.req.QueryDTO;
+import com.electromagnetic.industry.software.manage.service.serviceimpl.ChatService;
+import lombok.AllArgsConstructor;
+import lombok.NoArgsConstructor;
+import org.springframework.ai.chat.model.ChatResponse;
+import reactor.core.publisher.Flux;
+
+import java.util.concurrent.Callable;
+
+@AllArgsConstructor
+@NoArgsConstructor
+public class ChatTaskThread3 implements Callable<Flux<ChatResponse>> {
+
+    private ChatService chatService;
+    private QueryDTO queryDTO;
+
+    @Override
+    public Flux<ChatResponse> call() throws Exception {
+        return chatService.chatStreamResponse(queryDTO.getMsg());
+    }
+}

electrmangnetic/src/main/java/com/electromagnetic/industry/software/manage/tasks/ThreadUtil.java

@@ -0,0 +1,13 @@
+package com.electromagnetic.industry.software.manage.tasks;
+
+import java.util.concurrent.*;
+
+public final class ThreadUtil {
+
+    private static final ExecutorService THREAD_POOL = new ThreadPoolExecutor(1, 1, 3, TimeUnit.SECONDS, new LinkedBlockingDeque<>(100), Executors.defaultThreadFactory(), new ThreadPoolExecutor.DiscardOldestPolicy());
+
+    public static ExecutorService getThreadPool() {
+        return THREAD_POOL;
+    }
+
+}

electrmangnetic/src/main/resources/application.properties

@@ -9,6 +9,21 @@ spring.datasource.type=com.alibaba.druid.pool.DruidDataSource
 spring.jackson.time-zone=GMT+8
 spring.servlet.multipart.max-file-size=500MB
 spring.servlet.multipart.max-request-size=500MB
+spring.mvc.async.request-timeout=3600000
+# es
+spring.elasticsearch.password=_oO*of_l-b+4mrzXo6B0
+spring.elasticsearch.username=elastic
+spring.elasticsearch.uris=http://139.196.179.195:9200
+#ai
+spring.ai.ollama.base-url=http://localhost:11434
+spring.ai.ollama.chat.options.model=gemma3:latest
+spring.ai.ollama.chat.options.num-g-p-u=1
+spring.ai.ollama.embedding.enabled=true
+spring.ai.ollama.embedding.model=nomic-embed-text
+spring.ai.vectorstore.elasticsearch.initialize-schema=true
+spring.ai.vectorstore.elasticsearch.index-name=rag
+spring.ai.vectorstore.elasticsearch.dimensions=768
+spring.ai.vectorstore.elasticsearch.similarity=cosine
 # mybatis
 mybatis-plus.mapper-locations=classpath:sqlmapper/*.xml
 mybatis-plus.configuration.log-impl=org.apache.ibatis.logging.stdout.StdOutImpl
@@ -37,7 +52,3 @@ backup.remote.host=127.0.0.1
 backup.remote.port=1111
 backup.mysql.path=/workspace/mysqlbak/test
 backup.mysql.script.path=/workspace/mysqlbak/back_dev.sh
-
-# ai options
-ai.remote.host=127.0.0.1
-ai.remote.port=8186