Merge pull request #3115 from xinnan-tech/fix-knowledge-sync-ragflow-documents

fix: 实现知识库与RAGFlow双向同步（知识库名称/简介的修改、知识库删除、文档上传/删除）
2026-07-22 23:23:55 +08:00 · 2026-04-20 16:51:06 +08:00
parent 97b9c95c2d 0681550bab
commit 6f52d004e8
6 changed files with 322 additions and 29 deletions
@@ -187,4 +187,5 @@ public class RedisKeys {
    public static String getOtaUploadCountKey(Long username) {
        return "ota:upload:count:" + username;
    }
+
 }
@@ -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);

        /**
@@ -120,4 +120,13 @@ public interface KnowledgeFilesService {
         * 同步所有处于 RUNNING 状态的文档 (供定时任务调用)
         */
        void syncRunningDocuments();
+
+        /**
+         * 从RAGFlow全量同步文档到本地影子表
+         * 拉取远端所有文档，与本地影子表对比，插入缺失的记录
+         *
+         * @param datasetId 数据集ID
+         * @return 新同步的文档数量
+         */
+        int syncDocumentsFromRAG(String datasetId);
 }
@@ -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);
@@ -4,6 +4,8 @@ import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;

 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.BeanUtils;
@@ -76,6 +78,13 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
            throw new RenException(ErrorCode.RAG_DATASET_ID_AND_MODEL_ID_NOT_NULL);
        }

+        // 全量对账同步: 从RAGFlow拉取远端文档，实时同步确保立即感知远端变更
+        try {
+            self.syncDocumentsFromRAG(datasetId);
+        } catch (Exception e) {
+            log.warn("从RAGFlow全量同步文档失败(不影响本地查询): datasetId={}, error={}", datasetId, e.getMessage());
+        }
+
        // 1. 获取本地影子表数据 (MyBatis-Plus 分页)
        Page<DocumentEntity> pageParams = new Page<>(page, limit);
        QueryWrapper<DocumentEntity> queryWrapper = new QueryWrapper<>();
@@ -408,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);
@@ -453,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
@@ -738,6 +761,167 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
        this.deleteDocuments(datasetId, req);
    }

