diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/req/AnalysisQueryReq.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/req/AnalysisQueryReq.java new file mode 100644 index 0000000..5f8bd7d --- /dev/null +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/req/AnalysisQueryReq.java @@ -0,0 +1,27 @@ +package com.volvo.ai.analytic.center.dto.req; + +import lombok.Data; + +import javax.validation.constraints.NotNull; + +/** + * + * @ClassName: AnalysisRequest + * @author: renzhen + * @Description: Ai查询请求参数 + * @date: 2025-04-15 13:39 + */ +@Data +public class AnalysisQueryReq { + + + private Object data; + + // aiId + @NotNull(message = "aiAnalysisRequestId不能为空") + private String aiAnalysisRequestId; + + // 是否重试：默认：否 + private String retryAnalyze; + +} diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/req/AnalysisReq.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/req/AnalysisReq.java new file mode 100644 index 0000000..1c3e557 --- /dev/null +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/req/AnalysisReq.java @@ -0,0 +1,56 @@ +package com.volvo.ai.analytic.center.dto.req; + +import com.alibaba.csp.sentinel.util.StringUtil; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import javax.validation.Valid; +import javax.validation.constraints.NotBlank; +import javax.validation.constraints.NotNull; + +/** + * + * @ClassName: AnalysisRequest + * @author: renzhen + * @Description: Ai解析请求参数 + * @date: 2025-04-15 13:39 + */ +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +public class AnalysisReq { + + // 请求的所有参数 + @Valid + @NotNull(message = "请求语料不能为空") + private Object data; + + // 回调地址 + @Valid + @NotBlank(message = "回调地址不能为空") + private String callbackUrl; + + // aiId + private String aiAnalysisRequestId; + + // 业务类型 + @Valid + @NotBlank(message = "业务类型不能为空") + private String aiAnalysisRequestType; + + public void validate() { + if (null == data){ + throw new IllegalArgumentException("请求语料不能为空"); + } + if (StringUtil.isBlank(aiAnalysisRequestType)){ + throw new IllegalArgumentException("业务类型不能为空"); + } + if (StringUtil.isBlank(callbackUrl)){ + throw new IllegalArgumentException("回调地址不能为空"); + } + } + +} diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/resp/AnalysisDifyResultDTO.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/resp/AnalysisDifyResultDTO.java new file mode 100644 index 0000000..b47b7ea --- /dev/null +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/resp/AnalysisDifyResultDTO.java @@ -0,0 +1,24 @@ +package com.volvo.ai.analytic.center.dto.resp; + +import lombok.Data; + +/** + * + * @ClassName: AnalysisRequest + * @author: renzhen + * @Description: 工作流处理结果 + * @date: 2025-04-15 13:39 + */ +@Data +public class AnalysisDifyResultDTO{ + + // 响应 + private String workflowRunId; + private String workflowAppId; + private String workUserId; + // 分析中心唯一ID + private String aiAnalysisRequestId; + private String difyResponse; + + private String aiAnalysisRequestType; +} diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/resp/AnalysisResp.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/resp/AnalysisResp.java new file mode 100644 index 0000000..21a11ab --- /dev/null +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/dto/resp/AnalysisResp.java @@ -0,0 +1,57 @@ +package com.volvo.ai.analytic.center.dto.resp; + +import com.volvo.common.core.constant.CommonConstants; +import com.volvo.common.core.util.ResultMsg; +import lombok.Data; + +/** + * + * @ClassName: AnalysisRequest + * @author: renzhen + * @Description: Ai解析响应 + * @date: 2025-04-15 13:39 + */ +@Data +public class AnalysisResp { + + // 响应 + private T data; + + // 分析中心唯一ID + private String aiAnalysisRequestId; + + private int code; + private String msg; + + public static AnalysisResp success(String message) { + return (AnalysisResp) analysisResp((Object)null, CommonConstants.SUCCESS, message); + } + public static AnalysisResp success(T data, String aiAnalysisRequestId) { + AnalysisResp analysisResp = new AnalysisResp(); + analysisResp.setAiAnalysisRequestId(aiAnalysisRequestId); + analysisResp.setData(data); + analysisResp.setMsg("ok"); + analysisResp.setCode(CommonConstants.SUCCESS); + return analysisResp; + } + public static AnalysisResp failed(String message) { + return (AnalysisResp) analysisResp((Object)null, CommonConstants.FAIL, message); + } + + private static AnalysisResp analysisResp(T data, int code, String msg) { + AnalysisResp analysisResp = new AnalysisResp(); + analysisResp.setCode(code); + analysisResp.setData(data); + analysisResp.setMsg(msg); + return analysisResp; + } + + private static AnalysisResp analysisResp(T data, String aiAnalysisRequestId) { + AnalysisResp analysisResp = new AnalysisResp(); + analysisResp.setAiAnalysisRequestId(aiAnalysisRequestId); + analysisResp.setData(data); + analysisResp.setMsg("ok"); + analysisResp.setCode(CommonConstants.SUCCESS); + return analysisResp; + } +} diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/entity/AiAnalysisRequestLogs.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/entity/AiAnalysisRequestLogs.java index 9ef50ac..5c4b493 100644 --- a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/entity/AiAnalysisRequestLogs.java +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/entity/AiAnalysisRequestLogs.java @@ -43,6 +43,18 @@ public class AiAnalysisRequestLogs { @TableField("dify_agent_key") private String difyAgentKey; + @TableField("workflow_run_id") + private String workflowRunId; + + @TableField("workflow_app_id") + private String workflowAppId; + + @TableField("work_user_id") + private String workUserId; + + @TableField("callback_url") + private String callbackUrl; + @TableField("is_deleted") @TableLogic private Integer isDeleted; diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/entity/TcBusinessType.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/entity/TcBusinessType.java new file mode 100644 index 0000000..30765f5 --- /dev/null +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/entity/TcBusinessType.java @@ -0,0 +1,70 @@ +package com.volvo.ai.analytic.center.entity; + +import com.baomidou.mybatisplus.annotation.*; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.util.Date; + +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +@TableName("tc_business_type") +public class TcBusinessType { + + @TableId(value = "id", type = IdType.AUTO) + private Long id; + + @TableField("business_request_type") + private String businessRequestType; + + @TableField("business_request_desc") + private String businessRequestDesc; + + @TableField("business_type_topic") + private String businessTypeTopic; // JSON 字符串 + + @TableField("business_type_topic_tag") + private String businessTypeTopicTag; + + @TableField("workflow_api_key") + private String workflowApiKey; // JSON 字符串 + + @TableField("workflow_user") + private String workflowUser; + + @TableField("is_deleted") + @TableLogic + private Integer isDeleted; + + @TableField("versions") + @Version + private Integer versions; + + /** + * 创建者 + */ + @TableField("create_by") + private String createBy; + + /** + * 创建时间 + */ + @TableField("create_time") + private Date createTime; + + /** + * 更新者 + */ + @TableField("update_by") + private String updateBy; + + /** + * 更新时间 + */ + @TableField("update_time") + private Date updateTime; +} diff --git a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/feign/DiFyFeign.java b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/feign/DiFyFeign.java index 46595e2..93a01cb 100644 --- a/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/feign/DiFyFeign.java +++ b/ai-analytic-center-api/src/main/java/com/volvo/ai/analytic/center/feign/DiFyFeign.java @@ -21,7 +21,7 @@ public interface DiFyFeign { @PostMapping(value = "/v1/files/upload" , consumes = MediaType.MULTIPART_FORM_DATA_VALUE) JSONObject fileUpload(@RequestHeader(value = "Authorization") String authorization, @RequestPart("file") MultipartFile file); - @PostMapping(value = "/v1/workflows/run/:{workflowRunId}" , consumes = MediaType.APPLICATION_JSON_VALUE + @GetMapping(value = "/v1/workflows/run/{workflowRunId}" , consumes = MediaType.APPLICATION_JSON_VALUE , produces = MediaType.APPLICATION_JSON_VALUE) JSONObject queryWorkFlowById(@RequestHeader(value = "Authorization") String authorization, @PathVariable("workflowRunId") String workflowRunId); } diff --git a/ai-analytic-center-biz/pom.xml b/ai-analytic-center-biz/pom.xml index 1a1a413..35dc96d 100644 --- a/ai-analytic-center-biz/pom.xml +++ b/ai-analytic-center-biz/pom.xml @@ -232,6 +232,11 @@ 2.3.0 + + + org.springframework.boot + spring-boot-starter-data-redis + org.apache.commons @@ -239,6 +244,18 @@ 1.9 + + + + + + + + + + + + diff --git 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 new file mode 100644 index 0000000..30eceb0 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/AiAnalysisDifyController.java @@ -0,0 +1,81 @@ +package com.volvo.ai.analytic.center.controller; + + +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.ObjectMapper; +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; +import com.volvo.common.core.util.ResultMsg; +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; +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() +@Api(tags = "分析中心接口") +@Slf4j +@RefreshScope +@RequestMapping("analysis") +public class AiAnalysisDifyController { + + @Autowired + private RocketMQTemplate rocketMQTemplate; + + @Autowired + private AiAnalysisDifyService aiAnalysisDifyService; + + + + @PostMapping("/updateByAiId") + @ApiOperation(value = "更新dify结果") + public ResultMsg updateByAiId(@RequestBody String message) { + log.info("updateByAiId message: {}", message); + aiAnalysisDifyService.updateAiDifyResult(message); + return ResultMsg.ok("ok"); + } + + @PostMapping("/aiAnalyze") + @ApiOperation(value = "Ai解析接口") + public AnalysisResp aiAnalyze(@Validated @RequestBody String message) { + log.info("aiAnalyze data: {}",message); + ObjectMapper objectMapper = new ObjectMapper(); + objectMapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true); // 允许未转义的控制字符 + objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); + try { + AnalysisReq analysisReq = objectMapper.readValue(message, AnalysisReq.class); + analysisReq.validate(); + return aiAnalysisDifyService.aiAnalyze(analysisReq); + } catch (JsonProcessingException e) { + log.info("aiAnalyze data error: {}",e); + return AnalysisResp.failed("解析失败"); + } + } + + @PostMapping("/query") + @ApiOperation(value = "Ai解析结果查询") + public AnalysisResp query(@RequestBody @Validated AnalysisQueryReq analysisQueryReq) { + log.info("aiAnalyze query data: {}",analysisQueryReq); + return aiAnalysisDifyService.query(analysisQueryReq); + } + @PostMapping("/callback") + @ApiOperation(value = "Ai解析结果查询") + public AnalysisResp testCallback(@RequestBody AnalysisResp analysisResp) { + log.info("aiAnalyze testCallback data: {}",analysisResp); + return AnalysisResp.success("ok"); + } + + +} + diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/AiDifyResultController.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/AiDifyResultController.java deleted file mode 100644 index afde080..0000000 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/AiDifyResultController.java +++ /dev/null @@ -1,48 +0,0 @@ -package com.volvo.ai.analytic.center.controller; - - -import com.alibaba.fastjson.JSONObject; -import com.volvo.ai.analytic.center.entity.AiAnalysisRequestLogs; -import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService; -import com.volvo.common.core.util.ResultMsg; -import io.swagger.annotations.Api; -import io.swagger.annotations.ApiOperation; -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.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 -@Api(tags = "AiDifyResult") -@RequestMapping("") -@Slf4j -@RefreshScope -public class AiDifyResultController { - - @Autowired - private RocketMQTemplate rocketMQTemplate; - - @Autowired - private AiAnalysisRequestLogsService aiAnalysisRequestLogsService; - - - - @PostMapping("/updateByAiId") - @ApiOperation(value = "更新dify结果") - public ResultMsg updateByAiId(@RequestBody String message) { - JSONObject messageJson = JSONObject.parseObject(message); - AiAnalysisRequestLogs aiAnalysisRequestLogs = new AiAnalysisRequestLogs(); - aiAnalysisRequestLogs.setAiAnalysisRequestId(messageJson.getString("aiAnalysisRequestId")); - aiAnalysisRequestLogs.setDifyResponse(messageJson.getString("difyResponse")); - aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(aiAnalysisRequestLogs); - return ResultMsg.ok("ok"); - } - - -} - diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/RedisTestController.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/RedisTestController.java new file mode 100644 index 0000000..e36818c --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/controller/RedisTestController.java @@ -0,0 +1,80 @@ +package com.volvo.ai.analytic.center.controller; + + +import com.alibaba.fastjson.JSONObject; +import com.volvo.ai.analytic.center.service.AiAnalysisDifyService; +import com.volvo.ai.analytic.center.utils.RedisZSetUtil; +import com.volvo.common.core.util.ResultMsg; +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; +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.data.redis.core.RedisTemplate; +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; + +import java.time.Instant; +import java.util.Set; + + +@RestController() +@Api(tags = "分析中心接口") +@Slf4j +@RefreshScope +@RequestMapping("redis") +public class RedisTestController { + + @Autowired + private RocketMQTemplate rocketMQTemplate; + + @Autowired + private AiAnalysisDifyService aiAnalysisDifyService; + + + @Autowired + private RedisTemplate redisTemplate; + @Autowired + private RedisZSetUtil redisZSetUtil; + + @PostMapping("/testRedis") + public void test(@RequestBody String message) { + String zsetKey = "myZSet"; + JSONObject json = JSONObject.parseObject(message); + + // 添加成员并设置过期时间戳 + long expireTime1 = Instant.now().plusSeconds(30).toEpochMilli(); // 30秒后过期 + long expireTime2 = Instant.now().plusSeconds(60).toEpochMilli(); // 60秒后过期 + redisZSetUtil.addWithExpire(zsetKey, "member1", expireTime1); + redisZSetUtil.addWithExpire(zsetKey, "member2", expireTime2); + // 获取未过期的成员 + Set validMembers = redisZSetUtil.getValidMembers(zsetKey); + log.info("Valid members: " + validMembers); + } + + @PostMapping("/queryRedis") + @ApiOperation(value = "queryRedis") + public ResultMsg queryRedis(@RequestBody String key) { + log.info("aiAnalyze queryRedis : {}", redisTemplate.opsForZSet().zCard(key)); + Set getredisSet = redisTemplate.opsForZSet().range(key, 0, -1); + log.info("queryRedis:{}", getredisSet); + return ResultMsg.ok(getredisSet); + } + @PostMapping("/del") + @ApiOperation(value = "del") + public ResultMsg del(@RequestBody String key) { + log.info("aiAnalyze queryRedis : {}",key); + redisTemplate.opsForZSet().removeRange(key, 0, -1); + Set getRedisSet = redisTemplate.opsForZSet().range(key, 0, -1); + log.info("removeRange:{}" , getRedisSet); + redisTemplate.delete(key); + Set getredisSet2 = redisTemplate.opsForZSet().range(key, 0, -1); + log.info("delete:{}", getredisSet2); + return ResultMsg.ok(getredisSet2); + } + +} + diff --git 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 new file mode 100644 index 0000000..dad7e3e --- /dev/null +++ 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 workflowRunIdList = Arrays.asList(param.split(",")); + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.in(AiAnalysisRequestLogs::getWorkflowRunId, workflowRunIdList); + queryWrapper.eq(AiAnalysisRequestLogs::getIsDeleted, "0"); + List 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(); + } + + + +} diff --git 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 deleted file mode 100644 index 1abfadd..0000000 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/job/DccCorpusJob.java +++ /dev/null @@ -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(); - } - - - -} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mapper/TcBusinessTypeMapper.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mapper/TcBusinessTypeMapper.java new file mode 100644 index 0000000..d381fdd --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mapper/TcBusinessTypeMapper.java @@ -0,0 +1,11 @@ +package com.volvo.ai.analytic.center.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.volvo.ai.analytic.center.entity.TcBusinessType; +import org.apache.ibatis.annotations.Mapper; + +@Mapper +public interface TcBusinessTypeMapper extends BaseMapper { + + +} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/AnalysisDifyCallbackMqConsumer.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/AnalysisDifyCallbackMqConsumer.java new file mode 100644 index 0000000..c8aa099 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/AnalysisDifyCallbackMqConsumer.java @@ -0,0 +1,91 @@ + +package com.volvo.ai.analytic.center.mq; + +import com.alibaba.fastjson.JSONObject; +import com.fasterxml.jackson.databind.ObjectMapper; +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.service.AiAnalysisRequestLogsService; +import com.volvo.ai.analytic.center.service.DiFyService; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; +import org.apache.rocketmq.spring.core.RocketMQListener; +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.http.HttpEntity; +import org.springframework.http.HttpHeaders; +import org.springframework.http.ResponseEntity; +import org.springframework.stereotype.Component; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.client.RestTemplate; + +/** + * @ClassName AnalysisDifyMqConsumer + * @Description AI解析MQ-Callback处理 + * @Author renzhen + * @Date 2025-03-04 10:18 + * @Version 1.0 + **/ + +@Slf4j +@Component +@RefreshScope +@RestController +@RocketMQMessageListener(topic = "${rocketmq.consumer.analysisDify.callbackTopic}",consumerGroup = "${rocketmq.consumer.analysisDify.callbackGroup}", + instanceName = "analysisDifyCallbackMqConsumer", + consumeThreadNumber = 40, + enableMsgTrace = true) +public class AnalysisDifyCallbackMqConsumer implements RocketMQListener { + + + @Autowired + private AiAnalysisRequestLogsService aiAnalysisRequestLogsService; + + @Value("${dify.corpus.checkDccRepeat}") + private String checkDccRepeat; + + @Autowired + private DiFyService diFyService; + + @Autowired + private RestTemplate restTemplate; + + private final ObjectMapper objectMapper = new ObjectMapper(); + + @Override + public void onMessage(MessageExt messageExt) { + long startTime = System.currentTimeMillis(); + try { + log.info("analysisDifyCallbackMqConsumer 当前线程: {}, 线程ID: {}", Thread.currentThread().getName(), Thread.currentThread().getId()); + String message = new String(messageExt.getBody()); + log.info("analysisDifyCallbackMqConsumer message: " + message); + AnalysisDifyResultDTO analysisRestDto = JSONObject.parseObject(message, AnalysisDifyResultDTO.class); + AiAnalysisRequestLogs aiAnalysisRequestLogs = aiAnalysisRequestLogsService.queryByAiAnalysisRequestId(analysisRestDto.getAiAnalysisRequestId()); + AnalysisResp analysisResp = new AnalysisResp(); + analysisResp.setAiAnalysisRequestId(analysisRestDto.getAiAnalysisRequestId()); + JSONObject difyJson = JSONObject.parseObject(analysisRestDto.getDifyResponse()); + String outputs = difyJson.getString("outputs"); + analysisResp.setData(outputs); + + HttpHeaders headers = new HttpHeaders(); + headers.set("Content-Type", "application/json"); + // 封装请求体和请求头 + HttpEntity requestEntity = new HttpEntity<>(analysisResp, headers); + ResponseEntity response = restTemplate.postForEntity(aiAnalysisRequestLogs.getCallbackUrl(), requestEntity, String.class); // 响应类型); + if (response.getStatusCode().is2xxSuccessful()) { + log.info("analysisDifyCallbackMqConsumer aiAnalysisRequestId：{},回调请求成功，url:{}: " ,analysisResp.getAiAnalysisRequestId(), aiAnalysisRequestLogs.getCallbackUrl()); + } else { + log.info("analysisDifyCallbackMqConsumer aiAnalysisRequestId：{},回调请求失败，url:{}: " ,analysisResp.getAiAnalysisRequestId(), aiAnalysisRequestLogs.getCallbackUrl()); + } + aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(AiAnalysisRequestLogs.builder().aiAnalysisRequestId(analysisResp.getAiAnalysisRequestId()).businessResponse(outputs).build()); + log.info(" analysisDifyCallbackMqConsumer，耗时：{}", System.currentTimeMillis() - startTime); + } catch (Exception e) { + log.info(" analysisDifyCallbackMqConsumer mq 处理失败：{}", e.getMessage()); + } + } + +} + diff --git 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 new file mode 100644 index 0000000..f89d9cc --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/AnalysisDifyMqConsumer.java @@ -0,0 +1,141 @@ + +package com.volvo.ai.analytic.center.mq; + +import com.alibaba.fastjson.JSONObject; +import com.volvo.ai.analytic.center.dto.req.DiFyReq; +import com.volvo.ai.analytic.center.entity.AiAnalysisErrors; +import com.volvo.ai.analytic.center.service.AiAnalysisErrorsService; +import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService; +import com.volvo.ai.analytic.center.service.DiFyService; +import com.volvo.ai.analytic.center.utils.ConstantStr; +import com.volvo.ai.analytic.center.utils.RedisCounterRateLimiter; +import com.volvo.ai.analytic.center.utils.RedisLockService; +import com.volvo.ai.analytic.center.utils.RedisZSetUtil; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; +import org.apache.rocketmq.spring.core.RocketMQListener; +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.data.redis.core.RedisTemplate; +import org.springframework.stereotype.Component; +import org.springframework.web.bind.annotation.RestController; + +import java.time.Instant; +import java.util.concurrent.CompletableFuture; + +/** + * @ClassName AnalysisDifyMqConsumer + * @Description AI解析 MQ处理 + * @Author renzhen + * @Date 2025-03-04 10:18 + * @Version 1.0 + **/ + +@Slf4j +@Component +@RefreshScope +@RestController +@RocketMQMessageListener(topic = "${rocketmq.consumer.analysisDify.topic}",consumerGroup = "${rocketmq.consumer.analysisDify.group}", + instanceName = "analysisDifyMqConsumer", + consumeThreadNumber = 20, + enableMsgTrace = true) +public class AnalysisDifyMqConsumer implements RocketMQListener { + + + @Autowired + private AiAnalysisRequestLogsService aiAnalysisRequestLogsService; + + @Value("${dify.corpus.checkDccRepeat}") + private String checkDccRepeat; + + @Value("${rocketmq.consumer.analysisDify.difyLimit}") + private int difyLimit; + + @Value("${rocketmq.consumer.analysisDify.expire}") + private Long expire; + @Autowired + private DiFyService diFyService; + + @Autowired + private RedisCounterRateLimiter redisCounterRateLimiter; + + @Autowired + private AiAnalysisErrorsService aiAnalysisErrorsService; + + @Autowired + private RedisTemplate redisTemplate; + + @Autowired + private RedisZSetUtil redisZSetUtil; + + @Autowired + private RedisLockService redisLockService; + @Override + public void onMessage(MessageExt messageExt) { + log.info("analysisDifyMqConsumer 当前线程: {}, 线程ID: {}", Thread.currentThread().getName(), Thread.currentThread().getId()); + long startTime = System.currentTimeMillis(); + String lockKey = "LOCK_PREFIX:" + ConstantStr.DIFY_COUNTERRATELIMIT; +// RLock lock = redissonClient.getLock("LOCK_PREFIX:" + ConstantStr.DIFY_COUNTERRATELIMIT); + try { + + boolean isLocked = redisLockService.tryLock(lockKey,ConstantStr.DIFY_COUNTERRATELIMIT, expire); // 不等待，立即尝试获取 + if (isLocked) { + Long count = redisZSetUtil.zCard(ConstantStr.DIFY_COUNTERRATELIMIT); + log.info("redis计数数量: {}", count); + if(count>=difyLimit){ + log.info("analysisDifyMqConsumer 请求dify超过基数 ：{}，稍后请求: " + redisZSetUtil.getValidMembers(ConstantStr.DIFY_COUNTERRATELIMIT)); + throw new RuntimeException("请求dify超过基数 " ); + } + String message = new String(messageExt.getBody()); + log.info("analysisDifyMqConsumer message: " + message); + DiFyReq difyReq = JSONObject.parseObject(message, DiFyReq.class); + JSONObject difyRequest = JSONObject.parseObject(JSONObject.toJSONString(difyReq.getInputs()), JSONObject.class); + String aiAnalysisRequestId = difyRequest.getString("aiAnalysisRequestId"); + long expireTime = Instant.now().plusSeconds(expire).toEpochMilli(); // 60秒后过期 + redisZSetUtil.addWithExpire(ConstantStr.DIFY_COUNTERRATELIMIT, aiAnalysisRequestId, expireTime); + + CompletableFuture future = diFyService.asyncExecuteDifyFlow(difyReq); + future.thenAccept(result -> { + log.info("异步处理asyncExecuteDifyFlow完成aiAnalysisRequestId: {},处理结果：{}", aiAnalysisRequestId, result); + JSONObject data = result.getJSONObject("data"); + + if(null == result || data.get("status").equals("failed")){ + aiAnalysisErrorsService.saveAiAnalysisErrors(AiAnalysisErrors.builder() + .aiAnalysisRequestId(aiAnalysisRequestId) + .aiAnalysisRequestType(difyRequest.getString("aiAnalysisRequestType")) + .aiAnalysisErrorHandlingStatus("0") + .aiAnalysisErrorMessage(data.getString("error")) + .build()); + } + }).exceptionally(ex -> { + aiAnalysisErrorsService.saveAiAnalysisErrors(AiAnalysisErrors.builder() + .aiAnalysisRequestId(aiAnalysisRequestId) + .aiAnalysisRequestType(difyRequest.getString("aiAnalysisRequestType")) + .aiAnalysisErrorHandlingStatus("0") + .aiAnalysisErrorMessage(ex.getMessage()) + .build()); + log.error("异步处理asyncExecuteDifyFlow 失败aiAnalysisRequestId: {} ,{}", aiAnalysisRequestId,ex.getMessage()); + return null; + }); + redisLockService.releaseLock(lockKey, ConstantStr.DIFY_COUNTERRATELIMIT); + log.info("analysisDifyMqConsumer处理完成，耗时：{}", System.currentTimeMillis() - startTime); + }else{ + log.info("redis锁未获取到 "); + throw new RuntimeException("锁超时 " ); + } + + } catch (Exception e){ + log.error("analysisDifyMqConsumer 异常：{}", e.getMessage()); + throw new RuntimeException("请求dify超过基数 " ); + }finally { + redisLockService.releaseLock(lockKey, ConstantStr.DIFY_COUNTERRATELIMIT); + // 建议记录解锁日志 + log.info("释放锁成功，锁KEY: {}", ConstantStr.DIFY_COUNTERRATELIMIT); + + } + } + +} + diff --git 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 new file mode 100644 index 0000000..3663dd4 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/AiAnalysisDifyService.java @@ -0,0 +1,19 @@ +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.AnalysisDifyResultDTO; +import com.volvo.ai.analytic.center.dto.resp.AnalysisResp; + +public interface AiAnalysisDifyService { + + public boolean updateAiDifyResult(String message); + + + AnalysisResp aiAnalyze(AnalysisReq analysisReq); + + AnalysisResp query(AnalysisQueryReq analysisQueryReq); + + + +} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/DiFyService.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/DiFyService.java index 7159e60..3bfba4a 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/DiFyService.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/DiFyService.java @@ -3,6 +3,8 @@ package com.volvo.ai.analytic.center.service; import com.alibaba.fastjson.JSONObject; import com.volvo.ai.analytic.center.dto.req.DiFyReq; +import java.util.concurrent.CompletableFuture; + public interface DiFyService { @@ -11,4 +13,6 @@ public interface DiFyService { public JSONObject executeDifyFlow(DiFyReq diFyReq, String businessType, String businessData, String aiAnalysisRequestId); public JSONObject executeDifyFlow(DiFyReq diFyReq); + public CompletableFuture asyncExecuteDifyFlow(DiFyReq diFyReq); + } diff --git 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 new file mode 100644 index 0000000..b9e01d0 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/AiAnalysisDifyServiceImpl.java @@ -0,0 +1,204 @@ +package com.volvo.ai.analytic.center.service.impl; + +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +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.mapper.TcBusinessTypeMapper; +import com.volvo.ai.analytic.center.mapper.TmTelephoneCorpusMapper; +import com.volvo.ai.analytic.center.service.AiAnalysisDifyService; +import com.volvo.ai.analytic.center.service.AiAnalysisRequestLogsService; +import com.volvo.ai.analytic.center.service.TmTelephoneCorpusService; +import com.volvo.ai.analytic.center.utils.AiAnalysisUtils; +import com.volvo.ai.analytic.center.utils.ConstantStr; +import com.volvo.ai.analytic.center.utils.RedisCounterRateLimiter; +import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.spring.core.RocketMQTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.stereotype.Service; + +import javax.annotation.Resource; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Collectors; + +@Slf4j +@Service +public class AiAnalysisDifyServiceImpl implements AiAnalysisDifyService { + + @Autowired + private AiAnalysisRequestLogsService aiAnalysisRequestLogsService; + + @Autowired + private TmTelephoneCorpusMapper tmTelephoneCorpusMapper; + + @Autowired + private TmTelephoneCorpusService tmTelephoneCorpusService; + + @Autowired + private TcBusinessTypeMapper tcBusinessTypeMapper; + + @Value("${rocketmq.producer.analysisDify.topic}") + private String analysisDifyTopic; + + @Value("${rocketmq.producer.analysisDify.callbackTopic}") + private String callbackTopic; + @Resource + private RocketMQTemplate rocketMqTemplate; + + @Autowired + private RedisCounterRateLimiter redisCounterRateLimiter; + + @Autowired + private RedisTemplate redisTemplate; + @Override + public boolean updateAiDifyResult(String message) { + + + if(StringUtils.isNotEmpty(message)){ + AnalysisDifyResultDTO analysisResp = JSONObject.parseObject(message, AnalysisDifyResultDTO.class); + // 计数-1 + Long removedCount = redisTemplate.opsForZSet().remove(ConstantStr.DIFY_COUNTERRATELIMIT, analysisResp.getAiAnalysisRequestId()); + log.info("redisdecrement计数数量: " + removedCount); + + AiAnalysisRequestLogs oldAiAnalysisRequestLogs = Optional.ofNullable(aiAnalysisRequestLogsService.queryByAiAnalysisRequestId(analysisResp.getAiAnalysisRequestId())) + .orElseThrow(() -> new IllegalArgumentException("AiAnalysisRequestId查询对象为空！")); + + AiAnalysisRequestLogs aiAnalysisRequestLogs = new AiAnalysisRequestLogs(); + aiAnalysisRequestLogs.setAiAnalysisRequestId(analysisResp.getAiAnalysisRequestId()); + aiAnalysisRequestLogs.setDifyResponse(analysisResp.getDifyResponse()); + aiAnalysisRequestLogs.setWorkflowRunId(analysisResp.getWorkflowRunId()); + aiAnalysisRequestLogs.setWorkflowAppId(analysisResp.getWorkflowRunId()); + aiAnalysisRequestLogs.setWorkUserId(analysisResp.getWorkUserId()); + + JSONObject difyJson = JSONObject.parseObject(analysisResp.getDifyResponse()); + aiAnalysisRequestLogs.setBusinessResponse(difyJson.getString("outputs")); + aiAnalysisRequestLogsService.saveAiAnalysisRequestLogs(aiAnalysisRequestLogs); + + if(StringUtils.isNotEmpty(oldAiAnalysisRequestLogs.getCallbackUrl())){ + // 发送 mq + sendMq(callbackTopic, analysisResp); + } + return true; + } + + return false; + } + + @Override + public AnalysisResp aiAnalyze(AnalysisReq analysisReq) { + + Optional.ofNullable(analysisReq).orElseThrow(() -> { + log.info("请求Ai解析对象为空！"); + return new IllegalArgumentException("请求Ai解析对象为空！"); + }); + + String aiAnalysisRequestId = StringUtils.isEmpty(analysisReq.getAiAnalysisRequestId())? AiAnalysisUtils.getAiAnalysisRequestId(analysisReq.getAiAnalysisRequestType()):analysisReq.getAiAnalysisRequestId(); + + Map queryTcBusinessType = queryTcBusinessType(); + 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.setFlowId(tcBusinessType.getWorkflowApiKey()); + JSONObject difyRequest = JSONObject.parseObject(JSONObject.toJSONString(analysisReq.getData()), JSONObject.class); + difyRequest.put("aiAnalysisRequestId", aiAnalysisRequestId); + difyRequest.put("aiAnalysisRequestType", tcBusinessType.getBusinessRequestType() ); + 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())) + .difyAgentKey(diFyReq.getFlowId()) + .difyRequest(JSON.toJSONString(diFyReq)) + .aiAnalysisRequestType(analysisReq.getAiAnalysisRequestType()) + .callbackUrl(analysisReq.getCallbackUrl()) + .build()); + } + + @Override + public AnalysisResp query(AnalysisQueryReq analysisQueryReq) { + if(null == analysisQueryReq){ + log.info("请求Ai查询对象为空！"); + return AnalysisResp.failed("请求Ai查询对象为空！"); + } + + Optional 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解析不存在！"); + }); + + } + + + + public Map queryTcBusinessType() { + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(TcBusinessType::getIsDeleted, "0"); + List tcBusinessTypeList =tcBusinessTypeMapper.selectList(queryWrapper); + return tcBusinessTypeList.stream() + .collect(Collectors.toMap( + TcBusinessType::getBusinessRequestType, + tcBusinessType -> tcBusinessType + )); + } + + + + private void sendMq(String topic, Object message){ + rocketMqTemplate.asyncSend(topic, MessageBuilder.withPayload(message).build(), + new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + log.info("请求AI解析发送MQ成功消息体:{}", message); + } + @Override + public void onException(Throwable e) { + log.error("请求AI解析发送MQ异常消息体:{}, 异常：", message, e); + } + }, 10000); + } + + +} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/DiFyServiceImpl.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/DiFyServiceImpl.java index dce5453..716c3ce 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/DiFyServiceImpl.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/service/impl/DiFyServiceImpl.java @@ -17,6 +17,7 @@ import org.springframework.stereotype.Service; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CompletableFuture; @Slf4j @Service @@ -117,4 +118,13 @@ public class DiFyServiceImpl implements DiFyService{ return data; } + @Override + public CompletableFuture asyncExecuteDifyFlow(DiFyReq diFyReq) { + Map map = new HashMap<>(); + map.put("inputs",diFyReq.getInputs()); + map.put("user",diFyReq.getUser()); + return CompletableFuture.supplyAsync(() -> diFyFeign.runWorkflows("Bearer "+diFyReq.getFlowId(),map)); + + } + } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/ConstantStr.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/ConstantStr.java index 6ef828c..3fcf89d 100644 --- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/ConstantStr.java +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/ConstantStr.java @@ -21,4 +21,6 @@ public class ConstantStr { public static final String customerFlowIds = "customerFlowIds"; public static final String aiAnalysisRequestId = "aiAnalysisRequestId"; + + public static final String DIFY_COUNTERRATELIMIT = "DIFY_COUNTERRATELIMIT"; } diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisCounterRateLimiter.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisCounterRateLimiter.java new file mode 100644 index 0000000..6e9f906 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisCounterRateLimiter.java @@ -0,0 +1,80 @@ +package com.volvo.ai.analytic.center.utils; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.script.DefaultRedisScript; +import org.springframework.stereotype.Service; + +import java.util.Collections; + +@Service +public class RedisCounterRateLimiter { + private final StringRedisTemplate redisTemplate; + + @Autowired + public RedisCounterRateLimiter(StringRedisTemplate redisTemplate) { + this.redisTemplate = redisTemplate; + } + + /** + * 增加计数并检查是否超过限制 + * @param key 限流key + * @param limit 最大限制数 + * @param expire 过期时间(秒) + * @return true-允许请求; false-超过限制 + */ + public boolean incrementAndCheck(String key, int limit, long expire) { + // 使用Lua脚本保证原子性 + String luaScript = + "local current = redis.call('GET', KEYS[1]) or '0'\n" + + "local num = tonumber(current)\n" + + "if num >= tonumber(ARGV[1]) then\n" + + " return 0\n" + + "else\n" + + " redis.call('INCR', KEYS[1])\n" + + " if num == 0 then\n" + + " redis.call('EXPIRE', KEYS[1], ARGV[2])\n" + + " end\n" + + " return 1\n" + + "end"; + + DefaultRedisScript script = new DefaultRedisScript<>(); + script.setScriptText(luaScript); + script.setResultType(Long.class); + + Long result = redisTemplate.execute(script, Collections.singletonList(key), + String.valueOf(limit), String.valueOf(expire)); + + return result != null && result == 1L; + } + + /** + * 减少计数 + * @param key 限流key + */ + public void decrement(String key) { + // 使用Lua脚本防止减到负数 + String luaScript = + "local current = redis.call('GET', KEYS[1]) or '0'\n" + + "if tonumber(current) > 0 then\n" + + " redis.call('DECR', KEYS[1])\n" + + "end\n" + + "return 1"; + + DefaultRedisScript script = new DefaultRedisScript<>(); + script.setScriptText(luaScript); + script.setResultType(Long.class); + + redisTemplate.execute(script, Collections.singletonList(key)); + } + + /** + * 获取当前计数 + * @param key 限流key + * @return 当前计数值 + */ + public int getCurrentCount(String key) { + String count = redisTemplate.opsForValue().get(key); + return count == null ? 0 : Integer.parseInt(count); + } +} \ No newline at end of file diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisLockService.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisLockService.java new file mode 100644 index 0000000..58b91ca --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisLockService.java @@ -0,0 +1,43 @@ +package com.volvo.ai.analytic.center.utils; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.stereotype.Service; + +import java.util.concurrent.TimeUnit; + +@Service +public class RedisLockService { + + @Autowired + private StringRedisTemplate stringRedisTemplate; + + /** + * 尝试获取分布式锁 + * + * @param lockKey 锁的键名 + * @param requestId 请求标识（用于解锁时验证） + * @param expireTime 锁的过期时间（秒） + * @return 是否获取锁成功 + */ + public boolean tryLock(String lockKey, String requestId, long expireTime) { + return stringRedisTemplate.opsForValue() + .setIfAbsent(lockKey, requestId, expireTime, TimeUnit.SECONDS); + } + + /** + * 释放分布式锁 + * + * @param lockKey 锁的键名 + * @param requestId 请求标识（用于验证） + * @return 是否释放锁成功 + */ + public boolean releaseLock(String lockKey, String requestId) { + String currentValue = stringRedisTemplate.opsForValue().get(lockKey); + if (currentValue != null && currentValue.equals(requestId)) { + stringRedisTemplate.delete(lockKey); + return true; + } + return false; + } +} diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisZSetUtil.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisZSetUtil.java new file mode 100644 index 0000000..14e9e10 --- /dev/null +++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/utils/RedisZSetUtil.java @@ -0,0 +1,49 @@ +package com.volvo.ai.analytic.center.utils; + +import org.springframework.beans.factory.annotation.Autowired; + + +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +import java.time.Instant; +import java.util.Set; + +@Service +public class RedisZSetUtil { + +// @Autowired + private RedisTemplate redisTemplate; + @Autowired + public RedisZSetUtil(RedisTemplate redisTemplate) { + this.redisTemplate = redisTemplate; + } + /** + * 添加成员到 ZSet，并设置过期时间戳 + */ + public void addWithExpire(String zsetKey, String member, long expireTimeMillis) { + redisTemplate.opsForZSet().add(zsetKey, member, expireTimeMillis); + } + + /** + * 清理过期的成员 + */ + @Scheduled(fixedRate = 1000) + public void cleanupExpiredMembers() { + long now = Instant.now().toEpochMilli(); + redisTemplate.opsForZSet().removeRangeByScore(ConstantStr.DIFY_COUNTERRATELIMIT, 0, now); + } + + /** + * 获取未过期的成员 + */ + public Set getValidMembers(String zsetKey) { + long now = Instant.now().toEpochMilli(); + return redisTemplate.opsForZSet().rangeByScore(zsetKey, now, Double.MAX_VALUE); + } + + public Long zCard(String zsetKey) { + return redisTemplate.opsForZSet().zCard(zsetKey); + } +} \ No newline at end of file