langchain父子分块实现

在RAG（Retrieval Augmented Generation 检索增强生成）系统中，最核心的部分是语义检索返回的结果，也就是召回率，而影响召回率的重要因素是召回的块或片段是否是我们想要的，怎么分块才能获得我们想要的块，并且是较为完整的块？

本文结合langchain的两种分块方式（SemanticChunker和RecursiveCharacterTextSplitter）实现父子分块

两种分块策略介绍

这里介绍langchain中的两种分块方式——语义分块和滑动窗口分块

SemanticChunker

SemanticChunker（语义分块器），顾名思义，基于语义相似度进行分块

简要说明语义分块器流程：首先根据正则表达式分割句子（注意：若无正则表达式分割句子规则，那么只会返回一个块），得到句子列表，使用embeddings获取句子列表中每个句子对应的向量列表，再根据断点规则（断点类型和断点阈值）合并句子，使其最小块为min_chunk_size。

分块器的具体运行过程：

1、按照正则表达式的规则分割句子

• 若分割出的句子列表的长度为1或2、或breakpoint_threshold_type==“gradient”，直接返回句子列表，即为分出来的块

2、使用embeddings模型对句子列表进行向量化

• 期间会根据buff_size大小合并句子

3、若number_of_chunks不为None，则会按照某个规则（具体规则可以看源码）返回number_of_chunks的数量的块

4、number_of_chunks为None，则按照breakpoint_threshold_type进行句子合并，这里以percentile为例

• 首先根据传入的breakpoint_threshold_amount值，假如为40，找到句子列表对应的向量列表中排名为第40%的一个向量值β
  • 比如向量列表distances → [0.1, 0.2, 0.3, 0.5, 0.8, 0.9, 1.2]，第 40 百分位数落在 40% * (7 - 1) ≈第 2.4 个元素（索引从 0），也就是说第 2 和第 3 个元素之间 → 元素 0.3 和 0.5 之间。插值的话可能得到类似 ~0.38 的值。
• 接着用原始的句子向量列表过滤出大于上面找出的向量值β，得到超过向量值β的索引列表
• 根据索引列表合并块，判断块是否小于min_chunk_size，小于的话则继续合并，否则添加到chunks列表中，最后返回chunks列表

分块示例演示

from langchain_experimental.text_splitter import SemanticChunker
semantic_text_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",  # 或 'gradient'、'standard_deviation' 等
    breakpoint_threshold_amount=10.0,  # 降低阈值，例如切分频率更高
    min_chunk_size=500,  # 最小 chunk 字符数/句子数
    sentence_split_regex='(?<=[。；：:;.])'
)
docs = semantic_text_splitter.split_text(text)

RecursiveCharacterTextSplitter

RecursiveCharacterTextSplitter（递归字符文本分词器），类似一种滑动窗口的分词器，每个窗口内的句子作为一个块，并重叠一部分块保持上下文

分词器运行流程：先根据separators进行句子分割，假如窗口大小（chunk_size）为100，则将窗口内的句子合并为一个块（不会截断句子，若超过100则会舍去最后一个句子），之后窗口移动75（chunk_size-chunk_overlap），即重叠25（chunk_overlap）个字以上，再将窗口内的句子合并为一块。

分块演示示例

from langchain_text_splitters import RecursiveCharacterTextSplitter

chunk_size = 1024
recursive_text_splitter = RecursiveCharacterTextSplitter(
    separators=[
        "\n\n", "\n", " ", ".", ",", "\u3002", "\uff0c", "；", "。", "，"
    ],
    chunk_size=chunk_size,
    chunk_overlap=chunk_size * 0.25,
    length_function=len,
    is_separator_regex=False,
)

docs = recursive_text_splitter.split_text(text)

父子分块实现

虽然语义分块已经能够很好的对文章进行分块，而langchain中的语义分块并未设置最大值，因此分出来的块可能会特别大块，另一方面，分出来的块没有足够的上下文。

有人会说：要包含足够多的上下文，那我分块大小设置为大一点不就好了？

这会有一个问题，分块太大的话，检索的精度就会降低，因此我们这里使用父子分块的方式对文章进行分块。

什么是父子分块？

父块：将文章分为较大的块，保证足够多的上下文，例如父块大小为1000

子块：在每个父块的基础上，将父块分为较小的子块，例如子块大小为500

父子分块：即先将文章分为大块，再将大块分为小块，之后在检索的时候，先检索小块，再根据小块找到对应的大块，最后返回这些大块。

实现思路

整体思路：使用langchain中的SemanticChunker和RecursiveCharacterTextSplitter进行分块，先用SemanticChunker将文章进行语义分块，分成大块；判断每块大小是否要再次分块（由于SemanticChunker未设置上限，因此分出来的块会太大），若要再次分块则使用RecursiveCharacterTextSplitter分成大块。之后对于每个大块，再使用SemanticChunker和RecursiveCharacterTextSplitter进行分块，步骤类似大块，这样就得到了父子块了。

