对大量文本进行自动信息检索和摘要有许多有用的应用。其中最成熟的应用是 “检索增强生成”（Retrieval Augmented Generation，RAG），它包括从大型语料库中提取相关的文本块--通常是通过语义搜索或其他过滤步骤--以回答用户的问题。然后，由 LLM 对这些文本块进行解释或总结，以提供高质量的准确答案。为了使提取的语块尽可能与问题相关，语块在语义上的一致性非常有帮助，这意味着每个语块都 “涉及 ”一个特定的概念，并且本身就包含有用的信息包。

分块法的应用范围也超出了 RAG。想象一下，我们有一本复杂的文档，比如一本书或一篇期刊论文，想要快速了解其中包含哪些关键概念。如果能将文本聚类为语义连贯的群组，然后以某种方式对每个群组进行总结，这确实有助于加快洞察时间。优秀的软件包 BertTopic（请参阅本文的精彩概述）可以在这方面提供帮助。

无论是作为最终产品，还是在开发过程中，对组块进行可视化处理也会很有帮助。人类是视觉学习者，我们的大脑能更快地从图表和图像中收集信息，而不是文本流。根据我的经验，如果不以某种方式将分块可视化，或者不阅读所有分块，就很难理解分块算法对文本做了哪些处理，以及最佳参数是什么，而这在大型文档中是不切实际的。

在本文中，我们将探讨一种将文本分割成有语义意义的文本块的方法，重点是使用图表来了解发生了什么。在此过程中，除了使用 LLM总结文本块以便快速了解其中的信息外，我们还将涉及降维和嵌入向量的分层聚类。我将在这里使用 Python 3.9、LangChain 和 Seaborn。

什么是语义分块？

语义分块有几种标准类型，它也为本文提供了灵感。假设我们处理的是英文文本，那么最简单的分块形式就是基于字符的分块，即我们选择一个固定的字符窗口，然后简单地将文本分割成相应长度的块。此外，我们还可以在文本块之间添加重叠部分，以保留它们之间的顺序关系。这种方法在计算上非常简单，但不能保证分块具有语义意义，甚至不能保证是完整的句子。

递归分块法通常更有用，在许多应用中被视为首选算法。该过程需要分隔符的分层列表（LangChain 的默认值为[“\n\n”, “\n”, “ ”, “”] ）和目标长度。然后，它会使用分隔符以递归方式分割文本，向下推进列表，直到每个分块小于或等于目标长度。这种方法能更好地保留完整的段落和句子，这一点很好，因为它能使文本块更有可能连贯。不过，它并没有考虑语义： 如果一个句子紧接着上一个句子，又恰好处于语块窗口的末尾，那么这些句子就会被分开。

在语义分块（LangChain 和 LlamaIndex 中都有实现）中，拆分是根据连续分块嵌入之间的余弦距离进行的。因此，我们首先要将文本分成小而连贯的组，或许可以使用递归分块器。

接下来，我们使用训练有素的模型对每个分块进行矢量化，以生成有意义的嵌入。通常情况下，我们会使用基于变换器的双编码器，或者 OpenAI 的 text-embeddings-3-small 等终端，我们在这里使用的就是这种终端。最后，我们会查看后续语块嵌入之间的余弦距离，并选择距离较大的断点。理想情况下，这有助于创建既连贯又有语义区别的文本组。

最近对这一方法进行了扩展，称为语义双块合并（详见本文），试图通过进行第二次合并并使用一些重新分组逻辑来扩展这一方法。因此，举例来说，如果第一次合并在语块 1 和语块 2 之间断开，但语块 1 和语块 3 又非常相似，那么就会产生一个包括语块 1、语块 2 和语块 3 的新组。如果数据块 2 是一个数学公式或代码块，这就很有用。

然而，在进行任何类型的语义分块时，仍然存在一些关键问题： 在我们设置断点之前，分块嵌入之间的距离能有多大，这些分块究竟代表什么？我们关心这个问题吗？这些问题的答案取决于相关应用和文本。

探索断点

