掌握RAG：深入探讨文本分割技巧

在这里，我们将尝试在这里掌握各个对成功实施RAG至关重要的主题。这里有一个RAG架构的例子。

感谢 Dipanjan

我们先来谈谈一个叫做“文本分割”的概念。

在这里，我们可能需要考虑输入数据加载后，根据不同的文本分割方法被分割成不同的块。我们先来看看不同的文本分割方法，并对每种方法对应的向量存储进行比较。

简单来说，文本分割器是如何工作的如下所示：

1. 将文本分成小的、有意义的语义片段（通常是句子）。
2. 开始将这些小段落组合成更大的段落，直到达到某个大小（通过某个函数来测量）。
3. 当达到该大小时，将这一部分作为独立的文本片段，并开始创建新的文本片段，让它与前一部分有重叠（以便保持片段之间的上下文）。

这意味着你有两个不同的方式在两个不同的维度上自定义你的文本分割器的方式，进行调整。

1. 文本是怎么分割的
2. 分块的大小是如何测量的

引入库：

    import os  

    from langchain.text_splitter import (  
        CharacterTextSplitter,  
        RecursiveCharacterTextSplitter,  
        SentenceTransformersTokenTextSplitter,  
        TextSplitter,  
        TokenTextSplitter,  
    )  
    from langchain_community.document_loaders import TextLoader  
    from langchain_community.vectorstores import Chroma  
    from langchain_openai import OpenAIEmbeddings

• 本节导入脚本所需的模块和类，包括文本拆分器（Text Splitters）、文档加载器、向量存储和嵌入（embeddings）。使用 langchain_community 和 langchain_openai 库来加载文档，并将其拆分成易于管理的块，同时创建嵌入。

文件夹设置:

    # 注释：定义包含文本文件的路径  
    current_dir = os.path.dirname(os.path.abspath(__file__))  # 获取当前文件所在目录  
    file_path = os.path.join(current_dir, "books", "romeo_and_juliet.txt")  # 文件路径 = os.path.join(当前目录, 'books', 'romeo_and_juliet.txt')  
    db_dir = os.path.join(current_dir, "db")  # 数据库目录 = os.path.join(当前目录, 'db')

• 代码定义了脚本所在目录，文本文件路径（romeo_and_juliet.txt），以及用于保存向量文件的目录（db 文件夹）。

文件存在检测 ：

    # 检查文本文件是否存在  
    if not os.path.exists(file_path):  
        raise FileNotFoundError(  
            f"文件 {file_path} 不存在，请确认路径是否正确。"  
        )

• 这个代码会检查指定的文本文件是否真的存在。如果该文件不存在，它会抛出一个 FileNotFoundError，这会停止脚本并显示一个错误消息。

加载文本。

    # 读取文件中的文本内容  
    # TextLoader 用于从文件加载文本  
    loader = TextLoader(file_path)  # file_path 表示文件路径  
    documents = loader.load()

• TextLoader 类用于将文本文件的内容加载到脚本里。内容被加载到 documents 变量里。

定义嵌入式模型：

    # 定义嵌入模型如下  
    embeddings = OpenAIEmbeddings(  
      model="text-embedding-3-small"  
    )  # 如有需要，请更换为有效的嵌入模型

• 该代码设置了嵌入模型，将文本数据转换成数值向量。此特定模型（text-embedding-3-small）用于嵌入文本，并且如果需要，可以进行更新。

创建并持久化向量数据库的函数：

    # 创建并持久化向量库的函数  
    def create_vector_store(docs, store_name):  
        persistent_directory = os.path.join(db_dir, store_name)  
        if not os.path.exists(persistent_directory):  
            print(f"\n--- 正在创建向量库 {store_name}，请稍等... ---")  
            db = Chroma.from_documents(  
                docs, embeddings, persist_directory=persistent_directory  
            )  
            print(f"--- 已完成创建向量库 {store_name}，请查收... ---")  
        else:  
            print(f"向量库 {store_name} 已存在。无需重新创建。")

