Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-22 15:13:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(knowledge):重构知识库模块架构并优化DTO结构

Browse Source 
- 将BotController暂时注释以重新设计Agent集成方案
- 在多个DTO类中添加@JsonIgnoreProperties注解提升JSON序列化兼容性- 修改ChunkDTO中positions字段类型为嵌套列表并添加token字段
-为DatasetDTO和DocumentDTO添加更多时间日期字段映射
-重构KnowledgeBaseAdapter接口参数结构使用统一的请求对象
- 实现DocumentStatusSyncTask定时任务同步文档处理状态
-优化KnowledgeBaseService统计信息更新机制
- 修复数据集删除时的级联删除逻辑防止孤儿数据
- 统一本地实体ID与RAGFlow ID避免前端调用错误
This commit is contained in:
GZH
2026-02-14 12:55:56 +08:00
parent 861df73eab
commit 16a38bbfc4
32 changed files with 1090 additions and 736 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/manager-api/src/main/java/xiaozhi/common/config/SwaggerConfig.java
+16
View File
@@ -79,6 +79,22 @@ public class SwaggerConfig {
.build();
}
@Bean
public GroupedOpenApi knowledgeApi() {
return GroupedOpenApi.builder()
.group("knowledge")
.pathsToMatch("/datasets/**")
.build();
}
@Bean
public GroupedOpenApi botApi() {
return GroupedOpenApi.builder()
.group("bot")
.pathsToMatch("/api/v1/**")
.build();
}
@Bean
public OpenAPI customOpenAPI() {
return new OpenAPI().info(new Info()
main/manager-api/src/main/java/xiaozhi/common/exception/RenException.java
+4 -2
View File
@@ -13,23 +13,25 @@ public class RenException extends RuntimeException {
private String msg;
public RenException(int code) {
super(MessageUtils.getMessage(code));
this.code = code;
this.msg = MessageUtils.getMessage(code);
}
public RenException(int code, String... params) {
super(MessageUtils.getMessage(code, params));
this.code = code;
this.msg = MessageUtils.getMessage(code, params);
}
public RenException(int code, Throwable e) {
super(e);
super(MessageUtils.getMessage(code), e);
this.code = code;
this.msg = MessageUtils.getMessage(code);
}
public RenException(int code, Throwable e, String... params) {
super(e);
super(MessageUtils.getMessage(code, params), e);
this.code = code;
this.msg = MessageUtils.getMessage(code, params);
}
main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/OtaServiceImpl.java
+1 -1
View File
@@ -66,7 +66,7 @@ public class OtaServiceImpl extends BaseServiceImpl<OtaDao, OtaEntity> implement
// 同类固件只保留最新的一条
List<OtaEntity> otaList = baseDao.selectList(queryWrapper);
if (otaList != null && otaList.size() > 0) {
OtaEntity otaBefore = otaList.getFirst();
OtaEntity otaBefore = otaList.get(0);
entity.setId(otaBefore.getId());
baseDao.updateById(entity);
return true;
main/manager-api/src/main/java/xiaozhi/modules/knowledge/config/RAGTaskConfig.java
+13
View File
@@ -0,0 +1,13 @@
package xiaozhi.modules.knowledge.config;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableScheduling;
/**
* 知识库模块定时任务配置
* 启用 Spring Schedule 能力
*/
@Configuration
@EnableScheduling
public class RAGTaskConfig {
}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/controller/BotController.java
+148 -148
View File
@@ -1,148 +1,148 @@
package xiaozhi.modules.knowledge.controller;
import java.util.HashMap;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import xiaozhi.common.exception.ErrorCode;
import xiaozhi.common.exception.RenException;
import xiaozhi.modules.agent.entity.AgentEntity;
import xiaozhi.modules.agent.service.AgentService;
import xiaozhi.modules.knowledge.dto.bot.BotDTO;
import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapter;
import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapterFactory;
import xiaozhi.modules.knowledge.service.KnowledgeBaseService;
import xiaozhi.modules.model.entity.ModelConfigEntity;
import xiaozhi.modules.model.service.ModelConfigService;
@Tag(name = "外部机器人服务")
@RestController
@RequestMapping("/api/v1") // 保持与外部 API 一致的风格，或者映射到 RAGFlow 风格
@RequiredArgsConstructor
@Slf4j
public class BotController {//todo:这个类还有待完成，待完善
private final AgentService agentService;
private final ModelConfigService modelConfigService;
private final KnowledgeBaseService knowledgeBaseService;
// SearchBot (暂未完全实现，预留接口)
@Operation(summary = "SearchBot 提问")
@PostMapping(value = "/searchbots/ask", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter searchBotAsk(@RequestBody BotDTO.SearchAskReq request) {
// TODO: SearchBot 通常需要指定 Dataset 或 KnowledgeBase
// 这里需要明确 SearchBot 的 ID 来源。
// 暂时返回未实现
throw new RenException("SearchBot API 暂未开放");
}
// AgentBot
@Operation(summary = "AgentBot 对话 (流式)")
@PostMapping(value = "/agentbots/{id}/completions", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter agentBotCompletion(@PathVariable("id") String agentId,
@RequestBody BotDTO.AgentCompletionReq request) {
SseEmitter emitter = new SseEmitter(5 * 60 * 1000L); // 5 min timeout
new Thread(() -> {
try {
// 1. 获取本地 Agent
// 注意：这里的 id 是本地 AgentId
AgentEntity agent = agentService.getAgentById(agentId); // getAgentById actually returns VO but we need
// Entity fields
// 直接查 DB 或用 Service
// Service returns AgentInfoVO, let's better use getById from BaseService (which
// returns Entity) if exposed
// AgentService extends BaseService<AgentEntity>
AgentEntity entity = agentService.selectById(agentId);
if (entity == null) {
throw new RenException(ErrorCode.AGENT_NOT_FOUND);
}
// 2. 获取关联的 RAG 配置
String llmModelId = entity.getLlmModelId();
if (StringUtils.isBlank(llmModelId)) {
throw new RenException("Agent未配置LLM模型");
}
// 3. 所有的 AgentBot 请求都通过关联的 RAGConfig 下发
Map<String, Object> ragConfig = knowledgeBaseService.getRAGConfig(llmModelId);
String adapterType = (String) ragConfig.getOrDefault("type", "ragflow");
KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(adapterType, ragConfig);
// 4. 发起请求
// 注意：这里需要传递给 RAGFlow 的 Agent ID 可能是 config 中的某些字段，
// 或者我们这里构建的 Agent 实际上是 RAGFlow 中的 Agent？
// 根据 Deep Dive，ChatService 是自己组装 prompt 调用 chat/completions。
// 而 BotController 似乎旨在暴露 RAGFlow 的 AgentBot 能力。
// 如果我们是“Java主导”，那么其实这里的逻辑应该和 ChatService 类似：
// 即：我们自己组装参数，调用 RAGFlow 的 Chat/Completion 接口，而不是调用 AgentBot 接口。
// 除非 RAGFlow 侧真的有一个 "Agent" 对象对应我们的 Agent。
// 鉴于 Step 8 中我们决定 "Java-Side Master"，且 RAGFlow 是推理引擎，
// 我们应该复用 ChatService 的逻辑，或者调用 POST /chat/completions (RAGFlow)
// 但为了符合 BotDTO 的定义 (/agentbots/{id}/completions)，我们这里模拟这个行为
// 且尽量复用 ChatService 的流式能力。
// 这里的 adapter.postAgentBotCompletion 实现的是调用 /api/v1/agentbots/{id}/completions
// 这要求 RAGFlow 侧必须存在这个 ID。
// 但我们的 Agent 是本地的。
// **修正策略**：
// 如果 RAGFlow 侧没有对应 Agent，我们不能调这个接口。
// 我们应该降级为调用 Chat 接口 (/chat/completions)，带上 system prompt。
// 出于本步骤目标 "Bot Service (Public Access)"，我们为了兼容外部 Bot 协议，
// 应该要把本地请求转换为 RAGFlow 的 Chat 请求。
Map<String, Object> chatBody = new HashMap<>();
chatBody.put("messages", java.util.List.of(
Map.of("role", "user", "content", request.getQuestion())));
chatBody.put("stream", request.getStream());
chatBody.put("model", entity.getAgentName()); // Mock name
if (StringUtils.isNotBlank(entity.getSystemPrompt())) {
chatBody.put("system_prompt", entity.getSystemPrompt());
}
adapter.postStream("/api/v1/chat/completions", chatBody, line -> {
try {
if (line.startsWith("data:")) {
String dataContent = line.substring(5).trim();
if (!"[DONE]".equals(dataContent)) {
emitter.send(SseEmitter.event().data(dataContent));
}
}
} catch (Exception e) {
log.error("SSE Error", e);
throw new RuntimeException(e);
}
});
emitter.send(SseEmitter.event().data("[DONE]"));
emitter.complete();
} catch (Exception e) {
log.error("AgentBot Error", e);
try {
emitter.send(SseEmitter.event().name("error").data(e.getMessage()));
emitter.completeWithError(e);
} catch (Exception ex) {
// ignore
}
}
}).start();
return emitter;
}
}
//package xiaozhi.modules.knowledge.controller;
//
//import java.util.HashMap;
//import java.util.Map;
//
//import org.apache.commons.lang3.StringUtils;
//import org.springframework.http.MediaType;
//import org.springframework.web.bind.annotation.PathVariable;
//import org.springframework.web.bind.annotation.PostMapping;
//import org.springframework.web.bind.annotation.RequestBody;
//import org.springframework.web.bind.annotation.RequestMapping;
//import org.springframework.web.bind.annotation.RestController;
//import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
//
//import io.swagger.v3.oas.annotations.Operation;
//import io.swagger.v3.oas.annotations.tags.Tag;
//import lombok.RequiredArgsConstructor;
//import lombok.extern.slf4j.Slf4j;
//import xiaozhi.common.exception.ErrorCode;
//import xiaozhi.common.exception.RenException;
//import xiaozhi.modules.agent.entity.AgentEntity;
//import xiaozhi.modules.agent.service.AgentService;
//import xiaozhi.modules.knowledge.dto.bot.BotDTO;
//import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapter;
//import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapterFactory;
//import xiaozhi.modules.knowledge.service.KnowledgeBaseService;
//import xiaozhi.modules.model.entity.ModelConfigEntity;
//import xiaozhi.modules.model.service.ModelConfigService;
//
//@Tag(name = "外部机器人服务")
//@RestController
//@RequestMapping("/api/v1") // 保持与外部 API 一致的风格，或者映射到 RAGFlow 风格
//@RequiredArgsConstructor
//@Slf4j
//public class BotController {//todo:这个类还有待完成，待完善。这里的agent和其他模块的agent不一样，这里的agent是存在于ragflow中等agent/工作流，不独立存在，需要借助rag作为平台进行交流交互
//
// private final AgentService agentService;
// private final ModelConfigService modelConfigService;
// private final KnowledgeBaseService knowledgeBaseService;
//
// // SearchBot (暂未完全实现，预留接口)
// @Operation(summary = "SearchBot 提问")
// @PostMapping(value = "/searchbots/ask", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
// public SseEmitter searchBotAsk(@RequestBody BotDTO.SearchAskReq request) {
// // TODO: SearchBot 通常需要指定 Dataset 或 KnowledgeBase
// // 这里需要明确 SearchBot 的 ID 来源。
// // 暂时返回未实现
// throw new RenException("SearchBot API 暂未开放");
// }
//
// // AgentBot
// @Operation(summary = "AgentBot 对话 (流式)")
// @PostMapping(value = "/agentbots/{id}/completions", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
// public SseEmitter agentBotCompletion(@PathVariable("id") String agentId,
// @RequestBody BotDTO.AgentCompletionReq request) {
// SseEmitter emitter = new SseEmitter(5 * 60 * 1000L); // 5 min timeout
//
// new Thread(() -> {
// try {
// // 1. 获取本地 Agent
// // 注意：这里的 id 是本地 AgentId
// AgentEntity agent = agentService.getAgentById(agentId); // getAgentById actually returns VO but we need
// // Entity fields
// // 直接查 DB 或用 Service
// // Service returns AgentInfoVO, let's better use getById from BaseService (which
// // returns Entity) if exposed
// // AgentService extends BaseService<AgentEntity>
// AgentEntity entity = agentService.selectById(agentId);
//
// if (entity == null) {
// throw new RenException(ErrorCode.AGENT_NOT_FOUND);
// }
//
// // 2. 获取关联的 RAG 配置
// String llmModelId = entity.getLlmModelId();
// if (StringUtils.isBlank(llmModelId)) {
// throw new RenException("Agent未配置LLM模型");
// }
//
// // 3. 所有的 AgentBot 请求都通过关联的 RAGConfig 下发
// Map<String, Object> ragConfig = knowledgeBaseService.getRAGConfig(llmModelId);
// String adapterType = (String) ragConfig.getOrDefault("type", "ragflow");
// KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(adapterType, ragConfig);
//
// // 4. 发起请求
// // 注意：这里需要传递给 RAGFlow 的 Agent ID 可能是 config 中的某些字段，
// // 或者我们这里构建的 Agent 实际上是 RAGFlow 中的 Agent？
// // 根据 Deep Dive，ChatService 是自己组装 prompt 调用 chat/completions。
// // 而 BotController 似乎旨在暴露 RAGFlow 的 AgentBot 能力。
// // 如果我们是“Java主导”，那么其实这里的逻辑应该和 ChatService 类似：
// // 即：我们自己组装参数，调用 RAGFlow 的 Chat/Completion 接口，而不是调用 AgentBot 接口。
// // 除非 RAGFlow 侧真的有一个 "Agent" 对象对应我们的 Agent。
// // 鉴于 Step 8 中我们决定 "Java-Side Master"，且 RAGFlow 是推理引擎，
// // 我们应该复用 ChatService 的逻辑，或者调用 POST /chat/completions (RAGFlow)
//
// // 但为了符合 BotDTO 的定义 (/agentbots/{id}/completions)，我们这里模拟这个行为
// // 且尽量复用 ChatService 的流式能力。
//
// // 这里的 adapter.postAgentBotCompletion 实现的是调用 /api/v1/agentbots/{id}/completions
// // 这要求 RAGFlow 侧必须存在这个 ID。
// // 但我们的 Agent 是本地的。
// // **修正策略**：
// // 如果 RAGFlow 侧没有对应 Agent，我们不能调这个接口。
// // 我们应该降级为调用 Chat 接口 (/chat/completions)，带上 system prompt。
//
// // 出于本步骤目标 "Bot Service (Public Access)"，我们为了兼容外部 Bot 协议，
// // 应该要把本地请求转换为 RAGFlow 的 Chat 请求。
//
// Map<String, Object> chatBody = new HashMap<>();
// chatBody.put("messages", java.util.List.of(
// Map.of("role", "user", "content", request.getQuestion())));
// chatBody.put("stream", request.getStream());
// chatBody.put("model", entity.getAgentName()); // Mock name
// if (StringUtils.isNotBlank(entity.getSystemPrompt())) {
// chatBody.put("system_prompt", entity.getSystemPrompt());
// }
//
// adapter.postStream("/api/v1/chat/completions", chatBody, line -> {
// try {
// if (line.startsWith("data:")) {
// String dataContent = line.substring(5).trim();
// if (!"[DONE]".equals(dataContent)) {
// emitter.send(SseEmitter.event().data(dataContent));
// }
// }
// } catch (Exception e) {
// log.error("SSE Error", e);
// throw new RuntimeException(e);
// }
// });
//
// emitter.send(SseEmitter.event().data("[DONE]"));
// emitter.complete();
//
// } catch (Exception e) {
// log.error("AgentBot Error", e);
// try {
// emitter.send(SseEmitter.event().name("error").data(e.getMessage()));
// emitter.completeWithError(e);
// } catch (Exception ex) {
// // ignore
// }
// }
// }).start();
//
// return emitter;
// }
//}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/controller/KnowledgeBaseController.java
+10 -5
View File
@@ -36,6 +36,7 @@ import xiaozhi.modules.security.user.SecurityUser;
public class KnowledgeBaseController {
private final KnowledgeBaseService knowledgeBaseService;
private final xiaozhi.modules.knowledge.service.KnowledgeManagerService knowledgeManagerService;
@GetMapping
@Operation(summary = "分页查询知识库列表")
@@ -96,6 +97,8 @@ public class KnowledgeBaseController {
throw new RenException(ErrorCode.NO_PERMISSION);
}
// [FIX] 注入 ID，防止 Service 层找不到记录
knowledgeBaseDTO.setId(existingKnowledgeBase.getId());
knowledgeBaseDTO.setDatasetId(datasetId);
KnowledgeBaseDTO resp = knowledgeBaseService.update(knowledgeBaseDTO);
return new Result<KnowledgeBaseDTO>().ok(resp);
@@ -117,7 +120,8 @@ public class KnowledgeBaseController {
throw new RenException(ErrorCode.NO_PERMISSION);
}
knowledgeBaseService.deleteByDatasetId(datasetId);
// [Architecture Fix] 通过编排层级联删除，防止孤儿数据并解决循环依赖
knowledgeManagerService.deleteDatasetWithFiles(datasetId);
return new Result<>();
}
@@ -133,15 +137,16 @@ public class KnowledgeBaseController {
// 获取当前登录用户ID
Long currentUserId = SecurityUser.getUserId();
List<String> idList = Arrays.asList(ids.split(","));
List<KnowledgeBaseDTO> knowledgeBaseDTOs = Optional.ofNullable(knowledgeBaseService.getByDatasetIdList(idList)).orElseGet(ArrayList::new);
List<KnowledgeBaseDTO> knowledgeBaseDTOs = Optional.ofNullable(knowledgeBaseService.getByDatasetIdList(idList))
.orElseGet(ArrayList::new);
if (ToolUtil.isNotEmpty(knowledgeBaseDTOs)) {
knowledgeBaseDTOs.forEach(item->{
knowledgeBaseDTOs.forEach(item -> {
// 检查权限：用户只能删除自己创建的知识库
if (item.getCreator() == null || !item.getCreator().equals(currentUserId)) {
throw new RenException(ErrorCode.NO_PERMISSION);
}
//删除
knowledgeBaseService.deleteByDatasetId(item.getDatasetId());
// [Architecture Fix] 通过编排层级联删除
knowledgeManagerService.deleteDatasetWithFiles(item.getDatasetId());
});
}
return new Result<>();
main/manager-api/src/main/java/xiaozhi/modules/knowledge/controller/KnowledgeFilesController.java
+28 -28
View File
@@ -4,6 +4,7 @@ import java.util.List;
import java.util.Map;
import org.apache.shiro.authz.annotation.RequiresPermissions;
import org.springdoc.core.annotations.ParameterObject;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
@@ -11,7 +12,6 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.AllArgsConstructor;
import xiaozhi.common.exception.ErrorCode;
@@ -21,6 +21,7 @@ import xiaozhi.common.utils.Result;
import xiaozhi.modules.knowledge.dto.KnowledgeBaseDTO;
import xiaozhi.modules.knowledge.dto.KnowledgeFilesDTO;
import xiaozhi.modules.knowledge.dto.document.ChunkDTO;
import xiaozhi.modules.knowledge.dto.document.DocumentDTO;
import xiaozhi.modules.knowledge.dto.document.RetrievalDTO;
import xiaozhi.modules.knowledge.service.KnowledgeBaseService;
import xiaozhi.modules.knowledge.service.KnowledgeFilesService;
@@ -113,16 +114,15 @@ public class KnowledgeFilesController {
return new Result<KnowledgeFilesDTO>().ok(resp);
}
@DeleteMapping("/documents/{document_id}")
@Operation(summary = "删除单个文档")
@Parameter(name = "document_id", description = "文档ID", required = true)
@DeleteMapping("/documents")
@Operation(summary = "批量删除文档")
@RequiresPermissions("sys:role:normal")
public Result<Void> delete(@PathVariable("dataset_id") String datasetId,
@PathVariable("document_id") String documentId) {
@RequestBody DocumentDTO.BatchIdReq req) {
// 验证知识库权限
validateKnowledgeBasePermission(datasetId);
knowledgeFilesService.deleteByDocumentId(documentId, datasetId);
knowledgeFilesService.deleteDocuments(datasetId, req);
return new Result<>();
}
@@ -153,17 +153,19 @@ public class KnowledgeFilesController {
public Result<ChunkDTO.ListVO> listChunks(
@PathVariable("dataset_id") String datasetId,
@PathVariable("document_id") String documentId,
@RequestParam(required = false) String keywords,
@RequestParam(required = false, defaultValue = "1") Integer page,
@RequestParam(required = false, defaultValue = "50") Integer page_size,
@RequestParam(required = false) String id) {
@ParameterObject ChunkDTO.ListReq req) {
// 验证权限 (内部已包含知识库存在性校验与归属权校验)
validateKnowledgeBasePermission(datasetId);
// 设置默认值
if (req.getPage() == null)
req.setPage(1);
if (req.getPageSize() == null)
req.setPageSize(50);
// 调用服务层获取强类型切片列表
ChunkDTO.ListVO result = knowledgeFilesService.listChunks(datasetId,
documentId, keywords, page, page_size, id);
ChunkDTO.ListVO result = knowledgeFilesService.listChunks(datasetId, documentId, req);
return new Result<ChunkDTO.ListVO>().ok(result);
}
@@ -177,23 +179,21 @@ public class KnowledgeFilesController {
// 验证知识库权限
validateKnowledgeBasePermission(datasetId);
// 调用检索服务，返回强类型聚合对象
RetrievalDTO.ResultVO result = knowledgeFilesService.retrievalTest(
req.getQuestion(),
req.getDatasetIds() != null && !req.getDatasetIds().isEmpty() ? req.getDatasetIds()
: java.util.Arrays.asList(datasetId),
null,
1,
100,
req.getSimilarityThreshold(),
req.getVectorSimilarityWeight(),
req.getTopK(),
req.getRerankId(),
req.getKeyword(),
req.getHighlight(),
null,
null);
// 业务下沉逻辑：如果未指定知识库ID，则设为当前路径中的 datasetId
if (req.getDatasetIds() == null || req.getDatasetIds().isEmpty()) {
req.setDatasetIds(java.util.Arrays.asList(datasetId));
}
// [Reinforce] 强管控分页参数，防止 RAGFlow 端出现 Negative Slicing 报错
if (req.getPage() == null || req.getPage() < 1) {
req.setPage(1);
}
if (req.getPageSize() == null || req.getPageSize() < 1) {
req.setPageSize(100);
}
// 调用检索服务，返回强类型聚合对象
RetrievalDTO.ResultVO result = knowledgeFilesService.retrievalTest(req);
return new Result<RetrievalDTO.ResultVO>().ok(result);
}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/dao/KnowledgeBaseDao.java
+7 -11
View File
@@ -20,20 +20,16 @@ public interface KnowledgeBaseDao extends BaseDao<KnowledgeBaseEntity> {
void deletePluginMappingByKnowledgeBaseId(@Param("knowledgeBaseId") String knowledgeBaseId);
/**
* 删除文档后更新数据集统计信息
* 通用维度原子更新知识库统计信息
*
* @param datasetId 数据集ID
* @param chunkDelta 分块减少量
* @param tokenDelta Token减少量
* @param docDelta 文档数增量
* @param chunkDelta 分块数增量
* @param tokenDelta Token数增量
*/
void updateStatsAfterDelete(@Param("datasetId") String datasetId, @Param("chunkDelta") Long chunkDelta,
void updateStatsAfterChange(@Param("datasetId") String datasetId,
@Param("docDelta") Integer docDelta,
@Param("chunkDelta") Long chunkDelta,
@Param("tokenDelta") Long tokenDelta);
/**
* 上传文档后更新数据集统计信息 (递增)
*
* @param datasetId 数据集ID
*/
void updateStatsAfterUpload(@Param("datasetId") String datasetId);
}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/dto/KnowledgeFilesDTO.java
+2
View File
@@ -6,10 +6,12 @@ import java.util.Date;
import java.util.Map;
import io.swagger.v3.oas.annotations.media.Schema;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.Data;
@Data
@Schema(description = "知识库文档")
@JsonIgnoreProperties(ignoreUnknown = true)
public class KnowledgeFilesDTO implements Serializable {
@Serial
main/manager-api/src/main/java/xiaozhi/modules/knowledge/dto/dataset/DatasetDTO.java
+19
View File
@@ -6,6 +6,7 @@ import java.io.Serializable;
import java.util.List;
import java.util.Map;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import jakarta.validation.constraints.*;
/**
@@ -15,6 +16,7 @@ import jakarta.validation.constraints.*;
* </p>
*/
@Schema(description = "知识库管理聚合 DTO")
@JsonIgnoreProperties(ignoreUnknown = true)
public class DatasetDTO {
// ========== 通用内部类 ==========
@@ -27,6 +29,7 @@ public class DatasetDTO {
@AllArgsConstructor
@Builder
@Schema(description = "解析器配置")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class ParserConfig implements Serializable {
@Schema(description = "分块 token 数量", example = "128")
@@ -62,6 +65,7 @@ public class DatasetDTO {
@AllArgsConstructor
@Builder
@Schema(description = "创建知识库请求")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class CreateReq implements Serializable {
@NotBlank(message = "知识库名称不能为空")
@@ -98,6 +102,7 @@ public class DatasetDTO {
@AllArgsConstructor
@Builder
@Schema(description = "更新知识库请求")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class UpdateReq implements Serializable {
@Schema(description = "知识库名称", example = "updated_dataset")
@@ -136,6 +141,7 @@ public class DatasetDTO {
@AllArgsConstructor
@Builder
@Schema(description = "查询知识库列表请求")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class ListReq implements Serializable {
@Schema(description = "页码 (从 1 开始)", example = "1")
@@ -218,6 +224,7 @@ public class DatasetDTO {
@AllArgsConstructor
@Builder
@Schema(description = "异步任务 ID 响应")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class TaskIdVO implements Serializable {
@Schema(description = "GraphRAG 任务 ID", example = "task_uuid_12345678")
@@ -239,6 +246,7 @@ public class DatasetDTO {
@AllArgsConstructor
@Builder
@Schema(description = "知识库详情 VO")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class InfoVO implements Serializable {
@Schema(description = "知识库 ID", example = "abc123")
@@ -291,6 +299,14 @@ public class DatasetDTO {
@Schema(description = "总 Token 数", example = "102400")
@JsonProperty("token_num")
private Long tokenNum;
@Schema(description = "创建日期 (格式: yyyy-MM-dd HH:mm:ss)")
@JsonProperty("create_date")
private String createDate;
@Schema(description = "最后更新日期 (格式: yyyy-MM-dd HH:mm:ss)")
@JsonProperty("update_date")
private String updateDate;
}
/**
@@ -341,6 +357,7 @@ public class DatasetDTO {
@AllArgsConstructor
@Builder
@Schema(description = "图谱节点")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Node implements Serializable {
@Schema(description = "节点 ID", example = "node_001")
@@ -367,6 +384,7 @@ public class DatasetDTO {
@AllArgsConstructor
@Builder
@Schema(description = "图谱边")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Edge implements Serializable {
@Schema(description = "源节点 ID", example = "node_001")
@@ -393,6 +411,7 @@ public class DatasetDTO {
@AllArgsConstructor
@Builder
@Schema(description = "异步任务追踪 VO")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class TaskTraceVO implements Serializable {
@Schema(description = "任务 ID", example = "task_001")
main/manager-api/src/main/java/xiaozhi/modules/knowledge/dto/document/ChunkDTO.java
+14 -2
View File
@@ -5,12 +5,14 @@ import io.swagger.v3.oas.annotations.media.Schema;
import java.io.Serializable;
import java.util.List;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import jakarta.validation.constraints.*;
/**
* 切片管理聚合 DTO
*/
@Schema(description = "切片管理聚合 DTO")
@JsonIgnoreProperties(ignoreUnknown = true)
public class ChunkDTO {
/**
@@ -21,6 +23,7 @@ public class ChunkDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "新增切片请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class AddReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -44,6 +47,7 @@ public class ChunkDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "更新切片请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class UpdateReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -66,6 +70,7 @@ public class ChunkDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "获取切片列表请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class ListReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -91,6 +96,7 @@ public class ChunkDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "批量删除切片请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class RemoveReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -108,6 +114,7 @@ public class ChunkDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "文档切片信息")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class InfoVO implements Serializable {
private static final long serialVersionUID = 1L;
@@ -143,8 +150,12 @@ public class ChunkDTO {
@Schema(description = "切片是否可用 (true: 参与检索, false: 被禁用)")
private Boolean available;
@Schema(description = "切片在原文中的位置索引列表")
private List<Integer> positions;
@Schema(description = "切片在原文中的位置索引列表 (RAGFlow返回嵌套数组, 如 [[start, end, filename]])")
private List<List<Object>> positions;
@Schema(description = "Token ID 列表")
@JsonProperty("token")
private List<Integer> token;
}
/**
@@ -155,6 +166,7 @@ public class ChunkDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "分片列表聚合响应")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class ListVO implements Serializable {
private static final long serialVersionUID = 1L;
main/manager-api/src/main/java/xiaozhi/modules/knowledge/dto/document/DocumentDTO.java
+30 -8
View File
@@ -4,17 +4,17 @@ import lombok.*;
import io.swagger.v3.oas.annotations.media.Schema;
import java.io.Serializable;
import java.util.List;
import java.util.Date;
import java.util.Map;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonAlias;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import jakarta.validation.constraints.*;
/**
* 文档管理聚合 DTO
*/
@Schema(description = "文档管理聚合 DTO")
@JsonIgnoreProperties(ignoreUnknown = true)
public class DocumentDTO {
/**
@@ -25,6 +25,7 @@ public class DocumentDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "上传文档请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class UploadReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -33,10 +34,25 @@ public class DocumentDTO {
@NotBlank(message = "知识库ID不能为空")
private String datasetId;
@Schema(description = "虚拟父级目录路径 (默认为 /)")
@Schema(description = "文件名 (如果指定，则覆盖原始文件名)")
private String name;
@Schema(description = "分块方法")
@JsonProperty("chunk_method")
private DocumentDTO.InfoVO.ChunkMethod chunkMethod;
@Schema(description = "解析参数配置")
@JsonProperty("parser_config")
private DocumentDTO.InfoVO.ParserConfig parserConfig;
@Schema(description = "虚拟文件夹路径 (默认为 /)")
@JsonProperty("parent_path")
private String parentPath;
@Schema(description = "元数据字段")
@JsonProperty("meta")
private Map<String, Object> metaFields;
@Schema(description = "文件二进制流 (支持 PDF, DOCX, TXT, MD 等多种格式)", requiredMode = Schema.RequiredMode.REQUIRED)
@NotNull(message = "上传文件不能为空")
private org.springframework.web.multipart.MultipartFile file;
@@ -50,6 +66,7 @@ public class DocumentDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "更新文档请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class UpdateReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -76,6 +93,7 @@ public class DocumentDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "获取文档列表请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class ListReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -124,6 +142,7 @@ public class DocumentDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "批量文档操作请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class BatchIdReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -142,6 +161,7 @@ public class DocumentDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "知识库文档信息")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class InfoVO implements Serializable {
private static final long serialVersionUID = 1L;
@@ -202,10 +222,9 @@ public class DocumentDTO {
@JsonProperty("progress_msg")
private String progressMsg;
@Schema(description = "开始处理的时间戳")
@Schema(description = "开始处理的时间戳 (RAGFlow返回RFC1123格式)")
@JsonProperty("process_begin_at")
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date processBeginAt;
private String processBeginAt;
@Schema(description = "处理总耗时 (单位: 秒)")
@JsonProperty("process_duration")
@@ -228,7 +247,7 @@ public class DocumentDTO {
@JsonProperty("create_time")
private Long createTime;
@Schema(description = "创建日期 (格式: yyyy-MM-dd HH:mm:ss)")
@Schema(description = "创建日期 (RAGFlow返回RFC1123格式)")
@JsonProperty("create_date")
private String createDate;
@@ -236,7 +255,7 @@ public class DocumentDTO {
@JsonProperty("update_time")
private Long updateTime;
@Schema(description = "最后更新日期 (格式: yyyy-MM-dd HH:mm:ss)")
@Schema(description = "最后更新日期 (RAGFlow返回RFC1123格式)")
@JsonProperty("update_date")
private String updateDate;
@@ -320,6 +339,7 @@ public class DocumentDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "文档解析器参数配置")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class ParserConfig implements Serializable {
private static final long serialVersionUID = 1L;
@@ -362,6 +382,7 @@ public class DocumentDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "RAPTOR (递归摘要索引) 配置")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class RaptorConfig implements Serializable {
private static final long serialVersionUID = 1L;
@Schema(description = "是否启用 RAPTOR 索引")
@@ -374,6 +395,7 @@ public class DocumentDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "GraphRAG (图增强检索) 配置")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class GraphRagConfig implements Serializable {
private static final long serialVersionUID = 1L;
@Schema(description = "是否启用 GraphRAG 索引")
main/manager-api/src/main/java/xiaozhi/modules/knowledge/dto/document/RetrievalDTO.java
+34 -2
View File
@@ -6,12 +6,15 @@ import java.io.Serializable;
import java.util.List;
import java.util.Map;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import jakarta.validation.constraints.*;
/**
* 检索与元数据管理聚合 DTO
*/
@Schema(description = "检索与元数据管理聚合 DTO")
@JsonIgnoreProperties(ignoreUnknown = true)
public class RetrievalDTO {
/**
@@ -22,6 +25,7 @@ public class RetrievalDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "文档聚合信息")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class DocAggVO implements Serializable {
private static final long serialVersionUID = 1L;
@@ -45,6 +49,8 @@ public class RetrievalDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "检索测试请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonInclude(JsonInclude.Include.NON_NULL)
public static class TestReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -53,10 +59,21 @@ public class RetrievalDTO {
@NotEmpty(message = "知识库ID列表不能为空")
private List<String> datasetIds;
@Schema(description = "文档 ID 列表 (可选，用于限定检索范围)")
@JsonProperty("document_ids")
private List<String> documentIds;
@Schema(description = "检索问题", requiredMode = Schema.RequiredMode.REQUIRED)
@NotBlank(message = "检索问题不能为空")
private String question;
@Schema(description = "页码 (默认 1)")
private Integer page;
@Schema(description = "每页数量 (默认 10)")
@JsonProperty("page_size")
private Integer pageSize;
@Schema(description = "相似度阈值 (默认 0.2)")
@JsonProperty("similarity_threshold")
private Float similarityThreshold;
@@ -78,6 +95,14 @@ public class RetrievalDTO {
@Schema(description = "是否启用关键词检索")
private Boolean keyword;
@Schema(description = "跨语言翻译列表 (可选)")
@JsonProperty("cross_languages")
private List<String> crossLanguages;
@Schema(description = "元数据过滤条件 (JSON 对象)")
@JsonProperty("metadata_condition")
private Map<String, Object> metadataCondition;
}
/**
@@ -88,6 +113,7 @@ public class RetrievalDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "检索命中切片详情")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class HitVO implements Serializable {
private static final long serialVersionUID = 1L;
@@ -141,8 +167,8 @@ public class RetrievalDTO {
@JsonProperty("image_id")
private String imageId;
@Schema(description = "位置索引")
private List<Integer> positions;
@Schema(description = "位置索引 (RAGFlow返回嵌套数组, 如 [[start, end, filename]])")
private Object positions;
}
/**
@@ -153,6 +179,7 @@ public class RetrievalDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "知识库元数据摘要信息")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class MetaSummaryVO implements Serializable {
private static final long serialVersionUID = 1L;
@@ -185,6 +212,7 @@ public class RetrievalDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "批量更新元数据请求参数")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class MetaBatchReq implements Serializable {
private static final long serialVersionUID = 1L;
@@ -205,6 +233,7 @@ public class RetrievalDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "元数据更新筛选器")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Selector implements Serializable {
private static final long serialVersionUID = 1L;
@@ -225,6 +254,7 @@ public class RetrievalDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "元数据更新项")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class UpdateItem implements Serializable {
private static final long serialVersionUID = 1L;
@@ -243,6 +273,7 @@ public class RetrievalDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "元数据删除项")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class DeleteItem implements Serializable {
private static final long serialVersionUID = 1L;
@@ -259,6 +290,7 @@ public class RetrievalDTO {
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "召回测试结果聚合响应")
@JsonIgnoreProperties(ignoreUnknown = true)
public static class ResultVO implements Serializable {
private static final long serialVersionUID = 1L;
main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/KnowledgeBaseAdapter.java
+29 -48
View File
@@ -3,10 +3,11 @@ package xiaozhi.modules.knowledge.rag;
import java.util.List;
import java.util.Map;
import org.springframework.web.multipart.MultipartFile;
import xiaozhi.modules.knowledge.dto.dataset.DatasetDTO;
import xiaozhi.common.page.PageData;
import xiaozhi.modules.knowledge.dto.KnowledgeFilesDTO;
import xiaozhi.modules.knowledge.dto.document.DocumentDTO;
/**
* 知识库API适配器抽象基类
@@ -46,35 +47,24 @@ public abstract class KnowledgeBaseAdapter {
* @return 分页数据
*/
public abstract PageData<KnowledgeFilesDTO> getDocumentList(String datasetId,
Map<String, Object> queryParams,
Integer page,
Integer limit);
DocumentDTO.ListReq req);
/**
* 根据文档ID获取文档详情
*
* @param datasetId 知识库ID
* @return 文档详情
* @param datasetId 知识库ID
* @param documentId 文档ID
* @return 文档详情 (强类型 InfoVO)
*/
public abstract KnowledgeFilesDTO getDocumentById(String datasetId, String documentId);
public abstract DocumentDTO.InfoVO getDocumentById(String datasetId, String documentId);
/**
* 上传文档到知识库
*
* @param datasetId 知识库ID
* @param file 上传的文件
* @param name 文档名称
* @param metaFields 元数据字段
* @param chunkMethod 分块方法
* @param parserConfig 解析器配置
* @param req 上传请求参数
* @return 上传的文档信息
*/
public abstract KnowledgeFilesDTO uploadDocument(String datasetId,
MultipartFile file,
String name,
Map<String, Object> metaFields,
String chunkMethod,
Map<String, Object> parserConfig);
public abstract KnowledgeFilesDTO uploadDocument(DocumentDTO.UploadReq req);
/**
* 根据状态分页查询文档列表
@@ -91,12 +81,12 @@ public abstract class KnowledgeBaseAdapter {
Integer limit);
/**
* 删除文档
* 删除文档 (支持批量删除)
*
* @param datasetId 知识库ID
* @param documentId 文档ID
* @param datasetId 知识库ID
* @param req 包含文档ID列表的请求对象
*/
public abstract void deleteDocument(String datasetId, String documentId);
public abstract void deleteDocument(String datasetId, DocumentDTO.BatchIdReq req);
/**
* 解析文档（切块）
@@ -112,32 +102,21 @@ public abstract class KnowledgeBaseAdapter {
*
* @param datasetId 知识库ID
* @param documentId 文档ID
* @param keywords 关键词过滤
* @param page 页码
* @param pageSize 每页数量
* @param chunkId 切片ID
* @return 切片列表信息
* @param req 列表请求参数 (分页、关键词等)
* @return 切片列表VO
*/
public abstract xiaozhi.modules.knowledge.dto.document.ChunkDTO.ListVO listChunks(String datasetId,
String documentId,
String keywords,
Integer page,
Integer pageSize,
String chunkId);
xiaozhi.modules.knowledge.dto.document.ChunkDTO.ListReq req);
/**
* 召回测试 - 从知识库中检索相关切片
*
* @param question 用户查询
* @param datasetIds 数据集ID列表
* @param documentIds 文档ID列表
* @param retrievalParams 检索参数
* @param req 检索测试请求参数
* @return 召回测试结果
*/
public abstract xiaozhi.modules.knowledge.dto.document.RetrievalDTO.ResultVO retrievalTest(String question,
List<String> datasetIds,
List<String> documentIds,
Map<String, Object> retrievalParams);
public abstract xiaozhi.modules.knowledge.dto.document.RetrievalDTO.ResultVO retrievalTest(
xiaozhi.modules.knowledge.dto.document.RetrievalDTO.TestReq req);
/**
* 测试连接
@@ -170,25 +149,27 @@ public abstract class KnowledgeBaseAdapter {
/**
* 创建数据集
*
* @param createParams 创建参数
* @return 数据集ID
* @param req 创建参数
* @return 数据集详情
*/
public abstract Map<String, Object> createDataset(Map<String, Object> createParams);
public abstract DatasetDTO.InfoVO createDataset(DatasetDTO.CreateReq req);
/**
* 更新数据集
*
* @param datasetId 数据集ID
* @param updateParams 更新参数
* @param datasetId 数据集ID
* @param req 更新参数
* @return 数据集详情
*/
public abstract void updateDataset(String datasetId, Map<String, Object> updateParams);
public abstract DatasetDTO.InfoVO updateDataset(String datasetId, DatasetDTO.UpdateReq req);
/**
* 删除数据集
*
* @param datasetId 数据集ID
* @param req 删除请求参数（包含ID列表）
* @return 批量操作结果
*/
public abstract void deleteDataset(String datasetId);
public abstract DatasetDTO.BatchOperationVO deleteDataset(DatasetDTO.BatchIdReq req);
/**
* 获取数据集的文档数量
main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/RAGFlowClient.java
+17 -4
View File
@@ -43,6 +43,10 @@ public class RAGFlowClient {
this.baseUrl = baseUrl.endsWith("/") ? baseUrl.substring(0, baseUrl.length() - 1) : baseUrl;
this.apiKey = apiKey;
this.objectMapper = new ObjectMapper();
// [Reinforce] 兼容 RAGFlow 返回的 RFC 1123 日期格式 (如: Tue, 10 Feb 2026 10:27:35 GMT)
this.objectMapper
.setDateFormat(new java.text.SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss 'GMT'", java.util.Locale.US));
this.objectMapper.setTimeZone(java.util.TimeZone.getTimeZone("GMT"));
// 优先从 Spring 上下文中获取池化的 RestTemplate Bean (Issue 3: 连接池化)
RestTemplate pooledTemplate = null;
@@ -52,11 +56,12 @@ public class RAGFlowClient {
log.warn("无法从 SpringContext 获取池化 RestTemplate，将退化为简单连接模式: {}", e.getMessage());
}
if (pooledTemplate != null) {
if (false) { // Force new RestTemplate for debugging
this.restTemplate = pooledTemplate;
log.debug("RAGFlowClient 已成功挂载全局池化 RestTemplate");
} else {
// 兜底方案：配置超时并创建简单 RestTemplate
log.info("RAGFlowClient 初始化: 使用独立 RestTemplate (Debug Mode)");
org.springframework.http.client.SimpleClientHttpRequestFactory factory = new org.springframework.http.client.SimpleClientHttpRequestFactory();
factory.setConnectTimeout(timeoutSeconds * 1000);
factory.setReadTimeout(timeoutSeconds * 1000);
@@ -78,8 +83,14 @@ public class RAGFlowClient {
*/
public Map<String, Object> post(String endpoint, Object body) {
String url = buildUrl(endpoint, null);
log.debug("POST {}", url);
return execute(url, HttpMethod.POST, body);
log.info("RAGFlow Client POST Request: URL={}, BodyType={}", url,
body != null ? body.getClass().getName() : "null");
try {
return execute(url, HttpMethod.POST, body);
} catch (Exception e) {
log.error("RAGFlow Client POST Failed: URL={}", url, e);
throw e;
}
}
/**
@@ -163,7 +174,9 @@ public class RAGFlowClient {
} catch (RenException re) {
throw re;
} catch (Exception e) {
log.error("RAGFlow Client Execute Error: {}", e.getMessage(), e);
log.error("RAGFlow Client Execute Error! URL: {}, Method: {}, Body Type: {}", url, method,
requestEntity.getBody() != null ? requestEntity.getBody().getClass().getName() : "null");
log.error("Full exception stack trace: ", e);
throw new RenException(ErrorCode.RAG_API_ERROR, "Request Failed: " + e.getMessage());
}
}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/impl/RAGFlowAdapter.java
+203 -176
View File
@@ -4,7 +4,6 @@ import java.io.IOException;
import java.io.InputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
@@ -15,8 +14,8 @@ import org.springframework.core.io.AbstractResource;
import org.springframework.util.LinkedMultiValueMap;
import org.springframework.util.MultiValueMap;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.util.UriComponentsBuilder;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
@@ -25,6 +24,8 @@ import xiaozhi.common.exception.RenException;
import xiaozhi.common.page.PageData;
import xiaozhi.modules.knowledge.dto.KnowledgeFilesDTO;
import xiaozhi.modules.knowledge.dto.dataset.DatasetDTO;
import xiaozhi.modules.knowledge.dto.document.ChunkDTO;
import xiaozhi.modules.knowledge.dto.document.RetrievalDTO;
import xiaozhi.modules.knowledge.dto.document.DocumentDTO;
import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapter;
import xiaozhi.modules.knowledge.rag.RAGFlowClient;
@@ -60,13 +61,16 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
public void initialize(Map<String, Object> config) {
this.config = config;
validateConfig(config);
String baseUrl = (String) config.get("base_url");
String apiKey = (String) config.get("api_key");
String baseUrl = getConfigValue(config, "base_url", "baseUrl");
String apiKey = getConfigValue(config, "api_key", "apiKey");
// 初始化 Client，默认超时 30s，可通过 config 扩展
int timeout = 30;
if (config.containsKey("timeout")) {
Object timeoutObj = getConfigValue(config, "timeout", "timeout");
if (timeoutObj != null) {
try {
timeout = Integer.parseInt(config.get("timeout").toString());
timeout = Integer.parseInt(timeoutObj.toString());
} catch (Exception e) {
log.warn("解析超时配置失败，使用默认值 30s");
}
@@ -81,8 +85,8 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
throw new RenException(ErrorCode.RAG_CONFIG_NOT_FOUND);
}
String baseUrl = (String) config.get("base_url");
String apiKey = (String) config.get("api_key");
String baseUrl = getConfigValue(config, "base_url", "baseUrl");
String apiKey = getConfigValue(config, "api_key", "apiKey");
if (StringUtils.isBlank(baseUrl)) {
throw new RenException(ErrorCode.RAG_API_ERROR_URL_NULL);
@@ -103,6 +107,19 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
return true;
}
/**
* 辅助方法：支持多种键名获取配置（兼容 camelCase 和 snake_case）
*/
private String getConfigValue(Map<String, Object> config, String snakeKey, String camelKey) {
if (config.containsKey(snakeKey)) {
return (String) config.get(snakeKey);
}
if (config.containsKey(camelKey)) {
return (String) config.get(camelKey);
}
return null;
}
/**
* 辅助方法：确保 Client 已初始化
*/
@@ -126,43 +143,19 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
}
@Override
public PageData<KnowledgeFilesDTO> getDocumentList(String datasetId, Map<String, Object> queryParams, Integer page,
Integer limit) {
public PageData<KnowledgeFilesDTO> getDocumentList(String datasetId, DocumentDTO.ListReq req) {
try {
log.info("=== [RAGFlow] 获取文档列表: datasetId={} ===", datasetId);
// 构造参数
Map<String, Object> params = new HashMap<>();
if (page != null && page > 0)
params.put("page", page);
if (limit != null && limit > 0)
params.put("page_size", limit);
if (queryParams != null) {
// 兼容旧逻辑的特殊 parameter mapping
if (queryParams.containsKey("name")) {
params.put("keywords", queryParams.get("name"));
}
if (queryParams.containsKey("orderby")) {
params.put("orderby", queryParams.get("orderby"));
}
if (queryParams.containsKey("desc")) {
params.put("desc", queryParams.get("desc"));
}
if (queryParams.containsKey("id")) {
params.put("id", queryParams.get("id"));
}
if (queryParams.containsKey("run")) { // Run status
params.put("run", queryParams.get("run"));
}
// 处理时间范围等其他参数，旧逻辑中有很多 if，这里简化透传，RAGFlow Client 会处理 map
// 如果需要严格兼容旧逻辑的 parameter transform，可以在这里补全，但 queryParams 大多 key 是直接透传的
}
// 使用 Jackson 将 DTO 转为 Map 作为查询参数
@SuppressWarnings("unchecked")
Map<String, Object> params = objectMapper.convertValue(req, Map.class);
Map<String, Object> response = getClient().get("/api/v1/datasets/" + datasetId + "/documents", params);
Object dataObj = response.get("data");
return parseDocumentListResponse(dataObj, page != null ? page : 1, limit != null ? limit : 10);
return parseDocumentListResponse(dataObj, req.getPage() != null ? req.getPage() : 1,
req.getPageSize() != null ? req.getPageSize() : 10);
} catch (Exception e) {
log.error("获取文档列表失败", e);
@@ -171,43 +164,59 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
}
@Override
public KnowledgeFilesDTO getDocumentById(String datasetId, String documentId) {
public DocumentDTO.InfoVO getDocumentById(String datasetId, String documentId) {
try {
log.info("=== [RAGFlow] 获取文档详情: datasetId={}, documentId={} ===", datasetId, documentId);
Map<String, Object> queryParams = new HashMap<>();
queryParams.put("id", documentId);
PageData<KnowledgeFilesDTO> list = getDocumentList(datasetId, queryParams, 1, 1);
if (list != null && list.getList() != null && !list.getList().isEmpty()) {
return list.getList().get(0);
DocumentDTO.ListReq req = DocumentDTO.ListReq.builder()
.id(documentId)
.page(1)
.pageSize(1)
.build();
@SuppressWarnings("unchecked")
Map<String, Object> params = objectMapper.convertValue(req, Map.class);
Map<String, Object> response = getClient().get("/api/v1/datasets/" + datasetId + "/documents", params);
Object dataObj = response.get("data");
if (dataObj instanceof Map) {
Map<String, Object> dataMap = (Map<String, Object>) dataObj;
List<?> documents = (List<?>) dataMap.get("docs");
if (documents != null && !documents.isEmpty()) {
return objectMapper.convertValue(documents.get(0), DocumentDTO.InfoVO.class);
}
}
return null;
} catch (Exception e) {
log.error("获取文档详情失败: {}", documentId, e);
log.error("获取文档详情失败: documentId={}", documentId, e);
throw convertToRenException(e);
}
}
@Override
public KnowledgeFilesDTO uploadDocument(String datasetId, MultipartFile file, String name,
Map<String, Object> metaFields, String chunkMethod,
Map<String, Object> parserConfig) {
public KnowledgeFilesDTO uploadDocument(DocumentDTO.UploadReq req) {
String datasetId = req.getDatasetId();
MultipartFile file = req.getFile();
try {
log.info("=== [RAGFlow] 上传文档: datasetId={} ===", datasetId);
MultiValueMap<String, Object> body = new LinkedMultiValueMap<>();
body.add("file", new MultipartFileResource(file));
if (StringUtils.isNotBlank(name)) {
body.add("name", name);
if (StringUtils.isNotBlank(req.getName())) {
body.add("name", req.getName());
}
if (metaFields != null && !metaFields.isEmpty()) {
body.add("meta", objectMapper.writeValueAsString(metaFields));
if (req.getMetaFields() != null && !req.getMetaFields().isEmpty()) {
body.add("meta", objectMapper.writeValueAsString(req.getMetaFields()));
}
if (StringUtils.isNotBlank(chunkMethod)) {
body.add("chunk_method", chunkMethod);
if (req.getChunkMethod() != null) {
// 将枚举值转为 RAGFlow 期待的字符串（如 NAIVE -> naive）
body.add("chunk_method", req.getChunkMethod().name().toLowerCase());
}
if (parserConfig != null && !parserConfig.isEmpty()) {
body.add("parser_config", objectMapper.writeValueAsString(parserConfig));
if (req.getParserConfig() != null) {
body.add("parser_config", objectMapper.writeValueAsString(req.getParserConfig()));
}
if (StringUtils.isNotBlank(req.getParentPath())) {
body.add("parent_path", req.getParentPath());
}
Map<String, Object> response = getClient().postMultipart("/api/v1/datasets/" + datasetId + "/documents",
@@ -225,45 +234,45 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
@Override
public PageData<KnowledgeFilesDTO> getDocumentListByStatus(String datasetId, Integer status, Integer page,
Integer limit) {
Map<String, Object> queryParams = new HashMap<>();
List<DocumentDTO.InfoVO.RunStatus> runStatusList = null;
if (status != null) {
String runStatus;
runStatusList = new ArrayList<>();
switch (status) {
case 0:
runStatus = "UNSTART";
runStatusList.add(DocumentDTO.InfoVO.RunStatus.UNSTART);
break;
case 1:
runStatus = "RUNNING";
runStatusList.add(DocumentDTO.InfoVO.RunStatus.RUNNING);
break;
case 2:
runStatus = "CANCEL";
runStatusList.add(DocumentDTO.InfoVO.RunStatus.CANCEL);
break;
case 3:
runStatus = "DONE";
runStatusList.add(DocumentDTO.InfoVO.RunStatus.DONE);
break;
case 4:
runStatus = "FAIL";
runStatusList.add(DocumentDTO.InfoVO.RunStatus.FAIL);
break;
default:
runStatus = status.toString(); // Support implicit numbers
break;
}
queryParams.put("run", runStatus);
}
return getDocumentList(datasetId, queryParams, page, limit);
DocumentDTO.ListReq req = DocumentDTO.ListReq.builder()
.run(runStatusList)
.page(page)
.pageSize(limit)
.build();
return getDocumentList(datasetId, req);
}
@Override
public void deleteDocument(String datasetId, String documentId) {
public void deleteDocument(String datasetId, DocumentDTO.BatchIdReq req) {
try {
log.info("=== [RAGFlow] 删除文档: {} ===", documentId);
Map<String, Object> body = new HashMap<>();
body.put("ids", Collections.singletonList(documentId));
getClient().delete("/api/v1/datasets/" + datasetId + "/documents", body);
log.info("=== [RAGFlow] 批量删除文档: datasetId={}, count={} ===", datasetId,
req.getIds() != null ? req.getIds().size() : 0);
getClient().delete("/api/v1/datasets/" + datasetId + "/documents", req);
} catch (Exception e) {
log.error("删除文档失败", e);
log.error("批量删除文档失败: datasetId={}", datasetId, e);
throw convertToRenException(e);
}
}
@@ -284,85 +293,75 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
}
@Override
public xiaozhi.modules.knowledge.dto.document.ChunkDTO.ListVO listChunks(String datasetId, String documentId,
String keywords,
Integer page, Integer pageSize, String chunkId) {
public ChunkDTO.ListVO listChunks(String datasetId, String documentId, ChunkDTO.ListReq req) {
try {
Map<String, Object> params = new HashMap<>();
if (StringUtils.isNotBlank(keywords))
params.put("keywords", keywords);
if (page != null)
params.put("page", page);
if (pageSize != null)
params.put("page_size", pageSize);
if (StringUtils.isNotBlank(chunkId))
params.put("id", chunkId);
// [提灯重构] 使用 objectMapper 动态转换查询参数，消除硬编码
Map<String, Object> params = objectMapper.convertValue(req, new TypeReference<Map<String, Object>>() {
});
Map<String, Object> response = getClient()
.get("/api/v1/datasets/" + datasetId + "/documents/" + documentId + "/chunks", params);
// [提灯审计] 暗礁 2 & 6: 增加 NPE 检查并使用强类型 DTO 转换
Object dataObj = response.get("data");
if (dataObj == null) {
log.warn("[RAGFlow] listChunks 响应 data 为空, docId={}", documentId);
return xiaozhi.modules.knowledge.dto.document.ChunkDTO.ListVO.builder()
return ChunkDTO.ListVO.builder()
.chunks(new ArrayList<>())
.total(0L)
.build();
}
// 直接转换 DTO，保证字段全量映射
xiaozhi.modules.knowledge.dto.document.ChunkDTO.ListVO result = objectMapper.convertValue(dataObj,
xiaozhi.modules.knowledge.dto.document.ChunkDTO.ListVO.class);
// [提灯审计] 暗礁 5: 增加 Total 兜底处理
ChunkDTO.ListVO result = objectMapper.convertValue(dataObj, ChunkDTO.ListVO.class);
if (result.getTotal() == null) {
result.setTotal(0L);
}
return result;
} catch (Exception e) {
log.error("获取切片失败", e);
log.error("获取切片失败: docId={}", documentId, e);
throw convertToRenException(e);
}
}
@Override
public xiaozhi.modules.knowledge.dto.document.RetrievalDTO.ResultVO retrievalTest(String question,
List<String> datasetIds, List<String> documentIds,
Map<String, Object> retrievalParams) {
public RetrievalDTO.ResultVO retrievalTest(RetrievalDTO.TestReq req) {
try {
Map<String, Object> body = new HashMap<>();
body.put("question", question);
if (datasetIds != null)
body.put("dataset_ids", datasetIds);
if (documentIds != null)
body.put("document_ids", documentIds);
if (retrievalParams != null)
body.putAll(retrievalParams);
// [Production Reinforce] 参数防御性对齐：RAGFlow Python 端对 0 或负数分页敏感
// 解决 ValueError('Search does not support negative slicing.')
if (req.getPage() != null && req.getPage() < 1) {
req.setPage(1);
}
if (req.getPageSize() != null && req.getPageSize() < 1) {
req.setPageSize(10); // 默认 10 条
}
if (req.getTopK() != null && req.getTopK() < 1) {
req.setTopK(1024); // RAGFlow 内部默认 TopK
}
// 相似度阈值归一化 (0.0 ~ 1.0)
if (req.getSimilarityThreshold() != null) {
if (req.getSimilarityThreshold() < 0f)
req.setSimilarityThreshold(0.2f);
if (req.getSimilarityThreshold() > 1f)
req.setSimilarityThreshold(1.0f);
}
Map<String, Object> response = getClient().post("/api/v1/retrieval", body);
// [提灯重构] 直接透传强类型 DTO，由 getClient 处理序列化
Map<String, Object> response = getClient().post("/api/v1/retrieval", req);
// [提灯审计] DTO 化重构：增加 NPE 防护
Object dataObj = response.get("data");
if (dataObj == null) {
log.warn("[RAGFlow] retrievalTest 响应 data 为空");
return xiaozhi.modules.knowledge.dto.document.RetrievalDTO.ResultVO.builder()
return RetrievalDTO.ResultVO.builder()
.chunks(new ArrayList<>())
.docAggs(new ArrayList<>())
.total(0L)
.build();
}
xiaozhi.modules.knowledge.dto.document.RetrievalDTO.ResultVO result = objectMapper.convertValue(dataObj,
xiaozhi.modules.knowledge.dto.document.RetrievalDTO.ResultVO.class);
RetrievalDTO.ResultVO result = objectMapper.convertValue(dataObj, RetrievalDTO.ResultVO.class);
if (result.getTotal() == null) {
result.setTotal(0L);
}
return result;
} catch (Exception e) {
log.error("召回测试失败", e);
throw convertToRenException(e);
@@ -407,18 +406,40 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
}
@Override
public Map<String, Object> createDataset(Map<String, Object> createParams) {
public DatasetDTO.InfoVO createDataset(DatasetDTO.CreateReq req) {
try {
// Service 层已经处理了命名前缀逻辑
Map<String, Object> response = getClient().post("/api/v1/datasets", createParams);
// [Production Fix] 强化默认值处理，防止 RAGFlow API 因空字符串或缺失字段报错 (Code 101)
// 解决 "Field: <avatar> - Message: <Missing MIME prefix>" 等校验失败
if (StringUtils.isBlank(req.getPermission())) {
req.setPermission("me");
}
if (StringUtils.isBlank(req.getChunkMethod())) {
req.setChunkMethod("naive");
}
// 🤖 自动补全嵌入模型：优先使用请求传参，其次使用配置中的默认模型
if (StringUtils.isBlank(req.getEmbeddingModel())) {
String defaultModel = (String) getConfigValue(config, "embedding_model", "embeddingModel");
if (StringUtils.isNotBlank(defaultModel)) {
log.info("RAGFlow: 使用配置中的默认嵌入模型: {}", defaultModel);
req.setEmbeddingModel(defaultModel);
}
// 若配置中也无默认值，则留空由 RAGFlow 服务端自行兜底（或抛出业务异常）
}
// 🖼️ 自动补全头像：若为空则提供一个 1x1 透明像素，防止 RAGFlow 校验 MIME Prefix 失败
if (StringUtils.isBlank(req.getAvatar())) {
req.setAvatar(
"data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg==");
}
// 直接将强类型请求对象传给 Client，Jackson 会处理 JsonProperty 映射
Map<String, Object> response = getClient().post("/api/v1/datasets", req);
// 安全地获取 data 并通过 DatasetDTO.InfoVO 进行全量映射
Object dataObj = response.get("data");
if (dataObj instanceof Map) {
DatasetDTO.InfoVO info = objectMapper.convertValue(dataObj, DatasetDTO.InfoVO.class);
return objectMapper.convertValue(info,
new com.fasterxml.jackson.core.type.TypeReference<Map<String, Object>>() {
});
if (dataObj != null) {
return objectMapper.convertValue(dataObj, DatasetDTO.InfoVO.class);
}
throw new RenException(ErrorCode.RAG_API_ERROR, "Invalid response from createDataset: missing data object");
} catch (Exception e) {
@@ -428,16 +449,16 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
}
@Override
public void updateDataset(String datasetId, Map<String, Object> updateParams) {
public DatasetDTO.InfoVO updateDataset(String datasetId, DatasetDTO.UpdateReq req) {
try {
// RAGFlow API 更新通常需要 Body 里带 ID，或者是 PUT /datasets (id在Body)
// 根据 reverse analysis，原逻辑就是调 adapter.updateDataset.
// 确保 ID 存在
if (!updateParams.containsKey("id")) {
updateParams.put("id", datasetId);
// RAGFlow API 更新建议路径带 ID
Map<String, Object> response = getClient().put("/api/v1/datasets/" + datasetId, req);
Object dataObj = response.get("data");
if (dataObj != null) {
return objectMapper.convertValue(dataObj, DatasetDTO.InfoVO.class);
}
// 使用 PUT /api/v1/datasets
getClient().put("/api/v1/datasets", updateParams);
return null;
} catch (Exception e) {
log.error("更新数据集失败", e);
throw convertToRenException(e);
@@ -445,13 +466,18 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
}
@Override
public void deleteDataset(String datasetId) {
public DatasetDTO.BatchOperationVO deleteDataset(DatasetDTO.BatchIdReq req) {
try {
Map<String, Object> body = new HashMap<>();
body.put("ids", Collections.singletonList(datasetId));
getClient().delete("/api/v1/datasets", body);
// RAGFlow 批量删除接口使用 DELETE /api/v1/datasets
Map<String, Object> response = getClient().delete("/api/v1/datasets", req);
Object dataObj = response.get("data");
if (dataObj != null) {
return objectMapper.convertValue(dataObj, DatasetDTO.BatchOperationVO.class);
}
return null;
} catch (Exception e) {
log.error("删除数据集失败", e);
log.error("批量删除数据集失败", e);
throw convertToRenException(e);
}
}
@@ -459,19 +485,31 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
@Override
public Integer getDocumentCount(String datasetId) {
try {
// 改为调用 /datasets/{id} 获取详情
Map<String, Object> response = getClient().get("/api/v1/datasets/" + datasetId, null);
// [Fix] 使用列表过滤接口获取详情 (GET /datasets?id={id})
Map<String, Object> params = new HashMap<>();
params.put("id", datasetId);
params.put("page", 1);
params.put("page_size", 1);
Map<String, Object> response = getClient().get("/api/v1/datasets", params);
Object dataObj = response.get("data");
if (dataObj instanceof Map) {
Object countObj = ((Map<?, ?>) dataObj).get("doc_count");
if (countObj instanceof Number) {
return ((Number) countObj).intValue();
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 0;
} catch (Exception e) {
log.warn("获取文档数量失败: {}", e.getMessage());
// 降级：不抛错，返回 0 (for Stats loop safety)
return 0;
}
}
@@ -520,47 +558,36 @@ public class RAGFlowAdapter extends KnowledgeBaseAdapter {
}
// 复用原有的辅助解析方法，保持兼容
// [Bug Fix] 不再吞掉反序列化异常，避免上层误判"文档已删除"
private PageData<KnowledgeFilesDTO> parseDocumentListResponse(Object dataObj, long curPage, long pageSize) {
try {
if (dataObj == null) {
return new PageData<>(new ArrayList<>(), 0);
}
Map<String, Object> dataMap = (Map<String, Object>) dataObj;
List<Map<String, Object>> documents = (List<Map<String, Object>>) dataMap.get("docs");
if (documents == null || documents.isEmpty()) {
return new PageData<>(new ArrayList<>(), 0);
}
List<KnowledgeFilesDTO> list = new ArrayList<>();
for (Object docObj : documents) {
DocumentDTO.InfoVO info = objectMapper.convertValue(docObj, DocumentDTO.InfoVO.class);
list.add(mapToKnowledgeFilesDTO(info, null)); // datasetId is usually in InfoVO
}
long total = 0;
if (dataMap.containsKey("total")) {
total = ((Number) dataMap.get("total")).longValue();
}
return new PageData<>(list, total);
} catch (Exception e) {
log.error("解析文档列表失败", e);
if (dataObj == null) {
return new PageData<>(new ArrayList<>(), 0);
}
}
// Helper to parse time from Number or String
private Date parseTime(Object obj) {
if (obj instanceof Number)
return new Date(((Number) obj).longValue());
if (obj instanceof String) {
Map<String, Object> dataMap = (Map<String, Object>) dataObj;
List<Map<String, Object>> documents = (List<Map<String, Object>>) dataMap.get("docs");
if (documents == null || documents.isEmpty()) {
// RAGFlow 明确返回了空文档列表，这是合法的"真空"
return new PageData<>(new ArrayList<>(), 0);
}
List<KnowledgeFilesDTO> list = new ArrayList<>();
for (Object docObj : documents) {
try {
return new Date(Long.parseLong((String) obj));
// 单文档转换容错：一个文档反序列化失败不影响其他文档
DocumentDTO.InfoVO info = objectMapper.convertValue(docObj, DocumentDTO.InfoVO.class);
list.add(mapToKnowledgeFilesDTO(info, null));
} catch (Exception e) {
log.warn("[RAGFlow] 单文档 DTO 转换失败，跳过该文档: {}", e.getMessage());
}
}
return new Date();
long total = 0;
if (dataMap.containsKey("total")) {
total = ((Number) dataMap.get("total")).longValue();
}
return new PageData<>(list, total);
}
private KnowledgeFilesDTO parseUploadResponse(Object dataObj, String datasetId, MultipartFile file) {
main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/KnowledgeBaseService.java
+10
View File
@@ -93,4 +93,14 @@ public interface KnowledgeBaseService extends BaseService<KnowledgeBaseEntity> {
* @return RAG模型列表
*/
List<ModelConfigEntity> getRAGModels();
/**
* 更新知识库统计信息 (用于被文件服务回调)
*
* @param datasetId 知识库ID
* @param docDelta 文档数增量
* @param chunkDelta 分块数增量
* @param tokenDelta Token数增量
*/
void updateStatistics(String datasetId, Integer docDelta, Long chunkDelta, Long tokenDelta);
}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/KnowledgeFilesService.java
+32 -39
View File
@@ -7,6 +7,9 @@ import org.springframework.web.multipart.MultipartFile;
import xiaozhi.common.page.PageData;
import xiaozhi.modules.knowledge.dto.KnowledgeFilesDTO;
import xiaozhi.modules.knowledge.dto.document.ChunkDTO;
import xiaozhi.modules.knowledge.dto.document.RetrievalDTO;
import xiaozhi.modules.knowledge.dto.document.DocumentDTO;
/**
* 知识库文档服务接口
@@ -28,9 +31,9 @@ public interface KnowledgeFilesService {
*
* @param documentId 文档ID
* @param datasetId 知识库ID
* @return 文档详情
* @return 文档详情 (强类型 InfoVO)
*/
KnowledgeFilesDTO getByDocumentId(String documentId, String datasetId);
DocumentDTO.InfoVO getByDocumentId(String documentId, String datasetId);
/**
* 上传文档到知识库
@@ -48,12 +51,12 @@ public interface KnowledgeFilesService {
Map<String, Object> parserConfig);
/**
* 根据文档ID和知识库ID删除文档
* 批量删除文档
*
* @param documentId 文档ID
* @param datasetId 知识库ID
* @param datasetId 知识库ID
* @param req 删除请求参数 (含文档ID列表)
*/
void deleteByDocumentId(String documentId, String datasetId);
void deleteDocuments(String datasetId, DocumentDTO.BatchIdReq req);
/**
* 获取RAG配置信息
@@ -77,40 +80,18 @@ public interface KnowledgeFilesService {
*
* @param datasetId 知识库ID
* @param documentId 文档ID
* @param keywords 关键词过滤
* @param page 页码
* @param pageSize 每页数量
* @param chunkId 切片ID
* @param req 切片列表请求参数
* @return 切片列表信息
*/
xiaozhi.modules.knowledge.dto.document.ChunkDTO.ListVO listChunks(String datasetId, String documentId,
String keywords,
Integer page, Integer pageSize, String chunkId);
ChunkDTO.ListVO listChunks(String datasetId, String documentId, ChunkDTO.ListReq req);
/**
* 召回测试 - 从指定数据集或文档中检索相关切片
* 召回测试
*
* @param question 用户查询或查询关键词
* @param datasetIds 数据集ID列表
* @param documentIds 文档ID列表
* @param page 页码
* @param pageSize 每页数量
* @param similarityThreshold 最小相似度阈值
* @param vectorSimilarityWeight 向量相似度权重
* @param topK 参与向量余弦计算的切片数量
* @param rerankId 重排模型ID
* @param keyword 是否启用关键词匹配
* @param highlight 是否启用高亮显示
* @param crossLanguages 跨语言翻译列表
* @param metadataCondition 元数据过滤条件
* @param req 检索测试请求参数
* @return 召回测试结果
*/
xiaozhi.modules.knowledge.dto.document.RetrievalDTO.ResultVO retrievalTest(String question,
List<String> datasetIds, List<String> documentIds,
Integer page, Integer pageSize, Float similarityThreshold,
Float vectorSimilarityWeight, Integer topK, String rerankId,
Boolean keyword, Boolean highlight, List<String> crossLanguages,
Map<String, Object> metadataCondition);
RetrievalDTO.ResultVO retrievalTest(RetrievalDTO.TestReq req);
/**
* 保存文档影子记录
@@ -119,12 +100,24 @@ public interface KnowledgeFilesService {
Map<String, Object> parserConfig);
/**
* 删除文档影子记录并更新统计信息
* 批量删除文档影子记录并同步统计数据
*
* @param documentId 文档ID
* @param datasetId 数据集ID
* @param chunkDelta 待扣减的分块数
* @param tokenDelta 待扣减的Token数
* @param documentIds 文档ID列表
* @param datasetId 数据集ID
* @param chunkDelta 待扣减的总分块数
* @param tokenDelta 待扣减的总Token数
*/
void deleteDocumentShadow(String documentId, String datasetId, Long chunkDelta, Long tokenDelta);
void deleteDocumentShadows(List<String> documentIds, String datasetId, Long chunkDelta, Long tokenDelta);
/**
* 根据数据集ID清理所有关联文档 (级联删除专用)
*
* @param datasetId 数据集ID
*/
void deleteDocumentsByDatasetId(String datasetId);
/**
* 同步所有处于 RUNNING 状态的文档 (供定时任务调用)
*/
void syncRunningDocuments();
}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/KnowledgeManagerService.java
+24
View File
@@ -0,0 +1,24 @@
package xiaozhi.modules.knowledge.service;
import java.util.List;
/**
* 知识库模块领域编排服务
* 用于处理跨 KnowledgeBase 和 KnowledgeFiles 的复杂业务流程，彻底解决 Service 间的循环依赖问题。
*/
public interface KnowledgeManagerService {
/**
* 级联删除知识库及其下属所有文档 (包括本地 DB 和 RAGFlow 远程数据)
*
* @param datasetId 知识库 ID
*/
void deleteDatasetWithFiles(String datasetId);
/**
* 批量级联删除知识库
*
* @param datasetIds 知识库 ID 列表
*/
void batchDeleteDatasetsWithFiles(List<String> datasetIds);
}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeBaseServiceImpl.java
+73 -85
View File
@@ -18,9 +18,9 @@ import xiaozhi.common.redis.RedisUtils;
import xiaozhi.common.service.impl.BaseServiceImpl;
import xiaozhi.common.utils.ConvertUtils;
import xiaozhi.common.utils.JsonUtils;
import xiaozhi.common.utils.ToolUtil;
import xiaozhi.modules.knowledge.dao.KnowledgeBaseDao;
import xiaozhi.modules.knowledge.dto.KnowledgeBaseDTO;
import xiaozhi.modules.knowledge.dto.dataset.DatasetDTO;
import xiaozhi.modules.knowledge.entity.KnowledgeBaseEntity;
import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapter;
import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapterFactory;
@@ -30,6 +30,7 @@ import xiaozhi.modules.model.entity.ModelConfigEntity;
import xiaozhi.modules.model.service.ModelConfigService;
import xiaozhi.modules.security.user.SecurityUser;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -50,7 +51,6 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
private final RedisUtils redisUtils;
@Override
@SuppressWarnings("deprecation")
public PageData<KnowledgeBaseDTO> getPageList(KnowledgeBaseDTO knowledgeBaseDTO, Integer page, Integer limit) {
Page<KnowledgeBaseEntity> pageInfo = new Page<>(page, limit);
QueryWrapper<KnowledgeBaseEntity> queryWrapper = new QueryWrapper<>();
@@ -102,8 +102,12 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
if (StringUtils.isBlank(datasetId)) {
throw new RenException(ErrorCode.PARAMS_GET_ERROR);
}
// [Production Fix] 兼容性查找：优先通过 dataset_id 找，找不到通过主键 id 找，确保前端传哪种 UUID 都能命中
KnowledgeBaseEntity entity = knowledgeBaseDao
.selectOne(new QueryWrapper<KnowledgeBaseEntity>().eq("dataset_id", datasetId));
.selectOne(new QueryWrapper<KnowledgeBaseEntity>()
.eq("dataset_id", datasetId)
.or()
.eq("id", datasetId));
if (entity == null) {
throw new RenException(ErrorCode.Knowledge_Base_RECORD_NOT_EXISTS);
}
@@ -112,7 +116,6 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
@Override
@Transactional(rollbackFor = Exception.class)
@SuppressWarnings("deprecation")
public KnowledgeBaseDTO save(KnowledgeBaseDTO dto) {
// 1. Validation
checkDuplicateName(dto.getName(), null);
@@ -135,66 +138,49 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
adapter = KnowledgeBaseAdapterFactory.getAdapter((String) ragConfig.get("type"),
ragConfig);
Map<String, Object> createParams = new HashMap<>();
createParams.put("name", SecurityUser.getUser().getUsername() + "_" + dto.getName());
if (StringUtils.isNotBlank(dto.getDescription())) {
createParams.put("description", dto.getDescription());
}
DatasetDTO.CreateReq createReq = ConvertUtils.sourceToTarget(dto, DatasetDTO.CreateReq.class);
createReq.setName(SecurityUser.getUser().getUsername() + "_" + dto.getName());
Map<String, Object> ragResponse = adapter.createDataset(createParams);
if (ragResponse == null || !ragResponse.containsKey("id")) {
DatasetDTO.InfoVO ragResponse = adapter.createDataset(createReq);
if (ragResponse == null || StringUtils.isBlank(ragResponse.getId())) {
throw new RenException(ErrorCode.RAG_API_ERROR, "RAG创建返回无效: 缺失ID");
}
datasetId = (String) ragResponse.get("id");
datasetId = ragResponse.getId();
// 3. Local Save (Shadow)
KnowledgeBaseEntity entity = ConvertUtils.sourceToTarget(dto, KnowledgeBaseEntity.class);
entity.setId(null);
// [Production Fix] 统一本地 ID 与 RAGFlow ID，防止前端调用 /delete 或 /update 时因 ID 混淆（本地
// UUID vs RAG UUID）导致 10163 错误
entity.setId(datasetId);
entity.setDatasetId(datasetId);
entity.setStatus(1); // Default Enabled
// ✅ FULL PERSISTENCE: 严格全量回写 (User Requirement)
if (ragResponse.containsKey("tenant_id")) {
entity.setTenantId((String) ragResponse.get("tenant_id"));
}
if (ragResponse.containsKey("chunk_method")) {
entity.setChunkMethod((String) ragResponse.get("chunk_method"));
}
if (ragResponse.containsKey("embedding_model")) {
entity.setEmbeddingModel((String) ragResponse.get("embedding_model"));
}
if (ragResponse.containsKey("permission")) {
entity.setPermission((String) ragResponse.get("permission"));
}
if (ragResponse.containsKey("avatar") && StringUtils.isBlank(entity.getAvatar())) {
entity.setAvatar((String) ragResponse.get("avatar"));
// 使用强类型 DTO 属性获取，不再从 Map 中手动解析 Key
entity.setTenantId(ragResponse.getTenantId());
entity.setChunkMethod(ragResponse.getChunkMethod());
entity.setEmbeddingModel(ragResponse.getEmbeddingModel());
entity.setPermission(ragResponse.getPermission());
if (StringUtils.isBlank(entity.getAvatar())) {
entity.setAvatar(ragResponse.getAvatar());
}
// Parse Config (JSON)
if (ragResponse.containsKey("parser_config")) {
Object parserConfig = ragResponse.get("parser_config");
entity.setParserConfig(JsonUtils.toJsonString(parserConfig));
}
// Numeric defaults
if (ragResponse.containsKey("chunk_count")) {
Object val = ragResponse.get("chunk_count");
if (val instanceof Number)
entity.setChunkCount(((Number) val).longValue());
} else {
entity.setChunkCount(0L);
if (ragResponse.getParserConfig() != null) {
entity.setParserConfig(JsonUtils.toJsonString(ragResponse.getParserConfig()));
}
if (ragResponse.containsKey("document_count")) {
Object val = ragResponse.get("document_count");
if (val instanceof Number)
entity.setDocumentCount(((Number) val).longValue());
} else {
entity.setDocumentCount(0L);
}
// Numeric fields
entity.setChunkCount(ragResponse.getChunkCount() != null ? ragResponse.getChunkCount() : 0L);
entity.setDocumentCount(ragResponse.getDocumentCount() != null ? ragResponse.getDocumentCount() : 0L);
entity.setTokenNum(ragResponse.getTokenNum() != null ? ragResponse.getTokenNum() : 0L);
// TokenNum (Default 0 as requested)
entity.setTokenNum(0L);
// 清空 creator/updater，让 FieldMetaObjectHandler 从 SecurityUser 自动填充
// ConvertUtils 会把 DTO 中的 creator=0 拷贝过来，导致 strictInsertFill 跳过填充
entity.setCreator(null);
entity.setUpdater(null);
knowledgeBaseDao.insert(entity);
return ConvertUtils.sourceToTarget(entity, KnowledgeBaseDTO.class);
@@ -204,7 +190,8 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
if (StringUtils.isNotBlank(datasetId)) {
try {
if (adapter != null)
adapter.deleteDataset(datasetId);
adapter.deleteDataset(
DatasetDTO.BatchIdReq.builder().ids(Collections.singletonList(datasetId)).build());
} catch (Exception rollbackEx) {
log.error("RAG回滚失败: {}", datasetId, rollbackEx);
}
@@ -220,9 +207,12 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
@Transactional(rollbackFor = Exception.class)
@SuppressWarnings("deprecation")
public KnowledgeBaseDTO update(KnowledgeBaseDTO dto) {
log.info("Update Service Called: ID={}, DatasetID={}", dto.getId(), dto.getDatasetId());
KnowledgeBaseEntity entity = knowledgeBaseDao.selectById(dto.getId());
if (entity == null)
if (entity == null) {
log.error("Update failed: Entity not found for ID={}", dto.getId());
throw new RenException(ErrorCode.Knowledge_Base_RECORD_NOT_EXISTS);
}
checkDuplicateName(dto.getName(), dto.getId());
@@ -245,38 +235,36 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
dto.setRagModelId(entity.getRagModelId());
}
// [FIX] 智能补全：如果 DTO 里的关键字段为空，则使用 Entity 里的旧值
// 确保发给 RAGFlow 的请求包含所有必填项 (Partial Update Support)
if (StringUtils.isBlank(dto.getPermission())) {
dto.setPermission(entity.getPermission());
}
if (StringUtils.isBlank(dto.getChunkMethod())) {
dto.setChunkMethod(entity.getChunkMethod());
}
KnowledgeBaseAdapter adapter = getAdapterByModelId(dto.getRagModelId());
if (adapter != null) {
Map<String, Object> updateParams = new HashMap<>();
// 1. 必填/核心字段
updateParams.put("name", SecurityUser.getUser().getUsername() + "_" + dto.getName());
DatasetDTO.UpdateReq updateReq = ConvertUtils.sourceToTarget(dto, DatasetDTO.UpdateReq.class);
// 2. 修复回退：描述字段
if (dto.getDescription() != null) {
updateParams.put("description", dto.getDescription());
// 1. 必填/核心字段前缀处理
if (StringUtils.isNotBlank(dto.getName())) {
updateReq.setName(SecurityUser.getUser().getUsername() + "_" + dto.getName());
}
// 3. 增强：支持更多元数据同步
if (dto.getPermission() != null)
updateParams.put("permission", dto.getPermission());
if (dto.getAvatar() != null)
updateParams.put("avatar", dto.getAvatar());
if (dto.getChunkMethod() != null)
updateParams.put("chunk_method", dto.getChunkMethod());
if (dto.getEmbeddingModel() != null)
updateParams.put("embedding_model", dto.getEmbeddingModel());
// 4. 解析配置 (JSON String -> Object)
// 2. 解析器配置支持 (如果 DTO 里有字符串形式的配置，尝试转换，但优先建议 DTO 化)
if (StringUtils.isNotBlank(dto.getParserConfig())) {
try {
Map<String, Object> configMap = JsonUtils.parseObject(dto.getParserConfig(), Map.class);
updateParams.put("parser_config", configMap);
DatasetDTO.ParserConfig parserConfig = JsonUtils.parseObject(dto.getParserConfig(),
DatasetDTO.ParserConfig.class);
updateReq.setParserConfig(parserConfig);
} catch (Exception e) {
log.warn("解析 parser_config 失败，跳过同步", e);
}
}
adapter.updateDataset(entity.getDatasetId(), updateParams);
adapter.updateDataset(entity.getDatasetId(), updateReq);
log.info("RAG更新成功: {}", entity.getDatasetId());
}
} catch (Exception e) {
@@ -300,7 +288,6 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
@Override
@Transactional(rollbackFor = Exception.class)
@SuppressWarnings("deprecation")
public void deleteByDatasetId(String datasetId) {
if (StringUtils.isBlank(datasetId)) {
throw new RenException(ErrorCode.PARAMS_GET_ERROR);
@@ -324,7 +311,8 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
try {
KnowledgeBaseAdapter adapter = getAdapterByModelId(entity.getRagModelId());
if (adapter != null) {
adapter.deleteDataset(datasetId);
adapter.deleteDataset(
DatasetDTO.BatchIdReq.builder().ids(Collections.singletonList(datasetId)).build());
}
apiDeleteSuccess = true;
} catch (Exception e) {
@@ -354,19 +342,13 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
@Override
public List<KnowledgeBaseDTO> getByDatasetIdList(List<String> datasetIdList) {
// 1. 入参判空 (Match Old Logic)
if (ToolUtil.isEmpty(datasetIdList)) {
throw new RenException(ErrorCode.PARAMS_GET_ERROR);
if (datasetIdList == null || datasetIdList.isEmpty()) {
return Collections.emptyList();
}
List<KnowledgeBaseEntity> list = knowledgeBaseDao
.selectList(new QueryWrapper<KnowledgeBaseEntity>().in("dataset_id", datasetIdList));
// 2. 结果命中校验 (Match Old Logic)
if (ToolUtil.isEmpty(list)) {
throw new RenException(ErrorCode.Knowledge_Base_RECORD_NOT_EXISTS);
}
// [Production Fix] 批量兼容性查找
QueryWrapper<KnowledgeBaseEntity> queryWrapper = new QueryWrapper<>();
queryWrapper.in("dataset_id", datasetIdList).or().in("id", datasetIdList);
List<KnowledgeBaseEntity> list = knowledgeBaseDao.selectList(queryWrapper);
return ConvertUtils.sourceToTarget(list, KnowledgeBaseDTO.class);
}
@@ -376,7 +358,6 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
}
@Override
@SuppressWarnings("deprecation")
public Map<String, Object> getRAGConfigByDatasetId(String datasetId) {
KnowledgeBaseEntity entity = knowledgeBaseDao
.selectOne(new QueryWrapper<KnowledgeBaseEntity>().eq("dataset_id", datasetId));
@@ -386,6 +367,13 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
return getRAGConfig(entity.getRagModelId());
}
@Override
@Transactional(rollbackFor = Exception.class)
public void updateStatistics(String datasetId, Integer docDelta, Long chunkDelta, Long tokenDelta) {
log.info("递增更新知识库统计: datasetId={}, docs={}, chunks={}, tokens={}", datasetId, docDelta, chunkDelta, tokenDelta);
knowledgeBaseDao.updateStatsAfterChange(datasetId, docDelta, chunkDelta, tokenDelta);
}
@Override
public List<ModelConfigEntity> getRAGModels() {
return modelConfigDao.selectList(new QueryWrapper<ModelConfigEntity>()
main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeFilesServiceImpl.java
+214 -162
View File
@@ -2,13 +2,13 @@ package xiaozhi.modules.knowledge.service.impl;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
@@ -17,19 +17,22 @@ import org.springframework.web.multipart.MultipartFile;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import xiaozhi.common.exception.ErrorCode;
import org.springframework.util.CollectionUtils;
import xiaozhi.common.exception.RenException;
import xiaozhi.modules.knowledge.dto.KnowledgeFilesDTO;
import xiaozhi.modules.knowledge.dto.document.ChunkDTO;
import xiaozhi.modules.knowledge.dto.document.RetrievalDTO;
import xiaozhi.modules.knowledge.dto.document.DocumentDTO;
import xiaozhi.common.page.PageData;
import xiaozhi.common.redis.RedisKeys;
import xiaozhi.common.redis.RedisUtils;
import xiaozhi.common.service.impl.BaseServiceImpl;
import xiaozhi.modules.knowledge.dao.DocumentDao;
import xiaozhi.modules.knowledge.dao.KnowledgeBaseDao;
import xiaozhi.modules.knowledge.dto.KnowledgeFilesDTO;
import xiaozhi.modules.knowledge.entity.DocumentEntity;
import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapter;
import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapterFactory;
@@ -37,17 +40,26 @@ import xiaozhi.modules.knowledge.service.KnowledgeBaseService;
import xiaozhi.modules.knowledge.service.KnowledgeFilesService;
@Service
@AllArgsConstructor
@Slf4j
public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, DocumentEntity>
implements KnowledgeFilesService {
private final KnowledgeBaseService knowledgeBaseService;
private final KnowledgeBaseDao knowledgeBaseDao;
private final DocumentDao documentDao;
private final ObjectMapper objectMapper;
private final RedisUtils redisUtils;
public KnowledgeFilesServiceImpl(KnowledgeBaseService knowledgeBaseService,
DocumentDao documentDao,
ObjectMapper objectMapper,
RedisUtils redisUtils) {
this.knowledgeBaseService = knowledgeBaseService;
this.documentDao = documentDao;
this.objectMapper = objectMapper;
this.redisUtils = redisUtils;
}
@Lazy
@Autowired
private KnowledgeFilesService self;
@@ -90,18 +102,25 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
PageData<KnowledgeFilesDTO> pageData = new PageData<>(dtoList, iPage.getTotal());
// 4. 动态状态同步 (带限流与保护)
// [Bug Fix] P1: 扩大同步白名单，CANCEL/FAIL 也允许低频同步以支持自愈
if (pageData.getList() != null && !pageData.getList().isEmpty()) {
KnowledgeBaseAdapter adapter = null;
for (KnowledgeFilesDTO dto : pageData.getList()) {
// 只有处于“解析中”或“待解析”状态的文档才尝试同步
boolean needSync = "RUNNING".equals(dto.getRun()) || "UNSTART".equals(dto.getRun());
String runStatus = dto.getRun();
// 高优先级同步: RUNNING/UNSTART (5秒冷却)
boolean isActiveSync = "RUNNING".equals(runStatus) || "UNSTART".equals(runStatus);
// 低频自愈同步: CANCEL/FAIL (60秒冷却), 防止错误状态永久锁死
boolean isRecoverySync = "CANCEL".equals(runStatus) || "FAIL".equals(runStatus);
boolean needSync = isActiveSync || isRecoverySync;
if (needSync) {
// Issue 5: 限流保护，5秒内不重复向 RAGFlow 发起同一文档的状态查询
// 限流保护：活跃状态 5 秒冷却，自愈状态 60 秒冷却
long cooldownMs = isActiveSync ? 5000 : 60000;
DocumentEntity localEntity = documentDao.selectOne(new QueryWrapper<DocumentEntity>()
.eq("document_id", dto.getDocumentId()));
if (localEntity != null && localEntity.getLastSyncAt() != null) {
long diff = System.currentTimeMillis() - localEntity.getLastSyncAt().getTime();
if (diff < 5000) {
if (diff < cooldownMs) {
continue;
}
}
@@ -116,7 +135,18 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
break;
}
}
// [关键修复] 记录同步前的 Token 数，用于计算增量
Long oldTokenCount = dto.getTokenCount() != null ? dto.getTokenCount() : 0L;
syncDocumentStatusWithRAG(dto, adapter);
// 计算增量并更新知识库统计 (与定时任务保持一致)
Long newTokenCount = dto.getTokenCount() != null ? dto.getTokenCount() : 0L;
Long tokenDelta = newTokenCount - oldTokenCount;
if (tokenDelta != 0) {
knowledgeBaseService.updateStatistics(datasetId, 0, 0L, tokenDelta);
log.info("懒加载同步: 修正知识库统计, docId={}, tokenDelta={}", dto.getDocumentId(), tokenDelta);
}
}
}
}
@@ -152,7 +182,7 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
if (StringUtils.isNotBlank(entity.getMetaFields())) {
try {
dto.setMetaFields(objectMapper.readValue(entity.getMetaFields(),
new com.fasterxml.jackson.core.type.TypeReference<Map<String, Object>>() {
new TypeReference<Map<String, Object>>() {
}));
} catch (Exception e) {
log.warn("反序列化 MetaFields 失败, entityId: {}, error: {}", entity.getId(), e.getMessage());
@@ -190,11 +220,14 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
String datasetId = dto.getDatasetId();
try {
// 使用精准的 List API 配合 ID 过滤来获取状态
Map<String, Object> queryParams = new HashMap<>();
queryParams.put("id", documentId);
// 使用强类型 ListReq 配合 ID 过滤来获取状态
DocumentDTO.ListReq listReq = DocumentDTO.ListReq.builder()
.id(documentId)
.page(1)
.pageSize(1)
.build();
PageData<KnowledgeFilesDTO> remoteList = adapter.getDocumentList(datasetId, queryParams, 1, 1);
PageData<KnowledgeFilesDTO> remoteList = adapter.getDocumentList(datasetId, listReq);
if (remoteList != null && remoteList.getList() != null && !remoteList.getList().isEmpty()) {
KnowledgeFilesDTO remoteDto = remoteList.getList().get(0);
@@ -202,10 +235,11 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
// 核心状态对齐判别逻辑
boolean statusChanged = remoteStatus != null && !remoteStatus.equals(dto.getStatus());
boolean runChanged = remoteDto.getRun() != null && !remoteDto.getRun().equals(dto.getRun());
boolean isProcessing = "RUNNING".equals(remoteDto.getRun()) || "UNSTART".equals(remoteDto.getRun());
// 只要状态有变，或者文件仍在解析中（实时刷进度），就执行同步
if (statusChanged || isProcessing) {
// 只要状态有变，或者运行状态有变，或者文件仍在解析中（实时刷进度），就执行同步
if (statusChanged || runChanged || isProcessing) {
log.info("影子同步：状态变化={}，解析中={}，文档={}，最新状态={}，进度={}",
statusChanged, isProcessing, documentId, remoteStatus, remoteDto.getProgress());
@@ -250,9 +284,11 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
documentDao.update(null, updateWrapper);
}
} else {
// Issue 6: 远程列表为空，说明 RAGFlow 侧可能已物理删除文档。
// 我们不应直接返回，而应同步将影子库记录标记为逻辑失效或取消，防止形成“僵尸记录”
log.warn("远程同步感知：文档在 RAGFlow 侧已消失，准备清理影子状态, docId={}", documentId);
// Issue 6: 远程列表为空，可能是文档已删除，也可能是适配器调用出了问题
// [Bug Fix] P2: 仅当远程确实返回了合法空列表时才标记 CANCEL
// 同时更新 last_sync_at，配合 P1 冷却机制防止高频误判
log.warn("远程同步感知：RAGFlow 返回空文档列表, docId={}, 当前本地状态={}",
documentId, dto.getRun());
dto.setRun("CANCEL");
dto.setError("文档在远程服务中已被删除");
@@ -260,15 +296,19 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
.set("run", "CANCEL")
.set("error", "文档在远程服务中已被删除")
.set("updated_at", new Date())
.set("last_sync_at", new Date())
.eq("document_id", documentId));
}
} catch (Exception e) {
log.debug("同步文档状态失败，documentId: {}, error: {}", documentId, e.getMessage());
// [Bug Fix] P2: 适配器调用异常时不标记 CANCEL，避免因网络/反序列化问题导致误判
// 仅记录日志，等下次同步周期重试
log.warn("同步文档状态时适配器调用失败(不标记CANCEL), documentId: {}, error: {}",
documentId, e.getMessage());
}
}
@Override
public KnowledgeFilesDTO getByDocumentId(String documentId, String datasetId) {
public DocumentDTO.InfoVO getByDocumentId(String documentId, String datasetId) {
if (StringUtils.isBlank(documentId) || StringUtils.isBlank(datasetId)) {
throw new RenException(ErrorCode.RAG_DATASET_ID_AND_MODEL_ID_NOT_NULL);
}
@@ -287,11 +327,11 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(adapterType, ragConfig);
// 使用适配器获取文档详情
KnowledgeFilesDTO dto = adapter.getDocumentById(datasetId, documentId);
DocumentDTO.InfoVO info = adapter.getDocumentById(datasetId, documentId);
if (dto != null) {
if (info != null) {
log.info("获取文档详情成功，documentId: {}", documentId);
return dto;
return info;
} else {
throw new RenException(ErrorCode.Knowledge_Base_RECORD_NOT_EXISTS);
}
@@ -330,9 +370,30 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
Map<String, Object> ragConfig = knowledgeBaseService.getRAGConfigByDatasetId(datasetId);
KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(extractAdapterType(ragConfig), ragConfig);
// 构造强类型请求 DTO
DocumentDTO.UploadReq uploadReq = DocumentDTO.UploadReq.builder()
.datasetId(datasetId)
.file(file)
.name(fileName)
.metaFields(metaFields)
.build();
// 转换分块方法 (String -> Enum)
if (StringUtils.isNotBlank(chunkMethod)) {
try {
uploadReq.setChunkMethod(DocumentDTO.InfoVO.ChunkMethod.valueOf(chunkMethod.toUpperCase()));
} catch (Exception e) {
log.warn("无效的分块方法: {}, 将使用后台默认配置", chunkMethod);
}
}
// 转换解析配置 (Map -> DTO)
if (parserConfig != null && !parserConfig.isEmpty()) {
uploadReq.setParserConfig(objectMapper.convertValue(parserConfig, DocumentDTO.InfoVO.ParserConfig.class));
}
// 执行远程上传 (耗时 IO，在事务之外)
KnowledgeFilesDTO result = adapter.uploadDocument(datasetId, file, fileName,
metaFields, chunkMethod, parserConfig);
KnowledgeFilesDTO result = adapter.uploadDocument(uploadReq);
if (result == null || StringUtils.isBlank(result.getDocumentId())) {
throw new RenException(ErrorCode.RAG_API_ERROR, "远程上传成功但未返回有效 DocumentID");
@@ -396,84 +457,87 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
documentDao.insert(entity);
// Issue 4: 同步递增数据集文档总数统计，保持父子表一致
knowledgeBaseDao.updateStatsAfterUpload(datasetId);
knowledgeBaseService.updateStatistics(datasetId, 1, 0L, 0L);
log.info("已同步递增数据集统计: datasetId={}", datasetId);
}
@Override
@Transactional(propagation = Propagation.NOT_SUPPORTED)
public void deleteByDocumentId(String documentId, String datasetId) {
if (StringUtils.isBlank(documentId) || StringUtils.isBlank(datasetId)) {
public void deleteDocuments(String datasetId, DocumentDTO.BatchIdReq req) {
if (StringUtils.isBlank(datasetId) || req == null || req.getIds() == null || req.getIds().isEmpty()) {
throw new RenException(ErrorCode.RAG_DATASET_ID_AND_MODEL_ID_NOT_NULL);
}
log.info("=== 开始根据documentId删除文档 (地狱级加固版) ===");
List<String> documentIds = req.getIds();
log.info("=== 开始批量删除文档: datasetId={}, count={} ===", datasetId, documentIds.size());
// 1. 权限与物理归属权预审 (防止 ID 劫持漏洞)
DocumentEntity entity = documentDao.selectOne(
// 1. 批量权限与状态预审
List<DocumentEntity> entities = documentDao.selectList(
new QueryWrapper<DocumentEntity>()
.eq("dataset_id", datasetId)
.eq("document_id", documentId));
.in("document_id", documentIds));
if (entity == null) {
log.warn("尝试删除不存在或归属权异常的文档: docId={}, datasetId={}", documentId, datasetId);
if (entities.size() != documentIds.size()) {
log.warn("部分文档不存在或归属权异常: 预期={}, 实际={}", documentIds.size(), entities.size());
throw new RenException(ErrorCode.NO_PERMISSION);
}
// 2. 状态校验 (拦截解析中文件的删除，防止 RAG 侧产生僵尸数据 - Issue 4)
// 修正逻辑：status 是 String 类型，必须使用字符串比较或状态码转换判断
if (entity.getStatus() != null && "1".equals(entity.getStatus())) {
log.warn("拦截解析中文件的删除请求: docId={}, status=解析中", documentId);
throw new RenException(ErrorCode.RAG_DOCUMENT_PARSING_DELETE_ERROR);
long totalChunkDelta = 0;
long totalTokenDelta = 0;
for (DocumentEntity entity : entities) {
// 拦截正在解析中的文档的删除请求
// [Bug Fix] 判断解析中应该用 run 字段(RUNNING), 而非 status 字段
// status="1" 是"启用/正常"的意思, 不是"解析中"
if ("RUNNING".equals(entity.getRun())) {
log.warn("拦截解析中文件的删除请求: docId={}", entity.getDocumentId());
throw new RenException(ErrorCode.RAG_DOCUMENT_PARSING_DELETE_ERROR);
}
totalChunkDelta += entity.getChunkCount() != null ? entity.getChunkCount() : 0L;
totalTokenDelta += entity.getTokenCount() != null ? entity.getTokenCount() : 0L;
}
// 记录统计信息偏移量，用于后续影子表同步
Long chunkDelta = entity.getChunkCount() != null ? entity.getChunkCount() : 0L;
Long tokenDelta = entity.getTokenCount() != null ? entity.getTokenCount() : 0L;
// 3. 获取适配器 (非事务性)
// 2. 获取适配器 (非事务性)
Map<String, Object> ragConfig = knowledgeBaseService.getRAGConfigByDatasetId(datasetId);
KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(extractAdapterType(ragConfig), ragConfig);
// 4. 执行远程删除 (耗时 IO，已被 NOT_SUPPORTED 物理挂起事务)
// 3. 执行远程删除
try {
adapter.deleteDocument(datasetId, documentId);
log.info("远程请求删除成功: documentId={}", documentId);
adapter.deleteDocument(datasetId, req);
log.info("远程批量删除请求成功");
} catch (Exception e) {
log.warn("远程删除请求失败 (可能文件已不存在): {}", e.getMessage());
log.warn("远程删除请求部分或全部失败: {}", e.getMessage());
}
// 5. 原子化清理本地影子记录并同步统计数据 (通过 self 调用激活 @Transactional)
log.info("5. 同步清理本地影子记录并更新统计: documentId={}", documentId);
self.deleteDocumentShadow(documentId, datasetId, chunkDelta, tokenDelta);
// 4. 原子化清理本地影子记录并同步统计数据
self.deleteDocumentShadows(documentIds, datasetId, totalChunkDelta, totalTokenDelta);
// 6. 清理缓存 (Issue 7: 缓存孤岛修复)
// 5. 清理缓存
try {
String cacheKey = RedisKeys.getKnowledgeBaseCacheKey(datasetId);
redisUtils.delete(cacheKey);
log.info("已精准驱逐数据集缓存: {}", cacheKey);
log.info("已驱逐数据集缓存: {}", cacheKey);
} catch (Exception e) {
log.warn("驱逐 Redis 缓存失败 (非核心链路，忽略): {}", e.getMessage());
log.warn("驱逐 Redis 缓存失败: {}", e.getMessage());
}
log.info("=== 文档物理清理与统计同步成功 ===");
log.info("=== 批量文档清理完成 ===");
}
/**
* 原子化删除影子记录并同步父表统计，确保 Local-First 全链路一致性
* 批量原子化删除影子记录并同步父表统计
*/
@Override
@Transactional(rollbackFor = Exception.class)
public void deleteDocumentShadow(String documentId, String datasetId, Long chunkDelta, Long tokenDelta) {
public void deleteDocumentShadows(List<String> documentIds, String datasetId, Long chunkDelta, Long tokenDelta) {
// 1. 物理删除记录
int deleted = documentDao.delete(
new QueryWrapper<DocumentEntity>()
.eq("dataset_id", datasetId)
.eq("document_id", documentId));
.in("document_id", documentIds));
if (deleted > 0) {
// 2. 同步更新数据集统计信息 (原子操作，防止并发漂移)
knowledgeBaseDao.updateStatsAfterDelete(datasetId, chunkDelta, tokenDelta);
// 2. 同步更新数据集统计信息
knowledgeBaseService.updateStatistics(datasetId, -documentIds.size(), -chunkDelta, -tokenDelta);
log.info("已同步扣减数据集统计: datasetId={}, chunks={}, tokens={}", datasetId, chunkDelta, tokenDelta);
}
}
@@ -603,37 +667,21 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
}
@Override
public xiaozhi.modules.knowledge.dto.document.ChunkDTO.ListVO listChunks(String datasetId, String documentId,
String keywords,
Integer page, Integer pageSize, String chunkId) {
public ChunkDTO.ListVO listChunks(String datasetId, String documentId, ChunkDTO.ListReq req) {
if (StringUtils.isBlank(datasetId) || StringUtils.isBlank(documentId)) {
throw new RenException(ErrorCode.RAG_DATASET_ID_AND_MODEL_ID_NOT_NULL);
}
log.info("=== 开始列出切片 ===");
log.info("datasetId: {}, documentId: {}, keywords: {}, page: {}, pageSize: {}, chunkId: {}",
datasetId, documentId, keywords, page, pageSize, chunkId);
log.info("=== 开始列出切片: datasetId={}, documentId={}, req={} ===", datasetId, documentId, req);
try {
// 获取RAG配置
Map<String, Object> ragConfig = knowledgeBaseService.getRAGConfigByDatasetId(datasetId);
KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(extractAdapterType(ragConfig),
ragConfig);
// 提取适配器类型
String adapterType = extractAdapterType(ragConfig);
// 获取知识库适配器
KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(adapterType, ragConfig);
log.debug("查询参数: documentId: {}, keywords: {}, page: {}, pageSize: {}, chunkId: {}",
documentId, keywords, page, pageSize, chunkId);
// 调用适配器列出切片 (直接返回强类型 DTO)
xiaozhi.modules.knowledge.dto.document.ChunkDTO.ListVO result = adapter.listChunks(datasetId, documentId,
keywords, page, pageSize, chunkId);
log.info("切片列表获取成功，datasetId: {}, documentId: {}", datasetId, documentId);
ChunkDTO.ListVO result = adapter.listChunks(datasetId, documentId, req);
log.info("切片列表获取成功: datasetId={}, total={}", datasetId, result.getTotal());
return result;
} catch (Exception e) {
log.error("列出切片失败: {}", e.getMessage(), e);
String errorMessage = e.getMessage() != null ? e.getMessage() : "null";
@@ -647,88 +695,21 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
}
@Override
public xiaozhi.modules.knowledge.dto.document.RetrievalDTO.ResultVO retrievalTest(String question,
List<String> datasetIds, List<String> documentIds,
Integer page, Integer pageSize, Float similarityThreshold,
Float vectorSimilarityWeight, Integer topK, String rerankId,
Boolean keyword, Boolean highlight, List<String> crossLanguages,
Map<String, Object> metadataCondition) {
public RetrievalDTO.ResultVO retrievalTest(RetrievalDTO.TestReq req) {
if (CollectionUtils.isEmpty(req.getDatasetIds())) {
throw new RenException("未指定召回测试的知识库");
}
log.info("=== 开始召回测试 ===");
log.info("问题: {}, 数据集ID: {}, 文档ID: {}, 页码: {}, 每页数量: {}",
question, datasetIds, documentIds, page, pageSize);
log.info("=== 开始召回测试: req={} ===", req);
try {
// 获取RAG配置
Map<String, Object> ragConfig = knowledgeBaseService.getRAGConfigByDatasetId(datasetIds.get(0));
Map<String, Object> ragConfig = knowledgeBaseService.getRAGConfigByDatasetId(req.getDatasetIds().get(0));
KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(extractAdapterType(ragConfig),
ragConfig);
// 提取适配器类型
String adapterType = extractAdapterType(ragConfig);
// 获取知识库适配器
KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(adapterType, ragConfig);
// 构建检索参数
Map<String, Object> retrievalParams = new HashMap<>();
retrievalParams.put("question", question);
if (datasetIds != null && !datasetIds.isEmpty()) {
retrievalParams.put("datasetIds", datasetIds);
}
if (documentIds != null && !documentIds.isEmpty()) {
retrievalParams.put("documentIds", documentIds);
}
if (page != null && page > 0) {
retrievalParams.put("page", page);
}
if (pageSize != null && pageSize > 0) {
retrievalParams.put("pageSize", pageSize);
}
if (similarityThreshold != null) {
retrievalParams.put("similarityThreshold", similarityThreshold);
}
if (vectorSimilarityWeight != null) {
retrievalParams.put("vectorSimilarityWeight", vectorSimilarityWeight);
}
if (topK != null && topK > 0) {
retrievalParams.put("topK", topK);
}
if (rerankId != null) {
retrievalParams.put("rerankId", rerankId);
}
if (keyword != null) {
retrievalParams.put("keyword", keyword);
}
if (highlight != null) {
retrievalParams.put("highlight", highlight);
}
if (crossLanguages != null && !crossLanguages.isEmpty()) {
retrievalParams.put("crossLanguages", crossLanguages);
}
if (metadataCondition != null) {
retrievalParams.put("metadataCondition", metadataCondition);
}
log.debug("检索参数: {}", retrievalParams);
// 调用适配器进行检索测试 (直接返回结果 DTO)
xiaozhi.modules.knowledge.dto.document.RetrievalDTO.ResultVO result = adapter.retrievalTest(question,
datasetIds, documentIds, retrievalParams);
log.info("召回测试成功，返回数: {}", result != null ? result.getTotal() : 0);
RetrievalDTO.ResultVO result = adapter.retrievalTest(req);
log.info("召回测试成功: total={}", result != null ? result.getTotal() : 0);
return result;
} catch (Exception e) {
log.error("召回测试失败: {}", e.getMessage(), e);
String errorMessage = e.getMessage() != null ? e.getMessage() : "null";
@@ -740,4 +721,75 @@ public class KnowledgeFilesServiceImpl extends BaseServiceImpl<DocumentDao, Docu
log.info("=== 召回测试操作结束 ===");
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public void deleteDocumentsByDatasetId(String datasetId) {
log.info("级联清理数据集文档: datasetId={}", datasetId);
List<DocumentEntity> list = documentDao
.selectList(new QueryWrapper<DocumentEntity>().eq("dataset_id", datasetId));
if (list == null || list.isEmpty())
return;
List<String> docIds = list.stream().map(DocumentEntity::getDocumentId).toList();
// 封包调用现有删除逻辑 (含 RAG 物理删除)
DocumentDTO.BatchIdReq req = DocumentDTO.BatchIdReq.builder().ids(docIds).build();
this.deleteDocuments(datasetId, req);
}
@Override
public void syncRunningDocuments() {
// 1. 查询所有 RUNNING 状态的文档
List<DocumentEntity> runningDocs = documentDao.selectList(
new QueryWrapper<DocumentEntity>()
.eq("run", "RUNNING")
.eq("status", "1") // 仅同步启用的文档
);
if (runningDocs == null || runningDocs.isEmpty()) {
return;
}
log.info("定时任务: 发现 {} 个文档正在解析中，开始同步...", runningDocs.size());
// 2. 按 DatasetID 分组，复用 Adapter
Map<String, List<DocumentEntity>> groupedDocs = runningDocs.stream()
.collect(java.util.stream.Collectors.groupingBy(DocumentEntity::getDatasetId));
groupedDocs.forEach((datasetId, docs) -> {
KnowledgeBaseAdapter adapter = null;
try {
// 初始化 Adapter (每个数据集只初始化一次)
Map<String, Object> ragConfig = knowledgeBaseService.getRAGConfigByDatasetId(datasetId);
adapter = KnowledgeBaseAdapterFactory.getAdapter(extractAdapterType(ragConfig), ragConfig);
} catch (Exception e) {
log.warn("无法为数据集 {} 初始化适配器，跳过同步: {}", datasetId, e.getMessage());
return;
}
for (DocumentEntity doc : docs) {
try {
// 构造临时 DTO 传给同步方法
KnowledgeFilesDTO dto = convertEntityToDTO(doc);
// 记录同步前的 Token 数
Long oldTokenCount = dto.getTokenCount() != null ? dto.getTokenCount() : 0L;
syncDocumentStatusWithRAG(dto, adapter);
// 3. [关键修复] 计算增量并更新知识库统计
Long newTokenCount = dto.getTokenCount() != null ? dto.getTokenCount() : 0L;
Long tokenDelta = newTokenCount - oldTokenCount;
// 仅当状态变为 SUCCESS 且 Token 数有变化时更新统计
if (tokenDelta != 0) {
knowledgeBaseService.updateStatistics(datasetId, 0, 0L, tokenDelta);
log.info("定时任务: 同步修正知识库统计, docId={}, tokenDelta={}", dto.getDocumentId(), tokenDelta);
}
} catch (Exception e) {
log.error("同步文档 {} 失败: {}", doc.getDocumentId(), e.getMessage());
}
}
});
}
}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeManagerServiceImpl.java
+47
View File
@@ -0,0 +1,47 @@
package xiaozhi.modules.knowledge.service.impl;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import xiaozhi.modules.knowledge.service.KnowledgeBaseService;
import xiaozhi.modules.knowledge.service.KnowledgeFilesService;
import xiaozhi.modules.knowledge.service.KnowledgeManagerService;
import java.util.List;
@Service
@Slf4j
@RequiredArgsConstructor
public class KnowledgeManagerServiceImpl implements KnowledgeManagerService {
private final KnowledgeBaseService knowledgeBaseService;
private final KnowledgeFilesService knowledgeFilesService;
@Override
@Transactional(rollbackFor = Exception.class)
public void deleteDatasetWithFiles(String datasetId) {
log.info("=== 级联删除开始: datasetId={} ===", datasetId);
// 1. 先调用文件服务，清理该数据集下的所有文档记录 (含 RAGFlow 端)
log.info("Step 1: 清理关联文档...");
knowledgeFilesService.deleteDocumentsByDatasetId(datasetId);
// 2. 再调用知识库服务，彻底注销数据集 (含 RAGFlow 端)
log.info("Step 2: 删除数据集主体...");
knowledgeBaseService.deleteByDatasetId(datasetId);
log.info("=== 级联删除成功: datasetId={} ===", datasetId);
}
@Override
@Transactional(rollbackFor = Exception.class)
public void batchDeleteDatasetsWithFiles(List<String> datasetIds) {
if (datasetIds == null || datasetIds.isEmpty())
return;
log.info("=== 批量级联删除开始: count={} ===", datasetIds.size());
for (String id : datasetIds) {
deleteDatasetWithFiles(id);
}
}
}
main/manager-api/src/main/java/xiaozhi/modules/knowledge/task/DocumentStatusSyncTask.java
+39
View File
@@ -0,0 +1,39 @@
package xiaozhi.modules.knowledge.task;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import xiaozhi.modules.knowledge.service.KnowledgeFilesService;
/**
* 知识库文档状态同步定时任务
*
* 作用：
* 1. 自动扫描处于 "RUNNING" (解析中) 状态的文档
* 2. 调用 RAGFlow 接口获取最新状态
* 3. 状态翻转 (RUNNING -> SUCCESS/FAIL) 时，同步更新数据库
* 4. [关键] 解析成功时，补偿更新知识库的统计信息 (TokenCount)
*/
@Component
@AllArgsConstructor
@Slf4j
public class DocumentStatusSyncTask {
private final KnowledgeFilesService knowledgeFilesService;
/**
* 每 30 秒执行一次同步
* 采用 fixedDelay，确保上一次执行完 30 秒后才开始下一次，防止积压
*/
@Scheduled(fixedDelay = 30000)
public void syncRunningDocuments() {
try {
// log.debug("开始执行文档状态同步任务...");
knowledgeFilesService.syncRunningDocuments();
} catch (Exception e) {
log.error("文档状态同步任务异常", e);
}
}
}
main/manager-api/src/main/java/xiaozhi/modules/sys/service/impl/SysUserServiceImpl.java
+5 -5
View File
@@ -56,7 +56,7 @@ public class SysUserServiceImpl extends BaseServiceImpl<SysUserDao, SysUserEntit
if (users == null || users.isEmpty()) {
return null;
}
SysUserEntity entity = users.getFirst();
SysUserEntity entity = users.get(0);
return ConvertUtils.sourceToTarget(entity, SysUserDTO.class);
}
@@ -197,7 +197,7 @@ public class SysUserServiceImpl extends BaseServiceImpl<SysUserDao, SysUserEntit
*/
private String generatePassword() {
StringBuilder password = new StringBuilder();
// 确保包含至少一个数字
password.append("0123456789".charAt(random.nextInt(10)));
// 确保包含至少一个小写字母
@@ -206,12 +206,12 @@ public class SysUserServiceImpl extends BaseServiceImpl<SysUserDao, SysUserEntit
password.append("ABCDEFGHIJKLMNOPQRSTUVWXYZ".charAt(random.nextInt(26)));
// 确保包含至少一个特殊符号
password.append("!@#$%^&*()".charAt(random.nextInt(10)));
// 生成剩余的8个字符
for (int i = 4; i < 12; i++) {
password.append(CHARACTERS.charAt(random.nextInt(CHARACTERS.length())));
}
// 打乱密码中字符的顺序
char[] passwordChars = password.toString().toCharArray();
for (int i = 0; i < passwordChars.length; i++) {
@@ -220,7 +220,7 @@ public class SysUserServiceImpl extends BaseServiceImpl<SysUserDao, SysUserEntit
passwordChars[i] = passwordChars[randomIndex];
passwordChars[randomIndex] = temp;
}
return new String(passwordChars);
}
main/manager-api/src/main/resources/i18n/messages.properties
+2 -1
View File
@@ -201,4 +201,5 @@
10192=\u9002\u914D\u5668\u7C7B\u578B\u672A\u627E\u5230
10193=\u8BBE\u5907ID\u4E0D\u80FD\u4E3A\u7A7A
10194=\u8BBE\u5907\u4E0D\u5B58\u5728\u6216\u4E0D\u5728\u7EBF
10195=OTA\u4E0A\u4F20\u6B21\u6570\u8D85\u8FC7\u9650\u5236
10195=OTA\u4E0A\u4F20\u6B21\u6570\u8D85\u8FC7\u9650\u5236
10196=\u89E3\u6790\u4E2D\u6587\u4EF6\u6682\u4E0D\u652F\u6301\u6B64\u64CD\u4F5C
main/manager-api/src/main/resources/i18n/messages_de_DE.properties
+3 -2
View File
@@ -1,4 +1,4 @@
#Deutsch
#Deutsch
500=Server interne Ausnahme
401=Nicht autorisiert
403=Zugriff verweigert, keine Berechtigungen
@@ -201,4 +201,5 @@
10192=Adaptertyp nicht gefunden
10193=Adapter-ID darf nicht leer sein
10194=Adapter nicht gefunden oder nicht online
10195=OTA-Upload-Anzahl \u00FCberschreitet das Limit
10195=OTA\u4E0A\u4F20\u6B21\u6570\u8D85\u904E\u9650\u5236
10196=Dateianalyse läuft, dieser Vorgang wird nicht unterstützt
main/manager-api/src/main/resources/i18n/messages_en_US.properties
+3 -2
View File
@@ -1,4 +1,4 @@
#English
#English
500=Server internal exception
401=Unauthorized
403=Access denied, no permissions
@@ -201,4 +201,5 @@
10192=Adapter type not found
10193=Device ID cannot be empty
10194=Device not found or not online
10195=OTA upload times exceed the limit
10195=OTA\u4E0A\u4F20\u6B21\u6570\u8D85\u904E\u9650\u5236
10196=Parsing in progress, this operation is not supported
main/manager-api/src/main/resources/i18n/messages_vi_VN.properties
+3 -2
View File
@@ -1,4 +1,4 @@
#Ti\u1EBFng Vi\u1EC7t
#Ti\u1EBFng Vi\u1EC7t
500=Ngo\u1EA1i l\u1EC7 n\u1ED9i b\u1ED9 m\u00E1y ch\u1EE7
401=Kh\u00F4ng \u0111\u01B0\u1EE3c \u1EE7y quy\u1EC1n
403=Truy c\u1EADp b\u1ECB t\u1EEB ch\u1ED1i, kh\u00F4ng c\u00F3 quy\u1EC1n
@@ -201,4 +201,5 @@
10192=Kh\u00F4ng t\u00ECm th\u1EA5y lo\u1EA1i b\u1ED9 chuy\u1EC3n \u0111\u1ED5i
10193=ID thi\u1EBFt b\u1ECB kh\u00F4ng th\u1EC3 tr\u1ED1ng
10194=Kh\u00F4ng t\u00ECm th\u1EA5y thi\u1EBFt b\u1ECB ho\u1EB7c thi\u1EBFt b\u1ECB kh\u00F4ng tr\u1EF1c tuy\u1EBFn
10195=S\u1ED1 l\u1EA7n t\u1EA3i l\u00EAn OTA v\u01B0\u1EE3t qu\u00E1 gi\u1EDBi h\u1EA1n
10195=OTA\u4E0A\u4F20\u6B21\u6570\u8D85\u904E\u9650\u5236
10196=Tệp đang được phân tích, thao tác này không được hỗ trợ
main/manager-api/src/main/resources/i18n/messages_zh_CN.properties
+2 -1
View File
@@ -201,4 +201,5 @@
10192=\u9002\u914D\u5668\u7C7B\u578B\u672A\u627E\u5230
10193=\u8BBE\u5907ID\u4E0D\u80FD\u4E3A\u7A7A
10194=\u8BBE\u5907\u4E0D\u5B58\u5728\u6216\u4E0D\u5728\u7EBF
10195=OTA\u4E0A\u4F20\u6B21\u6570\u8D85\u8FC7\u9650\u5236
10195=OTA\u4E0A\u4F20\u6B21\u6570\u8D85\u8FC7\u9650\u5236
10196=\u89E3\u6790\u4E2D\u6587\u4EF6\u6682\u4E0D\u652F\u6301\u6B64\u64CD\u4F5C
main/manager-api/src/main/resources/i18n/messages_zh_TW.properties
+3 -2
View File
@@ -1,4 +1,4 @@
#\u7E41\u9AD4\u4E2D\u6587
#\u7E41\u9AD4\u4E2D\u6587
500=\u670D\u52D9\u5668\u5167\u90E8\u7570\u5E38
401=\u672A\u6388\u6B0A
403=\u62D2\u7D55\u8A2A\u554F\uFF0C\u6C92\u6709\u6743\u9650
@@ -201,4 +201,5 @@
10192=\u9002\u914D\u5668\u985E\u578B\u672A\u627E\u5230
10193=\u8A2D\u5099ID\u4E0D\u80FD\u4E3A\u7A7A
10194=\u8A2D\u5099\u4E0D\u5B58\u5728\u6216\u672A\u5728\u7DDA
10195=OTA\u4E0A\u4F20\u6B21\u6578\u8D85\u904E\u9650\u5236
10195=OTA\u4E0A\u4F20\u6B21\u6570\u8D85\u904E\u9650\u5236
10196=\u89E3\u6790\u4E2D\u6587\u4EF6\u6682\u4E0D\u652F\u6301\u6B64\u64CD\u4F5C
main/manager-api/src/main/resources/mapper/knowledge/KnowledgeBaseDao.xml
+10
View File
@@ -23,4 +23,14 @@
WHERE plugin_id = #{knowledgeBaseId}
</delete>
<!-- 通用维度原子更新知识库统计信息 -->
<update id="updateStatsAfterChange">
UPDATE ai_rag_dataset
SET document_count = document_count + #{docDelta},
chunk_count = chunk_count + #{chunkDelta},
token_num = token_num + #{tokenDelta},
updated_at = NOW()
WHERE dataset_id = #{datasetId}
</update>
</mapper>