跳转至

向 spaCy 添加指定分词器(Jieba,CKIP Transformers)

spaCy支持多种人类语言，包括中文。为了将中文文本分割成单词，spaCy支持使用Jieba或PKUSeg。但是，在繁体中文方面，它们都没有在准确性上击败CKIP Transformers（请参阅我的帖子 进行比较）。因此，我将展示如何插入 CKIP Transformers 以spaCy充分利用两者。

出于演示的目的，我将把这种集成放在从文本中提取关键字的管道中。与其他 NLP 任务相比，关键字提取是一项相对容易的工作。TextRank 和 RAKE 似乎是最广泛采用的关键字提取算法之一。我尝试了本文中提到的大多数方法，但似乎没有任何简单的 TextRank 或 RAKE 实现可以为繁体中文文本产生不错的结果。所以这篇文章的第一部分介绍了一个实际工作的管道，第二部分记录了其他失败的方法。

设置变量

让我们从定义我们的关键字提取程序的用户可能想要修改的两个变量开始：CUSTOM_STOPWORDS对于用户肯定希望从关键字候选中排除的单词列表以及KW_NUM他们希望从文档中提取的关键字的数量。

1
2
CUSTOM_STOPWORDS  =  [ "人群" , "朋友" , "市民" , "人群" , "全民" , "市民" , "人士" , "里民" , "影本" , "系统"  , "项目" , "证人" , "资格" , "公民" , "对象" , "个人" ]
KW_NUM = 10

预处理文本

我以高雄市政府土地管理局的公告作为示例文本，但您基本上可以使用任何繁体中文文本来测试程序。

  • 单击Open in Colab此页面右上角的 。
  • 单击File，然后单击Save a copy in Drive。
  • 用您自己的文字替换以下文字。
  • 单击Runtime，然后单击Run all。
  • 转到部分Put it together以查看结果。

我发现这个轻量级库nlp2对于文本清理非常方便。该clean_all函数删除 URL 链接、HTML 元素和未使用的标记。 安装 !pip install nlp2

1
2
3
from  nlp2  import  clean_all
text = clean_all(raw_text) #raw_text是文本
text[-300:]

安装spacy和ckip-transformers

1
2
3
4
!pip install -U pip setuptools wheel
!pip install -U spacy
!python -m spacy download zh_core_web_sm
!pip install -U ckip-transformers

标记文本ckip-transformers

让我们创建一个用于分词的驱动程序和一个用于词性的驱动程序。CKIP Transformers 还具有用于命名实体识别的内置驱动程序，即 CkipNerChunker

提示：默认情况下，colab使用 CPU。如果你想使用 GPU 来加速分词，ws_driver可以这样初始化：ws_driver = CkipWordSegmenter(device=-1)

1
2
3
from ckip_transformers.nlp import CkipWordSegmenter, CkipPosTagger
ws_driver  = CkipWordSegmenter()
pos_driver = CkipPosTagger()

重要提示：ws_driver()即使您只处理单个文本，也要确保输入是一个列表。否则，单词将无法正确分割。请注意， pos_driver()的输入是ws_driver()的输出。

1
2
3
4
5
6
7
8
ws  = ws_driver([text])
pos = pos_driver(ws)
tokens = ws[0]
print(tokens)

#结巴分词
import jieba
print(list(jieba.cut(text)))

将标记化结果提供给spacy使用WhitespaceTokenizer

spaCy的官方网站描述了添加自定义标记器的几种方法。最简单的是定义 WhitespaceTokenizer类，它在空格字符上标记文本。然后可以将标记化的输出馈送到管道中的后续操作，包括tagger用于词性 (POS) 标记、parser依赖解析和ner命名实体识别。这主要是因为tokenizer创建了一个Doc对象，而其他三个步骤对Doc对象进行操作，如下图所示。

1
2
3
4
5
6
7
from spacy.tokens import Doc
class WhitespaceTokenizer:
    def __init__(self, vocab):
        self.vocab = vocab
    def __call__(self, text):
        words = text.strip().split()
        return Doc(self.vocab, words=words)

