attention-transformer-test

一级目录

二级目录

三级目录

import math
import torch
import numpy as np
import torch.nn as nn
import torch.optim as optim
import torch.utils.data as Data

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

数据准备

# S: Symbol that shows starting of decoding input
# E: Symbol that shows starting of decoding output
# P: Symbol that will fill in blank sequence if current batch data size is short than time steps
sentences = [['ich mochte ein bier P', 'S i want a beer .', 'i want a beer . E'],['ich mochte ein cola P', 'S i want a coke .', 'i want a coke . E']
]
# 一行代表一个样本，数据集只有两个样本，
# 每个样本三个东西 encoder input 德语     decoder input 英语   decoder output 真实值
# P表示padding
# Padding Should be Zero

训练阶段
输入是目标序列的右移版本（Shifted Right）。

例如，在机器翻译任务中，假设目标句子是 [“<sos>”, “A”, “B”, “C”, “<eos>”]，

则解码器的输入会被右移一位，变成 [“<sos>”, “A”, “B”, “C”]，

而模型需要预测的输出是 [“A”, “B”, “C”, “<eos>”]。

为什么右移？
这是为了在训练时通过Teacher Forcing策略让模型学习预测下一个词（即输入是 t-1 时刻的真实词，输出是 t 时刻的词）。同时，解码器的自注意力机制会通过掩码（Mask）确保每个位置只能看到前面的词，避免信息泄漏。

此时输入是真实的句子，但经过了右移和掩码处理。

encoder input得到隐藏变量，传入decoder，同时decoder input也输入，最终得到预测句子与decoder output计算loss

构建词典

# 德语和英语分别构建词库# 字母与索引对应
src_vocab = {'P' : 0, 'ich' : 1, 'mochte' : 2, 'ein' : 3, 'bier' : 4, 'cola' : 5}
src_vocab_size = len(src_vocab)tgt_vocab = {'P' : 0, 'i' : 1, 'want' : 2, 'a' : 3, 'beer' : 4, 'coke' : 5, 'S' : 6, 'E' : 7, '.' : 8}
tgt_vocab_size = len(tgt_vocab)

# 目标序列索引与字母对应
idx2word = {i: w for i , w in enumerate(tgt_vocab)}

{0: ‘P’,
1: ‘i’,
2: ‘want’,
3: ‘a’,
4: ‘beer’,
5: ‘coke’,
6: ‘S’,
7: ‘E’,
8: ‘.’}

# 词库大小
src_vocab_size = 6
tgt_vocab_size = 9

# 句子长度最大
src_len = 5 # enc_input max sequence length 
tgt_len = 6 # dec_input(=dec_output) max sequence length

模型参数

字嵌入 & 位置嵌入的维度，这俩值是相同的，因此用一个变量就行了

FeedForward 层隐藏神经元个数

Q、K、V 向量的维度，其中 Q 与 K 的维度必须相等，V 的维度没有限制，不过为了方便起见，我都设为 64

Encoder 和 Decoder 的个数

多头注意力中 head 的数量

# Transformer Parameters
d_model = 512  # Embedding Size
d_ff = 2048 # FeedForward dimension
d_k = d_v = 64  # dimension of K(=Q), V
n_layers = 6  # number of Encoder of Decoder Layer
n_heads = 8  # number of heads in Multi-Head Attention

数据构建

sentences

[[‘ich mochte ein bier P’, ‘S i want a beer .’, ‘i want a beer . E’],

[‘ich mochte ein cola P’, ‘S i want a coke .’, ‘i want a coke . E’]]

src_vocab = {‘P’ : 0, ‘ich’ : 1, ‘mochte’ : 2, ‘ein’ : 3, ‘bier’ : 4, ‘cola’ : 5}

tgt_vocab = {‘P’ : 0, ‘i’ : 1, ‘want’ : 2, ‘a’ : 3, ‘beer’ : 4, ‘coke’ : 5, ‘S’ : 6, ‘E’ : 7, ‘.’ : 8}

根据字典得到向量输入和输出

def make_data(sentences):enc_inputs,dec_inputs,dec_outputs = [],[],[]for i in range(len(sentences)): # 2# 第一个样本，第二个样本，一共两个句子# 每个句子的encoder input 德语     decoder input 英语   decoder output单词对应的数字索引enc_input = [ [ src_vocab[n] for n in sentences[i][0].split() ] ]  # [[1, 2, 3, 4, 0], [1, 2, 3, 5, 0]]dec_input = [ [ tgt_vocab[n] for n in sentences[i][1].split() ] ]  # [[6, 1, 2, 3, 4, 8], [6, 1, 2, 3, 5, 8]]dec_output = [ [ tgt_vocab[n] for n in sentences[i][2].split() ] ]  # [[1, 2, 3, 4, 8, 7], [1, 2, 3, 5, 8, 7]]enc_inputs.extend(enc_input)dec_inputs.extend(dec_input)dec_outputs.extend(dec_output)return torch.LongTensor(enc_inputs),torch.LongTensor(dec_inputs),torch.LongTensor(dec_outputs)

sentences[0][0].split()

[‘ich’, ‘mochte’, ‘ein’, ‘bier’, ‘P’]

src_vocab = {‘P’ : 0, ‘ich’ : 1, ‘mochte’ : 2, ‘ein’ : 3, ‘bier’ : 4, ‘cola’ : 5}

[src_vocab[n] for n in sentences[0][0].split()]

[1, 2, 3, 4, 0]

enc_inputs, dec_inputs, dec_outputs = make_data(sentences)

enc_inputs tensor([[1, 2, 3, 4, 0]])

dec_inputs tensor([[6, 1, 2, 3, 4, 8]])

dec_outputs tensor([[1, 2, 3, 4, 8, 7]])

class MyDataSet(Data.Dataset):def __init__(self,enc_inputs,dec_inputs,dec_outputs):super(MyDataSet,self).__init__()self.enc_inputs = enc_inputsself.dec_inputs = dec_inputsself.dec_outputs = dec_outputsdef __len__(self):return self.enc_inputs.shape[0]def __getitem__(self,idx):return self.enc_inputs[idx], self.dec_inputs[idx], self.dec_outputs[idx]loader = Data.DataLoader(MyDataSet(enc_inputs, dec_inputs, dec_outputs), 2, True)

Positional Encoding

$P{E}_{(pos,2i)}=sin(pos/10000^{2i/d})$

