一种简单的文本聚类算法：single-pass

0引言

在生活里，我们经常需要将一些物品按类别存放。比如，我们需要把所有的袜子集合起来，放在一个抽屉里；把所有的拖鞋集合起来，放在鞋架上；等等。

类似的，数据类业务里，经常需要将样本按照类别聚合，然后以类别为单位对数据进行处理或使用。例如，我们可以训练一个KNN,对文档的主题类别进行判断，这样所有的文档就带有的主题标签，在搜索或者推荐的时候就可以基于主题标签对搜索结果进行某种过滤。当然了，这个方案要求我们定义一下主题的类别体系，还需要收集一定量的有标签数据(通常来说，这就是要标注数据了)。

当我们的聚合粒度比较小，到了“事件”这么小，人工定义分类体系就不现实了。一般我们采用聚类算法来实现较小粒度的文档聚合。

最有名的聚类算法应该是k-means了。这个算法思想简单直观，效果还不错，大家因此比较喜欢。实际上，还有一些比k-means更简单的聚类算法，比如single-pass。

1 single-pass简介

Single-pass就像它的名字一样，直白简洁。

1.1 single-pass的工作流程

如图1-1，我在遍历文本。一开始的时候，一个簇都没有。

图1-1 混沌未开，没有簇

然后，我遍历到了第1篇文档。通过考察1号文档与所有簇的匹配度，发现没有足够匹配的。因此，我认为1号文档表述了一个独特的主题，就以1号文档为核心定义了一个新的簇，如图1-2。

图1-2 混沌初开，有一个簇

接下来遍历到了第2篇文档，它和1号簇(的核心文档)非常匹配，我认为它俩描述了同一个主题，因此把2号文档添加到了1号簇中。

然后是3号文档。它和所有的簇都不够匹配，描述了一个新的主题。因此，我以3号文档为核心定义了一个新的簇，如图1-3。

重复图1-2和1-3的操作，直到遍历完毕，我们会得到若干簇。分析一下每个簇内文档的内容，就可以给它们各自添加一些标题、标签之类的描述——这就算完成无监督的聚合了。

图1-3 定义一个新的簇

1.2 single-pass的关键组件

1.2.1一个簇的结构与创建条件

在single-pass中，一个簇有两个必备组成部分：（1）簇成员;（2）簇核心。

簇成员指的是构成一个簇的所有文档。聚类任务的结果，就是将原始数据集划分为若干小集合——这些小集合内部有一定的相似性，就是常说的“簇”。“簇”是从“cluster”翻译过来的一个词语，本意是“聚集的团或堆”。

簇核心是我们对一个簇的简要表示。最简单情况下，我们会以第一个进入这个簇的文档作为簇核心。如图1-1，混沌未开的时候，1号文档由于找不到匹配的簇，只好自立门户、建立一个新的簇，并自荐为这个簇的核心。

簇核心是用来干什么的呢？“匹配”是什么呢？

1.2.2“匹配”与否的判断——文本相似度和阈值

当一篇文档内容与某个簇的内容足够接近或者相关(比较多的情况是相似)，我们称这篇文档和该簇匹配。为什么这里要强调“匹配”呢？首先，文档和簇不是一个维度的概念，前者是一篇文档，而后者是若干文档的集合；其次，我们有时候会用相似度计算之外的算法来判断，一篇文档是否应该加入一个簇，比如用分类算法如图1-2，当遍历到2号文档的时候，我需要判断已有簇中，有没有和2号文档“匹配”的。

具体计算方式就是，直接计算所有簇的核心与2号文档的匹配度(这里是相似度)，然后将2号文档添加到相似度高于阈值的簇中——为了简单，这里规定一篇文档只能属于一个簇，因此在发现第一个匹配的簇时，就可以停止计算了。

1.2.3计算复杂度分析

我们来捋一下原始版single-pass的计算时间复杂度。为了简单，这里定义，一次文本相似度计算的代价是1；其他的都是毛毛雨啦。

极端悲观情况下，每一篇文档都特立独行、各自成簇：（1）1号文档不需要计算相似度，当前簇个数是1；（2）2号文档需要和一个簇计算相似度，当前簇个数是2；（3）3号文档需要和2个簇计算相似度，当前簇个数是3；…;N号文档需要和N-1个簇计算相似度。用高斯小朋友的算法可以快速算出，我们需要进行的相似度计算次数是:

计算时间复杂度是O(N方)。假设一次相似度计算耗时0.001秒，当输入是1万篇文档的时候，聚类任务的总耗时就是10万秒，也就是超过一天。这时候，老板的脸色就如图1-4。为了大家的饭碗，我们要想办法让聚类任务算的快一点。

