diff --git a/.gitignore b/.gitignore index 5412251..f67be1b 100644 --- a/.gitignore +++ b/.gitignore @@ -173,3 +173,4 @@ spring-boot-*.jar # Kubernetes *.yaml.bak *.yml.bak + diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/CorpusPortraitServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/CorpusPortraitServiceImpl.java index 502e835..be96aba 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/CorpusPortraitServiceImpl.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/CorpusPortraitServiceImpl.java @@ -1,5 +1,4 @@ package com.volvo.ai.analytic.center.service.impl; - import cn.hutool.core.date.DatePattern; import cn.hutool.core.date.DateUtil; import com.alibaba.fastjson.JSONArray; @@ -27,15 +26,12 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.cloud.context.config.annotation.RefreshScope; import org.springframework.stereotype.Service; - import javax.annotation.Resource; import java.time.ZonedDateTime; import java.time.format.DateTimeFormatter; import java.util.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; - - /** * @description 铭牌语料表-同步表 * @author rz diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java index e0e4d33..7489429 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java @@ -134,8 +134,6 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl messageList = tmOdsVdqwMessagearchivingMapper.queryOdsVdqwMessageByData(statTime, endTime, offset, pageSize, retry); // 处理查询到的数据 - // 使用多线程并行执行两个业务场景 - ExecutorService executorService = Executors.newFixedThreadPool(1); log.info("待处理企微聊天列表messageList",messageList.toString()); // 使用 CompletableFuture 并行处理 messageList.forEach(item -> { @@ -151,7 +149,7 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl { - // 获取配置 - JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT_QIWEI.getCode(), JSONObject.toJSONString(corpusReportDTO),null); - log.info("runDify execDifyFlow , 企微场景，总结需求的dify返回 ：{}", execDifyFlow); - if (null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")) { - JSONObject text = execDifyFlow.getJSONObject("outputs"); - // 发送MQ - log.info("send mq企微总结 {}", text); - tmTelephoneCorpusService.sendMq( CategoryEnum.ENTERPRISE_WECHAT.getCode(), text.toJSONString()); - - try { - ttVdqwRecordMapper.insert(TtVdqwRecord.builder().acceptUserId(item.getFromUserId()).fromUserId(item.getAcceptUserId()).msgTime(maxMsgTimeItem.getMsgTime()).from_accept_user_id(item.getFromAcceptUserId()).build()); - aiAnalysisRequestLogsService.saveOrUpdateAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString("aiAnalysisRequestId")).businessResponse(text.toJSONString()).build()); - } catch (Exception e) { - log.info(" 企业语料处理保存报告异常processItem：{} ", e); - } - } - log.info("runDify execDifyFlow ,企微语料 ，总结和分类业务，返回： {}", execDifyFlow); - }, executorService); + CompletableFuture summaryTask = CompletableFuture.runAsync(() -> { + executeSummaryTask(diFyImageReq, corpusReportDTO, item, maxMsgTimeItem); + }, executor); + + long startTime2 = System.currentTimeMillis(); + CompletableFuture portraitTask = CompletableFuture.runAsync(() -> { + executePortraitTask(unionId, maxMsgTimeItem, chatList, corpusReportDTO); + }, executor); + + // 等待两个任务完成 + CompletableFuture.allOf(summaryTask, portraitTask).join(); + long endTime1 = System.currentTimeMillis(); log.info("第一个业务场景（总结和分类）执行时间: {} ms", (endTime1 - startTime)); - //第一个业务场景， 结束 - - long startTime2 = System.currentTimeMillis(); - CompletableFuture.runAsync(() -> { - //"analysisScene": "分析类型", // 1、企微会话 2、AI通话录音 3、AI铭牌(客流) 4、AI铭牌(试驾) - DiFyReq diFyImageReq2 = new DiFyReq(); - diFyImageReq2.setUser(ConstantStr.corpus_user); - Map inputMap2 = new HashMap<>(); - inputMap2.put("businessId", unionId); - inputMap2.put("communicateDate", DateUtil.format(maxMsgTimeItem.getMsgTime(), DatePattern.NORM_DATETIME_PATTERN)); - inputMap2.put("analysisScene", "1"); - inputMap2.put("version", 2); - inputMap2.put("chat", chatList.toString()); - diFyImageReq2.setInputs(inputMap2); - // 创建新的DiFyReq对象以避免线程安全问题 - diFyImageReq2.setFlowId(oneTokenPortrait); - log.info("runDify execDifyFlow ,客户画像场景 ，token-{} , 对象： {}", oneTokenPortrait, diFyImageReq2); - JSONObject execDifyFlowForPortrait = diFyService.executeDifyFlow(diFyImageReq2, BusinessTypeEnum.CORPUS_PORTRAIT_NAMEPLATE.getCode(), - JSONObject.toJSONString(corpusReportDTO), null); - log.info("runDify execDifyFlow ,客户画像场景 ，返回-{} ", execDifyFlowForPortrait); - - JSONObject text = execDifyFlowForPortrait.getJSONObject("outputs"); - // 发送MQ - log.info("send mq企微客户画像 {}", text); - tmTelephoneCorpusService.sendMq( CategoryEnum.ENTERPRISE_WECHAT.getCode(), text.toJSONString()); - - aiAnalysisRequestLogsService.saveOrUpdateAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlowForPortrait.getString("aiAnalysisRequestId")).businessResponse(text.toJSONString()).build()); - - log.info("runDify execDifyFlow ,企微语料 ,客户画像场景，返回： {}", execDifyFlowForPortrait); - - }, executorService); long endTime2 = System.currentTimeMillis(); log.info("第二个业务场景（用户画像）执行时间: {} ms", (endTime2 - startTime2)); @@ -318,5 +277,67 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl inputMap2 = new HashMap<>(); + inputMap2.put("businessId", unionId); + inputMap2.put("communicateDate", DateUtil.format(maxMsgTimeItem.getMsgTime(), DatePattern.NORM_DATETIME_PATTERN)); + inputMap2.put("analysisScene", "1"); + inputMap2.put("version", 2); + inputMap2.put("chat", chatList.toString()); + diFyImageReq2.setInputs(inputMap2); + // 创建新的DiFyReq对象以避免线程安全问题 + diFyImageReq2.setFlowId(oneTokenPortrait); + log.info("runDify execDifyFlow ,客户画像场景 ，token-{} , 对象： {}", oneTokenPortrait, diFyImageReq2); + JSONObject execDifyFlowForPortrait = diFyService.executeDifyFlow(diFyImageReq2, BusinessTypeEnum.CORPUS_PORTRAIT_NAMEPLATE.getCode(), + JSONObject.toJSONString(corpusReportDTO), null); + log.info("runDify execDifyFlow ,客户画像场景 ，返回-{} ", execDifyFlowForPortrait); + + JSONObject text = execDifyFlowForPortrait.getJSONObject("outputs"); + // 发送MQ + log.info("send mq企微客户画像 {}", text); + tmTelephoneCorpusService.sendMq( CategoryEnum.ENTERPRISE_WECHAT.getCode(), text.toJSONString()); + + aiAnalysisRequestLogsService.saveOrUpdateAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlowForPortrait.getString("aiAnalysisRequestId")).businessResponse(text.toJSONString()).build()); + + log.info("runDify execDifyFlow ,企微语料 ,客户画像场景，返回： {}", execDifyFlowForPortrait); + } catch (Exception e) { + log.error("执行客户画像任务异常", e); + } + } + } \ No newline at end of file