import torch import torch.nn as nn from transformers import BertConfig, BertModel, BertForSequenceClassification class UIEModel(nn.Module): def __init__(self, args): super(UIEModel, self).__init__() self.args = args self.tasks = args.tasks bert_dir = args.bert_dir self.bert_config = BertConfig.from_pretrained(bert_dir) self.bert_model = BertModel.from_pretrained(bert_dir) # self.bert_model.load_state_dict(torch.load(self.args.bert_pt_dir, map_location="cuda")) # self.bert_model = BertForSequenceClassification.from_pretrained(args.bert_pt_dir) if "ner" in args.tasks: self.ner_num_labels = args.ner_num_labels self.module_start_list = nn.ModuleList() self.module_end_list = nn.ModuleList() self.module_content_list = nn.ModuleList() # 增加一层内容判断 for i in range(args.ner_num_labels): self.module_start_list.append(nn.Linear(self.bert_config.hidden_size, 1)) self.module_end_list.append(nn.Linear(self.bert_config.hidden_size, 1)) self.module_content_list.append(nn.Linear(self.bert_config.hidden_size, 1)) self.ner_criterion = nn.BCEWithLogitsLoss() self.dropout = nn.Dropout(self.bert_config.hidden_dropout_prob) @staticmethod def build_dummpy_inputs(): inputs = {} inputs['ner_input_ids'] = torch.LongTensor( torch.randint(low=1, high=10, size=(32, 56))) inputs['ner_attention_mask'] = torch.ones(size=(32, 56)).long() inputs['ner_token_type_ids'] = torch.zeros(size=(32, 56)).long() inputs['ner_start_labels'] = torch.zeros(size=(32, 8, 56)).float() inputs['ner_end_labels'] = torch.zeros(size=(32, 8, 56)).float() return inputs def get_pointer_loss(self, start_logits, end_logits, attention_mask, start_labels, end_labels, criterion): start_logits = start_logits.view(-1) end_logits = end_logits.view(-1) active_loss = attention_mask.view(-1) == 1 active_start_logits = start_logits[active_loss] active_end_logits = end_logits[active_loss] active_start_labels = start_labels.view(-1)[active_loss] active_end_labels = end_labels.view(-1)[active_loss] start_loss = criterion(active_start_logits, active_start_labels) end_loss = criterion(active_end_logits, active_end_labels) loss = start_loss + end_loss return loss def ner_forward_1(self, ner_input_ids, ner_attention_mask, ner_start_labels=None, ner_end_labels=None): # 四个参数格式均为[tensor(), tensor(), ...] # 一次传入batch_size个样本，每个样本含多条句子 # 编码还需要一个个样本进行，若每个样本句子太长，还需截断分批处理 all_start_logits = [] all_end_logits = [] ner_loss = None for i in range(len(ner_end_labels)): # 有len(ner_end_labels)个样本/文档 input_ids = ner_input_ids[i].to(self.args.device) attention_mask = ner_attention_mask[i].to(self.args.device) # start_labels = ner_start_labels[i].to(self.args.device) # end_labels = ner_end_labels[i].to(self.args.device) # 根据sent_num的大小分段进行编码（sent_num太大时，显存不够） max_encoder_len = self.args.max_encoder_sent_len batch_num = int(input_ids.size(0) / max_encoder_len) last_hidden_states = [] if batch_num > 0: for i in range(batch_num): truncated_input_ids = input_ids[i*max_encoder_len:(i+1)*max_encoder_len, :] truncated_attention_mask = attention_mask[i*max_encoder_len:(i+1)*max_encoder_len, :] truncated_outputs = self.bert_model(truncated_input_ids, attention_mask=truncated_attention_mask) last_hidden_states.append(truncated_outputs.last_hidden_state) # .detach().cpu() if input_ids.size(0) - batch_num * max_encoder_len > 0: truncated_input_ids = input_ids[batch_num*max_encoder_len:, :] truncated_attention_mask = attention_mask[batch_num*max_encoder_len:, :] truncated_outputs = self.bert_model(truncated_input_ids, attention_mask=truncated_attention_mask) last_hidden_states.append(truncated_outputs.last_hidden_state) # .detach().cpu() if len(last_hidden_states) > 1: seq_bert_output = torch.cat(last_hidden_states, dim=0) else: seq_bert_output = last_hidden_states[0] # [sent_num, seq_len, hidden_dim] # 忽略padding求均值的方法 seq_bert_output = seq_bert_output[:, 1:-1, :] #.to(self.args.device) # 忽略[CLS]和[SEP] ??? expanded_attention_mask = attention_mask[:,1:-1].unsqueeze(-1).expand_as(seq_bert_output) # [sent_num,seq_len,hidden_size] sum_of_non_padded_output = (seq_bert_output * expanded_attention_mask).sum(dim=1) # 仅对有效位置求和 mean_encoder_outputs = sum_of_non_padded_output / expanded_attention_mask.sum(dim=1) # [sent_num, hidden_size] # dropout pooled_output = self.dropout(mean_encoder_outputs) # 计算Pointer位置的loss # for i in range(self.ner_num_labels): # 每个ner任务单独计算 if self.ner_num_labels == 1: # 对每个pointer位置接个线性层 start_logit = self.module_start_list[0](pooled_output).squeeze(1) #[sent_num] end_logit = self.module_end_list[0](pooled_output).squeeze(1) #[sent_num] # print(start_logit, start_logit.size()) all_start_logits.append(start_logit) all_end_logits.append(end_logit) # 将批次数据合成一个loss值 concat_start_logits = torch.cat(all_start_logits, dim=0) concat_end_logits = torch.cat(all_end_logits, dim=0) all_start_labels = torch.cat(ner_start_labels, dim=0).to(self.args.device) all_end_labels = torch.cat(ner_end_labels, dim=0).to(self.args.device) start_loss = self.ner_criterion(concat_start_logits, all_start_labels) # 起始位置loss值 end_loss = self.ner_criterion(concat_end_logits, all_end_labels) # 结束位置loss值 if ner_loss is None: ner_loss = start_loss + end_loss else: ner_loss += (start_loss + end_loss) res = { "ner_start_logits": [a.detach().cpu() for a in all_start_logits], "ner_end_logits": [a.detach().cpu() for a in all_end_logits], "ner_loss": ner_loss, } return res def ner_bc_forward(self, ner_input_ids, ner_attention_mask, ner_start_labels=None, ner_end_labels=None, ner_content_labels=None): """ ner:topic识别任务； bc:二分类任务 这里将试题的开始、结束位置预测，与试题的判断(是否属于试题内容)任务合并在一起！！！ 原因：根据预测标签进行试题切分时，按start的位置划分错误最少，但会出现题型行也会被划分到试题中，故需要单独判断！ # 四个参数格式均为[tensor(), tensor(), ...] # 一次传入batch_size=1个样本被截取的一部分，每个样本含多条句子 # 编码还需要一个个样本进行，若每个样本句子太长，需截断分批处理 """ all_start_logits = [] all_end_logits = [] all_content_logits = [] ner_loss = None outputs = self.bert_model(ner_input_ids, attention_mask=ner_attention_mask) # 取cls位置的表示 cls_bert_output = outputs.pooler_output # [sent_num, hidden_size] # dropout pooled_output = self.dropout(cls_bert_output) # 对每个pointer位置接个线性层 # 对试题判断任务也接个线性层 start_logit = self.module_start_list[0](pooled_output).squeeze(1) #[sent_num] end_logit = self.module_end_list[0](pooled_output).squeeze(1) #[sent_num] content_logit = self.module_content_list[0](pooled_output).squeeze(1) #[sent_num] all_start_logits.append(start_logit) all_end_logits.append(end_logit) all_content_logits.append(content_logit) if ner_start_labels is not None and ner_end_labels is not None: start_loss = self.ner_criterion(start_logit, ner_start_labels) # 起始位置loss值 end_loss = self.ner_criterion(end_logit, ner_end_labels) # 结束位置loss值 content_loss = self.ner_criterion(content_logit, ner_content_labels) if ner_loss is None: ner_loss = start_loss + end_loss + content_loss else: ner_loss += (start_loss + end_loss + content_loss) res = { "ner_start_logits": [a.detach().cpu() for a in all_start_logits], "ner_end_logits": [a.detach().cpu() for a in all_end_logits], "ner_content_logits": [a.detach().cpu() for a in all_content_logits], "ner_loss": ner_loss, } return res def ner_forward(self, ner_input_ids, ner_attention_mask, ner_start_labels=None, ner_end_labels=None): """ # 四个参数格式均为[tensor(), tensor(), ...] # 一次传入batch_size=1个样本被截取的一部分，每个样本含多条句子 # 编码还需要一个个样本进行，若每个样本句子太长，还需截断分批处理 """ all_start_logits = [] all_end_logits = [] ner_loss = None res = { "ner_start_logits": None, "ner_end_logits": None, "ner_loss": None } outputs = self.bert_model(ner_input_ids, attention_mask=ner_attention_mask) # 取每个token位置的表示再求平均 # last_hidden_states = outputs.last_hidden_state # [sent_num, seq_len, hidden_dim] # seq_bert_output = last_hidden_states[:, 1:-1, :] # 忽略[CLS]和[SEP] ??? # # 忽略padding求均值的方法 # expanded_attention_mask = ner_attention_mask[:,1:-1].unsqueeze(-1).expand_as(seq_bert_output) # [sent_num,seq_len,hidden_size] # sum_of_non_padded_output = (seq_bert_output * expanded_attention_mask).sum(dim=1) # 仅对有效位置求和 # mean_encoder_outputs = sum_of_non_padded_output / expanded_attention_mask.sum(dim=1) # [sent_num, hidden_size] # 取cls位置的表示 cls_bert_output = outputs.pooler_output # [sent_num, hidden_size] # dropout pooled_output = self.dropout(cls_bert_output) # 计算Pointer位置的loss if self.ner_num_labels == 1: # 对每个pointer位置接个线性层 # tensor.squeeze(1):移除大小为1的第二个维度 start_logit = self.module_start_list[0](pooled_output).squeeze(1) #[sent_num] end_logit = self.module_end_list[0](pooled_output).squeeze(1) #[sent_num] all_start_logits.append(start_logit) all_end_logits.append(end_logit) if ner_start_labels is not None and ner_end_labels is not None: start_loss = self.ner_criterion(start_logit, ner_start_labels) # 起始位置loss值 end_loss = self.ner_criterion(end_logit, ner_end_labels) # 结束位置loss值 if ner_loss is None: ner_loss = start_loss + end_loss else: ner_loss += (start_loss + end_loss) res = { "ner_start_logits": [a.detach().cpu() for a in all_start_logits], "ner_end_logits": [a.detach().cpu() for a in all_end_logits], "ner_loss": ner_loss, } return res def forward(self, ner_input_ids=None, # ner_token_type_ids=None, ner_attention_mask=None, ner_start_labels=None, ner_end_labels=None, ner_content_labels=None, ): res = { "ner_output": None, "re_output": None, "event_output": None } if "ner" in self.tasks: # ner_output = self.ner_forward( # ner_input_ids, # # ner_token_type_ids, # ner_attention_mask, # ner_start_labels, # ner_end_labels, # ) ner_output = self.ner_bc_forward( ner_input_ids, # ner_token_type_ids, ner_attention_mask, ner_start_labels, ner_end_labels, ner_content_labels, ) res["ner_output"] = ner_output return res if __name__ == '__main__': inputs = UIEModel.build_dummpy_inputs() class Args: bert_dir = "../chinese-bert-wwm-ext/" ner_num_labels = 8 re_num_labels = 16 tasks = ["re_rel"] args = Args() model = UIEModel(args) res = model( ner_input_ids=inputs["ner_input_ids"], ner_token_type_ids=inputs["ner_token_type_ids"], ner_attention_mask=inputs["ner_attention_mask"], ner_start_labels=inputs["ner_start_labels"], ner_end_labels=inputs["ner_end_labels"], ) print(res)