1-4 等你算出来，公司就黄了

1.3 single-pass文本聚类的python实现

<code>'''
Created on 2019年11月9日 

@author: Administrator
'''
from pyhanlp import HanLP
HanLP.Config.ShowTermNature = False

class Document():

    def __init__(self, doc_id, content, features):
        self.doc_id = doc_id
        self.features = features#文本的特征，这里是分词结果
        self.content = content#原文
       

class Cluster():

    def __init__(self, cluster_id, center_doc_id):
        self.cluster_id = cluster_id#簇的id，用来从map中获取这个簇的信息
        self.center_doc_id = center_doc_id#核心文档的id，用来从map红获取这个文档的信息。为了减少文档信息的备份数量，簇里只存储这个
        self.members = [center_doc_id]#簇成员的id列表。由于只遍历一遍(这是single-pass的核心竞争力之一)，不存在重复的可能，这里使用list

    def add_doc(self, doc_id):
        self.members.append(doc_id)


#一个简单的single-pass文本聚类算法
class SimpleSinglePass():

    def __init__(self):

        self.document_map = {}#存储文档信息，id-content结构。当然value也可以使用对象存储文档的其他信息。
        self.cluster_list = []#存储簇的信息，id-cluster_object结构。

    #提取文本特征
    def get_words(self, text): 

        words = HanLP.segment(text)
        words = list(map(str, words))
        return words

    #输入文档列表，进行聚类。现实中，我们遇到的文档会带有id等信息，这里为了简单，只有文本内容，所以需要生成id,一遍存取。
    def fit(self, document_list):
        #对文档进行预处理
        self.preprocess(document_list)
        self.clutering()

    def similar(self, cluster, document):
        cluster_feature = set(self.document_map[cluster.center_doc_id].features)
        document_feature = set(document.features)
#         print(cluster_feature, document_feature)
        similarity = len(cluster_feature & document_feature) / len(cluster_feature | document_feature)
        if similarity > 0.2:
            return True
        else:
            return False

    def clutering(self):
        for doc_id in self.document_map:
#             print(doc_id, self.document_map[doc_id])
            if_特立独行 =  True
            for cluster in self.cluster_list:
                if self.similar(cluster, self.document_map[doc_id]):
                    cluster.add_doc(doc_id)
                    if_特立独行 = False
                    break
            if if_特立独行:
                new_cluser_id = "cluster_" + str(len(self.cluster_list))
                print(new_cluser_id)
                new_cluster = Cluster(new_cluser_id, doc_id)
                self.cluster_list.append(new_cluster)

    #对所有文档分词，并生成id
    def preprocess(self, document_list):
        for i in range(len(document_list)):
            doc_id = "document_" + str(i)
            content = document_list[i]
            words = self.get_words(content)
            document = Document(doc_id, content, words) 

            self.document_map[doc_id] = document

    #打印所有簇的简要内容
    def show_clusters(self):
        for cluster in self.cluster_list:
            print(cluster.cluster_id, cluster.center_doc_id, cluster.members)

if __name__ == '__main__':
    docs = ["我爱北京天安门，天安门上太阳升。",
            "我要开着火车去北京，看天安门升旗。",
            "我们的家乡，在希望的田野上。",
            "我的老家是一片充满希望的田野。"]
    single_passor = SimpleSinglePass()
    single_passor.fit(docs)
    single_passor.show_clusters()/<code>

2升级single-pass的主要策略

2.1用倒排索引加速聚类

前面的方案中，每遍历到一个文档，我们都需要从所有的簇中找一个匹配的，作为文档的归属。我们可以换一个角度来理解这个操作：从已有簇中，搜索一个与当前文档匹配的簇。

一般来说，在大量数据中搜索时，我们会用倒排索引来提升搜索的效率。

2.1.1倒查索引的构建方式

如果文档A的所有词语，没有出现在文档B中，即A和B的词语列表的交集是空集。如果两篇文档没有一个“共同词语”，真相只有一个，那就是二者的内容完全无关(当然凡事无绝对，总有漏网之鱼)；反过来讲，只有文档B包含A中的某一个词语，B才有可能与A相似，即成为A的候选相似文档。一般来说，文本切分(分词，ngram之类)后得到的特征是比较稀疏的，除了一些高频词，大部分词语只出现在一小部分文档里。“小部分”的意思是比全体文档要小若干数量级。我们只要在这一小部分文档里去找与A相似的。

