🫐Python发票合同 | 解缠注意力语言模型

Python | 语言 | 儿童发展 | 常识 | 生物 | 物理 | 偏见 | 基准 | 注意力 | 行为 | 知识图 | 多标签 | 多模态 | 数据点 | 生物医学 | 商业文本 | 对话 | 混合密集 | 稀疏表征 | GPU | MATLAB | C++ | 基因 | 蛋白质 | 图像合成

🏈指点迷津 | Brief

🎯要点

🎯语言模型可解释和可视化：Python记忆组合透明度语言模型

🍇Python机器翻译和摘要中序列到序列任务

序列到序列任务表示输入和输出是序列（不一定具有相同长度）的任务。该领域的热门任务包括机器翻译和摘要。为此，我们通常有一个 Transformer 编码器用于解释输入序列，以及一个解码器用于以自回归方式生成输出。然而，在这里，我们将回到一个更简单的示例任务并仅使用编码器。给定 0 到 M 之间的 N 数字序列，任务是反转输入序列。在 Numpy 表示法中，如果我们的输入是 x，则输出应该是 x[::-1]。尽管这个任务听起来非常简单，但递归神经网络可能会遇到问题，因为该任务需要长期依赖。 Transformer 就是为了支持这种功能而构建的，因此，我们希望它能够表现良好。

首先，让我们在下面创建一个数据集类。

 class ReverseDataset(data.Dataset):
 
     def __init__(self, num_categories, seq_len, size):
         super().__init__()
         self.num_categories = num_categories
         self.seq_len = seq_len
         self.size = size
 
         self.data = torch.randint(self.num_categories, size=(self.size, self.seq_len))
 
     def __len__(self):
         return self.size
 
     def __getitem__(self, idx):
         inp_data = self.data[idx]
         labels = torch.flip(inp_data, dims=(0,))
         return inp_data, labels

我们创建 0 到 num_categories-1 之间任意数量的随机数字序列。标签只是在序列维度上翻转的张量。我们可以在下面创建相应的数据加载器。

 dataset = partial(ReverseDataset, 10, 16)
 train_loader = data.DataLoader(dataset(50000), batch_size=128, shuffle=True, drop_last=True, pin_memory=True)
 val_loader   = data.DataLoader(dataset(1000), batch_size=128)
 test_loader  = data.DataLoader(dataset(10000), batch_size=128)

让我们看一下数据集的任意样本：

 inp_data, labels = train_loader.dataset[0]
 print("Input data:", inp_data)
 print("Labels:    ", labels)

 Input data: tensor([9, 6, 2, 0, 6, 2, 7, 9, 7, 3, 3, 4, 3, 7, 0, 9])
 Labels:     tensor([9, 0, 7, 3, 4, 3, 3, 7, 9, 7, 2, 6, 0, 2, 6, 9])

在训练期间，我们将输入序列传递给 Transformer 编码器，并预测每个输入标记的输出。我们使用标准交叉熵损失来执行此操作。每个数字都表示为一个独热向量。请记住，将类别表示为单个标量会极大地降低模型的表达能力，因为在我们的示例中和并不比和更紧密相关。独热向量的替代方法是使用学习到的嵌入向量，因为它由 PyTorch 模块 nn.Embedding 提供。但是，像我们的例子一样使用带有附加线性层的独热向量具有与嵌入层相同的效果（self.input_net 将独热向量映射到密集向量，其中权重矩阵的每一行代表特定类别的嵌入）。

为了实现动态训练，我们创建一个继承自 TransformerPredictor 的新类，并覆盖训练、验证和测试步骤函数。

 class ReversePredictor(TransformerPredictor):
 
     def _calculate_loss(self, batch, mode="train"):
 
         inp_data, labels = batch
         inp_data = F.one_hot(inp_data, num_classes=self.hparams.num_classes).float()
 
         preds = self.forward(inp_data, add_positional_encoding=True)
         loss = F.cross_entropy(preds.view(-1,preds.size(-1)), labels.view(-1))
         acc = (preds.argmax(dim=-1) == labels).float().mean()
 
         self.log(f"{mode}_loss", loss)
         self.log(f"{mode}_acc", acc)
         return loss, acc
 
     def training_step(self, batch, batch_idx):
         loss, _ = self._calculate_loss(batch, mode="train")
         return loss
 
     def validation_step(self, batch, batch_idx):
         _ = self._calculate_loss(batch, mode="val")
 
     def test_step(self, batch, batch_idx):
         _ = self._calculate_loss(batch, mode="test")

