修复知识库文件解析和删除异常

This commit is contained in:
baiyanyun
2026-07-13 09:08:07 +08:00
parent f8ee4023ff
commit a69bb4b8a8
10 changed files with 175 additions and 62 deletions

View File

@@ -12,8 +12,11 @@ import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException; import java.net.MalformedURLException;
import java.net.URL; import java.net.URL;
import java.net.URLConnection;
import java.util.Objects; import java.util.Objects;
import java.util.UUID; import java.util.UUID;
@@ -150,6 +153,26 @@ public class FileAccessServiceImpl implements IFileAccessService {
} }
} }
@Override
public InputStream openFileStream(String fileUrl) throws IOException {
String fileKey = extractFileKeyFromUrl(fileUrl);
if (StringUtils.isNotBlank(fileKey)) {
FileRecordDomain fileRecord = fileManagementService.getFileByKey(fileKey);
if (fileRecord != null) {
try {
return fileManagementService.downloadFile(fileKey);
} catch (RuntimeException e) {
throw new IOException("Failed to open stored file: " + fileKey, e);
}
}
}
URLConnection connection = new URL(fileUrl).openConnection();
connection.setConnectTimeout(5000);
connection.setReadTimeout(30000);
return connection.getInputStream();
}
private static String getUrlPath(String fileUrl) throws MalformedURLException { private static String getUrlPath(String fileUrl) throws MalformedURLException {
URL url; URL url;
url = new URL(fileUrl); url = new URL(fileUrl);
@@ -172,4 +195,20 @@ public class FileAccessServiceImpl implements IFileAccessService {
return null; return null;
} }
private String extractFileKeyFromUrl(String fileUrl) {
if (StringUtils.isBlank(fileUrl)) {
return null;
}
int apiIndex = fileUrl.indexOf("/api/f/");
if (apiIndex < 0) {
return null;
}
String fileKey = fileUrl.substring(apiIndex + "/api/f/".length());
int queryIndex = fileKey.indexOf('?');
if (queryIndex >= 0) {
fileKey = fileKey.substring(0, queryIndex);
}
return StringUtils.isBlank(fileKey) ? null : fileKey;
}
} }

View File