• 此功能检查指定目录中是否有向量存储已经存在。如果不存在，则使用提供的文档和嵌入创建并保存新的向量存储。向量存储将保存在 db_dir 目录下，文件名由 store_name 指定。

这篇文章最关键的部分是我们要如何分割输入的数据集。这里我们将看看不同的分割数据的方法，以及看看结果怎么样。

1. 基于字符的拆分:

     # 1. 基于字符的切分  
    # 根据指定的字符数将文本分割成段落。  
    # 无论文本结构如何，都能保持一致的分割长度。  
    print("\n--- 使用基于字符的切分 ---")  
    char_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=100)  
    char_docs = char_splitter.split_documents(documents)  
    # 将切分后的文档存储到向量数据库中  
    create_vector_store(char_docs, "chroma_db_char")

文本被分割成每个1000字符的片段，每两个部分之间有100字符的重叠。这种方法确保每个片段的大小一致，不受文本内容结构的影响。生成的片段存储在一个名为chroma_db_char的向量存储中。

2. 基于句子的拆分:

     # 2. 句子级别的分割  
    # 根据句子将文本划分为多个段落，确保每个段落完整地结束于句子的末尾。  
    # 适合保持每个段落内的语义连贯性。  
    print("\n--- 使用句子级别的分割 ---")  
    sent_splitter = SentenceTransformersTokenTextSplitter(chunk_size=1000)  # 创建一个 SentenceTransformersTokenTextSplitter 对象，并设置 chunk_size 为 1000  
    sent_docs = sent_splitter.split_documents(documents)  
    create_vector_store(sent_docs, "chroma_db_sent")  # 将拆分的文档保存到向量存储中，存储名称为 'chroma_db_sent'

文本被分割成句子块，每块最多包含1000个字符，以确保。这样做保证每个块都有明确的意义。这些块会被存储在一个名为chroma_db_sent的向量数据库中。

3. 基于令牌的拆分:

     # 3. 基于标记的拆分  
    # 根据标记（单词或子词）将文本拆分成片段，使用如GPT-2的分词器。  
    # 对于那些有严格标记限制的Transformer模型特别有用。  
    print("\n--- 打印分割提示信息 ---")  
    # 定义一个标记拆分器  
    token_splitter = TokenTextSplitter(chunk_overlap=0, chunk_size=512)  
    # 使用拆分器将文档拆分成标记片段  
    token_docs = token_splitter.split_documents(documents)  
    # 创建向量存储  
    create_vector_store(token_docs, "chroma_db_token")

文本根据如单词或子词之类的标记分割成小块。这种方法在处理具有严格令牌限制的变压器模型时特别有用。得到的小块存储在一个名为“chroma_db_token”的向量数据库中。

4. 递归的字符级拆分

    # 4. 递归字符级拆分  
    # 尝试在字符限制内，在自然边界（句子、段落）处拆分文本。  
    # 平衡连贯性和遵守字符限制。  
    print("\n--- 使用递归字符级拆分 ---")  
    rec_char_splitter = RecursiveCharacterTextSplitter(  
        chunk_size=1000, chunk_overlap=100)  
    rec_char_docs = rec_char_splitter.split_documents(documents)  
    create_vector_store(rec_char_docs, "chroma_db_rec_char")

此方法尝试在自然边界（如句子或段落）处拆分文本，同时确保遵守字符限制。它在保持连贯性和控制片段长度在指定限制内之间找到平衡。生成的块将存储在名为 chroma_db_rec_char 的向量存储中。

5. 自定义分割:,

     # 5. 自定义分割  
    # 允许根据特定需求自定义分割逻辑。  
    # 对于那些标准分割器无法处理的独特结构文档非常有用。  
    print("\n--- 自定义分割 ---")  

    class CustomTextSplitter(TextSplitter):  
        def split_text(self, text):  
            # 自定义文本分割逻辑  
            return text.split("\n\n")  # 例如：按段落分割  

    custom_splitter = CustomTextSplitter()  
    custom_docs = custom_splitter.split_documents(documents)  
    create_vector_store(custom_docs, "chroma_db_custom")

