fix: 实现知识库与RAGFlow双向同步（知识库名称/简介的修改、知识库删除、文档上传/删除），移除冷却机制确保实时生效，修复文档影子表UNIQUE约束冲突

2026-07-22 15:13:55 +08:00 · 2026-04-20 16:49:24 +08:00
parent 6ce4503b22
commit 0681550bab
6 changed files with 191 additions and 117 deletions
@@ -188,10 +188,4 @@ public class RedisKeys {
        return "ota:upload:count:" + username;
    }

-    /**
-     * 文档全量同步冷却标记key
-     */
-    public static String getDocumentSyncKey(String datasetId) {
-        return "knowledge:doc:sync:" + datasetId;
-    }
 }
@@ -176,12 +176,21 @@ public abstract class KnowledgeBaseAdapter {

        /**
         * 获取数据集的文档数量
-         * 
+         *
         * @param datasetId 数据集ID
         * @return 文档数量
         */
        public abstract Integer getDocumentCount(String datasetId);

+        /**
+         * 获取数据集完整信息（名称、简介、文档数量等）
+         * 用于检测 RAGFlow 端是否已删除、同步名称/简介变更
+         *
+         * @param datasetId 数据集ID
+         * @return 数据集详情，若 RAGFlow 端不存在则返回 null
+         */
+        public abstract DatasetDTO.InfoVO getDatasetInfo(String datasetId);
+
        /**
         * 发送流式请求 (SSE)
         * 
@@ -486,7 +486,20 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
    @Override
    public Integer getDocumentCount(String datasetId) {
        try {
-            // [Fix] 使用列表过滤接口获取详情 (GET /datasets?id={id})
+            DatasetDTO.InfoVO info = getDatasetInfo(datasetId);
+            if (info != null && info.getDocumentCount() != null) {
+                return info.getDocumentCount().intValue();
+            }
+            return 0;
+        } catch (Exception e) {
+            log.warn("获取文档数量失败: {}", e.getMessage());
+            return 0;
+        }
+    }
+
+    @Override
+    public DatasetDTO.InfoVO getDatasetInfo(String datasetId) {
+        try {
            Map<String, Object> params = new HashMap<>();
            params.put("id", datasetId);
            params.put("page", 1);
@@ -498,20 +511,14 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
            if (dataObj instanceof List) {
                List<?> list = (List<?>) dataObj;
                if (!list.isEmpty()) {
-                    Object firstItem = list.get(0);
-                    if (firstItem instanceof Map) {
-                        Object countObj = ((Map<?, ?>) firstItem).get("document_count");
-                        if (countObj instanceof Number) {
-                            return ((Number) countObj).intValue();
-                        }
-                    }
+                    return objectMapper.convertValue(list.get(0), DatasetDTO.InfoVO.class);
                }
            }
-            // 降级：未找到或结构不匹配
-            return 0;
+            // RAGFlow 端不存在该数据集
+            return null;
        } catch (Exception e) {
-            log.warn("获取文档数量失败: {}", e.getMessage());
-            return 0;
+            log.warn("获取数据集信息失败: datasetId={}, error={}", datasetId, e.getMessage());
+            return null;
        }
    }

@@ -96,7 +96,7 @@ public interface KnowledgeFilesService {
        /**
         * 保存文档影子记录
         */
-        void saveDocumentShadow(String datasetId, KnowledgeFilesDTO result, String originalName, String chunkMethod,
+        boolean saveDocumentShadow(String datasetId, KnowledgeFilesDTO result, String originalName, String chunkMethod,
                        Map<String, Object> parserConfig);

        /**
@@ -19,6 +19,8 @@ import xiaozhi.common.service.impl.BaseServiceImpl;
 import xiaozhi.common.utils.ConvertUtils;
 import xiaozhi.common.utils.JsonUtils;
 import xiaozhi.modules.knowledge.dao.KnowledgeBaseDao;
+import xiaozhi.modules.knowledge.dao.DocumentDao;
+import xiaozhi.modules.knowledge.entity.DocumentEntity;
 import xiaozhi.modules.knowledge.dto.KnowledgeBaseDTO;
 import xiaozhi.modules.knowledge.dto.dataset.DatasetDTO;
 import xiaozhi.modules.knowledge.entity.KnowledgeBaseEntity;
@@ -46,6 +48,7 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
        implements KnowledgeBaseService {

    private final KnowledgeBaseDao knowledgeBaseDao;
+    private final DocumentDao documentDao;
    private final ModelConfigService modelConfigService;
    private final ModelConfigDao modelConfigDao;
    private final RedisUtils redisUtils;
@@ -67,27 +70,107 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,

        // Enrich with Document Count from RAG (Optional / Lazy)
        if (pageData != null && pageData.getList() != null) {
-            for (KnowledgeBaseDTO dto : pageData.getList()) {
+            pageData.getList().removeIf(dto -> {
                enrichDocumentCount(dto);
-            }
+                // syncDatasetFromRAG 检测到 RAGFlow 端已删除时，会将本地记录清理
+                // 此时 datasetId 被置空作为标记，需要在列表中移除该条目
+                return dto.getDatasetId() == null;
+            });
        }
        return pageData;
    }

    private void enrichDocumentCount(KnowledgeBaseDTO dto) {
+        syncDatasetFromRAG(dto);
+    }
+
+    /**
+     * 从 RAGFlow 同步数据集信息：检测删除、同步名称/简介、获取文档数量
+     * 每次列表刷新时实时查询 RAGFlow，确保立即感知远端变更
+     */
+    private void syncDatasetFromRAG(KnowledgeBaseDTO dto) {
        try {
-            if (StringUtils.isNotBlank(dto.getDatasetId()) && StringUtils.isNotBlank(dto.getRagModelId())) {
-                KnowledgeBaseAdapter adapter = getAdapterByModelId(dto.getRagModelId());
-                if (adapter != null) {
-                    dto.setDocumentCount(adapter.getDocumentCount(dto.getDatasetId()));
+            if (StringUtils.isBlank(dto.getDatasetId()) || StringUtils.isBlank(dto.getRagModelId())) {
+                return;
+            }
+
+            KnowledgeBaseAdapter adapter = getAdapterByModelId(dto.getRagModelId());
+            if (adapter == null) {
+                return;
+            }
+
+            DatasetDTO.InfoVO datasetInfo = adapter.getDatasetInfo(dto.getDatasetId());
+
+            if (datasetInfo == null) {
+                // RAGFlow 端已删除 → 本地级联清理
+                log.info("数据集 {} 在 RAGFlow 端不存在，执行本地清理", dto.getDatasetId());
+                cleanupLocalDataset(dto.getDatasetId(), dto.getId());
+                // 标记为已删除，让上层从列表中移除
+                dto.setDatasetId(null);
+                return;
+            }
+
+            // 同步名称（去掉 username_ 前缀）
+            String ragflowName = datasetInfo.getName();
+            if (StringUtils.isNotBlank(ragflowName)) {
+                String localName = ragflowName.contains("_") ? ragflowName.substring(ragflowName.indexOf('_') + 1) : ragflowName;
+                if (!localName.equals(dto.getName())) {
+                    log.info("同步知识库名称: {} -> {}", dto.getName(), localName);
+                    KnowledgeBaseEntity entity = knowledgeBaseDao.selectById(dto.getId());
+                    if (entity != null) {
+                        entity.setName(localName);
+                        knowledgeBaseDao.updateById(entity);
+                        dto.setName(localName);
+                    }
                }
            }
+
+            // 同步简介
+            String ragflowDesc = datasetInfo.getDescription();
+            String localDesc = dto.getDescription();
+            boolean descChanged = (ragflowDesc == null && localDesc != null) || (ragflowDesc != null && !ragflowDesc.equals(localDesc));
+            if (descChanged) {
+                log.info("同步知识库简介: datasetId={}", dto.getDatasetId());
+                KnowledgeBaseEntity entity = knowledgeBaseDao.selectById(dto.getId());
+                if (entity != null) {
+                    entity.setDescription(ragflowDesc);
+                    knowledgeBaseDao.updateById(entity);
+                    dto.setDescription(ragflowDesc);
+                }
+            }
+
+            // 设置文档数量（保留原有功能）
+            if (datasetInfo.getDocumentCount() != null) {
+                dto.setDocumentCount(datasetInfo.getDocumentCount().intValue());
+            }
+
        } catch (Exception e) {
-            log.warn("无法获取知识库 {} 的文档计数: {}", dto.getName(), e.getMessage());
+            log.warn("同步数据集信息失败 {}: {}", dto.getName(), e.getMessage());
            dto.setDocumentCount(0);
        }
    }

+    /**
+     * 本地级联清理：RAGFlow 端已删除时，清理本地所有关联数据
+     * 不调用 RAGFlow 删除 API
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void cleanupLocalDataset(String datasetId, String entityId) {
+        try {
+            // 1. 删除文档影子记录
+            documentDao.delete(new QueryWrapper<DocumentEntity>().eq("dataset_id", datasetId));
+            // 2. 删除插件映射
+            knowledgeBaseDao.deletePluginMappingByKnowledgeBaseId(entityId);
+            // 3. 删除知识库记录
+            knowledgeBaseDao.deleteById(entityId);
+            // 4. 清理缓存
+            redisUtils.delete(RedisKeys.getKnowledgeBaseCacheKey(entityId));
+            log.info("本地级联清理完成: datasetId={}, entityId={}", datasetId, entityId);
+        } catch (Exception e) {
+            log.error("本地级联清理失败: datasetId={}, entityId={}", datasetId, entityId, e);
+        }
+    }
+
    @Override
    public KnowledgeBaseDTO getById(String id) {
        KnowledgeBaseEntity entity = knowledgeBaseDao.selectById(id);
@@ -78,25 +78,11 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
            throw new RenException(ErrorCode.RAG_DATASET_ID_AND_MODEL_ID_NOT_NULL);
        }

-        // 全量对账同步: 从RAGFlow拉取远端文档，补充本地影子表缺失的记录
-        // 当本地影子表为空时，强制同步（无视冷却），避免删除后重新上传无法感知
-        String syncKey = RedisKeys.getDocumentSyncKey(datasetId);
-        boolean cooldownActive = redisUtils.get(syncKey) != null;
-        boolean localEmpty = false;
-        if (cooldownActive) {
-            Long localCount = documentDao.selectCount(
-                    new QueryWrapper<DocumentEntity>().eq("dataset_id", datasetId));
-            localEmpty = localCount == null || localCount == 0;
-        }
-        if (!cooldownActive || localEmpty) {
-            try {
-                self.syncDocumentsFromRAG(datasetId);
-                // 60秒冷却，避免每次翻页都触发远端调用（本地为空时缩短为10秒）
-                int cooldownSec = localEmpty ? 10 : 60;
-                redisUtils.set(syncKey, "1", cooldownSec);
-            } catch (Exception e) {
-                log.warn("从RAGFlow全量同步文档失败(不影响本地查询): datasetId={}, error={}", datasetId, e.getMessage());
-            }
+        // 全量对账同步: 从RAGFlow拉取远端文档，实时同步确保立即感知远端变更
+        try {
+            self.syncDocumentsFromRAG(datasetId);
+        } catch (Exception e) {
+            log.warn("从RAGFlow全量同步文档失败(不影响本地查询): datasetId={}, error={}", datasetId, e.getMessage());
        }

        // 1. 获取本地影子表数据 (MyBatis-Plus 分页)
@@ -431,10 +417,13 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
    }

    /**
-     * 原子化保存影子记录，确保本地数据绝对一致
+     * 原子化保存影子记录（Upsert 语义）
+     * 若 document_id 已存在则更新，不存在则插入，避免 UNIQUE 约束冲突
+     *
+     * @return true=新插入, false=更新已有记录
     */
    @Transactional(rollbackFor = Exception.class)
-    public void saveDocumentShadow(String datasetId, KnowledgeFilesDTO result, String originalName, String chunkMethod,
+    public boolean saveDocumentShadow(String datasetId, KnowledgeFilesDTO result, String originalName, String chunkMethod,
            Map<String, Object> parserConfig) {
        DocumentEntity entity = new DocumentEntity();
        entity.setDatasetId(datasetId);
@@ -476,12 +465,23 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
        entity.setCreatedAt(result.getCreatedAt() != null ? result.getCreatedAt() : new Date());
        entity.setUpdatedAt(result.getUpdatedAt() != null ? result.getUpdatedAt() : new Date());

-        // 插入影子表 (若失败将抛出异常，触发调用方报错，确保 Local-First 列表一致性)
-        documentDao.insert(entity);
+        // Upsert: 检查 document_id 是否已存在，存在则更新，不存在则插入
+        DocumentEntity existing = documentDao.selectOne(
+                new QueryWrapper<DocumentEntity>().eq("document_id", entity.getDocumentId()));

-        // Issue 4: 同步递增数据集文档总数统计，保持父子表一致
-        knowledgeBaseService.updateStatistics(datasetId, 1, 0L, 0L);
-        log.info("已同步递增数据集统计: datasetId={}", datasetId);
+        if (existing != null) {
+            entity.setId(existing.getId());
+            entity.setCreatedAt(existing.getCreatedAt()); // 保留原始创建时间
+            documentDao.updateById(entity);
+            log.info("影子记录已更新: documentId={}", entity.getDocumentId());
+            return false;
+        } else {
+            documentDao.insert(entity);
+            // 新增记录时递增数据集文档总数统计
+            knowledgeBaseService.updateStatistics(datasetId, 1, 0L, 0L);
+            log.info("影子记录已插入: documentId={}, datasetId={}", entity.getDocumentId(), datasetId);
+            return true;
+        }
    }

    @Override
@@ -535,7 +535,7 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
        // 4. 原子化清理本地影子记录并同步统计数据
        self.deleteDocumentShadows(documentIds, datasetId, totalChunkDelta, totalTokenDelta);

-        // 5. 清理缓存，并清除同步冷却标记以允许立即感知可能的重新上传
+        // 5. 清理缓存
        try {
            String cacheKey = RedisKeys.getKnowledgeBaseCacheKey(datasetId);
            redisUtils.delete(cacheKey);
@@ -543,12 +543,6 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
        } catch (Exception e) {
            log.warn("驱逐 Redis 缓存失败: {}", e.getMessage());
        }
-        try {
-            String syncKey = RedisKeys.getDocumentSyncKey(datasetId);
-            redisUtils.delete(syncKey);
-        } catch (Exception e) {
-            log.warn("清除同步冷却标记失败: {}", e.getMessage());
-        }

        log.info("=== 批量文档清理完成 ===");
    }
@@ -855,9 +849,8 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
            }
        }

-        // 7. 更新: 远端和本地都存在但状态不一致的文档（处理RAGFlow复用documentId重传场景）
-        // 当文档在RAGFlow被删除后重新上传，RAGFlow可能复用同一个documentId，
-        // 导致本地影子记录仍保留旧的CANCEL/FAIL状态而不会被步骤5/6处理
+        // 7. 全量更新: 远端和本地都存在的文档，以远端为准同步所有字段
+        // 处理 RAGFlow 复用 documentId 重传、远端编辑后元数据变化等场景
        Map<String, KnowledgeFilesDTO> remoteDocMap = allRemoteDocs.stream()
                .filter(doc -> doc.getDocumentId() != null)
                .collect(Collectors.toMap(KnowledgeFilesDTO::getDocumentId, doc -> doc, (a, b) -> b));
@@ -874,62 +867,50 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
            }
            KnowledgeFilesDTO remote = entry.getValue();

-            // 判断是否需要更新：run状态或status不同，说明远端有变化
-            boolean runChanged = remote.getRun() != null && !remote.getRun().equals(local.getRun());
-            boolean statusChanged = remote.getStatus() != null && !remote.getStatus().equals(local.getStatus());
-            boolean nameChanged = remote.getName() != null && !remote.getName().equals(local.getName());
+            // 全量字段更新（以远端为准），确保本地与 RAGFlow 完全一致
+            UpdateWrapper<DocumentEntity> updateWrapper = new UpdateWrapper<DocumentEntity>()
+                    .set("run", remote.getRun())
+                    .set("status", remote.getStatus() != null ? remote.getStatus() : local.getStatus())
+                    .set("progress", remote.getProgress())
+                    .set("chunk_count", remote.getChunkCount())
+                    .set("token_count", remote.getTokenCount())
+                    .set("size", remote.getFileSize())
+                    .set("error", remote.getError())
+                    .set("process_duration", remote.getProcessDuration())
+                    .set("updated_at", new Date())
+                    .set("last_sync_at", new Date())
+                    .eq("document_id", docId)
+                    .eq("dataset_id", datasetId);

-            if (runChanged || statusChanged || nameChanged) {
-                log.info("影子更新：检测到远端文档状态变化, docId={}, 本地run={}, 远端run={}, 本地status={}, 远端status={}",
-                        docId, local.getRun(), remote.getRun(), local.getStatus(), remote.getStatus());
-
-                UpdateWrapper<DocumentEntity> updateWrapper = new UpdateWrapper<DocumentEntity>()
-                        .set("run", remote.getRun())
-                        .set("status", remote.getStatus() != null ? remote.getStatus() : local.getStatus())
-                        .set("progress", remote.getProgress())
-                        .set("chunk_count", remote.getChunkCount())
-                        .set("token_count", remote.getTokenCount())
-                        .set("error", remote.getError())
-                        .set("process_duration", remote.getProcessDuration())
-                        .set("updated_at", new Date())
-                        .set("last_sync_at", new Date())
-                        .eq("document_id", docId)
-                        .eq("dataset_id", datasetId);
-
-                if (remote.getName() != null) {
-                    updateWrapper.set("name", remote.getName());
-                }
-                if (remote.getThumbnail() != null) {
-                    updateWrapper.set("thumbnail", remote.getThumbnail());
-                }
-                if (remote.getMetaFields() != null) {
-                    try {
-                        updateWrapper.set("meta_fields", objectMapper.writeValueAsString(remote.getMetaFields()));
-                    } catch (Exception e) {
-                        log.warn("同步更新元数据序列化失败: docId={}, error={}", docId, e.getMessage());
-                    }
-                }
-
-                documentDao.update(null, updateWrapper);
-
-                // 如果状态从终态(CANCEL/FAIL)变为活跃态(UNSTART/RUNNING)，需要同步token统计
-                boolean wasTerminal = "CANCEL".equals(local.getRun()) || "FAIL".equals(local.getRun());
-                boolean isActive = "RUNNING".equals(remote.getRun()) || "UNSTART".equals(remote.getRun());
-                Long remoteTokenCount = remote.getTokenCount() != null ? remote.getTokenCount() : 0L;
-                Long localTokenCount = local.getTokenCount() != null ? local.getTokenCount() : 0L;
-                long remoteChunkCount = remote.getChunkCount() != null ? remote.getChunkCount().longValue() : 0L;
-                long localChunkCount = local.getChunkCount() != null ? local.getChunkCount().longValue() : 0L;
-                if (wasTerminal && isActive) {
-                    long tokenDelta = remoteTokenCount - localTokenCount;
-                    long chunkDelta = remoteChunkCount - localChunkCount;
-                    if (tokenDelta != 0 || chunkDelta != 0) {
-                        knowledgeBaseService.updateStatistics(datasetId, 0, chunkDelta, tokenDelta);
-                        log.info("影子更新: 修正知识库统计, docId={}, chunkDelta={}, tokenDelta={}", docId, chunkDelta, tokenDelta);
-                    }
-                }
-
-                updateCount++;
+            if (remote.getName() != null) {
+                updateWrapper.set("name", remote.getName());
            }
+            if (remote.getThumbnail() != null) {
+                updateWrapper.set("thumbnail", remote.getThumbnail());
+            }
+            if (remote.getMetaFields() != null) {
+                try {
+                    updateWrapper.set("meta_fields", objectMapper.writeValueAsString(remote.getMetaFields()));
+                } catch (Exception e) {
+                    log.warn("同步更新元数据序列化失败: docId={}, error={}", docId, e.getMessage());
+                }
+            }
+
+            documentDao.update(null, updateWrapper);
+
+            // 同步统计差异（chunk/token 计数变化时修正父表）
+            Long remoteTokenCount = remote.getTokenCount() != null ? remote.getTokenCount() : 0L;
+            Long localTokenCount = local.getTokenCount() != null ? local.getTokenCount() : 0L;
+            long remoteChunkCount = remote.getChunkCount() != null ? remote.getChunkCount().longValue() : 0L;
+            long localChunkCount = local.getChunkCount() != null ? local.getChunkCount().longValue() : 0L;
+            long tokenDelta = remoteTokenCount - localTokenCount;
+            long chunkDelta = remoteChunkCount - localChunkCount;
+            if (tokenDelta != 0 || chunkDelta != 0) {
+                knowledgeBaseService.updateStatistics(datasetId, 0, chunkDelta, tokenDelta);
+                log.info("影子更新: 修正知识库统计, docId={}, chunkDelta={}, tokenDelta={}", docId, chunkDelta, tokenDelta);
+            }
+
+            updateCount++;
        }

        if (syncCount == 0 && deletedDocs.isEmpty() && updateCount == 0) {