15 · ETL 管道批量灌库 学完能做什么把 Reader→Splitter→富化→Embedding→VectorStore 串成一条可复用、可幂等、可分批的灌库流水线并了解大批量与增量更新要点。⏱️ 预计耗时50 分钟含动手 依赖前置第 12、13、14 章 难度一句话ETL 管道 读 → 切 → 富化 → 灌四步串成一个方法生产级还要做到幂等、分批、可增量更新。1. 完整管道全景Resource文件 │ Reader第12章 ▼ ListDocument长文档 │ Splitter第13章 ▼ ListDocumentchunks │ 富化 metadata第14章 ▼ ListDocument带标签 chunks │ vectorStore.add第7/9章内部自动 Embedding ▼ pgvector2. 一条可复用的灌库管道ServiceclassIngestPipeline{privatefinalVectorStorevectorStore;IngestPipeline(VectorStorevectorStore){this.vectorStorevectorStore;}/** 读 → 切 → 富化 → 灌 */publicintingest(Resourcefile,Stringsource,MapString,ObjectbaseMeta){// 1) 读ListDocumentrawnewTikaDocumentReader(file).get();// 2) 切ListDocumentchunksnewTokenTextSplitter(500,200,5,10000,true).apply(raw);// 3) 富化 稳定 id幂等关键inti0;for(Documentc:chunks){c.getMetadata().putAll(baseMeta);c.getMetadata().put(source,source);c.getMetadata().put(ingestTime,Instant.now().toString());i;}// 4) 灌分批batchAdd(chunks,100);returnchunks.size();}privatevoidbatchAdd(ListDocumentdocs,intbatch){for(inti0;idocs.size();ibatch){vectorStore.add(docs.subList(i,Math.min(ibatch,docs.size())));}}}3. 幂等重复灌同一文件不产生重复灌库最容易踩的坑是「重跑一次数据翻倍」。解决办法用可推导的稳定 id 先删后加回顾第 09 章。// 为每个 chunk 生成稳定 id来源 序号Stringidsource#index;DocumentcnewDocument(id,content,metadata);// 重灌前先删掉该 source 的所有旧 chunk再整体重灌vectorStore.delete(newFilterExpressionBuilder().eq(source,source).build());vectorStore.add(newChunks); 策略以「文件/文档」为单位做整体替换——删掉旧 source 的全部 chunk再灌新的。简单且不会残留。4. 分批与限速大批量必做海量灌库时一次性 add 会撑爆内存、触发 embedding API 限流、单事务过大。要点✅ 分批提交如每批 100 条 ✅ 控制并发避免 embedding 接口被限流配合第 36 章重试 ✅ 大文件流式读取别一次性全load 进内存 ✅ 记录进度失败可从断点续灌5. 增量更新与删除知识库不是灌一次就完事文档会更新、下线// 文档更新整体替换该 sourcepublicvoidreindex(Resourcefile,Stringsource,MapString,Objectmeta){vectorStore.delete(newFilterExpressionBuilder().eq(source,source).build());ingest(file,source,meta);}// 文档下线直接删publicvoidremove(Stringsource){vectorStore.delete(newFilterExpressionBuilder().eq(source,source).build());}底层的 VACUUM、索引膨胀与重建等长期维护问题见 pgvector 文档33。6. 触发灌库的几种方式方式适用启动时PostConstruct小型固定知识库、demo手动 REST 接口触发运营上传文档后灌库定时任务扫描目录批量同步文件夹消息队列驱动大规模、解耦、可重试的生产方案PostMapping(/ingest)MapString,Objectingest(RequestParamStringpath,RequestParamStringsource){intnpipeline.ingest(newFileSystemResource(path),source,Map.of(category,doc));returnMap.of(source,source,chunks,n);}7. 灌库后验收// 1) 数量对不对LongcntjdbcTemplate.queryForObject(SELECT count(*) FROM vector_store WHERE metadata-source ?,Long.class,source);// 2) 随手检索一条确认能召回ListDocumenthitsvectorStore.similaritySearch(SearchRequest.builder().query(退货几天).topK(1).build());8. 一句话总结ETL 管道把「读→切→富化→灌」串成一个方法生产化的三要点是稳定 id 幂等先删后加、分批限速、以 source 为单位增量替换——这样知识库才能长期可维护。第三阶段完成你现在能把真实文档读进来、合理分块、打好标签、幂等地批量灌库——知识库有了真正的内容。➡️ 下一阶段第 16–22 章进入RAG 应用构建用 Advisor 打通检索问答、定制提示、加记忆、做流式与溯源真正做出一个能用的知识库问答。