pytorch实现textCNN的具体操作

# 获取文件内容
def getData(file):
    f = open(file,'r')
    raw_data = f.readlines()
    return raw_data
# 转换文件格式
def d2csv(raw_data,label_map,name):
    texts = []
    labels = []
    i = 0
    for line in raw_data:
        d = eval(line) #将每行字符串转换为字典
        if len(d['type']) <= 1 or len(d['text']) <= 1: #筛掉无效数据
            continue
        y = label_map[d['type']] #根据label_map将label转换为数字表示
        x = d['text']
        texts.append(x)
        labels.append(y)
        i+=1
        if i%1000 == 0:
            print(i)
    df = pd.DataFrame({'text':texts,'label':labels})
    df.to_csv('data/'+name+'.csv',index=False,sep='\t') # 保存文件
label_map = {'执行':0,'刑事':1,'民事':2,'行政':3}
train_data = getData('data/train.txt') #22000+行
d2csv(train_data,label_map,'train')
valid_data = getData('data/valid.txt') # 2000+行
d2csv(valid_data,label_map,'valid')
test_data = getData('data/test.txt') # 2000+行
d2csv(test_data,label_map,'test')

train_text = []
for line in train_data:
    d = eval(line)
    t = jieba.cut(d['text'])
    train_text.append(t)
sentence_length = [len(x) for x in train_text] #train_text是train.csv中每一行分词之后的数据
%matplotlib notebook
import matplotlib.pyplot as plt
plt.hist(sentence_length,1000,nORMed=1,cumulative=True)
plt.xlim(0,1000)
plt.show()

def tokenizer(x):
    res = [w for w in jieba.cut(x)]
    return res

stop_words = []
print('build stop words set')
with open('data/stopwords.dat') as f:
    for l in f.readlines():
        stop_words.append(l.strip())

TEXT = data.Field(sequential=True, tokenize=tokenizer,fix_length=1000,stop_words=stop_words)
LABEL = data.Field(sequential=False,use_vocab=False)

train,valid,test = data.TabularDataset.splits(path='data',train='train.csv',
                                              validation='valid.csv',test='test.csv',
                                              format='csv',
                                              skip_header=True,csv_reader_params={'delimiter':'\t'},
                                              fields=[('text',TEXT),('label',LABEL)])

TEXT.build_vocab(train)

train_iter, val_iter, test_iter = data.Iterator.splits((train,valid,test),
                                                             batch_sizes=(args.batch_size,args.batch_size,args.batch_size),
                                                             device=args.device,
                                                             sort_key=lambda x:len(x.text),
                                                             sort_within_batch=False,
                                                             repeat=False)

import torch
import torch.nn as nn
import torch.nn.functional as F
class textCNN(nn.Module):
    def __init__(self, args):
        super(textCNN, self).__init__()
        self.args = args
        
        Vocab = args.embed_num ## 已知词的数量
        Dim = args.embed_dim ##每个词向量长度
        Cla = args.class_num ##类别数
        Ci = 1 ##输入的channel数
        Knum = args.kernel_num ## 每种卷积核的数量
        Ks = args.kernel_sizes ## 卷积核list，形如[2,3,4]
        
        self.embed = nn.Embedding(Vocab,Dim) ## 词向量，这里直接随机
        
        self.convs = nn.ModuleList([nn.Conv2d(Ci,Knum,(K,Dim)) for K in Ks]) ## 卷积层
        self.dropout = nn.Dropout(args.dropout) 
        self.fc = nn.Linear(len(Ks)*Knum,Cla) ##全连接层
        
    def forward(self,x):
        x = self.embed(x) #(N,W,D)
        
        x = x.unsqueeze(1) #(N,Ci,W,D)
        x = [F.relu(conv(x)).squeeze(3) for conv in self.convs] # len(Ks)*(N,Knum,W)
        x = [F.max_pool1d(line,line.size(2)).squeeze(2) for line in x]  # len(Ks)*(N,Knum)
        
        x = torch.cat(x,1) #(N,Knum*len(Ks))
        
        x = self.dropout(x)
        logit = self.fc(x)
        return logit

import os
import sys
import torch
import torch.autograd as autograd
import torch.nn.functional as F
def train(train_iter, dev_iter, model, args):
    if args.cuda:
        model.cuda(args.device)    
    optimizer = torch.optim.Adam(model.parameters(), lr=args.lr)
    
    steps = 0
    best_acc = 0
    last_step = 0
    model.train()
    print('training...')
    for epoch in range(1, args.epochs+1):
        for batch in train_iter:
            feature, target = batch.text, batch.label #(W,N) (N)
            feature.data.t_()
            
            if args.cuda:
                feature, target = feature.cuda(), target.cuda()
            
            optimizer.zero_grad()
            logit = model(feature)
            loss = F.cross_entropy(logit, target)
            loss.backward()
            optimizer.step()
            
            steps += 1
            if steps % args.log_interval == 0:
                result = torch.max(logit,1)[1].view(target.size())
                corrects = (result.data == target.data).sum()
                accuracy = corrects*100.0/batch.batch_size
                sys.stdout.write('\rBatch[{}] - loss: {:.6f} acc: {:.4f}$({}/{})'.format(steps,
                                                                                        loss.data.item(),
                                                                                        accuracy,
                                                                                        corrects,
                                                                                        batch.batch_size))
            if steps % args.dev_interval == 0:
                dev_acc = eval(dev_iter, model, args)
                if dev_acc > best_acc:
                    best_acc = dev_acc
                    last_step = steps
                    if args.save_best:
                        save(model,args.save_dir,'best',steps)
                else:
                    if steps - last_step >= args.early_stop:
                        print('early stop by {} steps.'.format(args.early_stop))
            elif steps % args.save_interval == 0:
                save(model,args.save_dir,'snapshot',steps)

def save(model, save_dir, save_prefix, steps):
    if not os.path.isdir(save_dir):
        os.makedirs(save_dir)
    save_prefix = os.path.join(save_dir,save_prefix)
    save_path = '{}_steps_{}.pt'.format(save_prefix,steps)
    torch.save(model.state_dict(),save_path)

def eval(data_iter, model, args):
    model.eval()
    corrects, avg_loss = 0,0
    for batch in data_iter:
        feature, target = batch.text, batch.label
        feature.data.t_()
        
        if args.cuda:
            feature, target = feature.cuda(), target.cuda()
        
        logit = model(feature)
        loss = F.cross_entropy(logit,target)
        
        avg_loss += loss.data[0]
        result = torch.max(logit,1)[1]
        corrects += (result.view(target.size()).data == target.data).sum()
    
    size = len(data_iter.dataset)
    avg_loss /= size 
    accuracy = 100.0 * corrects/size
    print('\nEvaluation - loss: {:.6f} acc: {:.4f}%({}/{}) \n'.format(avg_loss,accuracy,corrects,size))
    return accuracy