LangChain 实现 RAG

Written: 2026.06

1. LangChain RAG 组件与流程

1.1 LangChain组件

LangChain 框架提供了丰富的组件帮助我们搭建 RAG 应用，下面是关于这些核心组件的介绍：

LangChain组件	作用	常用组件类
`文档加载器`	对各种格式的文档信息进行加载	`Document`(文档组件)、 `UnstructuredPDFLoader`(PDF文档加载器)、 `UnstructuredFileLoader`(文件文档加载器)、 `UnstructuredMarkdownLoader`(markdown文档文本加载器)
`文档分割器`	将加载的文档分割成文档片段	`RecursiveCharacterTextSplitter` 递归字符文本分割器
`文本嵌入模型组件`	将文本信息向量化	`OpenAIEmbeddings` OpenAI文本嵌入模型 `HuggingFaceEmbeddings` HuggingFace文本嵌入模型
`向量数据库组件`	将向量和元数据信息保存到向量数据库	`VectorStore` 向量数据库，不同向量数据库有不同的实现类
`文本检索器`	根据用户提问在向量数据库中进行检索	`VectorStoreRetriever` 向量数据库检索器

1.2 LangChain 实现流程

在RAG准备阶段，LangChain通过文档加载器对各种格式的文档进行加载，转换为LangChain中的文档对象，之后对文档对象进行分割，根据分割规则，分割成文档片段

之后，将文档片段通过文本嵌入模型组件，转换为向量，通过向量数据库组件，保存到向量数据库，每个向量通常还会绑定原文内容、文档ID、来源等元数据信息，便于后面数据检索

在RAG的使用阶段，用户首先提出问题，使用文本嵌入模型组件，将提问文本转换为向量数据，通过向量数据库检索器组件，进行相似性检索，返回关联的文本片段将相关的文档片段内容渲染到提示词模板中，作为提问问题的上下文传递给大语言模型，大语言模型输出结果，再传递给输出解析器，最终结果通过输出解析器处理后返回给用户，这样就完成了一次RAG检索

2. 文档加载器

在LangChain中，文档加载器用于将各种格式的文档转换为Document对象，LangChain提供了大量的文档加载器，支持从各种来源加载文档，如文件、数据源、URL等

2.1 LangChain 文档加载器

文档加载器	作用
`CSVLoader`	从CSV加载文档
`JSONLoader`	从JSON数据加载文档
`PyPDFLoader`	从PDF数据加载文档
`UnstructuredHTMLLoader`	从HTML数据加载文档
`UnstructuredMarkdownLoader`	从Markdown加载文档
`UnstructuredExcelLoader`	从Excel文件加载数据

每一个文档加载器都有自己特定的参数和方法，但它们有一个统一的load()方法来完成文档的加载，load()方法会返回一个Document类的对象列表，因为这些文档加载器都继承自BaseLoader基类，它们的继承关系如下：

在BaseLoader类中，定义了load()方法，用来加载文档对象，在方法内部又调用了lazy_load()懒加载方法

load( )

def load(self) -> List[Document]:
    """Load data into Document objects."""
    return list(self.lazy_load())

在lazy_load()方法中，判断了子类是否重写了load()方法，如果重写了，则调用当前类的load()方法，如果没有重写则抛出异常，因此在子类中，要重写load()方法或lazy_load()方法

lazy_load( )

def lazy_load(self) -> Iterator[Document]:
    """A lazy loader for Documents."""
    if type(self).load != BaseLoader.load:
        return iter(self.load())
    raise NotImplementedError(
        f"{self.__class__.__name__} does not implement lazy_load()"
    )

如果LangChain提供的文档加载器无法满足业务需求，我们也可以自己实现自定义加载器，通过继承BaseLoader，并实现其中的load()方法，来编写自定义文档加载器的加载逻辑

2.2 Document文档类

文档加载器无论从什么来源进行文档加载，最终都是为了将文档信息解析为Document对象，下面一起来看看Document类中重要属性： Document类中，主要包含两个重要属性： page_content：表示文档的内容，类型是字符串 metadata ：与文档本身无关的元数据信息。可以保存文档 ID、文件名等任意信息，类型是字典

2.3 文档加载器使用

下面以UnstructuredMarkdownLoader为例来介绍文档加载器的用法，使用UnstructuredMarkdownLoader读取md文件示例如下：

UnstructuredMarkdownLoader示例

from langchain_community.document_loaders import UnstructuredMarkdownLoader

# 1.创建文档加载器，并指定路径
document_load = UnstructuredMarkdownLoader(file_path="LangChain框架入门09：什么是RAG？.md")

# 2.加载文档
documents = document_load.load()

# 3.打印文档内容
print(f"文档数量：{len(documents)}")
for document in documents:
    print(f"文档内容：{document.page_content}")
    print(f"文档元数据：{document.metadata}")

执行结果如下，默认情况下UnstructuredMarkdownLoader把md文档内容加载成了一个Document对象，并且自动将文件名添加到了Document对象的元数据中

打印文档内容

文档数量：1
文档内容：什么是 RAG
(文档内容省略...)
文档元数据：{'source': 'RAG入门.md'}

在底层Unstructured包会为不同的文本片段创建不同的“元素”。默认情况下会将这些元素合并在一起，可以通过指定 mode="elements" 来将不同元素进行分离，解析成多个文档

UnstructuredMarkdownLoader示例

document_load = UnstructuredMarkdownLoader(file_path="RAG入门.md", mode="elements")

重新执行代码，可以看到加载的文档数为12，文档的内容也按照不同元素进行了拆分

打印文档内容

文档数量：92
文档内容：什么是 RAG
文档元数据：{'source': 'RAG入门.md', 'category_depth': 0, 'languages': ['zho'], 'filename': 'RAG入门.md', 'filetype': 'text/markdown', 'last_modified': '2025-09-06T21:50:19', 'category': 'Title', 'element_id': '97fc7081ad1cf916d8fc1eead4f1e5f9'}
(省略...)

2.4 自定义文档加载器

在实际开发中，基于文件的不同类型和不同格式，有时通过这些LangChain提供的文档加载器很难满足业务需求，例如需要根据特定规则提取文本片段，这时就需要开发自定义加载器，只需要定义一个自定义文档加载器类，并继承前面提到的BaseLoader类假设有如下需求，对 faq.txt文件进行文档加载，内容如下，要求将问题和答案加载成一个文档，并添加文件创建日期元数据

自定义文档加载器

Q：在线支付取消订单后钱怎么返还？
订单取消后，款项会在一个工作日内，直接返还到您的美团账户余额。
Q：怎么查看退款是否成功？
退款会在一个工作日之内到美团账户余额，可在“账号管理——我的账号”中查看是否到账。
Q：美团账户里的余额怎么提现？
余额可到美团网(meituan.com)——“我的美团→美团余额”里提取到您的银行卡或者支付宝账号，另外，余额也可直接用于支付外卖订单(限支持在线支付的商家)。
Q：余额提现到账时间是多久？
1-7个工作日内可退回您的支付账户。由于银行处理可能有延迟，具体以账户的到账时间为准。
Q：申请退款后，商家拒绝了怎么办？
申请退款后，如果商家拒绝，此时回到订单页面点击“退款申诉”，美团客服介入处理。
Q：怎么取消退款呢？
请在订单页点击“不退款了”，商家还会正常送餐的。
Q：前面下了一个在线支付的单子，由于未付款，订单自动取消了，这单会计算我的参与活动次数吗？
不会。如果是未支付的在线支付订单，可以先将订单取消(如果不取消需要15分钟后系统自动取消)，订单无效后，此时您再下单仍会享受活动的优惠。
Q：为什么我用微信订餐，却无法使用在线支付？
目前只有网页版和美团外卖手机App(非美团手机客户端)订餐，才能使用在线支付，请更换到网页版和美团外卖手机App下单。
Q：如何进行付款？
美团外卖现在支持货到付款与在线支付，其中微信版与手机触屏版暂不支持在线支付。

自定义文档加载器代码如下：

SimpleQALoader示例

import os
from datetime import datetime
from langchain_core.documents import Document
from langchain.document_loaders.base import BaseLoader