接下来，让我们加载zh_core_web_sm中文模型，我们将需要它来进行词性标注。那么关键的部分来了：nlp.tokenizer = WhitespaceTokenizer(nlp.vocab). 这行代码将 Jieba 的默认分词器设置为WhitespaceTokenizer，我们刚刚在上面定义。

1
2
3
4
5
6
7
8
9
import spacy
nlp = spacy.load("zh_core_web_trf")
nlp.tokenizer = WhitespaceTokenizer(nlp.vocab)

token_str = " ".join(tokens)

doc = nlp(token_str)
print([token.text for token in doc])
print([token.pos_ for token in doc])

请注意 spaCy 使用粗略的标签，例如NOUN和VERB。相比之下，CKIP Transformers 采用更细粒度的标记集，例如Nc用于位置名词和Nd时间名词。这是 CKIP Transformers 生成的相同文本的 POS 标签。我们将使用 spaCy 的 POS 标记来过滤掉我们不想要的关键字候选池中的单词。

1
2
pos_tags = pos[0]
print(pos_tags)

将停用词spaCy从简体转换为台湾繁体

spaCy 带有一组内置的停用词（基本上是我们想忽略的词），可通过spacy.lang.zh.stop_words. 为了充分利用它，让我们借助OpenCC. OpenCC不只是机械地转换字符。它能够将单词从简体字转换为台湾普通话的等效措辞，这是由s2twp.json.

!pip install OpenCC

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
import opencc

from spacy.lang.zh.stop_words import STOP_WORDS
converter = opencc.OpenCC('s2twp.json')
spacy_stopwords_sim = list(STOP_WORDS)
print(spacy_stopwords_sim[:5])
spacy_stopwords_tra = [converter.convert(w) for w in spacy_stopwords_sim]
print(spacy_stopwords_tra[:5])
#['因为', '奇', '嘿嘿', '其次', '偏偏']
#['因為', '奇', '嘿嘿', '其次', '偏偏']

定义一个用于实现 TextRank 的类

如果你正在处理英文文本，你可以很容易地实现 TextRank textaCy，它的标语是NLP, before and after spaCy. 但我无法让它适用于中文文本，所以我不得不从头开始实现 TextRank。幸运的是，我从这个gist中得到了一个快速启动，它为以下定义提供了蓝图。

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
from collections import OrderedDict
import numpy as np

