1. Johngo学长首页
  2. 技术点详细复盘专题
  3. 人工智能

机器翻译实战（英译汉）Transformer代码学习详解

• 人工智能 • 阅读 95

任务目标

基于Transformer实现英语翻译汉语。如有疏忽请多指教

数据

Hi. 嗨。
Hi. 你好。
Run.    你用跑的。
Wait!   等等！
Hello!  你好。
I try.  让我来。
I won!  我赢了。
Oh no!  不会吧。
Cheers! 乾杯!

Got it? 你懂了吗？
He ran. 他跑了。
Hop in. 跳进来。
I lost. 我迷失了。
I quit. 我退出。
I'm OK. 我沒事。
Listen. 听着。
No way! 不可能！
No way! 没门！
Really? 你确定？
Try it. 试试吧。
We try. 我们来试试。
Why me? 为什么是我？
……     ……

代码&数据来源

Github：transformer-simple
哈弗NLP

Transformer模型结构

Transformer的组成
    1. Encoder
        a. 若干个EncoderLayer（两个子层）
            i. Feed Forward Neural Network
            connected layer.子层间使用Add & Normalization 相连
            ii. Self-Attention
    2. Decoder
        a. 若干个DecoderLayer（三个子层）
            i. Feed Forward Neural Network
            connected layer.子层间使用Add & Normalization 相连
            ii. Encoder-Decoder-Attention，常规注意力机制
            connected layer.子层间使用Add & Normalization 相连
            iii. Self-Attention，自注意力机制

机器翻译实战（英译汉）Transformer代码学习详解

Batch and Masking

Mask策略一

一种就是普通的mask，就是自然语言处理中将某些字符（如标点符号，空格等）进行mask的操作


class Batch:
"""
    Batches and Masking
    "Object for holding a batch of data with mask during training." 在训练期间使用mask处理数据
"""
    def __init__(self, src, trg=None, pad=0):
"""
        构造函数
        @param src: 源数据
        @param trg: 目标数据
        @param pad: 需要mask掉的字符，默认为0
        一共有两种mask的方式：
        一种就是普通的mask，就是自然语言处理中将某些字符（如标点符号，空格等）进行mask的操作
        另一种就是对目标数据的mask，其原因是为了不让decoder在训练中看到后续的内容（即，我对于下一个字符的预测，只来源于前面的字符）
        对于src的mask就是第一种mask，而对于tgt的mask是第一种加第二种
"""

        src = torch.from_numpy(src).to(args.device).long()
        trg = torch.from_numpy(trg).to(args.device).long()

        self.src = src

        self.src_mask = (src != pad).unsqueeze(-2)
        if trg is not None:
            self.trg = trg[:, :-1]
            self.trg_y = trg[:, 1:]
            self.trg_mask = self.make_std_mask(self.trg, pad)
            self.ntokens = (self.trg_y != pad).data.sum()

    @staticmethod
    def make_std_mask(tgt, pad):
"""
        mask 目标数据
        "Create a mask to hide padding and future words." 翻译：创造一个mask来屏蔽补全词和字典外的词进行屏蔽
        @param tgt: 即构造函数中的trg，目标数据
        @param pad: 需要mask的字符，默认为0
        @return: 返回mask后的目标数据
"""

        tgt_mask = (tgt != pad).unsqueeze(-2)

        tgt_mask = tgt_mask & Variable(subsequent_mask(tgt.size(-1)).type_as(tgt_mask.data))
        return tgt_mask

Mask策略二

另一种就是对目标数据的mask，其原因是为了不让decoder在训练中看到后续的内容（即，我对于下一个字符的预测，只来源于前面的字符）

def subsequent_mask(size):
"""
    第二种mask策略
    "Mask out subsequent positions."
    @param size: 句子长度
    @return:
"""
    attn_shape = (1, size, size)

    subsequent_mask = np.triu(np.ones(attn_shape), k=1).astype('uint8')
    return torch.from_numpy(subsequent_mask) == 0

make_model

构造Transformer模型