让我们用一个例子来说明如何使用语义分块生成断点。我们将实现自己版本的算法，不过如上所述，也有开箱即用的实现方法。我们的演示文本就在这里，它由 GPT-4o 撰写的三篇事实性短文组成，并附录在一起。第一篇是关于保护树木的一般重要性，第二篇是关于纳米比亚的历史，第三篇是关于保护树木用于医疗目的的重要性的深入探讨。题目的选择其实并不重要，但语料库是一个有趣的测试，因为第一篇和第三篇文章有些相似，但又被第二篇截然不同的文章分开。每篇文章还分成几个部分，分别侧重于不同的内容。

我们可以使用一个基本的 RecursiveCharacterTextSplitter 来制作初始块。这里最重要的参数是分块大小和分隔符列表，如果不了解文章的主题，我们通常不知道它们应该是多少。在这里，我选择了相对较小的语块大小，因为我希望初始语块最长只有几句话。我还选择了分隔符，以避免拆分句子。

# tools from the text chunking package mentioned in this article
from text_chunking.SemanticClusterVisualizer import SemanticClusterVisualizer
# put your open ai api key in a .env file in the top level of the package
from text_chunking.utils.secrets import load_secrets
# the example text we're talking about 
from text_chunking.datasets.test_text_dataset import TestText
# basic splitter
from langchain_text_splitters import RecursiveCharacterTextSplitter
import seaborn as sns

splitter = RecursiveCharacterTextSplitter(
        chunk_size=250,250,
        chunk_overlap=0,
        separators=["\n\n", "\n", "."],
        is_separator_regex=False
)

接下来，我们可以分割文本。如果分割器生成的文本块小于min_chunk_len参数，min_chunk_len参数就会发挥作用。如果出现这种情况，该分块就会被附加到前一个分块的末尾。

original_split_texts = semantic_chunker.split_documents(
    splitter, 
    TestText.testing_text, 
    min_chunk_len=100, 100, 
    verbose=True
)
### Output
# 2024-09-14 16:17:55,014 - Splitting text with original splitter
# 2024-09-14 16:17:55,014 - Creating 53 chunks
# Mean len: 178.88679245283018
# Max len: 245
# Min len: 103

现在，我们可以使用嵌入模型来嵌入拆分。你会在 SemanticClusterVisualizer 的类中看到，默认情况下我们使用 text-embeddings-3-small 模型。这将创建一个包含53个向量的列表，每个向量的长度为1536。直观地说，这意味着每个语块的语义都是在 1536 维空间中表示的。这对于可视化来说并不是一件好事，因此我们稍后将讨论降维问题。

original_split_text_embeddings = semantic_chunker.embed_original_document_splits(original_split_texts)

运行语义分块器会生成这样一张图。我们可以把它想象成一个时间序列，其中 x 轴代表整个文本中的字符距离。y 轴表示后续分块嵌入之间的余弦距离。当距离值超过第 95 百分位数时，就会出现断点。

根据我们对文本的了解，这种模式是合理的--有三个大主题，每个主题有几个不同的部分。不过，除了两个大的峰值外，其他断点的位置并不明确。

这就涉及到主观性和迭代的问题--根据我们的应用，我们可能需要更大或更小的区块，而使用图表来帮助引导我们真正阅读哪些区块是非常重要的。

有几种方法可以将文本分成更细小的块。第一种方法是降低百分位数阈值，以设置断点。

这样就产生了 4 个非常小的语块和 8 个较大的语块。例如，如果我们看一下前 4 个分块，从语义上讲，这些分块似乎是合理的，但我认为第 4 个分块有点过长，因为它包含了第一篇文章中 “经济重要性”、“社会重要性 ”和 “结论 ”部分的大部分内容。

另一种方法是递归应用相同的阈值，而不是仅仅改变百分位数阈值。我们首先在整个文本上创建断点。然后，对于每个新创建的文本块，如果该文本块高于某个长度阈值，我们就在该文本块内创建断点。如此反复，直到所有文本块都低于长度阈值。虽然有些主观，但我认为这更接近人类的做法，因为他们会首先识别出非常不同的文本组，然后反复缩小每个文本组的大小。

