diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentService.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentService.java index 4e912913..d4e971b6 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentService.java +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentService.java @@ -292,8 +292,11 @@ public class MuseKnowledgeDocumentService { uploadResult.externalId(), parseResult); return new UploadOutcome(ragflowContext, "failed"); } + // Dify 摄入把索引批次 batch 透出到 uploadResult.summary().get("batch"),是其 indexing-status 轮询主键; + // RAGFlow 无此字段 → runtimeBatchId 为 null,轮询回退 documentId、行为逐字不变。 + String runtimeBatchId = uploadResult.summary() == null ? null : (String) uploadResult.summary().get("batch"); processingTaskService.markRagflowParseAccepted(ragflowContext.version().getId(), ragflowContext.task().getTaskId(), - ragflowContext.ragflowDatasetId(), uploadResult.externalId(), parseResult); + ragflowContext.ragflowDatasetId(), uploadResult.externalId(), runtimeBatchId, parseResult); return new UploadOutcome(ragflowContext, "processing"); } diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorker.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorker.java index 069721fb..645a6996 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorker.java +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorker.java @@ -76,8 +76,11 @@ public class MuseKnowledgeParseStatusPollWorker { int advanced = 0; for (MuseKnowledgeProcessingTaskDO task : tasks) { // 缺租户/外部 id 的任务无法安全轮询,跳过(避免空指针与跨租户写串)。 + // 轮询键防御性放宽:ragflowDocumentId 与 runtimeBatchId 都空才跳过——Dify 场景两者都会有值, + // RAGFlow 只有 documentId,任一非空即可据以轮询。 if (task.getTenantId() == null || !StringUtils.hasText(task.getRagflowDatasetId()) - || !StringUtils.hasText(task.getRagflowDocumentId())) { + || (!StringUtils.hasText(task.getRagflowDocumentId()) + && !StringUtils.hasText(task.getRuntimeBatchId()))) { continue; } advanced += TenantUtils.execute(task.getTenantId(), () -> pollOne(task)); @@ -87,10 +90,14 @@ public class MuseKnowledgeParseStatusPollWorker { private int pollOne(MuseKnowledgeProcessingTaskDO task) { KnowledgeRuntimeClient.RuntimeResult result; + // 轮询主键择键:Dify 上传落 runtimeBatchId(indexing-status 按 batch 查)→ 优先用之; + // RAGFlow 无 batch → 回退 ragflowDocumentId,保持原逐字行为。skip gate 已保证两者至少一非空,pollKey 不为 null。 + String pollKey = StringUtils.hasText(task.getRuntimeBatchId()) + ? task.getRuntimeBatchId() : task.getRagflowDocumentId(); try { result = ragFlowClient.pollDocumentStatuses(new KnowledgeRuntimeClient.PollDocumentStatusesCommand( task.getTenantId(), task.getOwnerUserId(), task.getKbId(), task.getRagflowDatasetId(), - List.of(task.getRagflowDocumentId()), task.getTaskId(), + List.of(pollKey), task.getTaskId(), pollCorrelationId(task), pollRequestHash(task), 1)); } catch (RuntimeException ex) { // RAGFlow 临时不可达:保持 parsing,下轮重试,不误判失败。 diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeProcessingTaskService.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeProcessingTaskService.java index 4962d48c..fff224a4 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeProcessingTaskService.java +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeProcessingTaskService.java @@ -20,6 +20,7 @@ import jakarta.annotation.Resource; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; +import org.springframework.util.StringUtils; import java.time.LocalDateTime; import java.util.LinkedHashMap; @@ -178,7 +179,8 @@ public class MuseKnowledgeProcessingTaskService { @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class) public void markRagflowParseAccepted(Long documentVersionId, String taskId, String ragflowDatasetId, - String ragflowDocumentId, KnowledgeRuntimeClient.RuntimeResult result) { + String ragflowDocumentId, String runtimeBatchId, + KnowledgeRuntimeClient.RuntimeResult result) { MuseKnowledgeProcessingTaskDO task = processingTaskMapper.selectByTaskId(taskId); if (task != null) { task.setStatus("parsing"); @@ -187,6 +189,11 @@ public class MuseKnowledgeProcessingTaskService { // 新建 dataset 的上传链路必须把外部 datasetId 同步固化到 task,后续轮询、补偿和 live 取证都依赖这条事实。 task.setRagflowDatasetId(ragflowDatasetId); task.setRagflowDocumentId(ragflowDocumentId); + // Dify 上传返回的索引批次是其 indexing-status 轮询主键,必须固化;RAGFlow 无 batch → runtimeBatchId 为空, + // 用 hasText 判定避免把已落值覆盖成 null(poll worker 据此在 batch / documentId 间择键)。 + if (StringUtils.hasText(runtimeBatchId)) { + task.setRuntimeBatchId(runtimeBatchId); + } task.setRetryable(false); task.setProgressMessage("RAGFlow parse accepted; polling required"); task.setResultSummary(JsonUtils.toJsonString(Map.of("ragflow", safeSummary(result)))); diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/dataobject/muse/MuseKnowledgeProcessingTaskDO.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/dataobject/muse/MuseKnowledgeProcessingTaskDO.java index 9dded18c..bd505508 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/dataobject/muse/MuseKnowledgeProcessingTaskDO.java +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/main/java/cn/iocoder/muse/module/knowledge/dal/dataobject/muse/MuseKnowledgeProcessingTaskDO.java @@ -42,6 +42,12 @@ public class MuseKnowledgeProcessingTaskDO extends TenantBaseDO { private String status; private String ragflowDatasetId; private String ragflowDocumentId; + /** + * 运行时索引批次 id:Dify Datasets 摄入返回的顶层 batch,是其 indexing-status 轮询主键 + * (GET /v1/datasets/{id}/documents/{batch}/indexing-status)。RAGFlow 上传不产生 batch,留空, + * 轮询回退到 ragflowDocumentId,行为逐字不变。命名沿用 provider 无关口径,避免与 ragflowDocumentId 混淆。 + */ + private String runtimeBatchId; /** RAGFlow document run/progress 轮询得到的解析进度。 */ private Integer parseProgress; private String progressMessage; diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentServiceTest.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentServiceTest.java index 11848d31..2a6a2195 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentServiceTest.java +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeDocumentServiceTest.java @@ -323,11 +323,12 @@ class MuseKnowledgeDocumentServiceTest extends BaseMockitoUnitTest { }); when(processingTaskMapper.selectByTaskId("knowledge-processing-6001")).thenReturn(processingTask(6001L)); when(processingTaskMapper.insert(any(MuseKnowledgeProcessingTaskDO.class))).thenAnswer(invocation -> 1); + // 模拟 Dify 上传把索引批次 batch 透出到 summary,验证其被抽取并落库到 task.runtimeBatchId(RAGFlow 无此字段)。 when(ragFlowClient.uploadDocuments(any())).thenReturn(new KnowledgeRuntimeClient.RuntimeResult( "rag-doc-1", KnowledgeRuntimeClient.Operation.UPLOAD_DOCUMENTS, "uploadGlobalKBDocument:cmd-upload-file-success", "hash-file-success", 1, 12L, KnowledgeRuntimeClient.Status.SUCCEEDED, null, null, - "{\"status\":\"succeeded\"}", List.of(), java.util.Map.of())); + "{\"status\":\"succeeded\"}", List.of(), java.util.Map.of("batch", "dify-batch-1"))); when(ragFlowClient.startParseDocuments(any())).thenReturn(new KnowledgeRuntimeClient.RuntimeResult( null, KnowledgeRuntimeClient.Operation.START_PARSE_DOCUMENTS, "uploadGlobalKBDocument:cmd-upload-file-success", "hash-file-success", 1, 8L, @@ -380,6 +381,7 @@ class MuseKnowledgeDocumentServiceTest extends BaseMockitoUnitTest { "parsing".equals(task.getStatus()) && "rag-dataset-1".equals(task.getRagflowDatasetId()) && "rag-doc-1".equals(task.getRagflowDocumentId()) + && "dify-batch-1".equals(task.getRuntimeBatchId()) && "processing".equals(task.getResultSummary()) == false)); } diff --git a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorkerTest.java b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorkerTest.java index 25dc4763..85d075bf 100644 --- a/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorkerTest.java +++ b/muse-cloud/muse-module-knowledge/muse-module-knowledge-server/src/test/java/cn/iocoder/muse/module/knowledge/application/muse/MuseKnowledgeParseStatusPollWorkerTest.java @@ -9,6 +9,7 @@ import cn.iocoder.muse.module.knowledge.application.muse.facade.KnowledgeRuntime import cn.iocoder.muse.module.knowledge.dal.dataobject.muse.MuseKnowledgeProcessingTaskDO; import cn.iocoder.muse.module.knowledge.dal.mysql.muse.MuseKnowledgeProcessingTaskMapper; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import java.util.List; @@ -129,6 +130,7 @@ class MuseKnowledgeParseStatusPollWorkerTest extends BaseMockitoUnitTest { @Test void should_skip_whenMissingRagflowIds() { + // documentId 与 runtimeBatchId 都空 → 无可用轮询键 → 跳过。 MuseKnowledgeProcessingTaskDO t = parsingTask(); t.setRagflowDocumentId(null); when(processingTaskMapper.selectParsingForPoll(anyInt())).thenReturn(List.of(t)); @@ -138,4 +140,37 @@ class MuseKnowledgeParseStatusPollWorkerTest extends BaseMockitoUnitTest { assertEquals(0, advanced); verifyNoInteractions(ragFlowClient); } + + @Test + void should_pollByRuntimeBatchId_whenDifyBatchPresent() { + // Dify 场景:上传落 runtimeBatchId(=索引 batch)与 ragflowDocumentId(=document.id);轮询主键取 batch。 + MuseKnowledgeProcessingTaskDO t = parsingTask(); + t.setRuntimeBatchId("dify-batch-9"); + when(processingTaskMapper.selectParsingForPoll(anyInt())).thenReturn(List.of(t)); + when(ragFlowClient.pollDocumentStatuses(any())).thenReturn( + poll(Status.SUCCEEDED, new DocumentStatus("doc-1", "processing", "RUNNING", 50, "running"))); + + worker(true).pollOnce(); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(KnowledgeRuntimeClient.PollDocumentStatusesCommand.class); + verify(ragFlowClient).pollDocumentStatuses(captor.capture()); + assertEquals(List.of("dify-batch-9"), captor.getValue().ragflowDocumentIds()); + } + + @Test + void should_pollByDocumentId_whenRuntimeBatchIdAbsent() { + // RAGFlow 场景:无 batch,runtimeBatchId 留空 → 轮询回退 ragflowDocumentId,逐字不变。 + MuseKnowledgeProcessingTaskDO t = parsingTask(); + when(processingTaskMapper.selectParsingForPoll(anyInt())).thenReturn(List.of(t)); + when(ragFlowClient.pollDocumentStatuses(any())).thenReturn( + poll(Status.SUCCEEDED, new DocumentStatus("doc-1", "processing", "RUNNING", 50, "running"))); + + worker(true).pollOnce(); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(KnowledgeRuntimeClient.PollDocumentStatusesCommand.class); + verify(ragFlowClient).pollDocumentStatuses(captor.capture()); + assertEquals(List.of("doc-1"), captor.getValue().ragflowDocumentIds()); + } } diff --git a/muse-cloud/sql/muse/V37__add_knowledge_processing_task_runtime_batch_id.sql b/muse-cloud/sql/muse/V37__add_knowledge_processing_task_runtime_batch_id.sql new file mode 100644 index 00000000..bb502504 --- /dev/null +++ b/muse-cloud/sql/muse/V37__add_knowledge_processing_task_runtime_batch_id.sql @@ -0,0 +1,12 @@ +-- S6 知识摄入轮询键从 documentId 泛化到 batch:为 Dify Datasets 索引状态轮询补运行时批次列。 +-- +-- 背景:KnowledgeRuntimeClient 是 provider 无关端口(muse.knowledge.runtime-provider 切换 ragflow / dify)。 +-- Dify uploadDocuments 返回 document.id(落 ragflow_document_id)+ 顶层 batch,其索引状态轮询打 +-- GET /v1/datasets/{id}/documents/{batch}/indexing-status,轮询主键=batch;RAGFlow 仍按 documentId 轮询、 +-- 不产生 batch。此前摄入 caller 未持久化 batch,poll worker 只能传 documentId → Dify 轮询打错端点必坏。 +-- 本迁移新增可空列承载 Dify 批次:Dify 上传落 batch → 按 batch 轮询;RAGFlow 留空 → 回退 documentId、行为逐字不变。 +-- +-- 迁移方式:沿用本目录 append-only 约定(V24/V27/V30/V31 均以新迁移 ALTER 既有表),不回编 V14 建表语句, +-- 避免破坏已应用库(dev / live)的 Flyway 校验和。 +ALTER TABLE muse_knowledge_processing_task + ADD COLUMN runtime_batch_id VARCHAR(64);