$P{E}_{(pos,2i+1)}=cos(pos/10000^{2i/d})$

对于一个行是token，列是维度的矩阵，上式中 $pos$
指的是一句话中某个字的位置，取值范围是 [0,max_seqence_length]
，
$i$指的是字向量的维度序号，取值范围是 [0,embedding_demension/2]
，
$d$指的是 embedding_dimension​的值

class PositionalEncoding(nn.Module):def __init__(self,d_model,dropout=0.1,max_len=5000):super(PositionalEncoding,self).__init__()self.dropout = nn.Dropout(p = dropout)pe = torch.zeros(max_len,d_model)# 初始化矩阵，词长度 * 模型维度position = torch.arange(0, max_len, dtype = torch.float).unsqueeze(1)# 从0到最大长度,标量转化为n*1的维度div_term = torch.exp(torch.arange(0, d_model, 2).float()*(-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).transpose(0, 1)# 增加第一个维度，便于调整批次大小，转换第一个和第二个维度# pe  =  max_len * b * d_modelself.register_buffer('pe', pe)def forward(self,x):'''x: [seq_len, batch_size, d_model]'''x = x + self.pe[:x.size(0), :]return self.dropout(x)

self.register_buffer(‘pe’, pe)

在PyTorch中，self.register_buffer(‘pe’, pe) 的作用是将位置编码张量 pe 注册为模块的缓冲区（buffer），使其成为模型的一部分，但不作为可训练参数。以下是详细解释：

关键作用
持久化保存

当模型保存（torch.save）或加载（torch.load）时，注册的缓冲区 pe 会自动包含在模型的 state_dict 中，无需手动处理。

如果直接赋值 self.pe = pe，则 pe 不会被保存到 state_dict 中，导致模型加载时丢失位置编码。

设备一致性

当模型被移动到 GPU 或 CPU（如 model.to(device)）时，pe 会自动跟随模型迁移到对应设备，无需手动管理设备同步。

避免梯度计算

pe 是预先计算好的固定编码（非可学习参数），注册为缓冲区后，PyTorch 不会对其计算梯度，减少内存和计算开销。

为什么不用 nn.Parameter？
nn.Parameter 用于定义需要训练的参数（如权重矩阵），而位置编码通常是固定的（按公式预先计算），因此不需要梯度更新。

如果想让位置编码可学习（如某些变体模型），则需改用 self.pe = nn.Parameter(pe)。

位置编码

位置编码的目的是为模型提供 序列中每个位置的几何或顺序信息，无论该位置上的单词是什么。例如：

位置0的编码表示“第一个位置”，无论这个位置是单词“apple”还是“banana”。

位置1的编码表示“第二个位置”，依此类推。

设计原则：位置编码与单词内容解耦，专注于位置本身。

假设两个批次输入：

批次1：序列 [“I”, “love”, “you”]（位置0,1,2）。

批次2：序列 [“He”, “hates”, “me”]（位置0,1,2）。表示每一句话中，单词的位置顺序关系

位置编码应用：

两个批次均使用 self.pe[:3, :]（前3行位置编码）。

尽管单词不同，但位置0的编码相同，位置1的编码相同，位置2的编码相同。

模型通过位置编码知道“love”在位置1，“hates”也在位置1，从而捕捉到它们都是序列中的第二个token。

e x p { 2 i ∗ − [ l o g ( 10000 ) / d ] } exp\{2i*-[log(10000)/d]\} exp{2i∗−[log(10000)/d]} = e x p { − l o g [ 1000 0 2 i / d ] } exp\{-log[10000^{2i/d}]\} exp{−log[100002i/d]} = $1/10000^{2i/d}$

torch.arange(0, 6, dtype = torch.float).unsqueeze(1)

tensor([[0.],[1.],[2.],[3.],[4.],[5.]])

torch.exp(torch.arange(0, 10, 2).float()*(-math.log(10000.0)/d_model))

tensor([1.0000, 0.9647, 0.9306, 0.8977, 0.8660])

torch.arange(0, 6, dtype = torch.float).unsqueeze(1)*torch.exp(torch.arange(0, 10, 2).float()*(-math.log(10000.0)/d_model))

tensor([[0.0000, 0.0000, 0.0000, 0.0000, 0.0000],[1.0000, 0.9647, 0.9306, 0.8977, 0.8660],[2.0000, 1.9293, 1.8611, 1.7954, 1.7319],[3.0000, 2.8940, 2.7917, 2.6931, 2.5979],[4.0000, 3.8586, 3.7223, 3.5907, 3.4639],[5.0000, 4.8233, 4.6529, 4.4884, 4.3298]])

torch.zeros(6,10).shape

torch.Size([6, 10])

torch.zeros(6,10).unsqueeze(0).shape

torch.Size([1, 6, 10])

torch.zeros(6,10).unsqueeze(0).transpose(0, 1).shape

torch.Size([6, 1, 10])

torch.arange(0, 6, dtype = torch.float)

tensor([0., 1., 2., 3., 4., 5.])

torch.Size([6])

torch.arange(0, 6, dtype = torch.float).unsqueeze(1)

torch.Size([6, 1])

tensor([[0.],

    [1.],[2.],[3.],[4.],[5.]])

Pad Mask注意力机制的填充掩码（Padding Mask）

通过掩码操作（masked_fill）应用到注意力分数（QK^T）上

目的是在计算注意力权重时，屏蔽输入序列中的填充位置（Padding Tokens，通常用0表示）

填充掩码 (Padding Mask) 屏蔽输入序列中的填充位置（如补零的无效词），防止模型关注这些位置。 编码器和解码器的输入序列处理 输入序列 [I, love, , ]，掩码标记 位置为无效。

序列掩码 (Sequence Mask) 限制模型在生成时只能看到当前位置之前的信息（防止未来信息泄露），即因果掩码（Causal Mask）。 解码器的自注意力（训练时） 生成第3个词时，掩码会屏蔽第4、5、6…位置的词，确保预测仅依赖已生成的词。

由于在 Encoder 和 Decoder 中都需要进行 mask 操作，因此就无法确定这个函数的参数中 seq_len 的值，如果是在 Encoder 中调用的，seq_len 就等于 src_len；如果是在 Decoder 中调用的，seq_len 就有可能等于 src_len，也有可能等于 tgt_len（因为 Decoder 有两次 mask）

