Source code for speechbrain.nnet.containers

"""Library for implementing cascade (sequences) of different neural modules.

 * Peter Plantinga 2020

import torch
import inspect
import logging
import operator
import functools
from speechbrain.nnet.linear import Linear
from speechbrain.utils.callchains import lengths_arg_exists

logger = logging.getLogger(__name__)

[docs] class Sequential(torch.nn.ModuleDict): """A sequence of modules with potentially inferring shape on construction. If layers are passed with names, these can be referenced with dot notation. Arguments --------- input_shape : iterable A list or tuple of ints or None, representing the expected shape of an input tensor. None represents a variable-length dimension. If no ``input_shape`` is passed, no shape inference will be performed. *layers, **named_layers The inputs are treated as a list of layers to be applied in sequence. The output shape of each layer is used to infer the shape of the following layer. If a tuple is returned, only the shape of the first element is used to determine input shape of the next layer (e.g. RNN returns output, hidden). Example ------- >>> inputs = torch.rand(10, 40, 50) >>> model = Sequential(input_shape=inputs.shape) >>> model.append(Linear, n_neurons=100, layer_name="layer1") >>> model.append(Linear, n_neurons=200, layer_name="layer2") >>> outputs = model(inputs) >>> outputs.shape torch.Size([10, 40, 200]) >>> outputs = model.layer1(inputs) >>> outputs.shape torch.Size([10, 40, 100]) """ def __init__(self, *layers, input_shape=None, **named_layers): super().__init__() # Make sure either layers or input_shape is passed if not layers and input_shape is None and not named_layers: raise ValueError("Must pass either layers or input shape") # Keep track of what layers need "lengths" passed self.length_layers = [] # Replace None dimensions with arbitrary value self.input_shape = input_shape if input_shape and None in input_shape: self.input_shape = list(input_shape) for i, dim in enumerate(self.input_shape): # To reduce size of dummy tensors, use 1 for batch dim if i == 0 and dim is None: dim = 1 # Use 64 as nice round arbitrary value, big enough that # halving this dimension a few times doesn't reach 1 self.input_shape[i] = dim or 256 # Append non-named layers for layer in layers: self.append(layer) # Append named layers for name, layer in named_layers.items(): self.append(layer, layer_name=name)
[docs] def append(self, layer, *args, layer_name=None, **kwargs): """Add a layer to the list of layers, inferring shape if necessary. Arguments --------- layer : A torch.nn.Module class or object If the layer is a class, it should accept an argument called ``input_shape`` which will be inferred and passed. If the layer is a module object, it is added as-is. layer_name : str The name of the layer, for reference. If the name is in use, ``_{count}`` will be appended. *args, **kwargs These are passed to the layer if it is constructed. """ # Compute layer_name if layer_name is None: layer_name = str(len(self)) elif layer_name in self: index = 0 while f"{layer_name}_{index}" in self: index += 1 layer_name = f"{layer_name}_{index}" # Check if it needs to be constructed with input shape if self.input_shape: argspec = inspect.getfullargspec(layer) if "input_shape" in argspec.args + argspec.kwonlyargs: input_shape = self.get_output_shape() layer = layer(*args, input_shape=input_shape, **kwargs) # Finally, append the layer. try: self.add_module(layer_name, layer) except TypeError: raise ValueError( "Must pass `input_shape` at initialization and use " "modules that take `input_shape` to infer shape when " "using `append()`." )
[docs] def get_output_shape(self): """Returns expected shape of the output. Computed by passing dummy input constructed with the ``self.input_shape`` attribute. """ with torch.no_grad(): dummy_input = torch.zeros(self.input_shape) dummy_output = self(dummy_input) return dummy_output.shape
[docs] def forward(self, x): """Applies layers in sequence, passing only the first element of tuples. Arguments --------- x : torch.Tensor The input tensor to run through the network. """ for layer in self.values(): x = layer(x) if isinstance(x, tuple): x = x[0] return x
[docs] class LengthsCapableSequential(Sequential): """Sequential model that can take ``lengths`` in the forward method. This is useful for Sequential models that include RNNs where it is important to avoid padding, or for some feature normalization layers. Unfortunately, this module is not jit-able because the compiler doesn't know ahead of time if the length will be passed, and some layers don't accept the length parameter. """ def __init__(self, *args, **kwargs): self.takes_lengths = [] super().__init__(*args, **kwargs)
[docs] def append(self, *args, **kwargs): """Add a layer to the list of layers, inferring shape if necessary. """ # Add lengths arg inference here. super().append(*args, **kwargs) latest_forward_method = list(self.values())[-1].forward self.takes_lengths.append(lengths_arg_exists(latest_forward_method))
[docs] def forward(self, x, lengths=None): """Applies layers in sequence, passing only the first element of tuples. In addition, forward the ``lengths`` argument to all layers that accept a ``lengths`` argument in their ``forward()`` method (e.g. RNNs). Arguments --------- x : torch.Tensor The input tensor to run through the network. lengths : torch.Tensor The relative lengths of each signal in the tensor. """ for layer, give_lengths in zip(self.values(), self.takes_lengths): if give_lengths: x = layer(x, lengths=lengths) else: x = layer(x) if isinstance(x, tuple): x = x[0] return x
[docs] class ModuleList(torch.nn.Module): """This class implements a wrapper to torch.nn.ModuleList with a forward() method to forward all the layers sequentially. For some pretrained model with the SpeechBrain older implementation of Sequential class, user can use this class to load those pretrained models Arguments --------- *layers : torch class Torch objects to be put in a ModuleList. """ def __init__(self, *layers): super().__init__() self.layers = torch.nn.ModuleList(layers)
[docs] def forward(self, x): """Applies the computation pipeline.""" for layer in self.layers: x = layer(x) if isinstance(x, tuple): x = x[0] return x
[docs] def append(self, module): """Appends module to the layers list.""" self.layers.append(module)
[docs] def extend(self, modules): """Appends module to the layers list.""" self.layers.extend(modules)
[docs] def insert(self, index, module): """Inserts module to the layers list.""" self.layers.insert(module)
[docs] class ConnectBlocks(torch.nn.Module): """Connect a sequence of blocks with shortcut connections. Note: all shortcuts start from the output of the first block, since the first block may change the shape significantly. Arguments --------- input_shape : tuple The shape of the shortcut_type : str One of: * "residual" - first block output passed to final output, * "dense" - input of each block is from all previous blocks, * "skip" - output of each block is passed to final output. shortcut_projection : bool Only has an effect if `shortcut_type` is passed. Whether to add a linear projection layer to the shortcut connection before combining with the output, to handle different sizes. shortcut_combine_fn : str or function Either a pre-defined function (one of "add", "sub", "mul", "div", "avg", "cat") or a user-defined function that takes the shortcut and next input, and combines them, as well as `init_params` in case parameters need to be initialized inside of the function. Example ------- >>> inputs = torch.rand(10, 100, 20) >>> model = ConnectBlocks( ... input_shape=inputs.shape, shortcut_projection=True ... ) >>> model.append(Linear, n_neurons=10) >>> model.append(Linear, n_neurons=10, end_of_block=True) >>> model.append(Linear, n_neurons=10) >>> model.append(Linear, n_neurons=10, end_of_block=True) >>> outputs = model(inputs) >>> outputs.shape torch.Size([10, 100, 10]) """ def __init__( self, input_shape, shortcut_type="residual", shortcut_projection=False, shortcut_combine_fn=torch.add, ): super().__init__() self.first_input_shape = input_shape self.block_input_shape = input_shape self.new_block = True self.blocks = torch.nn.ModuleList() if shortcut_type not in ["residual", "dense", "skip"]: raise ValueError( "'shortcuts' must be one of 'residual', 'dense', or 'skip'" ) self.shortcut_type = shortcut_type self.shortcut_projection = shortcut_projection if shortcut_projection: self.projections = torch.nn.ModuleList() self.shortcut_combine_fn = shortcut_combine_fn
[docs] def append(self, layer, *args, **kwargs): """Appends the specified module to the shortcut model. Arguments --------- layer : torch.nn.Module class This layer will get initialized with *args and **kwargs. Also, the argument ``input_shape`` will be passed if the layer takes it. *args, **kwargs Passed unchanged to the layer **EXCEPT** the kwarg ``end_of_block`` which is used to indicate that the shortcut should be added in. """ if self.new_block: self.blocks.append(Sequential(input_shape=self.block_input_shape)) self.new_block = False end_of_block = False if "end_of_block" in kwargs: end_of_block = kwargs["end_of_block"] del kwargs["end_of_block"] self.blocks[-1].append(layer, *args, **kwargs) # When we reach the end of the block, prepare to add shortcut if end_of_block: # Use dummy input to find shape of next block dummy_input = torch.zeros(self.block_input_shape) dummy_output = self.blocks[-1](dummy_input) # Initialize projection if necessary if self.shortcut_projection: projection_size = functools.reduce( operator.mul, dummy_output.shape[2:], 1 ) if self.shortcut_type == "residual": shape = self.first_input_shape dummy_input = torch.zeros(self.first_input_shape) else: shape = self.block_input_shape self.projections.append( Linear( n_neurons=projection_size, input_shape=shape, bias=False, combine_dims=True, ) ) # Prepare for next block self.new_block = True dummy_output = self._combine(dummy_input, dummy_output, -1) self.block_input_shape = dummy_output.shape
[docs] def forward(self, x): """ Arguments --------- x : torch.Tensor The inputs to the replicated modules. """ shortcut = x for i, block in enumerate(self.blocks): x = block(x) if self.shortcut_type == "skip": shortcut = self._combine(shortcut, x, i) if self.shortcut_type == "dense": x = shortcut = self._combine(shortcut, x, i) if self.shortcut_type == "residual": x = self._combine(shortcut, x, i) if self.shortcut_type == "skip": return shortcut else: return x
def _combine(self, shortcut, x, block_index=0): """Handle combining shortcut with outputs.""" # Apply projection if self.shortcut_projection: shortcut = self.projections[block_index](shortcut) shortcut = shortcut.reshape(x.shape) return self.shortcut_combine_fn(shortcut, x)