|
|
@@ -0,0 +1,388 @@
|
|
|
+package com.cy.guoyan.admin.module.ai.service.aiservice;
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSONArray;
|
|
|
+import com.alibaba.fastjson.JSONObject;
|
|
|
+import com.cy.guoyan.admin.framework.common.util.object.BeanUtils;
|
|
|
+import com.cy.guoyan.admin.framework.mybatis.core.query.LambdaQueryWrapperX;
|
|
|
+import com.cy.guoyan.admin.framework.security.core.util.SecurityFrameworkUtils;
|
|
|
+import com.cy.guoyan.admin.module.ai.controller.admin.chatdocument.vo.ChatDocumentSaveReqVO;
|
|
|
+import com.cy.guoyan.admin.module.ai.controller.admin.chatknowledgedataset.vo.ChatKnowledgeDatasetSaveReqVO;
|
|
|
+import com.cy.guoyan.admin.module.ai.controller.admin.chatmessage.vo.ChatMessageRespVO;
|
|
|
+import com.cy.guoyan.admin.module.ai.controller.admin.chatmessage.vo.ChatMessageSendReqVO;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.dataobject.chatapp.ChatAppDO;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.dataobject.chatdocument.ChatDocumentDO;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.dataobject.chatknowledgedataset.ChatKnowledgeDatasetDO;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.dataobject.chatmessage.ChatMessageDO;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.dataobject.chatsession.ChatSessionDO;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.mysql.chatapp.ChatAppMapper;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.mysql.chatdocument.ChatDocumentMapper;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.mysql.chatknowledgedataset.ChatKnowledgeDatasetMapper;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.mysql.chatmessage.ChatMessageMapper;
|
|
|
+import com.cy.guoyan.admin.module.ai.dal.mysql.chatsession.ChatSessionMapper;
|
|
|
+import com.cy.guoyan.admin.module.ai.service.chatdocument.ChatDocumentService;
|
|
|
+import com.cy.guoyan.admin.module.ai.util.DifyClient;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.apache.commons.lang3.StringUtils;
|
|
|
+import org.springframework.beans.factory.annotation.Qualifier;
|
|
|
+import org.springframework.beans.factory.annotation.Value;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.web.multipart.MultipartFile;
|
|
|
+import javax.annotation.Resource;
|
|
|
+import java.io.*;
|
|
|
+import java.net.HttpURLConnection;
|
|
|
+import java.net.URL;
|
|
|
+import java.nio.charset.StandardCharsets;
|
|
|
+import java.time.Instant;
|
|
|
+import java.time.LocalDateTime;
|
|
|
+import java.time.ZoneId;
|
|
|
+import java.util.*;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+import static com.cy.guoyan.admin.framework.common.exception.util.ServiceExceptionUtil.exception;
|
|
|
+import static com.cy.guoyan.admin.module.system.enums.ErrorCodeConstants.CHAT_APP_NOT_EXISTS;
|
|
|
+import static com.cy.guoyan.admin.module.system.enums.ErrorCodeConstants.CHAT_KNOWLEDGE_DATASET_NOT_EXISTS;
|
|
|
+import static java.nio.charset.StandardCharsets.UTF_8;
|
|
|
+
|
|
|
+@Service
|
|
|
+@Qualifier("dify")
|
|
|
+@Slf4j
|
|
|
+public class DifyService implements AiService {
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private ChatKnowledgeDatasetMapper chatKnowledgeDatasetMapper;
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private ChatDocumentService chatDocumentService;
|
|
|
+
|
|
|
+ @Value("${dify.datasetsKey}")
|
|
|
+ private String difyApiKey;
|
|
|
+
|
|
|
+ @Value("${dify.baseUrl}")
|
|
|
+ private String difyBaseUrl;
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private ChatSessionMapper chatSessionMapper;
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private ChatMessageMapper chatMessageMapper;
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private ChatDocumentMapper chatDocumentMapper;
|
|
|
+
|
|
|
+ @Resource
|
|
|
+ private ChatAppMapper chaAppMapper;
|
|
|
+
|
|
|
+ private final DifyClient difyClient;
|
|
|
+
|
|
|
+ public DifyService(DifyClient difyClient) {
|
|
|
+ this.difyClient = difyClient;
|
|
|
+ }
|
|
|
+
|
|
|
+ private HttpURLConnection prepare(String method, String path) throws IOException {
|
|
|
+ URL url = new URL(difyBaseUrl + path);
|
|
|
+ HttpURLConnection conn = (HttpURLConnection) url.openConnection();
|
|
|
+ conn.setRequestMethod(method);
|
|
|
+ conn.setRequestProperty("Authorization", "Bearer " + difyApiKey);
|
|
|
+ return conn;
|
|
|
+ }
|
|
|
+
|
|
|
+ /** 读取响应 body 或抛异常 */
|
|
|
+ private String readResponse(HttpURLConnection conn) throws IOException {
|
|
|
+ int code = conn.getResponseCode();
|
|
|
+ InputStream is = (code >= 200 && code < 300) ? conn.getInputStream() : conn.getErrorStream();
|
|
|
+ if (is == null) throw new IOException("无返回流，HTTP " + code);
|
|
|
+ String body;
|
|
|
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, UTF_8))) {
|
|
|
+ body = reader.lines().collect(Collectors.joining());
|
|
|
+ }
|
|
|
+ if (code < 200 || code >= 300) {
|
|
|
+ throw new IOException("Dify 错误 HTTP " + code + "，" + body);
|
|
|
+ }
|
|
|
+ return body;
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public JSONObject createDataset(ChatKnowledgeDatasetSaveReqVO vo)throws IOException {
|
|
|
+ JSONObject body = new JSONObject();
|
|
|
+ body.put("name", vo.getName());
|
|
|
+ body.put("description", vo.getDescription());
|
|
|
+ body.put("indexing_technique", "high_quality");
|
|
|
+// body.put("embedding_model", vo.getEmbeddingModel());
|
|
|
+ body.put("auto_update", vo.getAutoUpdate());
|
|
|
+ body.put("provider", "vendor");
|
|
|
+ body.put("permission", "all_team_members");
|
|
|
+
|
|
|
+ HttpURLConnection conn = prepare("POST", "/v1/datasets");
|
|
|
+ conn.setDoOutput(true);
|
|
|
+ conn.setRequestProperty("Content-Type", "application/json");
|
|
|
+ try (OutputStream os = conn.getOutputStream()) {
|
|
|
+ os.write(body.toJSONString().getBytes(UTF_8));
|
|
|
+ }
|
|
|
+ String resp = readResponse(conn);
|
|
|
+ return JSONObject.parseObject(resp);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void updateDataset(ChatKnowledgeDatasetSaveReqVO updateReqVO) {
|
|
|
+ // 校验存在
|
|
|
+ validateChatKnowledgeDatasetExists(updateReqVO.getId());
|
|
|
+ // 更新
|
|
|
+ ChatKnowledgeDatasetDO updateObj = BeanUtils.toBean(updateReqVO, ChatKnowledgeDatasetDO.class);
|
|
|
+ chatKnowledgeDatasetMapper.updateById(updateObj);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public String deleteDataset(String datasetId) throws IOException {
|
|
|
+ HttpURLConnection conn = prepare("DELETE", "/v1/datasets/" + datasetId);
|
|
|
+ log.info("删除知识库：{}", datasetId);
|
|
|
+ return readResponse(conn);
|
|
|
+ }
|
|
|
+
|
|
|
+ private void validateChatKnowledgeDatasetExists(Long id) {
|
|
|
+ if (chatKnowledgeDatasetMapper.selectById(id) == null) {
|
|
|
+ throw exception(CHAT_KNOWLEDGE_DATASET_NOT_EXISTS);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public JSONObject createDocumentByFile(MultipartFile file, ChatDocumentSaveReqVO createReqVO,File tmpFile) throws IOException {
|
|
|
+ String datasetId = createReqVO.getDatasetId();
|
|
|
+ String originalFilename = createReqVO.getName();
|
|
|
+
|
|
|
+ Map<String, Object> requestData = new HashMap<>();
|
|
|
+ requestData.put("datasetId", datasetId);
|
|
|
+ requestData.put("name", originalFilename);
|
|
|
+ requestData.put("indexing_technique", "high_quality");
|
|
|
+ requestData.put("doc_form", "text_model");
|
|
|
+
|
|
|
+ Map<String, Object> processRule = new HashMap<>();
|
|
|
+ if (Objects.equals("custom",createReqVO.getProcessRuleMode())) {
|
|
|
+ processRule.put("mode", "custom");
|
|
|
+ // 预处理规则
|
|
|
+ Map<String, Object> rules = new HashMap<>();
|
|
|
+ List<Map<String, Object>> preProcessingRules = new ArrayList<>();
|
|
|
+ Map<String, Object> rule1 = new HashMap<>();
|
|
|
+ rule1.put("id", "remove_extra_spaces");
|
|
|
+ rule1.put("enabled", true);
|
|
|
+ preProcessingRules.add(rule1);
|
|
|
+ Map<String, Object> rule2 = new HashMap<>();
|
|
|
+ rule2.put("id", "remove_urls_emails");
|
|
|
+ rule2.put("enabled", true);
|
|
|
+ preProcessingRules.add(rule2);
|
|
|
+ rules.put("pre_processing_rules", preProcessingRules);
|
|
|
+
|
|
|
+ // 分段规则
|
|
|
+ Map<String, Object> segmentation = new HashMap<>();
|
|
|
+ segmentation.put("separator", "\n\n");
|
|
|
+ segmentation.put("max_tokens", 500);
|
|
|
+ rules.put("segmentation", segmentation);
|
|
|
+ processRule.put("rules", rules);
|
|
|
+ }else {
|
|
|
+ processRule.put("mode", "automatic");
|
|
|
+ }
|
|
|
+ requestData.put("process_rule", processRule);
|
|
|
+
|
|
|
+ return difyClient.createDocumentByFile(tmpFile, originalFilename, requestData);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public JSONObject getDocumentIndexingStatus(String datasetId,String documentId) throws IOException {
|
|
|
+
|
|
|
+ ChatDocumentDO doc = chatDocumentMapper.selectById(documentId);
|
|
|
+
|
|
|
+ new Thread(() -> {
|
|
|
+ try {
|
|
|
+ int waited = 0;
|
|
|
+ while (waited < 60*60*24) {
|
|
|
+ JSONObject status = difyClient.getDocumentIndexingStatus(datasetId, doc.getBatch());
|
|
|
+ JSONArray array = status.getJSONArray("data");
|
|
|
+ if (array == null || array.isEmpty()) {
|
|
|
+ log.warn("dify文档处理状态为空 name={} 状态：{}", doc.getName(), status);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ JSONObject item = array.getJSONObject(0);
|
|
|
+ String state = item.getString("indexing_status");
|
|
|
+ Integer completedSegments = item.getInteger("completed_segments");
|
|
|
+ log.info("dify文档处理状态 name={} 状态：{},分段：{}", doc.getName(), state, completedSegments);
|
|
|
+ if ("completed".equals(state)) {
|
|
|
+ if (doc.getId() == null) {
|
|
|
+ ChatDocumentDO aDo = chatDocumentMapper.selectOne("dataset_id", doc.getDatasetId(), "batch", doc.getBatch());
|
|
|
+ doc.setId(aDo.getId());
|
|
|
+ doc.setCreator(aDo.getCreator());
|
|
|
+ doc.setUpdater(aDo.getUpdater());
|
|
|
+ doc.setName(aDo.getName());
|
|
|
+ }
|
|
|
+ log.info("dify文档处理完成 name={} ✅", doc.getName());
|
|
|
+// doc.setHitCount(completedSegments);
|
|
|
+ doc.setIndexingStatus("completed");
|
|
|
+ doc.setCompletedSegments(completedSegments);
|
|
|
+ try {
|
|
|
+ JSONObject jsonObject = difyClient.listDocuments(datasetId, doc.getName(), null, null);
|
|
|
+ JSONArray dataArray = jsonObject.getJSONArray("data");
|
|
|
+ if (!dataArray.isEmpty()) {
|
|
|
+ JSONObject firstDocument = dataArray.getJSONObject(0);
|
|
|
+ // 提取 "word_count" 字段
|
|
|
+ int wordCount = firstDocument.getIntValue("word_count");
|
|
|
+ doc.setWordCount(wordCount);
|
|
|
+ }
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("dify获取文档信息失败", e);
|
|
|
+ }
|
|
|
+ chatDocumentMapper.updateById(doc);
|
|
|
+ return;
|
|
|
+ } else if ("failed".equals(state)) {
|
|
|
+ log.warn("dify文档处理失败 name={} ❌ 错误：{}", doc.getName(), status.getString("error"));
|
|
|
+ // 可更新失败状态
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ Thread.sleep(5000);
|
|
|
+ waited += 5;
|
|
|
+ }
|
|
|
+ log.warn("轮询超时 name={} ⏰", doc.getName());
|
|
|
+
|
|
|
+
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("异步轮询 Dify 状态失败", e);
|
|
|
+ }
|
|
|
+ }).start();
|
|
|
+
|
|
|
+
|
|
|
+ return difyClient.getDocumentIndexingStatus(datasetId, doc.getBatch());
|
|
|
+
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void deleteDocument(String datasetId, String documentId) throws IOException {
|
|
|
+ difyClient.deleteDocument(datasetId, documentId);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public ChatMessageRespVO sendChatMessageBlock(ChatMessageSendReqVO sendReqVO) throws IOException {
|
|
|
+ String conversationId = sendReqVO.getConversationId();
|
|
|
+ // 1. 从登录上下文取用户昵称
|
|
|
+ String nickname = String.valueOf(SecurityFrameworkUtils.getLoginUser().getInfo().get("nickname"));
|
|
|
+ ChatMessageDO msg = new ChatMessageDO();
|
|
|
+ msg.setUserId(nickname);
|
|
|
+ msg.setContent(sendReqVO.getQuery());
|
|
|
+ msg.setRole("user");
|
|
|
+ if (StringUtils.isNotBlank(conversationId)) {
|
|
|
+ msg.setConversationId(conversationId);
|
|
|
+ }
|
|
|
+ chatMessageMapper.insert(msg);
|
|
|
+ ChatAppDO appDO = getAppKey(sendReqVO);
|
|
|
+
|
|
|
+ // 2. 构造请求体
|
|
|
+ JSONObject body = new JSONObject();
|
|
|
+ body.put("query", sendReqVO.getQuery());
|
|
|
+ body.put("inputs", Optional.ofNullable(sendReqVO.getInputs()).orElse(new HashMap<>()));
|
|
|
+ if (StringUtils.isNotBlank(conversationId)) {
|
|
|
+ body.put("conversation_id", conversationId);
|
|
|
+ }
|
|
|
+ body.put("user", nickname);
|
|
|
+ body.put("response_mode", "blocking");
|
|
|
+ body.put("auto_generate_name", true);
|
|
|
+ body.put("enable_retrieval", true);
|
|
|
+// ChatKnowledgeDatasetDO datasetDO = chatKnowledgeDatasetMapper.selectById(sendReqVO.getDatasetId());
|
|
|
+// body.put("datasets", Collections.singletonList(datasetDO.getDatasetId()));
|
|
|
+ log.info("Dify 请求体: {}", body.toJSONString());
|
|
|
+ // 3. 发起 HTTP POST
|
|
|
+ URL url = new URL(difyBaseUrl + "/v1/chat-messages");
|
|
|
+ HttpURLConnection conn = (HttpURLConnection) url.openConnection();
|
|
|
+ conn.setRequestProperty("Authorization", "Bearer " + appDO.getAppKey());
|
|
|
+ conn.setRequestProperty("Content-Type", "application/json");
|
|
|
+ conn.setDoOutput(true);
|
|
|
+ conn.setRequestMethod("POST");
|
|
|
+ try (OutputStream os = conn.getOutputStream()) {
|
|
|
+ os.write(body.toJSONString().getBytes(StandardCharsets.UTF_8));
|
|
|
+ }
|
|
|
+
|
|
|
+ // 5. 读取响应（
|
|
|
+ int code = conn.getResponseCode();
|
|
|
+ InputStream rawStream = (code >= 200 && code < 300)
|
|
|
+ ? conn.getInputStream()
|
|
|
+ : conn.getErrorStream();
|
|
|
+
|
|
|
+ String respStr;
|
|
|
+ if (rawStream == null) {
|
|
|
+ respStr = "Dify API 无返回流，HTTP 状态码：" + code;
|
|
|
+ } else {
|
|
|
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(rawStream, StandardCharsets.UTF_8))) {
|
|
|
+ respStr = reader.lines().collect(Collectors.joining());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ if (code < 200 || code >= 300) {
|
|
|
+ throw new IOException("Dify API返回错误: HTTP " + code + "，响应: " + respStr);
|
|
|
+ }
|
|
|
+
|
|
|
+ // 6. 解析并保存
|
|
|
+ return parseAndSave(respStr, nickname,msg, appDO);
|
|
|
+ }
|
|
|
+ private ChatMessageRespVO parseAndSave(String jsonStr, String nickname, ChatMessageDO msg, ChatAppDO appDO) {
|
|
|
+ JSONObject json = JSONObject.parseObject(jsonStr);
|
|
|
+ String did = json.getString("conversation_id");
|
|
|
+
|
|
|
+ // 6.1 本地如果尚未存 Dify 的 conversation_id，则更新
|
|
|
+ ChatSessionDO session = chatSessionMapper.selectOne(new LambdaQueryWrapperX<ChatSessionDO>().eq(ChatSessionDO::getConversationId, did));
|
|
|
+ if (session == null) {
|
|
|
+ session = new ChatSessionDO();
|
|
|
+ session.setName(msg.getContent());
|
|
|
+ session.setUserId(nickname);
|
|
|
+ session.setConversationId(did);
|
|
|
+ session.setAppId(appDO.getId());
|
|
|
+ chatSessionMapper.insert(session);
|
|
|
+ // ======== 补充：从 Dify 获取会话名称（同步 + 异步）=======
|
|
|
+// asyncUpdateSessionName(did, session.getId());
|
|
|
+
|
|
|
+ }
|
|
|
+ chatMessageMapper.updateById(msg.setConversationId(did).setSessionId(session.getId()));
|
|
|
+ JSONObject metadata = json.getJSONObject("metadata");
|
|
|
+ String metadataStr = metadata.toJSONString();
|
|
|
+ // 6.2 保存消息
|
|
|
+ ChatMessageDO m = new ChatMessageDO();
|
|
|
+ m.setConversationId(did);
|
|
|
+ m.setSessionId(session.getId());
|
|
|
+ m.setUserId(nickname);
|
|
|
+ m.setMessageId(json.getString("message_id"));
|
|
|
+ m.setContent(json.getString("answer"));
|
|
|
+ m.setRole("assistant");
|
|
|
+ m.setMetadataJson(metadataStr);
|
|
|
+ chatMessageMapper.insert(m);
|
|
|
+
|
|
|
+ // 7. 组装 VO 返回
|
|
|
+ ChatMessageRespVO vo = new ChatMessageRespVO();
|
|
|
+ vo.setSessionId(session.getId());
|
|
|
+ vo.setConversationId(did);
|
|
|
+ vo.setUserId(nickname);
|
|
|
+ vo.setMessageId(m.getMessageId());
|
|
|
+ vo.setContent(m.getContent());
|
|
|
+ vo.setRole(m.getRole());
|
|
|
+ vo.setMessageTime(m.getCreateTime());
|
|
|
+ vo.setStatus("completed");
|
|
|
+ vo.setResponseJson(jsonStr);
|
|
|
+ vo.setMetadataJson(metadata);
|
|
|
+ return vo;
|
|
|
+ }
|
|
|
+
|
|
|
+ public ChatAppDO getAppKey(ChatMessageSendReqVO sendReqVO) {
|
|
|
+ ChatAppDO appDO = new ChatAppDO();
|
|
|
+ if (StringUtils.isBlank(sendReqVO.getName())) {
|
|
|
+ throw exception(CHAT_APP_NOT_EXISTS);
|
|
|
+ } else {
|
|
|
+ if (sendReqVO.getNetworking() == true) {
|
|
|
+ appDO = chaAppMapper.selectOne(new LambdaQueryWrapperX<ChatAppDO>().like(ChatAppDO::getAppName, sendReqVO.getName()).eq(ChatAppDO::getNetworking, 1).eq(ChatAppDO::getAppType, 1));
|
|
|
+ }else {
|
|
|
+ appDO = chaAppMapper.selectOne(new LambdaQueryWrapperX<ChatAppDO>().like(ChatAppDO::getAppName, sendReqVO.getName()).eq(ChatAppDO::getNetworking, 0).eq(ChatAppDO::getAppType, 1));
|
|
|
+ }
|
|
|
+ if (appDO == null) {
|
|
|
+ throw exception(CHAT_APP_NOT_EXISTS);
|
|
|
+ }
|
|
|
+ return appDO;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+}
|
yangshun