这个函数最核心的一句代码是 seq_k.data.eq(0)，这句的作用是返回一个大小和 seq_k 一样的 tensor，只不过里面的值只有 True 和 False。如果 seq_k 某个位置的值等于 0，那么对应位置就是 True，否则即为 False。举个例子，输入为 seq_data = [1, 2, 3, 4, 0]，seq_data.data.eq(0) 就会返回 [False, False, False, False, True]

生成一个掩码，标记 seq_k 中所有填充位置（值为0的位置），并将该掩码扩展到与 seq_q 的每个位置对齐，最终形状为 [batch_size, seq_len_q, seq_len_k]。

计算注意力分数：
首先计算查询（Q）和键（K）的相似度分数：

Attention Scores=Q⋅KT
形状为
[batch_size,len_q,len_k]

应用掩码：
使用生成的掩码（pad_attn_mask）将填充位置的注意力分数设置为一个极小的值（如-1e9），使得这些位置在Softmax后权重趋近于0：

scores = scores.masked_fill(pad_attn_mask, -1e9) # 填充位置的分数被屏蔽

def get_attn_pad_mask(seq_q,seq_k):  #  Q*V--K'''seq_q: [batch_size, seq_len]seq_k: [batch_size, seq_len]seq_len could be src_len or it could be tgt_lenseq_len in seq_q and seq_len in seq_k maybe not equal'''# print('seq_q',seq_q,seq_q.shape)# print('seq_k',seq_k,seq_k.shape)# print(seq_k.size())batch_size,len_q = seq_q.size()batch_size,len_k = seq_k.size()# eq(zero) is PAD token# 对两个张量Tensor进行逐元素的比较，若相同位置的两个元素相同，则返回True；若不同，返回False。pad_attn_mask = seq_k.data.eq(0).unsqueeze(1)'''unsqueeze(1)：在维度1（序列长度维度）上增加一个维度，变为 [batch_size, 1, seq_len_k]，目的是后续通过广播（Broadcast）机制扩展该掩码。'''# [batch_size, 1, len_k], True is maskedreturn pad_attn_mask.expand(batch_size,len_q,len_k)'''expand()：将掩码从 [batch_size, 1, seq_len_k] 扩展到 [batch_size, seq_len_q, seq_len_k]。扩展逻辑：每个批次中的每个查询位置（共 seq_len_q 个）都会复制一份相同的键序列掩码，最终每个查询位置都共享相同的键填充掩码。# [batch_size, len_q, len_k]'''

torch.randn([3,4])

tensor([[-0.2164,  0.7981,  2.0642,  1.5489],[-1.3097, -0.7571,  2.3407, -0.2603],[-0.3206, -0.0143, -0.0538, -0.9373]])

torch.rand([3,4]).data.eq(0)
# 3*4

tensor([[False, False, False, False],[False, False, False, False],[False, False, False, False]])

torch.rand([3,4]).data.eq(0).unsqueeze(1)
# 3-1-4

tensor([[[False, False, False, False]],[[False, False, False, False]],[[False, False, False, False]]])

torch.rand([3,4]).data.eq(0).unsqueeze(1).expand(3,2,4)

tensor([[[False, False, False, False],[False, False, False, False]],[[False, False, False, False],[False, False, False, False]],[[False, False, False, False],[False, False, False, False]]])

例子

假设输入：

seq_k 是 [[1, 2, 0, 0], [3, 0, 0, 0]]（两个样本，填充符为0）。

seq_q 的 len_q 是3。

生成的掩码过程：

seq_k = torch.tensor( [[1, 2, 0, 0], [3, 0, 0, 0]] )
seq_k.eq(0)
# torch.Size([2, 4])  batch_size*len_k

tensor([[False, False,  True,  True],[False,  True,  True,  True]])

seq_k.eq(0).unsqueeze(1)
# torch.Size([2, 1, 4])  batch_size*1*len_k

tensor([[[False, False,  True,  True]],[[False,  True,  True,  True]]])

seq_k.eq(0).unsqueeze(1).expand(2, 3, 4)
# torch.Size([2, 3, 4])  batch_size*len_q*seq_k# 每个批次中的每个查询位置（共 seq_len_q 个）都会复制一份相同的键序列掩码，最终每个查询位置都共享相同的键填充掩码。

tensor([[[False, False,  True,  True],[False, False,  True,  True],[False, False,  True,  True]],[[False,  True,  True,  True],[False,  True,  True,  True],[False,  True,  True,  True]]])