如下图所示，它可以通过堆栈来实现。

def get_breakpoints(
    embeddings: List[np.ndarray],
    start: int = 0,
    end: int = None,
    threshold: float = 0.95,
) -> np.ndarray:
    """
    Identifies breakpoints in embeddings based on cosine distance threshold.
    Args:
        embeddings (List[np.ndarray]): A list of embeddings.
        start (int, optional): The starting index for processing. Defaults to 0.
        end (int, optional): The ending index for processing. Defaults to None.
        threshold (float, optional): The percentile threshold for determining significant distance changes. Defaults to 0.95.
    Returns:
        np.ndarray: An array of indices where breakpoints occur.
    """
    if end is not None:
        embeddings_windowed = embeddings[start:end]
    else:
        embeddings_windowed = embeddings[start:]
    len_embeddings = len(embeddings_windowed)
    cdists = np.empty(len_embeddings - 1)
    
    # get the cosine distances between each chunk and the next one
    for i in range(1, len_embeddings):
        cdists[i - 1] = cosine(embeddings_windowed[i], embeddings_windowed[i - 1])
    
    # get the breakpoints
    difference_threshold = np.percentile(cdists, 100 * threshold, axis=0)
    difference_exceeding = np.argwhere(cdists >= difference_threshold).ravel()
    return difference_exceeding
def build_chunks_stack(
    self, length_threshold: int = 20000, cosine_distance_percentile_threshold: float = 0.95
) -> np.ndarray:
    """
    Builds a stack of text chunks based on length and cosine distance thresholds.
    Args:
        length_threshold (int, optional): Minimum length for a text chunk to be considered valid. Defaults to 20000.
        cosine_distance_percentile_threshold (float, optional): Cosine distance percentile threshold for determining breakpoints. Defaults to 0.95.
    Returns:
        np.ndarray: An array of indices representing the breakpoints of the chunks.
    """
    
    # self.split texts are the original split texts 
    # self.split text embeddings are their embeddings 
    S = [(0, len(self.split_texts))]
    all_breakpoints = set()
    while S:
        # get the start and end of this chunk
        id_start, id_end = S.pop()
        # get the breakpoints for this chunk
        updated_breakpoints = self.get_breakpoints(
            self.split_text_embeddings,
            start=id_start,
            end=id_end,
            threshold=cosine_distance_percentile_threshold,
        )
        updated_breakpoints += id_start
        # add the updated breakpoints to the set
        updated_breakpoints = np.concatenate(
            (np.array([id_start - 1]), updated_breakpoints, np.array([id_end]))
        )
        
        # for each updated breakpoint, add its bounds to the set and
        # to the stack if it is long enough
        for index in updated_breakpoints:
            text_group = self.split_texts[id_start : index + 1]
            if (len(text_group) > 2) and (
                self.get_text_length(text_group) >= length_threshold
            ):
                S.append((id_start, index))
            id_start = index + 1
        all_breakpoints.update(updated_breakpoints)
    
    # get all the breakpoints except the start and end (which will correspond to the start
    # and end of the text splits)
    return np.array(sorted(all_breakpoints))[1:-1]

我们对length_threshold的选择也是主观的，可以从绘图中获得信息。在本例中，1000 的阈值似乎效果不错。它很好地将文章划分为短小而有意义的不同部分。

通过观察第一篇文章对应的语块，我们可以发现它们与 GPT4-o 在撰写这篇文章时创建的不同部分非常吻合。很明显，就这篇特殊的文章而言，我们本可以只在“\n\n ”上进行拆分就完事了，但我们想要一种更通用的方法。

对语义分块进行聚类

既然我们已经有了一些候选语义块，那么看看它们之间的相似程度可能会有所帮助。这将有助于我们了解它们包含哪些信息。我们将对语义块进行嵌入，然后使用 UMAP 将嵌入结果的维度降低到二维，以便绘制它们。

UMAP 是 Uniform Manifold Approximation and Projection 的缩写，是一种强大的通用降维技术，可以捕捉非线性关系。有关其工作原理的完整解释，请点击此处。这里使用它的目的是在二维绘图中捕捉 1536-D 空间中嵌入的块之间存在的一些关系