+    @Override
+    public int syncDocumentsFromRAG(String datasetId) {
+        log.info("=== 开始从RAGFlow全量同步文档到本地影子表: datasetId={} ===", datasetId);
+
+        // 1. 获取适配器
+        Map<String, Object> ragConfig = knowledgeBaseService.getRAGConfigByDatasetId(datasetId);
+        KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(extractAdapterType(ragConfig), ragConfig);
+
+        // 2. 分页拉取远端所有文档
+        List<KnowledgeFilesDTO> allRemoteDocs = new ArrayList<>();
+        int pageNum = 1;
+        int pageSize = 100;
+        long totalRemote = Long.MAX_VALUE;
+
+        while ((long) (pageNum - 1) * pageSize < totalRemote) {
+            DocumentDTO.ListReq req = DocumentDTO.ListReq.builder()
+                    .page(pageNum)
+                    .pageSize(pageSize)
+                    .build();
+            PageData<KnowledgeFilesDTO> remotePage = adapter.getDocumentList(datasetId, req);
+            if (remotePage == null || remotePage.getList() == null || remotePage.getList().isEmpty()) {
+                break;
+            }
+            allRemoteDocs.addAll(remotePage.getList());
+            totalRemote = remotePage.getTotal();
+            pageNum++;
+        }
+
+        // 3. 获取本地已有文档
+        List<DocumentEntity> localDocs = documentDao.selectList(
+                new QueryWrapper<DocumentEntity>().eq("dataset_id", datasetId));
+        Set<String> localDocIds = localDocs.stream()
+                .map(DocumentEntity::getDocumentId)
+                .collect(Collectors.toSet());
+
+        // 4. 远端文档ID集合
+        Set<String> remoteDocIds = allRemoteDocs.stream()
+                .map(KnowledgeFilesDTO::getDocumentId)
+                .filter(id -> id != null)
+                .collect(Collectors.toSet());
+
+        // 5. 补充: 插入远端存在但本地缺失的文档
+        List<KnowledgeFilesDTO> newDocs = allRemoteDocs.stream()
+                .filter(doc -> doc.getDocumentId() != null && !localDocIds.contains(doc.getDocumentId()))
+                .collect(Collectors.toList());
+
+        int syncCount = 0;
+        if (!newDocs.isEmpty()) {
+            for (KnowledgeFilesDTO doc : newDocs) {
+                try {
+                    self.saveDocumentShadow(datasetId, doc, doc.getName(), doc.getChunkMethod(), doc.getParserConfig());
+                    // 同步远端已有的 token/chunk 统计
+                    Long tokenCount = doc.getTokenCount() != null ? doc.getTokenCount() : 0L;
+                    long chunkCount = doc.getChunkCount() != null ? doc.getChunkCount().longValue() : 0L;
+                    if (tokenCount > 0 || chunkCount > 0) {
+                        knowledgeBaseService.updateStatistics(datasetId, 0, chunkCount, tokenCount);
+                    }
+                    syncCount++;
+                } catch (Exception e) {
+                    log.warn("同步单个文档影子记录失败: docId={}, error={}", doc.getDocumentId(), e.getMessage());
+                }
+            }
+            log.info("从RAGFlow新增同步 {} 个文档影子记录, datasetId={}", syncCount, datasetId);
+        }
+
+        // 6. 清理: 删除远端已不存在但本地仍保留的影子记录
+        List<DocumentEntity> deletedDocs = localDocs.stream()
+                .filter(entity -> !remoteDocIds.contains(entity.getDocumentId()))
+                .collect(Collectors.toList());
+
+        if (!deletedDocs.isEmpty()) {
+            List<String> deletedDocIds = new ArrayList<>();
+            long totalChunkDelta = 0;
+            long totalTokenDelta = 0;
+
+            for (DocumentEntity entity : deletedDocs) {
+                deletedDocIds.add(entity.getDocumentId());
+                totalChunkDelta += entity.getChunkCount() != null ? entity.getChunkCount() : 0L;
+                totalTokenDelta += entity.getTokenCount() != null ? entity.getTokenCount() : 0L;
+            }
+            try {
+                self.deleteDocumentShadows(deletedDocIds, datasetId, totalChunkDelta, totalTokenDelta);
+                log.info("清理远端已删除的影子记录: {} 个, datasetId={}", deletedDocs.size(), datasetId);
+            } catch (Exception e) {
+                log.warn("清理远端已删除的影子记录失败: datasetId={}, error={}", datasetId, e.getMessage());
+            }
+        }
+
+        // 7. 全量更新: 远端和本地都存在的文档，以远端为准同步所有字段
+        // 处理 RAGFlow 复用 documentId 重传、远端编辑后元数据变化等场景
+        Map<String, KnowledgeFilesDTO> remoteDocMap = allRemoteDocs.stream()
+                .filter(doc -> doc.getDocumentId() != null)
+                .collect(Collectors.toMap(KnowledgeFilesDTO::getDocumentId, doc -> doc, (a, b) -> b));
+
+        Map<String, DocumentEntity> localDocMap = localDocs.stream()
+                .collect(Collectors.toMap(DocumentEntity::getDocumentId, e -> e, (a, b) -> b));
+
+        int updateCount = 0;
+        for (Map.Entry<String, KnowledgeFilesDTO> entry : remoteDocMap.entrySet()) {
+            String docId = entry.getKey();
+            DocumentEntity local = localDocMap.get(docId);
+            if (local == null) {
+                continue; // 不在本地，由步骤5处理
+            }
+            KnowledgeFilesDTO remote = entry.getValue();
+
+            // 全量字段更新（以远端为准），确保本地与 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 (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) {
+            log.info("本地影子表已与RAGFlow完全同步, datasetId={}", datasetId);
+        } else {
+            log.info("同步完成: 新增={}, 清理={}, 更新={}, datasetId={}", syncCount, deletedDocs.size(), updateCount, datasetId);
+        }
+
+        return syncCount;
+    }
+
    @Override
    public void syncRunningDocuments() {
        // 1. 查询所有 RUNNING 状态的文档
@@ -755,7 +939,7 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu

        // 2. 按 DatasetID 分组，复用 Adapter
        Map<String, List<DocumentEntity>> groupedDocs = runningDocs.stream()
-                .collect(java.util.stream.Collectors.groupingBy(DocumentEntity::getDatasetId));
+                .collect(Collectors.groupingBy(DocumentEntity::getDatasetId));

        groupedDocs.forEach((datasetId, docs) -> {
            KnowledgeBaseAdapter adapter = null;