1、seq2seq_attention_机器翻译
seq2seq_attention是一种基于神经网络的 机器翻译 模型，它通过 编码器 和 解码器 两个部分实现翻译功能。编码器将源语言句子转换为一个固定长度的向量表示，解码器则将这个向量作为输入，生成目标语言句子的翻译结果。
在seq2seq_attention中，编码器和解码器都是由 循环神经网络（RNN）组成的。 编码器 将源语言句子中的每个单词依次输入RNN，每个时刻RNN的输出都会被传递到下一个时刻，直到最后一个时刻，最终得到源语言句子的向量表示。 解码器 的工作方式类似，但它不仅要考虑源语言句子的信息，还要根据当前生成的目标语言单词来不断调整生成下一个单词的概率分布。这就需要在解码器中引入注意力机制（attention mechanism），用来关注源语言句子中与当前要翻译的目标语言单词相关的部分，以便更准确地生成翻译结果。
具体来说，解码器会将每个时刻的输出向量与编码器中所有时刻的输出向量进行加权平均，以得到一个新的上下文向量。这个加权平均的权重是由注意力模型计算得出的，它会考虑源语言句子中每个单词与当前目标语言单词的相关性。最终，这个上下文向量会与当前时刻的解码器输入向量一起输入到解码器的RNN中，以生成下一个目标语言单词的概率分布。这个过程会不断迭代，直到生成了完整的目标语言句子。
seq2seq_attention 相较于传统的seq2seq模型，能够更好地处理长句子和复杂的语法结构，从而提高翻译质量。
2、数据预处理
实现中文到英文的机器翻译
数据下载地址：
www.manythings.org/anki and tatoeba.org
3、加载数据集
datasets.py : 构建中文和英文样本对
sentence1 = normalizeString(l[0]) # 英文，英文文本处理(大写转小写，过滤非法字符等) sentence2 = cht_to_chs(l[1]) # 中文，繁体转简体 因为原始数据中有一些繁体字和中文大写问题，需要转换
import jieba
from utils import normalizeString
from utils import cht_to_chs
SOS_token = 0 # 起始符
EOS_token = 1 # 终止符
MAX_LENGTH = 10 # 将长度过长的句子去掉
class Lang:
def __init__(self, name):
self.name = name
self.word2index = {} # 记录词对应的索引
self.word2count = {} # 记录每个词的词频
self.index2word = {
0: "SOS", 1: "EOS"
} # 记录索引到词
self.n_words = 2 # 记录语料库中有多少种词，初始值为2(起始符+终止符)
# 对词进行统计
def addWord(self, word):
if word not in self.word2index: # 如果词不在统计表中，添加进统计表
self.word2index[word] = self.n_words # 词的索引为该词是第几种的词
self.word2count[word] = 1
self.index2word[self.n_words] = word
self.n_words += 1 # 字典中的词数量+1
else: # 该词在统计表中
self.word2count[word] += 1
# 对句子进行分词
def addSentence(self, sentence):
for word in sentence.split(" "): # 将 "你 吃饭 了 吗 ？" 分割为 ["你"，“吃”,“吃饭”," 了"," 吗"] 的list数组
self.addWord(word) # 依次将每个词统计
# 文本解析
def readLangs(lang1, lang2, path):
lines = open(path, encoding='utf-8').readlines() # 拿到文本的所有行
lang1_cls = Lang(lang1)
lang2_cls = Lang(lang2)
pairs = [] # 记录样本对
for l in lines: # 逐行处理
l = l.split("\t") # 以Tab分割
sentence1 = normalizeString(l[0]) # 英文，英文文本处理(大写转小写，过滤非法字符等)
sentence2 = cht_to_chs(l[1]) # 中文，繁体转简体
seg_list = jieba.cut(sentence2, cut_all=False) # 调用结巴分词对中文进行分割，得到分词后的数组
sentence2 = " ".join(seg_list) #将中文句子分词后的数组拼接为字符串。join() 方法用于把数组中的所有元素放入一个字符串。元素是通过指定的分隔符进行分隔的。
# 英文文本是天然分词的，不需要分词 # 向英文一样，通过空格拼接中文分词结果
if len(sentence1.split(" ")) > MAX_LENGTH: # 过滤一些长句,大于10个词的的不统计
continue # 忽略当前的一次循环
if len(sentence2.split(" ")) > MAX_LENGTH:
continue
pairs.append([sentence1, sentence2]) # [[“what are you doing?”,"你 在 干 什么"],....]
lang1_cls.addSentence(sentence1) # 统计每种语言的词频
lang2_cls.addSentence(sentence2)
return lang1_cls, lang2_cls, pairs
# 测试
lang1 = "en"
lang2 = "cn"
path = "../data/cmn.txt"
lang1_cls, lang2_cls, pairs = readLangs(lang1, lang2, path)
print(len(pairs))
print(lang1_cls.n_words)
print(lang1_cls.index2word)
print(lang2_cls.n_words)
print(lang2_cls.index2word)
代码是一个语言模型读取数据的预处理部分，目的是将源语言和目标语言的文本进行读取、分词和处理，以便在机器翻译模型中使用。
具体功能： 定义了一个Lang类，该类主要是记录每个词的出现频率和对应的索引，同时提供了一个方法用于将句子进行分词并更新词表。
readLangs函数用于读取源语言和目标语言的文本，将其分别进行预处理，并返回三个值：源语言Lang类、目标语言Lang类和处理好的样本对列表。在readLangs函数中，逐行读取源文本中的内容，对每个样本进行处理：用"\t"分割源语言和目标语言；对源语言进行简单的处理，去除空格和标点符号等；对目标语言进行繁体转简体和结巴分词处理；将处理好的样本加入到样本对列表中，并更新源语言和目标语言的词表。最后输出了样本对数量、源语言和目标语言的词汇量和词表，以及词表中的词汇和对应的索引。
4、搭建模型结构
import torch
import torch.nn as nn
import torch.nn.functional as F
from datasets import MAX_LENGTH
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 编码器
class EncoderRNN(nn.Module):
def __init__(self, input_size, hidden_size):
super(EncoderRNN, self).__init__() # 完成类的初始化
self.hidden_size = hidden_size
self.embedding = nn.Embedding(input_size, hidden_size) # 词嵌入层, 第一个参数：字典大小，第二个参数：有多少维的向量表征单词。
self.gru = nn.GRU(input_size=hidden_size, hidden_size=hidden_size) # gru层。也可以选择lstm层或者其他网络作为编码结果
def forward(self, input, hidden):
embedded = self.embedding(input).view(1, 1, -1) # 转化为3维的，因为gru的输入要求是3维的
output = embedded
output, hidden = self.gru(output, hidden)
return output, hidden #返回gru输出的结果和隐藏层信息
# 初始化隐藏状态h0
def initHidden(self):
return torch.zeros(1, 1, self.hidden_size, device=device) # gru的输入为3维的
# 实现两种解码RNN(不带attention + 带attention)
# 不带attention的解码器
class DecoderRNN(nn.Module):
def __init__(self, hidden_size, output_size):
super(DecoderRNN, self).__init__()
self.embedding = nn.Embedding(output_size, hidden_size)
self.gru = nn.GRU(input_size=hidden_size, hidden_size=hidden_size)
self.out = nn.Linear(hidden_size, output_size)
self.softmax = nn.LogSoftmax(dim=1)
def forward(self, input, hidden):
output = self.embedding(input).view(1, 1, -1)
output = F.relu(output)
output, hidden = self.gru(output, hidden)
output = self.softmax(self.out(output[0]))
return output, hidden
# 初始化隐藏状态h0
def initHidden(self):
return torch.zeros(1, 1, self.hidden_size, device=device)
# 带attention
class AttenDecoderRNN(nn.Module):
def __init__(self, hidden_size, output_size, dropout_p=0.1, max_len=MAX_LENGTH):
super(AttenDecoderRNN, self).__init__()
self.hidden_size = hidden_size
self.output_size = output_size
self.dropout_p = dropout_p
self.max_len = max_len
self.embedding = nn.Embedding(self.output_size, self.hidden_size)
self.attn = nn.Linear(self.hidden_size * 2, self.max_len) #要对两个结果进行连接，因此要乘以2
self.attn_combine = nn.Linear(self.hidden_size * 2, self.hidden_size)
self.dropout = nn.Dropout(self.dropout_p)
self.gru = nn.GRU(input_size=self.hidden_size, hidden_size=self.hidden_size)
self.out = nn.Linear(self.hidden_size, self.output_size)
def forward(self, input, hidden, encoder_outputs):
embedded = self.embedding(input).view(1, 1, -1) # 一个
embedded = self.dropout(embedded)
atten_weight = F.softmax(
self.attn(torch.cat([embedded[0], hidden[0]], 1)), # 将embedded和hidden进行拼接，来学习attention权重
dim=1
)
att_applied = torch.bmm(
atten_weight.unsqueeze(0),
encoder_outputs.unsqueeze(0)
)
output = torch.cat([embedded[0], att_applied[0]], dim=1)
output = self.attn_combine(output).unsqueeze(0)
output = F.relu(output)
output, hidden = self.gru(output, hidden)
output = F.log_softmax(self.out(output[0]), dim=1)
return output, hidden, atten_weight
# 初始化隐藏状态h0
def initHidden(self):
return torch.zeros(1, 1, self.hidden_size, device=device)
if __name__ == '__main__':
encoder_net = EncoderRNN(5000, 256)
decoder_net = DecoderRNN(256, 5000)
atten_decoder_net = AttenDecoderRNN(256, 5000)
tensor_in = torch.tensor([12, 14, 16, 18], dtype=torch.long).view(-1, 1) # 定义输入并调整shape
hidden_in = torch.zeros(1, 1, 256)
# 测试编码网络
encoder_out, encoder_hidden = encoder_net(tensor_in[0], hidden_in)
print(encoder_out)
print(encoder_hidden)
# 测试解码网络
tensor_in = torch.tensor([100])
hidden_in = torch.zeros(1, 1, 256)
encoder_out = torch.zeros(10, 256) # 第一维大小取决于MAX_LENGTH,此处为10
out1, out2, out3 = atten_decoder_net(tensor_in, hidden_in, encoder_out)
print(out1, out2, out3)
out1, out2 = decoder_net(tensor_in, hidden_in)
print(out1, out2)
这是一个PyTorch的代码实现，用于构建一个Seq2Seq模型。该模型由编码器和解码器两个部分组成，其中编码器采用了GRU，解码器可以选择不带Attention或者带Attention。
具体分析代码的功能：
导入需要的PyTorch模块和变量（如设备类型、最大长度等）。 实现编码器部分，包括：初始化函数、词嵌入层、GRU层，以及前向传播函数。其中前向传播函数的输入是一个输入序列（input）和一个隐藏状态（hidden），输出是一个输出张量（output）和一个隐藏状态（hidden）。 实现不带Attention的解码器部分，包括：初始化函数、词嵌入层、GRU层、输出层和softmax层，以及前向传播函数。其中前向传播函数的输入是一个输入序列（input）和一个隐藏状态（hidden），输出是一个输出张量（output）和一个隐藏状态（hidden）。 实现带Attention的解码器部分，包括：初始化函数、词嵌入层、Attention层、GRU层、输出层和softmax层，以及前向传播函数。其中前向传播函数的输入是一个输入序列（input）、一个隐藏状态（hidden）和编码器的输出张量（encoder_outputs），输出是一个输出张量（output）、一个隐藏状态（hidden）和注意力权重（atten_weight）。 实现初始化隐藏状态的函数initHidden，用于初始化隐藏状态。
注意，此处使用了一个三层的神经网络：GRU层、输出层和softmax层。其中，GRU层的输出作为输入传递给输出层，输出层再将结果传递给softmax层进行计算。最终的输出结果是一个向量，代表每个单词的概率分布。由于使用了softmax函数，因此输出结果之和等于1，可以作为概率分布使用。
5、训练脚本的搭建
import random
import time
import torch
import torch.nn as nn
from torch import optim
from datasets import readLangs, SOS_token, EOS_token, MAX_LENGTH
from models import EncoderRNN, AttenDecoderRNN
from utils import timeSince
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
MAX_LENGTH += 1 # 添加了终止符,比dataset中的的最大长度多1，因为要加入终止符
# 本任务完成英文到中文的翻译。若要倒过来，则要修改lang1和lang2的位置，还有pairs中的中英文词样本对的位置
lang1 = "en"
lang2 = "cn"
path = "../data/cmn.txt"
input_lang, output_lang, pairs = readLangs(lang1, lang2, path)
# print(len(pairs))
# print(input_lang.n_words)
# print(input_lang.index2word)
# print(output_lang.n_words)
# print(output_lang.index2word)
def listTotensor(input_lang, data):
indexes_in = [input_lang.word2index[word] for word in data.split(" ")] #得到句子所对应的索引列表[3,6,3,...]，经过embedding层，变为二维向量
indexes_in.append(EOS_token) # 在最后加入终止符,所以要比dataset中得MAX_LENGTH大1
input_tensor = torch.tensor(indexes_in,
dtype=torch.long,
device=device).view(-1, 1)
return input_tensor # 转换为张量并输出
#把pairs下的序列转换为输入tensor，并在tensor中插入一个终止符
# 将一个样本对转化为tensor
def tensorsFromPair(pair):
input_tensor = listTotensor(input_lang, pair[0]) # 将样本对前半部分英文转化为索引列表
output_tensor = listTotensor(output_lang, pair[1]) # 将样本对后半部分中文转化为索引列表
return (input_tensor, output_tensor)
# 计算loss
def loss_func(input_tensor, output_tensor, encoder, decoder, encoder_optimizer, decoder_optimizer,criterion):
encoder_hidden = encoder.initHidden() #初始化隐藏层
encoder_optimizer.zero_grad() #优化器梯度置零
decoder_optimizer.zero_grad()
input_len = input_tensor.size(0) # 输入输出长度，input_tensor,output_tensor均为二维张量。# 一句话的长度，
output_len = output_tensor.size(0) # input_tensor.size(1):为一个词的表示维度(embedding层的输出大小)
encoder_outputs = torch.zeros(MAX_LENGTH, encoder.hidden_size, device=device) # encoder的输出
#每次从input_tensor中取一个出来利用隐藏层信息进行encoder
for ei in range(input_len): # 将一个一句话的每个词依次编码
encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)
encoder_outputs[ei] = encoder_output[0, 0] #编码结果， # encoder_output为3维的向量
# encoder_outputs为一个句子的编码结果，为二维张量[[],[]...]
# 定义解码器
decoder_hidden = encoder_hidden
decoder_input = torch.tensor([[SOS_token]], device=device) #第一个解码输入定义为起始符SOS_token
# 加入随机因子，随机修改当前隐藏层的输入为真实的label，让模型收敛更快
use_teacher_forcing = True if random.random() < 0.5 else False
loss = 0 #loss初始化为0
if use_teacher_forcing: # 满足条件，使用
for di in range(output_len):
decoder_output, decoder_hidden, decoder_attention = decoder(
decoder_input, decoder_hidden, encoder_outputs
) # encoder_outputs:要解码的内容
loss += criterion(decoder_output, output_tensor[di]) # 计算loss, output_tensor:期待的输出(也就是label)
decoder_input = output_tensor[di] #下一次循环的输入直接定义为真实的label
else:
for di in range(output_len): # 不满足条件
decoder_output, decoder_hidden, decoder_attention = decoder(
decoder_input, decoder_hidden, encoder_outputs
)
loss += criterion(decoder_output, output_tensor[di])
# 定义下一次的输入为当前的预测结果
topV, topi = decoder_output.topk(1)
decoder_input = topi.squeeze().detach()
# 判断解码是否结束
if decoder_input.item() == EOS_token: # 等于终止符，解码结束
break
loss.backward() #梯度传播
encoder_optimizer.step()
decoder_optimizer.step()
return loss.item() / output_len
######
# 定义网络
hidden_size = 256
encoder = EncoderRNN(input_lang.n_words, hidden_size).to(device)
decoder = AttenDecoderRNN(hidden_size, output_lang.n_words,
max_len = MAX_LENGTH,
dropout_p=0.1).to(device)
lr = 0.01
encoder_optimizer = optim.SGD(encoder.parameters(), lr=lr) # 编码器优化器
decoder_optimizer = optim.SGD(decoder.parameters(), lr=lr) # 解码器优化器
#设置学习率调整 # 学习率的调整策略
scheduler_encoder = torch.optim.lr_scheduler.StepLR(encoder_optimizer,
step_size=1,
gamma=0.95)
scheduler_decoder = torch.optim.lr_scheduler.StepLR(decoder_optimizer,
step_size=1,
gamma=0.95)
# 定义损失函数
criterion = nn.NLLLoss()
# 不使用dataset,dataloader
# 直接生成样本对训练
n_iters = 10000 # 最大迭代次数
training_pairs = [
tensorsFromPair(random.choice(pairs)) for i in range(n_iters) # 挑选1000000个样本对
]
print_every = 1000 # 每迭代1000词打印一次信息
save_every = 10000
print_loss_total = 0
start = time.time()
for iter in range(1, n_iters+1):
training_pair = training_pairs[iter - 1]
input_tensor = training_pair[0]
output_tensor = training_pair[1]
loss = loss_func(input_tensor,
output_tensor,
encoder,
decoder,
encoder_optimizer,
decoder_optimizer,
criterion)
print_loss_total += loss
if iter % print_every == 0:
print_loss_avg = print_loss_total / print_every
print_loss_total = 0
print("{},{},{},{}".format(timeSince(start, iter/n_iters),
iter, iter / n_iters * 100,
print_loss_avg))
#保存模型
if iter % save_every == 0:
torch.save(encoder.state_dict(),
"../models/encoder_{}.pth".format(iter))
torch.save(decoder.state_dict(),
"../models/decoder_{}.pth".format(iter))
#更新学习率
if iter % 1000:
scheduler_encoder.step()
scheduler_decoder.step()
5、测试脚本的搭建
"""
利用训练好的模型进行推理计算
复用train.py的代码
去掉loss_func、学习率和优化器等部分代码
加载已经训练好的参数
"""
import random
import torch
import torch.nn as nn
from torch import optim
from datasets import readLangs, SOS_token, EOS_token, MAX_LENGTH
from models import EncoderRNN, AttenDecoderRNN
from utils import timeSince
import time
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
MAX_LENGTH = MAX_LENGTH + 1
lang1 = "en"
lang2 = "cn"
path = "../data/cmn.txt"
input_lang, output_lang, pairs = readLangs(lang1, lang2, path)
print(len(pairs))
print(input_lang.n_words)
print(input_lang.index2word)
print(output_lang.n_words)
print(output_lang.index2word)
def listTotensor(input_lang, data):
indexes_in = [input_lang.word2index[word] for word in data.split(" ")]
indexes_in.append(EOS_token)
input_tensor = torch.tensor(indexes_in,
dtype=torch.long,
device=device).view(-1, 1)
return input_tensor
def tensorsFromPair(pair):
input_tensor = listTotensor(input_lang, pair[0])
output_tensor = listTotensor(output_lang, pair[1])
return (input_tensor, output_tensor)
hidden_size = 256
encoder = EncoderRNN(input_lang.n_words, hidden_size).to(device)
decoder = AttenDecoderRNN(hidden_size,
output_lang.n_words,
max_len=MAX_LENGTH,
dropout_p=0.1).to(device)
# 加载已经训练好的参数
encoder.load_state_dict(torch.load("../models/encoder_10000.pth"))
decoder.load_state_dict(torch.load("../models/decoder_10000.pth"))
n_iters = 10
train_sen_pairs = [
random.choice(pairs) for i in range(n_iters)
]
training_pairs = [
tensorsFromPair(train_sen_pairs[i]) for i in range(n_iters)
]
for i in range(n_iters):
input_tensor, output_tensor = training_pairs[i]
encoder_hidden = encoder.initHidden()
input_len = input_tensor.size(0)
encoder_outputs = torch.zeros(MAX_LENGTH, encoder.hidden_size, device=device)
for ei in range(input_len):
encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)
encoder_outputs[ei] = encoder_output[0, 0]
decoder_hidden = encoder_hidden
decoder_input = torch.tensor([[SOS_token]], device=device)
use_teacher_forcing = True if random.random() < 0.5 else False
decoder_words = []
for di in range(MAX_LENGTH):
decoder_output, decoder_hidden, decoder_attention = decoder(
decoder_input, decoder_hidden, encoder_outputs
)
topV, topi = decoder_output.topk(1)
decoder_input = topi.squeeze().detach()
# 如果预测结果==终止符
if topi.item() == EOS_token: # 加入终止符
decoder_words.append("
break
else: # 加入预测结果
decoder_words.append(output_lang.index2word[topi.item()])
print(train_sen_pairs[i][0]) # input
print(train_sen_pairs[i][1]) # output
print(decoder_words)
可以上github把整个项目download下来
https://github.com/yingzhang123/Text_Sentiment_Classification
Time:2023.4.27 (周四) 五一小长假~~~ 如果上面代码对您有帮助，欢迎点个赞！！！
好文链接
您阅读本篇文章共花了：
发表评论