def make_model(src_vocab, tgt_vocab, N = 6, d_model = 512, d_ff = 2048, h = 8, dropout = 0.1):
"""
    定义了一个接收超参数并生成完整模型的函数。
    @param src_vocab: 源数据字典长度
    @param tgt_vocab: 目标数据字典长度
    @param N: 层数layer
    @param d_model: 表征后的维度
    @param d_ff: FeedForward输出维度
    @param h: attention机制，head多头个数
    @param dropout:
    @return:
"""
    c = copy.deepcopy
    attn = MultiHeadedAttention(h, d_model).to(args.device)
    ff = PositionwiseFeedForward(d_model, d_ff, dropout).to(args.device)
    position = PositionalEncoding(d_model, dropout).to(args.device)

    model = Transformer(
        Encoder(EncoderLayer(d_model, c(attn), c(ff), dropout).to(args.device), N).to(args.device),
        Decoder(DecoderLayer(d_model, c(attn), c(attn), c(ff), dropout).to(args.device), N).to(args.device),
        nn.Sequential(Embeddings(d_model, src_vocab).to(args.device), c(position)),
        nn.Sequential(Embeddings(d_model, tgt_vocab).to(args.device), c(position)),
        Generator(d_model, tgt_vocab)).to(args.device)

    for p in model.parameters():
        if p.dim() > 1:
            nn.init.xavier_uniform_(p)
    return model.to(args.device)

模型核心代码

Transformer（EncoderDecoder）模型结构

class Transformer(nn.Module):
    def __init__(self, encoder, decoder, src_embed, tgt_embed, generator):
        super(Transformer, self).__init__()

        self.encoder = encoder
        self.decoder = decoder
        self.src_embed = src_embed
        self.tgt_embed = tgt_embed
        self.generator = generator

    def encode(self, src, src_mask):
"""
        对src进行embedding，并嵌入位置信息
        @param src: self.src_embed(src)，调用self.src_embed对应类中的forward函数，对batch.src进行embedding操作
        包含两个部分：一个是对输入的句子进行了embedding，第二个就是添加了位置信息
        @param src_mask: batch中的masking后的源数据
        @return: 调用nn.Embedding 对输入的src进行了向量化
"""
        return self.encoder(self.src_embed(src), src_mask)

    def decode(self, memory, src_mask, tgt, tgt_mask):
"""
        对tgt进行embedding，并嵌入位置信息
        @param memory: 下面forward()中的self.encode(src, src_mask)
        @param src_mask:
        @param tgt: tgt_embed的参数
        @param tgt_mask:
        @return:
"""
        return self.decoder(self.tgt_embed(tgt), memory, src_mask, tgt_mask)

    def forward(self, src, tgt, src_mask, tgt_mask):
"""
        调用decode函数和encode函数，其中encode的输出作为decode的输入
        @param src: encode的参数
        @param tgt:
        @param src_mask: encode和decode的参数
        @param tgt_mask:
        @return:
"""
        return self.decode(self.encode(src, src_mask), src_mask, tgt, tgt_mask)

嵌入

Embedding

class Embeddings(nn.Module):
    def __init__(self, d_model, vocab):
        super(Embeddings, self).__init__()
        self.lut = nn.Embedding(vocab, d_model)
        self.d_model = d_model

    def forward(self, x):
        return self.lut(x) * math.sqrt(self.d_model)

PositionalEncoding

class PositionalEncoding(nn.Module):
"""
    嵌入位置信息
"""
    def __init__(self, d_model, dropout, max_len=5000):
        super(PositionalEncoding, self).__init__()
        self.dropout = nn.Dropout(p=dropout)

        pe = torch.zeros(max_len, d_model, device=args.device)
        position = torch.arange(0., max_len,  device=args.device).unsqueeze(1)

        div_term = torch.exp(torch.arange(0., d_model, 2,  device=args.device) *- (math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        pe = pe.unsqueeze(0)
        self.register_buffer('pe', pe)

    def forward(self, x):
        x = x + Variable(self.pe[:, :x.size(1)], requires_grad=False)
        return self.dropout(x)

编码器

Encoder

"""
继承nn.Module，并重写构造函数forward函数
Q：何时调用forward函数？
A：实际上model(data)是等价于model.forward(data)，因为Module中定义了__call__()函数，该函数调用了forward()函数
"""
class Encoder(nn.Module):

    def __init__(self, layer, N):
        super(Encoder, self).__init__()
        self.layers = clones(layer, N)
        self.norm = LayerNorm(layer.size)

    def forward(self, x, mask):

        for layer in self.layers:
            x = layer(x, mask)
        return self.norm(x)

EncoderLayer

class EncoderLayer(nn.Module):
"""
    "Encoder is made up of self-attn and feed forward (defined below)"
"""
    def __init__(self, size, self_attn, feed_forward, dropout):
        super(EncoderLayer, self).__init__()

        self.self_attn = self_attn
        self.feed_forward = feed_forward

        self.sublayer = clones(SublayerConnection(size, dropout), 2)
        self.size = size

"""
    SublayerConnection的作用就是把self_attn和feed_forward连在一起，只不过每一层输出之后都要先norm再残差
"""
    def forward(self, x, mask):

        x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, mask))

        return self.sublayer[1](x, self.feed_forward)

