| from typing import List |
| import torch |
|
|
| from common import utils |
| from common.utils import OutputData, InputData |
| from torch import Tensor |
|
|
| def argmax_for_seq_len(inputs, seq_lens, padding_value=-100): |
| packed_inputs = utils.pack_sequence(inputs, seq_lens) |
| outputs = torch.argmax(packed_inputs, dim=-1, keepdim=True) |
| return utils.unpack_sequence(outputs, seq_lens, padding_value).squeeze(-1) |
|
|
|
|
| def decode(output: OutputData, |
| target: InputData = None, |
| pred_type="slot", |
| multi_threshold=0.5, |
| ignore_index=-100, |
| return_list=True, |
| return_sentence_level=True, |
| use_multi=False, |
| use_crf=False, |
| CRF=None) -> List or Tensor: |
| """ decode output logits |
| |
| Args: |
| output (OutputData): output logits data |
| target (InputData, optional): input data with attention mask. Defaults to None. |
| pred_type (str, optional): prediction type in ["slot", "intent", "token-level-intent"]. Defaults to "slot". |
| multi_threshold (float, optional): multi intent decode threshold. Defaults to 0.5. |
| ignore_index (int, optional): align and pad token with ignore index. Defaults to -100. |
| return_list (bool, optional): if True return list else return torch Tensor. Defaults to True. |
| return_sentence_level (bool, optional): if True decode sentence level intent else decode token level intent. Defaults to True. |
| use_multi (bool, optional): whether to decode to multi intent. Defaults to False. |
| use_crf (bool, optional): whether to use crf. Defaults to False. |
| CRF (CRF, optional): CRF function. Defaults to None. |
| |
| Returns: |
| List or Tensor: decoded sequence ids |
| """ |
| if pred_type == "slot": |
| inputs = output.slot_ids |
| else: |
| inputs = output.intent_ids |
|
|
| if pred_type == "slot": |
| if not use_multi: |
| if use_crf: |
| res = CRF.decode(inputs, mask=target.attention_mask) |
| else: |
| res = torch.argmax(inputs, dim=-1) |
| else: |
| raise NotImplementedError("Multi-slot prediction is not supported.") |
| elif pred_type == "intent": |
| if not use_multi: |
| res = torch.argmax(inputs, dim=-1) |
| else: |
| res = (torch.sigmoid(inputs) > multi_threshold).nonzero() |
| if return_list: |
| res_index = res.detach().cpu().tolist() |
| res_list = [[] for _ in range(len(target.seq_lens))] |
| for item in res_index: |
| res_list[item[0]].append(item[1]) |
| return res_list |
| else: |
| return res |
| elif pred_type == "token-level-intent": |
| if not use_multi: |
| res = torch.argmax(inputs, dim=-1) |
| if not return_sentence_level: |
| return res |
| if return_list: |
| res = res.detach().cpu().tolist() |
| attention_mask = target.attention_mask |
| for i in range(attention_mask.shape[0]): |
| temp = [] |
| for j in range(attention_mask.shape[1]): |
| if attention_mask[i][j] == 1: |
| temp.append(res[i][j]) |
| else: |
| break |
| res[i] = temp |
| return [max(it, key=lambda v: it.count(v)) for it in res] |
| else: |
| seq_lens = target.seq_lens |
|
|
| if not return_sentence_level: |
| token_res = torch.cat([ |
| torch.sigmoid(inputs[i, 0:seq_lens[i], :]) > multi_threshold |
| for i in range(len(seq_lens))], |
| dim=0) |
| return utils.unpack_sequence(token_res, seq_lens, padding_value=ignore_index) |
|
|
| intent_index_sum = torch.cat([ |
| torch.sum(torch.sigmoid(inputs[i, 0:seq_lens[i], :]) > multi_threshold, dim=0).unsqueeze(0) |
| for i in range(len(seq_lens))], |
| dim=0) |
|
|
| res = (intent_index_sum > torch.div(seq_lens, 2, rounding_mode='floor').unsqueeze(1)).nonzero() |
| if return_list: |
| res_index = res.detach().cpu().tolist() |
| res_list = [[] for _ in range(len(seq_lens))] |
| for item in res_index: |
| res_list[item[0]].append(item[1]) |
| return res_list |
| else: |
| return res |
| else: |
| raise NotImplementedError("Prediction mode except ['slot','intent','token-level-intent'] is not supported.") |
| if return_list: |
| res = res.detach().cpu().tolist() |
| return res |
|
|
|
|
| def compute_loss(pred: OutputData, |
| target: InputData, |
| criterion_type="slot", |
| use_crf=False, |
| ignore_index=-100, |
| loss_fn=None, |
| use_multi=False, |
| CRF=None): |
| """ compute loss |
| |
| Args: |
| pred (OutputData): output logits data |
| target (InputData): input golden data |
| criterion_type (str, optional): criterion type in ["slot", "intent", "token-level-intent"]. Defaults to "slot". |
| ignore_index (int, optional): compute loss with ignore index. Defaults to -100. |
| loss_fn (_type_, optional): loss function. Defaults to None. |
| use_crf (bool, optional): whether to use crf. Defaults to False. |
| CRF (CRF, optional): CRF function. Defaults to None. |
| |
| Returns: |
| Tensor: loss result |
| """ |
| if criterion_type == "slot": |
| if use_crf: |
| return -1 * CRF(pred.slot_ids, target.slot, target.get_slot_mask(ignore_index).byte()) |
| else: |
| pred_slot = utils.pack_sequence(pred.slot_ids, target.seq_lens) |
| target_slot = utils.pack_sequence(target.slot, target.seq_lens) |
| return loss_fn(pred_slot, target_slot) |
| elif criterion_type == "token-level-intent": |
| |
| intent_target = target.intent.unsqueeze(1) |
| if not use_multi: |
| intent_target = intent_target.repeat(1, pred.intent_ids.shape[1]) |
| else: |
| intent_target = intent_target.repeat(1, pred.intent_ids.shape[1], 1) |
| intent_pred = utils.pack_sequence(pred.intent_ids, target.seq_lens) |
| intent_target = utils.pack_sequence(intent_target, target.seq_lens) |
| return loss_fn(intent_pred, intent_target) |
| else: |
| return loss_fn(pred.intent_ids, target.intent) |
|
|
