Pytorch

在前面的Pytorch库的介绍中，我们主要是了解诶Pytorch的整体知识。在这里我们强调在深度学习中的Pytorch的基本功，比如常用函数、常见代码、网络层搭建等

常用函数

• 使用Hook函数获取梯度、修改梯度

import torch
x = torch.tensor([1.0, 2.0, 3.0], requires_grad=True)

def print_hook(grad):
    print(f"Gradient: {grad}")

hook_handle = x.register_hook(print_hook)
y = x * 2
z = y.sum()
z.backward()

# 输出：
# Gradient: tensor([2., 2., 2.])

• 从分布中采样/上采样、下采样

# 从均值为 mean，方差为 var 的高斯分布中采样
mean + (var ** 0.5) * eps

# 上采样和下采样（unet）
self.conv = nn.ConvTranspose2d(n_channels, n_channels, (4, 4), (2, 2), (1, 1))
self.conv = nn.Conv2d(n_channels, n_channels, (3, 3), (2, 2), (1, 1))

常见结构

Multi-Headed Attention

Code from : Attention Is All You Need

下面介绍的是上三角Mask和填充Mask！

# 输入：[seq_len_q, seq_len_k, batch_size]
# 输出：[seq_len_q, seq_len_k, batch_size, heads]

def prepare_mask(self, mask: torch.Tensor, query_shape: List[int], key_shape:List[int]):
    assert ...
    mask = mask.unsqueeze(-1)
    return mask

# 上三角Mask
def generate_upper_triangular_mask(seq_len, batch_size):
    # 生成一个 [seq_len, seq_len] 的上三角矩阵
    # diagonal=1 确保对角线上的元素为零
    mask = torch.triu(torch.ones(seq_len, seq_len), diagonal=1)
    # 扩展到 [seq_len, seq_len, batch_size]
    mask = mask.unsqueeze(-1).expand(seq_len, seq_len, batch_size)
    return mask

# 填充Mask
def generate_padding_mask(padded_sequences, pad_token=0):

    seq_len, batch_size = padded_sequences.shape
    mask = (padded_sequences == pad_token).unsqueeze(0).expand(seq_len, seq_len, batch_size)

    return mask

在多头注意力机制中，我们将输入分成多个头（heads），每个头独立地执行注意力机制，然后将结果拼接在一起。这使模型可以关注输入的不同部分或特征。整体的代码结构如下所示：

# 输入：qkv [seq_len, batch_size, d_model]

def forward(self, *, query: torch.Tensor, key: torch.Tensor, value: torch.Tensor,  mask: Optional[torch.Tensor] = None):

    seq_len, batch_size, _ = query.shape
    if mask is not None:
        mask = self.prepare_mask(mask, query.shape, key.shape)

    # [seq_len, batch_size, heads, d_k]
    query = self.query(query)
    key = self.key(key)
    value = self.value(value)

    scores = self.get_scores(query, key)
    scores *= self.scale
    if mask is not None:
        scores = scores.masked_fill(mask, float('-inf'))
    attn = self.softmax(scores)
    tracker.debug('attn', attn)
    attn = self.dropout(attn)
    x = torch.einsum("ijbh,jbhd->ibhd", attn, value) # attn * value
    self.attn = attn.detach()

    # [seq_len, batch_size, heads * d_k]
    x = x.reshape(seq_len, batch_size, -1)
    return self.output(x)

Dropout层

[1] 在训练期间随机将一部分神经元的输出设置为零，以此来破坏神经元之间可能出现的依赖关系，从而增强模型的泛化能力。

[2] 为了保持输入的期望值不变，剩余未置零的神经元的输出会除以 1 - dropout_prob，以进行尺度调整。例如，如果 dropout_prob=0.5，那么剩下的神经元的输出将乘以 2。

[3] 在模型评估或推理过程中，Dropout 层不进行任何操作，所有神经元的输出都保持不变。这保证了在推理阶段模型的输出稳定且可预测。

class MyDropout(nn.Module):
    def __init__(self, p=0.5):
        super(MyDropout, self).__init__()
        self.p = p

    def forward(self, x):
        if self.training:
            mask = (torch.rand_like(x) > self.p).float()
            return x * mask / (1 - self.p)
        else:
            return x

BN层 (affine + momentum)

使用动量更新的方式可以平滑地估计整个训练数据集的均值和方差，而不仅仅依赖单个批次的统计信息。这样可以减少由于批次间差异导致的不稳定性。

class BatchNorm(nn.Moudle):
    def __init__(self, num_features, eps=1e-5, momentum=0.1, affine=True):
        super(BatchNorm, self).__init__()
        self.num_features = num_features
        self.eps = eps
        self.momentum = momentum

        if self.affine:
            # 缩放参数 & 偏移参数
            self.gamma = nn.Parameter(torch.ones(num_features))
            self.beta = nn.Parameter(torch.zeros(num_features))
        else :
            self.register_parameter('gamma', None)
            self.register_parameter('beta', None)

        self.register_buffer('running_mean', torch.zeros(num_features))
        self.register_buffer('running_var', torch.ones(num_features))

    def forward(self, x):
        if self.training:
            mean = x.mean(dim=0)
            var = x.var(dim=0)

            self.running_mean = self.momentum * self.running_mean + (1 - self.momentum) * mean
            self.running_var = self.momentum * self.running_var + (1 - self.momentum) * var

            x = (x - mean) / torch.sqrt(var + self.eps)
        else:
            x = (x - self.running_mean) / torch.sqrt(self.running_var + self.eps)

        if self.affine:
            x = x * self.gamma + self.beta
        return x

知识积累

• 随机张量生成

torch.rand，生成一个均匀分布的随机数张量，随机数的范围在 [0, 1) 之间。

torch.randn，生成一个正态分布（标准正态分布，均值为 0，标准差为 1）的随机数张量。

• NLP和CV维度信息（for transformer）

• 参数传递规范化

# 在Python函数定义中，使用*是为了指示从此位置开始，
# 后面的参数必须通过关键字参数（keyword arguments）来传递，
# 而不能通过位置参数（positional arguments）来传递。
def forward(self, *, query: torch.Tensor, key: torch.Tensor):

• 目标函数中的正则化项的计算

# L2 正则化项
l2_reg = torch.tensor(0.0)
for param in model.parameters():
    l2_reg += torch.norm(param, 2) ** 2
loss += lambda_l2 * l2_reg

# L1 正则化项
l1_reg = torch.tensor(0.0)
for param in model.parameters():
    l1_reg += torch.norm(param, 1)
loss += lambda_l1 * l1_reg