admin/vlv-ai-analytic-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dcc重构代码

Browse Source
This commit is contained in:
ZLI263
2025-09-22 15:30:43 +08:00
parent 43a8cf9b54
commit f497a90e32
4 changed files with 123 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/CorpusProcessKafkaProducer.java
View File

@@ -1,6 +1,7 @@
package com.volvo.ai.analytic.center.mq;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.volvo.ai.analytic.center.constant.Constant;
import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
@@ -48,19 +49,24 @@ public class CorpusProcessKafkaProducer {
private final ObjectMapper objectMapper = new ObjectMapper();
@Value("${rocketmq.producer.corpus.dcctopic}")
private String dccMqTipic;
private String dccMqTipic; // =ai_corpus_Dcc_topic
@Resource
private RocketMQTemplate rocketMqTemplate;
@Value("${batch.threadNumDCC}")
int threadNum =2;
ExecutorService executor = Executors.newFixedThreadPool(threadNum);
//spring.kafka.topic=topic_voc_covert_text_log spring.kafka.group=topic_group_covert_text_log_voc_ai_ac_uat
@KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.group}")
public void listen(List<ConsumerRecord<String, Object>> recordMessages) {
long startTime = System.currentTimeMillis();
log.info("CorpusProcessKafkaProducer Received message: {}", recordMessages);
// 获取消息列表
int optimalThreadPoolSize = Runtime.getRuntime().availableProcessors() + 2;
log.info("获取的线程数：{}", optimalThreadPoolSize);
ExecutorService executor = Executors.newFixedThreadPool(optimalThreadPoolSize);
log.info("获取的线程数：{}", threadNum);
try {
if(CollectionUtils.isNotEmpty(recordMessages)){
log.info("CorpusProcessKafkaProducer List size: {}", recordMessages.size());
@@ -70,6 +76,15 @@ public class CorpusProcessKafkaProducer {
try {
String message = (String) record.value();
AicorpusTelephoneDTO aicorpusTelephone = objectMapper.readValue(message, AicorpusTelephoneDTO.class);
JSONObject jsonObject = JSONObject.parseObject( aicorpusTelephone.getDisplay());
Long audioDuration = jsonObject.getLong("audio_duration"); // 毫秒
if(audioDuration <= 10000){//时间单位是 毫秒
log.info("电话语料时长小于10秒，不进行dify处理");
return;
}
log.info("aicorpusTelephone categoryCode:{}, display: {}", aicorpusTelephone.getCategoryCode(), aicorpusTelephone.getDisplay());
DisplayDTO display = objectMapper.readValue(aicorpusTelephone.getDisplay(), DisplayDTO.class);
log.info("aicorpusTelephone display getSegments: {}", display.getSegments());

2
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmNameplateCorpusServiceImpl.java
View File

@@ -70,7 +70,7 @@ public class TmNameplateCorpusServiceImpl extends ServiceImpl<TmNameplateCorpusM
@Autowired
@Resource(name = "threadPoolTaskExecutor")
private ThreadPoolTaskExecutor executor;
@Value("${batch.threadNum}")
@Value("${batch.threadNumNameplate}")
int threadNum =2;
private Semaphore semaphore = new Semaphore(threadNum); // 限制并发数

5
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmOdsVdqwMessagearchivingServiceImpl.java
View File

@@ -38,7 +38,6 @@ import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
@@ -92,7 +91,7 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
@Autowired
private TmTelephoneCorpusService tmTelephoneCorpusService;
@Value("${batch.threadNum}")
@Value("${batch.threadNumQiWei}")
int threadNum =2;
String consultantIdStr ="consultantId";
@@ -264,7 +263,7 @@ public class TmOdsVdqwMessagearchivingServiceImpl extends ServiceImpl<TmOdsVdqwM
// 创建新的DiFyReq对象以避免线程安全问题
diFyImageReq2.setFlowId(oneTokenPortrait);
log.info("runDify execDifyFlow ,客户画像场景 ，token-{} , 对象： {}", oneTokenPortrait, diFyImageReq2);
JSONObject execDifyFlowForPortrait = diFyService.executeDifyFlow(diFyImageReq2, BusinessTypeEnum.CORPUS_PORTRAIT_NAMEPLATE.getCode(),
JSONObject execDifyFlowForPortrait = diFyService.executeDifyFlow(diFyImageReq2, BusinessTypeEnum.CORPUS_PORTRAIT_QIWEI.getCode(),
JSONObject.toJSONString(corpusReportDTO), null);
log.info("runDify execDifyFlow ,客户画像场景 ，返回-{} ", execDifyFlowForPortrait);

130
ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/TmTelephoneCorpusServiceImpl.java
View File

@@ -42,6 +42,9 @@ import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
@@ -66,6 +69,8 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
private String topic;
@Value("${dify.corpus.telephoneToken}")
private String telephoneToken;
@Value("${dify.corpus.portrait.oneToken}")
private String oneTokenPortrait;
@Value("${dify.intelligentCustomer.channel4In1Token}")
private String channel4In1Token;
@@ -87,6 +92,13 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
@Autowired
CorpusQuestionProducer corpusQuestionProducer;
@Value("${batch.threadNumDCC}")
int threadNum =2;
ExecutorService executor = Executors.newFixedThreadPool(threadNum);
private String communicateDateStr = "communicateDate";
private String outputsStr="outputs";
private String aiAnalysisRequestIdStr = "aiAnalysisRequestId";
@Override
@Transactional
public void saveTelephoneCorpus(TmTelephoneCorpus tmTelephoneCorpus) {
@@ -96,7 +108,7 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
@Override
public void runTelephoneCorpusDify(AicorpusTelephoneDTO aicorpusTelephone) {
if(null != aicorpusTelephone){
if(null != aicorpusTelephone) {
List<DataMaskingRule> maskingRuleItems = dataMaskingRuleService.getDataMaskingRuleListByApplicationChannel(BusinessTypeEnum.SMART_ASSISTANT.getCode());
@@ -109,10 +121,10 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
diFyImageReq.setFlowId(telephoneToken);
JSONObject jsonObject = JSONObject.parseObject( aicorpusTelephone.getDisplay());
JSONObject jsonObject = JSONObject.parseObject(aicorpusTelephone.getDisplay());
JSONArray segments = jsonObject.getJSONArray("segments");
Long audioDuration = jsonObject.getLong("audio_duration"); // 毫秒
if(audioDuration/1000 <=10){
if (audioDuration / 1000 <= 10) {
log.info("电话语料时长小于10秒，不进行dify处理");
return;
}
@@ -129,11 +141,11 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
String text = result.getString("text");
JSONObject analysisInfo = result.getJSONObject("analysis_info");
String role = analysisInfo.getString("role");
String title="";
if(role.equals("AGENT")){
title="顾问";
}else{
title="客户";
String title = "";
if (role.equals("AGENT")) {
title = "顾问";
} else {
title = "客户";
}
// 拼接 role 和 text
String chat = title + ": " + text;
@@ -147,12 +159,12 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
String formattedDateStartTime = zonedDateTime.format(formatter);
String carModel = getCarModelList();
inputMap.put("chat",chatList.toString());
inputMap.put("model",carModel);
inputMap.put("chat", chatList.toString());
inputMap.put("model", carModel);
inputMap.put("analysisScene", "2");
inputMap.put("recordId", aicorpusTelephone.getSourceId());
inputMap.put("communicateDate", formattedDateStartTime);
inputMap.put("version",2);
inputMap.put(communicateDateStr, formattedDateStartTime);
inputMap.put("version", 2);
diFyImageReq.setInputs(inputMap);
CorpusReportDTO corpusReportDTO = new CorpusReportDTO();
corpusReportDTO.setCorpusTime(formattedDateStartTime);
@@ -160,32 +172,92 @@ public class TmTelephoneCorpusServiceImpl extends ServiceImpl<TmTelephoneCorpusM
corpusReportDTO.setAnalysisScene(2l);
corpusReportDTO.setCarModel(carModel);
// 获取配置
JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT.getCode(), JSONObject.toJSONString(corpusReportDTO), aicorpusTelephone.getAiAnalysisRequestId());
log.info("runDify execDifyFlow {}",execDifyFlow);
if(null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")){
JSONObject text = execDifyFlow.getJSONObject("outputs");
String aiAnalysisRequestId = execDifyFlow.getString("aiAnalysisRequestId");
// 发送MQ
log.info("send mq {}",text);
sendMq( CategoryEnum.PHONE_VOICE.getCode(), text.toJSONString());
long startTime = System.currentTimeMillis();
CompletableFuture.runAsync(() -> {
try {
if(StringUtils.isNotEmpty(aicorpusTelephone.getAiAnalysisRequestId())){
AiAnalysisErrors aiAnalysisErrors = new AiAnalysisErrors();
aiAnalysisErrors.setAiAnalysisRequestId(aiAnalysisRequestId);
aiAnalysisErrors.setAiAnalysisErrorHandlingStatus("1");
aiAnalysisErrorsService.updateAiAnalysisErrors(aiAnalysisErrors);
}
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString("aiAnalysisRequestId")).businessResponse(text.toJSONString()).build());
diFyImageReq.setFlowId(telephoneToken);
log.info("铭牌语料telephoneToken：{}", telephoneToken);
JSONObject execDifyFlow = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.SMART_ASSISTANT.getCode(),
JSONObject.toJSONString(corpusReportDTO), aicorpusTelephone.getAiAnalysisRequestId());
log.info("dcc总结场景,总结和分类 runDify execDifyFlow 返回 ： {}", execDifyFlow);
parseDfiyResult(execDifyFlow, aicorpusTelephone.getSourceId(), formattedDateStartTime, aicorpusTelephone.getAiAnalysisRequestId(),
BusinessTypeEnum.SMART_ASSISTANT.getCode());
} catch (Exception e) {
log.info(" 电话语料处理保存报告异常processItem：{} ", e);
log.error("runDify 异常", e);
}
}
}, executor);
long endTime = System.currentTimeMillis();
log.info("第一个业务场景（dcc总结和分类）执行时间: {} ms", (endTime - startTime));
//第一个业务场景， 结束
long startTime2 = System.currentTimeMillis();
CompletableFuture.runAsync(() -> {
try {
// 创建新的DiFyReq对象以避免线程安全问题
inputMap.put("businessId", aicorpusTelephone.getSourceId());
diFyImageReq.setInputs(inputMap);
diFyImageReq.setFlowId(oneTokenPortrait);
log.info("dcc总结场景,客户画像runDify execDifyFlow 返回 ： {}", oneTokenPortrait);
JSONObject execDifyFlowForPortrait = diFyService.executeDifyFlow(diFyImageReq, BusinessTypeEnum.CORPUS_PORTRAIT_DCC.getCode(),
JSONObject.toJSONString(corpusReportDTO), aicorpusTelephone.getAiAnalysisRequestId());
log.info("dcc客户画像场景 runDify execDifyFlow 返回 , dcc： {}", execDifyFlowForPortrait);
parseDfiyResult(execDifyFlowForPortrait, aicorpusTelephone.getSourceId(), formattedDateStartTime,
aicorpusTelephone.getAiAnalysisRequestId(), BusinessTypeEnum.CORPUS_PORTRAIT_DCC.getCode());
} catch (Exception e) {
log.error("runDify 异常", e);
}
}, executor);
long endTime2 = System.currentTimeMillis();
log.info("第二个业务场景（dcc用户画像）执行时间: {} ms", (endTime2 - startTime2));
//第二个业务场景， 结束
}
}
private void parseDfiyResult(JSONObject execDifyFlow,String recordId,String communicateDate,
String aiAnalysisRequestIdDB,String businessType) {
log.info("dcc语料 parseDfiyResult ，businessType: {} execDifyFlow: {}, recordId: {},communicateDate: {}", businessType,execDifyFlow,recordId,communicateDate);
try {
if(null != execDifyFlow && execDifyFlow.get("status").equals("succeeded")){
JSONObject text = execDifyFlow.getJSONObject(outputsStr);
// 发送MQ
if (BusinessTypeEnum.CORPUS_PORTRAIT_DCC.getCode().equals(businessType)){//DCC 客户画像
log.info("send mq 电话语料场景，客户画像需求， {}",text);
sendMq( CategoryEnum.PORTRAIT_ALLIN.getCode(), text.toJSONString());
}else{ // dcc 总结
log.info("send mq 电话语料场景，总结需求， {}",text);
sendMq( CategoryEnum.PHONE_VOICE.getCode(), text.toJSONString());
}
try {
if(StringUtils.isNotEmpty(aiAnalysisRequestIdDB)){
AiAnalysisErrors aiAnalysisErrors = new AiAnalysisErrors();
aiAnalysisErrors.setAiAnalysisRequestId(aiAnalysisRequestIdDB);
aiAnalysisErrors.setAiAnalysisErrorHandlingStatus("1");
aiAnalysisErrorsService.updateAiAnalysisErrors(aiAnalysisErrors);
}
aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(execDifyFlow.getString(aiAnalysisRequestIdStr)).businessResponse(text.toJSONString()).build());
} catch (Exception e) {
log.info(" 电话语料处理保存报告异常processItem：{} ", e.getMessage());
}
}else {
log.info("dcc语料 parseDfiyResult 非正常状态");
}
}catch (Exception e){
log.error("dcc语料 parseDfiyResult 异常，businessType: {} error: {}", businessType, e);
}
}
/**
* 4in1语料处理