@@ -1,5 +1,8 @@
package com.xspaceagi.file.sdk; package com.xspaceagi.file.sdk;
import java.io.IOException;
import java.io.InputStream;
public interface IFileAccessService { public interface IFileAccessService {
String getFileUrlWithAk(String fileUrl); String getFileUrlWithAk(String fileUrl);
@@ -9,4 +12,6 @@ public interface IFileAccessService {
void checkFileUrlAk(String uri, String ak); void checkFileUrlAk(String uri, String ak);
void checkFileUrlAk0(String uri, String ak); void checkFileUrlAk0(String uri, String ak);
InputStream openFileStream(String fileUrl) throws IOException;
} }

View File

@@ -267,16 +267,12 @@ public class KnowledgeDocumentApplicationService implements IKnowledgeDocumentAp
var kbId = existObj.getKbId(); var kbId = existObj.getKbId();
this.vectorDBService.removeDoc(kbId, docId); this.vectorDBService.removeDoc(kbId, docId);
// 同步删除全文检索数据(在事务内) // 全文检索是外部索引,删除失败不应阻断文档重试。
// 注意:重试时需要删除全文检索数据,但这里是在 Application 层,不要直接调用 Domain 层服务
// 建议:这个方法应该移到 Domain 层,或者使用 Application 层的 fullTextSyncService
try { try {
fullTextSyncService.deleteDocumentFromQuickwit(docId, kbId, userContext); fullTextSyncService.deleteDocumentFromQuickwit(docId, kbId, userContext);
log.info("文档重试时删除全文检索数据成功: docId={}, kbId={}", docId, kbId); log.info("文档重试时删除全文检索数据成功: docId={}, kbId={}", docId, kbId);
} catch (Exception e) { } catch (Exception e) {
log.error("文档重试时删除全文检索数据失败: docId={}, kbId={}", docId, kbId, e); log.warn("文档重试时删除全文检索数据失败,已忽略: docId={}, kbId={}", docId, kbId, e);
// 抛出异常,触发事务回滚
throw KnowledgeException.build(BizExceptionCodeEnum.knowledgeDeleteDocFulltextFailed, e);
} }
// 删除重试任务 // 删除重试任务

View File

@@ -3,6 +3,7 @@ package com.xspaceagi.knowledge.domain.docparser.impl;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject; import com.alibaba.fastjson2.JSONObject;
import com.xspaceagi.file.sdk.IFileAccessService;
import com.xspaceagi.knowledge.core.spec.utils.Constants; import com.xspaceagi.knowledge.core.spec.utils.Constants;
import com.xspaceagi.knowledge.domain.docparser.parse.DocParser; import com.xspaceagi.knowledge.domain.docparser.parse.DocParser;
import com.xspaceagi.knowledge.domain.model.KnowledgeDocumentModel; import com.xspaceagi.knowledge.domain.model.KnowledgeDocumentModel;
@@ -16,7 +17,6 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.io.InputStream; import java.io.InputStream;
import java.net.URL;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
@@ -28,9 +28,12 @@ public class JSONParser implements DocParser {
@Resource @Resource
private IKnowledgeRawSegmentRepository knowledgeRawSegmentRepository; private IKnowledgeRawSegmentRepository knowledgeRawSegmentRepository;
@Resource
private IFileAccessService fileAccessService;
@Override @Override
public void chunk(KnowledgeDocumentModel documentDto, UserContext userContext) { public void chunk(KnowledgeDocumentModel documentDto, UserContext userContext) {
try (InputStream inputStream = new URL(documentDto.getDocUrl()).openStream()) { try (InputStream inputStream = fileAccessService.openFileStream(documentDto.getDocUrl())) {
String jsonContent = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8); String jsonContent = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
Object parsed = JSON.parse(jsonContent); Object parsed = JSON.parse(jsonContent);

View File

@@ -1,5 +1,6 @@
package com.xspaceagi.knowledge.domain.docparser.impl; package com.xspaceagi.knowledge.domain.docparser.impl;
import com.xspaceagi.file.sdk.IFileAccessService;
import com.xspaceagi.knowledge.domain.docparser.FileParseRequest; import com.xspaceagi.knowledge.domain.docparser.FileParseRequest;
import com.xspaceagi.knowledge.domain.docparser.FileParseService; import com.xspaceagi.knowledge.domain.docparser.FileParseService;
import com.xspaceagi.knowledge.domain.docparser.parse.DocParser; import com.xspaceagi.knowledge.domain.docparser.parse.DocParser;
@@ -15,8 +16,6 @@ import org.apache.pdfbox.pdmodel.PDDocument;
import org.apache.pdfbox.text.PDFTextStripper; import org.apache.pdfbox.text.PDFTextStripper;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.net.URI;
@Slf4j @Slf4j
@Service @Service
public class PDFParser implements DocParser { public class PDFParser implements DocParser {
@@ -25,12 +24,14 @@ public class PDFParser implements DocParser {
@Resource @Resource
private FileParseService fileParseService; private FileParseService fileParseService;
@Resource
private IFileAccessService fileAccessService;
@Override @Override
public void chunk(KnowledgeDocumentModel documentDto, UserContext userContext) { public void chunk(KnowledgeDocumentModel documentDto, UserContext userContext) {
try (PDDocument doc = Loader.loadPDF(RandomAccessReadBuffer.createBufferFromStream( try (PDDocument doc = Loader.loadPDF(RandomAccessReadBuffer.createBufferFromStream(
new URI(documentDto.getDocUrl()) fileAccessService.openFileStream(documentDto.getDocUrl())))) {
.toURL().openStream()))) {
PDFTextStripper stripper = new PDFTextStripper(); PDFTextStripper stripper = new PDFTextStripper();
String content = stripper.getText(doc); String content = stripper.getText(doc);

View File

@@ -1,5 +1,6 @@
package com.xspaceagi.knowledge.domain.docparser.impl; package com.xspaceagi.knowledge.domain.docparser.impl;
import com.xspaceagi.file.sdk.IFileAccessService;
import com.xspaceagi.knowledge.core.spec.enums.KnowledgeDataTypeEnum; import com.xspaceagi.knowledge.core.spec.enums.KnowledgeDataTypeEnum;
import com.xspaceagi.knowledge.domain.docparser.FileParseRequest; import com.xspaceagi.knowledge.domain.docparser.FileParseRequest;
import com.xspaceagi.knowledge.domain.docparser.FileParseService; import com.xspaceagi.knowledge.domain.docparser.FileParseService;
@@ -14,7 +15,6 @@ import org.springframework.stereotype.Service;
import java.io.BufferedReader; import java.io.BufferedReader;
import java.io.InputStreamReader; import java.io.InputStreamReader;
import java.net.URL;
import java.util.Objects; import java.util.Objects;
@Slf4j @Slf4j
@@ -24,6 +24,9 @@ public class TXTParser implements DocParser {
@Resource @Resource
private FileParseService fileParseService; private FileParseService fileParseService;
@Resource
private IFileAccessService fileAccessService;
@Override @Override
public void chunk(KnowledgeDocumentModel documentDto, UserContext userContext) { public void chunk(KnowledgeDocumentModel documentDto, UserContext userContext) {
@@ -38,7 +41,7 @@ public class TXTParser implements DocParser {
case URL_FILE -> { case URL_FILE -> {
StringBuilder content = new StringBuilder(); StringBuilder content = new StringBuilder();
try (BufferedReader reader = try (BufferedReader reader =
new BufferedReader(new InputStreamReader(new URL(documentDto.getDocUrl()).openStream()))) { new BufferedReader(new InputStreamReader(fileAccessService.openFileStream(documentDto.getDocUrl())))) {
content.append(reader.lines() content.append(reader.lines()
.toList()); .toList());

View File

@@ -8,6 +8,7 @@ import com.xspaceagi.agent.core.infra.component.agent.AgentContext;
import com.xspaceagi.agent.core.infra.component.workflow.WorkflowContext; import com.xspaceagi.agent.core.infra.component.workflow.WorkflowContext;
import com.xspaceagi.agent.core.infra.component.workflow.WorkflowExecutor; import com.xspaceagi.agent.core.infra.component.workflow.WorkflowExecutor;
import com.xspaceagi.agent.core.spec.enums.DataTypeEnum; import com.xspaceagi.agent.core.spec.enums.DataTypeEnum;
import com.xspaceagi.file.sdk.IFileAccessService;
import com.xspaceagi.knowledge.core.spec.enums.KnowledgeDataTypeEnum; import com.xspaceagi.knowledge.core.spec.enums.KnowledgeDataTypeEnum;
import com.xspaceagi.knowledge.domain.docparser.FileParseRequest; import com.xspaceagi.knowledge.domain.docparser.FileParseRequest;
import com.xspaceagi.knowledge.domain.docparser.FileParseService; import com.xspaceagi.knowledge.domain.docparser.FileParseService;
@@ -62,6 +63,9 @@ public class TikaParser implements DocParser {
@Resource @Resource
private WorkflowExecutor workflowExecutor; private WorkflowExecutor workflowExecutor;
@Resource
private IFileAccessService fileAccessService;
@Override @Override
public void chunk( public void chunk(
KnowledgeDocumentModel documentDto, KnowledgeDocumentModel documentDto,
@@ -103,10 +107,7 @@ public class TikaParser implements DocParser {
return; return;
} }
// 下载文件 try (InputStream stream = fileAccessService.openFileStream(docUrl)) {
URL url = new URL(docUrl);
URLConnection connection = url.openConnection();
try (InputStream stream = connection.getInputStream()) {
// 使用 Apache Tika 解析文件内容 // 使用 Apache Tika 解析文件内容
Parser parser = new AutoDetectParser(); Parser parser = new AutoDetectParser();
BodyContentHandler handler = new BodyContentHandler(-1); BodyContentHandler handler = new BodyContentHandler(-1);

View File

@@ -1,11 +1,10 @@
package com.xspaceagi.knowledge.domain.docparser.impl; package com.xspaceagi.knowledge.domain.docparser.impl;
import java.net.URL;
import org.apache.poi.xwpf.extractor.XWPFWordExtractor; import org.apache.poi.xwpf.extractor.XWPFWordExtractor;
import org.apache.poi.xwpf.usermodel.XWPFDocument; import org.apache.poi.xwpf.usermodel.XWPFDocument;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import com.xspaceagi.file.sdk.IFileAccessService;
import com.xspaceagi.knowledge.domain.docparser.FileParseRequest; import com.xspaceagi.knowledge.domain.docparser.FileParseRequest;
import com.xspaceagi.knowledge.domain.docparser.FileParseService; import com.xspaceagi.knowledge.domain.docparser.FileParseService;
import com.xspaceagi.knowledge.domain.docparser.parse.DocParser; import com.xspaceagi.knowledge.domain.docparser.parse.DocParser;
@@ -24,10 +23,13 @@ public class WordParser implements DocParser {
@Resource @Resource
private FileParseService fileParseService; private FileParseService fileParseService;
@Resource
private IFileAccessService fileAccessService;
@Override @Override
public void chunk(KnowledgeDocumentModel documentDto, UserContext userContext) { public void chunk(KnowledgeDocumentModel documentDto, UserContext userContext) {
try (XWPFDocument document = new XWPFDocument(new URL(documentDto.getDocUrl()).openStream()); try (XWPFDocument document = new XWPFDocument(fileAccessService.openFileStream(documentDto.getDocUrl()));
XWPFWordExtractor extractor = new XWPFWordExtractor(document)) { XWPFWordExtractor extractor = new XWPFWordExtractor(document)) {
String content = extractor.getText(); String content = extractor.getText();
FileParseRequest fileParseRequest = FileParseRequest.builder() FileParseRequest fileParseRequest = FileParseRequest.builder()

View File

@@ -8,6 +8,7 @@ import com.xspaceagi.file.sdk.IFileAccessService;
import com.xspaceagi.knowledge.core.adapter.client.dto.PushResult; import com.xspaceagi.knowledge.core.adapter.client.dto.PushResult;
import com.xspaceagi.knowledge.core.spec.enums.FulltextSyncStatusEnum; import com.xspaceagi.knowledge.core.spec.enums.FulltextSyncStatusEnum;
import com.xspaceagi.knowledge.core.spec.enums.KnowledgeTaskRunTypeEnum; import com.xspaceagi.knowledge.core.spec.enums.KnowledgeTaskRunTypeEnum;
import com.xspaceagi.knowledge.core.spec.enums.KnowledgeTaskStageStatusEnum;
import com.xspaceagi.knowledge.core.spec.enums.QaStatusEnum; import com.xspaceagi.knowledge.core.spec.enums.QaStatusEnum;
import com.xspaceagi.knowledge.core.spec.enums.RawTextFulltextSyncStatusEnum; import com.xspaceagi.knowledge.core.spec.enums.RawTextFulltextSyncStatusEnum;
import com.xspaceagi.knowledge.core.spec.utils.Constants; import com.xspaceagi.knowledge.core.spec.utils.Constants;
@@ -129,14 +130,12 @@ public class KnowledgeDocumentDomainService implements IKnowledgeDocumentDomainS
var kbId = existObj.getKbId(); var kbId = existObj.getKbId();
this.vectorDBService.removeDoc(kbId, id); this.vectorDBService.removeDoc(kbId, id);
// 删除全文检索数据(在事务内,保证一致性) // 全文检索是外部索引,删除失败不应阻断业务数据删除。
try { try {
fullTextSearchDomainService.deleteByDocId(id, kbId, userContext.getTenantId()); fullTextSearchDomainService.deleteByDocId(id, kbId, userContext.getTenantId());
log.info("Delete doc full-text OK: docId={}, kbId={}", id, kbId); log.info("Delete doc full-text OK: docId={}, kbId={}", id, kbId);
} catch (Exception e) { } catch (Exception e) {
log.error("Delete doc full-text failed: docId={}, kbId={}", id, kbId, e); log.warn("Delete doc full-text failed, ignored: docId={}, kbId={}", id, kbId, e);
// 抛出异常,触发事务回滚
throw KnowledgeException.build(BizExceptionCodeEnum.knowledgeDeleteDocFulltextFailed, e);
} }
// 更新知识库最后操作时间 // 更新知识库最后操作时间
@@ -277,6 +276,7 @@ public class KnowledgeDocumentDomainService implements IKnowledgeDocumentDomainS
public void workRunTaskForDocument(KnowledgeDocumentModel model, UserContext userContext, Long docId) { public void workRunTaskForDocument(KnowledgeDocumentModel model, UserContext userContext, Long docId) {
log.debug("Vector task submit start, docId={}", docId); log.debug("Vector task submit start, docId={}", docId);
try {
// 1. 执行文档分段 // 1. 执行文档分段
this.self.workRunTaskForRawSegment(model, userContext, docId); this.self.workRunTaskForRawSegment(model, userContext, docId);
log.info("Segment task done: kbId={}, docId={}", model.getKbId(), docId); log.info("Segment task done: kbId={}, docId={}", model.getKbId(), docId);
@@ -292,6 +292,10 @@ public class KnowledgeDocumentDomainService implements IKnowledgeDocumentDomainS
}); });
threadTenantUtil.obtainCommonExecutor().execute(tenantRunnable); threadTenantUtil.obtainCommonExecutor().execute(tenantRunnable);
} catch (Exception e) {
markTaskWaitRetry(model, userContext, docId, e);
throw e;
}
} }
@@ -330,6 +334,7 @@ public class KnowledgeDocumentDomainService implements IKnowledgeDocumentDomainS
public void workRetryRunTaskForDocument(KnowledgeTaskRunTypeEnum runTypeEnum, KnowledgeDocumentModel model, public void workRetryRunTaskForDocument(KnowledgeTaskRunTypeEnum runTypeEnum, KnowledgeDocumentModel model,
UserContext userContext, Long docId) { UserContext userContext, Long docId) {
try {
switch (runTypeEnum) { switch (runTypeEnum) {
case SEGMENT -> { case SEGMENT -> {
// 文档分段,全文检索,问答,向量化,全部走重试 // 文档分段,全文检索,问答,向量化,全部走重试
@@ -360,9 +365,46 @@ public class KnowledgeDocumentDomainService implements IKnowledgeDocumentDomainS
log.info("No retry needed, docId={}", docId); log.info("No retry needed, docId={}", docId);
} }
} }
} catch (Exception e) {
markTaskWaitRetry(model, userContext, docId, e);
throw e;
}
} }
private void markTaskWaitRetry(KnowledgeDocumentModel model, UserContext userContext, Long docId, Exception e) {
try {
KnowledgeTaskModel task = this.knowledgeTaskRepository.queryOneByDocId(docId);
if (Objects.isNull(task)) {
task = KnowledgeTaskModel.createEmptyTaskModel(docId, KnowledgeTaskRunTypeEnum.SEGMENT, userContext);
task.setKbId(model.getKbId());
task.setSpaceId(model.getSpaceId());
}
task.setStatus(KnowledgeTaskStageStatusEnum.WAIT_RETRY.getStatus());
task.setResult(buildTaskFailureMessage(e));
if (Objects.isNull(task.getId())) {
this.knowledgeTaskRepository.addInfo(task, userContext);
} else {
this.knowledgeTaskRepository.updateInfo(task, userContext);
}
} catch (Exception markException) {
log.error("Failed to mark knowledge task as wait retry, docId={}", docId, markException);
}
}
private String buildTaskFailureMessage(Throwable throwable) {
Throwable root = throwable;
while (root.getCause() != null) {
root = root.getCause();
}
String message = root.getMessage();
if (StringUtils.isBlank(message)) {
message = root.getClass().getSimpleName();
}
return StringUtils.abbreviate(message, 500);
}
@Override @Override
public void generateForQa(Long docId, UserContext userContext) { public void generateForQa(Long docId, UserContext userContext) {
var docModel = this.knowledgeDocumentRepository.queryOneInfoById(docId); var docModel = this.knowledgeDocumentRepository.queryOneInfoById(docId);

View File

@@ -24,6 +24,7 @@ import com.xspaceagi.knowledge.core.infra.translator.IKnowledgeDocumentTranslato
import com.xspaceagi.knowledge.sdk.enums.KnowledgeDocStatueEnum; import com.xspaceagi.knowledge.sdk.enums.KnowledgeDocStatueEnum;
import com.xspaceagi.knowledge.sdk.enums.KnowledgePubStatusEnum; import com.xspaceagi.knowledge.sdk.enums.KnowledgePubStatusEnum;
import com.xspaceagi.knowledge.core.spec.enums.KnowledgeTaskRunTypeEnum; import com.xspaceagi.knowledge.core.spec.enums.KnowledgeTaskRunTypeEnum;
import com.xspaceagi.knowledge.core.spec.enums.KnowledgeTaskStageStatusEnum;
import com.xspaceagi.knowledge.core.spec.enums.QaStatusEnum; import com.xspaceagi.knowledge.core.spec.enums.QaStatusEnum;
import com.xspaceagi.knowledge.core.spec.utils.CaffeineCacheUtil; import com.xspaceagi.knowledge.core.spec.utils.CaffeineCacheUtil;
import com.xspaceagi.knowledge.core.spec.utils.ThreadTenantUtil; import com.xspaceagi.knowledge.core.spec.utils.ThreadTenantUtil;
@@ -36,6 +37,7 @@ import com.xspaceagi.system.spec.tenant.thread.TenantRunnable;
import jakarta.annotation.Resource; import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
@Slf4j @Slf4j
@Repository @Repository
@@ -201,6 +203,17 @@ public class KnowledgeDocumentRepository implements IKnowledgeDocumentRepository
log.warn("Document {} analysis failed due to max retry count reached", document.getId()); log.warn("Document {} analysis failed due to max retry count reached", document.getId());
return true; return true;
} }
KnowledgeTaskStageStatusEnum taskStatus = Objects.nonNull(retryTask.getStatus())
? KnowledgeTaskStageStatusEnum.getByStatus(retryTask.getStatus())
: null;
if (KnowledgeTaskStageStatusEnum.WAIT_RETRY == taskStatus
|| KnowledgeTaskStageStatusEnum.RETRY_FAIL == taskStatus
|| KnowledgeTaskStageStatusEnum.FORBID_RETRY == taskStatus) {
document.setDocStatus(KnowledgeDocStatueEnum.ANALYZE_FAILED);
document.setDocStatusReason(buildTaskFailureReason(retryTask, taskStatus));
log.warn("Document {} analysis failed, taskStatus={}", document.getId(), taskStatus);
return true;
}
//如果重试状态是成功,则认为是成功了 //如果重试状态是成功,则认为是成功了
if (Objects.nonNull(retryTask.getType()) && KnowledgeTaskRunTypeEnum.SUCCESS.getType() == retryTask.getType()) { if (Objects.nonNull(retryTask.getType()) && KnowledgeTaskRunTypeEnum.SUCCESS.getType() == retryTask.getType()) {
document.setDocStatus(KnowledgeDocStatueEnum.ANALYZED); document.setDocStatus(KnowledgeDocStatueEnum.ANALYZED);
@@ -246,6 +259,14 @@ public class KnowledgeDocumentRepository implements IKnowledgeDocumentRepository
return false; return false;
} }
private String buildTaskFailureReason(KnowledgeTask retryTask, KnowledgeTaskStageStatusEnum taskStatus) {
String result = retryTask.getResult();
if (StringUtils.isBlank(result)) {
return taskStatus.getDesc();
}
return taskStatus.getDesc() + "" + result;
}
/** /**
* 检查向量化状态 * 检查向量化状态
*/ */