from umap import UMAP
dimension_reducer = UMAP(
            n_neighbors=5, 
            n_components=2, 
            min_dist=0.0, 
            metric="cosine", 
            random_state=0
)
reduced_embeddings = dimension_reducer.fit_transform(semantic_embeddings)
splits_df = pd.DataFrame(
            {
                "reduced_embeddings_x": reduced_embeddings[:, 0],
                "reduced_embeddings_y": reduced_embeddings[:, 1],
                "idx": np.arange(len(reduced_embeddings[:, 0])),
            }
)
splits_df["chunk_end"] = np.cumsum([len(x) for x in semantic_text_groups])
ax = splits_df.plot.scatter(
            x="reduced_embeddings_x", 
            y="reduced_embeddings_y", 
            c="idx", 
            cmap="viridis"
)
ax.plot(
            reduced_embeddings[:, 0],
            reduced_embeddings[:, 1],
            "r-",
            linewidth=0.5,
            alpha=0.5,
)

UMAP 对 n_neighbors 参数相当敏感。一般来说，n_neighbors 的值越小，算法就越注重利用局部结构来学习如何将数据投影到更低的维度。将 n_neighbors 设得太小会导致投影无法很好地捕捉数据的大尺度结构，一般来说，随着数据点数量的增加，n_neighbors 的值也应该增加。

我们的数据投影如下所示，其信息量相当大： 很明显，我们有三个意义相似的聚类，其中第一聚类和第三聚类之间的相似程度要高于第二聚类和第三聚类之间的相似程度。上图中的 idx 色条显示了数据块的编号，而红线则显示了数据块的顺序。

自动聚类如何？如果我们想将语块归类为更大的片段或主题，这将会很有帮助，例如，在混合搜索的 RAG 应用程序中，这可以作为筛选的有用元数据。我们还可以将文本中相距甚远但含义相似的语块进行分组。

有许多聚类方法可以在此使用。HDBSCAN 是一种可能的方法，也是 BERTopic 软件包推荐的默认方法。不过，在这种情况下，分层聚类似乎更有用，因为它可以让我们了解出现的任何群体的相对重要性。要运行分层聚类，我们首先要使用 UMAP 将数据集的维度降低到较少的分量。只要 UMAP 在这里运行良好，成分的确切数量应该不会对生成的聚类产生重大影响。然后，我们使用 scipy 的层次结构模块进行聚类，并使用 seaborn 绘制结果图。

from scipy.cluster import hierarchy
from scipy.spatial.distance import pdist
from umap import UMAP
import seaborn as sns
# set up the UMAP
dimension_reducer_clustering = UMAP(
            n_neighbors=umap_neighbors,
            n_components=n_components_reduced,
            min_dist=0.0,
            metric="cosine",
            random_state=0
)
reduced_embeddings_clustering = dimension_reducer_clustering.fit_transform(
    semantic_group_embeddings
)
# create the hierarchy
row_linkage = hierarchy.linkage(
    pdist(reduced_embeddings_clustering),
    method="average",
    optimal_ordering=True,
)
# plot the heatmap and dendogram
g = sns.clustermap(
    pd.DataFrame(reduced_embeddings_clustering),
    row_linkage=row_linkage,
    row_cluster=True,
    col_cluster=False,
    annot=True,
    linewidth=0.5,
    annot_kws={"size": 8, "color": "white"},
    cbar_pos=None,
    dendrogram_ratio=0.5
)
g.ax_heatmap.set_yticklabels(
  g.ax_heatmap.get_yticklabels(), rotation=0, size=8
)

结果也相当有参考价值。在这里，n_components_reduced 为 4，因此我们将嵌入的维度降低到 4D，从而得到一个包含 4 个特征的矩阵，其中每一行代表一个语义块。分层聚类确定了两个主要群组（即树木和纳米比亚）、树木中的两个大的子群组（即医疗用途和其他）以及其他一些可能值得探索的群组。

请注意，BERTopic 在主题可视化方面也使用了类似的技术，这可以看作是本文内容的延伸。

