基于浅层前馈神经网络和词窗口的命名实体识别实践

Introduction

命名实体识别作为自然语言处理领域中一项基础而关键的技术，被广泛应用于自然语言处理各个应用领域中，也是关系抽取、事件抽取、知识图谱、机器翻译、问答系统等诸多 NLP 任务的基础，其目标是对待识别文本中代表知识主体的命名实体（named entity, NE）进行标注。

命名实体一般指的是文本中具有特定意义或者指代性强的实体，通常包括人名、地名、组织机构名、日期时间、专有名词等。而在知识图谱等研究的驱动下，NER 识别的实体范围进一步扩大到各个专业知识领域。

目前通用领域的命名实体识别方法有两种，分别是浅层机器学习的方法和深层神经网络的方法。

NER 研究的发展趋势如下图：

本文主要使用词窗口和词嵌入技术对文本数据输入样本进行数据化处理，主要是降维工作。

很显然命名实体识别本质是一个分类任务，因此本文模型搭建了一个简单的前馈神经网络来完成分类预测的任务。

最后经过验证，该模型具有一定的有效性，证明了这个模型基本上是可以成立的。

相关工作

词窗口

这里的词窗口和 n-gram 模型有一定的类似之处，本文处理的语料是中文语料，取一个合适的窗口大小，对输入语料进行对应的前后扩展，使其按照该窗口分割出的窗口词数目刚好等于原输入语料词数目，从而使每个窗口词对应一个实体标签，形成词窗口对。

这里处理的主要目的使用局部窗口的概念，来限定词义的预测范围，使词与其近邻的上下文产生一定的关联。

具体如下图:

上面显示了10组词窗口对，词窗口大小为5，因此每个样本有五个词，每个样本对应一个标签。

标签含义如下：

人名 nr 
地名 ns
机构名称 nt
其他 o

本文处理的标签直接按照原始标签处理，没有转化为 BIO 标准的实体标注标签，由于标签体系过于简单，对于最后的准确率会有一定的影响。

词嵌入

NLP 与图像处理相比，其难点之一就在于自然语言不是天然数字量化的，图像可以使用各种数字标签直接转换，比如常见的 RGB、HSV等，这些数字量化的数据可以直接交给计算机处理。

而文字则不然，计算机并不认识文字，也无法将文字直接转换为合适的数学表示，因此，词嵌入技术便出现了。词嵌入就是描述的就是如何将主观的词用数字向量表示。

传统的方式有 one-hot ，但其问题在于既不能很好表示词义又存在维度灾难，相比起来词嵌入（word-embedding）是一种降维技术，他用相对紧凑较少的维度来表示词不同维度的特征。

本文使用的词嵌入使用的是 PyTorch 自带的 torch.nn.Embedding(m, n)函数实现。在本文中的词嵌入更像是一个随着神经网络训练产生的副产物，因为是随机初始化词嵌入向量的，不是预训练的词向量。

前馈神经网络

这个在之前的文章中已经有所介绍，此处就不在赘述了，本文所搭建的神经网络是一个非常简单的神经网络。

由于 NER 涉及的分类问题显然不是一个简单线性可分问题，因此直接使用线性回归模型是不可能解决的，

而神经网络的强大之处在于其能拟合任何函数，其关键之处在于加入非线性的因素，即是：激活函数。而这个线性嵌套非线性的过程叫激活，构成了隐藏层的神经元，这样的隐藏层越多，非线性越多，网络就越深，本文就用了两层。

然后输出层根据任务一般使用softmax多分类函数。

模型定义

总体网络架构

总体模型示意图如下：

可以看出一共有两层隐藏层，一个输入层和输出层。

模型定义代码如下：

