完整指南:5种最佳chunking技术提升rag应用性能!
Photo by Jason Abdilla on Unsplash
一个检索增强生成的效果好坏取决于其块的质量。 — Thuwarakesh (我)
如果你的检索增强生成应用程序没有达到预期的效果,也许是时候改变你的分块策略了。更好的块意味着更好的检索,这意味着高质量的响应。
然而,没有任何分块技术优于其他技术。你需要根据几个因素进行选择,包括你的项目预算。
更好的分块如何导致高质量的响应?
如果你正在阅读这篇文章,我可以假设你知道什么是分块和检索增强生成。尽管如此,简而言之,它就是这样。
大型语言模型是在海量公共数据集上训练的。然而,它们在此之后并没有更新。因此,大型语言模型在预训练截止日期之后并不知道任何信息。此外,你使用大型语言模型可能涉及到你组织的私有数据,而大型语言模型没有办法知道这些数据。
因此,出现了一种美妙的解决方案,称为检索增强生成。检索增强生成要求大型语言模型 根据提示中提供的上下文回答问题。我们甚至要求它在大型语言模型知道答案的情况下也不要回答,但提供的上下文不足。
我们如何获取上下文?你可以查询你的数据库和互联网,浏览几页PDF报告,或者做其他任何事情。
但在检索增强生成中有两个问题。
- 大型语言模型的 上下文窗口大小 是有限的(不再如此——我很快会提到这一点!)
- 大的上下文窗口具有高 信噪比。
首先,早期的大型语言模型窗口大小有限。例如,GPT 2 只有一个1024标记的上下文窗口。GPT 3 提供了一个2048标记的窗口。这些只是 典型博客文章的大小。
由于这些限制,大型语言模型的提示无法包含一个组织的整个知识库。工程师们被迫减少输入到大型语言模型中的信息量,以获得良好的响应。
然而,各种具有128k标记上下文窗口的模型出现了。这通常是 许多上市公司年度报告的大小。足够好,可以将文档上传到聊天机器人并提出问题。
但是,它并不总是表现如预期。那是因为上下文中的噪声。大型文档很容易包含许多无关的信息和必要的信息。这些无关的信息使大型语言模型失去其目标或产生幻觉。
这就是我们分块文档的原因。我们不是将大型文档发送给大型语言模型,而是将其分成更小的部分,只发送最相关的部分。
然而,这说起来容易做起来难。
有一百万种可能的方法将文档分成块。例如,你可以逐段分块,而我可以逐句分块。这两种方法都是有效的,但在特定情况下,一种方法可能比另一种更有效。
然而,我们不会讨论句子和段落的分隔,因为它们是微不足道的,在分块中用途不大。相反,我们将讨论稍微复杂一些的分块方法,以便用于检索增强生成。
在接下来的文章中,我将讨论我学习和应用的一些分块策略。
递归字符分割
这就是基础,你会说。
当然是。但在我看来,这也是使用最广泛的方法。不仅因为它易于理解和实现,还因为它是分块文档的最快和最便宜的技术。
递归字符分割策略使用固定块大小的移动窗口和重叠参数。它从开始处开始移动窗口,保持指定的重叠。
以下是相同内容的视觉说明。
递归字符分割的工作原理 — 作者插图。
在上面的插图中,块大小为20个字符,重叠为2。这显示了块是如何创建的。
这种方法非常简单快速。它可以在几分钟内分块整个年度报告。以下是我们如何使用Langchain实现它。
from langchain.text_splitter import RecursiveCharacterTextSplitter
text = """
Hydroponics is an intelligent way to grow veggies indoors or in small spaces. In hydroponics, plants are grown without soil, using only a substrate and nutrient solution. The global population is rising fast, and there needs to be more space to produce food for everyone. Besides, transporting food for long distances involves lots of issues. You can grow leafy greens, herbs, tomatoes, and cucumbers with hydroponics.
"""
rc_splits = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
chunk_size=20, chunk_overlap=2
).split_text(text)
这种方法是一种基于位置的分割技术。不幸的是,递归字符假设两个连续句子讨论类似的内容。但这并不一定是。
在一本书的同一章节中,我们可能会讨论几个不同的主题,并最终将它们结合起来以显示联系。但在连接被揭示之前,关系并不明确。
现在,如果你有足够大的块,你可以捕捉关于个别想法及其连接的整个片段。然而,如果你在处理一个大型项目,这种方法就不适用了。
我们需要一种基于文本语义意义进行分割的技术——而不是基于其位置。
语义分块
语义分块是一种不同的方法。我们不再依赖位置,而是依赖语义意义。
我们扫描文档,并在语义意义显著变化时进行分块。
请参见下面的语义分块文本的可视化。
语义分块的工作原理 — 作者插图。
前两句讨论水培农业,而接下来的两句讨论全球问题。然后,它又回到了水培。
后面的段落揭示了水培与全球人口问题之间的联系。然而,到这一段为止,它们是显著不同的段落。
如果你再看看插图,使用语义分块技术创建的块是 不同长度 的。这是因为我们等待语义意义发生变化,而不是在达到固定的单词长度时进行分块。
但这种方法说起来简单,做起来难。
挑战在于以编程方式确定句子的意义。这是通过嵌入模型来完成的。像 OpenAI的text-embedding-3-large 这样的嵌入模型将句子转换为向量,以捕捉其背后的语义意义。
语义分块过程需要五个步骤才能完成。
第一,我们将文本分成句子/段落,并通过合并相邻的句子创建初始块。然后,我们为这些块创建向量嵌入。第三步涉及计算每两个连续块之间的距离。我们现在可以使用阈值来分块,当上一步计算的距离超过某个点时 — 这就是第四步。
第五步是可选的可视化。但我建议你创建这个,因为它会让你对所创建的块更有信心。当然,你不能在大型项目中做到这一点。但你至少可以对一个样本进行。
这是执行此操作的完整代码。
sentences = re.split(r"(?<\=\[.?!\])\\s+", text)
initial_chunks = [
{"chunk": str(sentence), "index": i} for i, sentence in enumerate(sentences)
]
def combine_chunks(chunks):
for i in range(len(chunks)):
combined_chunk = ""
if i > 0:
combined_chunk += chunks[i - 1]["chunk"]
combined_chunk += chunks[i]["chunk"]
if i < len(chunks) - 1:
combined_chunk += chunks[i + 1]["chunk"]
chunks[i]["combined_chunk"] = combined_chunk
return chunks
combined_chunks = combine_chunks(initial_chunks)
chunk_embeddings = embeddings.embed_documents(
[chunk["combined_chunk"] for chunk in combined_chunks]
)
for i, chunk in enumerate(combined_chunks):
chunk["embedding"] = chunk_embeddings[i]
def calculate_cosine_distances(chunks):
distances = []
for i in range(len(chunks) - 1):
current_embedding = chunks[i]["embedding"]
next_embedding = chunks[i + 1]["embedding"]
similarity = cosine_similarity([current_embedding], [next_embedding])[0][0]
distance = 1 - similarity
distances.append(distance)
chunks[i]["distance_to_next"] = distance
return distances
distances = calculate_cosine_distances(combined_chunks)
import numpy as np
threshold_percentile = 90
threshold_value = np.percentile(distances, threshold_percentile)
crossing_points = [
i for i, distance in enumerate(distances) if distance > threshold_value
]
len(crossing_points)
import seaborn as sns
import matplotlib.pyplot as plt
import numpy as np
def visualize_cosine_distances_with_thresholds_multicolored(
cosine_distances, threshold_percentile=90
):
threshold_value = np.percentile(cosine_distances, threshold_percentile)
crossing_points = [0]
crossing_points += [
i
for i, distance in enumerate(cosine_distances)
if distance > threshold_value
]
crossing_points.append(
len(cosine_distances)
)
plt.figure(figsize=(14, 6))
sns.set(style="white")
sns.lineplot(
x=range(len(cosine_distances)),
y=cosine_distances,
color="blue",
label="余弦距离",
)
plt.axhline(
y=threshold_value,
color="red",
linestyle="--",
label=f"{threshold_percentile}百分位阈值",
)
colors = sns.color_palette(
"hsv", len(crossing_points) - 1
)
for i in range(len(crossing_points) - 1):
plt.axvspan(
crossing_points[i], crossing_points[i + 1], color=colors[i], alpha=0.3
)
plt.title(
"带有多彩阈值高亮的段落间余弦距离"
)
plt.xlabel("段落索引")
plt.ylabel("余弦距离")
plt.legend()
plt.xlim(0, len(cosine_distances) - 1)
plt.show()
return crossing_points
crossing_points = visualize_cosine_distances_with_thresholds_multicolored(
distances, threshold_percentile=threshold_percentile
)

