修改AI结果查询

2025-04-16 17:51:56 +08:00
parent 0f9e21e86c
commit c34149925d
11 changed files with 201 additions and 95 deletions
--- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/AiAnalysisDifyController.java
+++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/AiAnalysisDifyController.java
@@ -1,6 +1,7 @@
 package com.volvo.ai.analytic.center.controller;


+import com.volvo.ai.analytic.center.dto.req.AnalysisQueryReq;
 import com.volvo.ai.analytic.center.dto.req.AnalysisReq;
 import com.volvo.ai.analytic.center.dto.resp.AnalysisResp;
 import com.volvo.ai.analytic.center.service.AiAnalysisDifyService;
@@ -11,22 +12,25 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.context.config.annotation.RefreshScope;
+import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;


-@RestController("analysis")
+@RestController()
@Api(tags = "分析中心接口")
@Slf4j
@RefreshScope
+@RequestMapping("analysis")
 public class AiAnalysisDifyController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
-    private AiAnalysisDifyService AiDifyResultService;
+    private AiAnalysisDifyService aiAnalysisDifyService;



@@ -34,15 +38,28 @@ public class AiAnalysisDifyController {
    @ApiOperation(value = "更新dify结果")
    public ResultMsg<Object> updateByAiId(@RequestBody String message) {
        log.info("updateByAiId message: {}",message);
-        AiDifyResultService.updateAiDifyResult(message);
+        aiAnalysisDifyService.updateAiDifyResult(message);
        return ResultMsg.ok("ok");
    }

    @PostMapping("/aiAnalyze")
    @ApiOperation(value = "Ai解析接口")
-    public AnalysisResp<Object> aiAnalyze(@RequestBody AnalysisReq analysisReq) {
+    public AnalysisResp<Object> aiAnalyze(@Validated @RequestBody  AnalysisReq analysisReq) {
        log.info("aiAnalyze data: {}",analysisReq);
-        return AnalysisResp.success("ok");
+        return  aiAnalysisDifyService.aiAnalyze(analysisReq);
+    }
+
+    @PostMapping("/query")
+    @ApiOperation(value = "Ai解析结果查询")
+    public AnalysisResp<Object> query(@RequestBody  @Validated AnalysisQueryReq analysisQueryReq) {
+        log.info("aiAnalyze query data: {}",analysisQueryReq);
+        return  aiAnalysisDifyService.query(analysisQueryReq);
+    }
+    @PostMapping("/callback")
+    @ApiOperation(value = "Ai解析结果查询")
+    public AnalysisResp<Object> testCallback(@RequestBody AnalysisResp analysisResp) {
+        log.info("aiAnalyze testCallback data: {}",analysisResp);
+        return  AnalysisResp.success("ok");
    }


--- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/AiAnalysisDifyJob.java
+++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/AiAnalysisDifyJob.java
@@ -0,0 +1,74 @@
+package com.volvo.ai.analytic.center.job;
+
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.volvo.ai.analytic.center.entity.AiAnalysisRequestLogs;
+import com.volvo.ai.analytic.center.feign.DiFyFeign;
+import com.volvo.ai.analytic.center.mapper.AiAnalysisRequestLogsMapper;
+import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
+import com.volvo.common.core.util.ResultMsg;
+import com.xxl.job.core.context.XxlJobHelper;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ *
+ */
+@Slf4j
+@Component
+@RestController
+public class AiAnalysisDifyJob {
+
+    @Autowired
+    private AiAnalysisRequestLogsMapper aiAnalysisRequestLogsMapper;
+
+    @Autowired
+    private DiFyFeign diFyFeign;
+    @Autowired
+    private AiAnalysisRequestLogsService aiAnalysisRequestLogsService;
+
+    /**
+     * 失败的查询处理
+     */
+    @XxlJob("workflowRunIdFaile")
+    @PostMapping("workflowRunIdFaile")
+    public ResultMsg workflowRunIdFaile(@RequestBody String paramJson) {
+        try {
+            // 获取任务参数
+            String param = XxlJobHelper.getJobParam();
+            if(StringUtils.isEmpty(param)){
+                param = paramJson;
+            }
+            List<String> workflowRunIdList = Arrays.asList(param.split(","));
+            LambdaQueryWrapper<AiAnalysisRequestLogs> queryWrapper = new LambdaQueryWrapper<>();
+            queryWrapper.in(AiAnalysisRequestLogs::getWorkflowRunId, workflowRunIdList);
+            queryWrapper.eq(AiAnalysisRequestLogs::getIsDeleted, "0");
+            List<AiAnalysisRequestLogs> aiAnalysisRequestLogsList =  aiAnalysisRequestLogsMapper.selectList(queryWrapper);
+            aiAnalysisRequestLogsList.stream().forEach(aiAnalysisRequestLogs -> {
+                JSONObject jsonResult =  diFyFeign.queryWorkFlowById("Bearer "+aiAnalysisRequestLogs.getDifyAgentKey(),aiAnalysisRequestLogs.getWorkflowRunId());
+                String outputs = jsonResult.getString("outputs");
+                aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder()
+                        .aiAnalysisRequestId(aiAnalysisRequestLogs.getAiAnalysisRequestId())
+                        .difyResponse(outputs)
+                        .build());
+            });
+
+        } catch (Exception e) {
+            log.error("processMessageByTask 定时任务补偿处理消息异常",e.getMessage());
+            throw new RuntimeException(e);
+        }
+        return ResultMsg.ok();
+    }
+
+
+
+}
--- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/DccCorpusJob.java
+++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/DccCorpusJob.java
@@ -1,56 +0,0 @@
-package com.volvo.ai.analytic.center.job;
-
-import com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper;
-import com.volvo.ai.analytic.center.service.TmOdsVdqwMessagearchivingService;
-import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
-import com.volvo.common.core.util.ResultMsg;
-import com.xxl.job.core.context.XxlJobHelper;
-import com.xxl.job.core.handler.annotation.XxlJob;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RestController;
-
-
-@Slf4j
-@Component
-@RestController
-public class DccCorpusJob {
-
-    @Autowired
-    private TmOdsVdqwMessagearchivingService tmOdsVdqwMessagearchivingService;
-
-    @Autowired
-    private TmTelephoneCorpusService tmTelephoneCorpusService;
-
-    @Autowired
-    private  TmTelephoneCorpusMapper tmTelephoneCorpusMapper;
-
-    /**
-     * dcc语料处理
-     */
-    @XxlJob("dccCorpusJob")
-    public ResultMsg dccCorpusJob(@RequestBody String paramJson) {
-        try {
-            // 获取任务参数
-            String param = XxlJobHelper.getJobParam();
-            if(StringUtils.isEmpty(param)){
-                param = paramJson;
-            }
-
-            // 分页查询 过滤已跑批并发送的
-
-
-            tmOdsVdqwMessagearchivingService.runQiWeiCorpusDify(param);
-        } catch (Exception e) {
-            log.error("processMessageByTask 定时任务补偿处理消息异常",e.getMessage());
-            throw new RuntimeException(e);
-        }
-        return ResultMsg.ok();
-    }
-
-
-
-}
--- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/AnalysisDifyMqConsumer.java
+++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/AnalysisDifyMqConsumer.java
@@ -2,13 +2,10 @@
 package com.volvo.ai.analytic.center.mq;

 import com.alibaba.fastjson.JSONObject;
-import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
 import com.volvo.ai.analytic.center.dto.req.DiFyReq;
 import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService;
 import com.volvo.ai.analytic.center.service.DiFyService;
-import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
@@ -60,6 +57,7 @@ public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt>  {
 			log.info("analysisDifyMqConsumer message: " + message);
 			DiFyReq difyReq = JSONObject.parseObject(message, DiFyReq.class);
 			CompletableFuture<JSONObject> future = diFyService.asyncExecuteDifyFlow(difyReq);
+//			JSONObject json = future.get();
 			future.thenAccept(result -> {
 				JSONObject difyRequest = JSONObject.parseObject(JSONObject.toJSONString(difyReq.getInputs()), JSONObject.class);
 				String aiAnalysisRequestId = difyRequest.getString("aiAnalysisRequestId");
@@ -68,7 +66,7 @@ public class AnalysisDifyMqConsumer implements RocketMQListener<MessageExt>  {
 			});
 			log.info("analysisDifyMqConsumer处理完成，耗时：{}", System.currentTimeMillis() - startTime);
 		} catch (Exception e) {
-			log.info(" dcc mq 处理失败：{}", e.getMessage());
+			log.info(" analysisDifyMqConsumer 处理失败：{}", e.getMessage());
 		}
 	}

--- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/AiAnalysisDifyService.java
+++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/AiAnalysisDifyService.java
@@ -1,5 +1,6 @@
 package com.volvo.ai.analytic.center.service;

+import com.volvo.ai.analytic.center.dto.req.AnalysisQueryReq;
 import com.volvo.ai.analytic.center.dto.req.AnalysisReq;
 import com.volvo.ai.analytic.center.dto.resp.AnalysisResp;

@@ -10,4 +11,8 @@ public interface AiAnalysisDifyService {

 	AnalysisResp aiAnalyze(AnalysisReq analysisReq);

+	AnalysisResp query(AnalysisQueryReq analysisQueryReq);
+
+
+
 }
--- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/AiAnalysisDifyServiceImpl.java
+++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/AiAnalysisDifyServiceImpl.java
@@ -5,14 +5,13 @@ import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.volvo.ai.analytic.center.dto.corpus.AicorpusTelephoneDTO;
 import com.volvo.ai.analytic.center.dto.corpus.CorpusReportDTO;
+import com.volvo.ai.analytic.center.dto.req.AnalysisQueryReq;
 import com.volvo.ai.analytic.center.dto.req.AnalysisReq;
 import com.volvo.ai.analytic.center.dto.req.DiFyReq;
 import com.volvo.ai.analytic.center.dto.resp.AnalysisDifyResultDTO;
 import com.volvo.ai.analytic.center.dto.resp.AnalysisResp;
 import com.volvo.ai.analytic.center.entity.AiAnalysisRequestLogs;
 import com.volvo.ai.analytic.center.entity.TcBusinessType;
-import com.volvo.ai.analytic.center.enums.BusinessTypeEnum;
-import com.volvo.ai.analytic.center.enums.CategoryEnum;
 import com.volvo.ai.analytic.center.mapper.TcBusinessTypeMapper;
 import com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper;
 import com.volvo.ai.analytic.center.service.AiAnalysisDifyService;
@@ -34,10 +33,7 @@ import org.springframework.stereotype.Service;
 import javax.annotation.Resource;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.stream.Collectors;

@Slf4j
@@ -88,26 +84,39 @@ public class AiAnalysisDifyServiceImpl implements AiAnalysisDifyService {

 	@Override
 	public AnalysisResp aiAnalyze(AnalysisReq analysisReq) {
-		if(null == analysisReq){
+
+		Optional.ofNullable(analysisReq).orElseThrow(() -> {
 			log.info("请求Ai解析对象为空！");
-			return AnalysisResp.failed("请求Ai解析对象为空！");
-		}
+			return new IllegalArgumentException("请求Ai解析对象为空！");
+		});
+
 		String aiAnalysisRequestId = StringUtils.isEmpty(analysisReq.getAiAnalysisRequestId())? AiAnalysisUtils.getAiAnalysisRequestId(analysisReq.getAiAnalysisRequestType()):analysisReq.getAiAnalysisRequestId();

 		Map<String, TcBusinessType> queryTcBusinessType = queryTcBusinessType();
-		TcBusinessType tcBusinessType = queryTcBusinessType.get(analysisReq.getAiAnalysisRequestType());
-		if(null == tcBusinessType || StringUtils.isEmpty(tcBusinessType.getWorkflowApiKey())){
-			log.info("接入业务类型未配置！");
-			return AnalysisResp.failed("接入业务类型未配置！");
-		}
+		TcBusinessType tcBusinessType = Optional.ofNullable(queryTcBusinessType.get(analysisReq.getAiAnalysisRequestType()))
+				.filter(businessType -> StringUtils.isNotEmpty(businessType.getWorkflowApiKey()))
+				.orElseThrow(() -> {
+					log.info("接入业务类型未配置！");
+					return new IllegalArgumentException("接入业务类型未配置！");
+				});

+		DiFyReq diFyReq = createDiFyReq(analysisReq, tcBusinessType, aiAnalysisRequestId);
+		saveAiAnalysisRequestLogs(analysisReq, diFyReq, aiAnalysisRequestId);
+		sendMq(analysisDifyTopic, diFyReq);
+		return AnalysisResp.success(analysisReq.getData(),aiAnalysisRequestId);
+	}
+
+	private DiFyReq createDiFyReq(AnalysisReq analysisReq, TcBusinessType tcBusinessType, String aiAnalysisRequestId) {
 		DiFyReq diFyReq = new DiFyReq();
-		diFyReq.setUser(StringUtils.isEmpty(tcBusinessType.getWorkflowUser())?analysisReq.getAiAnalysisRequestType().concat("_USER"):tcBusinessType.getWorkflowUser());
+		diFyReq.setUser(StringUtils.isEmpty(tcBusinessType.getWorkflowUser()) ? analysisReq.getAiAnalysisRequestType().concat("_USER") : tcBusinessType.getWorkflowUser());
 		diFyReq.setFlowId(tcBusinessType.getWorkflowApiKey());
-		// 发送mq消息
 		JSONObject difyRequest = JSONObject.parseObject(JSONObject.toJSONString(analysisReq.getData()), JSONObject.class);
-		difyRequest.put("aiAnalysisRequestId",aiAnalysisRequestId);
+		difyRequest.put("aiAnalysisRequestId", aiAnalysisRequestId);
 		diFyReq.setInputs(difyRequest);
+		return diFyReq;
+	}
+
+	private void saveAiAnalysisRequestLogs(AnalysisReq analysisReq, DiFyReq diFyReq, String aiAnalysisRequestId) {
 		aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder()
 				.aiAnalysisRequestId(aiAnalysisRequestId)
 				.businessRequest(JSONObject.toJSONString(analysisReq.getData()))
@@ -116,8 +125,33 @@ public class AiAnalysisDifyServiceImpl implements AiAnalysisDifyService {
 				.aiAnalysisRequestType(analysisReq.getAiAnalysisRequestType())
 				.callbackUrl(analysisReq.getCallbackUrl())
 				.build());
-		sendMq(analysisDifyTopic, diFyReq);
-		return AnalysisResp.success(analysisReq.getData(),aiAnalysisRequestId);
+	}
+
+	@Override
+	public AnalysisResp query(AnalysisQueryReq analysisQueryReq) {
+		if(null == analysisQueryReq){
+			log.info("请求Ai查询对象为空！");
+			return AnalysisResp.failed("请求Ai查询对象为空！");
+		}
+
+		Optional<AiAnalysisRequestLogs> aiAnalysisRequestLogsOpt = Optional.ofNullable(
+				aiAnalysisRequestLogsService.queryByAiAnalysisRequestId(analysisQueryReq.getAiAnalysisRequestId())
+		);
+		return aiAnalysisRequestLogsOpt.map(logs -> {
+			if ("true".equals(analysisQueryReq.getRetryAnalyze())) {
+				log.info("请求Ai查询 需要重新生成AI解析 aiAnalysisRequestId:{}, retryAnalyze：{}", analysisQueryReq.getAiAnalysisRequestId(), analysisQueryReq.getRetryAnalyze());
+				return aiAnalyze(AnalysisReq.builder()
+						.data(logs.getBusinessRequest())
+						.aiAnalysisRequestType(logs.getAiAnalysisRequestType())
+						.callbackUrl(logs.getCallbackUrl())
+						.build());
+			}
+			return AnalysisResp.success(logs.getBusinessResponse(), logs.getAiAnalysisRequestId());
+		}).orElseGet(() -> {
+			log.info("查询的AI解析不存在！aiAnalysisRequestId:{}", analysisQueryReq.getAiAnalysisRequestId());
+			return AnalysisResp.failed("查询的AI解析不存在！");
+		});
+
 	}

 	Map<String, String> sendDccCorpus(AiAnalysisRequestLogs oldAiAnalysisRequestLogs,String  difyResponse ){
@@ -195,7 +229,7 @@ public class AiAnalysisDifyServiceImpl implements AiAnalysisDifyService {
 	}

 	private void sendMq(String topic, Object message){
-		rocketMqTemplate.asyncSend(callbackTopic, MessageBuilder.withPayload(message).build(),
+		rocketMqTemplate.asyncSend(topic, MessageBuilder.withPayload(message).build(),
 				new SendCallback() {
 					@Override
 					public void onSuccess(SendResult sendResult) {