检索与RAG

本章配套可运行示例：demo/demo_rag.py。demo 把书屋读书笔记硬编码在文件里，从切割、入库、检索一路跑到组装问答链；本机没启动 Ollama 时会自动降级，只演示检索部分并打印提示。

本章你将学会

• as_retriever()：把第 07 章建好的向量库包装成标准 Runnable，从此检索也能用 | 接进流水线
• 检索参数三件套：k 控制取几条、score_threshold 设相似度及格线、filter 按元数据先筛一遍
• MMR 检索：相关性和多样性一手抓，别让同一本书的五段笔记挤满上下文
• 组装经典 RAG 链：RunnableParallel 一手递问题、一手取笔记 → 提示词 → 模型 → 解析器
• 来源溯源：答案和「它引用了哪些文档」一起返回，答案自带引用列表
• 多查询检索：让模型把一个问题改写成多种说法，分头检索再合并去重
• 一张检索优化地图：答案不对时，从切块、查询改写到重排，按阶段排查

前两章我们把弹药备齐了：第 06 章把书屋的读书笔记切成了小块，第 07 章把小块变成了向量、存进了 InMemoryVectorStore。但库躺在那里不会自己干活——本章把「检索」和「生成」这两棒焊在一起，组装完整的 RAG（Retrieval-Augmented Generation，检索增强生成）。

比喻：大模型平时答题是闭卷考试——课本是训练时背下来的，拾光书屋的借阅规则上周刚改，它不可能知道。RAG 把考试改成开卷：答题前先把相关笔记从书架上抽出来摊在桌上（检索），再让模型照着笔记答题（生成）。模型不需要重新背书，书屋的笔记改了，重灌一次库就生效。

一条主线看全貌：

用户问题（str）
   │
   ▼
检索器 retriever ──► 相关文档片段 list[Document]
   │                        │
   ▼                        ▼  format_docs：拼成一段文本
RunnableParallel ──► {"context": 拼好的笔记, "question": 原问题}
   │
   ▼
提示词模板 ──► 模型 ──► 解析器 ──► 答案（顺带把引用的文档也交出来）

技术基线：Python + LangChain 1.x（langchain + langchain-ollama），本地 Ollama 跑 qwen3:4b（对话）和 nomic-embed-text（嵌入）。没装的话回 01 章看环境准备。

后面所有示例共用一份起步代码（嵌入与入库的细节在第 07 章，这里直接用）：

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_ollama import OllamaEmbeddings

embeddings = OllamaEmbeddings(model="nomic-embed-text")
vectorstore = InMemoryVectorStore(embedding=embeddings)  # 07 章的成果

notes = [
    Document(
        page_content="《夜航船》是明代张岱编的类书式笔记，条目短小，适合通勤时随手翻几页。",
        metadata={"book": "夜航船", "category": "古籍"},
    ),
    Document(
        page_content="拾光书屋营业时间为每天 10:00 到 22:00。会员一次最多借 5 本书。",
        metadata={"book": "店内公告", "category": "公告"},
    ),
]
vectorstore.add_documents(notes)

1. as_retriever：给向量库发一张 Runnable 工牌

比喻：第 07 章我们直接调 vectorstore.similarity_search(...)，相当于每次都自己冲进书库翻架子上——能用，但书库永远只是个「仓库」。as_retriever() 给它发一张工牌，聘出一位专职「取书员」：你递问题，他还你一沓相关笔记。关键是这位取书员是标准 Runnable——第 03 章学的 | 拼装、invoke/stream/batch 全套驾驶技巧，对他统统有效。

作用：把向量库（或任何实现了检索接口的存储）包装成 VectorStoreRetriever——输入一个问题字符串，输出 list[Document]。从此检索环节可以名正言顺地进 LCEL 流水线。

签名：

def as_retriever(self, **kwargs: Any) -> VectorStoreRetriever: ...

两个最常用的关键字参数（其余见第 2、3 节）：

• search_type：检索策略，"similarity"（默认，纯相似度）、"mmr"、"similarity_score_threshold"
• search_kwargs：传给底层搜索函数的参数字典，比如 {"k": 3}