class SimpleQALoader(BaseLoader):
    """
    简单的问答文件加载器

    该加载器用于从文本文件中加载问答对，文件格式要求每两行为一组，
    第一行为问题(Q)，第二行为答案(A)

    Args:
        file_path (str): 问答文件的路径
        time_fmt (str): 时间格式字符串，默认为 "%Y-%m-%d %H:%M:%S"
    """

    def __init__(self, file_path: str, time_fmt: str = "%Y-%m-%d %H:%M:%S"):
        self.file_path = file_path
        self.time_fmt = time_fmt

    def load(self):
        """
        加载并解析问答文件

        读取文件中的问答对，每两行构成一个问答文档，第一行为问题，第二行为答案。
        每个文档包含问题和答案的组合内容，以及文件的元数据信息。

        Returns:
            list[Document]: 包含问答内容的文档列表，每个文档包含page_content和metadata
        """
        with open(self.file_path, "r", encoding="utf-8") as f:
            lines = [line.strip() for line in f if line.strip()]

        docs = []
        created_ts = os.path.getctime(self.file_path)
        created_at = datetime.fromtimestamp(created_ts).strftime(self.time_fmt)

        # 每两行构成一个 Q/A
        for i in range(0, len(lines), 2):
            q = lines[i].lstrip("Q：:").strip()
            a = lines[i+1].lstrip("A：:").strip()
            page_content = f"Q: {q}\nA: {a}"

            doc = Document(
                page_content=page_content,
                metadata={
                    "source": self.file_path,
                    "created_at": created_at,
                }
            )
            docs.append(doc)

        return docs

# 使用示例
if __name__ == "__main__":
    loader = SimpleQALoader("faq.txt")
    docs = loader.load()
    print(f"共解析到 {len(docs)} 个文档")
    for i, d in enumerate(docs, 1):
        print(f"\n--- 文档 {i} ---")
        print(d.page_content)
        print("元数据：", d.metadata)

执行结果如下，faq.txt文件被解析成 9 个文档，并且每个文档的元数据都保存了created_at

使用示例

共解析到 9 个文档

--- 文档 1 ---
Q: 在线支付取消订单后钱怎么返还？
A: 订单取消后，款项会在一个工作日内，直接返还到您的美团账户余额。
元数据： {'source': 'faq.txt', 'created_at': '2025-09-06 22:15:47'}

--- 文档 2 ---
Q: 怎么查看退款是否成功？
A: 退款会在一个工作日之内到美团账户余额，可在“账号管理——我的账号”中查看是否到账。
元数据： {'source': 'faq.txt', 'created_at': '2025-09-06 22:15:47'}

--- 文档 3 ---
Q: 美团账户里的余额怎么提现？
A: 余额可到美团网(meituan.com)——“我的美团→美团余额”里提取到您的银行卡或者支付宝账号，另外，余额也可直接用于支付外卖订单(限支持在线支付的商家)。
元数据： {'source': 'faq.txt', 'created_at': '2025-09-06 22:15:47'}

--- 文档 4 ---
Q: 余额提现到账时间是多久？
A: 1-7个工作日内可退回您的支付账户。由于银行处理可能有延迟，具体以账户的到账时间为准。
元数据： {'source': 'faq.txt', 'created_at': '2025-09-06 22:15:47'}

--- 文档 5 ---
Q: 申请退款后，商家拒绝了怎么办？
A: 申请退款后，如果商家拒绝，此时回到订单页面点击“退款申诉”，美团客服介入处理。
元数据： {'source': 'faq.txt', 'created_at': '2025-09-06 22:15:47'}

--- 文档 6 ---
Q: 怎么取消退款呢？
A: 请在订单页点击“不退款了”，商家还会正常送餐的。
元数据： {'source': 'faq.txt', 'created_at': '2025-09-06 22:15:47'}

--- 文档 7 ---
Q: 前面下了一个在线支付的单子，由于未付款，订单自动取消了，这单会计算我的参与活动次数吗？
A: 不会。如果是未支付的在线支付订单，可以先将订单取消(如果不取消需要15分钟后系统自动取消)，订单无效后，此时您再下单仍会享受活动的优惠。
元数据： {'source': 'faq.txt', 'created_at': '2025-09-06 22:15:47'}

--- 文档 8 ---
Q: 为什么我用微信订餐，却无法使用在线支付？
A: 目前只有网页版和美团外卖手机App(非美团手机客户端)订餐，才能使用在线支付，请更换到网页版和美团外卖手机App下单。
元数据： {'source': 'faq.txt', 'created_at': '2025-09-06 22:15:47'}

--- 文档 9 ---
Q: 如何进行付款？
A: 美团外卖现在支持货到付款与在线支付，其中微信版与手机触屏版暂不支持在线支付。
元数据： {'source': 'faq.txt', 'created_at': '2025-09-06 22:15:47'}

3. 文本分割器

3.1 LangChain 文本分割器

LangChain提供了多种文本分割器，常用的有：

分割器	作用
`RecursiveCharacterTextSplitter`	递归按字符分割文本
`CharacterTextSplitter`	按指定字符分割文本
`MarkdownHeaderTextSplitter`	按Markdown标题分割
`PythonCodeTextSplitter`	专门分割Python代码
`TokenTextSplitter`	按Token数量分割
`HTMLHeaderTextSplitter`	按HTML标题分割

大部分文本分割器都继承自TextSplitter基类，该基类定义了分割文本的核心方法：

split_text()：将文本字符串分割成字符串列表
split_documents()：将Document对象列表分割成更小文本片段的Document对象列表
create_documents()：通过字符串列表创建Document对象

3.2 递归文本分割器用法

RecursiveCharacterTextSplitter是LangChain中最常用的通用文本分割器，它会根据指定的字符优先级递归分割文本，直到所有片段长度不超过指定上限首先介绍一下RecursiveCharacterTextSplitter构造函数几个核心参数： chunk_size：每个片段的最大字符数 chunk_overlap：片段之间的重叠字符数 length_function：计算长度函数 is_separator_regex：分隔符是否为正则表达式 separators：自定义分隔符

3.3 分割文本

首先介绍使用split_text()方法进行文本分割，使用示例如下，其中RecursiveCharacterTextSplitter中指定的块大小为100，片段重叠字符数为30，计算长度的函数使用len()

RecursiveCharacterTextSplitter示例

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 1.分割文本内容
content = (
    "大模型RAG(检索增强生成)是一种结合生成模型与外部知识检索的技术，通过从大规模文档或数据库中检索相关信息，辅助生成模型以提升回答的准确性和相关性。其核心流程包括用户输入查询、系统检索相关知识、生成模型基于检索结果生成内容，并输出最终答案。RAG的优势在于能够弥补生成模型的知识盲区，提供更准确、实时和可解释的输出，广泛应用于问答系统、内容生成、客服、教育和企业领域。然而，其也面临依赖高质量知识库、可能的响应延迟、较高的维护成本以及数据隐私等挑战。")
# 2.定义递归文本分割器
# 使用RecursiveCharacterTextSplitter创建文本分割器，设置块大小为100，重叠长度为30
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=30, length_function=len)

# 3.分割文本
# 将原始文本内容分割成多个文本块
splitter_texts = text_splitter.split_text(content)

# 4.转换为文档对象
# 将分割后的文本块转换为文档对象列表
splitter_documents = text_splitter.create_documents(splitter_texts)
print(f"原始文本大小：{len(content)}")
print(f"分割文档数量：{len(splitter_documents)}")
for splitter_document in splitter_documents:
    print(f"文档片段大小：{len(splitter_document.page_content)},文档内容：{splitter_document.page_content}")

执行结果如下，文本分割器将文本内容分割成了 3 个文本片段，且内容长度最大为100个字符

将分割后的文本块转换为文档对象列表

原始文本大小：225
分割文档数量：3
文档片段大小：100,文档内容：大模型RAG(检索增强生成)是一种结合生成模型与外部知识检索的技术，通过从大规模文档或数据库中检索相关信息，辅助生成模型以提升回答的准确性和相关性。其核心流程包括用户输入查询、系统检索相关知识、生成模
文档片段大小：100,文档内容：相关性。其核心流程包括用户输入查询、系统检索相关知识、生成模型基于检索结果生成内容，并输出最终答案。RAG的优势在于能够弥补生成模型的知识盲区，提供更准确、实时和可解释的输出，广泛应用于问答系统、内容
文档片段大小：85,文档内容：区，提供更准确、实时和可解释的输出，广泛应用于问答系统、内容生成、客服、教育和企业领域。然而，其也面临依赖高质量知识库、可能的响应延迟、较高的维护成本以及数据隐私等挑战。

3.4 分割文档对象

RecursiveCharacterTextSplitter不仅可以分割纯文本，还可以直接分割Document对象，使用示例如下：

RecursiveCharacterTextSplitter示例

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_unstructured import UnstructuredLoader

# 1.创建文档加载器，进行文档加载
loader = UnstructuredLoader("rag.txt")
documents = loader.load()

