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 ec2efc6b..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 @@ -188,10 +188,4 @@ public class RedisKeys { return "ota:upload:count:" + username; } - /** - * 文档全量同步冷却标记key - */ - public static String getDocumentSyncKey(String datasetId) { - return "knowledge:doc:sync:" + datasetId; - } } 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 f46b7417..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); /** 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 49a8820f..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 @@ -78,25 +78,11 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl().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 parserConfig) { DocumentEntity entity = new DocumentEntity(); entity.setDatasetId(datasetId); @@ -476,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 @@ -535,7 +535,7 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl 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 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 (runChanged || statusChanged || nameChanged) { - log.info("影子更新:检测到远端文档状态变化, docId={}, 本地run={}, 远端run={}, 本地status={}, 远端status={}", - docId, local.getRun(), remote.getRun(), local.getStatus(), remote.getStatus()); - - 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("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) {