最小示例：

retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

docs = retriever.invoke("哪本书适合通勤时读？")
for doc in docs:
    print(f"[{doc.metadata['book']}] {doc.page_content[:30]}…")

拿到 Runnable 工牌意味着第 04 章的技巧直接复用：retriever.batch(["问题一", "问题二"]) 一次查多个问题；retriever | format_docs 把它接进管道（第 4 节）。

常见坑：

• 输入是字符串，不是 dict。retriever.invoke("问题") 才对；传 {"question": "问题"} 会在嵌入那一步炸掉——BaseRetriever 的输入类型就是 str，想在链里喂 dict 得先拆出问题字段（第 5 节末尾有变体写法）。
• 输出是 list[Document]，不是拼好的文本。直接把 retriever 接在提示词模板前面，模板变量里灌进的是 Document 列表——中间必须有个 format_docs 转成字符串（第 4 节）。
• as_retriever() 返回的是向量库的「视图」而不是拷贝：库里后续 add_documents 新增的内容，同一个 retriever 立刻能检到，不需要重新包一次。

2. 检索参数三件套：k、score_threshold、filter

比喻：给取书员交代任务的三个口头禅——「一次抱几本」（k）、「分数不够的别上桌」（score_threshold）、「只在小说区找」（filter）。三个都写在 search_kwargs 里。

2.1 k：一次抱几本

retriever = vectorstore.as_retriever(search_kwargs={"k": 6})

k 决定每次检索返回几条文档，默认 4。它是一根两头都不讨好的撬棍：

• 调太小：上下文缺料，答案缺胳膊少腿——该引用的公告没被捞上来；
• 调太大：无关内容挤进提示词，把模型带偏（噪声稀释重点），token 开销也跟着涨。

经验值：从 3~5 起步，打印检索结果肉眼看质量，再决定往哪边调。

2.2 score_threshold：设一条相似度及格线

每条候选都有一个相似度分数。score_threshold 的意思是：低于这条及格线的，别进答案。

retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"k": 4, "score_threshold": 0.5},
)

注意一个大坑（实测）：这条路的底层调用是「把分数归一化到 0~1 再过滤」，而 InMemoryVectorStore 没有实现归一化函数，直接跑会抛 NotImplementedError。两条出路：

1. 换支持 relevance score 的向量库（FAISS、Chroma 等）；
2. 手写阈值过滤——用 similarity_search_with_score 拿到 (文档, 分数) 对，自己筛。书屋的真实工程就是这么干的：

def retrieve_above_threshold(
    query: str, k: int = 6, threshold: float = 0.4
) -> list[Document]:
    """先多抓一些候选，再按分数过滤。"""
    hits: list[tuple[Document, float]] = vectorstore.similarity_search_with_score(query, k=k)
    return [doc for doc, score in hits if score >= threshold]

更麻烦的是，分数刻度没有行业标准：InMemoryVectorStore 返回的是余弦相似度（越大越相关），有的库返回的是距离（越小越好）；不同嵌入模型的分数分布也各不相同——nomic-embed-text 对相关中文文本的余弦一般在 0.4~0.7 一带。所以阈值别抄别人的数字，先看自己语料的分数分布：

for doc, score in vectorstore.similarity_search_with_score("营业时间是几点", k=5):
    print(f"{score:.3f}  [{doc.metadata['book']}]")

2.3 filter：先锁定书架，再往里找

入库时给每条文档打的 metadata，检索时就能用来预先筛掉整片无关区域：

retriever = vectorstore.as_retriever(
    search_kwargs={
        "k": 3,
        "filter": lambda doc: doc.metadata.get("category") == "小说",
    }
)
docs = retriever.invoke("一个晚上能读完的故事")

又一个移植大坑（实测）：InMemoryVectorStore 的 filter 吃的是函数 Callable[[Document], bool]，不是 dict！传 {"category": "小说"} 会报 TypeError: 'dict' object is not callable。而 FAISS、Chroma 这类库恰恰相反，吃的是 dict。同一份代码换向量库，filter 的写法必须跟着改——这是 RAG 代码迁移时最阴的一脚。

