RAG系列：切分優(yōu)化 - 基于句子余弦距離的語(yǔ)義切分
引言
傳統(tǒng)的文檔切分方法通常采用基于特定字符和固定長(zhǎng)度的切分策略，這種方法雖然實(shí)現(xiàn)簡(jiǎn)單，但在實(shí)際應(yīng)用中容易割裂完整的語(yǔ)義單元，導(dǎo)致后續(xù)的信息檢索與理解受到影響。
相比之下，一種更智能的切分方法是基于句子余弦距離的語(yǔ)義切分。它不再依據(jù)特定字符和固定長(zhǎng)度進(jìn)行機(jī)械切分，而是對(duì)每個(gè)句子進(jìn)行 embedding，以此來(lái)計(jì)算相鄰句子的余弦距離，再通過(guò)算法算出一個(gè)相對(duì)合理的切分點(diǎn)（某個(gè)距離值），最后將不大于該閾值的相鄰句子聚合在一起作為一個(gè)文檔塊，從而實(shí)現(xiàn)文檔語(yǔ)義切分。
例如 句子_1、句子_2、句子_3 之間的余弦距離都小于該閾值，而 句子_3 與 句子_4 的余弦距離大于該閾值，則在 句子_3 和 句子_4 之間增加切分點(diǎn)，最終的切分結(jié)果就是把 句子_1、句子_2、句子_3 聚合在一個(gè)文檔塊中，句子_4 在其它的文檔塊中。
實(shí)現(xiàn)原理
基于余弦距離的語(yǔ)義切分大致分為以下5個(gè)步驟：
langchain-experimental中的 SemanticChunker[1] 實(shí)現(xiàn)了基于余弦距離的語(yǔ)義切分，因此本文我將通過(guò) SemanticChunker 的源碼來(lái)帶大家了解語(yǔ)義切分的實(shí)現(xiàn)原理。
以下是 SemanticChunker 的初始化參數(shù)，后面根據(jù)不同步驟所需要的參數(shù)來(lái)了解這些參數(shù)的具體作用。
class SemanticChunker(
# 向量模型
embeddings: Embeddings,
# 向前向后取 buffer_size 個(gè)句子一起 embedding
buffer_size: int = 1,
# 是否在元數(shù)據(jù)添加開(kāi)始切分的位置（以文檔字符長(zhǎng)度計(jì)算）
add_start_index: bool = False,
# 切分點(diǎn)計(jì)算方法
breakpoint_threshold_type: BreakpointThresholdType = "percentile",
# 切分點(diǎn)計(jì)算閾值
breakpoint_threshold_amount: float | None = None,
# 切分后的文檔塊數(shù)量
number_of_chunks: int | None = None,
# 句子切分規(guī)則
sentence_split_regex: str = r"(?<=[.?!])\s+",
# 最小文檔塊大小
min_chunk_size: int | None = None
)句子切分
這一步是通過(guò)特定規(guī)則將文檔切分為一個(gè)個(gè)句子，在 SemanticChunker 中通過(guò)參數(shù) sentence_split_regex 來(lái)設(shè)置規(guī)則進(jìn)行切分，默認(rèn)值為 r"(?<=[.?!])\s+"，這是以英文的句號(hào)、問(wèn)號(hào)、感嘆號(hào)來(lái)進(jìn)行切分的，而且是對(duì)比較規(guī)范的英文行文，也就是這三種標(biāo)點(diǎn)后還跟空白字符的。如果要對(duì)中文文檔切分，那就需要將這個(gè)正則表達(dá)式替換成能切分中文的，例如：r"(?<=[。？！\n])"，也就是以中文的句號(hào)、問(wèn)號(hào)、感嘆號(hào)以及換行符來(lái)進(jìn)行切分。
SemanticChunker的實(shí)現(xiàn)源碼如下：
import re
def _get_single_sentences_list(self, text: str) -> List[str]:
return re.split(self.sentence_split_regex, text)句子 embedding
這一步是將每個(gè)句子進(jìn)行 embedding，理論上接著就以每個(gè)句子 embedding 結(jié)果來(lái)計(jì)算相鄰句子的距離就可以了。但通過(guò)實(shí)際操作發(fā)現(xiàn)對(duì)單個(gè)句子處理噪音比較大，后續(xù)切分的效果并不理想，因此 SemanticChunker 通過(guò) buffer_size 來(lái)控制當(dāng)前句子前、后各取幾個(gè)句子組成一組來(lái)計(jì)算 embedding 并計(jì)算余弦距離。例如buffer_size設(shè)置為為1（默認(rèn)值），表示取當(dāng)前句子前、后各取1個(gè)句子組成一組來(lái)計(jì)算 embedding。
SemanticChunker的實(shí)現(xiàn)源碼如下：
首先根據(jù)buffer_size得到當(dāng)前句子的組合。
def combine_sentences(sentences: List[dict], buffer_size: int = 1) -> List[dict]:
for i inrange(len(sentences)):
# 創(chuàng)建一個(gè)字符串變量來(lái)保存連接的句子
combined_sentence = ""
# 添加當(dāng)前句子之前 buffer_size 個(gè)句子
for j inrange(i - buffer_size, i):
if j >= 0:
combined_sentence += sentences[j]["sentence"] + " "
# 添加當(dāng)前句子
combined_sentence += sentences[i]["sentence"]
# 添加當(dāng)前句子之后 buffer_size 個(gè)句子
for j inrange(i + 1, i + 1 + buffer_size):
if j < len(sentences):
combined_sentence += " " + sentences[j]["sentence"]
# 將合并好的句子存儲(chǔ)在當(dāng)前的句子 combined_sentence 中
sentences[i]["combined_sentence"] = combined_sentence
return sentences然后根據(jù)通過(guò)參數(shù) embeddings傳入的向量模型對(duì)句子組合進(jìn)行 embedding。
def _calculate_sentence_distances(self, single_sentences_list: List[str]) -> Tuple[List[float], List[dict]]:
_sentences = [
{"sentence": x, "index": i} for i, x in enumerate(single_sentences_list)
]
sentences = combine_sentences(_sentences, self.buffer_size)
embeddings = self.embeddings.embed_documents(
[x["combined_sentence"] for x in sentences]
)
for i, sentence in enumerate(sentences):
sentence["combined_sentence_embedding"] = embeddings[i]
return calculate_cosine_distances(sentences)計(jì)算相鄰句子（組）余弦距離
這一步就是通過(guò)計(jì)算相鄰句子（組） 的余弦相似度來(lái)得到相鄰句子（組） 的余弦距離。
將橫軸記為句子（組）的序號(hào)，縱軸為相鄰句子（組） 的余弦距離，就可得到下面類(lèi)似的圖：
SemanticChunker的實(shí)現(xiàn)源碼如下：
from langchain_community.utils.math import cosine_similarity
defcalculate_cosine_distances(sentences: List[dict]) -> Tuple[List[float], List[dict]]:
distances = []
for i inrange(len(sentences) - 1):
embedding_current = sentences[i]["combined_sentence_embedding"]
embedding_next = sentences[i + 1]["combined_sentence_embedding"]
# 計(jì)算余弦相似度
similarity = cosine_similarity([embedding_current], [embedding_next])[0][0]
# 轉(zhuǎn)換成余弦距離
distance = 1 - similarity
distances.append(distance)
# 保存余弦距離
sentences[i]["distance_to_next"] = distance
# 【可選】最后一個(gè)句子的處理
# sentences[-1]['distance_to_next'] = None # 或其它默認(rèn)值
return distances, sentences計(jì)算切分點(diǎn)
如何計(jì)算切分點(diǎn)，SemanticChunker給出了4種方法：
- percentile： 分位法，默認(rèn)方法。將所有余弦距離在第 X 分位數(shù)的值作為閾值，并在那些余弦距離超過(guò)該閾值的位置進(jìn)行切分。第 X 分位數(shù)可通過(guò)
breakpoint_threshold_amount設(shè)置，默認(rèn)為 95。還可以通過(guò)number_of_chunks指定切分后的文檔塊總數(shù)量，采用線(xiàn)性插值的方式反向推導(dǎo)出該分位數(shù)；
SemanticChunker的實(shí)現(xiàn)源碼如下：
import numpy as np
def_calculate_breakpoint_threshold(self, distances: List[float]) -> Tuple[float, List[float]]:
# 第一種方式：指定分位數(shù)
return cast(
float,
np.percentile(distances, self.breakpoint_threshold_amount),
), distances
# 第二種方式：通過(guò) number_of_chunks 反向推導(dǎo)分位數(shù)
x1, y1 = len(distances), 0.0
x2, y2 = 1.0, 100.0
x = max(min(self.number_of_chunks, x1), x2)
if x2 == x1:
y = y2
else:
y = y1 + ((y2 - y1) / (x2 - x1)) * (x - x1) # 線(xiàn)性插值
y = min(max(y, 0), 100)
return cast(
float,
np.percentile(distances, y),),
distances- standard_deviation： 標(biāo)準(zhǔn)差偏離法，是統(tǒng)計(jì)學(xué)中表示偏離的常規(guī)方法，這種方法比較適合正態(tài)分布。將所有余弦距離的平均值加上 X 倍的所有余弦距離標(biāo)準(zhǔn)差的值作為閾值，并在那些余弦距離超過(guò)該閾值的位置進(jìn)行切分。X 倍可通過(guò)
breakpoint_threshold_amount設(shè)置，默認(rèn)為 3，這是最常用的值；
SemanticChunker的實(shí)現(xiàn)源碼如下：
import numpy as np
def _calculate_breakpoint_threshold(self, distances: List[float]) -> Tuple[float, List[float]]:
return cast(
float,
np.mean(distances) +
self.breakpoint_threshold_amount * np.std(distances),),
distances- interquartile： 四分位距法，是統(tǒng)計(jì)學(xué)中表示偏離的另一種常規(guī)方法，這種方法計(jì)算分位數(shù)，所以數(shù)據(jù)分布不那么正態(tài)問(wèn)題也不大。將所有余弦距離的平均值加上 X 倍的所有余弦距離四分位距的值作為閾值，并在那些余弦距離超過(guò)該閾值的位置進(jìn)行切分。X 倍可通過(guò)
breakpoint_threshold_amount設(shè)置，默認(rèn)為 1.5，也是最常用的值；
SemanticChunker的實(shí)現(xiàn)源碼如下：
import numpy as np
def _calculate_breakpoint_threshold(self, distances: List[float]) -> Tuple[float, List[float]]:
# 取出25分位（下四分位）和75分位（上四分位）的數(shù)值
q1, q3 = np.percentile(distances, [25, 75])
# 計(jì)算兩個(gè)分位的差值（四分位距）
iqr = q3 - q1
return np.mean(distances) +
self.breakpoint_threshold_amount * iqr, distances- gradient： 梯度法。首先計(jì)算所有余弦距離的變化梯度，變化梯度計(jì)算出來(lái)后，就可以知道哪個(gè)地方余弦距離變化得快，然后將所有變化梯度在第 X 分位數(shù)的值作為閾值，并在那些余弦距離超過(guò)該閾值的位置進(jìn)行切分。第 X 分位數(shù)可通過(guò)
breakpoint_threshold_amount設(shè)置，默認(rèn)為 95。
SemanticChunker的實(shí)現(xiàn)源碼如下：
import numpy as np
def _calculate_breakpoint_threshold(self, distances: List[float]) -> Tuple[float, List[float]]:
# 計(jì)算所有余弦距離的變化梯度
distance_gradient = np.gradient(distances, range(0, len(distances)))
return cast(
float,
np.percentile(distance_gradient,
self.breakpoint_threshold_amount)),
distance_gradient按切分點(diǎn)切分
通過(guò)第4步各方法得到切分點(diǎn)后，就可以按切分點(diǎn)對(duì)文檔進(jìn)行切分（通過(guò)設(shè)置min_chunk_size控制合并較小的塊），就可得到下面類(lèi)似的圖（包括切分位置以及切片）：
SemanticChunker的實(shí)現(xiàn)源碼如下：
def split_text(self, text: str,) -> List[str]:
# 計(jì)算相鄰句子的余弦距離
distances, sentences = self._calculate_sentence_distances(single_sentences_list)
# 計(jì)算切分點(diǎn)
breakpoint_distance_threshold, breakpoint_array = self._calculate_breakpoint_threshold(distances)
indices_above_thresh = [
i
for i, x inenumerate(breakpoint_array)
if x > breakpoint_distance_threshold
]
chunks = []
start_index = 0
# 遍歷切分點(diǎn)來(lái)分割句子
for index in indices_above_thresh:
end_index = index
group = sentences[start_index : end_index + 1]
combined_text = " ".join([d["sentence"] for d in group])
# 通過(guò)設(shè)置 min_chunk_size 來(lái)合并較小的文檔塊
if (
self.min_chunk_size isnotNone
andlen(combined_text) < self.min_chunk_size
):
continue
chunks.append(combined_text)
start_index = index + 1
if start_index < len(sentences):
combined_text = " ".join([d["sentence"] for d in sentences[start_index:]])
chunks.append(combined_text)
return chunks代碼實(shí)踐
原 TypeScript 項(xiàng)目已使用 Python 進(jìn)行了重構(gòu)，后續(xù)將優(yōu)先使用 Python 進(jìn)行代碼實(shí)踐和講解。
其中：RAG.libs 中是封裝好的各種不同作用的模塊，如 RAG/libs/text_splitter.py 是封裝好的文檔切分器，RAG/libs/evaluator.py 是封裝好的評(píng)估器，因此文中不再貼具體的代碼，如需查看具體代碼實(shí)現(xiàn)，請(qǐng)移步到 github 代碼倉(cāng)庫(kù)中查看。
本文完整代碼地址[2]：https://github.com/laixiangran/ai-learn-python/blob/main/RAG/examples/06_semantic_splitting.py
先看下基于句子余弦距離的語(yǔ)義切分的評(píng)估結(jié)果：
從評(píng)估結(jié)果來(lái)看，相較于 RecursiveCharacterTextSplitter 的切分方法，在上下文召回率、上下文相關(guān)性以及答案準(zhǔn)確性都有不同程度的提升。
加載文件
from RAG.libs.file_loader import FileLoader
file_loader = FileLoader(
file_path="RAG/datas/2024少兒編程教育行業(yè)發(fā)展趨勢(shì)報(bào)告.md",
provider="customMarkdownLoader",
)
documents = file_loader.load()語(yǔ)義切分
因?yàn)槲覀兊奈臋n只要是中文，因此需要將 sentence_split_regex修改成可對(duì)中文切分的規(guī)則，如：r"(?<=[。？！\n])"。
from langchain_experimental.text_splitter import SemanticChunker
from RAG.libs.embedding import Embedding
# 向量模型
embeddings = Embedding(model="nomic-embed-text", provider="ollama")
# 使用 SemanticChunker 切分
text_splitter = SemanticChunker(
embeddings=embeddings,
breakpoint_threshold_type="percentile",
sentence_split_regex=r"(?<=[。？！\n])",
)
documents = text_splitter.split_documents(documents=documents)切分后處理
使用SemanticChunker進(jìn)行切分，會(huì)出現(xiàn)較短或者較長(zhǎng)的切片。比如通過(guò)percentile進(jìn)行切分后的結(jié)果可以看到，最小的文檔塊大小只有 1，最大的文檔塊大小有 3346。因此，為了更好的檢索效果，我們一般需要=對(duì)較長(zhǎng)的文檔做二次切分、對(duì)較短的文檔進(jìn)行合并和添加標(biāo)題等其它切分后處理。
count 75 // 總數(shù)
mean 731 // 平均值
std 685 // 標(biāo)準(zhǔn)差
min 1 // 最小值
25% 218 // 25分位值
50% 568 // 50分位值
75% 990 // 75分位值
90% 1535 // 90分位值
97% 2577 // 97分位值
99% 2876 // 99分位值
max 3346 // 最大值將較大的文檔塊進(jìn)行二次切分、合并較小的塊和添加標(biāo)題：
from RAG.libs.text_splitter import (
TextSplitter,
merge_small_documents,
add_headers_to_documents,
)
# 將較大的文檔塊進(jìn)行二次切分
text_splitter = TextSplitter(
provider="recursiveCharacter",
chunk_size=500,
chunk_overlap=0,
)
documents = text_splitter.split_documents(documents)
# 合并較小的塊和添加標(biāo)題
documents = merge_small_documents(documents, merge_max_length=100)
documents = add_headers_to_documents(documents)效果評(píng)估
from RAG.libs.evaluator import BatchEvaluator
eval_result = BatchEvaluator(
chat_model=chat_model,
vector_store=vector_store,
qa_data=qa_data,
top_k=3,
filter=filter,
output_path=output_path,
)結(jié)語(yǔ)
本文介紹了一種更智能的切分方法 - 基于句子余弦距離的語(yǔ)義切分，并通過(guò) langchain-experimental 中的 SemanticChunker 的源碼來(lái)帶大家了解了語(yǔ)義切分的實(shí)現(xiàn)原理。
從最后評(píng)估結(jié)果來(lái)看，相較于 RecursiveCharacterTextSplitter 的切分方法，在上下文召回率、上下文相關(guān)性以及答案準(zhǔn)確性都有不同程度的提升，這說(shuō)明通過(guò)基于句子余弦距離的語(yǔ)義切分方法對(duì)文檔切分優(yōu)化具有一定的可行性，大家可以根據(jù)自己的實(shí)際情況進(jìn)一步驗(yàn)證，歡迎大家留言交流。
引用鏈接
[1] SemanticChunker: https://github.com/langchain-ai/langchain-experimental/blob/main/libs/experimental/langchain_experimental/text_splitter.py#L99
[2] 本文完整代碼地址: https://github.com/laixiangran/ai-learn-python/blob/main/RAG/examples/06_semantic_splitting.py