本部分定义了一个自定义的文本拆分器，根据特定需求（例如，按段落）来拆分文本。这种方法对于具有独特结构的文档特别有用，标准拆分器可能不太适合处理这些文档。拆分后的片段会存放在名为 chroma_db_custom 的向量数据库中。

查询：向量库 ：

    # 查询向量存储的函数 `query_vector_store`  
    def query_vector_store(store_name, query):  
        persistent_directory = os.path.join(db_dir, store_name)  
        if os.path.exists(persistent_directory):  
            print(f"\n--- 查询向量库 {store_name} ---")  
            db = Chroma(  
                persist_directory=persistent_directory, embedding_function=embeddings  
            )  
            retriever = db.as_retriever(  
                search_type="similarity_score_threshold",  
                search_kwargs={"k": 1, "score_threshold": 0.1},  
            )  
            relevant_docs = retriever.invoke(query)  
            # 显示相关文档及其元数据  
            print(f"\n--- {store_name} 的相关文档 ---")  
            for i, doc in enumerate(relevant_docs, 1):  
                print(f"文档 {i}:\n{doc.page_content}\n")  
                if doc.metadata:  
                    print(f"来源：{doc.metadata.get('source', '未知')}\n")  
        else:  
            print(f"向量库 {store_name} 不存在。")

这个功能使用用户定义的查询来查找特定的向量库。它会确认该向量库是否存在，并根据查询检索相关文档，显示结果及元数据。

查询执行 ：

    # 定义用户的提问
    query = "朱丽叶是怎么死的？"

    # 查询每个向量数据库，获取相关信息
    query_vector_store("chroma_db_char", query)
    query_vector_store("chroma_db_sent", query)
    query_vector_store("chroma_db_token", query)
    query_vector_store("chroma_db_rec_char", query)
    query_vector_store("chroma_db_custom", query)

这段代码定义了这样一个查询（"朱丽叶是怎么死的?"），并使用它来查询之前创建的每个向量仓库。它对每个向量仓库调用query_vector_store函数，查找并展示相关的文档。

