From ac9e200a1552482aedd9255430c09c281e9ec9d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Mon, 10 Aug 2026 23:58:24 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=BC=BA=E5=8C=96=E6=96=87=E6=A1=A3?= =?UTF-8?q?=E5=88=86=E5=9D=97=E4=B8=8E=E7=B4=A2=E5=BC=95=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 对最终向量分块应用 BGE-M3 上下文硬限制 - 批量写入和删除关键词索引并校验执行结果 - 外部索引回滚失败时保留明确错误提示 - 补充自动重分块、快照清理和失败回滚测试 --- .../ai/service/impl/DocumentServiceImpl.java | 87 ++- ...ledgeDocumentImportTaskAppServiceTest.java | 509 +++++++++++++++--- .../impl/DocumentServiceImplRollbackTest.java | 100 ++++ .../service/impl/DocumentServiceImplTest.java | 31 ++ 4 files changed, 645 insertions(+), 82 deletions(-) create mode 100644 easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplRollbackTest.java diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java index e8a29d41..f037a43e 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java @@ -12,6 +12,7 @@ import com.easyagents.core.model.embedding.EmbeddingOptions; import com.easyagents.core.store.DocumentStore; import com.easyagents.core.store.StoreOptions; import com.easyagents.core.store.StoreResult; +import com.easyagents.rag.core.BgeM3ChunkSafety; import com.easyagents.rag.core.RagChunk; import com.easyagents.rag.core.RagDefaults; import com.easyagents.rag.core.RagStrategyCodes; @@ -73,6 +74,7 @@ import static tech.easyflow.ai.entity.table.DocumentTableDef.DOCUMENT; public class DocumentServiceImpl extends ServiceImpl implements DocumentService { protected Logger Log = LoggerFactory.getLogger(DocumentServiceImpl.class); private static final String SOURCE_RANGES_KEY = "sourceRanges"; + private static final String ROLLBACK_FAILURE_SUFFIX = "；外部索引回滚未完成，请联系管理员处理"; @Resource private DocumentMapper documentMapper; @@ -392,9 +394,9 @@ public class DocumentServiceImpl extends ServiceImpl i return Result.ok(); } catch (Exception e) { cleanupPersistedDocument(document); - rollbackStoredChunks(storeContext, validChunks); + boolean rollbackSucceeded = rollbackStoredChunks(storeContext, validChunks); Log.error("保存文档失败: documentId={}, title={}", document.getId(), document.getTitle(), e); - throw new BusinessException("保存失败：" + e.getMessage()); + throw new BusinessException(buildStoreFailureMessage("保存失败：", e, rollbackSucceeded)); } finally { closeStoreContext(storeContext); } @@ -541,8 +543,8 @@ public class DocumentServiceImpl extends ServiceImpl i updateKnowledgeAfterStore(storeContext); } catch (Exception e) { cleanupPersistedDocument(document); - rollbackStoredChunks(storeContext, session.getDocumentChunks()); - throw new BusinessException("提交导入失败：" + e.getMessage()); + boolean rollbackSucceeded = rollbackStoredChunks(storeContext, session.getDocumentChunks()); + throw new BusinessException(buildStoreFailureMessage("提交导入失败：", e, rollbackSucceeded)); } finally { closeStoreContext(storeContext); } @@ -897,6 +899,7 @@ public class DocumentServiceImpl extends ServiceImpl i private void storeDocumentChunks(StoreExecutionContext storeContext, List documentChunks) { List documents = new ArrayList<>(); for (DocumentChunk item : documentChunks) { + assertEmbeddingChunkWithinHardLimit(item); com.easyagents.core.document.Document document = new com.easyagents.core.document.Document(); document.setId(item.getId()); document.setContent(item.getContent()); @@ -921,30 +924,82 @@ public class DocumentServiceImpl extends ServiceImpl i throw new BusinessException("DocumentStore.store failed"); } - if (storeContext.searcher != null) { - for (com.easyagents.core.document.Document document : documents) { - storeContext.searcher.addDocument(document); - } + if (storeContext.searcher != null && !storeContext.searcher.addDocuments(documents)) { + throw new BusinessException("关键词索引写入失败"); } } - private void rollbackStoredChunks(StoreExecutionContext storeContext, List documentChunks) { + /** + * 校验最终分块没有超过 BGE-M3 向量上下文安全预算。 + * + * @param chunk 待向量化分块 + * @throws BusinessException 分块超过上下文安全上限 + */ + private void assertEmbeddingChunkWithinHardLimit(DocumentChunk chunk) { + String content = chunk == null ? null : chunk.getContent(); + int tokenEstimate = BgeM3ChunkSafety.estimateContentTokens(content); + if (tokenEstimate > RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT) { + BigInteger chunkId = chunk == null ? null : chunk.getId(); + throw new BusinessException( + "分块内容超过向量模型上下文上限，请重新分块：chunkId=" + + chunkId + "，tokenEstimate=" + tokenEstimate); + } + } + + /** + * 回滚已写入的向量和关键词索引。 + * + * @param storeContext 外部存储上下文 + * @param documentChunks 待回滚分块 + * @return 两类外部索引均回滚成功时返回 {@code true} + */ + private boolean rollbackStoredChunks(StoreExecutionContext storeContext, + List documentChunks) { try { - List chunkIds = new ArrayList<>(); + Set uniqueIds = new LinkedHashSet<>(); for (DocumentChunk chunk : documentChunks) { - chunkIds.add(chunk.getId()); - } - storeContext.documentStore.delete(chunkIds, storeContext.options); - if (storeContext.searcher != null) { - for (BigInteger chunkId : chunkIds) { - storeContext.searcher.deleteDocument(chunkId); + if (chunk != null && chunk.getId() != null) { + uniqueIds.add(chunk.getId()); } } + List chunkIds = new ArrayList<>(uniqueIds); + if (chunkIds.isEmpty()) { + return true; + } + StoreResult deleteResult = storeContext.documentStore.delete(chunkIds, storeContext.options); + if (deleteResult == null || !deleteResult.isSuccess()) { + String failReason = deleteResult == null ? "未返回结果" : deleteResult.getFailReason(); + throw new IllegalStateException("向量存储回滚失败: " + failReason); + } + if (storeContext.searcher != null + && !storeContext.searcher.deleteDocuments(chunkIds)) { + throw new IllegalStateException("关键词索引回滚失败"); + } + return true; } catch (Exception e) { Log.error("回滚向量文档失败: knowledgeId={}", storeContext.knowledge.getId(), e); + return false; } } + /** + * 生成文档写入失败信息，并显式提示未完成的外部索引回滚。 + * + * @param prefix 业务场景前缀 + * @param error 原始异常 + * @param rollbackSucceeded 外部索引是否完整回滚 + * @return 用户可见失败信息 + */ + private String buildStoreFailureMessage(String prefix, + Exception error, + boolean rollbackSucceeded) { + String detail = error != null && StringUtil.hasText(error.getMessage()) + ? error.getMessage() + : "未知错误"; + String message = prefix + detail; + return rollbackSucceeded ? message : message + ROLLBACK_FAILURE_SUFFIX; + } + private void updateKnowledgeAfterStore(StoreExecutionContext storeContext) { DocumentCollection documentCollection = new DocumentCollection(); documentCollection.setId(storeContext.knowledge.getId()); diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppServiceTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppServiceTest.java index ce6761b0..e71cbce9 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppServiceTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppServiceTest.java @@ -7,6 +7,7 @@ import com.easyagents.core.model.embedding.EmbeddingModel; import com.easyagents.core.store.DocumentStore; import com.easyagents.core.store.StoreOptions; import com.easyagents.core.store.StoreResult; +import com.easyagents.rag.core.BgeM3ChunkSafety; import com.easyagents.rag.core.RagChunk; import com.easyagents.rag.core.RagDefaults; import com.easyagents.rag.core.RagStrategyCodes; @@ -22,6 +23,8 @@ import org.junit.Test; import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import org.springframework.data.redis.RedisSystemException; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.web.multipart.MultipartFile; import tech.easyflow.ai.document.exception.DocumentParseBridgeException; @@ -56,6 +59,7 @@ import tech.easyflow.common.web.exceptions.BusinessException; import java.io.ByteArrayInputStream; import java.lang.reflect.Constructor; import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.lang.reflect.Proxy; import java.math.BigInteger; @@ -1387,6 +1391,298 @@ public class KnowledgeDocumentImportTaskAppServiceTest { ); } + /** + * 验证自动导入向量化失败后从分块阶段重试，并废弃旧分块快照。 + * + * @throws Exception 反射注入异常 + */ + @Test + public void retryIndexTaskShouldResplitAutoBatchAfterIndexFailure() + throws Exception { + BigInteger knowledgeId = BigInteger.valueOf(101); + BigInteger documentId = BigInteger.valueOf(102); + BigInteger batchId = BigInteger.valueOf(103); + BigInteger batchItemId = BigInteger.valueOf(104); + BigInteger splitTaskId = BigInteger.valueOf(105); + String staleSnapshotPath = "document-import/chunks/stale.snapshot"; + + DocumentCollection knowledge = new DocumentCollection(); + knowledge.setId(knowledgeId); + tech.easyflow.ai.entity.Document document = + new tech.easyflow.ai.entity.Document(); + document.setId(documentId); + document.setCollectionId(knowledgeId); + document.setProcessStatus(DocumentProcessStatus.INDEX_FAILED.name()); + Map options = new LinkedHashMap(); + options.put(DocumentImportKeys.KEY_DOCUMENT_IMPORT_BATCH_ID, + batchId.toString()); + options.put(DocumentImportKeys.KEY_DOCUMENT_IMPORT_BATCH_ITEM_ID, + batchItemId.toString()); + options.put(DocumentImportKeys.KEY_DOCUMENT_CHUNK_SNAPSHOT_PATH, + staleSnapshotPath); + document.setOptions(options); + + DocumentImportBatch batch = new DocumentImportBatch(); + batch.setId(batchId); + batch.setRequestedStrategyJson("{\"strategyCode\":\"AUTO\"}"); + + DocumentCollectionService knowledgeService = + Mockito.mock(DocumentCollectionService.class); + Mockito.when(knowledgeService.getById(knowledgeId)).thenReturn(knowledge); + DocumentMapper documentMapper = Mockito.mock(DocumentMapper.class); + Mockito.when(documentMapper.selectOneById(documentId)).thenReturn(document); + Mockito.when(documentMapper.updateByQuery( + Mockito.any(tech.easyflow.ai.entity.Document.class), + Mockito.any(com.mybatisflex.core.query.QueryWrapper.class))) + .thenReturn(1); + DocumentImportBatchTracker tracker = + Mockito.mock(DocumentImportBatchTracker.class); + Mockito.when(tracker.isAutoBatch(batchId)).thenReturn(true); + Mockito.when(tracker.requireBatch(batchId)).thenReturn(batch); + DocumentImportTaskService taskService = + Mockito.mock(DocumentImportTaskService.class); + AtomicReference createdTask = + new AtomicReference(); + Mockito.doAnswer(invocation -> { + DocumentImportTask task = invocation.getArgument(0); + task.setId(splitTaskId); + createdTask.set(task); + return true; + }).when(taskService).save(Mockito.any(DocumentImportTask.class)); + DocumentImportSplitTaskProducer splitTaskProducer = + Mockito.mock(DocumentImportSplitTaskProducer.class); + DocumentImportSnapshotCleanupService cleanupService = + Mockito.mock(DocumentImportSnapshotCleanupService.class); + ThreadPoolTaskExecutor taskExecutor = + Mockito.mock(ThreadPoolTaskExecutor.class); + + KnowledgeDocumentImportTaskAppService service = + new KnowledgeDocumentImportTaskAppService(); + setField(service, "knowledgeService", knowledgeService); + setField(service, "documentMapper", documentMapper); + setField(service, "documentImportBatchTracker", tracker); + setField(service, "documentImportTaskService", taskService); + setField(service, "splitTaskProducer", splitTaskProducer); + setField(service, "snapshotCleanupService", cleanupService); + setField(service, "documentImportTaskExecutor", taskExecutor); + setField(service, "documentImportTaskStatusStreamService", + new NoopTaskStatusStreamService()); + + DocumentImportDtos.TaskRetryRequest request = + new DocumentImportDtos.TaskRetryRequest(); + request.setKnowledgeId(knowledgeId); + request.setDocumentId(documentId); + DocumentImportDtos.TaskStartIndexResponse response = + service.retryIndexTask(request).getData(); + + Assert.assertNotNull(response); + Assert.assertEquals(splitTaskId, response.getTaskId()); + Assert.assertEquals(DocumentProcessStatus.SPLITTING.name(), + response.getProcessStatus()); + Assert.assertFalse(document.getOptions().containsKey( + DocumentImportKeys.KEY_DOCUMENT_CHUNK_SNAPSHOT_PATH)); + Assert.assertNotNull(createdTask.get()); + Assert.assertEquals(DocumentImportTaskPhase.SPLIT.name(), + createdTask.get().getPhase()); + Assert.assertEquals(batch.getRequestedStrategyJson(), + createdTask.get().getPayloadJson().get("strategyConfigJson")); + Mockito.verify(splitTaskProducer).send(splitTaskId); + Mockito.verify(cleanupService).scheduleChunkSnapshot( + staleSnapshotPath); + Mockito.verify(documentMapper).update( + Mockito.argThat(updated -> updated.getOptions() != null + && !updated.getOptions().containsKey( + DocumentImportKeys.KEY_DOCUMENT_CHUNK_SNAPSHOT_PATH)), + Mockito.eq(false)); + Mockito.verify(tracker).updateItem( + batchItemId, + DocumentImportBatchItemStage.SPLIT, + DocumentImportBatchItemStatus.PENDING, + null); + } + + /** + * 验证旧分块快照只会在事务提交后登记可靠清理。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void deleteChunkSnapshotShouldWaitForTransactionCommit() + throws Exception { + KnowledgeDocumentImportTaskAppService service = + new KnowledgeDocumentImportTaskAppService(); + DocumentImportSnapshotCleanupService cleanupService = + Mockito.mock(DocumentImportSnapshotCleanupService.class); + setField(service, "snapshotCleanupService", cleanupService); + Method method = KnowledgeDocumentImportTaskAppService.class + .getDeclaredMethod("deleteChunkSnapshotAfterCommit", String.class); + method.setAccessible(true); + + TransactionSynchronizationManager.setActualTransactionActive(true); + TransactionSynchronizationManager.initSynchronization(); + try { + method.invoke(service, "document-import/chunks/commit.snapshot"); + Mockito.verifyNoInteractions(cleanupService); + + List synchronizations = + TransactionSynchronizationManager.getSynchronizations(); + for (TransactionSynchronization synchronization : synchronizations) { + synchronization.afterCommit(); + synchronization.afterCompletion( + TransactionSynchronization.STATUS_COMMITTED); + } + Mockito.verify(cleanupService).scheduleChunkSnapshot( + "document-import/chunks/commit.snapshot"); + } finally { + TransactionSynchronizationManager.clearSynchronization(); + TransactionSynchronizationManager.setActualTransactionActive(false); + } + } + + /** + * 验证事务回滚时保留旧分块快照。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void deleteChunkSnapshotShouldKeepSnapshotAfterRollback() + throws Exception { + KnowledgeDocumentImportTaskAppService service = + new KnowledgeDocumentImportTaskAppService(); + DocumentImportSnapshotCleanupService cleanupService = + Mockito.mock(DocumentImportSnapshotCleanupService.class); + setField(service, "snapshotCleanupService", cleanupService); + Method method = KnowledgeDocumentImportTaskAppService.class + .getDeclaredMethod("deleteChunkSnapshotAfterCommit", String.class); + method.setAccessible(true); + + TransactionSynchronizationManager.setActualTransactionActive(true); + TransactionSynchronizationManager.initSynchronization(); + try { + method.invoke(service, "document-import/chunks/rollback.snapshot"); + for (TransactionSynchronization synchronization + : TransactionSynchronizationManager.getSynchronizations()) { + synchronization.afterCompletion( + TransactionSynchronization.STATUS_ROLLED_BACK); + } + Mockito.verifyNoInteractions(cleanupService); + } finally { + TransactionSynchronizationManager.clearSynchronization(); + TransactionSynchronizationManager.setActualTransactionActive(false); + } + } + + /** + * 验证 Office 超长正文也会应用统一的 BGE-M3 硬限制。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void officeChunksShouldUseSharedBgeM3HardLimit() + throws Exception { + KnowledgeDocumentImportTaskAppService service = + new KnowledgeDocumentImportTaskAppService(); + DocumentChunk source = new DocumentChunk(); + source.setId(BigInteger.valueOf(501)); + source.setDocumentId(BigInteger.valueOf(502)); + source.setDocumentCollectionId(BigInteger.valueOf(503)); + source.setContent("长".repeat( + RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT + 100)); + Map options = new LinkedHashMap(); + options.put( + DocumentImportKeys.KEY_DOCUMENT_RENDER_MARKDOWN, + source.getContent()); + source.setOptions(options); + + Method method = KnowledgeDocumentImportTaskAppService.class + .getDeclaredMethod("enforceDocumentChunkHardLimit", List.class); + method.setAccessible(true); + @SuppressWarnings("unchecked") + List result = + (List) method.invoke(service, List.of(source)); + + Assert.assertTrue(result.size() > 1); + Assert.assertTrue(result.stream().allMatch( + chunk -> BgeM3ChunkSafety.isWithinHardLimit(chunk.getContent()))); + Assert.assertTrue(result.stream().allMatch( + chunk -> BgeM3ChunkSafety.isWithinHardLimit(String.valueOf( + chunk.getOptions().get( + DocumentImportKeys.KEY_DOCUMENT_RENDER_MARKDOWN))))); + } + + /** + * 验证服务端拒绝无法推进游标的分块重叠参数。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void strategyOverlapShouldBeSmallerThanChunkSize() + throws Exception { + KnowledgeDocumentImportTaskAppService service = + new KnowledgeDocumentImportTaskAppService(); + StrategyConfig config = StrategyConfig.defaults(); + config.setChunkSize(128); + config.setOverlapSize(128); + Method method = KnowledgeDocumentImportTaskAppService.class + .getDeclaredMethod("validateStrategyOverlap", StrategyConfig.class); + method.setAccessible(true); + + try { + method.invoke(service, config); + Assert.fail("重叠大小等于分块大小时应拒绝请求"); + } catch (InvocationTargetException expected) { + Assert.assertTrue(expected.getCause() instanceof BusinessException); + } + } + + /** + * 验证 AUTO 空结果使用默认长度与重叠参数回退自然段长度拆分。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void autoEmptyChunksShouldFallbackWithDefaultParagraphSettings() + throws Exception { + assertAutoParagraphFallback(RagDefaults.CHUNK_SIZE, RagDefaults.OVERLAP_SIZE); + } + + /** + * 验证手动预览选择 AUTO 时，兜底保留页面传入的长度与重叠参数。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void autoEmptyChunksShouldFallbackWithRequestedParagraphSettings() + throws Exception { + assertAutoParagraphFallback(768, 192); + } + + /** + * 验证显式选择非 AUTO 策略时不触发兜底。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void explicitStrategyEmptyChunksShouldNotFallback() throws Exception { + KnowledgeDocumentImportTaskAppService service = + new KnowledgeDocumentImportTaskAppService(); + RagIngestionService ragIngestionService = Mockito.mock(RagIngestionService.class); + setField(service, "ragIngestionService", ragIngestionService); + AnalysisResult analysis = Mockito.mock(AnalysisResult.class); + StrategyConfig requestedStrategy = StrategyConfig.defaults(); + requestedStrategy.setStrategyCode(RagStrategyCodes.OUTLINE_SECTION); + StrategyConfig effectiveStrategy = requestedStrategy.copy(); + Mockito.when(ragIngestionService.split(analysis, effectiveStrategy)) + .thenReturn(List.of()); + + List chunks = invokeAutoParagraphFallback( + service, analysis, requestedStrategy, effectiveStrategy); + + Assert.assertTrue(chunks.isEmpty()); + Assert.assertEquals(RagStrategyCodes.OUTLINE_SECTION, effectiveStrategy.getStrategyCode()); + Mockito.verify(ragIngestionService).split(analysis, effectiveStrategy); + } + /** * 验证看门狗已经收口任务后，迟到执行者不能覆盖文档终态。 * @@ -1726,12 +2022,13 @@ public class KnowledgeDocumentImportTaskAppServiceTest { List.class ); method.setAccessible(true); - method.invoke(service, + Object rollbackResult = method.invoke(service, BigInteger.valueOf(701), BigInteger.valueOf(702), context, List.of(first, duplicate)); + Assert.assertEquals(Boolean.FALSE, rollbackResult); @SuppressWarnings("rawtypes") ArgumentCaptor idsCaptor = ArgumentCaptor.forClass(Collection.class); Mockito.verify(documentStore).delete(idsCaptor.capture(), Mockito.same(storeOptions)); @@ -1741,51 +2038,171 @@ public class KnowledgeDocumentImportTaskAppServiceTest { } /** - * 验证 AUTO 空结果使用默认长度与重叠参数回退自然段长度拆分。 + * 验证关键词索引删除失败会被回滚结果显式报告。 * * @throws Exception 反射调用异常 */ @Test - public void autoEmptyChunksShouldFallbackWithDefaultParagraphSettings() - throws Exception { - assertAutoParagraphFallback(RagDefaults.CHUNK_SIZE, RagDefaults.OVERLAP_SIZE); + public void rollbackStoredChunksShouldReportKeywordDeleteFailure() throws Exception { + KnowledgeDocumentImportTaskAppService service = new KnowledgeDocumentImportTaskAppService(); + DocumentStore documentStore = Mockito.mock(DocumentStore.class); + DocumentSearcher searcher = Mockito.mock(DocumentSearcher.class); + StoreOptions storeOptions = StoreOptions.ofCollectionName("knowledge-test"); + Mockito.when(documentStore.delete(Mockito.anyCollection(), Mockito.same(storeOptions))) + .thenReturn(StoreResult.success()); + Mockito.when(searcher.deleteDocuments(Mockito.anyCollection())).thenReturn(false); + + DocumentCollection knowledge = new DocumentCollection(); + knowledge.setId(BigInteger.valueOf(751)); + Class contextClass = Class.forName( + KnowledgeDocumentImportTaskAppService.class.getName() + "$StoreExecutionContext" + ); + Constructor constructor = contextClass.getDeclaredConstructor( + DocumentCollection.class, + EmbeddingModel.class, + DocumentStore.class, + StoreOptions.class, + DocumentSearcher.class + ); + constructor.setAccessible(true); + Object context = constructor.newInstance(knowledge, null, documentStore, storeOptions, searcher); + + DocumentChunk chunk = new DocumentChunk(); + chunk.setId(BigInteger.valueOf(752)); + Method method = KnowledgeDocumentImportTaskAppService.class.getDeclaredMethod( + "rollbackStoredChunks", + BigInteger.class, + BigInteger.class, + contextClass, + List.class + ); + method.setAccessible(true); + + Object rollbackResult = method.invoke(service, + BigInteger.valueOf(753), + BigInteger.valueOf(754), + context, + List.of(chunk)); + + Assert.assertEquals(Boolean.FALSE, rollbackResult); + Mockito.verify(searcher).deleteDocuments(List.of(chunk.getId())); } /** - * 验证手动预览选择 AUTO 时，兜底保留页面传入的长度与重叠参数。 + * 验证索引业务异常会保留准确原因，并在回滚失败时追加处置提示。 * * @throws Exception 反射调用异常 */ @Test - public void autoEmptyChunksShouldFallbackWithRequestedParagraphSettings() - throws Exception { - assertAutoParagraphFallback(768, 192); + public void resolveIndexFailureMessageShouldKeepBusinessReason() throws Exception { + KnowledgeDocumentImportTaskAppService service = new KnowledgeDocumentImportTaskAppService(); + Method method = KnowledgeDocumentImportTaskAppService.class.getDeclaredMethod( + "resolveIndexFailureMessage", + Exception.class, + boolean.class + ); + method.setAccessible(true); + + String message = (String) method.invoke( + service, + new BusinessException("关键词索引写入失败"), + false + ); + + Assert.assertEquals( + "关键词索引写入失败；外部索引回滚未完成，请联系管理员处理", + message + ); } /** - * 验证显式选择非 AUTO 策略时不触发兜底。 + * 验证关键词索引批量写入失败会终止导入，避免任务误报完成。 * * @throws Exception 反射调用异常 */ @Test - public void explicitStrategyEmptyChunksShouldNotFallback() throws Exception { - KnowledgeDocumentImportTaskAppService service = - new KnowledgeDocumentImportTaskAppService(); - RagIngestionService ragIngestionService = Mockito.mock(RagIngestionService.class); - setField(service, "ragIngestionService", ragIngestionService); - AnalysisResult analysis = Mockito.mock(AnalysisResult.class); - StrategyConfig requestedStrategy = StrategyConfig.defaults(); - requestedStrategy.setStrategyCode(RagStrategyCodes.OUTLINE_SECTION); - StrategyConfig effectiveStrategy = requestedStrategy.copy(); - Mockito.when(ragIngestionService.split(analysis, effectiveStrategy)) - .thenReturn(List.of()); + public void storeDocumentChunksShouldPropagateKeywordIndexFailure() throws Exception { + KnowledgeDocumentImportTaskAppService service = new KnowledgeDocumentImportTaskAppService(); + DocumentStore documentStore = Mockito.mock(DocumentStore.class); + DocumentSearcher searcher = Mockito.mock(DocumentSearcher.class); + StoreOptions storeOptions = StoreOptions.ofCollectionName("knowledge-test"); + Mockito.when(documentStore.store(Mockito.anyList(), Mockito.same(storeOptions))) + .thenReturn(StoreResult.success()); + Mockito.when(searcher.addDocuments(Mockito.anyList())).thenReturn(false); - List chunks = invokeAutoParagraphFallback( - service, analysis, requestedStrategy, effectiveStrategy); + DocumentCollection knowledge = new DocumentCollection(); + knowledge.setId(BigInteger.valueOf(801)); + Class contextClass = Class.forName( + KnowledgeDocumentImportTaskAppService.class.getName() + "$StoreExecutionContext" + ); + Constructor constructor = contextClass.getDeclaredConstructor( + DocumentCollection.class, + EmbeddingModel.class, + DocumentStore.class, + StoreOptions.class, + DocumentSearcher.class + ); + constructor.setAccessible(true); + Object context = constructor.newInstance(knowledge, null, documentStore, storeOptions, searcher); - Assert.assertTrue(chunks.isEmpty()); - Assert.assertEquals(RagStrategyCodes.OUTLINE_SECTION, effectiveStrategy.getStrategyCode()); - Mockito.verify(ragIngestionService).split(analysis, effectiveStrategy); + DocumentChunk chunk = new DocumentChunk(); + chunk.setId(BigInteger.valueOf(802)); + chunk.setContent("关键词索引失败测试"); + Method method = KnowledgeDocumentImportTaskAppService.class.getDeclaredMethod( + "storeDocumentChunks", + contextClass, + List.class + ); + method.setAccessible(true); + + try { + method.invoke(service, context, List.of(chunk)); + Assert.fail("关键词索引失败时应终止导入"); + } catch (InvocationTargetException expected) { + Assert.assertTrue(expected.getCause() instanceof BusinessException); + Assert.assertEquals("关键词索引写入失败", expected.getCause().getMessage()); + } + Mockito.verify(searcher).addDocuments(Mockito.anyList()); + } + + private static DocumentMapper mockDocumentMapper(tech.easyflow.ai.entity.Document persistedDocument, + AtomicReference updatedDocumentRef) { + return (DocumentMapper) Proxy.newProxyInstance( + DocumentMapper.class.getClassLoader(), + new Class[]{DocumentMapper.class}, + (proxy, method, args) -> { + if ("selectOneById".equals(method.getName())) { + return persistedDocument; + } + if ("update".equals(method.getName())) { + updatedDocumentRef.set((tech.easyflow.ai.entity.Document) args[0]); + return 1; + } + return defaultValue(method.getReturnType()); + } + ); + } + + private static FileStorageService mockFileStorageService(AtomicReference savedPrePathRef, + AtomicReference savedFilenameRef) { + return (FileStorageService) Proxy.newProxyInstance( + FileStorageService.class.getClassLoader(), + new Class[]{FileStorageService.class}, + (proxy, method, args) -> { + if ("save".equals(method.getName()) && args != null && args.length == 2 && args[0] instanceof MultipartFile file) { + savedPrePathRef.set((String) args[1]); + savedFilenameRef.set(file.getOriginalFilename()); + return "http://localhost:39000/easyflow/attachment/" + args[1] + "/" + file.getOriginalFilename(); + } + return defaultValue(method.getReturnType()); + } + ); + } + + private static void setField(Object target, String fieldName, Object value) throws Exception { + Field field = KnowledgeDocumentImportTaskAppService.class.getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); } private static void assertAutoParagraphFallback(int chunkSize, int overlapSize) throws Exception { @@ -1842,46 +2259,6 @@ public class KnowledgeDocumentImportTaskAppServiceTest { service, analysis, requestedStrategy, effectiveStrategy); } - private static DocumentMapper mockDocumentMapper(tech.easyflow.ai.entity.Document persistedDocument, - AtomicReference updatedDocumentRef) { - return (DocumentMapper) Proxy.newProxyInstance( - DocumentMapper.class.getClassLoader(), - new Class[]{DocumentMapper.class}, - (proxy, method, args) -> { - if ("selectOneById".equals(method.getName())) { - return persistedDocument; - } - if ("update".equals(method.getName())) { - updatedDocumentRef.set((tech.easyflow.ai.entity.Document) args[0]); - return 1; - } - return defaultValue(method.getReturnType()); - } - ); - } - - private static FileStorageService mockFileStorageService(AtomicReference savedPrePathRef, - AtomicReference savedFilenameRef) { - return (FileStorageService) Proxy.newProxyInstance( - FileStorageService.class.getClassLoader(), - new Class[]{FileStorageService.class}, - (proxy, method, args) -> { - if ("save".equals(method.getName()) && args != null && args.length == 2 && args[0] instanceof MultipartFile file) { - savedPrePathRef.set((String) args[1]); - savedFilenameRef.set(file.getOriginalFilename()); - return "http://localhost:39000/easyflow/attachment/" + args[1] + "/" + file.getOriginalFilename(); - } - return defaultValue(method.getReturnType()); - } - ); - } - - private static void setField(Object target, String fieldName, Object value) throws Exception { - Field field = KnowledgeDocumentImportTaskAppService.class.getDeclaredField(fieldName); - field.setAccessible(true); - field.set(target, value); - } - private static Object defaultValue(Class returnType) { if (returnType == boolean.class) { return false; diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplRollbackTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplRollbackTest.java new file mode 100644 index 00000000..196eead5 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplRollbackTest.java @@ -0,0 +1,100 @@ +package tech.easyflow.ai.service.impl; + +import com.easyagents.core.model.embedding.EmbeddingModel; +import com.easyagents.core.store.DocumentStore; +import com.easyagents.core.store.StoreOptions; +import com.easyagents.core.store.StoreResult; +import com.easyagents.search.engine.service.DocumentSearcher; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.Mockito; +import tech.easyflow.ai.entity.DocumentChunk; +import tech.easyflow.ai.entity.DocumentCollection; +import tech.easyflow.ai.entity.Model; + +import java.lang.reflect.Constructor; +import java.lang.reflect.Method; +import java.math.BigInteger; +import java.util.List; + +/** + * {@link DocumentServiceImpl} 外部索引回滚回归测试。 + */ +public class DocumentServiceImplRollbackTest { + + /** + * 验证手动导入回滚会检查关键词索引批量删除结果。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void rollbackStoredChunksShouldReportKeywordDeleteFailure() throws Exception { + DocumentServiceImpl service = new DocumentServiceImpl(); + DocumentStore documentStore = Mockito.mock(DocumentStore.class); + DocumentSearcher searcher = Mockito.mock(DocumentSearcher.class); + StoreOptions options = StoreOptions.ofCollectionName("knowledge-test"); + Mockito.when(documentStore.delete(Mockito.anyCollection(), Mockito.same(options))) + .thenReturn(StoreResult.success()); + Mockito.when(searcher.deleteDocuments(Mockito.anyCollection())).thenReturn(false); + + DocumentCollection knowledge = new DocumentCollection(); + knowledge.setId(BigInteger.valueOf(901)); + Class contextClass = Class.forName( + DocumentServiceImpl.class.getName() + "$StoreExecutionContext" + ); + Constructor constructor = contextClass.getDeclaredConstructor( + DocumentCollection.class, + Model.class, + EmbeddingModel.class, + DocumentStore.class, + StoreOptions.class, + DocumentSearcher.class + ); + constructor.setAccessible(true); + Object context = constructor.newInstance( + knowledge, null, null, documentStore, options, searcher); + + DocumentChunk chunk = new DocumentChunk(); + chunk.setId(BigInteger.valueOf(902)); + Method method = DocumentServiceImpl.class.getDeclaredMethod( + "rollbackStoredChunks", + contextClass, + List.class + ); + method.setAccessible(true); + + Object rollbackResult = method.invoke(service, context, List.of(chunk)); + + Assert.assertEquals(Boolean.FALSE, rollbackResult); + Mockito.verify(searcher).deleteDocuments(List.of(chunk.getId())); + } + + /** + * 验证手动导入会在外部索引回滚失败时追加明确处置提示。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void buildStoreFailureMessageShouldIncludeRollbackFailure() throws Exception { + DocumentServiceImpl service = new DocumentServiceImpl(); + Method method = DocumentServiceImpl.class.getDeclaredMethod( + "buildStoreFailureMessage", + String.class, + Exception.class, + boolean.class + ); + method.setAccessible(true); + + String message = (String) method.invoke( + service, + "提交导入失败：", + new IllegalStateException("关键词索引写入失败"), + false + ); + + Assert.assertEquals( + "提交导入失败：关键词索引写入失败；外部索引回滚未完成，请联系管理员处理", + message + ); + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplTest.java index 69f7abe7..743d5e4b 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/service/impl/DocumentServiceImplTest.java @@ -2,6 +2,7 @@ package tech.easyflow.ai.service.impl; import com.easyagents.core.store.DocumentStore; import com.easyagents.core.store.StoreResult; +import com.easyagents.rag.core.RagDefaults; import com.mybatisflex.core.query.QueryWrapper; import org.junit.Assert; import org.junit.Test; @@ -9,6 +10,7 @@ import org.mockito.ArgumentCaptor; import org.mockito.Mockito; import tech.easyflow.ai.config.SearcherFactory; import tech.easyflow.ai.entity.Document; +import tech.easyflow.ai.entity.DocumentChunk; import tech.easyflow.ai.entity.DocumentCollection; import tech.easyflow.ai.entity.Model; import tech.easyflow.ai.enums.DocumentProcessStatus; @@ -20,6 +22,8 @@ import tech.easyflow.common.filestorage.FileStorageService; import tech.easyflow.common.web.exceptions.BusinessException; import java.lang.reflect.Field; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; import java.math.BigInteger; import java.util.List; import java.util.Locale; @@ -304,6 +308,33 @@ public class DocumentServiceImplTest { Mockito.verifyNoInteractions(storageService); } + /** + * 验证旧版向量化入口同样拒绝超过 BGE-M3 上下文预算的分块。 + * + * @throws Exception 反射调用异常 + */ + @Test + public void embeddingGuardShouldRejectOversizedChunk() + throws Exception { + DocumentServiceImpl service = new DocumentServiceImpl(); + DocumentChunk chunk = new DocumentChunk(); + chunk.setId(BigInteger.valueOf(8801)); + chunk.setContent("长".repeat( + RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT + 1)); + Method method = DocumentServiceImpl.class.getDeclaredMethod( + "assertEmbeddingChunkWithinHardLimit", DocumentChunk.class); + method.setAccessible(true); + + try { + method.invoke(service, chunk); + Assert.fail("超过 BGE-M3 上下文预算的分块应被拒绝"); + } catch (InvocationTargetException expected) { + Assert.assertTrue(expected.getCause() instanceof BusinessException); + Assert.assertTrue(expected.getCause().getMessage().contains( + "tokenEstimate")); + } + } + /** * 通过反射注入测试依赖。 *