filter 还有一个高频用途——多文档问答：书屋支持读者上传多份笔记，每份入库时把「原始文件名」写进 metadata，之后 filter 指定文件名，就实现了「只在这一篇里问答」。

常见坑：

• score_threshold 只有搭配 search_type="similarity_score_threshold" 才生效，光在 search_kwargs 里塞一个阈值参数，走默认 similarity 检索时会被无视。
• 三个参数都住在 search_kwargs 这个 dict 里，别把它们平铺在 as_retriever() 的参数位上。
• filter 的形态跟着向量库走（函数 or dict），换库必查文档，报错信息不一定能一眼看出是 filter 的锅。

3. MMR：既要最相关，也要不重样

比喻：自助餐打菜。普通相似度检索是「哪盘最像我最爱的那口就拿哪盘」，结果端回来五碗几乎一样的汤；《山月记》的笔记被切成五块后，普通检索很可能把最相似的前四块全给了它，翻来覆去都是李征变虎，真正该上桌的《城南旧事》一块都挤不进来。MMR（Maximal Marginal Relevance，最大边际相关性）换了个打法：先拿最对味的那盘，之后每一次伸手前都问自己两个问题——这盘好不好吃（相关性）？跟我已经端的是不是同一道菜（多样性）？

作用：在「每条都尽量相关」和「彼此尽量不重复」之间找平衡，让有限的上下文窗口装下更宽的信息面。

签名：

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.5},
)

• fetch_k：先海选多少条候选（默认 20），MMR 从这个池子里精挑
• lambda_mult：天平上的砝码，1.0 只看相关性（退化成普通检索），0.0 只求多样性，默认 0.5 居中

最小示例：

plain = vectorstore.as_retriever(search_kwargs={"k": 4})
mmr = vectorstore.as_retriever(search_type="mmr", search_kwargs={"k": 4, "fetch_k": 20})

print("普通检索：", [d.metadata["book"] for d in plain.invoke("变虎的故事和老北京的童年")])
print("MMR 检索：", [d.metadata["book"] for d in mmr.invoke("变虎的故事和老北京的童年")])
# 笔记块主题重复时，MMR 结果里的书名明显更分散

常见坑：

• fetch_k 必须 ≥ k，候选池比要取的数量还小，MMR 就无从「挑」起。
• lambda_mult 一味压向多样性，可能把最相关的那条挤出去——「答案要准」的场景别把砝码全押多样性。
• MMR 是向量库端的能力，要求底层实现 max_marginal_relevance_search（InMemoryVectorStore 支持）；不支持的向量库上设 search_type="mmr" 会直接报错。

4. RAG 链组装：retriever | format_docs → prompt | model | parser

比喻：前面攒齐了灶台、锅和食材，这一节点火炒菜。先看不用 LCEL 的「散装版」，看清它别扭在哪：

question = "书屋的营业时间是几点？"

docs = retriever.invoke(question)        # 第 1 步：检索
context = format_docs(docs)              # 第 2 步：拼上下文

qa_chain = prompt | model | StrOutputParser()
answer = qa_chain.invoke({"context": context, "question": question})  # 第 3 步：生成

散装版能跑，但「检索」游离在链外：整条链没法统一 stream/batch，第 12 章的回调观测也照不见检索这一环。经典 LCEL 结构把它焊成一体：

作用：把「检索 → 拼上下文 → 填模板 → 生成 → 解析」装进一条链，一个 invoke 走完全程，并且整体获得流式、批量、回调、可观测能力。

签名（结构等式）：

rag_chain = (
    RunnableParallel(
        context=retriever | format_docs,   # 一手取笔记
        question=RunnablePassthrough(),    # 一手递问题
    )
    | prompt                               # 填模板
    | model                                # 生成
    | StrOutputParser()                    # 取出纯文本
)

最小示例：

from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_ollama import ChatOllama


