在构建 RAG 应用（检索增强生成）时，选择一套高效的工具链能够显著提升开发效率。本文将详细拆解整个流程，从环境配置到最终效果验证，逐步带你掌握核心步骤与实用技巧。

工具清单与选型

1. LangChain 框架（用于编排 RAG 链）
2. bge-small-zh-v1.5 嵌入模型（轻量中文语义向量）
3. Chroma 向量库（持久化存储和检索）
4. LCEL 语法构建的 RAG 链（LangChain 表达式语言）

整体流程概览

构建向量知识库

先梳理完整的实现步骤：

1. 数据采集：支持 TXT、PDF 等多种格式文件。
2. 数据加载：虽然 LangChain 提供 DirectoryLoader，但本例采用手动遍历文件夹加载方式，更易于理解底层逻辑。
3. 加载嵌入模型：本文选择的是 bge-small-zh-v1.5，兼顾准确性与速度。
4. 用该模型对文档进行向量化（转换为语义向量）。
5. 将向量存入 Chroma 向量库并持久化。

查询与检索知识库

1. 加载大语言模型（LLM），用于生成最终答案。
2. 加载与构建时相同的嵌入模型，确保向量空间一致。
3. 加载已存储的向量数据库，恢复索引。
4. 从数据库创建检索器（Retriever），指定返回文档数量。
5. 定义 Prompt 模板，约束 LLM 基于上下文回答。
6. 利用 LCEL 的 RAG 链将它们串联，形成端到端管道。
7. 用示例问题测试系统，验证效果。

分步实战详解

步骤一：构建知识库

1. 导入必要库

import os
from langchain.docstore.document import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_huggingface import HuggingFaceEmbeddings

2. 下载模型（建议预先克隆到本地）

git install lfs
$env:GIT_CLONE_PROTECTION_ACTIVE="false"; git clone https://huggingface.co/BAAI/bge-small-zh-v1.5

3. 选择嵌入模型（下面的MODEL_PATH需要换成自己下载好的路径）

model_kwargs = {'device': 'cpu'} # 若拥有 NVIDIA GPU，可改为 {'device': 'cuda'} 加速推理
encode_kwargs = {'normalize_embeddings': True} # 输出归一化向量，便于后续余弦相似度计算
embedding_function = HuggingFaceEmbeddings(
    model_name=MODEL_PATH,
    model_kwargs=model_kwargs,
    encode_kwargs=encode_kwargs
)

4. 加载文档

此处采用 Document 方式构建，每份文件都格式化为一个 Document 对象，便于统一管理。

• page_content (str)：文档的核心文本内容。
• metadata (dict)：描述文档的元数据，比如来源文件名、页码、作者等。元数据在后续高级应用（如过滤、溯源）中非常有用。目前支持两种格式，且该框架扩展性很高。

documents = []
try:
    for root, _, files in os.walk(SOURCE_DIRECTORY):
        print(f"正在扫描文件夹: {root}，查找 TXT 和 PDF 文件")
        for file in files:
            file_path = os.path.join(root, file)
            if file.endswith(".txt"):
                try:
                    with open(file_path, 'r', encoding='utf-8') as f:
                        text = f.read()
                    doc = Document(page_content=text, metadata={"source": file_path})
                    documents.append(doc)
                    print(f"成功手动加载 TXT 文件: {file_path}")
                except Exception as e:
                    print(f"加载 TXT 文件 {file_path} 时出错: {e}")
            elif file.endswith(".pdf"):
                try:
                    # 使用 LangChain 的 PyPDFLoader 处理 PDF，每页自动生成一个 Document
                    pdf_loader = PyPDFLoader(file_path)
                    pdf_docs = pdf_loader.load()
                    documents.extend(pdf_docs)
                    print(f"成功加载 PDF 文件: {file_path} (共 {len(pdf_docs)} 页)")
                except Exception as e:
                    print(f"加载 PDF 文件 {file_path} 时出错: {e}")

5. 文档切分（将长文档拆分为语义完整的短块）

• chunk_size=500：每个文本块的最大长度（字符数），尽量不超过 500 个字符，兼顾检索精度与上下文容量。
• chunk_overlap=100：相邻块之间的重叠字符数，下一个块的开头包含上一个块末尾的 100 个字符，避免因切分导致信息断裂。
• 切分策略：按优先级列表 ["\n\n", "\n", " ", ""] 进行递归分割。优先按段落（双换行符）切分，其次按单换行、空格，最后按字符。这种方式比简单按字符分割更符合语义结构。

# 2. 文档切分 (Chunking)
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100)
split_docs = text_splitter.split_documents(documents)
print(f"文档已切分为 {len(split_docs)} 个小块，准备向量化存储")

6. 向量化并存入 ChromaDB

db = Chroma.from_documents(
    split_docs,
    embedding_function,
    persist_directory=PERSIST_DIRECTORY  # 指定持久化路径，下次可直接加载
)
# 确保数据写入磁盘，避免丢失
db.persist()

