NLP基础：文本预处理、词元化与词表构建

文本预处理

读取数据集

import collections
import re
from d2l import torch as d2l

d2l.DATA_HUB['time_machine'] = (d2l.DATA_URL + 'timemachine.txt','090b5e7e70c295757f55df93cb0a180b9691891a')

def read_time_machine():
    with open(d2l.download('time_machine'),'r') as f:
        lines = f.readlines()
    return [re.sub('[^A-Za-z]+',' ',line).strip().lower() for line in lines]
lines = read_time_machine()
print(f'文本总行数:{len(lines)}')
print(lines[0])
print(lines[10])

文本总行数:3221
the time machine by h g wells
twinkled and his usually pale face was flushed and animated the

词元化

def tokenize(lines,token='word'):
    if token == 'word':
        return [line.split() for line in lines]
    elif token == 'char':
        return [list(line) for line in lines]
    else:
        print("错误: 未知的词元类型: " + token)

tokens = tokenize(lines)
for i in range(11):
    print(tokens[i])

['the', 'time', 'machine', 'by', 'h', 'g', 'wells']
[]
[]
[]
[]
['i']
[]
[]
['the', 'time', 'traveller', 'for', 'so', 'it', 'will', 'be', 'convenient', 'to', 'speak', 'of', 'him']
['was', 'expounding', 'a', 'recondite', 'matter', 'to', 'us', 'his', 'grey', 'eyes', 'shone', 'and']
['twinkled', 'and', 'his', 'usually', 'pale', 'face', 'was', 'flushed', 'and', 'animated', 'the']

词表

实际是一个字典，用来将字符串类型的词元映射到从0开始的数字索引中。我们对训练集的所有文档合并对他们的唯一词元进行统计，得到的结果为语料库。根据每个词元出现的频率为其分配索引。

class Vocab:
    def __init__(self, tokens=None, min_freq=0,reserved_tokens=None):
        if tokens is None:
            tokens = []
        if reserved_tokens is  None:
            reserved_tokens = []

        # 按照出现频率排序
        counter = count_corpus(tokens)
        """
        * counter.items() 是字典的一个方法。它返回一个包含字典所有项（(键, 值) 对，即 (key, value) tuples）的视图对象。例如，如果 counter 是 {'the': 10, 'a': 5}，counter.items() 会提供像 ('the', 10) 和 ('a', 5) 这样的元素。
        * sorted() 在对元素进行比较之前，会先将每个元素传递给 key 指定的函数，然后根据这些函数返回的结果进行排序。
        * lambda匿名函数见下
        """
        self._token_freqs = sorted(counter.items(), key=lambda x:x[1], reverse=True)

        # 未知词元的索引为0
        self.idx_to_token = ['<unk>'] + reserved_tokens
        self.token_to_idx = {token: idx for idx, token in enumerate(self.idx_to_token)}

        for token, freq in self._token_freqs:
            if freq < min_freq:
                break
            if token not in self.token_to_idx:
                self.idx_to_token.append(token)
                self.token_to_idx[token] = len(self.idx_to_token)

    # 定义指定方法`__len__`. 这个方法允许在这个类对象上使用内置函数 `len()`。 (e.g., `len(my_vocab)`).
    def __len__(self):
        return len(self.idx_to_token)

     # 定义特殊方法`__getitem__`。这允许使用方括号表示法进行查找（例如，`my_vocab['hello']`或`my_voab[['hello'，'world']]）。
    def __getitem__(self, tokens):
        if not isinstance(tokens, (list, tuple)):
            return self.token_to_idx.get(tokens,self.unk)
        return [self.__getitem__(token) for token in tokens]
	# 进行反向查找index--->token
    def to_tokens(self, indices):
        if not isinstance(indices, (list, tuple)):
            return self.idx_to_token[indices]
        return [self.idx_to_token[index]for index in indices]

      # `@property` 是一个装饰器，允许我们向调用属性的方式一样来调用函数。E.g., `my_vocab.unk` instead of `my_vocab.unk()`.
    @property
    def unk(self):
        return 0
    @property
    def token_freqs(self):
        return self._token_freqs

def count_corpus(tokens):
    """统计词元出现的频率"""
    if len(tokens) == 0 or isinstance(tokens[0], list):
        # 将词元展开成一个列表
        tokens = [token for line in tokens for token in line]
    return collections.Counter(tokens)

lambda 匿名函数 (Anonymous Function): lambda x: x[1] 定义了一个简单的匿名函数（没有名字的函数）。

• lambda: 关键字，表示这是一个 lambda 函数。
• x: 参数名，代表输入给这个 lambda 函数的单个元素。在这个上下文中，因为 sorted 正在处理 counter.items() 的结果，所以 x 会是一个 (键, 值) 对的元组，例如 ('the', 10)。
• : x[1]: 冒号后面是函数的返回值。x[1] 表示取元组 x 中索引为 1 的元素，也就是那个键值对中的值 (value)——即词元的频率。
• key=lambda x: x[1] 的整体作用: 告诉 sorted 函数，在排序 (词元, 频率) 对时，应该根据频率（元组的第二个元素 x[1]）来排序，而不是根据词元本身（元组的第一个元素 x[0]）。

vocab = Vocab(tokens)
print(list(vocab.token_to_idx.items())[:10])

[('<unk>', 0), ('the', 2), ('i', 3), ('and', 4), ('of', 5), ('a', 6), ('to', 7), ('was', 8), ('in', 9), ('that', 10)]

for i in [0,10]:
    print('文本: ', tokens[i])
    print('索引: ', vocab[tokens[i]])

文本:  ['the', 'time', 'machine', 'by', 'h', 'g', 'wells']
索引:  [2, 20, 51, 41, 2184, 2185, 401]
文本:  ['twinkled', 'and', 'his', 'usually', 'pale', 'face', 'was', 'flushed', 'and', 'animated', 'the']
索引:  [2187, 4, 26, 1045, 363, 114, 8, 1422, 4, 1046, 2]

def load_corpus_time_machine(max_tokens=-1):
    """返回时光机器数据集的词元索引列表和词表"""
    lines = read_time_machine()
    tokens = tokenize(lines,'char')
    vocab = Vocab(tokens)
    # 因为时光机器数据集中的每个文本行不一定是一个句子或段落，所以将所有的文本行展平到一个列表中
    corpus = [vocab[token] for line in tokens for token in lines]
    if max_tokens > 0:
        corpus = corpus[:max_tokens]
    return corpus,vocab

corpus,vocab = load_corpus_time_machine()
len(corpus),len(vocab)

(10374841, 28)