diff --git a/main/manager-api/src/main/java/xiaozhi/common/redis/RedisKeys.java b/main/manager-api/src/main/java/xiaozhi/common/redis/RedisKeys.java index 6b143930..6691e395 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/redis/RedisKeys.java +++ b/main/manager-api/src/main/java/xiaozhi/common/redis/RedisKeys.java @@ -187,4 +187,5 @@ public class RedisKeys { public static String getOtaUploadCountKey(Long username) { return "ota:upload:count:" + username; } + } diff --git a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/KnowledgeBaseAdapter.java b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/KnowledgeBaseAdapter.java index 9e6314e4..2c45e797 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/KnowledgeBaseAdapter.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/KnowledgeBaseAdapter.java @@ -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) * diff --git a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/impl/RAGFlowAdapter.java b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/impl/RAGFlowAdapter.java index 32cbc64b..13a6decf 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/impl/RAGFlowAdapter.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/impl/RAGFlowAdapter.java @@ -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 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; } } diff --git a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/KnowledgeFilesService.java b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/KnowledgeFilesService.java index fa420e38..f832c1b1 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/KnowledgeFilesService.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/KnowledgeFilesService.java @@ -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 parserConfig); /** @@ -120,4 +120,13 @@ public interface KnowledgeFilesService { * 同步所有处于 RUNNING 状态的文档 (供定时任务调用) */ void syncRunningDocuments(); + + /** + * 从RAGFlow全量同步文档到本地影子表 + * 拉取远端所有文档,与本地影子表对比,插入缺失的记录 + * + * @param datasetId 数据集ID + * @return 新同步的文档数量 + */ + int syncDocumentsFromRAG(String datasetId); } \ No newline at end of file diff --git a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeBaseServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeBaseServiceImpl.java index ab88d6e9..94958bd1 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeBaseServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeBaseServiceImpl.java @@ -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 { 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().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); diff --git a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeFilesServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeFilesServiceImpl.java index 3bf09b60..680abd6a 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeFilesServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeFilesServiceImpl.java @@ -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 pageParams = new Page<>(page, limit); QueryWrapper queryWrapper = new QueryWrapper<>(); @@ -408,10 +417,13 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl parserConfig) { DocumentEntity entity = new DocumentEntity(); entity.setDatasetId(datasetId); @@ -453,12 +465,23 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl().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 ragConfig = knowledgeBaseService.getRAGConfigByDatasetId(datasetId); + KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(extractAdapterType(ragConfig), ragConfig); + + // 2. 分页拉取远端所有文档 + List 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 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 localDocs = documentDao.selectList( + new QueryWrapper().eq("dataset_id", datasetId)); + Set localDocIds = localDocs.stream() + .map(DocumentEntity::getDocumentId) + .collect(Collectors.toSet()); + + // 4. 远端文档ID集合 + Set remoteDocIds = allRemoteDocs.stream() + .map(KnowledgeFilesDTO::getDocumentId) + .filter(id -> id != null) + .collect(Collectors.toSet()); + + // 5. 补充: 插入远端存在但本地缺失的文档 + List 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 deletedDocs = localDocs.stream() + .filter(entity -> !remoteDocIds.contains(entity.getDocumentId())) + .collect(Collectors.toList()); + + if (!deletedDocs.isEmpty()) { + List 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 remoteDocMap = allRemoteDocs.stream() + .filter(doc -> doc.getDocumentId() != null) + .collect(Collectors.toMap(KnowledgeFilesDTO::getDocumentId, doc -> doc, (a, b) -> b)); + + Map localDocMap = localDocs.stream() + .collect(Collectors.toMap(DocumentEntity::getDocumentId, e -> e, (a, b) -> b)); + + int updateCount = 0; + for (Map.Entry entry : remoteDocMap.entrySet()) { + String docId = entry.getKey(); + DocumentEntity local = localDocMap.get(docId); + if (local == null) { + continue; // 不在本地,由步骤5处理 + } + KnowledgeFilesDTO remote = entry.getValue(); + + // 全量字段更新(以远端为准),确保本地与 RAGFlow 完全一致 + UpdateWrapper updateWrapper = new UpdateWrapper() + .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> groupedDocs = runningDocs.stream() - .collect(java.util.stream.Collectors.groupingBy(DocumentEntity::getDatasetId)); + .collect(Collectors.groupingBy(DocumentEntity::getDatasetId)); groupedDocs.forEach((datasetId, docs) -> { KnowledgeBaseAdapter adapter = null;