RAG 全链路：从文档加载到检索增强问答

RAG 的本质是一条离线建库和一条在线问答两段流水线。离线段（加载→分块→向量化→入库）把文档变成可被相似度搜索的向量索引，只跑一次或定期更新；在线段（检索→注入→生成）在每次用户提问时执行，把问题向量化后从库里召回 top-k 片段，拼进 prompt 交给模型。理解这个划分，才知道哪些代码该缓存、哪些该实时跑。

0. 安装与组件分工

LangChain 0.3+ 把功能拆成多个独立包，RAG 需要装核心包加几个集成包。注意 向量库和 embedding 模型是独立安装的，按你选的技术栈装对应包即可。

# 核心 + OpenAI 集成（ChatModel 与 Embeddings）
pip install langchain langchain-openai langchain-community

# 文档加载依赖：PDF 用 pypdf，网页用 beautifulsoup4
pip install pypdf beautifulsoup4

# 文本分块（RecursiveCharacterTextSplitter 在这个包里）
pip install langchain-text-splitters

# 向量库二选一：
pip install langchain-chroma chromadb   # Chroma（自带持久化，开箱即用）
pip install faiss-cpu                    # FAISS（高性能，需配合 langchain_community.vectorstores）

# API key（OpenAIEmbeddings 与 ChatOpenAI 共用）
export OPENAI_API_KEY="sk-..."

1. DocumentLoader：把任意来源读成 Document

所有 loader 的 .load() 都返回 List[Document]，每个 Document 有 page_content（正文文本）和 metadata（来源、页码等元数据）。三种最常用的 loader 覆盖网页、PDF、纯文本。

from langchain_community.document_loaders import (
    WebBaseLoader,
    PyPDFLoader,
    TextLoader,
)

# --- 1) 网页：抓取并解析 HTML 正文 ---
web_loader = WebBaseLoader("https://python.langchain.com/docs/introduction/")
web_docs = web_loader.load()          # List[Document]，通常整页是 1 个 Document
print(web_docs[0].metadata)           # {'source': 'https://...', 'title': '...'}
print(web_docs[0].page_content[:200])

# --- 2) PDF：按页加载，每页一个 Document，metadata 含页码 ---
pdf_loader = PyPDFLoader("./docs/manual.pdf")
pdf_docs = pdf_loader.load()          # 一页一个 Document
print(len(pdf_docs), "pages")
print(pdf_docs[0].metadata)           # {'source': './docs/manual.pdf', 'page': 0}

# --- 3) 纯文本：整个文件一个 Document ---
txt_loader = TextLoader("./docs/notes.txt", encoding="utf-8")
txt_docs = txt_loader.load()
print(txt_docs[0].page_content[:200])

# 实战中常把多个来源合并成一个列表统一处理
all_docs = web_docs + pdf_docs + txt_docs

2. TextSplitter：为什么要切块，怎么切

不能把整篇文档直接向量化：一是 embedding 模型有 token 上限，二是一整页的向量语义太糊，检索时召回精度差。RecursiveCharacterTextSplitter 是首选——它优先按段落（\n\n）切，切不开再退而求其次按句子、按词、按字符切，尽量保住语义边界。

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每块目标长度（默认按字符数；中文≈字数）
    chunk_overlap=80,      # 相邻块重叠的字符数，防止语义在边界被腰斩
    # 递归切分的分隔符优先级：先段落，再换行，再空格，最后逐字
    separators=["\n\n", "\n", "。", "！", "？", " ", ""],
    length_function=len,   # 用字符长度衡量；想按 token 切见下方 callout
)

# split_documents 会保留每个 Document 的 metadata，并自动透传到切出的小块
splits = splitter.split_documents(all_docs)
print(f"切出 {len(splits)} 个块")
print(splits[0].page_content)
print(splits[0].metadata)   # 仍带 source / page，便于溯源

3. Embeddings + VectorStore：建库与持久化

OpenAIEmbeddings 把每个块转成向量；向量库负责存储和相似度搜索。下面给 Chroma 和 FAISS 两套完整代码，都包含建库、持久化、重新加载三步。记住一条铁律：写入和查询必须用同一个 embedding 模型，否则向量空间不一致，检索全乱。

from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

# 1) embedding 模型：建库与查询全程复用同一个实例
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 2) 从切好的块建库；persist_directory 一指定即自动落盘
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=embeddings,
    persist_directory="./chroma_db",   # 持久化目录
    collection_name="langchain_docs",
)
print("建库完成，向量数：", vectorstore._collection.count())

# 3) 下次直接从磁盘加载，无需重新 embedding（省钱省时间）
vectorstore = Chroma(
    persist_directory="./chroma_db",
    collection_name="langchain_docs",
    embedding_function=embeddings,     # 注意这里参数名是 embedding_function
)