我们可以用倒查索引来把包含一个词语的所有簇都存储起来，这样，就可以在遍历文档的时候，把一篇文档内词语对应的所有候选匹配簇找出来。接下来，遍历候选簇、找到匹配簇，然后加入它。

这个方案有个隐患：像“的”这样没啥信息量的词语，以及“我”这样的高频词，会出现在几乎每一个文档中。这会导致一些篇文档的候选簇个数接近全体规模。这个系统还需要优化一下。

2.1.2用python实现引入倒排的single-pass

2.2特征工程

2.2.1使用关键词作为为本特征

简单说来，2.1.1说到的“隐患”，是一些低质高频词语出现导致的。自然地，我们可以用停用词过滤、关键词抽取等方法，把词汇表精简一下。这样，每个词语的候选簇数量就可以控制得比较稳定。这时候，大家的饭碗就保住了，如图2-1。

图2-1 姚式表情

2.2.2基于关键词特征的single-pass

<code>import sys, os 
path = os.path.dirname(os.getcwd())
sys.path.append(path)
from single_pass_v1 import SinglePassV1, Cluster, Document

#增加倒排索引
class SinglePassV2(SinglePassV1):
    
    def __init__(self):
        self.document_map = {}#存储文档信息，id-content结构。当然value也可以使用对象存储文档的其他信息。
        self.cluster_map = {}#存储簇的信息，id-cluster_object结构。
        self.cluster_iindex = {}#word-cluster_ids结构

    #对所有文档分词，并生成id
    def preprocess(self, document_list):
        for i in range(len(document_list)):
            doc_id = "document_" + str(i)
            content = document_list[i]
            words = self.get_key_words(content)
            document = Document(doc_id, content, words)
            self.document_map[doc_id] = document
            
    #提取文本特征。这里使用文档内频次最高的K个词语。实际应用中可以用TF-IDF
    def get_key_words(self, text, K=5):
        words = self.get_words(text)
        word_freq = {}
        for word in words:
            word_freq[word] = word_freq.get(word, 0) + 1
        keywords = sorted(word_freq.items(), key=lambda x: x[1],reverse=True)[:K]
        keywords = list(map(lambda x: x[0], keywords))
        return keywords
        
    def clutering(self):
        for doc_id in self.document_map:
#             print(doc_id, self.document_map[doc_id]) 

            words = self.document_map[doc_id].features
            if_特立独行 =  True
            for cluster_id in self.get_cand_clusters(words):
                cluster = self.cluster_map[cluster_id]
                if self.similar(cluster, self.document_map[doc_id]):
                    cluster.add_doc(doc_id)
                    if_特立独行 = False
                    break
            if if_特立独行:
                new_cluser_id = "cluster_" + str(len(self.cluster_map))
                print(new_cluser_id)
                new_cluster = Cluster(new_cluser_id, doc_id)
                self.cluster_map[new_cluser_id] = new_cluster
                
                for word in self.document_map[new_cluster.center_doc_id].features:
                    if word not in self.cluster_iindex: self.cluster_iindex[word] = []
                    self.cluster_iindex[word].append(new_cluser_id)
                    
    def get_cand_clusters(self, words):
        cand_cluster_ids = []
        for word in words:
            cand_cluster_ids.extend(self.cluster_iindex.get(word, []))
        return cand_cluster_ids
        
    #打印所有簇的简要内容
    def show_clusters(self):
        for cluster_id in self.cluster_map:
            cluster = self.cluster_map[cluster_id]
            print(cluster.cluster_id, cluster.center_doc_id, cluster.members)
            
if __name__ == '__main__':
    docs = ["我爱北京天安门，天安门上太阳升。",
            "我要开着火车去北京，看天安门升旗。",
            "我们的家乡，在希望的田野上。",
            "我的老家是一片充满希望的田野。"]
    single_passor = SinglePassV2()
    single_passor.fit(docs)
    single_passor.show_clusters()/<code>

3结语

Single-pass是一个挺不错的聚类算法，我们可以在这个算法的基础上做很多变化，来满足各种各样的需求。

注意：本文为李鹏宇(知乎个人主页https://www.zhihu.com/people/py-li-34)原创作品，受到著作权相关法规的保护。如需引用、转载，请注明来源信息：（1）作者名，即“李鹏宇”；（2）原始网页链接，即https://zhuanlan.zhihu.com/p/91007237。如有疑问，可发邮件至我的邮箱:[email protected]。

閱讀更多 數據科學漫遊 的文章

關鍵字: 类别 Python 鞋架