wanghanlin
/
RAG
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

fix(document): 修复批量上传向量化只成功小文档问题并加固处理链路 

- 移除逐 chunk 串行 AI 关键词提取(MyKeywordEnricher)：产出 excerpt_keywords 全库无检索消费点，却将外部调用放大 chunkCount 倍且与在线对话共用模型抢限流
- 向量化改按批入库(默认 50 块/批, knowledge.vector.batch-size)逐批 try-catch 隔离：失败批只记录缺失区间继续，已入库块保留，error_message 聚合"已入库 x/y 块+缺失区间+原因"，chunk_count 改为实际入库块数
- 新增文档级失败自动整体重试(≤2 次, 仅 timeout/429/5xx 等瞬时错误)：重试前清残留向量防重复；处理期间删除文档即终止写入避免孤儿向量；外层兜底保证状态不悬挂 PROCESSING
- Embedding 客户端(DashScope/OpenAI 兼容/豆包多模态)补显式 connect 10s/read 60s 超时，避免慢响应无限挂起占死 documentExecutor 线程
- 文档同步 CLAUDE.md/README.md，新增 knowledge.vector.batch-size 配置说明
Spring-AI-1.1.2
wanghanlin 1 week ago
parent
459cabb1bc
commit
d0cd06f9fc
8 changed files with 410 additions and 142 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      CLAUDE.md
  2. 9
      README.md
  3. 30
      src/main/java/com/wok/supportbot/config/EmbeddingModelFactory.java
  4. 16
      src/main/java/com/wok/supportbot/config/VolcengineMultimodalEmbeddingModel.java
  5. 30
      src/main/java/com/wok/supportbot/document/transform/MyKeywordEnricher.java
  6. 452
      src/main/java/com/wok/supportbot/service/DocumentProcessingService.java
  7. 8
      src/main/java/com/wok/supportbot/service/DocumentService.java
  8. 3
      src/main/resources/application.yml

4
CLAUDE.md
View File

@ -67,7 +67,9 @@ AI 智能客服系统，基于 Spring AI Alibaba + 通义千问 + PGVector，支
- **Open API**: `OpenApiController` 已接入 `ChatPipeline`，补齐角色/RAG/FAQ/MCP/分类隔离能力
### 文档处理管道
`DocumentService.uploadDocument()` 统一流程：文档提取 → `MyTokenTextSplitter` 分块 → `MyKeywordEnricher` AI 关键词提取 → `pgVectorVectorStore.add()` 向量化存储。每个分块的 metadata 中注入 `documentId`、`chunkIndex`、`sourceName`、`title` 以关联 `knowledge_document` 表。
`DocumentService.uploadDocument()` 统一流程：文档提取 → `MyTokenTextSplitter` 分块 → 为每块写 metadata → 按批向量化（默认 50 块/批，配置项 `knowledge.vector.batch-size`）`pgVectorVectorStore.add(batch)` 入库。每个分块的 metadata 注入 `documentId`、`chunkIndex`、`sourceName`、`title`、`categoryId`、`enabled` 关联 `knowledge_document` 表。
**向量化加固**（`DocumentProcessingService`）：逐批 try-catch 隔离，失败批只记录缺失区间后继续，已入库块保留；`chunk_count` 记实际入库块数，`error_message` 聚合"已入库 x/y 块 + 缺失区间 + 原因"；文档级失败自动整体重试至多 2 次（仅瞬时/限流/超时类错误，4xx 不空转），重试前先清残留向量再重建。**无逐块 AI 关键词提取环节**（`MyKeywordEnricher` 已移除，其产出 `excerpt_keywords` 全库无检索消费点）。
## 关键配置

9
README.md
View File