最后，我们可以创建一个类似于我们在 PyTorch Lightning 中看到的训练函数。我们创建一个 pl.Trainer 对象，运行几个时期，登录 TensorBoard，并根据验证保存我们的最佳模型。之后，我们在测试集上测试我们的模型。我们在这里传递给训练器的另一个参数是 gradient_clip_val。这会在采取优化器步骤之前剪切所有参数的梯度范数，并防止模型在例如尖锐损失表面处获得非常高的梯度时发散。对于 Transformers，梯度剪切可以帮助在前几次迭代期间以及之后进一步稳定训练。在普通的 PyTorch 中，您可以通过 torch.nn.utils.clip_grad_norm_(...) 应用梯度剪切。剪辑值通常在 0.5 到 10 之间，具体取决于您想要剪辑大梯度的程度。解释完这些之后，我们来实现训练功能：

 def train_reverse(**kwargs):
 
     root_dir = os.path.join(CHECKPOINT_PATH, "ReverseTask")
     os.makedirs(root_dir, exist_ok=True)
     trainer = pl.Trainer(default_root_dir=root_dir,
                          callbacks=[ModelCheckpoint(save_weights_only=True, mode="max", monitor="val_acc")],
                          accelerator="gpu" if str(device).startswith("cuda") else "cpu",
                          devices=1,
                          max_epochs=10,
                          gradient_clip_val=5)
     trainer.logger._default_hp_metric = None 
 
 
     pretrained_filename = os.path.join(CHECKPOINT_PATH, "ReverseTask.ckpt")
     if os.path.isfile(pretrained_filename):
         print("Found pretrained model, loading...")
         model = ReversePredictor.load_from_checkpoint(pretrained_filename)
     else:
         model = ReversePredictor(max_iters=trainer.max_epochs*len(train_loader), **kwargs)
         trainer.fit(model, train_loader, val_loader)
 
 
     val_result = trainer.test(model, val_loader, verbose=False)
     test_result = trainer.test(model, test_loader, verbose=False)
     result = {"test_acc": test_result[0]["test_acc"], "val_acc": val_result[0]["test_acc"]}
 
     model = model.to(device)
     return model, result

最后，我们可以训练模型了。在这个设置中，我们将在多头注意力中使用单个编码器块和单个头。之所以选择这种方式，是因为任务比较简单，在这种情况下，注意力实际上可以被解释为预测的“解释”。

 reverse_model, reverse_result = train_reverse(input_dim=train_loader.dataset.num_categories,
                                               model_dim=32,
                                               num_heads=1,
                                               num_classes=train_loader.dataset.num_categories,
                                               num_layers=1,
                                               dropout=0.0,
                                               lr=5e-4,
                                               warmup=50)
 GPU available: True, used: True
 TPU available: False, using: 0 TPU cores
 LOCAL_RANK: 0 - CUDA_VISIBLE_DEVICES: [0]