class WindowClassifier(nn.Module): 
    def __init__(self, vocab_size, embedding_size, window_size, hidden_size, output_size):

        super(WindowClassifier, self).__init__()
        
        self.embed = nn.Embedding(vocab_size, embedding_size)
        self.h_layer1 = nn.Linear(embedding_size * (window_size * 2 + 1), hidden_size)
        self.h_layer2 = nn.Linear(hidden_size, hidden_size)
        self.o_layer = nn.Linear(hidden_size, output_size)
        self.relu = nn.ReLU()
        self.softmax = nn.LogSoftmax(dim=1)
        self.dropout = nn.Dropout(0.35)
        
    def forward(self, inputs, is_training=False): 
        embeds = self.embed(inputs) # BxWxD  torch.Size([128, 5, 100])
        concated = embeds.view(-1, embeds.size(1)*embeds.size(2)) # Bx(W*D)
        h0 = self.relu(self.h_layer1(concated))
        if is_training:
            h0 = self.dropout(h0)
        h1 = self.relu(self.h_layer2(h0))
        if is_training:
            h1 = self.dropout(h1)
        out = self.softmax(self.o_layer(h1))
        return out

激活函数使用的是 ReLU，最终是个多分类任务，所以说 LogSoftmax 进行归一化处理。

数据预处理

原始语料格式如下所示：

这里需要将 word 和 tag 分别取出，形成序列，目标格式如下:

从而方便生成处理窗口词对。

实验代码

导入环境

import torch
import torch.nn as nn
from torch.autograd import Variable
import torch.optim as optim
import torch.nn.functional as F
import nltk
import random
import numpy as np
from collections import Counter
from sklearn_crfsuite import metrics
import pickle

预定义

random.seed(1024)
flatten = lambda l: [item for sublist in l for item in sublist]

USE_CUDA = torch.cuda.is_available()

FloatTensor = torch.cuda.FloatTensor if USE_CUDA else torch.FloatTensor
LongTensor = torch.cuda.LongTensor if USE_CUDA else torch.LongTensor
ByteTensor = torch.cuda.ByteTensor if USE_CUDA else torch.ByteTensor

数据预处理

def preprocess_corpus(file_path, is_train = True):
    data = []
    if is_train :
        output_data = open('./original/train_process.txt', 'w', encoding='utf-8')
        output_pkl = open('train_data.pkl', 'wb')
    else :
        output_data = open('./original/test_process.txt', 'w', encoding='utf-8')
        output_pkl = open('test_data.pkl', 'wb')
    
    file = open(file_path, 'r', encoding='utf8')
    lines = file.readlines()
    for line in lines:
        # 移除字符串的头和尾的空格并分割
        word_list = line.strip().split()
        cor_line = list(words.split("/") for words in word_list)
        sent, tag = list(zip(*cor_line))
        data.append([sent, tag])
        # 持久化 Json 存储
        output_data.write("{\"words\":\"%s\", \"tags\":\"%s\"}" % (" ".join(sent), " ".join(tag)))
        output_data.write("\n")
    # Pickle dictionary using protocol 0.
    pickle.dump(data, output_pkl)
    output_data.close()  
    return data
    
train_data = preprocess_corpus("./original/train1.txt")

test_data =  preprocess_corpus("./original/testright1.txt", is_train = False)

输入样本生成

1. 词表、标签表及其映射

sents, tags = list(zip(*train_data)) #sent实际上就是句子序列组成的list，tags 是所有的标签 list
# 用集合去重
vocab = list(set(flatten(sents)))
tagset = list(set(flatten(tags)))

# 映射
# 词映射，<UNK> 为未登录词，<DUMMY> 为扩展填充词
word2index={'<UNK>' : 0, '<DUMMY>' : 1}for vo in vocab:
    if word2index.get(vo) is None:
        word2index[vo] = len(word2index)
index2word = {v:k for k, v in word2index.items()}

tag2index = {}
for tag in tagset:
    if tag2index.get(tag) is None:
        tag2index[tag] = len(tag2index)
index2tag={v:k for k, v in tag2index.items()}

2. 生成窗口词对

这一步的生成结果如文章开头的示意图

WINDOW_SIZE = 2
windows_train = []
for sample in train_data:  #每一个sample就是完整的一句话以及对应的单词tag
    dummy = ['<DUMMY>'] * WINDOW_SIZE
    window = list(nltk.ngrams(dummy + list(sample[0]) + dummy, WINDOW_SIZE * 2 + 1))
    windows_train.extend([[list(window[i]), sample[1][i]] for i in range(len(sample[0]))])