def format_docs(docs: list[Document]) -> str:
    """把文档列表拼成一段纯文本，塞进模板变量。"""
    return "\n\n".join(doc.page_content for doc in docs)


prompt = ChatPromptTemplate.from_messages([
    ("system",
     "你是拾光书屋的店员，只依据【书屋笔记】回答读者的问题；"
     "笔记里没有的信息就直说「书屋笔记里没有相关内容」，不要编造。\n\n"
     "【书屋笔记】\n{context}"),
    ("human", "{question}"),
])

model = ChatOllama(model="qwen3:4b", temperature=0, reasoning=False)

rag_chain = (
    RunnableParallel(
        context=retriever | format_docs,
        question=RunnablePassthrough(),
    )
    | prompt
    | model
    | StrOutputParser()
)

print(rag_chain.invoke("书屋的营业时间是几点？"))

每一环的「接口形状」心里要有数：

提示词里那句「只依据笔记回答，没有就说不知道」是 RAG 提示词的灵魂——它把「编一个像样的答案」挡在门外，是防幻觉的第一道闸。

常见坑：

• retriever 和 prompt 之间必须站一个 format_docs。retriever 吐 list[Document]，模板要字符串，少了这一环，模板变量里就是一堆 Document 对象的字符串表示。
• qwen3 默认会「思考」，正文里混着 <think>…</think> 标签。ChatOllama(reasoning=False) 直接关掉；想保留思考过程就用 reasoning=True，思考内容会被收进 additional_kwargs，不污染正文。
• 查营业时间这种事实型问答，temperature 放 0 附近——不需要创造性，需要老实。
• 链的输入是裸字符串（问题本身），不是 {"question": ...}——因为 RunnableParallel 的两个分支拿到的都是这个原始输入（下一节细说）。

5. RunnableParallel：一手递问题，一手取笔记

第 4 节的结构里最费解的就是开头那个 RunnableParallel，值得单独拆开看。

比喻：前台收发室。读者的问题递进来，收发室复印两份：一份送去书库（检索、拼文本，产出 context），另一份原样放到桌上（产出 question），最后两样一起打包交给模板。没有这层分身，模板的两个变量就凑不齐。

作用：让「检索拼上下文」和「原样传递问题」两条支路拿到同一份输入并行处理，输出合并成一个 dict——键名恰好对齐模板变量。

签名：

setup = RunnableParallel(
    context=retriever | format_docs,   # 分身一：检索并拼成文本
    question=RunnablePassthrough(),    # 分身二：原样放行
)

偷懒写法：直接把 dict 接进管道，框架自动把它转成 RunnableParallel，效果一字不差（第 03 章讲过的规则）：

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

最小示例（不需要模型，专看数据形状）：

from langchain_core.runnables import RunnableLambda, RunnableParallel, RunnablePassthrough

fan_out = RunnableParallel(
    context=RunnableLambda(lambda q: f"（假设这是检索到的笔记，来自问题：{q}）"),
    question=RunnablePassthrough(),
)
print(fan_out.invoke("营业时间是几点？"))
# {'context': '（假设这是检索到的笔记，来自问题：营业时间是几点？）',
#  'question': '营业时间是几点？'}

常见坑：

• RunnablePassthrough() 就是「原样放行」零件，等价于 lambda q: q，但语义更直白——它存在的意义就是占住 question 这条支路。
• 键名必须和模板变量严格对齐。模板里写 {question}，parallel 里写成 ques，模板校验立刻报错，不会静默通过。
• 两个分支拿到的是同一份完整输入（这里是问题字符串）。如果上游给的是 dict（比如 {"question": ..., "user_id": ...}），context 分支开头接一个取字段的函数即可：

from operator import itemgetter

setup = RunnableParallel(
    context=itemgetter("question") | retriever | format_docs,
    question=itemgetter("question"),
)

• 分支之间不该有先后依赖。真有依赖（比如「先判断要不要检索」）就别硬并行，串行或用条件路由（第 03 章）。

6. 来源溯源：让答案自带引用列表