找到预训练模型，正在加载...

 /home/anaconda3/envs/nlp1/lib/python3.7/site-packages/pytorch_lightning/utilities/distributed.py:45: UserWarning: The dataloader, test dataloader 0, does not have many workers which may be a bottleneck. Consider increasing the value of the `num_workers` argument` (try 16 which is the number of cpus on this machine) in the `DataLoader` init to improve performance.
   warnings.warn(*args, **kwargs)

PyTorch Lightning 关于 worker 数量的警告暂时可以忽略。由于数据集非常简单，并且 __getitem__ 的完成时间可以忽略不计，因此我们不需要子进程来为我们提供数据（事实上，更多的 worker 可能会减慢训练速度，因为进程之间存在通信开销）。

 print(f"Val accuracy:  {(100.0 * reverse_result['val_acc']):4.2f}%")
 print(f"Test accuracy: {(100.0 * reverse_result['test_acc']):4.2f}%")

 Val accuracy:  100.00%
 Test accuracy: 100.00%

正如我们所期望的，Transformer 可以正确地解决任务。然而，对于任意输入，多头注意力模块中的注意力看起来如何？让我们尝试在下面可视化它。

 data_input, labels = next(iter(val_loader))
 inp_data = F.one_hot(data_input, num_classes=reverse_model.hparams.num_classes).float()
 inp_data = inp_data.to(device)
 attention_maps = reverse_model.get_attention_maps(inp_data)

对象attention_maps是一个长度列表 𝑁 在哪里 𝑁 是层数。每个元素都是一个形状为 [Batch, Heads, SeqLen, SeqLen] 的张量，我们可以在下面验证。

 attention_maps[0].shape

 torch.Size([128, 1, 16, 16])

接下来，我们将编写一个绘图函数，该函数将序列、注意力图和一个索引作为输入，该索引指示我们想要可视化哪个批次元素的注意力图。我们将创建一个图，其中在行上有不同的层，而在列上显示不同的头部。请记住，softmax 已分别应用于每一行。

 def plot_attention_maps(input_data, attn_maps, idx=0):
     if input_data is not None:
         input_data = input_data[idx].detach().cpu().numpy()
     else:
         input_data = np.arange(attn_maps[0][idx].shape[-1])
     attn_maps = [m[idx].detach().cpu().numpy() for m in attn_maps]
 
     num_heads = attn_maps[0].shape[0]
     num_layers = len(attn_maps)
     seq_len = input_data.shape[0]
     fig_size = 4 if num_heads == 1 else 3
     fig, ax = plt.subplots(num_layers, num_heads, figsize=(num_heads*fig_size, num_layers*fig_size))
     if num_layers == 1:
         ax = [ax]
     if num_heads == 1:
         ax = [[a] for a in ax]
     for row in range(num_layers):
         for column in range(num_heads):
             ax[row][column].imshow(attn_maps[row][column], origin='lower', vmin=0)
             ax[row][column].set_xticks(list(range(seq_len)))
             ax[row][column].set_xticklabels(input_data.tolist())
             ax[row][column].set_yticks(list(range(seq_len)))
             ax[row][column].set_yticklabels(input_data.tolist())
             ax[row][column].set_title(f"Layer {row+1}, Head {column+1}")
     fig.subplots_adjust(hspace=0.5)
     plt.show()

最后，我们可以绘制训练好的 Transformer 在反向任务上的注意力图：

 plot_attention_maps(data_input, attention_maps, idx=0)

该模型已经学会了关注位于其自身翻转索引上的标记。因此，它实际上做了我们想要它做的事情。然而，我们看到它也会关注靠近翻转索引的值。这是因为模型不需要完美、严格的注意力来解决这个问题，但可以接受这种近似、嘈杂的注意力图。接近的索引是由位置编码的相似性引起的，这也是我们用位置编码所期望的。

PreviousPython俄罗斯方块可操纵卷积分类 | 稀疏辨识算法 | 微分方程神经求解器 NextPython记忆组合透明度语言模型

Last updated 5 months ago

class ReverseDataset(data.Dataset): def __init__(self, num_categories, seq_len, size): super().__init__() self.num_categories = num_categories self.seq_len = seq_len self.size = size self.data = torch.randint(self.num_categories, size=(self.size, self.seq_len)) def __len__(self): return self.size def __getitem__(self, idx): inp_data = self.data[idx] labels = torch.flip(inp_data, dims=(0,)) return inp_data, labels

dataset = partial(ReverseDataset, 10, 16) train_loader = data.DataLoader(dataset(50000), batch_size=128, shuffle=True, drop_last=True, pin_memory=True) val_loader = data.DataLoader(dataset(1000), batch_size=128) test_loader = data.DataLoader(dataset(10000), batch_size=128)

class ReversePredictor(TransformerPredictor): def _calculate_loss(self, batch, mode="train"): inp_data, labels = batch inp_data = F.one_hot(inp_data, num_classes=self.hparams.num_classes).float() preds = self.forward(inp_data, add_positional_encoding=True) loss = F.cross_entropy(preds.view(-1,preds.size(-1)), labels.view(-1)) acc = (preds.argmax(dim=-1) == labels).float().mean() self.log(f"{mode}_loss", loss) self.log(f"{mode}_acc", acc) return loss, acc def training_step(self, batch, batch_idx): loss, _ = self._calculate_loss(batch, mode="train") return loss def validation_step(self, batch, batch_idx): _ = self._calculate_loss(batch, mode="val") def test_step(self, batch, batch_idx): _ = self._calculate_loss(batch, mode="test")

def train_reverse(**kwargs): root_dir = os.path.join(CHECKPOINT_PATH, "ReverseTask") os.makedirs(root_dir, exist_ok=True) trainer = pl.Trainer(default_root_dir=root_dir, callbacks=[ModelCheckpoint(save_weights_only=True, mode="max", monitor="val_acc")], accelerator="gpu" if str(device).startswith("cuda") else "cpu", devices=1, max_epochs=10, gradient_clip_val=5) trainer.logger._default_hp_metric = None pretrained_filename = os.path.join(CHECKPOINT_PATH, "ReverseTask.ckpt") if os.path.isfile(pretrained_filename): print("Found pretrained model, loading...") model = ReversePredictor.load_from_checkpoint(pretrained_filename) else: model = ReversePredictor(max_iters=trainer.max_epochs*len(train_loader), **kwargs) trainer.fit(model, train_loader, val_loader) val_result = trainer.test(model, val_loader, verbose=False) test_result = trainer.test(model, test_loader, verbose=False) result = {"test_acc": test_result[0]["test_acc"], "val_acc": val_result[0]["test_acc"]} model = model.to(device) return model, result

reverse_model, reverse_result = train_reverse(input_dim=train_loader.dataset.num_categories, model_dim=32, num_heads=1, num_classes=train_loader.dataset.num_categories, num_layers=1, dropout=0.0, lr=5e-4, warmup=50) GPU available: True, used: True TPU available: False, using: 0 TPU cores LOCAL_RANK: 0 - CUDA_VISIBLE_DEVICES: [0]

/home/anaconda3/envs/nlp1/lib/python3.7/site-packages/pytorch_lightning/utilities/distributed.py:45: UserWarning: The dataloader, test dataloader 0, does not have many workers which may be a bottleneck. Consider increasing the value of the `num_workers` argument` (try 16 which is the number of cpus on this machine) in the `DataLoader` init to improve performance. warnings.warn(*args, **kwargs)