示例代码

from typing import List, Dict, Any
from langchain_experimental.text_splitter import SemanticChunker
from langchain_text_splitters.character import RecursiveCharacterTextSplitter
from langchain.schema import Document
from langchain_openai.embeddings import OpenAIEmbeddings

def split_text_into_hierarchy(
    text: str,
    embedding_model,
    # 参数：第一层的大块的最大字符数／最小字符数
    max_size_parent: int = 2000,
    min_size_parent: int = 500,
    # 参数：子块层的最大字符数／最小字符数
    max_size_child: int = 800,
    min_size_child: int = 200,
) -> List[Dict[str, Any]]:
    """
    返回一个父子块结构的 list，每个 item 包含：
     - parent 文本
     - parent 文档对象 metadata
     - 子块 list，子块也是文档对象或者文本
    """

    # 初始化 splitters
    semantic_text_splitter_parent = SemanticChunker(
    embeddings,
        breakpoint_threshold_type="percentile",  # 或 'gradient'、'standard_deviation' 等
        breakpoint_threshold_amount=10.0,  # 降低阈值，例如切分频率更高
        min_chunk_size=min_size_parent,  # 最小 chunk 字符数/句子数
        sentence_split_regex='(?<=[。；：:;.])'
	)

    semantic_text_splitter_child = SemanticChunker(
        embeddings,
        breakpoint_threshold_type="percentile",  # 或 'gradient'、'standard_deviation' 等
        breakpoint_threshold_amount=10.0,  # 降低阈值，例如切分频率更高
        min_chunk_size=min_size_child,  # 最小 chunk 字符数/句子数
        sentence_split_regex='(?<=[。；：:;.])'
	)

    recursive_splitter_parent = RecursiveCharacterTextSplitter(
        separators=[
        "\n\n", "\n", " ", ".", ",", "\u3002", "\uff0c", "；", "。", "，"
    ],
        chunk_size=max_size_parent,
        chunk_overlap=max_size_parent*0.25,
        length_function=len,
    )

    recursive_splitter_child = RecursiveCharacterTextSplitter(
        separators=[
        "\n\n", "\n", " ", ".", ",", "\u3002", "\uff0c", "；", "。", "，"
    ],
        chunk_size=max_size_child,
        chunk_overlap=max_size_child*0.25,
        length_function=len,
    )

    # 第一层：先用 semantic_chunker 分父块
    parent_docs: List[Document] = semantic_text_splitter_parent.create_documents([text])
    results = []

    for parent_doc in parent_docs:
        parent_text = parent_doc.page_content
        parent_metadata = parent_doc.metadata if hasattr(parent_doc, "metadata") else {}

        # 如果父块太大／太小，就用 recursive splitter 再拆父块，使得父块大小控制在 [min_size_parent, max_size_parent]
        parent_chunks = []
        if len(parent_text) > max_size_parent:
            # 分成多个父块
            docs1 = recursive_splitter_parent.create_documents([parent_text])
            for d in docs1:
                parent_chunks.append(d)
        else:
            parent_chunks.append(parent_doc)

        # 对每个父块，再做子块处理
        for pchunk in parent_chunks:
            child_list: List[Document] = []

            # 用 semantic chunker 再语义切分子块
            child_semantic = semantic_text_splitter_child.create_documents([pchunk.page_content])

            # 对每个语义子块，如果过大／过小，再用 recursive_splitter_child 拆
            for cdoc in child_semantic:
                ctext = cdoc.page_content
                if len(ctext) > max_size_child:
                    # 拆成小子块
                    small_chunks = recursive_splitter_child.create_documents([ctext])
                    child_list.extend(small_chunks)
                else:
                    child_list.append(cdoc)

            # 构建这一父块 + 子块的结构 ，此处仅仅做简单展示，可结合milvus等向量数据库一起使用
            results.append({
                "parent": {
                    "text": pchunk.page_content,
                    "metadata": pchunk.metadata,
                },
                "children": [
                    {
                        "text": c.page_content,
                        "metadata": c.metadata,
                    }
                    for c in child_list
                ],
            })

    return results


if __name__ == "__main__":
    # 示例用法
    embedding_model = OpenAIEmbeddings()

    long_text = """
    # 在这里放你的长文章内容
    Lorem ipsum dolor sit amet, consectetur adipiscing elit. ...
    （很多内容）
    """

    hierarchy = split_text_into_hierarchy(
        long_text,
        embedding_model,
        max_size_parent=2000,
        min_size_parent=500,
        max_size_child=800,
        min_size_child=200,
    )

    # 打印示例
    for i, item in enumerate(hierarchy):
        print(f"=== Parent Block {i} (length {len(item['parent']['text'])}) ===")
        for j, child in enumerate(item['children']):
            print(f"  Child {j} length {len(child['text'])}")
        print()