流媒体 新接口

yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/api/ai/AiApiImpl.java

@@ -65,7 +65,7 @@ public class AiApiImpl implements AiApi {
         
         return workflowRunService.getDifyStreaming(inputs,
                 SecurityFrameworkUtils.getLoginUserId().toString(),
-                apiKey, query, null)
+                apiKey, query, conversationId)
                 ;
         
     }

yudao-module-infra/yudao-module-infra-biz/src/main/java/cn/iocoder/yudao/module/infra/service/aipromptmanagement/utils/DifyApiUtils.java

@@ -1,5 +1,6 @@
 package cn.iocoder.yudao.module.infra.service.aipromptmanagement.utils;
 
+import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONObject;
 import cn.iocoder.yudao.module.infra.api.ai.dto.StreamResponseDTO;
 import cn.iocoder.yudao.module.infra.service.aipromptmanagement.StreamResponse;
@@ -92,7 +93,9 @@ public class DifyApiUtils {
         requestBody.put("inputs", inputs);
         requestBody.put("response_mode", "streaming");
         requestBody.put("user", user);
-//        requestBody.put("conversation_id", conversationId);
+        if (StrUtil.isNotBlank(conversationId)) {
+            requestBody.put("conversation_id", conversationId);
+        }
         requestBody.put("query", query);
         Flux<StreamResponseDTO> rs = webClient.post()
                 .uri("/v1/chat-messages")

yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/controller/admin/kefu/vo/conversation/KeFuConversationRespVO.java

@@ -50,4 +50,6 @@ public class KeFuConversationRespVO {
     private Integer relUnreadMessageCount;
     private LocalDateTime relLastMessageTime;
     private String relLastMessageContent;
+    
+    private String difyConversationId;
 }

yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/controller/app/kefu/AppKeFuMessageController.java

@@ -3,6 +3,7 @@ package cn.iocoder.yudao.module.promotion.controller.app.kefu;
 import cn.iocoder.yudao.framework.common.enums.UserTypeEnum;
 import cn.iocoder.yudao.framework.common.pojo.CommonResult;
 import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
+import cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils;
 import cn.iocoder.yudao.module.promotion.controller.admin.kefu.vo.message.KeFuMessageListReqVO;
 import cn.iocoder.yudao.module.promotion.controller.admin.kefu.vo.message.KeFuMessageRespVO;
 import cn.iocoder.yudao.module.promotion.controller.app.kefu.vo.message.AppKeFuMessageSendReqVO;
@@ -14,6 +15,7 @@ import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.annotation.Resource;
+import jakarta.servlet.http.HttpServletRequest;
 import jakarta.validation.Valid;
 import org.springframework.http.MediaType;
 import org.springframework.validation.annotation.Validated;
@@ -68,10 +70,11 @@ public class AppKeFuMessageController {
     
     @PostMapping(value = "/sendStream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
     @Operation(summary = "发送消息（流式）", description = "流式返回，响应较快")
-    public Flux<String> sendStream(@Valid @RequestBody AppKeFuMessageSendReqVO sendReqVO) {
+    public Flux<String> sendStream(@Valid @RequestBody AppKeFuMessageSendReqVO sendReqVO, HttpServletRequest request) {
         sendReqVO.setSenderId(getLoginUserId()).setSenderType(UserTypeEnum.MEMBER.getValue()); // 设置用户编号和类型
-        
-        
+        String token = SecurityFrameworkUtils.obtainAuthorization(request,
+                "Authorization", "token");
+        sendReqVO.setToken(token);
         return kefuMessageService.sendStream(sendReqVO);
     }
     

yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/controller/app/kefu/vo/message/AppKeFuMessageSendReqVO.java

@@ -15,6 +15,7 @@ public class AppKeFuMessageSendReqVO {
     @Schema(description = "消息", requiredMode = Schema.RequiredMode.REQUIRED)
     @NotEmpty(message = "消息不能为空")
     private String content;
+    private String token;
 
     // ========== 后端设置的参数，前端无需传递 ==========
 

yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/dal/dataobject/kefu/KeFuConversationDO.java

@@ -88,4 +88,5 @@ public class KeFuConversationDO extends BaseDO {
     private Integer relUnreadMessageCount;
     private LocalDateTime relLastMessageTime;
     private String relLastMessageContent;
+    private String difyConversationId;
 }

yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/service/kefu/KeFuConversationService.java

@@ -37,6 +37,8 @@ public interface KeFuConversationService {
      * @param updateReqVO 请求
      */
     void updateConversationPinnedByAdmin(KeFuConversationUpdatePinnedReqVO updateReqVO);
+    
+    void updateConversation(KeFuConversationRespVO updateReqVO);
 
     /**
      * 更新会话客服消息冗余信息

yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/service/kefu/KeFuConversationServiceImpl.java

@@ -1,6 +1,7 @@
 package cn.iocoder.yudao.module.promotion.service.kefu;
 
 import cn.hutool.core.util.StrUtil;
+import cn.iocoder.yudao.framework.common.util.object.BeanUtils;
 import cn.iocoder.yudao.framework.mybatis.core.query.LambdaQueryWrapperX;
 import cn.iocoder.yudao.framework.security.core.util.SecurityFrameworkUtils;
 import cn.iocoder.yudao.module.promotion.controller.admin.kefu.vo.conversation.KeFuConversationRespVO;
@@ -78,6 +79,15 @@ public class KeFuConversationServiceImpl implements KeFuConversationService {
         // 更新管理员会话置顶状态
         conversationMapper.updateById(new KeFuConversationDO().setId(updateReqVO.getId()).setAdminPinned(updateReqVO.getAdminPinned()));
     }
+    
+    @Override
+    public void updateConversation(KeFuConversationRespVO updateReqVO) {
+        // 校验存在
+        validateKefuConversationExists(updateReqVO.getId());
+        
+        // 更新管理员会话置顶状态
+        conversationMapper.updateById(BeanUtils.toBean(updateReqVO, KeFuConversationDO.class));
+    }
 
     @Override
     @Transactional(rollbackFor = Exception.class)

yudao-module-mall/yudao-module-promotion-biz/src/main/java/cn/iocoder/yudao/module/promotion/service/kefu/KeFuMessageServiceImpl.java

@@ -13,6 +13,7 @@ import cn.iocoder.yudao.module.infra.api.ai.AiApi;
 import cn.iocoder.yudao.module.infra.api.websocket.WebSocketSenderApi;
 import cn.iocoder.yudao.module.member.api.user.MemberUserApi;
 import cn.iocoder.yudao.module.member.api.user.dto.MemberUserRespDTO;
+import cn.iocoder.yudao.module.promotion.controller.admin.kefu.vo.conversation.KeFuConversationRespVO;
 import cn.iocoder.yudao.module.promotion.controller.admin.kefu.vo.message.KeFuMessageListReqVO;
 import cn.iocoder.yudao.module.promotion.controller.admin.kefu.vo.message.KeFuMessageRespVO;
 import cn.iocoder.yudao.module.promotion.controller.admin.kefu.vo.message.KeFuMessageSendReqVO;
@@ -113,11 +114,7 @@ public class KeFuMessageServiceImpl implements KeFuMessageService {
     public Long sendMessage(AppKeFuMessageSendReqVO sendReqVO) {
         // 1.1 设置会话编号
         KeFuMessageDO kefuMessage = BeanUtils.toBean(sendReqVO, KeFuMessageDO.class);
-        if (ObjUtil.isNull(kefuMessage.getConversationId())) {
-            KeFuConversationDO conversation = conversationService.getOrCreateConversation(sendReqVO.getSenderId(),
-                    sendReqVO.getRelUserId());
-            kefuMessage.setConversationId(conversation.getId());
-        }
+        
         kefuMessage.setReceiverId(sendReqVO.getRelUserId());
         // 1.2 保存消息
         keFuMessageMapper.insert(kefuMessage);
@@ -140,18 +137,38 @@ public class KeFuMessageServiceImpl implements KeFuMessageService {
         // 1.1 设置会话编号
         String apiKey = DictFrameworkUtils.parseDictDataValue("ai_key", "多轮对话");
         Map<String, Object> inputs = new HashMap<>();
-        inputs.put("type", "活林木");
+        if (sendReqVO.getRelUserId() == 1L) {
+            inputs.put("type", "系统客服");
+        } else if (sendReqVO.getRelUserId() == 2L) {
+            inputs.put("type", "贷款专家");
+        } else if (sendReqVO.getRelUserId() == 3L) {
+            inputs.put("type", "律师专家");
+        } else {
+            inputs.put("type", "其他");
+        }
+        inputs.put("token", sendReqVO.getToken());
+        String conversationId;
+        KeFuMessageDO kefuMessage = BeanUtils.toBean(sendReqVO, KeFuMessageDO.class);
+        if (ObjUtil.isNull(kefuMessage.getConversationId())) {
+            KeFuConversationDO conversation = conversationService.getOrCreateConversation(sendReqVO.getSenderId(),
+                    sendReqVO.getRelUserId());
+            kefuMessage.setConversationId(conversation.getId());
+            sendReqVO.setConversationId(conversation.getId());
+            conversationId = conversation.getDifyConversationId();
+        } else {conversationId = "";}
         sendMessage(sendReqVO);
         
+        
         AppKeFuMessageSendReqVO aiSendReqVO = sendReqVO;
         aiSendReqVO.setRelUserId(SecurityFrameworkUtils.getLoginUserId());
         aiSendReqVO.setSenderId(aiSendReqVO.getRelUserId());
         aiSendReqVO.setContentType(11);
         Long aiId = sendMessage(aiSendReqVO);
         StringBuffer contentBuffer = new StringBuffer();
+        StringBuffer contentBufferCon = new StringBuffer();
         return aiApi.getDifyMessageStreaming(inputs,
                         SecurityFrameworkUtils.getLoginUserId().toString(),
-                        apiKey, sendReqVO.getContent(), "132")
+                        apiKey, sendReqVO.getContent(), conversationId)
                 .flatMap(response -> {
                     log.info("流式结果：" + response.toString());
                     if (response.getEvent().equals("message")) {
@@ -164,6 +181,7 @@ public class KeFuMessageServiceImpl implements KeFuMessageService {
                     }
                     if (response.getEvent().equals("message_end")) {
                         log.info("进入message_end");
+                        contentBufferCon.append(response.getConversation_id());
                     }
                     return Flux.empty(); // 如果不是 workflow_finished 或 message_end，返回空 Flux
                 }).doOnComplete(() -> {
@@ -173,7 +191,14 @@ public class KeFuMessageServiceImpl implements KeFuMessageService {
                     // 1.2 保存消息
                     TenantUtils.executeIgnore(() ->
                             keFuMessageMapper.updateById(aiMessage));
-                    
+                    if (conversationId == null) {
+                        KeFuConversationRespVO updatePinnedReqVO = new KeFuConversationRespVO();
+                        updatePinnedReqVO.setId(sendReqVO.getConversationId());
+                        updatePinnedReqVO.setDifyConversationId(contentBufferCon.toString());
+                        TenantUtils.executeIgnore(() ->
+                                conversationService.updateConversation(updatePinnedReqVO));
+                        
+                    }
                 })
                 .onErrorResume(e -> {
                     log.error("流式处理出错：", e);