比喻：论文末尾的引用文献页。没有它，答案再漂亮也只是一段「来路不明的好话」；有了它，读者（和你自己）能一键核对模型有没有瞎编。对书屋来说，这还是「展示多文档上传价值」的关键——答案旁边列出「引用了哪几篇笔记」，读者才敢信。

作用：把「被检索到的原始 Document 列表」和「生成的答案」一起返回。做法：RunnableParallel 的 context 分支不做 format_docs、保留 Document 列表，再用 .assign() 在旁边并行生成答案——答案子链内部才把 context 拼成文本。

签名：

rag_with_sources = RunnableParallel(
    context=retriever,               # 保留原始 Document 列表
    question=RunnablePassthrough(),
).assign(answer=answer_chain)        # 拿着上面的 dict 算答案

最小示例：

answer_chain = (
    RunnablePassthrough.assign(context=lambda x: format_docs(x["context"]))
    | prompt
    | model
    | StrOutputParser()
)

rag_with_sources = RunnableParallel(
    context=retriever,
    question=RunnablePassthrough(),
).assign(answer=answer_chain)

result = rag_with_sources.invoke("会员借书有什么限制？")

print(result["answer"])                     # 答案
for doc in result["context"]:               # 引用列表
    print(f"出处：[{doc.metadata['book']}] {doc.page_content[:30]}…")

result 的形状：

{
    "context": [Document(...), Document(...)],  # 原始文档，溯源用
    "question": "会员借书有什么限制？",
    "answer": "会员一次最多可借 5 本书，借期 14 天……",  # 纯文本答案
}

常见坑：

• context 分支千万别接 format_docs——接了，回来的就是字符串，出处信息（metadata、原文）全没了。拼接动作挪进 answer_chain 里做。
• .assign() 里那个子链拿到的输入，是 RunnableParallel 输出的完整 dict（含 context 和 question 两个键），RunnablePassthrough.assign(context=...) 在保留原字段的基础上覆写 context。
• 检索为空的兜底：context 是空列表时链照跑，模型只能按提示词回答「笔记里没有」。更讲究的做法是先判 len(docs) == 0，空了就切换到一条不带笔记约束的「自由问答链」——书屋的完整工程里就是这么设计双链路的：有笔记走「依据文档回答」，没笔记走「自由回答」，两条链用的是两套提示词。
• 多文档上传场景，入库时把「上传文件名」写进每块的 metadata（{"source": "夜航船笔记.txt", ...}），溯源时把文件名渲染进引用列表——读者一看就知道答案出自自己上传的哪份文件。

7. 多查询检索：一个问题，换几种问法

比喻：图书馆问询台。你说「李征最后怎么了」，馆员不光按字面查一遍，还会顺手用「李征 结局」「变虎 诗人」再查两遍——同一件事的不同说法，命中不同书架上的卡片。这就是多查询检索（multi-query）的全部直觉。

先补一个度量词：召回率（recall）= 检索到的相关内容 / 实际存在的相关内容总数。实际相关 10 条、只捞上来 6 条，召回率 60%。多查询检索治的就是「单一措辞导致召回不足」：读者问「李征最后怎么了」，笔记里写的是「诗人李征最终化作猛虎」，字面重叠太少，一次检索可能漏掉——那就把问题改写成 N 种说法，每种各检一遍，合并去重，网撒开了，漏网之鱼自然少。

手写版（思路本身只有十几行，值得亲手写一次）：

from langchain_core.prompts import PromptTemplate

rewrite_prompt = PromptTemplate.from_template(
    "你是拾光书屋的检索助手。为了在书屋笔记里找到更多相关内容，"
    "把读者的问题改写成 3 种不同的说法：换角度、换用词，但不要改变原意。"
    "每行一个，只输出改写后的问题，不要编号、不要解释。\n"
    "问题：{question}"
)
rewriter = rewrite_prompt | model | StrOutputParser()