# 4) 裸搜索验证：返回最相似的 Document 列表
hits = vectorstore.similarity_search("LangChain 是什么", k=3)
for d in hits:
    print(d.metadata.get("source"), "->", d.page_content[:60])

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 1) FAISS 建库：纯内存索引，速度快
vectorstore = FAISS.from_documents(splits, embeddings)

# 2) 持久化：序列化到本地目录
vectorstore.save_local("./faiss_index")

# 3) 重新加载（FAISS 用 pickle 反序列化，需显式开启 allow_dangerous_deserialization）
vectorstore = FAISS.load_local(
    "./faiss_index",
    embeddings,
    allow_dangerous_deserialization=True,   # 仅加载你自己生成的索引时才设 True
)

hits = vectorstore.similarity_search("chunk_overlap 的作用", k=3)
for d in hits:
    print(d.page_content[:60])

4. Retriever：as_retriever 与 search_kwargs

向量库本身可以搜索，但 LCEL 链需要的是统一的 Retriever 接口（一个 Runnable：输入查询字符串，输出 List[Document]）。as_retriever() 把任意向量库包装成 retriever，search_kwargs 控制召回行为，search_type 切换检索策略。

# 1) 最常用：相似度 top-k 检索
retriever = vectorstore.as_retriever(
    search_type="similarity",     # 默认值，按向量相似度排序
    search_kwargs={"k": 4},       # 召回 4 个最相关的块
)

# 2) MMR：在相关性与多样性间平衡，避免召回内容高度重复
retriever_mmr = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.5},
)

# 3) 带相似度阈值：低于阈值的不召回（过滤弱相关）
retriever_thresh = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.5, "k": 4},
)

# retriever 是 Runnable，直接 invoke 即可拿到 Document 列表
docs = retriever.invoke("如何配置 chunk_size")
print([d.page_content[:40] for d in docs])

5. 用 LCEL 拼检索增强问答链（核心）

现在把所有环节用 LCEL 串起来。难点在于：prompt 同时需要 context（检索到的文档）和 question（原始问题），所以要用 RunnableParallel（这里用字典字面量等价表达）让两条分支并行——context 分支走 retriever | format_docs，question 分支用 RunnablePassthrough 原样透传。format_docs 是关键胶水：把 List[Document] 压成一个纯文本字符串注入 prompt。

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

# 1) 把检索到的 Document 列表压成一段纯文本
def format_docs(docs):
    return "\n\n".join(d.page_content for d in docs)

# 2) prompt：明确要求“只依据 context 作答”，抑制幻觉
prompt = ChatPromptTemplate.from_template(
    """你是文档问答助手。只依据下面的【上下文】回答问题，
如果上下文里没有答案，就直说“根据已有资料无法回答”，不要编造。

【上下文】
{context}

【问题】
{question}

【回答】"""
)

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 3) LCEL 拼装：
#    - context 分支：retriever 检索 -> format_docs 压成字符串
#    - question 分支：RunnablePassthrough 把原始输入原样传下去
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | model
    | StrOutputParser()
)

# 4) 直接传一个问题字符串即可（question 分支透传它，context 分支检索它）
answer = rag_chain.invoke("chunk_overlap 有什么作用？")
print(answer)

# 流式输出也开箱即用（LCEL 链天然支持 stream）
for token in rag_chain.stream("FAISS 和 Chroma 怎么选？"):
    print(token, end="", flush=True)

理解这条链只需抓住一点：输入的问题字符串被同时喂给字典的两个分支。question 分支靠 RunnablePassthrough 把它原样保留给 prompt；context 分支把它送进 retriever 检索出文档、再经 format_docs 变成文本。两个分支的结果汇成一个 {context, question} 字典，正好填满 prompt 的两个占位符——这就是 RAG 在 LCEL 里的标准范式。

6. create_retrieval_chain：官方封装写法

如果不想手写上面的字典分支，LangChain 提供了现成的 create_retrieval_chain + create_stuff_documents_chain。它内部帮你做了文档检索和拼接，prompt 里直接用 {context} 占位即可。代价是灵活性下降、输出结构固定。

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# prompt 必须包含 {context} 占位符；输入键默认是 input
qa_prompt = ChatPromptTemplate.from_template(
    """只依据下面的上下文回答问题，没有答案就说不知道。

上下文：
{context}

问题：{input}"""
)

# 1) “塞文档”链：负责把检索到的文档填进 {context}
combine_docs_chain = create_stuff_documents_chain(model, qa_prompt)

# 2) 检索链：retriever 检索 -> 交给 combine_docs_chain 生成答案
retrieval_chain = create_retrieval_chain(retriever, combine_docs_chain)

# 3) 调用：输入键是 input，输出是 dict（含 answer 和 context）
result = retrieval_chain.invoke({"input": "chunk_size 该怎么设？"})
print(result["answer"])                 # 模型生成的答案
print(len(result["context"]), "个引用片段")  # 检索到的源文档，便于溯源
for d in result["context"]:
    print("  -", d.metadata.get("source"))