这对我们探索语义分块有何帮助？根据结果，我们可能会选择将某些语块分组。这也是相当主观的，尝试几种不同类型的分组可能很重要。比方说，我们看了树枝图后决定要进行 8 个不同的分组。然后，我们就可以对层次结构进行相应的切割，返回与每个组相关的聚类标签并绘制它们。

cluster_labels = hierarchy.cut_tree(linkage, n_clusters=n_clusters).ravel()
dimension_reducer = UMAP(
  n_neighbors=umap_neighbors, 
  n_components=2, 2, 
  min_dist=0.0, 
  metric="cosine", 
  random_state=0
)
reduced_embeddings = dimension_reducer.fit_transform(semantic_embeddings)
splits_df = pd.DataFrame(
            {
                "reduced_embeddings_x": reduced_embeddings[:, 0],
                "reduced_embeddings_y": reduced_embeddings[:, 1],
                "cluster_label": cluster_labels,
            }
        )
splits_df["chunk_end"] = np.cumsum(
            [len(x) for x in semantic_text_groups]
        ).reshape(-1, 1)
ax = splits_df.plot.scatter(
            x="reduced_embeddings_x",
            y="reduced_embeddings_y",
            c="cluster_label",
            cmap="rainbow",
        )
ax.plot(
  reduced_embeddings[:, 0],
  reduced_embeddings[:, 1],
  "r-",  
  linewidth=0.5,
  alpha=0.5,
)

结果如下图所示。我们有 8 个聚类，它们在二维空间中的分布看起来很合理。这再次说明了可视化的重要性： 根据文本、应用和利益相关者的不同，正确的群组数量和分布很可能不同，而检查算法运行情况的唯一方法就是绘制这样的图表

标记聚类

假设经过上述步骤的几次反复之后，我们已经确定了自己满意的语义分割和聚类。那么我们就有必要问一问，这些聚类究竟代表了什么？显然，我们可以通过阅读文本来找出答案，但对于大型语料库来说，这并不现实。相反，我们可以使用 LLM 来帮忙。具体来说，我们将把与每个聚类相关的文本输入 GPT-4o-mini，并要求它生成摘要。对于 LangChain 来说，这是一项相对简单的任务，代码的核心部分如下所示

import langchain
from langchain.prompts import PromptTemplate
from langchain_core.output_parsers.string import StrOutputParser
from langchain.callbacks import get_openai_callback
from dataclasses import dataclass

@dataclass
class ChunkSummaryPrompt:
    system_prompt: str = """
        You are an expert at summarization and information extraction from text. You will be given a chunk of text from a document and your
        task is to summarize what's happening in this chunk using fewer than 10 words. 
        Read through the entire chunk first and think carefully about the main points. Then produce your summary.
        Chunk to summarize: {current_chunk}
    """
    prompt: langchain.prompts.PromptTemplate = PromptTemplate(
        input_variables=["current_chunk"],
        template=system_prompt,
    )
class ChunkSummarizer(object):
    def __init__(self, llm):
        self.prompt = ChunkSummaryPrompt()
        self.llm = llm
        self.chain = self._set_up_chain()
    def _set_up_chain(self):
        return self.prompt.prompt | self.llm | StrOutputParser()
    def run_and_count_tokens(self, input_dict):
        with get_openai_callback() as cb:
            result = self.chain.invoke(input_dict)
        return result, cb
llm_model = "gpt-4o-mini"
llm = ChatOpenAI(model=llm_model, temperature=0, api_key=api_key)
summarizer = ChunkSummarizer(llm)

在我们的 8 个集群上运行这个程序，并用数据图谱绘制结果，结果如下：

可视化这些群组的另一种方法与第 2 节中的图表类似，我们在 x 轴上绘制累积字符数，并显示群组之间的边界。回想一下，我们有 18 个语义块，现在将它们进一步分为 8 个群组。这样绘制可以显示文本语义内容从头到尾的变化，突出相似内容并不总是相邻的事实，并直观地显示语义块的相对大小。

在更大的语料库上进行测试