# 2.定义递归文本分割器
# 创建RecursiveCharacterTextSplitter实例，用于将文档分割成指定大小的文本块
# chunk_size: 每个文本块的最大字符数为100
# chunk_overlap: 相邻文本块之间的重叠字符数为30
# length_function: 使用len函数计算文本长度
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=30, length_function=len)

# 3.分割文本
# 使用文本分割器将加载的文档分割成多个较小的文档片段
splitter_documents = text_splitter.split_documents(documents)

# 输出分割后的文档信息
print(f"分割文档数量：{len(splitter_documents)}")
for splitter_document in splitter_documents:
    print(f"文档片段：{splitter_document.page_content}")
    print(f"文档片段大小：{len(splitter_document.page_content)}, 文档元数据：{splitter_document.metadata}")

输出分割后的文档信息

分割文档数量：3
文档片段：大模型RAG(检索增强生成)是一种结合生成模型与外部知识检索的技术，通过从大规模文档或数据库中检索相关信息，辅助生成模型以提升回答的准确性和相关性。其核心流程包括用户输入查询、系统检索相关知识、生成模
文档片段大小：100, 文档元数据：{'source': 'rag.txt', 'last_modified': '2025-09-06T23:10:05', 'languages': ['zho'], 'filename': 'rag.txt', 'filetype': 'text/plain', 'category': 'UncategorizedText', 'element_id': '2cfba084735e806b5c74d312ca68e815'}
文档片段：相关性。其核心流程包括用户输入查询、系统检索相关知识、生成模型基于检索结果生成内容，并输出最终答案。RAG的优势在于能够弥补生成模型的知识盲区，提供更准确、实时和可解释的输出，广泛应用于问答系统、内容
文档片段大小：100, 文档元数据：{'source': 'rag.txt', 'last_modified': '2025-09-06T23:10:05', 'languages': ['zho'], 'filename': 'rag.txt', 'filetype': 'text/plain', 'category': 'UncategorizedText', 'element_id': '2cfba084735e806b5c74d312ca68e815'}
文档片段：区，提供更准确、实时和可解释的输出，广泛应用于问答系统、内容生成、客服、教育和企业领域。然而，其也面临依赖高质量知识库、可能的响应延迟、较高的维护成本以及数据隐私等挑战。
文档片段大小：85, 文档元数据：{'source': 'rag.txt', 'last_modified': '2025-09-06T23:10:05', 'languages': ['zho'], 'filename': 'rag.txt', 'filetype': 'text/plain', 'category': 'UncategorizedText', 'element_id': '2cfba084735e806b5c74d312ca68e815'}

3.5 自定义分隔符

RecursiveCharacterTextSplitter默认按照["\n\n", "\n", " ", ""]的优先级进行分割，可以通过separators指定自定义分隔符

RecursiveCharacterTextSplitter示例

# 2.定义递归文本分割器
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100,
                                               chunk_overlap=30,
                                               length_function=len,
                                               separators=["。", "?", "\n\n", "\n", " ", ""]
                                              )

3.6 按标题分割Markdown文件

在对Markdown格式的文档进行分割时，一般不能像RecursiveCharacterTextSplitter默认分割规则方式进行分割，通常需要按照标题层次进行分割，LangChain提供了MarkdownHeaderTextSplitter类来实现这个功能在对Markdown文件进行分割时，对于那些很长的文档，可以先利用MarkdownHeaderTextSplitter按标题分割，将分割后的文档再使用RecursiveCharacterTextSplitter进行分割，使用示例如下：

MarkdownHeaderTextSplitter示例

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

# 1.文档加载
# 创建文本加载器并加载Markdown文档
loader = TextLoader(file_path="RAG入门.md")
documents = loader.load()
document_text = documents[0].page_content

# 2.定义文本分割器，设置指定要分割的标题
# 配置Markdown标题分割规则，指定不同级别的标题标记及其对应的元数据标签
headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2")
]
headers_text_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)

# 3.按标题分割文档
# 使用标题分割器将文档按Markdown标题结构进行分割
headers_splitter_documents = headers_text_splitter.split_text(document_text)

print(f"按标题分割文档数量：{len(headers_splitter_documents)}")
for splitter_document in headers_splitter_documents:
    print(f"按标题分割文档片段大小：{len(splitter_document.page_content)}, 文档元数据：{splitter_document.metadata}")

# 4.定义递归文本分割器
# 创建递归字符分割器，用于进一步细分过大的文档片段
# chunk_size: 每个文本块的目标大小为100个字符
# chunk_overlap: 相邻文本块之间的重叠字符数为30
# length_function: 使用len函数计算文本长度
text_splitter = RecursiveCharacterTextSplitter(chunk_size=100,
                                               chunk_overlap=30,
                                               length_function=len
                                              )

# 5.递归分割文本
# 对已按标题分割的文档片段进行二次递归分割，确保每个片段不超过指定大小
recursive_documents = text_splitter.split_documents(headers_splitter_documents)
print(f"第二次递归文本分割文档数量：{len(recursive_documents)}")
for recursive_document in recursive_documents:
    print(
        f"第二次递归文本分割文档片段大小：{len(recursive_document.page_content)}, 文档元数据：{recursive_document.metadata}")

执行结果如下，先用MarkdownHeaderTextSplitter将markdown文本内容分割成4个文档，之后在对每一个文档使用RecursiveCharacterTextSplitter进行分割，分割成了11个文档，并且在文档元数据中，还添加了文本片段所属的标题信息

执行结果如下

按标题分割文档数量：15
按标题分割文档片段大小：247, 文档元数据：{'Header 1': '什么是 RAG'}
按标题分割文档片段大小：446, 文档元数据：{'Header 1': '为什么需要RAG', 'Header 2': '缺陷一：大模型幻觉'}
按标题分割文档片段大小：713, 文档元数据：{'Header 1': '为什么需要RAG', 'Header 2': '缺陷二：有限的最大上下文'}
按标题分割文档片段大小：406, 文档元数据：{'Header 1': '为什么需要RAG', 'Header 2': '缺陷三：模型专业知识与时效性知识不足'}
按标题分割文档片段大小：806, 文档元数据：{'Header 1': 'RAG技术实现流程'}
按标题分割文档片段大小：673, 文档元数据：{'Header 1': 'RAG系统使用场景'}
按标题分割文档片段大小：196, 文档元数据：{'Header 1': 'RAG全栈技术体系介绍'}
按标题分割文档片段大小：815, 文档元数据：{'Header 1': 'RAG全栈技术体系介绍', 'Header 2': 'GraphRAG'}
按标题分割文档片段大小：521, 文档元数据：{'Header 1': 'RAG全栈技术体系介绍', 'Header 2': 'Agentic RAG'}
按标题分割文档片段大小：45, 文档元数据：{'Header 1': 'RAG热门开源项目&产品'}
…………

3.7 自定义文本分割器

当内置的的文本分割器无法满足业务需求时，可以继承TextSplitter类来实现自定义分割器，不过一般需要自定义文本分割器的情况非常少，假设我们有如下需求，在对文本分割时，按段落进行分割，并且每个段落只提取第一句话，下面通过实现自定义文本分割器，来实现这个功能，示例如下：

CustomTextSplitter示例

from typing import List

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import TextSplitter

class CustomTextSplitter(TextSplitter):
    """
    自定义文本分割器类

    该类继承自TextSplitter，用于将文本按照特定规则进行分割
    分割策略：首先按段落分割，然后对每个段落提取第一句话
    """

    def split_text(self, text: str) -> List[str]:
        """
        将输入文本分割成多个文本片段

        参数:
            text (str): 需要分割的原始文本字符串

        返回:
            List[str]: 分割后的文本片段列表，每个片段为段落的第一句话
        """
        text = text.strip()
        # 1.按段落进行分割
        text_array = text.split("\n\n")

        result_texts = []
        for text_item in text_array:
            strip_text_item = text_item.strip()
            if strip_text_item is None:
                continue
            # 2.按句进行分割
            result_texts.append(strip_text_item.split("。")[0])
        return result_texts

# 1.文档加载
loader = TextLoader(file_path="RAG入门.md")
documents = loader.load()
document_text = documents[0].page_content

# 2.定义文本分割器
splitter = CustomTextSplitter()

# 3.文本分割
splitter_texts = splitter.split_text(document_text)
for splitter_text in splitter_texts:
    print(
        f"文本分割片段大小：{len(splitter_text)}, 文本内容：{splitter_text}")

文本分割