class TextRank4Keyword():
    """Extract keywords from text"""

    def __init__(self):
        self.d = 0.85 # damping coefficient, usually is .85
        self.min_diff = 1e-5 # convergence threshold
        self.steps = 10 # iteration steps
        self.node_weight = None # save keywords and its weight

    def set_stopwords(self, custom_stopwords):  
        """Set stop words"""
        for word in set(spacy_stopwords_tra).union(set(custom_stopwords)):
            lexeme = nlp.vocab[word]
            lexeme.is_stop = True

    def sentence_segment(self, doc, candidate_pos, lower):
        """Store those words only in cadidate_pos"""
        sentences = []
        for sent in doc.sents:
            selected_words = []
            for token in sent:
                # Store words only with cadidate POS tag
                if token.pos_ in candidate_pos and token.is_stop is False:
                    if lower is True:
                        selected_words.append(token.text.lower())
                    else:
                        selected_words.append(token.text)
            sentences.append(selected_words)
        return sentences

    def get_vocab(self, sentences):
        """Get all tokens"""
        vocab = OrderedDict()
        i = 0
        for sentence in sentences:
            for word in sentence:
                if word not in vocab:
                    vocab[word] = i
                    i += 1
        return vocab

    def get_token_pairs(self, window_size, sentences):
        """Build token_pairs from windows in sentences"""
        token_pairs = list()
        for sentence in sentences:
            for i, word in enumerate(sentence):
                for j in range(i+1, i+window_size):
                    if j >= len(sentence):
                        break
                    pair = (word, sentence[j])
                    if pair not in token_pairs:
                        token_pairs.append(pair)
        return token_pairs

    def symmetrize(self, a):
        return a + a.T - np.diag(a.diagonal())

    def get_matrix(self, vocab, token_pairs):
        """Get normalized matrix"""
        # Build matrix
        vocab_size = len(vocab)
        g = np.zeros((vocab_size, vocab_size), dtype='float')
        for word1, word2 in token_pairs:
            i, j = vocab[word1], vocab[word2]
            g[i][j] = 1

        # Get Symmeric matrix
        g = self.symmetrize(g)

        # Normalize matrix by column
        norm = np.sum(g, axis=0)
        g_norm = np.divide(g, norm, where=norm!=0) # this is to ignore the 0 element in norm

        return g_norm

    # I revised this function to return keywords as a list
    def get_keywords(self, number=10):
        """Print top number keywords"""
        node_weight = OrderedDict(sorted(self.node_weight.items(), key=lambda t: t[1], reverse=True))
        keywords = []
        for i, (key, value) in enumerate(node_weight.items()):
            keywords.append(key)
            if i > number:
                break
        return keywords

    def analyze(self, text, 
                candidate_pos=['NOUN', 'VERB'], 
                window_size=5, lower=False, stopwords=list()):
        """Main function to analyze text"""

        # Set stop words
        self.set_stopwords(stopwords)

        # Pare text with spaCy
        doc = nlp(token_str)

        # Filter sentences
        sentences = self.sentence_segment(doc, candidate_pos, lower) # list of list of words

        # Build vocabulary
        vocab = self.get_vocab(sentences)

        # Get token_pairs from windows
        token_pairs = self.get_token_pairs(window_size, sentences)

        # Get normalized matrix
        g = self.get_matrix(vocab, token_pairs)

        # Initionlization for weight(pagerank value)
        pr = np.array([1] * len(vocab))

        # Iteration
        previous_pr = 0
        for epoch in range(self.steps):
            pr = (1-self.d) + self.d * np.dot(g, pr)
            if abs(previous_pr - sum(pr))  < self.min_diff:
                break
            else:
                previous_pr = sum(pr)

        # Get weight for each node
        node_weight = dict()
        for word, index in vocab.items():
            node_weight[word] = pr[index]

        self.node_weight = node_weight

现在我们可以创建一个类的实例并使用我们的变量TextRank4Keyword调用该set_stopwords函数。CUSTOM_STOPWORDS这创建了一组由我们的自定义停用词和 spaCy 的内置停用词联合产生的停用词。只有满足这两个条件的词才能成为关键字的候选者：

  • 它们不在停用词集合中；
  • 它们的 POS 标签是 中列出的标签之一，默认情况下candidate_pos包括NOUN和VERB。
1
2
tr4w = TextRank4Keyword()
tr4w.set_stopwords(CUSTOM_STOPWORDS)

合并代码

让我们通过定义一个用于关键字提取的主函数。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
def extract_keys_from_str(raw_text):
  text = clean_all(raw_text) #clean the raw text
  ws  = ws_driver([text]) #tokenize the text with CKIP Transformers
  tokenized_text = " ".join(ws[0]) #join a list into a string 
  tr4w.analyze(tokenized_text) #create a spaCy Doc object with the string and calculate weights for words
  keys = tr4w.get_keywords(KW_NUM) #get top 10 keywords, as set by the KW_NUM variable
  return keys

keys = extract_keys_from_str(raw_text)
keys = [k for k in keys if len(k) > 1]
keys
#Tokenization: 100%|██████████| 1/1 [00:00<00:00, 221.73it/s]
#Inference: 100%|██████████| 1/1 [00:05<00:00,  5.20s/it]
#['土地', '公園', '地政局', '文化', '推出', '面積', '標售', '道路', '優質', '投標']

凡本网注明"来源：XXX "的文/图/视频等稿件，本网转载出于传递更多信息之目的，并不意味着赞同其观点或证实其内容的真实性。如涉及作品内容、版权和其它问题，请与本网联系，我们将在第一时间删除内容！
作者: Haowen'S Ai Blog
来源： https://howard-haowen.rohan.tw/blog.ai/keyword-extraction/spacy/textacy/ckip-transformers/jieba/textrank/rake/2021/02/16/Adding-a-custom-tokenizer-to-spaCy-and-extracting-keywords.html