RAG 101：分块策略

![img](https://yg9538.kmgy.top/20241209170344533.png)

我们如何拆分球？（使用 Cava 生成）

T大型语言模型在单个请求中可以处理的最大令牌数称为上下文长度（或上下文窗口）。下表显示了[所有版本的 GPT-4 的上下文长度](https://platform.openai.com/docs/models/gpt-4-turbo-and-gpt-4)（截至 2024 年 9 月）。虽然上下文长度随着每次迭代和每个新模型而增加，但我们能为模型提供的信息仍然存在限制。此外，输入的大小与 LLM 生成的响应的上下文相关性之间存在负相关，简短而集中的输入比包含大量信息的长上下文产生更好的结果。这强调了将数据分解为更小、更相关的块以确保 LLM 做出更适当响应的重要性——至少在 LLM 无需重新训练即可处理大量数据之前。

gpt-4 模型的上下文窗口限制（引用自 [OpenAI](https://platform.openai.com/docs/models/gpt-4-turbo-and-gpt-4))

图像中表示的上下文窗口**包括输入和输出标记。**

# 为什么上下文长度很重要

虽然较长的上下文为模型提供了更全面的画面，并帮助它理解关系并做出更好的推理，但另一方面，较短的上下文减少了模型需要理解的数据量，从而减少了延迟，使模型更具响应性。它还有助于最大限度地减少 LLM 的幻觉，因为仅向模型提供相关数据。因此，这是性能、效率和数据复杂程度之间的平衡，我们需要运行实验，确定多少数据是用合理资源产生最佳结果的合适数据量。

GPT-4 模型的 128k 令牌可能看起来很多，所以让我们将它们转换为实际的单词并正确看待它们。来自 [OpenAI Tokenizer](https://platform.openai.com/tokenizer)：

> 一个有用的经验法则是，对于常见的英语文本，一个标记通常对应于 ~4 个字符的文本。这大约相当于一个单词的 3/4（所以 100 个标记 ~= 75 个单词）

让我们以 [Arthur Conan Doyle 的 The Hound of the Baskervilles](https://www.gutenberg.org/ebooks/2852)（Project Gutenberg License）为例。这本书长 7734 行，有 62303 个单词，大约有 83,700 个代币

如果您对精确计算代币而不仅仅是近似感兴趣，您可以使用 OpenAI 的 [tiktoken](https://github.com/openai/tiktoken)：

```
import request.
from tiktoken import encoding_for_model

url = "https://www.gutenberg.org/cache/epub/3070/pg3070.txt"

response = requests.get(url)
if response.status_code == 200:
    book_full_text = response.text

encoder = encoding_for_model("gpt-4o")
tokens = encoder.encode(book_full_text)

print(f"Number of tokens: {len(tokens)}")
```

它给出了要为`Number of tokens: 82069`

# 什么是分块

![img](https://yg9538.kmgy.top/20241209170345110.png)

块状奶酪！！（使用 Canva 生成）

我喜欢 wiki 对分块的[定义](https://en.wikipedia.org/wiki/Chunking_(psychology)#:~:text=In cognitive psychology%2C chunking is,whole later on in memory.)，因为它适用于 RAG，就像在认知心理学中一样。

> 分块是将一组信息中的一小部分绑定在一起的过程。这些块旨在提高材料的短期保留率，从而绕过工作内存的有限容量，使工作内存更有效

将大型数据集拆分为更小、有意义的信息片段，以便更有效地使用 LLM 的非参数内存的过程称为分块。有许多不同的方法可以拆分数据以改进 RAG 的块检索，我们需要根据正在使用的数据类型进行选择。

分块是 RAG 管道中一个关键的预检索步骤，它直接影响检索过程并显着影响最终输出。在本文中，我们将研究最常见的分块策略，并在数据上下文中评估它们的检索指标。

# 分离器的直觉

与其立即回顾不同库中可用的现有分块策略/拆分器，不如开始构建一个简单的拆分器并探索需要考虑的重要方面，以建立编写新拆分器的直觉。让我们从一个基本的 splitter 开始，然后通过解决它的缺点/限制来逐步改进它。

## 1. 朴素分块

当我们谈论拆分数据时，我们首先想到的是在换行符处拆分数据。让我们继续实施。但正如你所看到的，它留下了很多回程回车字符。此外，我们只假设*了 \n* 和 *\r*，因为我们只处理英语，但是如果我们想解析其他语言怎么办？让我们添加灵活性，以便也传入要拆分的字符。

```
def naive_splitter_v2(text: str, separators: List[str] = ["\n", "\r"]) -> List[str]:
    """Splits text at every separator"""
    splits = [text]
    for sep in separators:
        splits = [segment for part in result for segment in part.split(sep) if segment]

return splits
```

![img](https://yg9538.kmgy.top/20241209170344945.png)

naive_splitter_v2 的输出

您可能已经从输出中猜到了为什么我们将此方法称为 Naive。这个想法有很多缺点：

1. 没有 Chunk 限制。只要一行有分隔符之一，它就会中断，但是如果我们一个块没有这些分隔符怎么办，它可以达到任何长度。
2. 同样，正如您在输出中清楚地看到的那样，有些块太小了！如果没有周围的上下文，单个单词 chunks 没有任何意义。
3. 行间中断：根据提出的问题检索一个块，但如果我们在句子中间截断一个句子/行，它就完全不完整，甚至具有不同的含义。

让我们尝试一一解决这些问题。

## 2. 修复了窗口分块

让我们首先解决第一个问题，即块大小太长或太短。这次我们接受大小限制，并尝试在达到大小时准确拆分文本。

```
def fixed_window_splitter(text: str, chunk_size: int = 1000) -> List[str]:
    """Splits text at given chunk_size"""
    splits = []
    for i in range(0, len(text), chunk_size):
        splits.append(text[i:i + chunk_size])
    return splits
```

![img](https://yg9538.kmgy.top/20241209170344083.png)

fixed_window_splitter 的输出

我们确实解决了 chunk 的最小和最大边界，因为它总是会chunk_size。但单词之间的中断仍然保持不变。从输出中我们可以看到，我们正在失去块的含义，因为它在句子中间被分割。

## 3. 修复了具有重叠分块的窗口

确保我们不会在单词之间分裂的最简单方法是确保我们遍历到单词的末尾，然后停止。虽然这会使上下文不会太长，并且在预期的chunk_size范围内，但更好的方法是在实际开始位置后面开始下一个块的一些字符/单词/标记，以便上下文始终保留并且是连续的。`x`

```
def fixed_window_with_overlap_splitter(text: str, chunk_size: int = 1000, chunk_overlap: int = 10) -> List[str]:
    """Splits text at given chunk_size, and starts next chunk from start - chunk_overlap position"""
    chunks = []
    start = 0
    
    while start <= len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start = end - chunk_overlap
    
    return chunks
```

![img](https://yg9538.kmgy.top/20241209170344242.png)

fixed_window_with_overlap_splitter 的输出

## 4. 递归字符分块

有了 和 fixed，我们现在就可以解决单词中间或句子中间拆分的问题。这可以通过对我们最初的 Naive splitter 进行一些修改来解决。我们获取一个 separator 列表，并在我们增长到 chunk 大小时选择一个好的 separator。同时，我们仍将继续使用相同的数据块重叠。这是 LangChain 包中最受欢迎的拆分器之一，称为 [RecursiveCharacterTextSplitter](https://github.com/langchain-ai/langchain/blob/82b5b77940e97f65179efa0268031c47d0584a1c/libs/text-splitters/langchain_text_splitters/character.py#L58)。这与我们的方法相同：`Chunk Size``Chunk Overlap`

1. 从最高优先级分隔符开始，该分隔符从开头 *\n\n* 开始，然后移动到*分隔符*列表中的下一个。
2. 如果拆分超过chunk_size，它将应用下一个分隔符，直到当前拆分小于正确的大小。
3. 下一次分割从当前分割结尾后面的 chunk_overlap 个字符开始，从而保持上下文的连续性。

![img](https://yg9538.kmgy.top/20241209170344178.png)

recursive_character_splitter 的输出

## 4. 语义分块

到目前为止，我们只考虑了在何处拆分数据，无论是在段落末尾、换行符、制表符还是其他分隔符。但是我们还没有考虑何时拆分，即如何更好地捕获有意义的 chunk，而不仅仅是某个长度的 chunk。这种方法称为语义分块。让我们使用 [Flair](https://github.com/flairNLP/flair) 来检测句子边界或特定实体并创建有意义的块。使用 *SegtokSentenceSplitter* 将文本拆分为句子，这可确保在有意义的边界处拆分文本。我们保持大小调整逻辑不变，进行分组，直到达到*chunk_overlap的chunk_size*重叠和重叠，以确保保持上下文。

```
def semantic_splitter(text: str, chunk_size: int = 1000, chunk_overlap: int = 10) -> List[str]:
    from flair.models import SequenceTagger
    from flair.data import Sentence
    from flair.splitter import SegtokSentenceSplitter

splitter = SegtokSentenceSplitter()
    
    # Split text into sentences
    sentences = splitter.split(text)

chunks = []
    current_chunk = ""

for sentence in sentences:
        # Add sentence to the current chunk
        if len(current_chunk) + len(sentence.to_plain_string()) <= chunk_size:
            current_chunk += " " + sentence.to_plain_string()
        else:
            # If adding the next sentence exceeds max size, start a new chunk
            chunks.append(current_chunk.strip())
            current_chunk = sentence.to_plain_string()

# Add the last chunk if it exists
    if current_chunk:
        chunks.append(current_chunk.strip())

return chunks
```

![img](https://yg9538.kmgy.top/20241209170346208.png)

semantic_splitter 的输出

LangChain 有两个这样的拆分器，使用 [NLTK](https://github.com/langchain-ai/langchain/blob/master/libs/text-splitters/langchain_text_splitters/nltk.py) 和 [spaCy](https://github.com/langchain-ai/langchain/blob/master/libs/text-splitters/langchain_text_splitters/spacy.py) 库，所以一定要看看它们。

因此，一般来说，在静态分块方法中，是确定分块策略时要考虑的两个主要因素。Chunk size 是每个块的字符/单词/标记的数量，chunk overlap 是当前块中要包含的前一个块的数量，因此上下文是连续的。块重叠也可以表示为字符/单词/标记的数量或块大小的百分比。`Chunk Size``Chunk Overlap`

您可以使用很酷的 [ChunkViz](https://github.com/gkamradt/ChunkViz) 工具来可视化不同的分块策略在不同的数据块大小和重叠参数下的行为：

![img](https://yg9538.kmgy.top/20241209170346778.png)

[ChunkViz](https://github.com/gkamradt/ChunkViz) 上的 Hound Of Baskervilles

## 5. 嵌入分块

尽管语义分块可以完成工作，但 NLTK、spaCy 或 Flair 使用自己的模型/嵌入来理解给定的数据，并尝试为我们提供数据在语义上可以最佳分割的时间。当我们继续进行实际的 RAG 实现时，我们的 embeddings 可能与我们的 chunks 合并在一起的 embedding 不同，因此可以完全以不同的方式理解。因此，在这种方法中，我们首先拆分为句子，并根据我们稍后将用于 RAG 检索的相同嵌入模型形成块。为了以不同的方式做事，我们将使用 NLTK for 来拆分为句子，并使用 OpenAIEmbeddings 将它们合并以形成句子。

```
def embedding_splitter(text_data, chunk_size=400):
    import os
    import nltk
    from langchain_openai.embeddings import AzureOpenAIEmbeddings
    from sklearn.metrics.pairwise import cosine_similarity
    import numpy as np
    from dotenv import load_dotenv, find_dotenv
    from tqdm import tqdm
    from flair.splitter import SegtokSentenceSplitter
    
    load_dotenv(find_dotenv())
    
    
    # Set Azure OpenAI API environment variables (ensure these are set in your environment)
    # You can also set these in your environment directly
    # os.environ["OPENAI_API_KEY"] = "your-azure-openai-api-key"
    # os.environ["OPENAI_API_BASE"] = "your-azure-openai-api-endpoint"
    os.environ["OPENAI_API_VERSION"] = "2023-05-15"
    
    # Initialize OpenAIEmbeddings using LangChain's Azure support
    embedding_model = AzureOpenAIEmbeddings(deployment="text-embedding-ada-002-01")  # Use your Azure model name
    
    
    # Step 1: Split the text into sentences
    def split_into_sentences(text):
        splitter = SegtokSentenceSplitter()
    
        # Split text into sentences
        sentences = splitter.split(text)
        sentence_str = []
        for sentence in sentences:
            sentence_str.append(sentence.to_plain_string())
        return sentence_str[:100]
    
    # Step 2: Get embeddings for each sentence using the same Azure embedding model
    def get_embeddings(sentences):
        embeddings = []
        for sentence in tqdm(sentences, desc="Generating embeddings"):
            embedding = embedding_model.embed_documents([sentence])  # Embeds a single sentence
            embeddings.append(embedding[0])  # embed_documents returns a list, so take the first element
        return embeddings
    
    # Step 3: Form chunks based on sentence embeddings, a similarity threshold, and a max chunk character size
    def form_chunks(sentences, embeddings, similarity_threshold=0.7, chunk_size=500):
        chunks = []
        current_chunk = []
        current_chunk_emb = []
        current_chunk_length = 0  # Track the character length of the current chunk
    
        for i, (sentence, emb) in enumerate(zip(sentences, embeddings)):
            emb = np.array(emb)  # Ensure the embedding is a numpy array
            sentence_length = len(sentence)  # Calculate the length of the sentence
    
            if current_chunk:
                # Calculate similarity with the current chunk's embedding (mean of embeddings in the chunk)
                chunk_emb = np.mean(np.array(current_chunk_emb), axis=0).reshape(1, -1)  # Average embedding of the chunk
                similarity = cosine_similarity(emb.reshape(1, -1), chunk_emb)[0][0]
    
                if similarity < similarity_threshold or current_chunk_length + sentence_length > chunk_size:
                    # If similarity is below threshold or adding this sentence exceeds max chunk size, create a new chunk
                    chunks.append(current_chunk)
                    current_chunk = [sentence]
                    current_chunk_emb = [emb]
                    current_chunk_length = sentence_length  # Reset chunk length
                else:
                    # Else, add sentence to the current chunk
                    current_chunk.append(sentence)
                    current_chunk_emb.append(emb)
                    current_chunk_length += sentence_length  # Update chunk length
            else:
                current_chunk.append(sentence)
                current_chunk_emb = [emb]
                current_chunk_length = sentence_length  # Set initial chunk length
    
        # Add the last chunk
        if current_chunk:
            chunks.append(current_chunk)
    
        return chunks
    
    # Apply the sentence splitting
    sentences = split_into_sentences(text_data)
    
    # Get sentence embeddings
    embeddings = get_embeddings(sentences)
    
    # Form chunks based on embeddings
    chunks = form_chunks(sentences, embeddings, chunk_size=chunk_size)
    
    return chunks
```

![img](https://yg9538.kmgy.top/20241209170346426.png)

embedding_splitter 的输出

## 6. 代理分块

我们的 Embedding Chunking 应该更接近于使用创建的嵌入的余弦相似性来分割数据。虽然这效果很好，但我们有一个主要缺点：它无法理解文本的语义。“I Like You” 与 “I ***Like\*** You” 对 “like” 的讽刺，这两个句子将具有相同的嵌入，因此在计算时将对应于相同的余弦距离。这就是 Agentic（或基于 LLM）分块派上用场的地方。它分析内容，以根据独立性和语义一致性确定逻辑中断点。

```
def agentic_chunking(text_data):
    from langchain_openai import AzureChatOpenAI
    from langchain.prompts import PromptTemplate
    from langchain
    llm = AzureChatOpenAI(model="gpt-4o",
                           api_version="2023-03-15-preview",
                           verbose=True,
                           temperature=1)
    prompt = """I am providing a document below. 
    Please split the document into chunks that maintain semantic coherence and ensure that each chunk represents a complete and meaningful unit of information. 
    Each chunk should stand alone, preserving the context and meaning without splitting key ideas across chunks. 
    Use your understanding of the content’s structure, topics, and flow to identify natural breakpoints in the text. 
    Ensure that no chunk exceeds 1000 characters length, and prioritize keeping related concepts or sections together.

Do not modify the document, just split to chunks and return them as an array of strings, where each string is one chunk of the document.
    Return the entire book not dont stop in betweek some sentences.

Document:
    {document}
    """

prompt_template = PromptTemplate.from_template(prompt)

chain = prompt_template | llm

result = chain.invoke({"document": text_data})
    return result
```

# 评估

我们将在即将发布的文章中介绍 RAG 评估技术;在这篇文章中，我们将看到 [RAGAS](https://docs.ragas.io/en/stable/) 定义的两个指标，以及 ，它们决定了我们的分块策略的表现。`context_precision``context_relevance`

> [Context Precision （上下文精度](https://docs.ragas.io/en/stable/concepts/metrics/context_precision.html)） 是一个指标，用于评估上下文中存在的所有与真实相关的项目是否排名较高。理想情况下，所有相关的 chunk 都必须出现在最高排名中。此指标是使用 question、ground_truth 和 contexts 计算的，值范围介于 0 和 1 之间，其中分数越高表示精度越高。
>
> [Context Relevancy](https://docs.ragas.io/en/v0.1.6/concepts/metrics/context_relevancy.html) 衡量检索到的上下文的相关性，根据问题和上下文进行计算。这些值在 （0， 1） 范围内，值越高表示相关性越高。

在下一篇文章中，我们将介绍提案检索，这是代理拆分方法之一，并计算所有策略的 RAGAS 指标。

# 结论

在本文中，我们介绍了为什么需要分块，并培养了一种直觉来构建一些策略及其实现，以及在一些知名库中构建它们的相应代码。这些只是基本的分块策略，尽管每天都在发明越来越多的策略，以使更好的检索变得更好。