残差连接层&正则层

正则层

class LayerNorm(nn.Module):
"""
    #自己定义的Layer归一化，这个代码号称是不调用额外的包的，所以这个自己实现
    #对应的就是上面图中黄色方块中的norm操作
"""
    def __init__(self, features, eps=1e-6):
        super(LayerNorm, self).__init__()
        self.a_2 = nn.Parameter(torch.ones(features))
        self.b_2 = nn.Parameter(torch.zeros(features))
        self.eps = eps

    def forward(self, x):

        mean = x.mean(-1, keepdim=True)
        std = x.std(-1, keepdim=True)
        return self.a_2 * (x - mean) / (std + self.eps) + self.b_2

残差层

class SublayerConnection(nn.Module):
"""
    子层之间的连接层（Add&Normal）
    子层：self-Attention & Feed Forward Network
    残差连接residual connection后面是layerNorm
    其中Add代表了Residual Connection
    通过将一部分的前一层的信息无差的传递到下一层，可以有效的提升模型性能，防止梯度消失，加快收敛
"""
    def __init__(self, size, dropout):
        super(SublayerConnection, self).__init__()
        self.norm = LayerNorm(size)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x, sublayer):
"""
        前向逻辑
        @param x: 上一层的输出
        @param sublayer: 子层
        @return:
"""

        return x + self.dropout(sublayer(self.norm(x)))

注意力机制

多头注意力机制

class MultiHeadedAttention(nn.Module):
    def __init__(self, h, d_model, dropout=0.1):
        super(MultiHeadedAttention, self).__init__()

        assert d_model % h == 0
        self.d_k = d_model // h
        self.h = h

        self.linears = clones(nn.Linear(d_model, d_model), 4)
        self.attn = None
        self.dropout = nn.Dropout(p=dropout)

    def forward(self, query, key, value, mask=None):
"""
        query,key,value均为输入x，详见EncoderLayer类的forward函数
        @param query:
        @param key:
        @param value:
        @param mask:
        @return:
"""
        if mask is not None:
            mask = mask.unsqueeze(1)
        nbatches = query.size(0)

        for l, x in zip(self.linears, (query, key, value)):
            lx = l(x)
            print(lx.view(nbatches, -1, self.h, self.d_k).shape)
            print(l(x).view(nbatches, -1, self.h, self.d_k).transpose(1,2).shape)

        query, key, value = \
            [l(x).view(nbatches, -1, self.h, self.d_k).transpose(1, 2)
             for l, x in zip(self.linears, (query, key, value))]

        x, self.attn = attention(query, key, value, mask=mask, dropout=self.dropout)

        x = x.transpose(1, 2).contiguous().view(nbatches, -1, self.h * self.d_k)
        return self.linears[-1](x)

注意力分数计算

def attention(query, key, value, mask=None, dropout=None):
"""
    Compute 'Scaled Dot Product Attention
    @param query: (batch, #head头数, sequence length, feature dimension(d_k))
    @param key: (batch, #head头数, sequence length, feature dimension(d_k))
    @param value: K和Q的shape必须相同的，而V可以不同
    @param mask: (batch, 1, sequence length, sequence length)
    @param dropout:
    @return: p_attn，将注意力分数转换为概率的矩阵，p_attn与value的乘积
"""
    d_k = query.size(-1)