到目前为止，我们已经出于演示目的在相对较小的文本量上测试了这一工作流程。理想情况下，它也能在较大的语料库中发挥作用，而无需进行重大修改。为了测试这一点，让我们在从古腾堡计划下载的一本书上试试看，这里我选择了《绿野仙踪》。这项任务要困难得多，因为小说通常不像纪实散文那样按语义清晰的章节排列。虽然它们通常按章节编排，但故事线可能以连续的方式 “拱起”，或在不同主题之间跳来跳去。如果能利用语义块分析从不同作者的作品中了解他们的风格，那将是一件非常有趣的事情。

步骤 1：嵌入并生成断点

from text_chunking.SemanticClusterVisualizer import SemanticClusterVisualizer
from text_chunking.utils.secrets import load_secrets
from text_chunking.datasets.test_text_dataset import TestText, TestTextNovel
from langchain_text_splitters import RecursiveCharacterTextSplitter
secrets = load_secrets()
semantic_chunker = SemanticClusterVisualizer(api_key=secrets["OPENAI_API_KEY"])
splitter = RecursiveCharacterTextSplitter(
        chunk_size=250,
        chunk_overlap=0,
        separators=["\n\n", "\n", "."],
        is_separator_regex=False
)
original_split_texts = semantic_chunker.split_documents(
    splitter, 
    TestTextNovel.testing_text, 
    min_chunk_len=100, 
    verbose=True
)
original_split_text_embeddings = semantic_chunker.embed_original_document_splits(original_split_texts)
breakpoints, semantic_groups = semantic_chunker.generate_breakpoints(
    original_split_texts,
    original_split_text_embeddings,
    length_threshold=10000 #may need some iteration to find a good value for this parameter
)

这样就生成了 77 个大小不一的语义块。在这里做了一些抽查，让我确信它运行得相对较好，而且许多语块最终都是在章节边界或接近章节边界的地方划分的，这很有意义。

步骤 2：聚类并生成标签

在查看分层聚类树状图时，我决定尝试将聚类减少到 35 个。结果发现下图左上方有一个异常值（聚类标识 34），原来是文本末尾的一组大块内容，其中包含对该书发行条款的冗长描述。

除第一组外，其他各组都很好地概括了小说中的主要事件。对与每个组相关的实际文本进行的快速检查证实，它们都是相当准确的概括，不过，确定各组的边界也是非常主观的。

GPT-4o-mini 将异常群组标记为 “古腾堡计划允许自由传播不受保护的作品”。与这个标签相关的文本对我们来说并不特别有趣，因此我们将其删除并重新绘制结果图。这将使小说的结构更容易看清。

如果我们对更大的集群感兴趣呢？如果我们专注于高层次结构，树枝图显示大约有六个语义块群，绘制如下。

在这一语义空间中，有很多相距甚远的点之间来回跳跃，这表明主题经常会突然发生变化。考虑各组之间的联系也很有趣： 例如，4 和 5 之间没有任何联系，而 0 和 1 之间却有大量的往返。

我们能总结出这些较大的聚类吗？事实证明，我们的提示语似乎并不适合这种规模的群组，它所产生的描述要么过于具体到群组的某一部分（即群组 0 和 4），要么过于模糊，没有太大帮助。改进提示语的工程设计--可能涉及多个总结步骤--可能会改善这里的结果。

尽管这些名称并不实用，但这一按群组着色的文本片段情节仍然为我们选择性阅读文本提供了参考。我们看到，这本书的开头和结尾都在同一个群组上，很可能是关于多萝西、托托和他们的家的描述--这与《绿野仙踪》的故事架构一致，都是一次旅行和随后的回归。第 1 组主要是关于认识新角色，这主要发生在开头，但在整本书中也会定期出现。第 2 组和第 3 组与翡翠城和魔法师有关，而第 4 组和第 5 组则分别与旅行和战斗有关。

总结

在这里，我们深入探讨了语义分块的概念，以及它如何与降维、聚类和可视化相辅相成。这篇文章给我们的主要启示是，在决定最合适的方法之前，系统地探索不同的分块技术和参数对文本的影响非常重要。