文本分割片段大小：256, 文本内容：# 什么是 RAG
RAG，Retrieval-Augmented Generation，也被称作检索增强生成技术，最早在 Facebook AI(Meta AI)在 2020 年发表的论文《Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks》( https://arxiv.org/abs/2005.11401 )中正式提出，这种方法的核心思想是借助一些文本检索策略，让大模型每次问答前都带入相关文本，以此来改善大模型回答时的准确性
文本分割片段大小：92, 文本内容：# 为什么需要RAG
## 缺陷一：大模型幻觉
大家在使用大模型的时候，都会遇到大模型无中生有胡编乱造答案的情况，例如胡乱生成一些概念、一些论文甚至是一些实时等，这就是所谓的大模型幻觉
文本分割片段大小：106, 文本内容：![](https://api.cuiliangblog.cn/v1/public/imgProxy/?url=https://cdn.nlark.com/yuque/0/2025/png/2308212/1756388279716-93c6e99d-a511-4380-a9b6-8123e0db056b.png)
文本分割片段大小：53, 文本内容：大型语言模型之所以会产生幻觉，主要是因为它们的训练方式和内在机制决定了它们并不具备真正理解和验证事实的能力
文本分割片段大小：105, 文本内容：## 缺陷二：有限的最大上下文
由于大模型的本质其实是一个算法，不管是让大模型“知道”有哪些外部工具，还是要给大模型进行“背景设置”，或者是要给模型添加历史对话消息，以及本次对话的输出，都需要占用这个上下文窗口
文本分割片段大小：33, 文本内容：大型语言模型还存在最大上下文限制，这是由它们的架构和计算方式决定的
文本分割片段大小：168, 文本内容：早些时候的大模型普遍是8k最大上下文，相当于是8-10页中文PDF，伴随着大模型预训练技术的不断发展，顶尖的大模型，如Gemini 2.5 Pro和GPT-4.1等模型，已经达到了1M的最大上下文长度，相当于是一千页的PDF，相当于1.5本《红楼梦》，而普通的模型，也基本达到64K或128K最大上下文，相当于60-100也左右的PDF
文本分割片段大小：106, 文本内容：![](https://api.cuiliangblog.cn/v1/public/imgProxy/?url=https://cdn.nlark.com/yuque/0/2025/png/2308212/1756388280202-74832b75-9554-4311-8387-50958e801d44.png)
……

4. VectorStore存储组件

4.1 VectorStore组件介绍

对非结构化数据的存储与检索，最常用的方法是先将文本进行嵌入，转换为向量后存储到向量数据库中；在查询时，同样将查询文本嵌入生成向量，再将该向量传递给向量数据库，由数据库完成后续的相似度计算与检索过程在 LangChain 中，这一过程由顶层接口 VectorStore 统一管理，不同类型的向量数据库只需实现该接口中的抽象方法即可完成集成。VectorStore 接口提供了多个常用方法，例如：

add_texts：将文本列表转换为向量，并存储到向量数据库
add_documents：将文档列表转换为向量，并存储到向量数据库
as_retriever：返回向量数据库初始化的检索器
similarity_search_with_relevance_scores：进行相似性检索，返回文档及其相关性得分(范围在[0, 1]之间)
delete：根据向量id删除向量数据
from_texts：传入文本列表、元数据信息、文本嵌入模型，返回创建好的VectorStore

VectorStore接口，常用的实现类如下：

类名	描述
`InMemoryVectorStore`	基于内存实现的向量数据库
`ElasticsearchStore`	Elasticsearch为基础实现的向量数据库
`TencentVectorDB`	腾讯向量数据库，当前类还在community包下，没有独立出来
`PineconeVectorStore`	Pinecone向量数据库
`WeaviateVectorStore`	Weaviate向量数据库

4.2 RedisVectorStore数据存储

RedisVectorStore是 VectorStore 接口的一个实现类，下面将以它为例介绍 VectorStore 的用法。在使用 WeaviateVectorStore 之前，需要先安装 langchain-redis 依赖：

RedisVectorStore数据存储

pip install langchain-redis

Redis 向量存储可参考文档：https://www.langchain.com.cn/docs/integrations/vectorstores/redis/ Ollama Embedding 可参考文档：https://python.langchain.com/api_reference/community/embeddings/langchain_community.embeddings.ollama.OllamaEmbeddings.html 接下来，先展示如何Ollama Embedding 转为词向量后使用 RedisVectorStore 进行数据存储，示例程序如下

RedisVectorStore示例

from langchain_ollama import OllamaEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore
import dotenv

# 读取env配置
dotenv.load_dotenv()

# 初始化 Embedding 模型
embedding = OllamaEmbeddings(model="deepseek-r1:14b")

# ========== 存储数据 ==========
# 定义待处理的文本数据列表
texts = [
    "我喜欢吃苹果",
    "苹果是我最喜欢吃的水果",
    "我喜欢用苹果手机",
]

# 获取文本向量
# 使用embedding模型将文本转换为向量表示
embeddings = embedding.embed_documents(texts)

# 打印结果
# 遍历并打印每个文本及其对应的向量信息
for i, vec in enumerate(embeddings, 1):
    print(f"文本 {i}: {texts[i-1]}")
    print(f"向量长度: {len(vec)}")
    print(f"前5个向量值: {vec[:10]}\n")

# 定义每条文本对应的元数据信息
metadata = [{"segment_id": "1"}, {"segment_id": "2"}, {"segment_id": "3"}]

# 配置Redis连接参数和索引名称
config = RedisConfig(
    index_name="newsgroups",
    redis_url="redis://localhost:6379",
)

# 创建Redis向量存储实例
vector_store = RedisVectorStore(embedding, config=config)

# 将文本和元数据添加到向量存储中
ids = vector_store.add_texts(texts, metadata)

# 打印前5个存储记录的ID
print(ids[0:5])

打印前5个存储记录的ID

17:12:27 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"
文本 1: 我喜欢吃苹果
向量长度: 5120
前5个向量值: [-0.0048059775, 0.0110130375, 0.0008511835, 0.00023040122, 0.0031677634, 0.00023193852, -0.0036830187, 0.00031026654, 0.003424279, 0.002071732]
文本 2: 苹果是我最喜欢吃的水果
向量长度: 5120
前5个向量值: [-0.016867071, 0.006843345, 0.0056118295, -0.012666135, -0.00062124344, 0.0024224545, -0.0016708882, -0.0017295848, 0.01790834, 0.0042891167]
文本 3: 我喜欢用苹果手机
向量长度: 5120
前5个向量值: [0.0014743459, 0.0019745259, 0.003218666, 0.0019062049, -0.008469705, -0.003524136, -0.0010743507, -0.0026532735, -0.0065791565, 0.0043244734]
17:12:27 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"
17:12:27 redisvl.index.index INFO   Index already exists, not overwriting.
17:12:27 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"
['newsgroups::01K4HQ5H2JJ5Z5C3KW6RZVMFWZ', 'newsgroups::01K4HQ5H2JJ5Z5C3KW6RZVMFX0', 'newsgroups::01K4HQ5H2JJ5Z5C3KW6RZVMFX1']

4.3 RedisVectorStore数据检索

在上面的示例程序中，我们将文本信息和元数据信息都保存到了数据库中。接下来，使用 VectorStore的similarity_search_with_relevance_scores() 方法进行相似性检索。在调用该方法时，传入查询文本 query，并指定 k=3，即返回匹配分数最高的三条数据(k 的默认值为 4)

similarity_search_with_relevance_scores示例

from langchain_ollama import OllamaEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore
import dotenv

# 读取env配置
dotenv.load_dotenv()

# 初始化 Embedding 模型
embedding = OllamaEmbeddings(model="deepseek-r1:14b")

# 配置Redis连接参数和索引名称
config = RedisConfig(
    index_name="newsgroups",
    redis_url="redis://localhost:6379",
)

# 创建Redis向量存储实例
vector_store = RedisVectorStore(embedding, config=config)

# ========== 查询数据 ==========
# 定义查询文本
query = "我喜欢用什么手机"

# 将查询语句向量化，并在Redis中做相似度检索
results = vector_store.similarity_search_with_score(query, k=3)

print("=== 查询结果 ===")
for i, (doc, score) in enumerate(results, 1):
    similarity = 1 - score  #  score 是距离，可以转成相似度
    print(f"结果 {i}:")
    print(f"内容: {doc.page_content}")
    print(f"元数据: {doc.metadata}")
    print(f"相似度: {similarity:.4f}")

执行结果如下，返回了3个与查询文本最相关的文本信息

=== 查询结果 ===
结果 1:
内容: 我喜欢用苹果手机
元数据: {'segment_id': '3'}
相似度: 0.9202
结果 2:
内容: 我喜欢吃苹果
元数据: {'segment_id': '1'}
相似度: 0.8157
结果 3:
内容: 苹果是我最喜欢吃的水果
元数据: {'segment_id': '2'}
相似度: 0.6804

5. Retrievers检索器组件

5.1 BaseRetriever接口

BaseRetriever 是检索器相关类的顶层接口。当给定一个查询文本需要进行非结构化查询时，它比 VectorStore 更为通用。检索器本身不需要存储文档，只要能够对文档进行检索并返回检索到的文档即可。检索器可以通过 VectorStore 创建，也可以对诸如维基百科等数据源进行检索在 langchain 项目中， VectorStore 是底层存储接口，负责和具体的向量数据库交互(比如 Redis、Weaviate、Milvus、Pinecone 等) ，而 VectorStoreRetriever 是 LangChain 的 Retriever 抽象(Retriever 是统一的“检索器”接口，用于在链/Agent 中做检索) 最重要的是，BaseRetriever 是一个可运行组件，它可以方便地使用 LECL 表达式对检索器组件进行集成。检索器接受一个查询文本作为输入，返回一个 Document 对象列表作为输出。BaseRetriever 的 invoke 方法定义如下：

BaseRetriever.invoke示例

def invoke(
    self, input: str, config: Optional[RunnableConfig] = None, **kwargs: Any
) -> List[Document]:

5.2 VectorStoreRetriever使用

在 RAG 应用中，当需要基于向量数据库进行文档检索时，就可以使用VectorStoreRetriever。它封装了向量数据库检索的底层逻辑，能够直接调用 VectorStore 的方法，从向量数据库中检索最相关的文档在前面介绍 VectorStore 常用方法时，包括了 as_retriever() 方法，该方法可以构建一个检索器对象，这个检索器就是 VectorStoreRetriever 使用示例如下，使用as_retriever()方法创建了一个VectorStoreRetriever 对象，之后调用invoke()方法传入query进行文档检索

as_retriever示例

from langchain_ollama import OllamaEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore
import dotenv

# 读取env配置
dotenv.load_dotenv()

# 初始化 Embedding 模型
embedding = OllamaEmbeddings(model="deepseek-r1:14b")

# 配置Redis连接参数和索引名称
config = RedisConfig(
    index_name="newsgroups",
    redis_url="redis://localhost:6379",
)

# 创建Redis向量存储实例
vector_store = RedisVectorStore(embedding, config=config)

# 创建检索器，进行数据检索
retriever = vector_store.as_retriever()
documents = retriever.invoke("介绍一下我喜欢用什么手机")

for document in documents:
    print(document.page_content)
    print(document.metadata)
    print("=================================")

创建检索器，进行数据检索

我喜欢用苹果手机
{'segment_id': '3'}
=================================
我喜欢吃苹果
{'segment_id': '1'}
=================================
苹果是我最喜欢吃的水果
{'segment_id': '2'}
=================================

在默认情况下，检索器使用相似性检索方式进行检索。另一种检索方式是最大边际相关性检索(简称 MMR)，可以在调用 as_retriever() 方法时通过 search_type="mmr" 指定，但前提是检索器所使用的底层数据库必须支持该检索方式

创建检索器，进行数据检索

retriever = vector_store.as_retriever(search_type="mmr")

除了 search_type 之外，还可以使用 search_kwargs 将参数传递给 VectorStore 的底层搜索方法，例如传递 k 值，将默认匹配度最高的前三个文档返回(默认 k=4)

创建检索器，进行数据检索

retriever = vector_store.as_retriever(search_type="mmr", search_kwargs={"k": 3})

5.3 MultiQueryRetriever使用

在向量检索过程中，查询文本会被转换为向量，并通过计算向量间距离来检索相似文档。然而，检索结果的准确性可能会受到查询文本表达方式的影响因此，为了提升查询结果的准确性，可以将查询文本传递给大语言模型，由其生成多个不同表达方式的查询文本变体。随后，使用这些不同的查询文本分别进行文档检索，并将所有检索结果汇总、排序，返回最相关的文档 MultiQueryRetriever(多查询检索器)正是实现上述 RAG 检索优化逻辑的工具。可以使用 MultiQueryRetriever.from_llm() 方法创建一个多查询检索器。进入 from_llm() 源码可以看到，除了需要传递检索器对象和模型对象之外，还可以传入 prompt 参数，该参数用于调用大模型生成多个查询文本的提示词，并提供了默认值

该默认提示词为英文版，在使用时需要进行汉化，否则返回的查询文本将全部为英文，导致检索效果下降。

MultiQueryRetriever 使用示例如下，首先进行了日志设置，在调用大语言模型生成多个查询文本时，MultiQueryRetriever 会进行 INFO 级别的日志打印，将生成的文本输出，在创建 MultiQueryRetriever 时，需要传入 BaseRetriever 对象、模型对象以及汉化后的 prompt，之后同样通过调用invoke()方法传入查询文本进行检索

MultiQueryRetriever使用

from langchain.retrievers import MultiQueryRetriever
from langchain_core.prompts import PromptTemplate
from langchain_ollama import OllamaEmbeddings, ChatOllama
from langchain_redis import RedisConfig, RedisVectorStore
import dotenv

# 读取env配置
dotenv.load_dotenv()

# 初始化 Embedding 模型
embedding = OllamaEmbeddings(model="deepseek-r1:14b")
# 初始化大语言模型
llm = ChatOllama(model="deepseek-r1:14b", reasoning=False)
# 配置Redis连接参数和索引名称
config = RedisConfig(
    index_name="newsgroups",
    redis_url="redis://localhost:6379",
)

# 创建Redis向量存储实例
vector_store = RedisVectorStore(embedding, config=config)

# 创建多查询检索器
retriever = vector_store.as_retriever()
retriever_from_llm = MultiQueryRetriever.from_llm(
    retriever=retriever, llm=llm,
    prompt=PromptTemplate(
        input_variables=["question"],
        template="""你是一个 AI 语言模型助手。你的任务是：
        为给定的用户问题生成 3 个不同的版本，以便从向量数据库中检索相关文档。
        通过生成用户问题的多种视角(改写版本)，
        你的目标是帮助用户克服基于距离的相似性搜索的某些局限性。
        请将这些改写后的问题用换行符分隔开。原始问题：{question}""")
)

# 5.进行数据检索
documents = retriever_from_llm.invoke("介绍一下我喜欢使用的手机")

for document in documents:
    print(document.page_content)
    print(document.metadata)
    print("=================================")

执行结果如下。通过日志可以观察到，LLM 生成了三个查询文本，并且最终检索结果中排在前面的前两条文档与查询文本的信息最为相关

进行数据检索

17:31:41 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"
17:31:41 redisvl.index.index INFO   Index already exists, not overwriting.
17:31:41 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/chat "HTTP/1.1 200 OK"
17:31:42 langchain.retrievers.multi_query INFO   Generated queries: ['你常用的手机有哪些特点？  ', '你能描述一下你平时使用的一款手机吗？  ', '你最喜欢使用的手机是什么样子的？']
17:31:42 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"
17:31:42 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"
17:31:42 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"
苹果是我最喜欢吃的水果
{'segment_id': '2'}
=================================
我喜欢吃苹果
{'segment_id': '1'}
=================================
我喜欢用苹果手机
{'segment_id': '3'}
=================================

5.4 自定义检索器实现

在前面已经介绍过 BaseRetriever 接口，我们可以通过继承 BaseRetriever 来实现自定义检索器。查看 BaseRetriever 的 invoke 方法(省略部分代码)可以发现，最终真正执行检索的核心方法是 _get_relevant_documents

自定义检索器实现

def invoke(
    self, input: str, config: Optional[RunnableConfig] = None, **kwargs: Any
) -> List[Document]:
......
try:
    _kwargs = kwargs if self._expects_other_args else {}
    if self._new_arg_supported:
        result = self._get_relevant_documents(
            input, run_manager=run_manager, **_kwargs
        )
    else:
        result = self._get_relevant_documents(input, **_kwargs)
except Exception as e:
    run_manager.on_retriever_error(e)
    raise e
else:
    run_manager.on_retriever_end(
        result,
    )
    return result

并且 _get_relevant_documents 是一个抽象方法，需要由子类去实现

自定义检索器实现

@abstractmethod
def _get_relevant_documents(
    self, query: str, *, run_manager: CallbackManagerForRetrieverRun
) -> List[Document]:

因此，实现一个自定义检索器需要继承 BaseRetriever 并实现 _get_relevant_documents 方法假设有如下需求：需要一个自定义检索器，将传入的查询文本按空格拆分成关键词数组，并在文档中进行匹配。只要有任意一个关键词匹配成功，即返回该文档信息，同时支持通过传递参数控制检索器返回的文档数量

实现该需求的代码示例

from typing import List

from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever

class KeywordsRetriever(BaseRetriever):
    """自定义检索器

    该检索器根据查询中的关键词来检索相关文档，支持返回前k个匹配的文档

    Attributes:
        documents: 文档列表，用于检索的文档集合
        k: 返回文档数量，指定最多返回多少个相关文档
    """
    documents: List[Document]
    k: int

    def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> List[Document]:
        """根据查询关键词检索相关文档

        Args:
            query: 查询字符串，将被拆分为多个关键词进行匹配
            run_manager: 回调管理器，用于处理检索过程中的回调

        Returns:
            List[Document]: 包含匹配文档的列表，最多返回k个文档
        """
        # 获取返回文档数量参数
        k = self.k if self.k is not None else 3
        documents_result = []

        # 将查询字符串按空格拆分为关键词列表
        query_keywords = query.split(" ")

        # 遍历所有文档，筛选包含任一关键词的文档
        for document in self.documents:
            if any(query_keyword in document.page_content for query_keyword in query_keywords):
                documents_result.append(document)

        # 返回前k个匹配的文档
        return documents_result[:k]

# 定义文档列表，包含用于检索的文本内容
documents = [
    Document("苹果是我最喜欢吃的水果"),
    Document("我喜欢吃苹果"),
    Document("我喜欢用苹果手机"),
]

# 创建关键词检索器实例，设置文档集合和返回文档数量
retriever = KeywordsRetriever(documents=documents, k=1)

# 执行检索操作，根据查询"手机"查找相关文档
result = retriever.invoke("手机")

# 输出检索结果，打印匹配文档的内容
for document in result:
    print(document.page_content)
    print("===========================")

输出检索结果，打印匹配文档的内容

我喜欢用苹果手机
===========================

6. RAG 实战项目

6.1 项目介绍

6.1.1 项目背景

随着美团业务的不断扩展，客服人员需要应对海量的用户咨询，包括订单问题、退款流程、配送异常、优惠政策等。传统的知识库客服系统依赖规则匹配，回答僵硬，难以及时覆盖最新的业务规则为提升客户体验和客服效率，本项目基于 RAG(Retrieval-Augmented Generation，检索增强生成) 技术构建智能客服问答系统，将美团内部文档知识与大语言模型结合，实现更智能、更准确的自动化答复

6.1.2 项目功能

针对智能客服系统本身，我们将采用RAG + LLM来完成这一需求。结合前面所学的知识，这个智能客服系统应该具备以下功能：

支持历史记忆功能，并且能够实现历史记忆持久化
使用LCEL 表达式来构建链
支持RAG 检索功能，使大语言模型能够根据知识库文档内容进行作答
编写完善的提示词模板，内容包括历史对话信息、RAG 检索的上下文信息、用户提问，以及AI 作为客服的系统提示词

6.1.3 系统架构

6.2 RAG 准备阶段

在 RAG 准备阶段，我们需要进行文档收集、文档处理、文档数据向量化操作以及文档相似性检索测试

6.2.1 文档收集

收集美团客服相关知识文档，例如：

业务手册(退款规则、订单处理流程)
常见问题 FAQ
内部客服知识库
实时更新的运营公告

以美团外卖常见问题为例，文档地址：https://waimai.meituan.com/help/faq，我们通过playwright 工具爬虫获取页面数据并写入本地 txt 文件中

安装依赖包

# 安装浏览器插件库
pip install playwright chromium
playwright install
# 安装浏览器中文依赖
sudo apt update && sudo apt install fonts-wqy-zenhei fonts-wqy-microhei -y

文档收集

from playwright.sync_api import sync_playwright

def collect_faq(url):
    """
    收集指定URL页面中的FAQ内容

    参数:
        url (str): 目标网页URL地址

    返回:
        str: 提取的FAQ内容html代码
    """
    # 启动Playwright浏览器自动化工具
    with sync_playwright() as p:
        # 启动Chromium浏览器，设置为非无头模式并指定中文语言
        browser = p.chromium.launch(
            headless=False,
            args=['--lang=zh-CN']  # 浏览器语言
        )
        # 创建新页面，配置中文环境
        page = browser.new_page(
            locale='zh-CN',  # 页面 locale
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/139.0.0.0 Safari/537.36 Edg/139.0.0.0"
            ),
            extra_http_headers={
                "Accept-Language": "zh-CN,zh;q=0.9"
            }
        )
        # 访问目标URL并等待页面加载完成
        page.goto(url, timeout=30_000)
        page.wait_for_load_state("networkidle")

        # 提取FAQ列表区域的html代码
        raw_text = page.locator("#faq-list").first.inner_html()
        browser.close()
        return raw_text

def save_faq(cleaned_text: str, output_file: str):
    """
    将FAQ内容保存到指定文件

    参数:
        cleaned_text (str): 要保存的FAQ内容
        output_file (str): 输出文件路径
    """
    # 写入文件
    with open(output_file, "w", encoding="utf-8") as f:
        f.write(cleaned_text)

    print(f"FAQ 已保存到 {output_file}")

if __name__ == "__main__":
    cleaned_text = collect_faq(url="https://waimai.meituan.com/help/faq")
    output_file = "faq.html"
    save_faq(cleaned_text, output_file)

原始文件内容

# cat faq.html
  <ul>
        <li class="faq-head head1">
          <h1>在线支付问题</h1>
          <span></span>
        </li>
        <li>
          <dl>
            <dt><a href="javascript:;" class="questions">Q：在线支付取消订单后钱怎么返还？<i class="icon i-triangledown fr"></i></a></dt>
            <dd class="answers hidden ">
              订单取消后，款项会在一个工作日内，直接返还到您的美团账户余额。
            </dd>
          </dl>
        </li>
    ……

6.2.2 文档处理

我们已经爬取了 FAQ 文档，接下来就需要对收集到的文档进行统一处理，内容包括：

文本清洗(去除 HTML 标签、无关字符)
分段切分(按规则或语义将文档拆分成小片段，便于检索)
元数据标注(来源、时间、业务类别等)

文档处理

import json
from bs4 import BeautifulSoup
from langchain.schema import Document

def parse_faq_html(file_path):
    """
    解析FAQ HTML文件，提取问题和答案信息并封装为Document对象列表。

    参数:
        file_path (str): FAQ HTML文件的路径。

    返回:
        list: 包含Document对象的列表，每个对象的metadata包含分类、问题、答案和来源信息。
    """
    docs = []
    with open(file_path, "r", encoding="utf-8") as f:
        soup = BeautifulSoup(f, "html.parser")

    current_category = None

    # 遍历所有<ul>标签，解析其中的<li>元素
    for ul in soup.find_all("ul"):
        for li in ul.find_all("li", recursive=False):
            h1 = li.find("h1")
            if h1:  # 分类标题
                current_category = h1.get_text(strip=True)
                continue

            dl = li.find("dl")
            if dl:
                # 去掉 Q：
                question_raw = dl.find("dt").get_text(strip=True)
                question = question_raw.lstrip("Q：").strip()
                answer = dl.find("dd").get_text(strip=True)

                docs.append(
                    Document(
                        page_content="",
                        metadata={
                            "source": file_path,
                            "category": current_category,
                            "question": question,
                            "answer": answer
                        }
                    )
                )
    return docs

def save_docs_to_json(docs, output_file):
    """
    将Document对象列表保存为JSON格式文件。

    参数:
        docs (list): 包含Document对象的列表。
        output_file (str): 输出JSON文件的路径。

    返回:
        None
    """
    data = [
        {
            "question": doc.metadata["question"],
            "answer": doc.metadata["answer"],
            "category": doc.metadata["category"],
            "source": doc.metadata["source"]
        }
        for doc in docs
    ]
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    print(f"FAQ 已保存到 {output_file}")

if __name__ == "__main__":
    faq_docs = parse_faq_html("faq.html")
    for d in faq_docs:
        print(d.metadata)
    save_docs_to_json(faq_docs, "faq.json")

执行后的结果如下：

执行结果如下

# cat faq.json
[
  {
    "question": "在线支付取消订单后钱怎么返还？",
    "answer": "订单取消后，款项会在一个工作日内，直接返还到您的美团账户余额。",
    "category": "在线支付问题",
    "source": "faq.html"
  },
  ……

6.2.3 文档数据向量化

我们将 FAQ 数据格式化成 json 数据后，接下来就要转成向量数据并存储到向量数据库中，此处以 redis 为例，操作内容包括：

使用 向量化模型(Embedding Model，如 BGE、OpenAI Embedding) 将文档片段转换为向量表示
存储至向量数据库(如 Milvus、Weaviate、Redis Vector、Faiss)，支持高效的相似度搜索

代码如下

文档数据向量化

import json
from langchain_ollama import OllamaEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore

def insert_faq(texts, meta_data):
    """
    将FAQ文本数据插入到Redis向量存储中

    Args:
        texts (list): 包含问题文本的列表
        meta_data (list): 包含每个问题对应元数据的列表，每个元素为字典格式

    Returns:
        None
    """
    # 配置Redis连接参数和索引名称
    config = RedisConfig(
        index_name="faq",
        redis_url="redis://localhost:6379",
    )
    # 初始化 Embedding 模型
    embedding = OllamaEmbeddings(model="deepseek-r1:14b")
    # 创建Redis向量存储实例
    vector_store = RedisVectorStore(embedding, config=config)
    vector_store.add_texts(texts=texts, metadatas=meta_data)

def insert_from_file(file_path):
    """
    从JSON文件中读取FAQ数据并插入到向量存储中

    Args:
        file_path (str): 包含FAQ数据的JSON文件路径

    Returns:
        None
    """
    with open(file_path, "r", encoding="utf-8") as f:
        docs = json.load(f)
    texts = []
    meta_data = []
    # 解析文档数据，提取问题文本和元数据
    for doc in docs:
        texts.append(doc["question"])
        meta_data.append({
            "answer": doc["answer"],
            "category": doc["category"],
            "source": doc["source"]
        })
    insert_faq(texts, meta_data)

if __name__ == "__main__":
    # 程序入口：先创建索引再批量插入数据
    insert_from_file("faq.json")

6.2.4 文档数据相似性检索

文档向量数据写入数据库后，接下来就是测试验证召回数据准确性，主要内容包括：

用户提问后，将问题转换为向量，与向量数据库中的文档进行相似性匹配
召回与问题最相关的文档片段(如退款流程、配送延误规则)，并返回给上层系统

文档数据相似性检索

from langchain_ollama import OllamaEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore

def search_question(question):
    # 初始化 Embedding 模型
    embedding = OllamaEmbeddings(model="deepseek-r1:14b")

    # 配置Redis连接参数和索引名称
    config = RedisConfig(
        index_name="faq",
        redis_url="redis://localhost:6379",
    )

    # 创建Redis向量存储实例
    vector_store = RedisVectorStore(embedding, config=config)

    # 创建检索器，进行数据检索
    retriever = vector_store.as_retriever()
    documents = retriever.invoke(question)

    for document in documents:
        print(document.page_content)
        print(document.metadata)
        print("=================================")

if __name__ == "__main__":
    # 程序入口：先创建索引再批量插入数据
    search_question("在线支付取消订单后钱怎么返还给我呢")

文档数据相似性检索

21:02:58 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"
21:02:58 redisvl.index.index INFO   Index already exists, not overwriting.
21:02:58 httpx INFO   HTTP Request: POST http://127.0.0.1:11434/api/embed "HTTP/1.1 200 OK"
在线支付取消订单后钱怎么返还？
{'answer': '订单取消后，款项会在一个工作日内，直接返还到您的美团账户余额。', 'category': '在线支付问题', 'source': 'faq.html'}
=================================
在线支付的过程中，订单显示未支付成功，款项却被扣了，怎么办？
{'answer': '出现此问题，可能是银行/支付宝的数据没有即时传输至美团，请您不要担心，稍后刷新页面查看。 如半小时后仍显示"未付款"，请先联系银行/支付宝客服，获取您扣款的交易号，然后致电美团外卖客服4008507777，我们会协助您解决。', 'category': '在线支付问题', 'source': 'faq.html'}
=================================
在线支付订单如何退款？
{'answer': '商家接单前，您可以直接取消订单，订单金额会自动退款到美团余额；商家接单后，您在点击“申请退款”，在线申请。提交退款申请之后，商家有24小时处理您的退款申请。商家同意退款，或24小时内没有处理您的退款申请，您的支付金额会退款至您的美团余额。', 'category': '在线支付问题', 'source': 'faq.html'}
=================================
美团账户里的余额怎么提现？
{'answer': '余额可到美团网(meituan.com)——“我的美团→美团余额”里提取到您的银行卡或者支付宝账号，另外，余额也可直接用于支付外卖订单(限支持在线支付的商家)。', 'category': '在线支付问题', 'source': 'faq.html'}
=================================

6.2.5 构建提示词

把 用户问题 + 检索召回的上下文 拼接成一个高质量的 Prompt 送给大模型
提示词示例：

构建提示词

你是一个外卖公司的智能客服，接下来你将扮演一个专业客服的角色，对用户提出来的商品问题进行回答，一定要礼貌热情，如果用户提问与客服和商品无关的问题，礼貌委婉的表示拒绝或无法回答，只回答外卖服务相关的问题。

用户问题：
取消订单后多久能收到退款？

可用文档片段：
【文档片段1】
Q: 在线支付取消订单后钱怎么返还？
A: 订单取消后，款项会在一个工作日内，直接返还到您的美团账户余额。

【文档片段2】
Q: 怎么查看退款是否成功？
A: 退款会在一个工作日之内到美团账户余额，可在“账号管理——我的账号”中查看是否到账。

请基于以上信息，生成简洁明了的回答：

构建提示词

from langchain_ollama import OllamaEmbeddings
from langchain_redis import RedisConfig, RedisVectorStore
from langchain_core.prompts import PromptTemplate

def build_prompt(question: str):
    """
    使用向量检索技术查找相关文档，并通过 LangChain PromptTemplate 构造提示词。

    参数:
        question (str): 用户提出的问题。

    返回:
        str: 构造完成的提示词字符串。
    """
    # 初始化 Embedding 模型
    embedding = OllamaEmbeddings(model="deepseek-r1:14b")

    # Redis 配置
    config = RedisConfig(
        index_name="faq",
        redis_url="redis://localhost:6379",
    )

    # 创建 Redis 向量存储实例
    vector_store = RedisVectorStore(embedding, config=config)

    # 创建检索器，取 2 个最相关文档
    retriever = vector_store.as_retriever(search_kwargs={"k": 2})
    documents = retriever.invoke(question)

    # 组装 context
    context = "\n\n".join(
        f"【文档片段{i + 1}】\nQ: {doc.page_content}\nA: {doc.metadata.get('answer', '')}"
        for i, doc in enumerate(documents)
    )

    # 定义 Prompt 模板
    template = """
    你是一个外卖公司的智能客服，接下来你将扮演一个专业客服的角色，
    对用户提出来的商品问题进行回答，一定要礼貌热情，如果用户提问与客服和商品无关的问题，
    礼貌委婉的表示拒绝或无法回答，只回答外卖服务相关的问题。

    用户问题：
    {question}

    可用文档片段：
    {context}

    请基于以上信息，生成简洁明了的回答：
    """
    prompt_template = PromptTemplate(
        input_variables=["question", "context"], template=template
    )

    # 渲染提示词
    prompt = prompt_template.format(question=question, context=context)

    print("=== 提示词 ===")
    print(prompt)
    print("=================================")
    return prompt

if __name__ == "__main__":
    build_prompt("在线支付取消订单后钱怎么返还给我呢")

构建提示词

=== 提示词 ===

你是一个外卖公司的智能客服，接下来你将扮演一个专业客服的角色，
对用户提出来的商品问题进行回答，一定要礼貌热情，如果用户提问与客服和商品无关的问题，
礼貌委婉的表示拒绝或无法回答，只回答外卖服务相关的问题。

用户问题：
在线支付取消订单后钱怎么返还给我呢

可用文档片段：
【文档片段1】
Q: 在线支付取消订单后钱怎么返还？
A: 订单取消后，款项会在一个工作日内，直接返还到您的美团账户余额。

【文档片段2】
Q: 在线支付取消订单后钱怎么返还？
A: 订单取消后，款项会在一个工作日内，直接返还到您的美团账户余额。

请基于以上信息，生成简洁明了的回答：

=================================

6.3 RAG 系统实现

6.3.1 主要步骤

接下来，开始实现智能客服系统，主要包含以下 8 个步骤：

创建提示词模板

模板包括 系统消息、消息占位符、人类消息

系统消息：设置 AI 的身份和当前业务场景
消息占位符：传递聊天历史
人类消息：传递用户提问以及通过 RAG 检索到的上下文信息

构建模型

使用 deepseek-r1:14b 模型

创建输出解析器

创建一个 字符串输出解析器，用于结果输出

构建检索器

连接 Weaviate 数据库，创建 WeaviateVectorStore 对象，并传入 文本嵌入对象、Weaviate 客户端对象、存储文本信息 key、集合名称然后调用 WeaviateVectorStore.as_retriever() 方法生成检索器，并指定只返回一条最相关的文档数据

创建记忆组件

构建记忆组件，并将历史对话信息保存在 customer_service_history.txt 中

构建链

构建 LCEL 链。链的后半部分较为直观，这里重点介绍前半部分由于检索器需要接收一个字符串参数，因此使用字典进行构建：将检索器的输出信息通过 format_documents() 方法拼接成字符串，作为 context 参数，同时添加 query 参数，供下一个可运行组件使用这里利用了 RunnableParallel 的参数传递功能。在 LCEL 表达式中，使用字典结构包裹并通过管道符连接时，会自动被包装成 RunnableParallel

调用链

使用 stream() 方法调用链，传入用户提问stream() 可以实现流式输出，相比一次性返回结果，用户体验更好

保存记忆

调用 save_context()，将对话记忆进行持久化

6.3.2 代码编写

代码编写

from langchain_core.output_parsers import StrOutputParser
from langchain_ollama import OllamaEmbeddings, ChatOllama
from langchain_redis import RedisConfig, RedisVectorStore, RedisChatMessageHistory
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnableWithMessageHistory

# ---------- 工具 ----------

def format_docs(docs):
    """
    把检索到的文档格式化成上下文字符串，用于提供给语言模型作为参考信息。

    参数:
        docs (list): 文档对象列表，每个对象应包含 page_content 和 metadata 属性。

    返回:
        str: 格式化后的字符串，包含多个文档片段及其问答内容。
    """
    return "\n\n".join(
        f"【文档片段{i + 1}】\n"
        f"Q: {doc.page_content}\n"
        f"A: {doc.metadata.get('answer', '')}"
        for i, doc in enumerate(docs)
    )

def extract_question(x: str | list) -> str:
    """
    从 RunnableWithMessageHistory 的输入中提取用户的纯文本问题。

    参数:
        x (str | list): 输入可以是字符串或消息对象列表。

    返回:
        str: 提取到的用户问题文本。
    """
    if isinstance(x, str):
        return x
    # x 是 list[HumanMessage]
    return x[-1].content

# ---------- 构建链 ----------

def build_chain():
    """
    构建一个基于检索增强生成(RAG)的对话链，用于智能客服问答。

    返回:
        Chain: 一个可调用的 LangChain 链对象，用于处理用户问题并生成回答。
    """
    # 初始化嵌入模型
    embedding = OllamaEmbeddings(model="deepseek-r1:14b")

    # 配置 Redis 向量存储
    config = RedisConfig(index_name="faq", redis_url="redis://localhost:6379")
    vector_store = RedisVectorStore(embedding, config=config)

    # 创建文档检索器，最多返回2个相关文档
    retriever = vector_store.as_retriever(search_kwargs={"k": 2})

    # 定义提示模板
    template = """
    你是一个外卖公司的智能客服，接下来你将扮演一个专业客服的角色，
    对用户提出来的商品问题进行回答，一定要礼貌热情，如果用户提问与客服和商品无关的问题，
    礼貌委婉的表示拒绝或无法回答，只回答外卖服务相关的问题。

    用户问题：
    {question}

    可用文档片段：
    {context}

    请基于以上信息，生成简洁明了的回答：
    """
    prompt = PromptTemplate.from_template(template)

    # 初始化语言模型和输出解析器
    llm = ChatOllama(model="deepseek-r1:14b", reasoning=False)
    parser = StrOutputParser()

    # 构建处理链：提取问题 -> 检索文档 -> 格式化上下文 -> 拼接提示 -> 调用模型 -> 解析输出
    chain = (
        {
            "context": extract_question | retriever | format_docs,
            "question": extract_question,
        }
        | prompt
        | llm
        | parser
    )
    return chain

# ---------- 交互 ----------

def main():
    """
    主函数，启动智能客服交互系统。
    """
    # 构建对话链
    chain = build_chain()

    # 初始化 Redis 聊天历史记录
    history = RedisChatMessageHistory(session_id='rag', redis_url='redis://localhost:6379/0')

    # 将对话链包装为带历史记录的可运行对象
    runnable = RunnableWithMessageHistory(
        chain,
        get_session_history=lambda: history
    )

    # 启动交互循环
    print(">>> 欢迎使用外卖智能客服系统，输入 quit/exit 退出 <<<")
    while True:
        try:
            user = input("\n您：").strip()
        except (KeyboardInterrupt, EOFError):
            print("\nbye~")
            break
        if user.lower() in {"quit", "exit", "q"}:
            print("客服：祝您生活愉快，再见！")
            break
        answer = runnable.invoke(user)      # 自动把 user 包装成 HumanMessage
        print("客服：", answer)

if __name__ == "__main__":
    main()

执行效果如下：

执行结果如下

>>> 欢迎使用外卖智能客服系统，输入 quit/exit 退出 <<<
您：今天天气怎么样
客服： 您好，我是外卖公司的智能客服。关于天气问题，我无法提供相关信息。如需查询天气，请您开启天气应用查看实时情况哦！如果有任何外卖服务相关的问题，我会很乐意为您提供帮助。

您：在线支付取消订单后钱怎么返还给我呢
客服： 您好，关于在线支付取消订单后的退款问题，请您放心，订单取消后，款项会在一个工作日内直接返还到您的美团账户余额。如有任何疑问或需要进一步帮助，请随时联系我们。感谢您的理解与支持！

您：重复回答
客服： 您好，关于在线支付取消订单后的退款问题，请您放心，订单取消后，款项会在一个工作日内直接返还到您的美团账户余额。如有任何疑问或需要进一步帮助，请随时联系我们。感谢您的理解与支持！

​LangChain 实现 RAG

​1. LangChain RAG 组件与流程

​1.1 LangChain组件

​1.2 LangChain 实现流程

​2. 文档加载器

​2.1 LangChain 文档加载器

​2.2 Document文档类

​2.3 文档加载器使用

​2.4 自定义文档加载器

​3. 文本分割器

​3.1 LangChain 文本分割器

​3.2 递归文本分割器用法

​3.3 分割文本

​3.4 分割文档对象

​3.5 自定义分隔符

​3.6 按标题分割Markdown文件

​3.7 自定义文本分割器

​4. VectorStore存储组件

​4.1 VectorStore组件介绍

​4.2 RedisVectorStore数据存储

​4.3 RedisVectorStore数据检索

​5. Retrievers检索器组件

​5.1 BaseRetriever接口

​5.2 VectorStoreRetriever使用

​5.3 MultiQueryRetriever使用

​5.4 自定义检索器实现

​6. RAG 实战项目

​6.1 项目介绍

​6.1.1 项目背景

​6.1.2 项目功能

​6.1.3 系统架构

​6.2 RAG 准备阶段

​6.2.1 文档收集

​6.2.2 文档处理

​6.2.3 文档数据向量化

​6.2.4 文档数据相似性检索

​6.2.5 构建提示词

​6.3 RAG 系统实现

​6.3.1 主要步骤

​6.3.2 代码编写

LangChain 实现 RAG

1. LangChain RAG 组件与流程

1.1 LangChain组件

1.2 LangChain 实现流程

2. 文档加载器

2.1 LangChain 文档加载器

2.2 Document文档类

2.3 文档加载器使用

2.4 自定义文档加载器

3. 文本分割器

3.1 LangChain 文本分割器

3.2 递归文本分割器用法

3.3 分割文本

3.4 分割文档对象

3.5 自定义分隔符

3.6 按标题分割Markdown文件

3.7 自定义文本分割器

4. VectorStore存储组件

4.1 VectorStore组件介绍

4.2 RedisVectorStore数据存储

4.3 RedisVectorStore数据检索

5. Retrievers检索器组件

5.1 BaseRetriever接口

5.2 VectorStoreRetriever使用

5.3 MultiQueryRetriever使用

5.4 自定义检索器实现

6. RAG 实战项目

6.1 项目介绍

6.1.1 项目背景

6.1.2 项目功能

6.1.3 系统架构

6.2 RAG 准备阶段

6.2.1 文档收集

6.2.2 文档处理

6.2.3 文档数据向量化

6.2.4 文档数据相似性检索

6.2.5 构建提示词

6.3 RAG 系统实现

6.3.1 主要步骤

6.3.2 代码编写