"""
    torch.matmul() tensor乘法
    高维矩阵遵循的原则是：在多维矩阵相乘中，需最后两维满足shape匹配原则，最后两维才是有数据的矩阵，前面的维度只是矩阵的排列而已！
    这也是在MultiHeadedAttention这个函数中将数据输入attention函数时要进行transpose的原因
    因为只有seq_length跟 embedding才是要进行注意力点乘的关键啊，其他俩维度只是排列而已啊
"""

    scores = torch.matmul(query, key.transpose(-2, -1)) / math.sqrt(d_k)

    if mask is not None:
        scores = scores.masked_fill(mask == 0, -1e9)
    p_attn = F.softmax(scores, dim=-1)
    if dropout is not None:
        p_attn = dropout(p_attn)
    return torch.matmul(p_attn, value), p_attn

前馈层

class PositionwiseFeedForward(nn.Module):
    def __init__(self, d_model, d_ff, dropout=0.1):
        super(PositionwiseFeedForward, self).__init__()

        self.w_1 = nn.Linear(d_model, d_ff)
        self.w_2 = nn.Linear(d_ff, d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):

        return self.w_2(self.dropout(F.relu(self.w_1(x))))

解码器

Decoder

class Decoder(nn.Module):
    def __init__(self, layer, N):
        super(Decoder, self).__init__()
        self.layers = clones(layer, N)
        self.norm = LayerNorm(layer.size)

    def forward(self, x, memory, src_mask, tgt_mask):
"""
        @param x: 每个DecoderLayer的输出，即DecoderLayer层之间的信息流动
        @param memory: 是Encoder的输出
        @param src_mask: Encoder的mask用于padding
        @param tgt_mask: Decoder的mask用于隐藏后面的单词的输出
        @return:
"""
        for layer in self.layers:
            x = layer(x, memory, src_mask, tgt_mask)
        return self.norm(x)

DecoderLayer

class DecoderLayer(nn.Module):
"""
    每个DecoderLayer包含三个子层
"""
    def __init__(self, size, self_attn, src_attn, feed_forward, dropout):
        super(DecoderLayer, self).__init__()
        self.size = size
        self.self_attn = self_attn
        self.src_attn = src_attn
        self.feed_forward = feed_forward

        self.sublayer = clones(SublayerConnection(size, dropout), 3)

    def forward(self, x, memory, src_mask, tgt_mask):
        m = memory

        x = self.sublayer[0](x, lambda x: self.self_attn(x, x, x, tgt_mask))

        x = self.sublayer[1](x, lambda x: self.src_attn(x, m, m, src_mask))

        return self.sublayer[2](x, self.feed_forward)

阅读源代码过程中的参考资料

[En]

Reference materials in the process of reading source code

  1. Transformer代码记录整理
  2. 第二章：Transformer架构解析
  3. Transformer代码详解
  4. Transformer代码阅读
  5. pytorch yield
  6. unsqueeze和sequeeze
  7. np.triu()的用法
  8. nn.Sequential
  9. PositionalEmbedding
  10. python中copy()和deepcopy()详解
  11. make_model()中 nn.init.xavier_uniform_()初始化
  12. 深度学习中的Normalization
  13. 残差连接和归一化层
  14. zip函数
  15. nn.Module()什么时候调用forward()函数=>call ()函数
  16. torch.matmul()用法介绍
  17. 多维tensor的相乘到底在乘什么
  18. pytorch broadcasting广播
  19. pytorch transpose与contiguous
  20. Linear、Dense、MLP、FC的区别
  21. 保姆级硬核图解Transformer

Original: https://blog.csdn.net/weixin_44343282/article/details/124575684
Author: 九筒-
Title: 机器翻译实战（英译汉）Transformer代码学习详解

原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/527447/

转载文章受原作者版权保护。转载请注明原作者出处！

赞 (0)
0

【自取】最近整理的，有需要可以领取学习：



Linux核心资料大放送~

全栈面试题汇总（持续更新&可下载）

一个提高学习100%效率的工具！

【超详细】深度学习面试题目！

LeetCode Python刷题答案下载！

LeetCode Java版刷题答案下载！

LeetCode C++ 版本，抓紧保存！

LeetCode GO语言 刷题答案下载！

大家都在看

亲爱的 Coder【最近整理，可免费获取】👉 最新必读书单  | 👏 面试题下载  | 🌎 免费的AI知识星球