数据截至 (上游 commit daa7624a2755)
RAG:摄取与检索两条流水线
30 秒导读: RAG(Retrieval-Augmented Generation,检索增强生成)就是"回答前先去资料堆里翻几段相关的塞进 prompt"。LangChain4j 没有把它做成一个大类,而是切成两条互不认识的流水线:离线的摄取线把文件变成向量存起来,在线的检索线在每次提问时翻出内容拼进用户消息。本章讲这两条线各有哪些可插拔点,以及在线那条线的核心——
DefaultRetrievalAugmentor.augment的五段流水。
1. 这是什么(零基础也能懂)
-
一句话定义: RAG = 在把用户的问题发给大模型之前,先从你自己的资料库里检索出几段相关文本,追加到问题后面一起发过去。
-
解决什么问题: 模型不知道你公司内部文档里写了什么,也不可能把 5000 页手册塞进上下文窗口。RAG 的做法是"每次只塞相 关的那 3 段"。
-
LangChain4j 的取舍: 它不提供一个
RagPipeline大对象,而是提供两组接口 + 每个接口的默认实现。你可以只用默认值(3 行代码),也可以逐个插槽替换。
两条流水线的关系 —— 这是理解本章的地基:
离线:摄取(通常跑一次) 在线:检索(每次提问都跑)
原始文件 / 网页 / S3 用户的一句话
│ DocumentLoader + DocumentParser │
v v
Document ──DocumentSplitter──> TextSegment RetrievalAugmentor
│ │ 检索 → 融合 → 注入
EmbeddingModel v
│ 增强后的 UserMessage
v ^
【EmbeddingStore】─────┘
怎么读这张图:两条线只在 EmbeddingStore 处相遇。左边写、右边读,双方都不需要认识对方的类。这就是为什么你可以离线用 Spark 跑摄取、线上只跑检索。
用起来什么样 —— 最小的一个完整 RAG(示意,非源码):
// 1. 摄取:把一个目录里的文档灌进内存向量库
EmbeddingStore<TextSegment> store = new InMemoryEmbeddingStore<>();
List<Document> docs = FileSystemDocumentLoader.loadDocuments("/path/to/docs");
EmbeddingStoreIngestor.ingest(docs, store); // 切分/向量化用 SPI 默认值
// 2. 检索:把 store 挂到一个 AI Service 上
Assistant assistant = AiServices.builder(Assistant.class)
.chatModel(model)
.contentRetriever(EmbeddingStoreContentRetriever.from(store)) // 唯一的 RAG 配置
.build();
String answer = assistant.chat("我们的退款政策是几天?"); // 内部自动检索 + 注入
- 一句话直觉: 摄取线像"给一本书做索引卡片",检索线像"考试时按题目抽几张卡片夹进答题纸"。两件事在不同时间发生,靠"卡片盒"(EmbeddingStore)连接。
2. 四个数据模型:整条链路只搬运这四种东西
先认清数据,后面所有接口都只是在这四种类型之间做转换。
| 类型 | 是什么 | 关键字段 | 定义位置 |
|---|---|---|---|
Document | 一整份原始文档 | text() + metadata() | langchain4j-core/src/main/java/dev/langchain4j/data/document/Document.java:10 |
Metadata | 挂在文档/段上的键值对 | 只允许 String / UUID / int / long / float / double | langchain4j-core/src/main/java/dev/langchain4j/data/document/Metadata.java:35 |
TextSegment | 切分后的一小段文本 | text() + metadata()(同样的 Metadata 类) | langchain4j-core/src/main/java/dev/langchain4j/data/segment/TextSegment.java:16 |
Embedding | 一段文本的向量 | float[] vector | langchain4j-core/src/main/java/dev/langchain4j/data/embedding/Embedding.java:16 |
三个细节值得单独点出来:
其一,Document 和 TextSegment 结构完全一样。 都是"文本 + Metadata"。区别只是语义上的粒度。所以当你不切分时,Document.toTextSegment() 直接原样转换(Document.java:44)。
其二,Metadata 的值类型是白名单。 Metadata 在 put 时用 SUPPORTED_VALUE_TYPES(声明在 Metadata.java:37)校验(Metadata.java:80),不在白名单的类型直接抛异常。这是为了让下游各家向量库都能映射过去——它是跨向量库过滤的最小公约数。
其三,三个约定俗成的 metadata key 写死在 Document 接口上:
String FILE_NAME = "file_name"; // Document.java:15
String ABSOLUTE_DIRECTORY_PATH = "absolute_directory_path"; // :19
String URL = "url"; // :23
各个 DocumentSource 用这几个 key 写来源信息,检索时你就能用它们做元数据过滤(见 §9)。
3. 摄取线第一步:Loader + Parser(把字节变成 Document)
它要解决的小问题: "从哪儿读"和"怎么解析格式"是两件事,不该耦合。
LangChain4j 的答案是一个 8 行的胶水函数——DocumentLoader.load:
public static Document load(DocumentSource source, DocumentParser parser) {
try (InputStream inputStream = source.inputStream()) {
Document document = parser.parse(inputStream);
document.metadata().putAll(source.metadata().toMap()); // 来源信息回填
return document;
langchain4j-core/src/main/java/dev/langchain4j/data/document/DocumentLoader.java:23 · 符号 DocumentLoader.load。重点看最后那句 putAll:解析器只管把字节变成文本,"这段文本来自哪个文件/URL"由 source 补上。职责切得很干净。
3.1 三个开箱即用的 Loader
| Loader | 从哪儿读 | 有无批量能力 | 文件 |
|---|---|---|---|
FileSystemDocumentLoader | 本地文件/目录 | 有:loadDocuments / loadDocumentsRecursively,可带 PathMatcher glob | langchain4j/src/main/java/dev/langchain4j/data/document/loader/FileSystemDocumentLoader.java:27 |
ClassPathDocumentLoader | classpath 资源 | 有,同样支持 PathMatcher 与自定义 ClassLoader | langchain4j/src/main/java/dev/langchain4j/data/document/loader/ClassPathDocumentLoader.java:34 |
UrlDocumentLoader | 一个 URL | 无,只有单个 load(url, parser) | langchain4j/src/main/java/dev/langchain4j/data/document/loader/UrlDocumentLoader.java:20 |
批量方法有一个容错约定:加载失败的文件被跳过而不是让整批失败(FileSystemDocumentLoader 的批量方法 javadoc 明确写 "Skips any Documents that fail to load",见 FileSystemDocumentLoader.java:115)。
远程来源(S3、Azure Blob、GitHub、GCS、腾讯 COS、Selenium/Playwright 抓浏览器渲染后的页面)各自是独立模块,在克隆根的 document-loaders/ 目录下,共 7 个。
3.2 DocumentParser 是个 SPI,不是 if-else
DocumentParser 接口只有一个方法(DocumentParser.java:22):
Document parse(InputStream inputStream);
内置实现只有一个 TextDocumentParser——读全部字节、按 charset(默认 UTF-8)解码;若结果 isBlank() 就抛 BlankDocumentException(langchain4j/src/main/java/dev/langchain4j/data/document/parser/TextDocumentParser.java:26-31)。
其余格式全是可选模块 + SPI 自动发现。关键一行在 FileSystemDocumentLoader.java:31:
private static final DocumentParser DEFAULT_DOCUMENT_PARSER =
getOrDefault(DocumentParserLoader.loadDocumentParser(), TextDocumentParser::new);
意思是:classpath 上如果有某个 DocumentParserFactory 的 SPI 注册,就用它;否则退回纯文本解析。所以"支持 PDF"这件事在 LangChain4j 里等于"往 pom 里加一个依赖",代码一行不改。
可选解析器模块(克隆根 document-parsers/,共 6 个):
| 模块 | 主类 | 覆盖格式 |
|---|---|---|
langchain4j-document-parser-apache-pdfbox | ApachePdfBoxDocumentParser | |
langchain4j-document-parser-apache-poi | ApachePoiDocumentParser | MS Office(doc/xls/ppt 系) |
langchain4j-document-parser-apache-tika | ApacheTikaDocumentParser | 通吃(Tika 自动嗅探格式) |
langchain4j-document-parser-markdown | MarkdownDocumentParser | Markdown |
langchain4j-document-parser-docling | DoclingDocumentParser | 走 Docling 做版面理解 |
langchain4j-document-parser-yaml | YamlDocumentParser | YAML |
只有 tika 模块注册了 SPI(document-parsers/langchain4j-document-parser-apache-tika/src/main/resources/META-INF/services/dev.langchain4j.spi.data.document.parser.DocumentParserFactory),其余模块需要你手动 new 出来传进 loader。
4. 摄取线第二步:切分器的"逐级降级"
它要解决的小问题: 一段太长塞不进 maxSegmentSize 怎么办?硬切会把句子拦腰砍断。
思路: 优先按语义边界最大的单位切(段落),塞不下再降一级(行 → 句 → 词 → 字符)。降级不是全局的,而是只对那个塞不下的部分降级——其余部分仍按段落切。
段落 ──塞不下──> 行 ──塞不下──> 句子 ──塞不下──> 词 ──塞不下──> 字符
\R\R \R OpenNLP 模型 \s+ ""
读法:从左到右是降级顺序,能塞进 maxSegmentSize 就地停下,不再往下降。
4.1 五个具体切分器
各自只需实现三个方法:怎么切(split)、怎么拼回去(joinDelimiter)、切不下时降给谁(defaultSubSplitter)。
| 切分器 | split 用的规则 | 默认降级到 | 行号(langchain4j/src/main/java/dev/langchain4j/data/document/splitter/) |
|---|---|---|---|
DocumentByParagraphSplitter | \s*(?>\R)\s*(?>\R)\s*(空行) | DocumentBySentenceSplitter | DocumentByParagraphSplitter.java:56,66 |
DocumentByLineSplitter | \s*\R\s* | DocumentBySentenceSplitter | DocumentByLineSplitter.java:56,66 |
DocumentBySentenceSplitter | OpenNLP SentenceDetectorME | DocumentByWordSplitter | DocumentBySentenceSplitter.java:89,100 |
DocumentByWordSplitter | \s+ | DocumentByCharacterSplitter | DocumentByWordSplitter.java:56,66 |
DocumentByCharacterSplitter | ""(逐字符) | null(最底层,无处可降) | DocumentByCharacterSplitter.java:47,57 |
句子切分器带一个内置的英文 OpenNLP 模型 /opennlp/opennlp-en-ud-ewt-sentence-1.2-2.5.0.bin(DocumentBySentenceSplitter.java:83),也可以在构造时传自己的 SentenceModel。
链外还有一个 DocumentByRegexSplitter(按你给的正则切),它的 defaultSubSplitter() 也返回 null(DocumentByRegexSplitter.java:82-83),但它不出现在上面这条降级链的任何一级里。注意 别把它和表格末行的 DocumentByCharacterSplitter 混为一谈:后者同样返回 null,却是链的最底层,不是链外的独立品。
4.2 DocumentSplitters.recursive 与"默认降级链"不是一回事
推荐入口是工厂方法 DocumentSplitters.recursive(langchain4j/src/main/java/dev/langchain4j/data/document/splitter/DocumentSplitters.java:21),它显式串出四级:
return new DocumentByParagraphSplitter(maxSegmentSizeInTokens, maxOverlapSizeInTokens, tokenCountEstimator,
new DocumentByLineSplitter(..., new DocumentBySentenceSplitter(..., new DocumentByWordSplitter(...))));
注意对比上面的表:段落切分器的默认 subSplitter 是句子切分器(跳过了行)。只有你用 DocumentSplitters.recursive 时,"行"这一级才会被插进来。这是个容易看漏的差异。
4.3 真正的切分循环:HierarchicalDocumentSplitter.split
所有降级逻辑都在基类的一个方法里(langchain4j/src/main/java/dev/langchain4j/data/document/splitter/HierarchicalDocumentSplitter.java:120 · split(Document))。核心是一个"攒够就冲刷"的循环,配合一个 SegmentBuilder 当缓冲区:
for each part(本级切出来的块):
├─ 装得下? → segmentBuilder.append(part),继续 (:132)
├─ 缓冲区非空? → 冲刷成一个 TextSegment,算出 overlap,
│ 用 overlap 打底重开缓冲区;再试装一次 (:138-154)
└─ 还是装不下?
├─ subSplitter == null → 抛 RuntimeException(明确报错) (:158)
└─ 否则 → 把这块交给 subSplitter 递归切,结果全部收下 (:171)
三处非显然的设计:
-
重叠(overlap)只按整句取。
overlapFrom(:195)固定用一个惰性创建的DocumentBySentenceSplitter(1, 0, null, null)(:29-34)把上一段倒着按句拆,从尾部往前塞满maxOverlapSize为止。也就是说 overlap 永远不会是半句话。 -
防"纯 overlap 段"。 冲刷前会判断
!segmentText.equals(overlap)(:141),末尾收尾时也判一次(:182)。否则当一段内容全部来自上一段的重叠时,会产出一个完全冗余的段。 -
尺寸单位由
TokenCountEstimator决定。estimateSize(:224)在没给估算器时退化成text.length()(按字符),给了就按 token 算。同一个maxSegmentSize参数,单位随之变化——报错信息里也会跟着切换 "characters"/"tokens"(:164)。
SegmentBuilder(langchain4j/src/main/java/dev/langchain4j/data/document/splitter/SegmentBuilder.java:13,包级私有 @Internal)只做三件事:hasSpaceFor 判断(算上 join 分隔符的长度,:51-71)、append / prepend 拼接(:88、:101)、toString 时 trim()。
每个产出的段会继承文档的全部 metadata,并额外写一个 index 键记录段序(:241 · createSegment)。
5. 摄取线终点:EmbeddingStoreIngestor
这个类就是把前面几步串成一条线,方法体不到 30 行(langchain4j-core/src/main/java/dev/langchain4j/store/embedding/EmbeddingStoreIngestor.java:192 · ingest(List<Document>)):
documents
│ documentTransformer (可选:清洗/加标题) :181
v
documents
│ documentSplitter.splitAll —— 若为 null 则整篇当一段 :186-191
v
segments
│ textSegmentTransformer (可选:给每段前缀文档标题) :192
v
segments ──embeddingModel.embedAll──> embeddings :198
│
└──> embeddingStore.addAll(embeddings, segments) :202
四个可插拔点、两个必填项(embeddingModel、embeddingStore),而这两个必填项也能从 SPI 拿:loadDocumentSplitter(:86)与 loadEmbeddingModel(:102)都走 ServiceHelper.loadFactories,并且在发现多于一个实现时直接抛异常要求你显式指定——宁可失败也不猜。
返回值 IngestionResult 目前只裹了一个字段:向量化过程的 TokenUsage(langchain4j-core/src/main/java/dev/langchain4j/store/embedding/IngestionResult.java:16-22)。用来算这次摄取花了多少钱。
静态快捷方法 EmbeddingStoreIngestor.ingest(docs, store)(:144)则完全依赖 SPI —— 这正是 §10 讲的 easy-rag 生效的地方。
6. 检索线的核心:DefaultRetrievalAugmentor.augment 的五段流水
摄取线讲完了。剩下的全是在线部分。
6.1 入口在哪
RetrievalAugmentor 接口只有一个方法 augment(AugmentationRequest)(langchain4j-core/src/main/java/dev/langchain4j/rag/RetrievalAugmentor.java:23)。AI Service 在组装消 息时调用它(langchain4j/src/main/java/dev/langchain4j/service/DefaultAiServices.java:218-229):先用当前 UserMessage、SystemMessage、chat memory、InvocationContext 拼出一个 rag.query.Metadata,再拿返回的 augmentationResult.chatMessage() 顶替原始用户消息。AI Service 的整体装配见 02-ai-services。
6.2 五个插槽的装配线
UserMessage
│ Query.from(text, metadata)
v
Query ──①QueryTransformer──> Query, Query, …
│ ②QueryRouter:每个 Query 选一组 ContentRetriever
v
③ 多个 retriever 并发 retrieve()
v
Map<Query, Collection<List<Content>>>
│ ④ContentAggregator:融合 / 重排 / 截断
v
List<Content> ──⑤ContentInjector──> 新 UserMessage
augment 方法本体只有 6 行实质逻辑(langchain4j-core/src/main/java/dev/langchain4j/rag/DefaultRetrievalAugmentor.java:136-159):
Query originalQuery = Query.from(queryText, augmentationRequest.metadata());
Collection<Query> queries = queryTransformer.transform(originalQuery);
Map<Query, Collection<List<Content>>> queryToContents = process(queries);
List<Content> contents = contentAggregator.aggregate(queryToContents);
ChatMessage augmentedChatMessage = contentInjector.inject(contents, chatMessage);
一行一个插槽,毫无分支。分支全被藏进了 process。
"五段流水"和导读地图里的"四个可换的环节"说的是同一件事。 DefaultRetrievalAugmentor 身上只有四个可替换字段——queryTransformer / queryRouter / contentAggregator / contentInjector(:109-112,executor 不算编排环节);第 ③ 段的 ContentRetriever 不是它的字段,是 QueryRouter 在运行时交出来的。按字段数它是四步,按流水段数是五段。
顺带一提:非 UserMessage 的消息类型会在开头直接抛 IllegalArgumentException(:140-144)——这条流水线只处理用户提问。
6.3 并发扇出:process / retrieveFromAll / join
process(:161-187)按"查询数 × 检索器数"分三种走法:
| 情况 | 走法 | 用不用线程池 | 行号 |
|---|---|---|---|
| 1 个 query + 1 个 retriever | 当前线程直接 retrieve | 不用 | :165-168 |
| 1 个 query + N 个 retriever | retrieveFromAll(...).join() | 用 | :169-171 |
| N 个 query | 每个 query 先异步 route,再 thenCompose 扇出 | 用 | :175-183 |
| 0 个 query 或 0 个 retriever | 返回 emptyMap(),后续全部空转 | — | :172-174、:184-186 |
retrieveFromAll(:189-200)是标准的 supplyAsync + allOf + join 三段式:每个 retriever 一个 future,全部完成后按原顺序收集成 Collection<List<Content>>。join(:202-212)对多 query 的场景再做一次同样的汇聚,把 Map<Query, CompletableFuture<...>> 拍平成 Map<Query, Collection<List<Content>>>。
默认线程池是刻意调过的(:127-133 · createDefaultExecutor):
return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 1, SECONDS, new SynchronousQueue<>());
这是 Executors.newCachedThreadPool() 的改版,唯一的差别是 keepAlive 从 60 秒压到 1 秒(类 javadoc :99 明确说明)。理由不难推:RAG 扇出是短促突发型负载,线程留 60 秒是浪费。
6.4 为什么中间结果是 Map<Query, Collection<List<Content>>>
这个三层嵌套类型看着别扭,但它是后面能做 RRF 融合的前提:
- 最外层
Map的 key —— 哪个查询检出来的(重排时要知道拿哪个 query 去打分); - 中间的
Collection—— 哪个检索器检出来的(每个 retriever 一份独立排名); - 最内层的
List—— 该检索器内部的排序(下标即 rank)。
丢掉任何一层,排名信息就没了。
7. 四个可换插槽的现成实现
这里讲的是 §6.2 那条流水线上 DefaultRetrievalAugmentor 直接持有的四个字段;第 ③ 段的 ContentRetriever 单独放 §8 讲。
每个插槽都有一个"什么都不做"的默认实现 + 一到两个"用 LLM 做聪明事"的高级实现。默认值在构造函数里用 getOrDefault 兜底(DefaultRetrievalAugmentor.java:120-124),唯独 queryRouter 是 ensureNotNull —— 因为检索器必须由你提供,框架没法替你猜。
| 插槽 | 默认实现 | 高级实现 | 高级实现干什么 |
|---|---|---|---|
QueryTransformer | DefaultQueryTransformer | CompressingQueryTransformer | 把对话历史 + 新问题压成一句自足的查询 |
ExpandingQueryTransformer | 让 LLM 生成 n 个同义改写(默认 3) | ||
QueryRouter | DefaultQueryRouter | LanguageModelQueryRouter | 让 LLM 从带描述的数据源里选编号 |
ContentAggregator | DefaultContentAggregator | ReRankingContentAggregator | 用 ScoringModel(如 Cohere rerank)重排并按分截断 |
ContentInjector | DefaultContentInjector | —(自己实现) | — |
7.1 QueryTransformer:改写查询
DefaultQueryTransformer.transform 就一句 return singletonList(query)(langchain4j-core/src/main/java/dev/langchain4j/rag/query/transformer/DefaultQueryTransformer.java:28)。原样透传。
CompressingQueryTransformer(langchain4j-core/src/main/java/dev/langchain4j/rag/query/transformer/CompressingQueryTransformer.java:72 · transform)解决的是"用户说'那它多少钱?'——'它'指什么只有上文知道"。它把 query.metadata().chatMemory() 格式化成 User: ... / AI: ... 的对话文本,连同新问题丢给 LLM 重写成自足查询。两个细节:
- 没有 chat memory 时直接跳过,不白花一次 LLM 调用(
:75-78); - 格式化时丢弃带 tool 调用的 AiMessage(
format(ChatMessage)在hasToolExecutionRequests()时返回 null,:99-101)——工具调用的中间态对查询改写没有帮助。工具调用本身见 03-tool-calling。
ExpandingQueryTransformer(langchain4j-core/src/main/java/dev/langchain4j/rag/query/transformer/ExpandingQueryTransformer.java:76)反过来,让 LLM 生成 n 个措辞不同、语义相同的版本(DEFAULT_N = 3,:47)。它的 prompt 明确要求"每行一个、不要编号不要连字符"(:37-46),解析就是简单的 split("\n") + 过滤空行(:94-97)——约定优于解析,不做容错。
7.2 QueryRouter:选数据源
DefaultQueryRouter.route 无条件返回构造时给的全部检索器(langchain4j-core/src/main/java/dev/langchain4j/rag/query/router/DefaultQueryRouter.java:40-42)。
LanguageModelQueryRouter(langchain4j-core/src/main/java/dev/langchain4j/rag/query/router/LanguageModelQueryRouter.java:98 · route)在构造时就把 Map<ContentRetriever, String> 编成 1: 描述\n2: 描述 的选项文本(:73-89),运行时让 LLM 只回编号,parse 按逗号切开转 int 查表(:127-133)。
它的亮点是 FallbackStrategy 三选一(:139-155),因为"LLM 返回了非数字"是必然会发生的:
| 策略 | 行为 | 后果 |
|---|---|---|
DO_NOT_ROUTE(默认) | 返回 emptyList() | 整个 RAG 静默跳过,原样问模型 |
ROUTE_TO_ALL | 返回全部检索器 | 退化成 DefaultQueryRouter |
FAIL | 重新抛出异常 | 整次调用失败 |
默认选静默降级而非失败——这条选择和 04-structured-output-and-guardrails 里护栏的"宁可拦住"取向正好相反,值得注意。
7.3 ContentAggregator:RRF 融合
它要解决的小问题: 3 个查询 × 2 个检索器 = 6 份各自排好序的列表,但各家的分数不可比(向量库的余弦相似度和搜索引擎的相关度不是一个量纲)。怎么合成一份?
思路: 不看分数,只看名次。这就是 Reciprocal Rank Fusion(倒数排名融合):每个文档在每份列表里的贡献是 1/(k+rank),把所有贡献加起来排序。
int rank = i + 1;
double newScore = currentScore + 1.0 / (k + rank);
langchain4j-core/src/main/java/dev/langchain4j/rag/content/aggregator/ReciprocalRankFuser.java:56-57 · ReciprocalRankFuser.fuse。k 默认 60(:30),这个值来自经验研究:k 越大越抹平名次差异,越小越放大头部(javadoc :38-45 有完整说明)。
DefaultContentAggregator.aggregate 分两阶段跑同一个融合器(langchain4j-core/src/main/java/dev/langchain4j/rag/content/aggregator/DefaultContentAggregator.java:57-64):
阶段 1:对每个 query,把它的多个 retriever 结果融合成一份
阶段 2:把各 query 的那一份再融合成最终一份
为什么要分两阶段而不是一次性 fuse 全部?因为一次性会让"被更多 query 命中"和"被更多 retriever 命中"混为一谈。分阶段后,每个 query 先在内部归一化,再平权参与第二轮。类 javadoc 给了一个具体例子(:32-49)。
ReRankingContentAggregator(langchain4j-core/src/main/java/dev/langchain4j/rag/content/aggregator/ReRankingContentAggregator.java:93 · aggregate)先做完一模一样的两阶段 RRF(:103-106),再拿 ScoringModel.scoreAll 对融合结果重新打分、按 minScore 过滤、按 maxResults 截断(:125-144 · reRankAndFilter)。
它有一个必须知道的约束:ScoringModel 只能对一个 query 打分。多 query 时必须提供 querySelector,否则默认选择器直接抛异常并在消息里告诉你该怎么办(:50-61 · DEFAULT_QUERY_SELECTOR)。这 是一个"不猜,报错并给出路"的好例子。
重排后的分数写回 Content 的 ContentMetadata.RERANKED_SCORE(:141)。ContentMetadata 是个只有三个值的枚举:SCORE、RERANKED_SCORE、EMBEDDING_ID(langchain4j-core/src/main/java/dev/langchain4j/rag/content/ContentMetadata.java:3-7)。
7.4 ContentInjector:拼进 prompt
DefaultContentInjector 的模板就是把检索结果追加在用户消息后面(langchain4j-core/src/main/java/dev/langchain4j/rag/content/injector/DefaultContentInjector.java:47-52):
{{userMessage}}
Answer using the following information:
{{contents}}
inject(:79-92)有两条值得注意的行为:
contents为空时原样返回消息(:80-82),不会留下一句尴尬的 "Answer using the following information:" 加空白。这也是 §7.2 里DO_NOT_ROUTE能静默降级的前提。- 通过
metadataKeysToInclude可以把段的 metadata 一起写进 prompt,格式是content: 正文\nkey: value(:133-137)。给来源信息的常用手法:把file_name或url加进去,模型就能引用出处。
8. EmbeddingStoreContentRetriever:三个参数全部可动态化
这是最常用的检索器,也是 §6.2 流水线里第 ③ 段的主力。检索逻辑本身很直白(langchain4j-core/src/main/java/dev/langchain4j/rag/content/retriever/EmbeddingStoreContentRetriever.java:225-246 · retrieve):把 query 文本向量化 → 组 EmbeddingSearchRequest → embeddingStore.search → 把每个 match 包成 Content,顺手把 SCORE 和 EMBEDDING_ID 写进 content metadata(:240-244)。
真正的设计点在于:三个检索参数存的不是值,而是 Function<Query, T>。
public static final Function<Query, Integer> DEFAULT_MAX_RESULTS = (query) -> 3;
public static final Function<Query, Double> DEFAULT_MIN_SCORE = (query) -> 0.0;
public static final Function<Query, Filter> DEFAULT_FILTER = (query) -> null;
:59-61。静态值的 builder 方法(maxResults(Integer) 等)只是把常量包成 lambda(:154-173)。
这么做的价值在于 Query 身上挂着 rag.query.Metadata,里面有:
| 能拿到什么 | 方法 | 典型用途 |
|---|---|---|
| 当前用户/会话 ID | metadata().chatMemoryId()(langchain4j-core/src/main/java/dev/langchain4j/rag/query/Metadata.java:68) | 多租户隔离:按用户 ID 生成 filter |
| 完整对话历史 | metadata().chatMemory()(:76) | 依上下文调整召回数量 |
| 本次调用的系统消息 | metadata().systemMessage()(:60) | 按角色调整策略 |
| 调用级参数 | metadata().invocationContext()(:83) | 传业务侧的额外维度 |
于是"这个用户只能看到自己部门的文档"变成一行 dynamicFilter(q -> metadataKey("dept").isEqualTo(deptOf(q.metadata().chatMemoryId()))),不需要为每个租户建一个 retriever 实例。
一个坑: maxResults(Integer) 的合法性校验写在 lambda 内部(:156),minScore 同理(:163)。也就是说传 maxResults(-1) 在 build 时不报错,要等到第一次 retrieve 才抛。
另一个开箱即用的 检索器是 WebSearchContentRetriever(langchain4j-core/src/main/java/dev/langchain4j/rag/content/retriever/WebSearchContentRetriever.java:38 · retrieve):把 query 文本当搜索词丢给 WebSearchEngine,默认取 5 条(:30),结果直接 toTextSegments() 转 Content。它和 EmbeddingStoreContentRetriever 一起挂在 DefaultQueryRouter 下,就得到"本地知识库 + 联网"的混合检索,再由 RRF 融合——这正是 §6.4 那个三层类型存在的意义。
向量库的接口契约(以 InMemory 为参照)
本章不讲各家向量库实现,只用 InMemoryEmbeddingStore 说明检索器对 store 的期待(langchain4j/src/main/java/dev/langchain4j/store/embedding/inmemory/InMemoryEmbeddingStore.java:164 · search):
- 先按
filter过滤(:171),过滤只对TextSegment类型的 embedded 生效,测的是它的Metadata(:274-280·matchesFilter); - 再算余弦相似度并归一到
[0,1]的RelevanceScore(:176-177); - 按
minScore卡掉、按maxResults截断、分数降序返回。
EmbeddingSearchRequest 的默认值是 maxResults=3、minScore=0.0、filter=null(langchain4j-core/src/main/java/dev/langchain4j/store/embedding/EmbeddingSearchRequest.java:52-54)。各家向量库实现要做的,就是把 Filter 对象翻译成自己的原生过滤表达式。EmbeddingStore 接口本身长什么样、可选能力怎么用 default 方法表达,见 01-core-abstractions §7。
9. 元数据过滤:一套抽象语法,各家自己翻译
Filter 的类型体系与"一棵中立的树、两种消费方式(翻译成原生语法 / 直接 test() 求值)"的设计意图,在 01-core-abstractions §7.3 讲过;本节只讲它在检索线里怎么被用——重点是最后那一小节:让 LLM 替用户写 filter。
Filter 接口只有一个 boolean test(Object) 加三个静态组合方法 and / or / not(langchain4j-core/src/main/java/dev/langchain4j/store/embedding/filter/Filter.java:52-72)。它是一棵表达式树,不是查询字符串。
| 类别 | 实现类 | 位置 |
|---|---|---|
| 相等 | IsEqualTo、IsNotEqualTo | .../filter/comparison/ |
| 大小 | IsGreaterThan、IsGreaterThanOrEqualTo、IsLessThan、IsLessThanOrEqualTo | 同上 |
| 集合 | IsIn、IsNotIn | 同上 |
| 字符串包含 | ContainsString | 同上 |
| 逻辑 | And、Or、Not | .../filter/logical/ |
构造用流式 DSL metadataKey("year").isGreaterThan(2020)(langchain4j-core/src/main/java/dev/langchain4j/store/embedding/filter/MetadataFilterBuilder.java:36),每种比较都为 String / UUID / int / long / float / double 各重载一遍——和 §2 那张 Metadata 值类型白名单严格对应。
test 的实现都很保守,比如 ContainsString:非 Metadata 对象返回 false、key 不存在返回 false、值不是 String 也返回 false(langchain4j-core/src/main/java/dev/langchain4j/store/embedding/filter/comparison/ContainsString.java:34-46)。"不确定就不匹配",不做隐式类型转换。
9.1 让 LLM 写 filter(self-querying)
用户不会写 metadataKey("product").isEqualTo("iPhone"),他只会问"我手机屏幕怎么调亮?"。langchain4j-embedding-store-filter-parser-sql 模块把这中间的一步交给 LLM。
思路是借 SQL 当中间语言:
用户问题 ─┐
├─> LLM ─> SELECT * FROM documentation WHERE product = 'iPhone'
TableDefinition(把 metadata 描述成建表语句) │
│ SqlFilterParser(JSqlParser)
v
metadataKey("product").isEqualTo("iPhone")
为什么绕 SQL?因为 text-to-SQL 是模型见过最多的结构化生成任务之一,还有 SQLCoder 这类专门微调的小模型可用——类 javadoc 直接推荐了(embedding-store-filter-parsers/langchain4j-embedding-store-filter-parser-sql/src/main/java/dev/langchain4j/store/embedding/filter/builder/sql/LanguageModelSqlFilterBuilder.java:68-73)。
build(Query) 的容错是三级递降(:129-151 + :164 · fallback):
- 直接
sqlFilterParser.parse(cleanedSql); - 失败 →
extractSelectStatement从模型的啰嗦回复里抠出 SELECT 语句再 parse; - 还失败 → 返回
null,即不过滤(:168的 "Cannot extract SQL, giving up")。
第 3 步同样是"降级而非失败":过滤丢了,搜索范围变大,但用户仍能拿到答案。源码里还留了一段 TODO 列出未来可选策略(反馈错误重试、预定义 filter、部分 filter),:144-148。
用法就是把它接到上一节的动态 filter 上:.dynamicFilter(sqlFilterBuilder::build)。这个模块标了 @Experimental(:82)。
10. langchain4j-easy-rag:它替你做了哪三个默认选择
前面那么多插槽,新手一个都不想选。langchain4j-easy-rag 模块的全部作用,就是往 classpath 上塞 SPI 实现,让 EmbeddingStoreIngestor.ingest(docs, store) 这样的零配置调用能跑起来。
模块里只有一个 Java 类:
public DocumentSplitter create() {
TokenCountEstimator tokenCountEstimator = new HuggingFaceTokenCountEstimator();
return DocumentSplitters.recursive(300, 30, tokenCountEstimator);
}
langchain4j-easy-rag/src/main/java/dev/langchain4j/data/document/splitter/recursive/RecursiveDocumentSplitterFactory.java:12-15。
它做的三个选择,都是通过 pom 依赖 + SPI 注册完成的:
| 选择 | 选了什么 | 来自 |
|---|---|---|
| 解析器 | ApacheTikaDocumentParser(格式通吃) | 依赖 langchain4j-document-parser-apache-tika,该模块自带 DocumentParserFactory 的 SPI 注册 |
| 切分器 | recursive(300 tokens, 30 tokens overlap),按 HuggingFace tokenizer 计数 | 本模块的 RecursiveDocumentSplitterFactory + META-INF/services/dev.langchain4j.spi.data.document.splitter.DocumentSplitterFactory |
| 向量模型 | BgeSmallEnV15QuantizedEmbeddingModel(量化的本地 ONNX 模型,不联网) | 依赖 langchain4j-embeddings-bge-small-en-v15-q,该模块注册了 EmbeddingModelFactory |
三个选择的共同取向很明确:默认路径不需要任何外部 API key。向量模型是打包进 jar 的本地 ONNX 量化模型,解析器和切分器纯本地。加一个依赖就能跑通整条摄取线。
代价是这些默认值都是为英文调的(bge-small-en、OpenNLP 的英文句子模型)。中文场景至少要换掉向量模型和句子切分模型。
11. 巧妙之处(可借鉴)
其一,把"要不要并发"降级成一个 if。 process 在"1 个 query + 1 个 retriever"这条最常见的路径上根本不碰线程池,直接同线程调用(DefaultRetrievalAugmentor.java:165-168)。绝大多数用户的绝大多数调用零并发开销,只有真的扇出时才付代价。
其二,线程池参数是被想过的。 抄 newCachedThreadPool 但把 keepAlive 从 60s 改成 1s(:127-133),并在类 javadoc 里写清楚改了什么、为什么(:99)。这是"默认值也要解释"的好例子。
其三,融合只用名次不用分数。 RRF 让"向量库的相似度"和"搜索引擎的相关度"能同台竞争(ReciprocalRankFuser.java:48),而两阶段融合(先 per-query 再跨 query)避免了检索器数量多的 query 天然占优。
其四,降级切分只对"塞不下的那块"降级。 HierarchicalDocumentSplitter.split 的循环让绝大多数段落保持段落粒度,只有超长段落才被递归拆碎(:171)。整篇文档的语义边界因此得到最大保留。
其五,overlap 保证是整句。 固定用句子切分器倒序取重叠(:195-213),而不是简单地"取最后 N 个字符"。这让重叠区始终是可读的完整句子。
其六,几处关键失败点都选"静默降级"而非抛异常。 路由失败 → 不路由(LanguageModelQueryRouter.java:107-109);SQL 解析失败 → 不过滤(LanguageModelSqlFilterBuilder.java 的 fallback);检索为空 → 注入器原样返回消息(DefaultContentInjector.java:79-81)。RAG 是"锦上添花"而非"必要条件",这个取向是自洽的。
12. 边界与局限
-
摄取线没有增量/去重。
EmbeddingStoreIngestor.ingest每次都是全量 embed +addAll(:198-202),同一份文档灌两次就会在库里出现两份。"文档更新了怎么办"框架不管。 -
摄取线是同步单线程的。 没有并发,没有批次控制,没有失败重试。
embedAll一次把所有段丢给模型;大目录摄取要自己切批。检索线才有Executor。 -
IngestionResult信息很薄。 只有TokenUsage,没有"成功几段/跳过几个文件"这类统计(IngestionResult.java:8-22)。 -
重排必须挑一个 query。
ReRankingContentAggregator无法对"每个 content 用检出它的那个 query"打分,只能全体对同一个 query 打分;javadoc 承认了这个限制并建议自定义实现(ReRankingContentAggregator.java:29-33)。 -
ExpandingQueryTransformer的解析没有容错。 按行切分 + 过滤空行(:94-97),模型如果输出了编号或前言,那些噪声会原样变成查询。 -
参数校验时机偏晚。
EmbeddingStoreContentRetriever的maxResults/minScore校验在 lambda 内部,build 期不报错(:154-166)。 -
默认值面向英文。 句子切分模型和 easy-rag 的向量模型都是英文的(
DocumentBySentenceSplitter.java:83、easy-rag pom 的bge-small-en-v15-q)。 -
本章不覆盖向量库实现。 根
pom.xml的<!-- embedding / chat memory stores -->段共 21 个langchain4j-<vendor>模块(pom.xml:70-90),全仓非测试类implements EmbeddingStore的实现 27 个——同一个模块可以放多个实现,所以两个数字对不齐是正常的。它们各自把Filter翻译成原生表达式、各自处理连接与索引。契约见 §8 末尾,接口本身见 01-core-abstractions §7。
13. 横向对比与本组其它章
| 关切 | 本章的位置 | 去哪儿看 |
|---|---|---|
| 模型接口、消息模型、100+ 集成怎么共存 | RAG 用到的 ChatModel / EmbeddingModel / ScoringModel 都出自那套抽象,EmbeddingStore 与 Filter 的接口设计也在那一章 | 01-core-abstractions |
| RAG 怎么被自动挂到一个 Java 接口上 | AiServices.builder().contentRetriever(...) 背后就是本章的 DefaultRetrievalAugmentor | 02-ai-services |
| 工具调用 | CompressingQueryTransformer 会主动丢弃带工具调用的消息 | 03-tool-calling |
| 输出解析与护栏 | 护栏遇错倾向拦截,RAG 遇错倾向降级——两种相反取向 | 04-structured-output-and-guardrails |
| 多智能体编排 | RAG 常作为 agent 的一个能力被编排进循环 | 06-agentic |
14. 代码地图(导航索引)
按符号名 grep 比按行号更抗上游漂移。路径相对克隆根。
| 主题 | 文件路径(相对克隆根) | 符号名 |
|---|---|---|
| 检索线总编排 | langchain4j-core/src/main/java/dev/langchain4j/rag/DefaultRetrievalAugmentor.java | augment、process、retrieveFromAll、join、createDefaultExecutor |
| 检索线接口 | langchain4j-core/src/main/java/dev/langchain4j/rag/RetrievalAugmentor.java | RetrievalAugmentor.augment |
| AI Service 挂载点 | langchain4j/src/main/java/dev/langchain4j/service/DefaultAiServices.java | context.retrievalAugmentor.augment(约 :218-229) |
| 查询与查询上下文 | langchain4j-core/src/main/java/dev/langchain4j/rag/query/Query.java、.../query/Metadata.java | Query.from、Metadata.chatMemoryId、Metadata.chatMemory |
| 查询改写 | langchain4j-core/src/main/java/dev/langchain4j/rag/query/transformer/ | DefaultQueryTransformer、CompressingQueryTransformer、ExpandingQueryTransformer |
| 查询路由 | langchain4j-core/src/main/java/dev/langchain4j/rag/query/router/ | DefaultQueryRouter、LanguageModelQueryRouter、FallbackStrategy |
| 检索器 | langchain4j-core/src/main/java/dev/langchain4j/rag/content/retriever/ | ContentRetriever.retrieve、EmbeddingStoreContentRetriever、WebSearchContentRetriever |
| 融合与重排 | langchain4j-core/src/main/java/dev/langchain4j/rag/content/aggregator/ | ReciprocalRankFuser.fuse、DefaultContentAggregator.aggregate、ReRankingContentAggregator.reRankAndFilter |
| 内容注入 | langchain4j-core/src/main/java/dev/langchain4j/rag/content/injector/DefaultContentInjector.java | inject、DEFAULT_PROMPT_TEMPLATE、metadataKeysToInclude |
| 检索结果载体 | langchain4j-core/src/main/java/dev/langchain4j/rag/content/ | Content、ContentMetadata(SCORE/RERANKED_SCORE/EMBEDDING_ID) |
| 摄取线总编排 | langchain4j-core/src/main/java/dev/langchain4j/store/embedding/EmbeddingStoreIngestor.java | ingest、loadDocumentSplitter、loadEmbeddingModel |
| 摄取结果 | langchain4j-core/src/main/java/dev/langchain4j/store/embedding/IngestionResult.java | IngestionResult.tokenUsage |
| 数据模型 | langchain4j-core/src/main/java/dev/langchain4j/data/document/、.../data/segment/、.../data/embedding/ | Document、Metadata.SUPPORTED_VALUE_TYPES、TextSegment、Embedding |
| 加载与解析胶水 | langchain4j-core/src/main/java/dev/langchain4j/data/document/DocumentLoader.java | DocumentLoader.load |
| 文件系统加载器 | langchain4j/src/main/java/dev/langchain4j/data/document/loader/ | FileSystemDocumentLoader、ClassPathDocumentLoader、UrlDocumentLoader、DEFAULT_DOCUMENT_PARSER |
| 默认解析器 | langchain4j/src/main/java/dev/langchain4j/data/document/parser/TextDocumentParser.java | TextDocumentParser.parse、BlankDocumentException |
| 切分器基类 | langchain4j/src/main/java/dev/langchain4j/data/document/splitter/HierarchicalDocumentSplitter.java | split(Document)、overlapFrom、estimateSize、createSegment |
| 切分缓冲区 | langchain4j/src/main/java/dev/langchain4j/data/document/splitter/SegmentBuilder.java | hasSpaceFor、append、prepend |
| 切分器工厂 | langchain4j/src/main/java/dev/langchain4j/data/document/splitter/DocumentSplitters.java | DocumentSplitters.recursive |
| 元数据过滤 | langchain4j-core/src/main/java/dev/langchain4j/store/embedding/filter/ | Filter.test、MetadataFilterBuilder.metadataKey、And/Or/Not、IsIn、ContainsString |
| 自然语言转 filter | embedding-store-filter-parsers/langchain4j-embedding-store-filter-parser-sql/src/main/java/dev/langchain4j/store/embedding/filter/builder/sql/ | LanguageModelSqlFilterBuilder.build、fallback、TableDefinition |
| easy-rag 的默认选择 | langchain4j-easy-rag/src/main/java/dev/langchain4j/data/document/splitter/recursive/RecursiveDocumentSplitterFactory.java | RecursiveDocumentSplitterFactory.create |
| 参照用的向量库 | langchain4j/src/main/java/dev/langchain4j/store/embedding/inmemory/InMemoryEmbeddingStore.java | search、matchesFilter |