以下为供参考的完整代码。

    import os

    from langchain.text_splitter import (  
        CharacterTextSplitter,  
        RecursiveCharacterTextSplitter,  
        SentenceTransformersTokenTextSplitter,  
        TextSplitter,  
        TokenTextSplitter,  
    )  
    from langchain_community.document_loaders import TextLoader  
    from langchain_community.vectorstores import Chroma  
    from langchain_openai import OpenAIEmbeddings  

    # 定义包含文本文件的文件夹路径  
    current_dir = os.path.dirname(os.path.abspath(__file__))  
    file_path = os.path.join(current_dir, "books", "romeo_and_juliet.txt")  
    db_dir = os.path.join(current_dir, "db")  

    # 检查文本文件是否存在  
    if not os.path.exists(file_path):  
        raise FileNotFoundError(  
            f"文件 {file_path} 不存在，请检查路径。"  
        )  

    # 从文件中读取文本内容  
    loader = TextLoader(file_path)  
    documents = loader.load()  

    # 定义嵌入模型  
    embeddings = OpenAIEmbeddings(  
        model="text-embedding-3-small"  
    )  # 如有必要，请更新为有效的嵌入模型  

    # 创建并持久化向量存储的函数  
    def create_vector_store(docs, store_name):  
        persistent_directory = os.path.join(db_dir, store_name)  
        if not os.path.exists(persistent_directory):  
            print(f"\n--- 创建向量存储 {store_name} ---")  
            db = Chroma.from_documents(  
                docs, embeddings, persist_directory=persistent_directory  
            )  
            print(f"--- 完成创建向量存储 {store_name} ---")  
        else:  
            print(f"向量存储 {store_name} 已存在。无需初始化。")  

    # 1. 基于字符的分割  
    # 根据指定的字符数分割文本。  
    # 能够确保无论内容结构如何，分块大小一致。  
    print("\n--- 使用基于字符的分割 ---")  
    char_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=100)  
    char_docs = char_splitter.split_documents(documents)  
    create_vector_store(char_docs, "chroma_db_char")  

    # 2. 基于句子的分割  
    # 根据句子分割文本，确保分块在句子边界结束。  
    # 适合在块内保持语义连贯性。  
    print("\n--- 使用基于句子的分割 ---")  
    sent_splitter = SentenceTransformersTokenTextSplitter(chunk_size=1000)  
    sent_docs = sent_splitter.split_documents(documents)  
    create_vector_store(sent_docs, "chroma_db_sent")  

    # 3. 基于标记的分割  
    # 根据标记（单词或子词）分割文本，使用像GPT-2这样的分词器。  
    # 适合具有严格标记限制的转换模型。  
    print("\n--- 使用基于标记的分割 ---")  
    token_splitter = TokenTextSplitter(chunk_overlap=0, chunk_size=512)  
    token_docs = token_splitter.split_documents(documents)  
    create_vector_store(token_docs, "chroma_db_token")  

    # 4. 递归字符分割  
    # 尝试在字符限制内，在自然边界（句子、段落）处分割文本。  
    # 平衡保持连贯性和遵守字符限制。  
    print("\n--- 使用递归字符分割 ---")  
    rec_char_splitter = RecursiveCharacterTextSplitter(  
        chunk_size=1000, chunk_overlap=100)  
    rec_char_docs = rec_char_splitter.split_documents(documents)  
    create_vector_store(rec_char_docs, "chroma_db_rec_char")  

    # 5. 自定义分割  
    # 允许根据特定需求创建自定义分割逻辑。  
    # 对于标准分割器无法处理的独特结构的文档很有用。  
    print("\n--- 使用自定义分割 ---")  

    class CustomTextSplitter(TextSplitter):  
        def split_text(self, text):  
            # 自定义逻辑来分割文本  
            return text.split("\n\n")  # 示例：按段落分割  

    custom_splitter = CustomTextSplitter()  
    custom_docs = custom_splitter.split_documents(documents)  
    create_vector_store(custom_docs, "chroma_db_custom")  

    # 查询向量存储的函数  
    def query_vector_store(store_name, query):  
        persistent_directory = os.path.join(db_dir, store_name)  
        if os.path.exists(persistent_directory):  
            print(f"\n--- 查询向量存储 {store_name} ---")  
            db = Chroma(  
                persist_directory=persistent_directory, embedding_function=embeddings  
            )  
            retriever = db.as_retriever(  
                search_type="similarity_score_threshold",  
                search_kwargs={"k": 1, "score_threshold": 0.1},  
            )  
            relevant_docs = retriever.invoke(query)  
            # 显示相关结果及其元数据  
            print(f"\n--- 对于 {store_name} 的相关文档 ---")  
            for i, doc in enumerate(relevant_docs, 1):  
                print(f"文档 {i}:\n{doc.page_content}\n")  
                if doc.metadata:  
                    print(f"来源: {doc.metadata.get('source', '未知')}\n")  
        else:  
            print(f"向量存储 {store_name} 不存在。")  

    # 定义用户的查询  
    query = "朱丽叶是怎么去世的？"  

    # 查询每个向量存储  
    query_vector_store("chroma_db_char", query)  
    query_vector_store("chroma_db_sent", query)  
    query_vector_store("chroma_db_token", query)  
    query_vector_store("chroma_db_rec_char", query)  
    query_vector_store("chroma_db_custom", query)

以下是某些查询的结果：

这段代码提供了一套全面的解决方案，涵盖了文档处理、拆分、向量化处理和查询处理。它允许使用不同的文本拆分方法进行实验，并展示了如何使用 Chroma 向量存储来存储和检索文本嵌入。

参考:

1. https://python.langchain.com/v0.1/docs/modules/data_connection/document_transformers/?source=post_page-----3225af44e7ea-------------------------------- （链原文档）

2.https://python.langchain.com/v0.1/docs/modules/data_connection/text_embedding/?source=post_page-----3225af44e7ea--------------------------------

3. https://python.langchain.com/v0.2/docs/integrations/chat/（集成/聊天）

1. Brandon Hancock的技术链大师班 https://brandonhancock.io/langchain-master-class

2. https://platform.openai.com/docs/guides/embeddings/关于嵌入的介绍