Download utils/nn_utils.py from deepcode-ai/Compact_Facts: direct link, hf CLI and curl.
- Browser
- Download file 14.1 kB
-
https://huggingface.co/deepcode-ai/Compact_Facts/resolve/388a6d68da3e36482bb22f8f617a8b7796e3a137/utils/nn_utils.py
- Command line
-
hf download hf://deepcode-ai/Compact_Facts@388a6d68da3e36482bb22f8f617a8b7796e3a137/utils/nn_utils.py
-
curl -L -o nn_utils.py https://huggingface.co/deepcode-ai/Compact_Facts/resolve/388a6d68da3e36482bb22f8f617a8b7796e3a137/utils/nn_utils.py
14.1 kB
| import functools | |
| import logging | |
| import torch | |
| import torch.nn.functional as F | |
| import math | |
| import numpy as np | |
| logger = logging.getLogger(__name__) | |
| def get_device_of(tensor): | |
| """This function returns the device of the tensor | |
| refer to https://github.com/allenai/allennlp/blob/master/allennlp/nn/util.py | |
| Arguments: | |
| tensor {tensor} -- tensor | |
| Returns: | |
| int -- device | |
| """ | |
| if not tensor.is_cuda: | |
| return -1 | |
| else: | |
| return tensor.get_device() | |
| def get_range_vector(size, device): | |
| """This function returns a range vector with the desired size, starting at 0 | |
| the CUDA implementation is meant to avoid copy data from CPU to GPU | |
| refer to https://github.com/allenai/allennlp/blob/master/allennlp/nn/util.py | |
| Arguments: | |
| size {int} -- the size of range | |
| device {int} -- device | |
| Returns: | |
| torch.Tensor -- range vector | |
| """ | |
| if device > -1: | |
| return torch.cuda.LongTensor(size, device=device).fill_(1).cumsum(0) - 1 | |
| else: | |
| return torch.arange(0, size, dtype=torch.long) | |
| def flatten_and_batch_shift_indices(indices, sequence_length): | |
| """This function returns a vector that correctly indexes into the flattened target, | |
| the sequence length of the target must be provided to compute the appropriate offsets. | |
| refer to https://github.com/allenai/allennlp/blob/master/allennlp/nn/util.py | |
| Arguments: | |
| indices {tensor} -- index tensor | |
| sequence_length {int} -- sequence length | |
| Returns: | |
| tensor -- offset index tensor | |
| """ | |
| # Shape: (batch_size) | |
| if torch.max(indices) >= sequence_length or torch.min(indices) < 0: | |
| raise RuntimeError("All elements in indices should be in range (0, {})".format(sequence_length - 1)) | |
| offsets = get_range_vector(indices.size(0), get_device_of(indices)) * sequence_length | |
| for _ in range(len(indices.size()) - 1): | |
| offsets = offsets.unsqueeze(1) | |
| # Shape: (batch_size, d_1, ..., d_n) | |
| offset_indices = indices + offsets | |
| # Shape: (batch_size * d_1 * ... * d_n) | |
| offset_indices = offset_indices.view(-1) | |
| return offset_indices | |
| def batched_index_select(target, indices, flattened_indices=None): | |
| """This function returns selected values in the target with respect to the provided indices, | |
| which have size ``(batch_size, d_1, ..., d_n, embedding_size)`` | |
| refer to https://github.com/allenai/allennlp/blob/master/allennlp/nn/util.py | |
| Arguments: | |
| target {torch.Tensor} -- target tensor | |
| indices {torch.LongTensor} -- index tensor | |
| Keyword Arguments: | |
| flattened_indices {Optional[torch.LongTensor]} -- flattened index tensor (default: {None}) | |
| Returns: | |
| torch.Tensor -- selected tensor | |
| """ | |
| if flattened_indices is None: | |
| # Shape: (batch_size * d_1 * ... * d_n) | |
| flattened_indices = flatten_and_batch_shift_indices(indices, target.size(1)) | |
| # Shape: (batch_size * sequence_length, embedding_size) | |
| flattened_target = target.view(-1, target.size(-1)) | |
| # Shape: (batch_size * d_1 * ... * d_n, embedding_size) | |
| flattened_selected = flattened_target.index_select(0, flattened_indices) | |
| selected_shape = list(indices.size()) + [target.size(-1)] | |
| # Shape: (batch_size, d_1, ..., d_n, embedding_size) | |
| selected_targets = flattened_selected.view(*selected_shape) | |
| return selected_targets | |
| def get_padding_vector(size, dtype, device): | |
| """This function initializes padding unit | |
| Arguments: | |
| size {int} -- padding unit size | |
| dtype {torch.dtype} -- dtype | |
| device {int} -- device = -1 if cpu, device >= 0 if gpu | |
| Returns: | |
| tensor -- padding tensor | |
| """ | |
| pad = torch.zeros(size, dtype=dtype) | |
| if device > -1: | |
| pad = pad.cuda(device=device, non_blocking=True) | |
| return pad | |
| def array2tensor(array, dtype, device): | |
| """This function transforms numpy array to tensor | |
| Arguments: | |
| array {numpy.array} -- numpy array | |
| dtype {torch.dtype} -- torch dtype | |
| device {int} -- device = -1 if cpu, device >= 0 if gpu | |
| Returns: | |
| tensor -- tensor | |
| """ | |
| tensor = torch.as_tensor(array, dtype=dtype) | |
| if device > -1: | |
| tensor = tensor.cuda(device=device, non_blocking=True) | |
| return tensor | |
| def gelu(x): | |
| """Implementation of the gelu activation function. | |
| For information: OpenAI GPT's gelu is slightly different (and gives slightly different results): | |
| 0.5 * x * (1 + torch.tanh(math.sqrt(2 / math.pi) * (x + 0.044715 * torch.pow(x, 3)))) | |
| Also see https://arxiv.org/abs/1606.08415 | |
| refer to: https://github.com/huggingface/pytorch-transformers/blob/master/pytorch_transformers/modeling_bert.py | |
| """ | |
| return x * 0.5 * (1.0 + torch.erf(x / math.sqrt(2.0))) | |
| def pad_vecs(vecs, padding_size, dtype, device): | |
| """This function pads vectors for batch | |
| Arguments: | |
| vecs {list} -- vector list | |
| padding_size {int} -- padding dims | |
| dtype {torch.dtype} -- dtype | |
| device {int} -- device = -1 if cpu, device >= 0 if gpu | |
| Returns: | |
| tensor -- padded vectors | |
| """ | |
| max_length = max(len(vec) for vec in vecs) | |
| if max_length == 0: | |
| pad_vecs = torch.cat([get_padding_vector((1, padding_size), dtype, device).unsqueeze(0) for _ in vecs], 0) | |
| return pad_vecs | |
| pad_vecs = [] | |
| for vec in vecs: | |
| pad_vec = torch.cat(vec + [get_padding_vector((1, padding_size), dtype, device)] * (max_length - len(vec)), | |
| 0).unsqueeze(0) | |
| assert pad_vec.size() == (1, max_length, padding_size), "the size of pad vector is not correct" | |
| pad_vecs.append(pad_vec) | |
| return torch.cat(pad_vecs, 0) | |
| def get_bilstm_minus(batch_seq_encoder_repr, span_list, seq_lens): | |
| """This function gets span representation using bilstm minus | |
| Arguments: | |
| batch_seq_encoder_repr {list} -- batch sequence encoder representation | |
| span_list {list} -- span list | |
| seq_lens {list} -- sequence length list | |
| Returns: | |
| tensor -- span representation vector | |
| """ | |
| assert len(batch_seq_encoder_repr) == len( | |
| span_list), "the length of batch seq encoder repr is not equal to span list's length" | |
| assert len(span_list) == len(seq_lens), "the length of span list is not equal to batch seq lens's length" | |
| hidden_size = batch_seq_encoder_repr.size(-1) | |
| span_vecs = [] | |
| for seq_encoder_repr, (s, e), seq_len in zip(batch_seq_encoder_repr, span_list, seq_lens): | |
| rnn_output = seq_encoder_repr[:seq_len] | |
| forward_rnn_output, backward_rnn_output = rnn_output.split(hidden_size // 2, 1) | |
| forward_span_vec = get_forward_segment(forward_rnn_output, s, e, get_device_of(forward_rnn_output)) | |
| backward_span_vec = get_backward_segment(backward_rnn_output, s, e, get_device_of(backward_rnn_output)) | |
| span_vec = torch.cat([forward_span_vec, backward_span_vec], 0).unsqueeze(0) | |
| span_vecs.append(span_vec) | |
| return torch.cat(span_vecs, 0) | |
| def get_forward_segment(forward_rnn_output, s, e, device): | |
| """This function gets span representaion in forward rnn | |
| Arguments: | |
| forward_rnn_output {tensor} -- forward rnn output | |
| s {int} -- span start | |
| e {int} -- span end | |
| device {int} -- device | |
| Returns: | |
| tensor -- span representaion vector | |
| """ | |
| seq_len, hidden_size = forward_rnn_output.size() | |
| if s >= e: | |
| vec = torch.zeros(hidden_size, dtype=forward_rnn_output.dtype) | |
| if device > -1: | |
| vec = vec.cuda(device=device, non_blocking=True) | |
| return vec | |
| if s == 0: | |
| return forward_rnn_output[e - 1] | |
| return forward_rnn_output[e - 1] - forward_rnn_output[s - 1] | |
| def get_backward_segment(backward_rnn_output, s, e, device): | |
| """This function gets span representaion in backward rnn | |
| Arguments: | |
| forward_rnn_output {tensor} -- backward rnn output | |
| s {int} -- span start | |
| e {int} -- span end | |
| device {int} -- device | |
| Returns: | |
| tensor -- span representaion vector | |
| """ | |
| seq_len, hidden_size = backward_rnn_output.size() | |
| if s >= e: | |
| vec = torch.zeros(hidden_size, dtype=backward_rnn_output.dtype) | |
| if device > -1: | |
| vec = vec.cuda(device=device, non_blocking=True) | |
| return vec | |
| if e == seq_len: | |
| return backward_rnn_output[s] | |
| return backward_rnn_output[s] - backward_rnn_output[e] | |
| def get_dist_vecs(span_list, max_sent_len, device): | |
| """This function gets distance embedding | |
| Arguments: | |
| span_list {list} -- span list | |
| Returns: | |
| tensor -- distance embedding vector | |
| """ | |
| dist_vecs = [] | |
| for s, e in span_list: | |
| assert s <= e, "span start is greater than end" | |
| vec = torch.Tensor(np.eye(max_sent_len)[e - s]) | |
| if device > -1: | |
| vec = vec.cuda(device=device, non_blocking=True) | |
| dist_vecs.append(vec) | |
| return torch.stack(dist_vecs) | |
| def get_conv_vecs(batch_token_repr, span_list, span_batch_size, conv_layer): | |
| """This funciton gets span vector representation through convolution layer | |
| Arguments: | |
| batch_token_repr {list} -- batch token representation | |
| span_list {list} -- span list | |
| span_batch_size {int} -- span convolutuion batch size | |
| conv_layer {nn.Module} -- convolution layer | |
| Returns: | |
| tensor -- conv vectors | |
| """ | |
| assert len(batch_token_repr) == len(span_list), "the length of batch token repr is not equal to span list's length" | |
| span_vecs = [] | |
| for token_repr, (s, e) in zip(batch_token_repr, span_list): | |
| if s == e: | |
| span_vecs.append([]) | |
| continue | |
| span_vecs.append(list(token_repr[s:e].split(1))) | |
| span_conv_vecs = [] | |
| for id in range(0, len(span_vecs), span_batch_size): | |
| span_pad_vecs = pad_vecs(span_vecs[id:id + span_batch_size], conv_layer.get_input_dims(), | |
| batch_token_repr[0].dtype, get_device_of(batch_token_repr[0])) | |
| span_conv_vecs.append(conv_layer(span_pad_vecs)) | |
| return torch.cat(span_conv_vecs, dim=0) | |
| def get_n_trainable_parameters(model): | |
| """This function calculates the number of trainable parameters | |
| of the model | |
| Arguments: | |
| model {nn.Module} -- model | |
| Returns: | |
| int -- the number of trainable parameters of the model | |
| """ | |
| cnt = 0 | |
| for param in list(model.parameters()): | |
| if param.requires_grad: | |
| cnt += functools.reduce(lambda x, y: x * y, list(param.size()), 1) | |
| return cnt | |
| def js_div(p, q, reduction='batchmean'): | |
| """js_div caculate Jensen Shannon Divergence (JSD). | |
| Args: | |
| p (tensor): distribution p | |
| q (tensor): distribution q | |
| reduction (str, optional): reduction. Defaults to 'batchmean'. | |
| Returns: | |
| tensor: JS divergence | |
| """ | |
| m = 0.5 * (p + q) | |
| return (F.kl_div(p, m, reduction=reduction) + F.kl_div(q, m, reduction=reduction)) * 0.5 | |
| def load_weight_from_pretrained_model(model, pretrained_state_dict, prefix=""): | |
| """load_weight_from_pretrained_model This function loads weight from pretrained model. | |
| Arguments: | |
| model {nn.Module} -- model | |
| pretrained_state_dict {dict} -- state dict of pretrained model | |
| Keyword Arguments: | |
| prefix {str} -- prefix for pretrained model (default: {""}) | |
| """ | |
| model_state_dict = model.state_dict() | |
| # # load weight except decode weight | |
| # filtered_state_dict = { | |
| # k: pretrained_state_dict[k] | |
| # for k, v in model_state_dict.items() if k in pretrained_state_dict | |
| # and v.size() == pretrained_state_dict[k].size() and 'decoder' not in k | |
| # } | |
| # # load bert encoder & cnn | |
| # filtered_state_dict.update({ | |
| # k: pretrained_state_dict[k[k.find('.') + 1:]] | |
| # for k, v in model_state_dict.items() if k[k.find('.') + 1:] in pretrained_state_dict | |
| # and v.size() == pretrained_state_dict[k[k.find('.') + 1:]].size() and 'decoder' not in k | |
| # }) | |
| filtered_state_dict = {} | |
| for k, v in model_state_dict.items(): | |
| if 'decoder' in k: | |
| continue | |
| # if 'bert_encoder' not in k: | |
| # continue | |
| k = k.split('.') | |
| for candi_name in ['.'.join(k), '.'.join(k[1:]), '.'.join(k[2:])]: | |
| if candi_name in pretrained_state_dict and v.size() == pretrained_state_dict[candi_name].size(): | |
| filtered_state_dict['.'.join(k)] = pretrained_state_dict[candi_name] | |
| break | |
| candi_name = prefix + candi_name | |
| if candi_name in pretrained_state_dict and v.size() == pretrained_state_dict[candi_name].size(): | |
| filtered_state_dict['.'.join(k)] = pretrained_state_dict[candi_name] | |
| break | |
| # only load bert encoder | |
| # filtered_state_dict = {k: pretrained_state_dict[k[k.find('.') + 1:]] for k, v in model_state_dict.items() if 'bert_encoder' in k and k[k.find('.') + 1:] in pretrained_state_dict and v.size() == pretrained_state_dict[k[k.find('.') + 1:]].size() and 'decoder' not in k} | |
| logger.info("Load weights parameters:") | |
| for name in filtered_state_dict: | |
| logger.info(name) | |
| model_state_dict.update(filtered_state_dict) | |
| model.load_state_dict(model_state_dict) | |
| def clone_weights(first_module, second_module): | |
| """This function clones(ties) weights from first module to second module | |
| refers to: https://huggingface.co/transformers/v1.2.0/_modules/pytorch_transformers/modeling_utils.html#PreTrainedModel | |
| Arguments: | |
| first_module {nn.Module} -- first module | |
| second_module {nn.Module} -- second module | |
| """ | |
| first_module.weight = second_module.weight | |
| if hasattr(first_module, 'bias') and first_module.bias is not None: | |
| first_module.bias.data = torch.nn.functional.pad(first_module.bias.data, | |
| (0, first_module.weight.shape[0] - first_module.bias.shape[0]), | |
| 'constant', 0) | |