windows_test = []
for sample in test_data:  #每一个sample就是完整的一句话以及对应的单词tag
    dummy = ['<DUMMY>'] * WINDOW_SIZE
    window = list(nltk.ngrams(dummy + list(sample[0]) + dummy, WINDOW_SIZE * 2 + 1))
    windows_test.extend([[list(window[i]), sample[1][i]] for i in range(len(sample[0]))])

3. 划分训练集、测试集

#这里一个windows的含义就是一句话的窗口上下文标注对
random.shuffle(windows_train)
random.shuffle(windows_test)

train_data = windows_train# windows[:int(len(windows) * 0.9)]
test_data = windows_test# windows[int(len(windows) * 0.9):]

4. 样本集 batch 生成函数

def getBatch(batch_size, train_data):
    random.shuffle(train_data)
    sindex = 0
    eindex = batch_size
    while eindex < len(train_data):
        batch = train_data[sindex: eindex]
        temp = eindex
        eindex = eindex + batch_size
        sindex = temp
        yield batch
    
    if eindex >= len(train_data):
        batch = train_data[sindex:]
        yield batch

训练

1. 词序列、标签映射数字转换

def prepare_sequence(seq, word2index):
    idxs = list(map(lambda w: word2index[w] if word2index.get(w) is not None else word2index["<UNK>"], seq))
    return Variable(LongTensor(idxs))

def prepare_tag(tag,tag2index):
    return Variable(LongTensor([tag2index[tag]]))

2. 模型超参数设置

BATCH_SIZE = 128
EMBEDDING_SIZE = 100 # x (WINDOW_SIZE*2+1) = 500
HIDDEN_SIZE = 300
EPOCH = 3
LEARNING_RATE = 0.001

3. 训练

model = WindowClassifier(len(word2index), EMBEDDING_SIZE, WINDOW_SIZE, HIDDEN_SIZE, len(tag2index))
if USE_CUDA:
    model = model.cuda()
loss_function = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=LEARNING_RATE)

for epoch in range(EPOCH):
    losses = []
    for i,batch in enumerate(getBatch(BATCH_SIZE, train_data)):
        x,y=list(zip(*batch))
        inputs = torch.cat([prepare_sequence(sent, word2index).view(1, -1) for sent in x]) # view是变成二维 这里是把batch绑起来一个矩阵
        targets = torch.cat([prepare_tag(tag, tag2index) for tag in y])
        model.zero_grad()
        preds = model(inputs, is_training=True)
        loss = loss_function(preds, targets)
        losses.append(loss.item())
        loss.backward()
        optimizer.step()

        if i % 1000 == 0:
            print("[%d/%d] mean_loss : %0.2f" %(epoch, EPOCH, np.mean(losses)))
            losses = []
# 保存模型            
torch.save(model.state_dict(), "./simplennmodel.pkl")

测试及结果

for_f1_score = []

accuracy = 0
for test in test_data:
    x, y = test[0], test[1]
    input_ = prepare_sequence(x, word2index).view(1, -1)

    i = model(input_).max(1)[1]
    pred = index2tag[i.data.tolist()[0]]
    for_f1_score.append([pred, y])
    if pred == y:
        accuracy += 1

print(accuracy/len(test_data) * 100)

y_pred, y_test = list(zip(*for_f1_score))

sorted_labels = sorted(list(set(y_test) - {'o'}), key=lambda name: (name[1:], name[0]))

# 混淆矩阵
y_pred = [[y] for y in y_pred] # this is because sklearn_crfsuite.metrics function flatten inputs
y_test = [[y] for y in y_test]

print(metrics.flat_classification_report(y_test, y_pred, labels = sorted_labels, digits=3))

输出结果如下:

Next

这里的处理使用的是最直接的标签体系，神经网络模型的架构也比较简单，下一步计划将现有的预料数据标签体系转换为 BIO 标签，并搭建更为复杂的网络模型来尝试这一任务。