@ -188,8 +188,7 @@ src/main/java/com/wok/supportbot/
│ │ ├── JsonDocumentLoader.java # JSON 解析（3种模式）
│ │ └── SimpleStringDocumentReader.java # 纯文本读取
│ └── transform/ # 文档转换器
│ ├── MyTokenTextSplitter.java # Token 分块器
│ └── MyKeywordEnricher.java # AI 关键词提取
│ └── MyTokenTextSplitter.java # Token 分块器
├── entity/ # 数据实体类
│ ├── ChatMessage.java # 聊天消息实体
│ ├── KnowledgeDocument.java # 知识文档实体
@ -234,11 +233,11 @@ src/main/resources/
↓
[Token 分块] MyTokenTextSplitter (200 token / 100 overlap)
↓
[关键词提取] MyKeywordEnricher (AI 提取 Top-5 关键词)
[元数据标注] metadata.documentId / chunkIndex / sourceName / title / categoryId / enabled
↓
[向量化存储] DashScope text-embedding-v2 → PGVector
[分批向量化] 默认 50 块/批循环 add → PGVector（逐批隔离，失败批记录区间后继续，整体失败自动重试 ≤2 次）
↓
[元数据关联] metadata.documentId / chunkIndex / sourceName / title
[状态更新] READY / FAILED（已入库块保留，chunk_count = 实际入库块数，error_message 聚合缺失区间）
```
## 🖥️ 前端管理页面

30
src/main/java/com/wok/supportbot/config/EmbeddingModelFactory.java
View File

@ -13,10 +13,14 @@ import org.springframework.ai.openai.OpenAiEmbeddingModel;
import org.springframework.ai.openai.OpenAiEmbeddingOptions;
import org.springframework.ai.openai.api.OpenAiApi;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.client.JdkClientHttpRequestFactory;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestClient;
import java.net.http.HttpClient;
import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@ -110,7 +114,8 @@ public class EmbeddingModelFactory {
if ("dashscope".equalsIgnoreCase(config.getProvider())) {
log.info("创建 DashScope EmbeddingModel: model={}, baseUrl={}", config.getModelName(), baseUrl);
DashScopeApi.Builder apiBuilder = DashScopeApi.builder()
.apiKey(config.getApiKey());
.apiKey(config.getApiKey())
.restClientBuilder(timeoutRestClientBuilder());
// 团队版/私有化部署使用自定义 baseUrl
if (baseUrl != null && !baseUrl.isBlank()) {
apiBuilder.baseUrl(baseUrl);
@ -147,6 +152,7 @@ public class EmbeddingModelFactory {
.apiKey(config.getApiKey())
.baseUrl(baseUrl)
.embeddingsPath(embeddingsPath)
.restClientBuilder(timeoutRestClientBuilder())
.build();
OpenAiEmbeddingOptions options = OpenAiEmbeddingOptions.builder()
@ -223,6 +229,28 @@ public class EmbeddingModelFactory {
log.info("EmbeddingModel 缓存已清除");
}
/** 向量化模型连接超时：connect 10s 与聊天路径（ChatModelFactory）保持一致 */
private static final Duration EMBED_CONNECT_TIMEOUT = Duration.ofSeconds(10);
/**
* 向量化模型读取超时：embedding 为非流式单次请求，60s 足以容纳大体积批请求，
* 避免慢响应无限挂起占死 documentExecutor 线程
*/
private static final Duration EMBED_READ_TIMEOUT = Duration.ofSeconds(60);
/**
* 构建带显式 connect/read 超时的 RestClient（DashScope / OpenAI 兼容 embedding 客户端使用）。
* 否则第三方默认可能无限等待，长文档批量向量化时一次慢响应即拖垮整个任务
*/
private static RestClient.Builder timeoutRestClientBuilder() {
HttpClient httpClient = HttpClient.newBuilder()
.connectTimeout(EMBED_CONNECT_TIMEOUT)
.build();
JdkClientHttpRequestFactory requestFactory = new JdkClientHttpRequestFactory(httpClient);
requestFactory.setReadTimeout(EMBED_READ_TIMEOUT);
return RestClient.builder().requestFactory(requestFactory);
}
// ==================== F1: 连接测试 ====================
/**

16
src/main/java/com/wok/supportbot/config/VolcengineMultimodalEmbeddingModel.java
View File

@ -12,9 +12,12 @@ import org.springframework.ai.embedding.EmbeddingResponseMetadata;
import org.springframework.ai.chat.metadata.DefaultUsage;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.client.JdkClientHttpRequestFactory;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.web.client.RestClient;
import java.net.http.HttpClient;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@ -39,6 +42,12 @@ public class VolcengineMultimodalEmbeddingModel implements EmbeddingModel {
private static final String EMBEDDINGS_PATH = "/embeddings/multimodal";
/** 连接超时：与聊天路径（ChatModelFactory）保持一致 */
private static final Duration CONNECT_TIMEOUT = Duration.ofSeconds(10);
/** 读取超时：单条文本一次请求，60s 足以容纳慢冷启动/大体积文本，避免无限挂起占用文档处理线程 */
private static final Duration READ_TIMEOUT = Duration.ofSeconds(60);
private final String apiKey;
private final String baseUrl;
private final String modelName;
@ -54,8 +63,15 @@ public class VolcengineMultimodalEmbeddingModel implements EmbeddingModel {
this.modelName = modelName;
this.dimensions = dimensions;
this.retryTemplate = retryTemplate;
// 显式设置 connect/read 超时：embedding 为非流式单次请求，避免慢响应无限挂起占用文档处理线程
HttpClient httpClient = HttpClient.newBuilder()
.connectTimeout(CONNECT_TIMEOUT)
.build();
JdkClientHttpRequestFactory requestFactory = new JdkClientHttpRequestFactory(httpClient);
requestFactory.setReadTimeout(READ_TIMEOUT);
this.restClient = RestClient.builder()
.baseUrl(baseUrl)
.requestFactory(requestFactory)
.defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + apiKey)
.defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
.build();

30
src/main/java/com/wok/supportbot/document/transform/MyKeywordEnricher.java
View File

@ -1,30 +0,0 @@
package com.wok.supportbot.document.transform;
import com.wok.supportbot.config.ChatModelFactory;
import jakarta.annotation.Resource;
import org.springframework.ai.document.Document;
import org.springframework.ai.model.transformer.KeywordMetadataEnricher;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* 基于 AI 的文档元信息增强器（为文档补充元信息）
* 通过 ChatModelFactory 获取 ChatModel，支持多提供商动态切换
*/
@Component
public class MyKeywordEnricher {
@Resource
private ChatModelFactory chatModelFactory;
/**
* 使用 AI 提取关键词并添加到元数据
*/
public List<Document> enrichDocuments(List<Document> documents) {
KeywordMetadataEnricher enricher = new KeywordMetadataEnricher.Builder(chatModelFactory.getChatModel("CHAT"))
.keywordCount(5)
.build();
return enricher.apply(documents);
}
}

452
src/main/java/com/wok/supportbot/service/DocumentProcessingService.java
View File

@ -1,38 +1,57 @@
package com.wok.supportbot.service;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.wok.supportbot.dao.KnowledgeDocumentMapper;
import com.wok.supportbot.document.transform.MyKeywordEnricher;
import com.wok.supportbot.document.transform.MyTokenTextSplitter;
import com.wok.supportbot.entity.KnowledgeDocument;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.document.Document;
import org.springframework.ai.vectorstore.VectorStore;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import java.util.stream.Collectors;
/**
* 文档异步处理服务
* 负责文档的分块、关键词提取、向量化等耗时操作，在后台线程中执行
* 负责文档的分块与向量化（按批隔离、失败可感知、有限自动重试），在后台线程中执行
*/
@Service
@Slf4j
public class DocumentProcessingService {
/** 一批向量化的分块数：控制单批请求体积与"炸点"范围（含 volcengine-vision 逐条串行路径） */
@Value("${knowledge.vector.batch-size:50}")
private int embedBatchSize = 50;
/**
* 文档级自动重试次数（不含首次尝试）。
* 注意：批内 EmbeddingModel 已自带 3 次指数退避重试（EmbeddingModelFactory.createRetryTemplate），
* 文档级重试叠加在最外层；文档级只对"整篇仍有失败批"触发，多数瞬时错误已在内层耗掉，外层命中率低。
* 若厂商限流严重，可调小本值或调大 RETRY_BASE_DELAY_MS 冷却时间
*/
private static final int MAX_DOC_LEVEL_RETRIES = 2;
/** 重试等待基数（ms），指数放大（2s → 4s） */
private static final long RETRY_BASE_DELAY_MS = 2_000L;
/** 连续 N 批失败即中止本轮，交给文档级重试，避免打爆全部批次 */
private static final int MAX_CONSECUTIVE_BATCH_FAILURES = 3;
@Autowired
private KnowledgeDocumentMapper documentMapper;
@Autowired
private MyTokenTextSplitter myTokenTextSplitter;
@Autowired
private MyKeywordEnricher myKeywordEnricher;
@Autowired
private VectorStore pgVectorVectorStore;
@ -40,18 +59,18 @@ public class DocumentProcessingService {
private JdbcTemplate jdbcTemplate;
/**
* 异步处理文档：分块 → 关键词提取 → 向量化 → 更新状态
* 不加跨方法事务：AI 关键词提取与向量化属于慢速网络调用，期间不占用数据库连接，
* 异步处理文档：分块 → 向量化（分批入库）→ 更新状态
* 不加跨方法事务：向量化属于慢速网络调用，期间不占用数据库连接，
* 避免文件夹批量上传时多个异步任务把连接池占满导致连接超时。
*
* @param docId 文档ID
* @param documents 已解析的原始文档列表
* @param docId 文档ID
* @param documents 已解析的原始文档列表
* @param sourceName 源文件名
* @param title 文档标题
* @param title 文档标题
* @param categoryId 分类ID
* @param tags 标签列表
* @param chunkSize 分块大小（可选，覆盖全局配置）
* @param overlap 重叠大小（可选，覆盖全局配置）
* @param tags 标签列表
* @param chunkSize 分块大小（可选，覆盖全局配置）
* @param overlap 重叠大小（可选，覆盖全局配置）
*/
@Async("documentExecutor")
public void processDocumentAsync(Long docId, List<Document> documents, String sourceName,
@ -64,55 +83,19 @@ public class DocumentProcessingService {
return;
}
try {
// 1. 分块处理（使用 per-doc 参数或全局配置）
List<Document> splitDocuments = myTokenTextSplitter.splitDocuments(documents, chunkSize, overlap);
// 2. 为每个分块设置 metadata
for (int i = 0; i < splitDocuments.size(); i++) {
Document d = splitDocuments.get(i);
Map<String, Object> meta = new HashMap<>(d.getMetadata());
meta.put("documentId", String.valueOf(docId));
meta.put("chunkIndex", i);
meta.put("sourceName", sourceName);
meta.put("title", title);
if (categoryId != null && categoryId > 0) {
meta.put("categoryId", String.valueOf(categoryId));
}
if (tags != null && !tags.isEmpty()) {
meta.put("tags", tags);
}
// P1-2.1: 标记启用状态，用于 RAG 检索过滤
meta.put("enabled", String.valueOf(Boolean.TRUE.equals(doc.getEnabled())));
splitDocuments.set(i, new Document(d.getId(), d.getText(), meta));
}
// 3. 关键词提取
List<Document> enrichedDocuments = myKeywordEnricher.enrichDocuments(splitDocuments);
// 4. 向量化存储
pgVectorVectorStore.add(enrichedDocuments);
DocMeta meta = DocMeta.of(docId, sourceName, title, categoryId, tags,
Boolean.TRUE.equals(doc.getEnabled()));
// 5. 更新状态为 READY
doc.setStatus("READY");
doc.setChunkCount(enrichedDocuments.size());
documentMapper.updateById(doc);
log.info("异步处理文档完成: id={}, title={}, chunks={}", docId, title, enrichedDocuments.size());
} catch (Exception e) {
doc.setStatus("FAILED");
doc.setErrorMessage(e.getMessage());
documentMapper.updateById(doc);
log.error("异步处理文档失败: id={}, title={}", docId, title, e);
}
// 新文档首次处理无需清理（尚无向量）；若中途失败触发整体重试，框架内会先清残留向量再重建
runPipelineWithRetry(doc, meta, false,
() -> myTokenTextSplitter.splitDocuments(documents, chunkSize, overlap));
}
/**
* 异步重新处理文档（重新分块 + 向量化）
* 异步重新处理文档（清理旧向量后按批重建向量化）
* 使用文档存储的 per-doc 分块参数（extraConfig），如无则使用全局配置
*
* @param docId 文档ID
* @param docId 文档ID
* @param documents 解析后的文档列表
*/
@Async("documentExecutor")
@ -124,61 +107,265 @@ public class DocumentProcessingService {
return;
}
try {
// 删除旧向量
String sql = "SELECT id::text FROM vector_store WHERE metadata->>'documentId' = ?";
List<String> oldIds = jdbcTemplate.queryForList(sql, String.class, String.valueOf(docId));
if (!oldIds.isEmpty()) {
pgVectorVectorStore.delete(oldIds);
}
DocMeta meta = DocMeta.fromDocument(doc);
// 从 extraConfig 读取 per-doc 分块参数
// reprocess 兼容：首轮就清理旧向量后按批重建；分块参数从 extraConfig 读取
runPipelineWithRetry(doc, meta, true, () -> {
Integer chunkSize = null;
Integer overlap = null;
if (doc.getExtraConfig() != null) {
Object cs = doc.getExtraConfig().get("chunkSize");
Object ol = doc.getExtraConfig().get("overlap");
if (cs instanceof Number) chunkSize = ((Number) cs).intValue();
if (ol instanceof Number) overlap = ((Number) ol).intValue();
}
// 重新分块（使用 per-doc 参数或全局配置）
List<Document> splitDocuments = myTokenTextSplitter.splitDocuments(documents, chunkSize, overlap);
for (int i = 0; i < splitDocuments.size(); i++) {
Document d = splitDocuments.get(i);
Map<String, Object> meta = new HashMap<>(d.getMetadata());
meta.put("documentId", String.valueOf(docId));
meta.put("chunkIndex", i);
meta.put("sourceName", doc.getSourceName());
meta.put("title", doc.getTitle());
if (doc.getCategoryId() != null && doc.getCategoryId() > 0) {
meta.put("categoryId", String.valueOf(doc.getCategoryId()));
if (doc.getExtraConfig().get("chunkSize") instanceof Number cs) chunkSize = cs.intValue();
if (doc.getExtraConfig().get("overlap") instanceof Number ol) overlap = ol.intValue();
}
return myTokenTextSplitter.splitDocuments(documents, chunkSize, overlap);
});
}
/**
* 文档级处理框架（外层兜底）：切分 → 分批向量化 → 失败自动整体重试有限次数。
* 外层 try-catch 保证任何未预期异常下文档状态都从 PROCESSING 收敛到 FAILED，
* 避免 @Async 异常被 Spring 静默吞掉导致文档永久悬挂（前端轮询永不结束）
*/
private void runPipelineWithRetry(KnowledgeDocument doc, DocMeta meta,
boolean cleanBeforeFirstAttempt,
Supplier<List<Document>> splitter) {
try {
doPipelineWithRetry(doc, meta, cleanBeforeFirstAttempt, splitter);
} catch (Exception e) {
// 兜底落 FAILED；若状态写入也失败（如 DB 抖动），仅记日志，避免再次上抛造成悬挂
log.error("文档向量化处理出现未捕获异常: id={}", doc.getId(), e);
try {
String reason = e.getMessage() == null
? e.getClass().getSimpleName()
: truncate(e.getMessage());
patchStatus(doc.getId(), "FAILED", safeCountVectors(String.valueOf(doc.getId())),
"内部错误: " + reason);
} catch (Exception ex) {
log.error("兜底标记文档处理失败状态也失败: id={}, error={}", doc.getId(), ex.getMessage(), ex);
}
}
}
private void doPipelineWithRetry(KnowledgeDocument doc, DocMeta meta,
boolean cleanBeforeFirstAttempt,
Supplier<List<Document>> splitter) {
String docIdStr = String.valueOf(doc.getId());
for (int attempt = 0; attempt <= MAX_DOC_LEVEL_RETRIES; attempt++) {
// 每次尝试前复查文档仍存在（用户可能在重试等待/sleep 期间删除文档），避免向已删文档写孤儿向量
if (documentMapper.selectById(doc.getId()) == null) {
log.info("文档在处理期间已被删除，放弃处理: id={}", doc.getId());
return;
}
// 重试前：指数等待 + 清理上一轮残留向量，避免 split 重建产生新 Document.id 造成重复向量
if (attempt > 0) {
if (!sleepBackoff(attempt)) {
patchStatus(doc.getId(), "FAILED", safeCountVectors(docIdStr), "处理被中断");
return;
}
if (doc.getTags() != null && doc.getTags().containsKey("tags")) {
meta.put("tags", doc.getTags().get("tags"));
deleteVectorsByDocumentId(docIdStr);
log.warn("文档向量化整体重试第 {}/{} 次: id={}", attempt, MAX_DOC_LEVEL_RETRIES, doc.getId());
} else if (cleanBeforeFirstAttempt) {
deleteVectorsByDocumentId(docIdStr);
}
List<Document> chunks;
try {
chunks = splitter.get();
} catch (Exception e) {
if (attempt < MAX_DOC_LEVEL_RETRIES && isRetryable(e.getMessage())) {
continue;
}
// P1-2.1: 标记启用状态
meta.put("enabled", String.valueOf(Boolean.TRUE.equals(doc.getEnabled())));
splitDocuments.set(i, new Document(d.getId(), d.getText(), meta));
markFailed(doc, "文档分块失败: " + truncate(e.getMessage()));
return;
}
List<Document> enrichedDocuments = myKeywordEnricher.enrichDocuments(splitDocuments);
pgVectorVectorStore.add(enrichedDocuments);
if (chunks == null || chunks.isEmpty()) {
// 分块为空属于内容/参数问题，重试不会自愈
markFailed(doc, "文档分块结果为空（无可向量化文本或小于最小分块），请检查内容与分块参数");
return;
}
doc.setStatus("READY");
doc.setChunkCount(enrichedDocuments.size());
doc.setErrorMessage(null);
documentMapper.updateById(doc);
List<Document> annotated = annotateChunks(chunks, meta);
BatchResult result = vectorizeInBatches(docIdStr, annotated, doc.getId());
log.info("异步重新处理文档成功: id={}, title={}, chunks={}", docId, doc.getTitle(), enrichedDocuments.size());
// 处理期间文档被删除：终止且不再写状态（删除是用户明确意图，不覆盖为 FAILED）
if (result.cancelled) {
return;
}
int stored = safeCountVectors(docIdStr);
if (!result.hasFailure()) {
patchStatus(doc.getId(), "READY", stored, null);
log.info("文档向量化成功: id={}, title={}, chunks={}", doc.getId(), doc.getTitle(), stored);
return;
}
// 有失败批：若仍有余量且错误属瞬时/基础设施类，则整篇重试
if (attempt < MAX_DOC_LEVEL_RETRIES && isRetryable(result.firstError)) {
log.warn("存在失败批，准备整体重试: id={}, total={}, stored={}, firstError={}",
doc.getId(), result.totalChunks, stored, result.firstError);
continue;
}
// 已入库块保留在库（chunk_count 记实际入库数），缺失区间聚合进 error_message
markFailed(doc, result.buildSummary(stored));
return;
}
}
/**
* 分批向量化入库：按 embedBatchSize 逐批 add()，单批失败仅记录缺失区间后继续；
* 连续失败超过阈值且错误可重试时中止本轮（交给文档级重试），避免打满全部批次
*
* @return 各批结果聚合（成功批已写入 DB）
*/
private BatchResult vectorizeInBatches(String docIdStr, List<Document> annotated, Long docId) {
BatchResult result = new BatchResult(annotated.size());
int consecutiveFailures = 0;
int batchSize = effectiveBatchSize();
for (int start = 0; start < annotated.size(); start += batchSize) {
// 批前复查：文档若在异步处理期间被删除，立即终止后续写入，避免在 vector_store 留下孤儿向量
if (docId != null && documentMapper.selectById(docId) == null) {
log.info("处理期间文档已被删除，终止剩余批次: docId={}", docIdStr);
result.cancelled = true;
break;
}
int end = Math.min(start + batchSize, annotated.size());
List<Document> batch = annotated.subList(start, end);
try {
pgVectorVectorStore.add(batch);
consecutiveFailures = 0;
log.info("批次向量化成功: docId={}, 第{}~{}块, 累计进度 {}/{}",
docIdStr, start + 1, end, Math.min(end, result.totalChunks), result.totalChunks);
} catch (Exception e) {
consecutiveFailures++;
// 缺失区间：0-based chunkIndex 闭区间 [start, end-1]；错误消息截断防 error_message 过长/回显正文
result.recordFailure(start, end - 1, truncate(e.getMessage()));
log.error("批次向量化失败: docId={}, 第{}~{}块, error={}",
docIdStr, start + 1, end, truncate(e.getMessage()));
if (consecutiveFailures >= MAX_CONSECUTIVE_BATCH_FAILURES && isRetryable(e.getMessage())) {
result.aborted = true;
// 连续失败中止本轮：把尚未执行的后续批次区间一并记入，避免 error_message 缺失段不完整
if (end < annotated.size()) {
result.recordFailure(end, annotated.size() - 1, "连续失败已中止本轮，后续批次未执行");
}
break;
}
}
}
return result;
}
/**
* 为每个分块写入关联元数据：documentId/chunkIndex/sourceName/title/categoryId/tags/enabled
*/
private List<Document> annotateChunks(List<Document> splitDocuments, DocMeta meta) {
for (int i = 0; i < splitDocuments.size(); i++) {
Document d = splitDocuments.get(i);
Map<String, Object> m = new HashMap<>(d.getMetadata());
m.put("documentId", String.valueOf(meta.docId()));
m.put("chunkIndex", i);
m.put("sourceName", meta.sourceName());
m.put("title", meta.title());
if (meta.categoryId() != null && meta.categoryId() > 0) {
m.put("categoryId", String.valueOf(meta.categoryId()));
}
if (meta.tagsValue() instanceof List<?> tags && !tags.isEmpty()) {
m.put("tags", tags);
}
// P1-2.1: 标记启用状态，用于 RAG 检索过滤
m.put("enabled", String.valueOf(meta.enabled()));
splitDocuments.set(i, new Document(d.getId(), d.getText(), m));
}
return splitDocuments;
}
// ==================== 私有工具 ====================
/** 仅 patch 状态/分块数/错误信息，避免覆盖用户在异步期间的并发编辑（如改标题/分类/toggle enabled） */
private void patchStatus(Long docId, String status, Integer chunkCount, String errorMessage) {
documentMapper.update(null, new LambdaUpdateWrapper<KnowledgeDocument>()
.eq(KnowledgeDocument::getId, docId)
.set(KnowledgeDocument::getStatus, status)
.set(KnowledgeDocument::getChunkCount, chunkCount)
.set(KnowledgeDocument::getErrorMessage, errorMessage));
}
private void markFailed(KnowledgeDocument doc, String message) {
String docIdStr = String.valueOf(doc.getId());
patchStatus(doc.getId(), "FAILED", safeCountVectors(docIdStr), message);
log.error("文档向量化失败: id={}, title={}, error={}", doc.getId(), doc.getTitle(), message);
}
/** 统计实际已入库向量数（失败时容错，DB 抖动不阻断状态收敛） */
private int safeCountVectors(String docIdStr) {
try {
return countStoredVectors(docIdStr);
} catch (Exception e) {
doc.setStatus("FAILED");
doc.setErrorMessage(e.getMessage());
documentMapper.updateById(doc);
log.error("异步重新处理文档失败: id={}, title={}", docId, doc.getTitle(), e);
log.warn("统计已入库向量数失败: docId={}, error={}", docIdStr, e.getMessage());
return 0;
}
}
/** 截断错误消息，避免 error_message 过长或回显大段文档文本 */
private static String truncate(String msg) {
if (msg == null) {
return null;
}
return msg.length() <= 500 ? msg : msg.substring(0, 500) + "…(已截断)";
}
/** 实际已入库向量条数：以 DB 为准，天然吸收"失败批内偶发的部分写入" */
private int countStoredVectors(String docIdStr) {
Integer c = jdbcTemplate.queryForObject(
"SELECT count(*) FROM vector_store WHERE metadata->>'documentId' = ?",
Integer.class, docIdStr);
return c == null ? 0 : c;
}
/** 物理删除该文档在 vector_store 的全部向量（vector_store 无逻辑删除语义） */
private void deleteVectorsByDocumentId(String docIdStr) {
List<String> ids = jdbcTemplate.queryForList(
"SELECT id::text FROM vector_store WHERE metadata->>'documentId' = ?",
String.class, docIdStr);
if (!ids.isEmpty()) {
pgVectorVectorStore.delete(ids);
log.debug("清理旧向量: documentId={}, count={}", docIdStr, ids.size());
}
}
private int effectiveBatchSize() {
return embedBatchSize > 0 ? embedBatchSize : 50;
}
/** 简单指数退避等待；被中断时复位中断标志并返回 false（放弃处理） */
private static boolean sleepBackoff(int attempt) {
long ms = RETRY_BASE_DELAY_MS * (1L << (attempt - 1));
try {
Thread.sleep(ms);
return true;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
/** 仅对疑似瞬时/基础设施类错误做文档级重试，4xx 参数类错误不空转 */
private static boolean isRetryable(String msg) {
// 消息缺失的未知错误不做整篇重试（避免对 NPE 等非瞬时错误空转 2 次），仅对明确瞬时/限流特征重试
if (msg == null || msg.isBlank()) {
return false;
}
String m = msg.toLowerCase();
return m.contains("timeout") || m.contains("timed out") || m.contains("connection")
|| m.contains("429") || m.contains("too many request") || m.contains("rate limit")
|| m.contains(" 500") || m.contains(" 502") || m.contains(" 503") || m.contains(" 504")
|| m.contains("socket") || m.contains("i/o error") || m.contains("internal server error")
|| m.contains("service unavailable") || m.contains("bad gateway");
}
/**
@ -206,4 +393,73 @@ public class DocumentProcessingService {
}
return null;
}
/**
* 分批向量化结果聚合：成功批已入库；失败批记录缺失 chunkIndex 区间与首个错误
*/
private static final class BatchResult {
final int totalChunks;
/** 每项为 [start, end]（0-based chunkIndex 闭区间），最多保留前 8 段 */
final List<int[]> failedRanges = new ArrayList<>();
String firstError;
boolean aborted;
/** 处理期间文档被删除时置位：调用方应终止且不再写状态 */
boolean cancelled;
BatchResult(int totalChunks) {
this.totalChunks = totalChunks;
}
void recordFailure(int start, int end, String error) {
failedRanges.add(new int[]{start, end});
if (firstError == null) {
firstError = error;
}
}
boolean hasFailure() {
return !failedRanges.isEmpty();
}
/** 例：已入库 320/400 块，缺失第 321~400 块 向量化失败: <原因> */
String buildSummary(int storedChunks) {
String ranges = failedRanges.stream().limit(8)
.map(r -> r[0] == r[1]
? "第 " + (r[0] + 1) + " 块"
: "第 " + (r[0] + 1) + "~" + (r[1] + 1) + " 块")
.collect(Collectors.joining("、"));
if (failedRanges.size() > 8) {
ranges += "…共 " + failedRanges.size() + " 段";
}
String msg = "已入库 " + storedChunks + "/" + totalChunks + " 块，缺失 " + ranges + " 向量化失败";
if (aborted) {
msg += "（连续失败已中止本轮）";
}
if (firstError != null) {
msg += ": " + firstError;
}
return msg;
}
}
/**
* 分块元数据值对象：统一两个异步入口（新上传/reprocess）的 metadata 标注逻辑
*/
private record DocMeta(Long docId, String sourceName, String title, Long categoryId,
Object tagsValue, boolean enabled) {
static DocMeta of(Long docId, String sourceName, String title, Long categoryId,
List<String> tags, boolean enabled) {
return new DocMeta(docId, sourceName, title, categoryId, tags, enabled);
}
static DocMeta fromDocument(KnowledgeDocument doc) {
Object tagsValue = null;
if (doc.getTags() != null && doc.getTags().containsKey("tags")) {
tagsValue = doc.getTags().get("tags");
}
return new DocMeta(doc.getId(), doc.getSourceName(), doc.getTitle(),
doc.getCategoryId(), tagsValue, Boolean.TRUE.equals(doc.getEnabled()));
}
}
}

8
src/main/java/com/wok/supportbot/service/DocumentService.java
View File

@ -7,7 +7,6 @@ import com.wok.supportbot.document.extract.JsonDocumentLoader;
import com.wok.supportbot.document.extract.MarkdownDocumentLoader;
import com.wok.supportbot.document.extract.SimpleStringDocumentReader;
import com.wok.supportbot.document.extract.TikaDocumentReader;
import com.wok.supportbot.document.transform.MyKeywordEnricher;
import com.wok.supportbot.document.transform.MyTokenTextSplitter;
import com.wok.supportbot.entity.CategoryNode;
import com.wok.supportbot.entity.KnowledgeCategory;
@ -56,9 +55,6 @@ public class DocumentService {
@Autowired
private MyTokenTextSplitter myTokenTextSplitter;
@Autowired
private MyKeywordEnricher myKeywordEnricher;
@Autowired
private TikaDocumentReader tikaDocumentReader;
@ -937,9 +933,7 @@ public class DocumentService {
meta.put("enabled", String.valueOf(Boolean.TRUE.equals(doc.getEnabled())));
Document newDoc = new Document(vectorId, newContent, meta);
// 关键词提取
List<Document> enriched = myKeywordEnricher.enrichDocuments(List.of(newDoc));
pgVectorVectorStore.add(enriched);
pgVectorVectorStore.add(List.of(newDoc));
log.info("更新分块: docId={}, chunkIndex={}, vectorId={}", docId, chunkIndex, vectorId);
}

3
src/main/resources/application.yml
View File

@ -58,6 +58,9 @@ knowledge:
# 向量维度，需与 Embedding 模型输出维度一致
# 千问 text-embedding-v2: 1024 | 豆包 doubao-embedding-text-240515: 2048 | OpenAI text-embedding-3-small: 1536
dimension: 1024
# 向量化分批大小：单批分块数（逐批入库，失败批只记录缺失区间不扩散到整篇）
# 默认 50；豆包多模态(vision)模型逐条调用较慢，若大量使用可调小(如 20)减小单批串行耗时
batch-size: 50
role:
# 严格隔离：true=角色未绑定知识库分类时禁止检索任何内容；false=可检索全部知识库
strict-isolation: false

Write Preview
Loading…
Cancel
Save