def multi_retrieve(question: str, per_query_k: int = 2) -> list[Document]:
    """多路检索 + 去重合并。"""
    raw = rewriter.invoke({"question": question})
    variants = [line.strip() for line in raw.splitlines() if line.strip()]

    seen: set[str] = set()
    merged: list[Document] = []
    for q in [question, *variants]:        # 原问题也参加检索
        for doc in retriever.invoke(q)[:per_query_k]:
            key = doc.id or doc.page_content  # 去重键：优先用文档 id
            if key not in seen:
                seen.add(key)
                merged.append(doc)
    return merged

官方成品：MultiQueryRetriever——干的就是上面这些事：调用模型改写、逐条检索、合并去重。注意它在 LangChain 1.x 里搬了家，住进了 langchain-classic 包：

import logging

from langchain_classic.retrievers import MultiQueryRetriever

# 想在控制台看到模型改写出了哪些说法，开这个 logger 的 INFO 级别
logging.getLogger("langchain_classic.retrievers.multi_query").setLevel(logging.INFO)

mq_retriever = MultiQueryRetriever.from_llm(
    retriever=retriever,   # 底层的单查询检索器
    llm=model,             # 负责改写的模型
)
docs = mq_retriever.invoke("李征最后怎么了")

几个常用开关：

• 看改写日志：开 INFO 日志（或构造时传 verbose=True），控制台会打印 "Generated queries: ..."——对应其他语言版本里 verbose: true 的效果；
• 原问题也参战：MultiQueryRetriever.from_llm(retriever=..., llm=..., include_original=True)；
• 改写条数：没有现成参数（其他语言版本的 queryCount 在 Python 侧没有对应物），想改成 5 条就传自定义 prompt=，把提示词里的「3 种」改成「5 种」。

常见坑：

• 成本翻倍：改写是一次模型调用，N 路检索是 N 次向量查询。对延迟敏感的流式问答场景要先掂量。
• 改写跑偏就是灾难：改写悄悄变了原意，检回来的全是歪瓜裂枣。提示词里那句「不要改变原意」必须写死。
• 去重键：InMemoryVectorStore 返回的 Document 自带 id，直接用；换没有 id 的库就用 (doc.metadata, doc.page_content) 当键。
• 旧资料里的 from langchain.retrievers import MultiQueryRetriever 在 1.x 会报 ModuleNotFoundError——它现在的家是 langchain_classic.retrievers，装包用 pip install langchain-classic。

8. 检索优化地图：从切块调优到重排

RAG 答得不好，别第一时间去改提示词——多数时候是「检索」这一棒掉链子了。按流水线阶段排一张排查地图：

两个高频进阶件（都在 langchain-classic，了解思路即可，不必急着上）：

• ContextualCompressionRetriever + LLMChainExtractor：检索后处理——候选文档先让模型「只抽出与问题相关的句子」，再交给主链。省 token、聚焦，代价是每次检索多一次模型调用。
• 重排（rerank）：两段式选人——召回阶段讲究「快而全」（向量粗筛，大 k），精排阶段讲究「准」（用更贵的打分器对候选重新排序，取头部）。像选秀：海选看简历（向量相似度），面试看真本事（重排模型）。

排查口诀：答案不对，先看检索对了没有——把检索结果打印出来肉眼看；检索结果是对的、答案还不对，再回头查提示词和模型。顺着这张地图，90% 的「RAG 答非所问」都能定位到具体阶段。

9. 本章小结

一张速查表带走本章：

配套 demo（本章 demo/ 目录）：

• demo_rag.py：单文件完整 RAG——书屋读书笔记硬编码在文件里，切割 → 嵌入 → 入库 → 检索参数演示 → 组装问答链 → 来源溯源 → 多查询检索，一条龙跑通。需要本机 Ollama 已启动并拉取 qwen3:4b 和 nomic-embed-text；连不上时自动降级为「只演示检索部分」，文件头部注释写了全部前提。

到这里，拾光书屋的「知识库问答」已经完整落地。但还有一个明显的缺口：读者每次提问都是孤立的——「接着刚才说的，那借期呢？」会难倒它，因为链不记得上一轮聊过什么。下一章我们给对话装上记忆。

下一章：09 · 对话记忆