【AI达人创造营第二期】一文读懂textCNN模型原理

import pandas as pdimport numpy as npimport paddleimport paddle.nn as nnfrom paddle.io import DataLoader, Datasetimport paddle.optimizer as optimfrom paddlenlp.data import Padimport jiebafrom collections import Counter

data = pd.read_csv("data/data45260/Train.txt", sep='\t', names=['ClassNo', 'ClassName', 'Sentence'])print("Train Set")print(data.head())print(data.info())
pred_data = pd.read_csv("data/data45260/Test.txt", sep='\t', names=['Sentence'])print("Pred Set")print(pred_data.head())print(pred_data.info())

## 注意到训练集中语句与数据数目不匹配，需删除无效数据data.dropna(axis=0, how='any', inplace=True)
X = data['Sentence'].values
Y = data['ClassNo'].values
Y_name = data['ClassName'][np.sort(np.unique(data['ClassName'], return_index=True)[1])].values
Y_dict = dict([val for val in zip(np.unique(Y), Y_name)])

def preprocess(data):
    stopwords = open("data/data81223/stopwords.txt").read().split('\n')
    new_data = []    for row in data:
        row = jieba.lcut(row)  # 中文分词
        new_row = []        for word in row:            if word not in stopwords:
                new_row.append(word)
        new_data.append(new_row)    return np.array(new_data)

corpus = preprocess(X)print(corpus[42])

## 构建词表def build_vocab(data):
    word2id = {}
    vocab = Counter()    for row in data:
        vocab.update(row)
    vocab = sorted(vocab.items(), key=lambda x: x[1], reverse=True)
    vocab = [('<PAD>', 0)] + list(vocab) + [('<UNK>', 0)]
    word2id = {word[0]: i for i, word in enumerate(vocab)}    return word2iddef get_ids(data, word2id):
    ids =  []    for row in data:
        id_ = list(map(lambda x: word2id.get(x, word2id['<UNK>']), row))
        ids.append(id_)    return np.array(ids)def padding(data):
    return Pad(pad_val=0)(data)

word2id = build_vocab(corpus)
ids = get_ids(corpus, word2id)
padding_ids = padding(ids)print(padding_ids[42])print(padding_ids[56748])

## 重写Datasetclass MyDataset(Dataset):
    def __init__(self, X, Y):
        super(MyDataset, self).__init__()
        self.X = X
        self.Y = Y    def __len__(self):
        return self.X.shape[0]    def __getitem__(self, index):
        return self.X[index], self.Y[index]

## 划分数据集————训练集、测试集、开发集def split(X, Y):
    idx = [i for i in range(0, X.shape[0])]
    np.random.shuffle(idx)
    train_len = int(0.9 * len(idx))    return X[idx[:train_len]], Y[idx[:train_len]], \
           X[idx[train_len:]], Y[idx[train_len:]]

train_X, train_Y, test_X, test_Y = split(padding_ids, Y)print("Train Set: ", train_X.shape, train_Y.shape)print("Test Set: ", test_X.shape, test_Y.shape)

## 搭建模型class TextCNN(paddle.nn.Layer):
    def __init__(self, vocab_size, embedding_size, classes, pretrained=None, kernel_num=100, kernel_size=[3, 4, 5], dropout=0.5):
        super(TextCNN, self).__init__()
        self.vocab_size = vocab_size
        self.embedding_size = embedding_size
        self.classes = classes
        self.pretrained = pretrained
        self.kernel_num = kernel_num
        self.kernel_size = kernel_size
        self.dropout = dropout        if self.pretrained != None:
            self.embedding = nn.Embedding(num_embeddings=self.vocab_size, embedding_dim=self.embedding_size, padding_idx=0 ,_weight=pretrained)        else:
            self.embedding = nn.Embedding(num_embeddings=self.vocab_size, embedding_dim=self.embedding_size, padding_idx=0)
        self.convs = nn.LayerList([nn.Conv2D(1, self.kernel_num, (kernel_size_, embedding_size)) for kernel_size_ in self.kernel_size])
        self.dropout = nn.Dropout(self.dropout)
        self.linear = nn.Linear(3 * self.kernel_num, self.classes)    def forward(self, x):
        embedding = self.embedding(x).unsqueeze(1)
        convs = [nn.ReLU()(conv(embedding)).squeeze(3) for conv in self.convs]
        pool_out = [nn.MaxPool1D(block.shape[2])(block).squeeze(2) for block in convs]
        pool_out = paddle.concat(pool_out, 1)
        logits = self.linear(pool_out)        return logits