data_input, labels = next(iter(val_loader)) inp_data = F.one_hot(data_input, num_classes=reverse_model.hparams.num_classes).float() inp_data = inp_data.to(device) attention_maps = reverse_model.get_attention_maps(inp_data)

def plot_attention_maps(input_data, attn_maps, idx=0): if input_data is not None: input_data = input_data[idx].detach().cpu().numpy() else: input_data = np.arange(attn_maps[0][idx].shape[-1]) attn_maps = [m[idx].detach().cpu().numpy() for m in attn_maps] num_heads = attn_maps[0].shape[0] num_layers = len(attn_maps) seq_len = input_data.shape[0] fig_size = 4 if num_heads == 1 else 3 fig, ax = plt.subplots(num_layers, num_heads, figsize=(num_heads*fig_size, num_layers*fig_size)) if num_layers == 1: ax = [ax] if num_heads == 1: ax = [[a] for a in ax] for row in range(num_layers): for column in range(num_heads): ax[row][column].imshow(attn_maps[row][column], origin='lower', vmin=0) ax[row][column].set_xticks(list(range(seq_len))) ax[row][column].set_xticklabels(input_data.tolist()) ax[row][column].set_yticks(list(range(seq_len))) ax[row][column].set_yticklabels(input_data.tolist()) ax[row][column].set_title(f"Layer {row+1}, Head {column+1}") fig.subplots_adjust(hspace=0.5) plt.show()