如何向量化PyTorch自定义Channelwise_Linear层的forward方法?
类卷积全连接层的并行优化方案
我正在构建一种类2D卷积层的新型全连接层,每个输出通道对应输入通道全连接层输出的总和。目前已有可运行的原型,但forward方法里的嵌套for循环导致运行速度很慢,理论上这些操作都可以并行执行,请问该怎么实现并行优化?
原实现代码:
# 输入形状为 (BATCH X channels X feature_dim) class Channelwise_Linear(nn.Module): def __init__(self, in_channels, out_channels, in_feature_dim, out_feature_dim, drop_rate): super(Channelwise_Linear, self).__init__() self.in_channels = in_channels self.out_channels = out_channels self.in_feature_dim = in_feature_dim self.out_feature_dim = out_feature_dim self.linear_layers = [] self.bn_layers = [] self.activation_layers = [] self.do_layers = [] for out_channel in range(out_channels): temp_linear = nn.ModuleList() temp_bn = nn.ModuleList() temp_AL = nn.ModuleList() temp_do = nn.ModuleList() for in_channel in range(in_channels): temp_linear.append(nn.Linear(in_feature_dim, out_feature_dim)) temp_bn.append(nn.BatchNorm1d(out_feature_dim)) temp_AL.append(nn.LeakyReLU()) temp_do.append(nn.Dropout1d(drop_rate)) self.linear_layers.append(temp_linear) self.bn_layers.append(temp_bn) self.activation_layers.append(temp_AL) self.do_layers.append(temp_do) def forward(self, x): x = torch.transpose(x,0,1) # 现在通道维度在前 output_list = [] for out_channel in range(self.out_channels): temp = [] for in_channel in range(self.in_channels): intermediate = self.linear_layers[out_channel][in_channel](x[in_channel]) intermediate = self.bn_layers[out_channel][in_channel](intermediate) intermediate = self.activation_layers[out_channel][in_channel](intermediate) intermediate = self.do_layers[out_channel][in_channel](intermediate) temp.append(intermediate) temp = torch.sum(torch.stack(temp), axis=0) # 每个输出通道是对应输入通道全连接输出的和,类似卷积 output_list.append(temp) return torch.transpose(torch.stack(output_list), 0, 1) # 把batch维度换回第一维
优化思路
原代码的核心问题是通过嵌套ModuleList拆分了所有层,导致必须用循环逐个处理通道对,无法利用PyTorch的批量张量运算能力。优化方向是将分散的层参数合并为大张量,通过批量矩阵运算、归一化操作替代循环,充分利用GPU并行计算。
优化后代码
import torch import torch.nn as nn # 输入形状为 (BATCH X channels X feature_dim) class ChannelwiseLinear(nn.Module): def __init__(self, in_channels, out_channels, in_feature_dim, out_feature_dim, drop_rate): super().__init__() self.in_channels = in_channels self.out_channels = out_channels self.in_feature_dim = in_feature_dim self.out_feature_dim = out_feature_dim # 合并所有线性层参数:(输出通道数, 输入通道数, 输出特征维度, 输入特征维度) self.weight = nn.Parameter(torch.randn(out_channels, in_channels, out_feature_dim, in_feature_dim)) # 合并所有偏置参数:(输出通道数, 输入通道数, 输出特征维度) self.bias = nn.Parameter(torch.randn(out_channels, in_channels, out_feature_dim)) # 用GroupNorm实现每个(输出通道,输入通道)对的独立归一化,组数等于通道对总数 self.bn = nn.GroupNorm( num_groups=out_channels * in_channels, num_channels=out_feature_dim ) self.activation = nn.LeakyReLU() self.dropout = nn.Dropout1d(drop_rate) def forward(self, x): # x 初始形状: (BATCH, 输入通道数, 输入特征维度) batch_size = x.shape[0] # 用einsum一次性完成所有通道对的线性变换 # 运算逻辑:每个batch的每个输入通道特征,和对应输出通道的线性层权重做矩阵乘法,再加上偏置 x_linear = torch.einsum( 'b i f_in, o i f_out f_in -> b o i f_out', x, self.weight ) + self.bias # 调整形状适配GroupNorm:(BATCH, 通道对总数, 输出特征维度) x_bn = x_linear.reshape(batch_size, -1, self.out_feature_dim) # 批量执行归一化、激活、Dropout x_bn = self.bn(x_bn) x_bn = self.activation(x_bn) x_bn = self.dropout(x_bn) # 恢复原多维度形状 x_bn = x_bn.reshape(batch_size, self.out_channels, self.in_channels, self.out_feature_dim) # 对输入通道维度求和,得到每个输出通道的结果 x_sum = x_bn.sum(dim=2) # 最终输出形状: (BATCH, 输出通道数, 输出特征维度),与原代码一致 return x_sum
关键优化点
- 合并线性层参数:将所有分散的
nn.Linear层权重和偏置合并为大张量,通过torch.einsum实现批量矩阵乘法,彻底消除嵌套循环。 - 独立归一化适配:用
nn.GroupNorm替代独立的nn.BatchNorm1d组,每个通道对对应一个归一化组,严格保持原代码的归一化行为,同时支持批量运算。 - 批量处理激活与Dropout:通过张量形状变换,将多维度特征合并为适配标准层的形状,一次性完成所有通道对的激活和Dropout操作。
- 移除冗余转置:直接基于原始输入维度运算,减少不必要的张量变换开销。
这种优化方式完全利用PyTorch的并行计算能力,能大幅提升模型的前向传播速度,同时保持原模型的功能和行为一致。
内容的提问来源于stack exchange,提问作者Ambrose
相关产品推荐
相关产品推荐