Seaborn图表以说明语义分块 — 作者图片。
如你所见,每当块之间的距离超过0.12时,我们通过合并到该点的每个较小块来创建一个更大的块。这个过程创建了六个不同长度的块。
## 智能分块
这些日子你可能厌倦了听到“代理”这个词。我理解。
智能分块是一个创建分块的智能过程。我会说它比语义分块更进一步。
这里的主要区别是,人们的思维并不是严格有序的。我们的思维往往是跳来跳去的——我们的写作也是如此。
顺便提一下,这是判断 AI 是否写作的一个简单方法。
回到讨论中,
递归字符拆分和语义分块的缺点在于,它们假设相似的想法在文档中总是彼此靠近。
在我们看到的许多文档中,这种情况很少发生。
智能分块像真实的人一样拆分文档。我们 **使用大型语言模型** 来浏览文本以寻找新想法;如果文本属于现有的想法,则该部分被放入现有的桶(块)中。
但是如果代理第一次看到这个想法,它会创建一个新的桶(块)来放置这段文本。
以下是代码实现。
```python
from langchain import hub
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel, Field
obj = hub.pull("wfh/proposal-indexing")
llm = ChatOpenAI(model="gpt-4o")
class Sentences(BaseModel):
sentences: List[str]
extraction_llm = llm.with_structured_output(Sentences)
extraction_chain = obj | extraction_llm
paragraphs = text.split("\\n\\n")
propositions = []
for i, p in enumerate(paragraphs):
propositions = extraction_chain.invoke(p)
propositions.extend(propositions)
chunks = {}
class ChunkMeta(BaseModel):
title: str = Field(description="The title of the chunk.")
summary: str = Field(description="The summary of the chunk.")
def create_new_chunk(chunk_id, proposition):
summary_llm = llm.with_structured_output(ChunkMeta)
summary_prompt_template = ChatPromptTemplate.from_messages(
[
(
"system",
"Generate a new summary and a title based on the propositions.",
),
(
"user",
"propositions:{propositions}",
),
]
)
summary_chain = summary_prompt_template | summary_llm
chunk_meta = summary_chain.invoke(
{
"propositions": [proposition],
}
)
chunks[chunk_id] = {
"summary": chunk_meta.summary,
"title": chunk_meta.title,
"propositions": [proposition],
}
def add_proposition(chunk_id, proposition):
summary_llm = llm.with_structured_output(ChunkMeta)
summary_prompt_template = ChatPromptTemplate.from_messages(
[
(
"system",
"If the current_summary and title is still valid for the propositions return them."
"If not generate a new summary and a title based on the propositions.",
),
(
"user",
"current_summary:{current_summary}\\n\\ncurrent_title:{current_title}\\n\\npropositions:{propositions}",
),
]
)
summary_chain = summary_prompt_template | summary_llm
chunk = chunks[chunk_id]
current_summary = chunk["summary"]
current_title = chunk["title"]
current_propositions = chunk["propositions"]
all_propositions = current_propositions + [proposition]
chunk_meta = summary_chain.invoke(
{
"current_summary": current_summary,
"current_title": current_title,
"propositions": all_propositions,
}
)
chunk["summary"] = chunk_meta.summary
chunk["title"] = chunk_meta.title
chunk["propositions"] = all_propositions
def find_chunk_and_push_proposition(proposition):
class ChunkID(BaseModel):
chunk_id: int = Field(description="The chunk id.")
allocation_llm = llm.with_structured_output(ChunkID)
allocation_prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You have the chunk ids and the summaries"
"Find the chunk that best matches the proposition."
"If no chunk matches, return a new chunk id."
"Return only the chunk id.",
),
(
"user",
"proposition:{proposition}" "chunks_summaries:{chunks_summaries}",
),
]
)
allocation_chain = allocation_prompt | allocation_llm
chunks_summaries = {
chunk_id: chunk["summary"] for chunk_id, chunk in chunks.items()
}
best_chunk_id = allocation_chain.invoke(
{"proposition": proposition, "chunks_summaries": chunks_summaries}
).chunk_id
if best_chunk_id not in chunks:
best_chunk_id = create_new_chunk(best_chunk_id, proposition)
return
add_proposition(best_chunk_id, proposition)
在上述示例中,我们在处理分块过程之前进行了一个叫做命题的操作。这是一种将每个句子转换为自我解释的方法。考虑以下示例:
## 原始文本
一只乌鸦坐在池塘旁边。它是一只白色的乌鸦。
## 命题文本
一只乌鸦坐在池塘旁边。这只乌鸦是一只白色的。
这有助于大型语言模型更清晰地理解文本。它不必在每个句子中都去猜测“它”和“那”的意思。
使用智能分块方法创建的块不必仅包含相邻的文本。在一份大型文档中,智能分块可以将相似的想法带到任何地方。
## 摘要,
在这篇文章中,我讨论了三种常用的分块策略。每种都有其优点。但没有一种是每次都有效的灵丹妙药。
递归字符拆分是最简单且成本最低的。这是一个适合尝试的经济实惠选项。它非常适合原型设计,如果你幸运的话,在生产中也能获得不错的准确性。
语义分块速度较慢,但成本不高。它可以创建讨论一个特定想法的块。这在递归拆分中是无法实现的。因此,它可以产生更好的结果。
然而,这两种方法都假设所有相关内容都是一起说的。但实际上,我们并不是这样做的。更好的方法涉及大型语言模型——智能分块。不幸的是,这可能非常慢且昂贵。
但根据你的用例,你会发现其中一种是你的最爱。