# 第一个样本的掩码（共3个查询位置，每个位置共享相同的键掩码）：
[[[False, False,  True,  True],[False, False,  True,  True],[False, False,  True,  True]],# 第二个样本的掩码：[[False,  True,  True,  True],[False,  True,  True,  True],[False,  True,  True,  True]]]

[[[False, False, True, True],[False, False, True, True],[False, False, True, True]],[[False, True, True, True],[False, True, True, True],[False, True, True, True]]]

在注意力机制中的应用
在计算注意力权重时，将此掩码应用到注意力分数上：

效果：填充位置的注意力权重趋近于0，模型不会关注这些位置。

总结
这段代码生成的掩码用于屏蔽键序列（Key）中的填充位置，确保模型在计算注意力时忽略无效的填充信息。它在Transformer的编码器和解码器中广泛使用，例如处理变长输入序列或目标序列的生成任务。

'''
attention_scores = query @ key.transpose(-2, -1)  # [batch_size, seq_len_q, seq_len_k]attention_scores = attention_scores.masked_fill(pad_attn_mask, -1e9)  # 填充位置设为极小值attention_weights = F.softmax(attention_scores, dim=-1)
'''

'\nattention_scores = query @ key.transpose(-2, -1)  # [batch_size, seq_len_q, seq_len_k]\n\nattention_scores = attention_scores.masked_fill(pad_attn_mask, -1e9)  # 填充位置设为极小值\n\nattention_weights = F.softmax(attention_scores, dim=-1)\n'

自注意力（Self-Attention）

Q、K、V 来源：三者均来自同一输入序列。

长度关系：

len_q = len_k（因为查询和键来自同一序列）。

示例：

编码器自注意力：输入序列是源语言句子（如 “I love NLP”），len_q 和 len_k 均为源序列长度（如 3）。

解码器自注意力：输入序列是目标语言句子（如 “我 爱 自然语言处理”），len_q 和 len_k 均为目标序列长度（如 4）。

交叉注意力（Cross-Attention）

Q、K、V 来源：

Q（查询）来自解码器的输入（目标序列）。

K、V（键、值）来自编码器的输出（源序列）。

长度关系：

len_q 是目标序列长度（解码器输入长度）。

len_k 是源序列长度（编码器输出长度）。

示例：

源序列（编码器输入）长度为 5（如 “I love NLP tasks”）。

目标序列（解码器输入）长度为 3（如 “我 爱 自然语言处理”）。

此时 len_q = 3，len_k = 5。

Subsequence Mask

Subsequence Mask 只有 Decoder 会用到，主要作用是屏蔽未来时刻单词的信息。首先通过 np.ones() 生成一个全 1 的方阵，然后通过 np.triu() 生成一个上三角矩阵，下图是 np.triu() 用法

def get_attn_subsequence_mask(seq):'''seq: [batch_size, tgt_len]'''attn_shape = [seq.size(0), seq.size(1), seq.size(1)]subsequence_mask = np.triu( np.ones(attn_shape) , k = 1 )subsequence_mask = torch.from_numpy(subsequence_mask).byte()# .byte()：将数据类型转为ByteTensor（旧版PyTorch中布尔掩码常用此类型）。# 注意：新版PyTorch推荐使用 bool()，但 .byte() 仍然兼容。return subsequence_mask # [batch_size, tgt_len, tgt_len]

np.ones([2,4,4])
# (2, 4, 4)

array([[[1., 1., 1., 1.],[1., 1., 1., 1.],[1., 1., 1., 1.],[1., 1., 1., 1.]],[[1., 1., 1., 1.],[1., 1., 1., 1.],[1., 1., 1., 1.],[1., 1., 1., 1.]]])

np.triu( np.ones([2,4,4]) , k = 1 )

array([[[0., 1., 1., 1.],[0., 0., 1., 1.],[0., 0., 0., 1.],[0., 0., 0., 0.]],[[0., 1., 1., 1.],[0., 0., 1., 1.],[0., 0., 0., 1.],[0., 0., 0., 0.]]])

torch.from_numpy(np.triu( np.ones([2,4,4]) , k = 1 )).byte()

tensor([[[0, 1, 1, 1],[0, 0, 1, 1],[0, 0, 0, 1],[0, 0, 0, 0]],[[0, 1, 1, 1],[0, 0, 1, 1],[0, 0, 0, 1],[0, 0, 0, 0]]], dtype=torch.uint8)

np.triu(data,0)

data = np.ones(5)
np.triu(data,0)
# 0表示正常的上三角矩阵

array([[1., 1., 1., 1., 1.],[0., 1., 1., 1., 1.],[0., 0., 1., 1., 1.],[0., 0., 0., 1., 1.],[0., 0., 0., 0., 1.]])

np.triu(data,-1)
# -1表示对角线的位置下移动一个对角线

array([[1., 1., 1., 1., 1.],[1., 1., 1., 1., 1.],[0., 1., 1., 1., 1.],[0., 0., 1., 1., 1.],[0., 0., 0., 1., 1.]])

np.triu(data,1)
# -1表示对角线的位置上移动一个对角线

array([[0., 1., 1., 1., 1.],[0., 0., 1., 1., 1.],[0., 0., 0., 1., 1.],[0., 0., 0., 0., 1.],[0., 0., 0., 0., 0.]])

mask = [[0, 1, 1],  # 第1个位置只能关注自身（无未来词）[0, 0, 1],  # 第2个位置可关注前两个位置[0, 0, 0],  # 第3个位置可关注所有位置（但实际可能被填充掩码限制）
]

'''
固定性：因果掩码的矩阵在每次前向传播时固定生成，与训练步骤无关。动态性：掩码的“动态”体现在序列生成过程中，而非训练过程中：训练阶段：模型并行处理整个目标序列，但通过掩码强制每个位置只能看到前面的词。推断阶段：模型自回归生成词，每次生成第 i 个词时，仅使用前 i 个已生成的词。掩码始终固定，模型在训练时直接学习如何利用允许关注的位置，而非逐步调整掩码。
'''

'\n固定性：因果掩码的矩阵在每次前向传播时固定生成，与训练步骤无关。\n\n动态性：掩码的“动态”体现在序列生成过程中，而非训练过程中：\n\n训练阶段：模型并行处理整个目标序列，但通过掩码强制每个位置只能看到前面的词。\n\n推断阶段：模型自回归生成词，每次生成第 i 个词时，仅使用前 i 个已生成的词。\n\n掩码始终固定，模型在训练时直接学习如何利用允许关注的位置，而非逐步调整掩码。\n'

ScaledDotProductAttention

class ScaledDotProductAttention(nn.Module):def __init__(self):super(ScaledDotProductAttention,self).__init__()def forward( self,Q,K,V,attn_mask ):scores = '''Q: [batch_size, n_heads, len_q, d_k]K: [batch_size, n_heads, len_k, d_k]V: [batch_size, n_heads, len_v(=len_k), d_v]attn_mask: [batch_size, n_heads, seq_len, seq_len]'''scores = torch.matmul( Q,K.transpose(-1,-2) )/np.sqrt(d_k)  '''[batch_size, n_heads, len_q, d_k] * [batch_size, n_heads, d_k, len_k]'''# scores : [batch_size, n_heads, len_q, len_k]scores.masked_fill_(attn_mask,-1e9)# Fills elements of self tensor with value where mask is True.attn = nn.Softmax(dim = -1)(scores)context = torch.matmul(attn,V)# [batch_size, n_heads, len_q, d_v]return context, attn

torch.randn(2,2,3,4).transpose(-1,-2)
# torch.Size([2, 2, 4, 3])

tensor([[[[-0.1955,  0.6530, -0.9631],[ 1.2081, -0.2663, -0.3650],[-1.4615,  0.7132,  1.0989],[ 0.0347, -0.4724, -0.4844]],[[-1.7010,  1.5704, -0.6445],[-0.4314,  0.3485,  0.4670],[ 0.0839,  0.4096, -0.1395],[-0.8040, -1.7008,  0.8796]]],[[[ 0.4990,  0.3628, -0.0200],[-1.7474,  0.6363,  1.0624],[ 1.2794,  0.8372,  1.4988],[-0.4211, -0.5843,  0.5880]],[[-0.8321, -0.5741, -0.1154],[-0.2121, -0.5078, -0.5043],[ 0.0310,  0.5007,  0.0082],[ 0.3993, -0.2361, -0.1918]]]])

MultiHeadAttention

完整代码中一定会有三处地方调用 MultiHeadAttention()，Encoder Layer 调用一次，传入的 input_Q、input_K、input_V 全部都是 enc_inputs；Decoder Layer 中两次调用，第一次传入的全是 dec_inputs，第二次传入的分别是 dec_outputs，enc_outputs，enc_outputs

class MultiHeadAttention(nn.Module):def __init__(self):super(MultiHeadAttention,self).__init__()self.W_Q = nn.Linear(d_model , d_k * n_heads , bias = False)self.W_K = nn.Linear(d_model , d_k * n_heads , bias = False)self.W_V = nn.Linear(d_model , d_v * n_heads , bias = False)self.fc = nn.Linear(n_heads * d_v, d_model, bias=False)def forward(self , input_Q , input_K , input_V , attn_mask):'''input_Q: [batch_size, len_q, d_model]input_K: [batch_size, len_k, d_model]input_V: [batch_size, len_v(=len_k), d_model]attn_mask: [batch_size, seq_len, seq_len]'''residual , batch_size = input_Q , input_Q.size(0)# (B, S, D) -proj-> (B, S, D_new) -split-> (B, S, H, W) -trans-> (B, H, S, W)Q = self.W_Q(input_Q).view(batch_size , -1 , n_heads , d_k).transpose(1,2)# Q: [batch_size, n_heads, len_q, d_k]K = self.W_K(input_K).view(batch_size, -1, n_heads, d_k).transpose(1,2)  # K: [batch_size, n_heads, len_k, d_k]V = self.W_V(input_V).view(batch_size, -1, n_heads, d_v).transpose(1,2)  # V: [batch_size, n_heads, len_v(=len_k), d_v]attn_mask = attn_mask.unsqueeze(1).repeat(1,n_heads,1,1)# attn_mask : [batch_size, n_heads, seq_len, seq_len]# context: [batch_size, n_heads, len_q, d_v]# attn: [batch_size, n_heads, len_q, len_k]context, attn = ScaledDotProductAttention()(Q, K, V, attn_mask)context = context.transpose(1, 2).reshape(batch_size, -1, n_heads * d_v) # context: [batch_size, len_q, n_heads * d_v]# 这样做的目的是将多头的输出进行合并，把每个头的 d_v 维度信息拼接在一起，# 形成一个更高维度的特征表示，方便后续的层（如全连接层）处理。output = self.fc(context) # [batch_size, len_q, d_model]return nn.LayerNorm(d_model).cuda()(output + residual), attn

分头

输入 input_Q 的形状为 (batch_size, seq_len, d_model)（例如：(32, 100, 512)），经过 self.W_Q 后形状不变（线性层不改变前两维）。

batch_size：保留批次维度（第一维固定）。
-1：自动推断维度，确保总元素数不变。

由于 d_model = n_heads * d_k（例如：512 = 8 * 64），

则：
seq_len * d_model = seq_len * n_heads * d_k，因此 -1 会被推断为 seq_len。
n_heads, d_k：将最后一维 d_model 拆分为多头（n_heads）和头维度（d_k）。

view

在 Transformer 架构的多头注意力机制里，通常要求头的数量 n_heads 与每个头的特征维度 d_k 相乘等于原始的特征维度 d_model，即 d_model = n_heads * d_k ，下面从不同方面详细阐述这一要求：

理论层面
信息完整性：d_model 代表了模型在某一层中所处
理的特征的总维度，它包含了输入序列的所有特征信息。当采用多头注意力机制时，为了保证在分头计算注意力的过程中不丢失信息，所有头所处理的特征维度之和应该等于原始的特征维度。所以将 d_model 拆分为 n_heads 个 d_k 维度的头，并且满足 d_model = n_heads * d_k，这样就能确保在拆分和后续合并的过程中，特征信息的总量保持不变。

并行计算与特征多样性：多头注意力机制的目的是让模型能够从多个不同的子空间并行地学习特征。每个头的维度 d_k 相对较小，使得每个头可以专注于捕捉输入序列的某一方面的特征，而多个头并行计算可以同时捕捉不同方面的特征，从而增加模型的表达能力。通过 n_heads 和 d_k 的合理设置，可以在保证计算效率的同时，让模型学习到更丰富的特征表示。

view：要求张量在内存中是连续存储的。当张量经过 transpose、permute 等操作后，其内存布局可能不再连续，此时若使用 view 会引发错误。

reshape：不要求张量内存连续。它会在必要时对张量进行复制操作，以确保能够正确改变形状。因此，在 transpose 操作之后，使用 reshape 更为合适。

灵活性：
view：在张量内存不连续时功能受限，需要先调用 contiguous() 方法使其连续后才能使用。
reshape：更为灵活，无需额外步骤就能处理内存不连续的情况。

repeat

repeat 方法会按照指定的倍数在各个维度上对张量进行复制。

具体来说，repeat(1, n_heads, 1, 1) 会在第 0 个维度复制 1 次（即保持不变），在第 1 个维度复制 n_heads 次，在第 2 和第 3 个维度复制 1 次（即保持不变）。
经过 repeat(1, n_heads, 1, 1) 操作后，attn_mask 的形状会从 (batch_size, 1, seq_len, seq_len) 变为 (batch_size, n_heads, seq_len, seq_len)。
4. 为什么需要这样操作
在多头注意力机制里，每个注意力头都需要使用相同的掩码来屏蔽不需要的注意力计算。通过 repeat 操作，就能把原本单头的掩码复制成适用于 n_heads 个注意力头的掩码，保证每个头都使用相同的屏蔽策略。

实例化

1. ScaledDotProductAttention 是一个类
   在 PyTorch 里，ScaledDotProductAttention 通常是一个自定义的类，用于实现缩放点积注意力机制。这个类继承自 torch.nn.Module，它会定义一些属性和方法，并且在 init 方法中初始化必要的参数。
2. ScaledDotProductAttention() 是实例化对象
   实例化过程：ScaledDotProductAttention() 这一步是对 ScaledDotProductAttention 类进行实例化，也就是创建这个类的一个对象。在 Python 中，当你调用一个类名并跟上括号时，就会调用该类的 init 方法，从而创建一个新的对象。

attention = ScaledDotProductAttention()

为什么需要实例化：实例化的目的是为了创建一个具体的对象，这个对象可以保存类中定义的属性和状态。

在深度学习里，这些属性可能包括模型的参数（如权重、偏置等），实例化后的对象可以在不同的输入上进行重复使用。
3. attention(Q, K, V, attn_mask) 是调用对象的 call 方法
call 方法：在 Python 中，当你对一个对象使用括号并传入参数时，实际上是调用了该对象的 call 方法。

对于 torch.nn.Module 的子类来说，call 方法会调用 forward 方法。所以 attention(Q, K, V, attn_mask) 实际上是调用了 ScaledDotProductAttention 类中定义的 forward 方法，用于执行具体的注意力计算。

如果写成 ScaledDotProductAttention(Q, K, V, attn_mask)，这是尝试直接调用类，而不是类的实例。这样会把 Q, K, V, attn_mask 作为参数传递给 init 方法，而不是 forward 方法，这显然不符合我们的需求，因为 init 方法通常用于初始化对象的属性，而不是执行具体的计算逻辑。

layernorm

1. nn.LayerNorm(d_model)
   nn.LayerNorm 是 PyTorch 提供的一个类，用于实现层归一化操作。层归一化是一种归一化技术，与批量归一化（Batch Normalization）不同，它是对每个样本的特征维度进行归一化，能够有效缓解深度神经网络中的梯度消失和梯度爆炸问题，同时加速模型的收敛速度。

$d\_model$ 是模型的特征维度，作为 nn.LayerNorm 的参数传入，表示要对输入张量的最后一个维度（即特征维度）进行归一化。例如，如果输入张量的形状是 [batch_size, seq_len, d_model]，那么 nn.LayerNorm(d_model) 会对每个样本的 d_model 维度进行归一化。

2. .cuda()
   .cuda() 是 PyTorch 中用于将张量或模型移动到 GPU 上进行计算的方法。在深度学习中，使用 GPU 可以显著加速计算过程。通过调用 .cuda()，nn.LayerNorm 实例会被移动到 GPU 上，后续的归一化计算也会在 GPU 上进行。如果你的代码运行在没有 GPU 的环境中，这部分可以省略。

3. (output + residual)
   output 通常是经过某个模块（如多头注意力模块）处理后的输出结果。
   residual 是残差连接中的输入，也就是在经过该模块处理之前的原始输入。残差连接是 Transformer 架构中的一个重要组件，它可以帮助模型更好地学习特征，避免梯度消失问题。具体来说，残差连接通过将模块的输入和输出相加，使得模型能够更容易地学习到恒等映射，从而提高模型的性能。
   output + residual 表示将模块的输出和原始输入相加，得到一个包含了原始信息和处理后信息的结果。

4. nn.LayerNorm(d_model).cuda()(output + residual)
   这部分代码将 output + residual 的结果作为输入，传入到 nn.LayerNorm 实例中进行层归一化操作。最终得到的是经过归一化处理后的张量。

5. , attn
   attn 是注意力权重，通常是在多头注意力机制中计算得到的。它记录了输入序列中各个位置之间的注意力分布情况。在返回值中包含 attn 可以方便后续的分析和可视化。

FeedForward Layer

class PoswiseFeedForwardNet(nn.Module):def __init__(self):super(PoswiseFeedForwardNet,self).__init__()self.fc = nn.Sequential(nn.Linear(d_model , d_ff , bias = False),nn.ReLU(),nn.Linear(d_ff , d_model , bias = False))def forward(self,inputs):'''inputs: [batch_size, seq_len, d_model]'''residual = inputsoutput = self.fc(inputs)return nn.LayerNorm(d_model).cuda()(output + residual)# [batch_size, seq_len, d_model]

Encoder Layer

class EncoderLayer(nn.Module):def __init__(self):super(EncoderLayer,self).__init__()self.enc_self_attn = MultiHeadAttention()self.pos_ffn = PoswiseFeedForwardNet()def forward(self , enc_inputs , enc_self_attn_mask):'''enc_inputs: [batch_size, src_len, d_model]enc_self_attn_mask: [batch_size, src_len, src_len]'''# enc_outputs: [batch_size, src_len, d_model]# attn: [batch_size, n_heads, src_len, src_len]enc_outputs , attn = self.enc_self_attn(enc_inputs, enc_inputs, enc_inputs, enc_self_attn_mask)# enc_inputs to same Q,K,Venc_outputs = self.pos_ffn(enc_outputs) # enc_outputs: [batch_size, src_len, d_model]return enc_outputs , attn

Encoder

class Encoder(nn.Module):def __init__(self):super(Encoder,self).__init__()self.src_emb = nn.Embedding(src_vocab_size,d_model)self.pos_emb = PositionalEncoding(d_model)self.layers = nn.ModuleList( [ EncoderLayer() for  _ in range(n_layers) ] )def forward(self , ien_inputs):'''enc_inputs: [batch_size, src_len]'''enc_outputs = self.src_emb(enc_inputs) # [batch_size, src_len, d_model]enc_outputs = self.pos_emb( enc_outputs.transpose(0,1) ).transpose(0,1)# [batch_size, src_len, d_model]enc_self_attn_mask = get_attn_pad_mask(enc_inputs,enc_inputs)# [batch_size, src_len, src_len]enc_self_attns = []for layer in self.layers:# enc_outputs: [batch_size, src_len, d_model], # enc_self_attn: [batch_size, n_heads, src_len, src_len]           enc_outputs , enc_self_attn = layer( enc_outputs , enc_self_attn_mask )enc_self_attns.append(enc_self_attn)return enc_outputs, enc_self_attns

embeding

src_vocab_size # 6
nn.Embedding(src_vocab_size,d_model)# an Embedding module containing src_vocab_size tensors of size d_model

Embedding(6, 512)

nn.Embedding(10, 3) # an Embedding module containing 10 tensors of size 3

Embedding(10, 3)

torch.LongTensor([[1,2,4,5],[4,3,2,9]]) # a batch of 2 samples of 4 索引 each
# 2*4

tensor([[1, 2, 4, 5],[4, 3, 2, 9]])

nn.Embedding(10, 3)(torch.LongTensor([[1,2,4,5],[4,3,2,9]]))
# 2-4-3
# 所以说，我们待输入的张量[[1,2,4,5],[4,3,2,9]]，在经过nn.embedding后，
# 从[2, 4]维度变换为[2, 4, 3]，其实就是[2, 4]中的每个值作为索引去nn.embedding中取对应的权重。# '''
# 两个样本，emd矩阵10*3，每次个样本每个索引从权重索引出一个1*3隐藏
# '''

tensor([[[-0.8152, -1.6954,  0.7013],[-0.9970, -0.2400,  1.3145],[ 1.0676,  0.3986, -0.6844],[ 0.8887,  0.9975, -0.9938]],[[ 1.0676,  0.3986, -0.6844],[-0.0193, -1.0218,  0.2434],[-0.9970, -0.2400,  1.3145],[-1.2412,  0.2199,  0.4242]]], grad_fn=<EmbeddingBackward0>)

 # embedding.weight  (10-3)
'''根据原始词库得到的索引
'''

'\n   根据原始词库得到的索引\n'

nn.Embedding(4, 3)

Embedding(4, 3)

nn.Embedding(4, 3).weight

Parameter containing:
tensor([[ 0.2513,  0.8116,  1.2485],[ 0.1551,  0.8137,  0.4109],[-0.3992, -1.2102, -0.9165],[-1.2512,  1.1654, -0.1464]], requires_grad=True)

ModuleList

2. [EncoderLayer() for _ in range(n_layers)]
   这是一个列表推导式，其作用是创建一个包含 n_layers 个 EncoderLayer 实例的列表。range(n_layers) 会生成一个从 0 到 n_layers - 1 的整数序列，_ 是一个占位符，表示不使用这个循环变量。每次循环都会创建一个新的 EncoderLayer 实例，最终将这些实例存储在一个列表中。
3. nn.ModuleList()
   nn.ModuleList 是 PyTorch 提供的一个容器类，用于存储多个 nn.Module 子类的实例。与普通的 Python 列表不同，nn.ModuleList 会将其中的模块注册为当前模块的子模块，这样 PyTorch 就能正确地管理这些模块的参数，例如在调用 model.parameters() 时会包含 nn.ModuleList 中所有模块的参数，并且在将模型移动到 GPU 上时，nn.ModuleList 中的所有模块也会被正确地移动。

Decoder Layer

class DecoderLayer(nn.Module):def __init__(self):super(DecoderLayer,self).__init__()self.dec_self_attn = MultiHeadAttention()self.dec_enc_attn = MultiHeadAttention()self.pos_ffn = PoswiseFeedForwardNet()def forward( self , dec_inputs , enc_outputs , dec_self_attn_mask ,  dec_enc_attn_mask):'''dec_inputs: [batch_size, tgt_len, d_model]enc_outputs: [batch_size, src_len, d_model]dec_self_attn_mask: [batch_size, tgt_len, tgt_len]dec_enc_attn_mask: [batch_size, tgt_len, src_len]'''# dec_outputs: [batch_size, tgt_len, d_model]# dec_self_attn: [batch_size, n_heads, tgt_len, tgt_len]dec_outputs , dec_self_attn = self.dec_self_attn( dec_inputs, dec_inputs, dec_inputs, dec_self_attn_mask )# dec_outputs: [batch_size, tgt_len, d_model]# dec_enc_attn: [batch_size, h_heads, tgt_len, src_len]dec_outputs , dec_enc_attn = self.dec_enc_attn( dec_outputs, enc_outputs, enc_outputs, dec_enc_attn_mask )dec_outputs = self.pos_ffn(dec_outputs) # [batch_size, tgt_len, d_model]return dec_outputs, dec_self_attn, dec_enc_attn

在 Decoder Layer 中会调用两次 MultiHeadAttention，第一次是计算 Decoder Input 的 self-attention，得到输出 dec_outputs。

然后将 dec_outputs 作为生成 Q 的元素，enc_outputs 作为生成 K 和 V 的元素，

再调用一次 MultiHeadAttention，得到的是 Encoder 和 Decoder Layer 之间的 context vector。最后将 dec_outptus 做一次维度变换，然后返回

Decoder

class Decoder(nn.Module):def __init__(self):super(Decoder,self).__init__()self.tgt_emb = nn.Embedding(tgt_vocab_size, d_model)self.pos_emb = PositionalEncoding(d_model)self.layers = nn.ModuleList( [ DecoderLayer() for _ in range(n_layers) ] )def forward(self , dec_inputs , enc_inputs , enc_outputs):'''dec_inputs: [batch_size, tgt_len]enc_intpus: [batch_size, src_len]enc_outputs: [batch_size, src_len, d_model]'''dec_outputs = self.tgt_emb(dec_inputs)# [batch_size, tgt_len, d_model]dec_outputs = self.pos_emb(dec_outputs.transpose(0, 1)).transpose(0, 1).cuda() # [batch_size, tgt_len, d_model]dec_self_attn_pad_mask = get_attn_pad_mask(dec_inputs, dec_inputs).cuda() # [batch_size, tgt_len, tgt_len]dec_self_attn_subsequence_mask = get_attn_subsequence_mask(dec_inputs).cuda() # [batch_size, tgt_len, tgt_len]dec_self_attn_mask = torch.gt((dec_self_attn_pad_mask + dec_self_attn_subsequence_mask), 0).cuda() # [batch_size, tgt_len, tgt_len]dec_enc_attn_mask = get_attn_pad_mask(dec_inputs, enc_inputs) # [batc_size, tgt_len, src_len]dec_self_attns, dec_enc_attns = [], []for layer in self.layers:# dec_outputs: [batch_size, tgt_len, d_model],# dec_self_attn: [batch_size, n_heads, tgt_len, tgt_len]# dec_enc_attn: [batch_size, h_heads, tgt_len, src_len]dec_outputs, dec_self_attn, dec_enc_attn = layer(dec_outputs, enc_outputs, dec_self_attn_mask, dec_enc_attn_mask)dec_self_attns.append(dec_self_attn)dec_enc_attns.append(dec_enc_attn)return dec_outputs, dec_self_attns, dec_enc_attns

torch.gt

torch.gt 是 PyTorch 中的一个函数，用于比较张量中的每个元素是否大于给定的值（这里是 0）。如果元素大于 0，则返回 True，否则返回 False。通过这个操作，将相加后的掩码转换为一个布尔型张量，其中 True 表示需要屏蔽的位置，False 表示不需要屏蔽的位置。

torch.gt(a, value) 的意思是，将 a 中各个位置上的元素和 value 比较，若大于 value，则该位置取 1，否则取 0torch.gt(a, value) 的意思是，将 a 中各个位置上的元素和 value 比较，若大于 value，则该位置取 1，否则取 0

Transformer

class Transformer(nn.Module):def __init__(self):super(Transformer,self).__init__()self.encoder = Encoder().cuda()self.decoder = Decoder().cuda()self.projection = nn.Linear( d_model , tgt_vocab_size , bias = False ).cuda()def forward(self, enc_inputs, dec_inputs):'''enc_inputs: [batch_size, src_len]dec_inputs: [batch_size, tgt_len]'''# tensor to store decoder outputs# outputs = torch.zeros(batch_size, tgt_len, tgt_vocab_size).to(self.device)# enc_outputs: [batch_size, src_len, d_model]# enc_self_attns: [n_layers, batch_size, n_heads, src_len, src_len]enc_outputs, enc_self_attns = self.encoder(enc_inputs)# dec_outpus: [batch_size, tgt_len, d_model]# dec_self_attns: [n_layers, batch_size, n_heads, tgt_len, tgt_len]# dec_enc_attn: [n_layers, batch_size, tgt_len, src_len]dec_outputs, dec_self_attns, dec_enc_attns = self.decoder(dec_inputs, enc_inputs, enc_outputs)dec_logits = self.projection(dec_outputs) # dec_logits: [batch_size, tgt_len, tgt_vocab_size]return dec_logits.view(-1, dec_logits.size(-1)), enc_self_attns, dec_self_attns, dec_enc_attns

最后返回 dec_logits 的维度是 [batch_size * tgt_len, tgt_vocab_size]，可以理解为，一个句子，这个句子有 batch_size*tgt_len 个单词，每个单词有 tgt_vocab_size 种情况，取概率最大者

train

model = Transformer().cuda()
criterion = nn.CrossEntropyLoss(ignore_index=0)
optimizer = optim.SGD(model.parameters(), lr=1e-3, momentum=0.99)

enc_inputs

tensor([[1, 2, 3, 4, 0]], device='cuda:0')

for epoch in range(30):for enc_inputs, dec_inputs, dec_outputs in loader:'''enc_inputs: [batch_size, src_len]dec_inputs: [batch_size, tgt_len]dec_outputs: [batch_size, tgt_len]'''enc_inputs, dec_inputs, dec_outputs = enc_inputs.cuda(), dec_inputs.cuda(), dec_outputs.cuda()# outputs: [batch_size * tgt_len, tgt_vocab_size]outputs, enc_self_attns, dec_self_attns, dec_enc_attns = model(enc_inputs, dec_inputs)loss = criterion(outputs, dec_outputs.view(-1))print('Epoch:', '%04d' % (epoch + 1), 'loss =', '{:.6f}'.format(loss))optimizer.zero_grad()loss.backward()optimizer.step()

Epoch: 0001 loss = 2.828818
Epoch: 0002 loss = 2.544355
Epoch: 0003 loss = 2.170896
Epoch: 0004 loss = 1.899252
Epoch: 0005 loss = 1.507468
Epoch: 0006 loss = 1.345472
Epoch: 0007 loss = 1.184967
Epoch: 0008 loss = 0.809456
Epoch: 0009 loss = 0.601722
Epoch: 0010 loss = 0.455130
Epoch: 0011 loss = 0.329263
Epoch: 0012 loss = 0.230903
Epoch: 0013 loss = 0.156530
Epoch: 0014 loss = 0.141345
Epoch: 0015 loss = 0.087132
Epoch: 0016 loss = 0.060078
Epoch: 0017 loss = 0.050748
Epoch: 0018 loss = 0.033932
Epoch: 0019 loss = 0.019920
Epoch: 0020 loss = 0.011308
Epoch: 0021 loss = 0.007204
Epoch: 0022 loss = 0.006975
Epoch: 0023 loss = 0.003768
Epoch: 0024 loss = 0.002927
Epoch: 0025 loss = 0.001546
Epoch: 0026 loss = 0.001353
Epoch: 0027 loss = 0.001144
Epoch: 0028 loss = 0.000830
Epoch: 0029 loss = 0.000720
Epoch: 0030 loss = 0.000649

test

def greedy_decoder(model, enc_input, start_symbol):"""For simplicity, a Greedy Decoder is Beam search when K=1. This is necessary for inference as we don't know thetarget sequence input. Therefore we try to generate the target input word by word, then feed it into the transformer.Starting Reference: http://nlp.seas.harvard.edu/2018/04/03/attention.html#greedy-decoding:param model: Transformer Model:param enc_input: The encoder input:param start_symbol: The start symbol. In this example it is 'S' which corresponds to index 4:return: The target input"""enc_outputs, enc_self_attns = model.encoder(enc_input)dec_input = torch.zeros(1, 0).type_as(enc_input.data)terminal = Falsenext_symbol = start_symbolwhile not terminal:         dec_input = torch.cat([dec_input.detach(),torch.tensor([[next_symbol]],dtype=enc_input.dtype).cuda()],-1)dec_outputs, _, _ = model.decoder(dec_input, enc_input, enc_outputs)projected = model.projection(dec_outputs)prob = projected.squeeze(0).max(dim=-1, keepdim=False)[1]next_word = prob.data[-1]next_symbol = next_wordif next_symbol == tgt_vocab["."]:terminal = Trueprint(next_word)            return dec_input# Test
enc_inputs, _, _ = next(iter(loader))
enc_inputs = enc_inputs.cuda()
for i in range(len(enc_inputs)):greedy_dec_input = greedy_decoder(model, enc_inputs[i].view(1, -1), start_symbol=tgt_vocab["S"])predict, _, _, _ = model(enc_inputs[i].view(1, -1), greedy_dec_input)predict = predict.data.max(1, keepdim=True)[1]print(enc_inputs[i], '->', [idx2word[n.item()] for n in predict.squeeze()])

tensor(1, device='cuda:0')
tensor(2, device='cuda:0')
tensor(3, device='cuda:0')
tensor(4, device='cuda:0')
tensor(8, device='cuda:0')
tensor([1, 2, 3, 4, 0], device='cuda:0') -> ['i', 'want', 'a', 'beer', '.']