## 训练配置BATCH_SIZE = 50EMBEDDING_SIZE = 150LEARNING_RATE = 0.00005EPOCHS = 5 # 5device = paddle.device.get_device()  #注意配置环境必须是GPU，否则以下训练程序将无法运行print(device)

## 模型训练与评估Train_Loader = DataLoader(MyDataset(paddle.to_tensor(train_X), paddle.to_tensor(train_Y)), batch_size=BATCH_SIZE, shuffle=True)
Test_Loader = DataLoader(MyDataset(paddle.to_tensor(test_X), paddle.to_tensor(test_Y)), batch_size=BATCH_SIZE, shuffle=True)

model = TextCNN(vocab_size=len(word2id), embedding_size=EMBEDDING_SIZE, classes=len(Y_dict))print(model)
optimizer = optim.Adam(parameters=model.parameters(), learning_rate=LEARNING_RATE)
criterion = nn.CrossEntropyLoss()for epoch in range(0, EPOCHS):
    Train_Loss, Test_Loss = [], []
    Train_Acc, Test_Acc = [], []
    model.train()    for i, (x, y) in enumerate(Train_Loader):
        x = x.cuda()
        y = y.cuda()
        pred = model(x)
        loss = criterion(pred, y)
        Train_Loss.append(loss.item())
        Train_Acc.append(paddle.metric.accuracy(pred, y).numpy())
        loss.backward()
        optimizer.step()
        optimizer.clear_grad()
    model.eval()    for i, (x, y) in enumerate(Test_Loader):
        x = x.cuda()
        y = y.cuda()
        pred = model(x)
        Test_Loss.append(criterion(pred, y).item())
        Test_Acc.append(paddle.metric.accuracy(pred, y).numpy())    print(        "Epoch: [{}/{}] TrainLoss/TestLoss: {:.4f}/{:.4f} TrainAcc/TestAcc: {:.4f}/{:.4f}".format( \
        epoch + 1, EPOCHS, \
        np.mean(Train_Loss), np.mean(Test_Loss), \
        np.mean(Train_Acc), np.mean(Test_Acc) \
        ) \
    )

## 保存模型paddle.save(model.state_dict(), "TextCNN.pdparams")
paddle.save(optimizer.state_dict(), "Adam.pdparams")

## 预测数据准备pred_X = pred_data['Sentence'].values
pred_corpus = preprocess(pred_X)
pred_ids = get_ids(pred_corpus, word2id)
pred_padding_ids = padding(pred_ids)print(pred_padding_ids.shape)## 预测过程，并将预测结果按照输入的训练集形式进行了保存Pred_Loader = DataLoader(MyDataset(paddle.to_tensor(pred_padding_ids), paddle.to_tensor(pred_padding_ids)), batch_size=BATCH_SIZE, shuffle=False)
model.eval()
pred_Y = []for i, (x, y) in enumerate(Pred_Loader):
    x = x.cuda()
    pred = model(x)
    label = np.argmax(pred, axis=1)
    pred_Y += list(label)## 可视化一些文本的预测结果pred_Name = [Y_dict[name] for name in pred_Y]
MyPred = pd.DataFrame({'ClassNo': pred_Y, 'ClassName': pred_Name, 'Sentence': pred_X})print(MyPred.head())
MyPred.to_csv('pred.csv', index=False)

model_params = paddle.load('TextCNN.pdparams')
model_infer = TextCNN(vocab_size=len(word2id), embedding_size=EMBEDDING_SIZE, classes=len(Y_dict))
model_infer.set_state_dict(model_params)

sentence = input()
infer_corpus = preprocess([sentence])
infer_ids = get_ids(infer_corpus, word2id)
infer_padding_ids = padding(infer_ids)
model_infer.eval()
val = model_infer(paddle.to_tensor(infer_padding_ids))
label = np.argmax(val, axis=1)[0]print("Sentence: ", sentence, " label: ", Y_dict[label])