步骤二：查询与检索知识库

1. 加载 LLM 模型（注意自行配置 API 密钥）

secrets = load_secrets()
llm = ChatOpenAI(
    api_key=secrets.get("API_KEY"),
    base_url=secrets.get("BASE_URL"),
    model_name=secrets.get("MODEL"),
    temperature=0.0  # RAG 场景下希望答案严格基于事实，温度设为 0 以获得最确定性输出
)

2. 加载嵌入模型（必须与构建知识库时使用的模型完全一致）

model_kwargs = {'device': 'cpu'}
encode_kwargs = {'normalize_embeddings': True}
embedding_function = HuggingFaceEmbeddings(
    model_name=MODEL_PATH,
    model_kwargs=model_kwargs,
    encode_kwargs=encode_kwargs
)

3. 加载已持久化的 Chroma 数据库

db = Chroma(persist_directory=PERSIST_DIRECTORY, embedding_function=embedding_function)

4. 创建检索器（Retriever）

将向量数据库 db 转换为专门负责检索的 检索器（Retriever）。参数 k=3 表示：每次问答时返回与用户查询最相似的前 3 个文档块。

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

5. 定义 Prompt 模板（约束 LLM 仅基于上下文作答）

template = """请只根据以下提供的上下文信息来回答问题。
上下文:
{context}
问题:
{question}"""
prompt = ChatPromptTemplate.from_template(template)

6. 利用 LCEL 链构建端到端 RAG 链

• retriever：即上一步创建的检索器，负责从向量库中召回相关文档。
• RunnablePassthrough：将输入的问题原样传递，不做修改。
• | prompt：将上一步输出的字典（包含 context 和 question）传给 Prompt，生成完整的问答提示。
• | llm：将 Prompt 传给大语言模型，返回 AIMessage。
• | StrOutputParser()：提取 AIMessage 中的文本内容，最终返回干净的字符串答案。

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

7. 执行答案查询（调用 invoke 触发完整链）

invoke是 LCEL 链条的“执行”入口，会从头到尾依次调用每个组件。传入用户问题即可获得基于知识库的精准回答。

response = rag_chain.invoke(question)

效果验证：实际问答示例

以下是用上述 RAG 系统处理真实问题的输出效果，展示了系统对因果推断领域知识的检索与生成能力。

知识库内容（txt）：因果推断是统计学和数据科学中的一个核心领域，它的主要目标是确定一个事件（原因）对另一个事件（结果）的影响程度。这不仅仅是发现变量之间的相关性，而是要探究它们之间是否存在真正的因果关系。

一个经典的概念是“相关不等于因果”。例如，夏天的冰淇淋销量和溺水人数都同时上升，它们高度相关，但我们不能说是吃冰淇淋导致了溺水。这背后其实有一个共同的原因，即“炎热的天气”，这个变量被称为“混杂变量”（Confounding Variable）。

处理混杂变量是因果推断的关键挑战之一。常用的方法包括：

1. 随机对照试验（RCT）：这是黄金标准，通过随机分组来消除混杂因素的影响。
2. 工具变量法（Instrumental Variables）：当无法进行RCT时，寻找一个“工具变量”，它只影响原因而不直接影响结果。
3. 回归不连续性设计（Regression Discontinuity）：利用一个临界值来近似实现随机分组的效果。

RAG 系统响应示例：

用户问题: 因果推断是什么？它和相关性有什么区别？

--- RAG 响应 ---

根据知识库的上下文，因果推断是统计学和数据科学中的一个核心领域，它的主要目标是确定一个事件（原因）对另一个事件（结果）的影响程度。它不仅仅是发现变量之间的相关性，而是要探究它们之间是否存在真正的因果关系。

与相关性相比，因果推断的区别在于：

1. 相关性指的是两个变量之间是否存在统计上的关联，即它们是否经常一起变化。例如，夏天的冰淇淋销量和溺水人数可能同时上升，这表明它们之间存在相关性。
2. 因果关系则进一步探究一个变量是否真正导致了另一个变量的变化。例如，虽然夏天的冰淇淋销量和溺水人数可能同时上升，但因果推断会探究是否是吃冰淇淋导致了溺水，而不是其他共同的原因，如炎热的天气。

简而言之，相关性描述了变量之间的统计关联，而因果推断则试图确定一个变量是否是另一个变量的原因。

用户问题: 处理混杂变量的方法有哪些？

--- RAG 响应 ---

知识库中列举了处理混杂变量的三种常用方法：

1. 随机对照试验（RCT）：这是黄金标准，通过随机分组来消除混杂因素的影响。
2. 工具变量法（Instrumental Variables）：当无法进行RCT时，寻找一个“工具变量”，它只影响原因而不直接影响结果。
3. 回归不连续性设计（Regression Discontinuity）：利用一个临界值来近似实现随机分组的效果。

用户问题: Judea Pearl 是谁？

--- RAG 响应 ---

根据提供的知识库，我无法回答该问题。