Skip to content

Commit 0c56729

Browse files
ltrinthewindltrinthewind
committed
feat(knowledge): 实现文档资源异步清理
- 使用 RocketMQ 事务消息投递文档清理事件 - 在本地事务中删除文档、Chunk、调度、日志及 PgVector 数据 - 事务提交后异步清理 Milvus、ES、LightRAG 和对象文件 - 补充事务回查、幂等消费、失败重试及存储分阶段测试
1 parent 2911e60 commit 0c56729

17 files changed

Lines changed: 880 additions & 59 deletions

‎bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/KnowledgeBaseCleanupConsumer.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -93,7 +93,7 @@ public void onMessage(MessageWrapper<KnowledgeBaseCleanupEvent> message) {
9393
LightRagClient lightRagClient = lightRagClientProvider.getIfAvailable();
9494
if (lightRagClient != null) {
9595
try {
96-
lightRagClient.deleteByCollection(collectionName);
96+
lightRagClient.deleteByCollectionOrThrow(collectionName);
9797
} catch (Exception e) {
9898
allSucceeded = false;
9999
log.error("删除 LightRAG 图谱数据失败，collectionName={}", collectionName, e);

‎bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/KnowledgeDocumentCleanupConsumer.java‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package com.nageoffer.ai.ragent.knowledge.mq;
19+
20+
import com.nageoffer.ai.ragent.framework.exception.ServiceException;
21+
import com.nageoffer.ai.ragent.framework.mq.MessageWrapper;
22+
import com.nageoffer.ai.ragent.knowledge.mq.event.KnowledgeDocumentCleanupEvent;
23+
import com.nageoffer.ai.ragent.rag.core.graph.LightRagClient;
24+
import com.nageoffer.ai.ragent.rag.core.keyword.KeywordIndexService;
25+
import com.nageoffer.ai.ragent.rag.core.vector.VectorStoreService;
26+
import com.nageoffer.ai.ragent.rag.service.FileStorageService;
27+
import lombok.RequiredArgsConstructor;
28+
import lombok.extern.slf4j.Slf4j;
29+
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
30+
import org.apache.rocketmq.spring.core.RocketMQListener;
31+
import org.springframework.beans.factory.ObjectProvider;
32+
import org.springframework.stereotype.Component;
33+
import org.springframework.util.StringUtils;
34+
35+
/**
36+
* 文档删除清理消费者
37+
* <p>
38+
* 每个外部资源独立尝试，全部尝试结束后只要存在真实失败就抛错触发 MQ 重试。所有删除操作均须幂等
39+
*/
40+
@Slf4j
41+
@Component
42+
@RequiredArgsConstructor
43+
@RocketMQMessageListener(
44+
topic = "knowledge-document-cleanup_topic${unique-name:}",
45+
consumerGroup = "knowledge-document-cleanup_cg${unique-name:}"
46+
)
47+
public class KnowledgeDocumentCleanupConsumer
48+
implements RocketMQListener<MessageWrapper<KnowledgeDocumentCleanupEvent>> {
49+
50+
private final VectorStoreService vectorStoreService;
51+
private final FileStorageService fileStorageService;
52+
private final ObjectProvider<KeywordIndexService> keywordIndexServiceProvider;
53+
private final ObjectProvider<LightRagClient> lightRagClientProvider;
54+
55+
@Override
56+
public void onMessage(MessageWrapper<KnowledgeDocumentCleanupEvent> message) {
57+
KnowledgeDocumentCleanupEvent event = message.getBody();
58+
String docId = event.getDocId();
59+
String collectionName = event.getCollectionName();
60+
61+
log.info("[消费者] 开始清理文档外部资源，docId={}, collectionName={}", docId, collectionName);
62+
boolean allSucceeded = true;
63+
64+
try {
65+
// 此阶段仅会让外部向量数据库 Milvus 执行清理(若使用)
66+
vectorStoreService.deleteDocumentVectorsAfterCommit(collectionName, docId);
67+
} catch (Exception e) {
68+
allSucceeded = false;
69+
log.error("删除文档主向量失败，collectionName={}, docId={}", collectionName, docId, e);
70+
}
71+
72+
KeywordIndexService keywordIndexService = keywordIndexServiceProvider.getIfAvailable();
73+
if (keywordIndexService != null) {
74+
try {
75+
keywordIndexService.deleteDocumentIndex(collectionName, docId);
76+
} catch (Exception e) {
77+
allSucceeded = false;
78+
log.error("删除文档 ES 索引失败，collectionName={}, docId={}", collectionName, docId, e);
79+
}
80+
}
81+
82+
LightRagClient lightRagClient = lightRagClientProvider.getIfAvailable();
83+
if (lightRagClient != null) {
84+
try {
85+
lightRagClient.deleteByDocOrThrow(docId);
86+
} catch (Exception e) {
87+
allSucceeded = false;
88+
log.error("删除文档 LightRAG 数据失败，docId={}", docId, e);
89+
}
90+
}
91+
92+
if (StringUtils.hasText(event.getFileUrl())) {
93+
try {
94+
fileStorageService.deleteByUrl(event.getFileUrl());
95+
} catch (Exception e) {
96+
allSucceeded = false;
97+
log.error("删除文档对象文件失败，docId={}, fileUrl={}", docId, event.getFileUrl(), e);
98+
}
99+
}
100+
101+
if (!allSucceeded) {
102+
throw new ServiceException("文档外部资源清理存在失败项，触发重试");
103+
}
104+
}
105+
}

‎bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/KnowledgeDocumentCleanupTransactionChecker.java‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package com.nageoffer.ai.ragent.knowledge.mq;
19+
20+
import cn.hutool.json.JSONUtil;
21+
import com.nageoffer.ai.ragent.framework.mq.MessageWrapper;
22+
import com.nageoffer.ai.ragent.framework.mq.producer.DelegatingTransactionListener;
23+
import com.nageoffer.ai.ragent.framework.mq.producer.TransactionChecker;
24+
import com.nageoffer.ai.ragent.knowledge.dao.entity.KnowledgeDocumentDO;
25+
import com.nageoffer.ai.ragent.knowledge.dao.mapper.KnowledgeDocumentMapper;
26+
import com.nageoffer.ai.ragent.knowledge.mq.event.KnowledgeDocumentCleanupEvent;
27+
import jakarta.annotation.PostConstruct;
28+
import lombok.RequiredArgsConstructor;
29+
import lombok.extern.slf4j.Slf4j;
30+
import org.springframework.beans.factory.annotation.Value;
31+
import org.springframework.stereotype.Component;
32+
33+
/**
34+
* 文档删除清理事务消息回查器
35+
* <p>
36+
* 只以文档是否已逻辑删除作为本地事务提交凭据
37+
*/
38+
@Slf4j
39+
@Component
40+
@RequiredArgsConstructor
41+
public class KnowledgeDocumentCleanupTransactionChecker implements TransactionChecker {
42+
43+
private final KnowledgeDocumentMapper documentMapper;
44+
private final DelegatingTransactionListener transactionListener;
45+
46+
@Value("knowledge-document-cleanup_topic${unique-name:}")
47+
private String cleanupTopic;
48+
49+
@PostConstruct
50+
public void init() {
51+
transactionListener.registerChecker(cleanupTopic, this);
52+
}
53+
54+
@Override
55+
public boolean check(MessageWrapper<?> message) {
56+
log.info("[事务回查] 文档删除清理，消息体：{}", JSONUtil.toJsonStr(message));
57+
58+
KnowledgeDocumentCleanupEvent event = (KnowledgeDocumentCleanupEvent) message.getBody();
59+
// @TableLogic 会让已删除行对 selectById 不可见,删除事务消息只会在确认文档存在后发送
60+
KnowledgeDocumentDO document = documentMapper.selectById(event.getDocId());
61+
return document == null || Integer.valueOf(1).equals(document.getDeleted());
62+
}
63+
}

‎bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/mq/event/KnowledgeDocumentCleanupEvent.java‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package com.nageoffer.ai.ragent.knowledge.mq.event;
19+
20+
import lombok.AllArgsConstructor;
21+
import lombok.Builder;
22+
import lombok.Data;
23+
import lombok.NoArgsConstructor;
24+
25+
import java.io.Serial;
26+
import java.io.Serializable;
27+
28+
/**
29+
* 文档删除后的外部资源清理事件。
30+
* <p>
31+
* 文档 ID 永不复用，逻辑删除不可恢复，因此不可变资源标识足以支持幂等重试，无需携带操作版本。
32+
*/
33+
@Data
34+
@Builder
35+
@NoArgsConstructor
36+
@AllArgsConstructor
37+
public class KnowledgeDocumentCleanupEvent implements Serializable {
38+
39+
@Serial
40+
private static final long serialVersionUID = 1L;
41+
42+
private String docId;
43+
44+
private String collectionName;
45+
46+
private String fileUrl;
47+
}

‎bootstrap/src/main/java/com/nageoffer/ai/ragent/knowledge/service/impl/KnowledgeDocumentServiceImpl.java‎

Lines changed: 44 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,6 @@
3838
import com.nageoffer.ai.ragent.core.ingest.IngestionOutcome;
3939
import com.nageoffer.ai.ragent.core.ingest.IngestionSpec;
4040
import com.nageoffer.ai.ragent.core.ingest.VectorTarget;
41-
import com.nageoffer.ai.ragent.core.ingest.sink.ChunkIndexWriter;
4241
import com.nageoffer.ai.ragent.core.parser.registry.ParserRegistry;
4342
import com.nageoffer.ai.ragent.framework.context.UserContext;
4443
import com.nageoffer.ai.ragent.framework.exception.ClientException;
@@ -70,6 +69,7 @@
7069
import com.nageoffer.ai.ragent.knowledge.enums.SourceType;
7170
import com.nageoffer.ai.ragent.knowledge.handler.RemoteFileFetcher;
7271
import com.nageoffer.ai.ragent.knowledge.mq.event.KnowledgeDocumentChunkEvent;
72+
import com.nageoffer.ai.ragent.knowledge.mq.event.KnowledgeDocumentCleanupEvent;
7373
import com.nageoffer.ai.ragent.knowledge.schedule.CronScheduleHelper;
7474
import com.nageoffer.ai.ragent.knowledge.schedule.DocumentStatusHelper;
7575
import com.nageoffer.ai.ragent.knowledge.service.KnowledgeChunkService;
@@ -104,7 +104,6 @@ public class KnowledgeDocumentServiceImpl implements KnowledgeDocumentService {
104104
private final KnowledgeDocumentMapper documentMapper;
105105
private final ParserRegistry parserRegistry;
106106
private final IngestionKernel ingestionKernel;
107-
private final ChunkIndexWriter chunkIndexWriter;
108107
private final IngestionSpecCodec ingestionSpecCodec;
109108
private final FileStorageService fileStorageService;
110109
private final VectorStoreService vectorStoreService;
@@ -127,6 +126,9 @@ public class KnowledgeDocumentServiceImpl implements KnowledgeDocumentService {
127126
@Value("knowledge-document-chunk_topic${unique-name:}")
128127
private String chunkTopic;
129128

129+
@Value("knowledge-document-cleanup_topic${unique-name:}")
130+
private String cleanupTopic;
131+
130132
@Override
131133
@LogRecord(
132134
success = "上传文档：{{#bizChangeName}}",
@@ -416,7 +418,6 @@ private void markChunkFailed(String docId, String documentVersion) {
416418
}
417419

418420
@Override
419-
@Transactional(rollbackFor = Exception.class)
420421
@LogRecord(
421422
success = "删除文档：{{#bizChangeName}}",
422423
fail = "删除文档失败：{{#_errorMsg}}",
@@ -431,36 +432,49 @@ public void delete(String docId) {
431432
Assert.notNull(documentDO, () -> new ClientException("文档不存在"));
432433
bizChangeLogContext.putName(documentDO.getDocName());
433434
KnowledgeDocumentDO before = BeanUtil.copyProperties(documentDO, KnowledgeDocumentDO.class);
435+
KnowledgeBaseDO kbDO = knowledgeBaseMapper.selectById(documentDO.getKbId());
436+
Assert.notNull(kbDO, () -> new ClientException("知识库不存在"));
437+
String collectionName = kbDO.getCollectionName();
438+
String operator = UserContext.getUsername();
439+
String snapshotVersion = documentDO.getDocumentVersion();
440+
KnowledgeDocumentCleanupEvent event = KnowledgeDocumentCleanupEvent.builder()
441+
.docId(docId)
442+
.collectionName(collectionName)
443+
.fileUrl(documentDO.getFileUrl())
444+
.build();
434445

435-
String deletingVersion = documentStatusHelper.tryMarkDeleting(
436-
docId, documentDO.getDocumentVersion(), UserContext.getUsername());
437-
if (deletingVersion == null) {
438-
throw new ClientException("文档状态或版本已变化，请稍后重试");
439-
}
440-
documentDO.setStatus(DocumentStatus.DELETING.getCode());
441-
documentDO.setDocumentVersion(deletingVersion);
442-
443-
scheduleService.deleteByDocId(docId);
444-
chunkLogMapper.delete(Wrappers.lambdaQuery(KnowledgeDocumentChunkLogDO.class)
445-
.eq(KnowledgeDocumentChunkLogDO::getDocId, docId));
446+
// half 消息成功后执行本地事务；提交后由消费者清理 Milvus、ES、LightRAG 与对象文件。
447+
messageQueueProducer.sendInTransaction(
448+
cleanupTopic,
449+
docId,
450+
"文档删除清理",
451+
event,
452+
ignored -> {
453+
String deletingVersion = documentStatusHelper.tryMarkDeleting(docId, snapshotVersion, operator);
454+
if (deletingVersion == null) {
455+
throw new ClientException("文档状态或版本已变化，请稍后重试");
456+
}
446457

447-
int deleted = documentMapper.update(
448-
Wrappers.lambdaUpdate(KnowledgeDocumentDO.class)
449-
.set(KnowledgeDocumentDO::getDeleted, 1)
450-
.set(KnowledgeDocumentDO::getUpdatedBy, UserContext.getUsername())
451-
.set(KnowledgeDocumentDO::getUpdateTime, new Date())
452-
.eq(KnowledgeDocumentDO::getId, docId)
453-
.eq(KnowledgeDocumentDO::getDeleted, 0)
454-
.eq(KnowledgeDocumentDO::getStatus, DocumentStatus.DELETING.getCode())
455-
.eq(KnowledgeDocumentDO::getDocumentVersion, deletingVersion));
456-
if (deleted != 1) {
457-
throw new ClientException("文档删除所有权已失效");
458-
}
458+
int deleted = documentMapper.update(
459+
Wrappers.lambdaUpdate(KnowledgeDocumentDO.class)
460+
.set(KnowledgeDocumentDO::getDeleted, 1)
461+
.set(KnowledgeDocumentDO::getUpdatedBy, operator)
462+
.set(KnowledgeDocumentDO::getUpdateTime, new Date())
463+
.eq(KnowledgeDocumentDO::getId, docId)
464+
.eq(KnowledgeDocumentDO::getDeleted, 0)
465+
.eq(KnowledgeDocumentDO::getStatus, DocumentStatus.DELETING.getCode())
466+
.eq(KnowledgeDocumentDO::getDocumentVersion, deletingVersion));
467+
if (deleted != 1) {
468+
throw new ClientException("文档删除所有权已失效");
469+
}
459470

460-
// 一次调用覆盖全部落点：关系库块与向量都在扇出里，未来加索引后端也自动跟随
461-
KnowledgeBaseDO kbDO = knowledgeBaseMapper.selectById(documentDO.getKbId());
462-
chunkIndexWriter.deleteDocument(vectorTargetResolver.resolve(kbDO), documentRef(documentDO));
463-
deleteStoredFileQuietly(documentDO);
471+
scheduleService.deleteByDocId(docId);
472+
chunkLogMapper.delete(Wrappers.lambdaQuery(KnowledgeDocumentChunkLogDO.class)
473+
.eq(KnowledgeDocumentChunkLogDO::getDocId, docId));
474+
chunkMapper.delete(Wrappers.lambdaQuery(KnowledgeChunkDO.class)
475+
.eq(KnowledgeChunkDO::getDocId, docId));
476+
vectorStoreService.deleteDocumentVectorsInTransaction(collectionName, docId);
477+
});
464478
bizChangeLogContext.put(docId, before, null);
465479
}
466480

@@ -885,14 +899,4 @@ public String preview(String docId) {
885899
}
886900
}
887901

888-
private void deleteStoredFileQuietly(KnowledgeDocumentDO documentDO) {
889-
if (documentDO == null || !StringUtils.hasText(documentDO.getFileUrl())) {
890-
return;
891-
}
892-
try {
893-
fileStorageService.deleteByUrl(documentDO.getFileUrl());
894-
} catch (Exception e) {
895-
log.warn("删除文档存储文件失败, docId={}, fileUrl={}", documentDO.getId(), documentDO.getFileUrl(), e);
896-
}
897-
}
898902
}

0 commit comments

Comments
 (0)