IE101TW / models /kg.py
DeepLearning101's picture
Upload kg.py
3169cc9
raw
history blame
9.28 kB
# -*- coding: utf-8 -*-
# @Time : 2022/2/17 11:26 上午
# @Author : JianingWang
# @File : kg.py
import torch
from torch import nn
from torch.nn import CrossEntropyLoss
import torch.nn.functional as F
from collections import OrderedDict
from transformers.models.bert import BertPreTrainedModel, BertModel
from transformers.models.bert.modeling_bert import BertOnlyMLMHead
class MLPLayer(nn.Module):
"""
Head for getting sentence representations over RoBERTa/BERT"s CLS representation.
"""
def __init__(self, config):
super().__init__()
self.dense = nn.Linear(config.hidden_size, config.hidden_size)
self.activation = nn.Tanh()
def forward(self, features, **kwargs):
x = self.dense(features)
x = self.activation(x)
return x
class Similarity(nn.Module):
"""
Dot product or cosine similarity
"""
def __init__(self, temp):
super().__init__()
self.temp = temp
self.cos = nn.CosineSimilarity(dim=-1)
def forward(self, x, y):
return self.cos(x, y) / self.temp
class BertForPretrainWithKG(BertPreTrainedModel):
def __init__(self, config):
super().__init__(config)
self.num_labels = config.num_labels
self.config = config
self.bert = BertModel(config)
classifier_dropout = (
config.classifier_dropout if config.classifier_dropout is not None else config.hidden_dropout_prob
)
self.dropout = nn.Dropout(classifier_dropout)
self.cls = BertOnlyMLMHead(config)
self.classifiers = nn.ModuleList([nn.Linear(config.hidden_size, config.num_ner_labels) for _ in range(config.entity_type_num)])
self.post_init()
def forward(
self,
input_ids=None,
attention_mask=None,
token_type_ids=None,
position_ids=None,
head_mask=None,
inputs_embeds=None,
encoder_hidden_states=None,
encoder_attention_mask=None,
labels=None,
ner_labels=None,
output_attentions=None,
output_hidden_states=None,
return_dict=None,
):
return_dict = return_dict if return_dict is not None else self.config.use_return_dict
outputs = self.bert(
input_ids,
attention_mask=attention_mask,
token_type_ids=token_type_ids,
position_ids=position_ids,
head_mask=head_mask,
inputs_embeds=inputs_embeds,
encoder_hidden_states=encoder_hidden_states,
encoder_attention_mask=encoder_attention_mask,
output_attentions=output_attentions,
output_hidden_states=output_hidden_states,
return_dict=return_dict,
)
sequence_output = outputs.last_hidden_state
# mlm
prediction_scores = self.cls(sequence_output)
# ner
sequence_output = self.dropout(sequence_output)
ner_logits = torch.stack([classifier(sequence_output) for classifier in self.classifiers]).movedim(1, 0)
# mlm
masked_lm_loss, ner_loss, total_loss = None, None, None
if labels is not None:
loss_fct = CrossEntropyLoss() # -100 index = padding token
masked_lm_loss = loss_fct(prediction_scores.view(-1, self.config.vocab_size), labels.view(-1))
if ner_labels is not None:
loss_fct = CrossEntropyLoss()
# Only keep active parts of the loss
active_loss = attention_mask.repeat(self.config.entity_type_num, 1, 1).view(-1) == 1
active_logits = ner_logits.reshape(-1, self.config.num_ner_labels)
active_labels = torch.where(
active_loss, ner_labels.view(-1), torch.tensor(loss_fct.ignore_index).type_as(ner_labels)
)
ner_loss = loss_fct(active_logits, active_labels)
if masked_lm_loss:
total_loss = masked_lm_loss + ner_loss * 4
return OrderedDict([
("loss", total_loss),
("mlm_loss", masked_lm_loss.unsqueeze(0)),
("ner_loss", ner_loss.unsqueeze(0)),
("logits", prediction_scores.argmax(2)),
("ner_logits", ner_logits.argmax(3))
])
# MaskedLMOutput(
# loss=total_loss,
# logits=prediction_scores.argmax(2),
# ner_l
# hidden_states=outputs.hidden_states,
# attentions=outputs.attentions,
# )
class BertForPretrainWithKGV2(BertPreTrainedModel):
def __init__(self, config):
super().__init__(config)
self.num_labels = config.num_labels
self.config = config
self.bert = BertModel(config)
classifier_dropout = (
config.classifier_dropout if config.classifier_dropout is not None else config.hidden_dropout_prob
)
self.dropout = nn.Dropout(classifier_dropout)
self.cls = BertOnlyMLMHead(config)
self.classifiers = nn.ModuleList([nn.Linear(config.hidden_size, config.num_ner_labels) for _ in range(config.entity_type_num)])
self.mlp = MLPLayer(config)
self.sim = Similarity(0.05)
self.post_init()
def forward(
self,
input_ids=None,
attention_mask=None,
token_type_ids=None,
position_ids=None,
head_mask=None,
inputs_embeds=None,
encoder_hidden_states=None,
encoder_attention_mask=None,
labels=None,
ner_labels=None,
output_attentions=None,
output_hidden_states=None,
return_dict=None,
):
return_dict = return_dict if return_dict is not None else self.config.use_return_dict
outputs = self.bert(
input_ids,
attention_mask=attention_mask,
token_type_ids=token_type_ids,
position_ids=position_ids,
head_mask=head_mask,
inputs_embeds=inputs_embeds,
encoder_hidden_states=encoder_hidden_states,
encoder_attention_mask=encoder_attention_mask,
output_attentions=output_attentions,
output_hidden_states=output_hidden_states,
return_dict=return_dict,
)
sequence_output = outputs.last_hidden_state
# mlm
prediction_scores = self.cls(sequence_output)
# ner
sequence_output = self.dropout(sequence_output)
ner_logits = torch.stack([classifier(sequence_output) for classifier in self.classifiers]).movedim(1, 0)
# mlm
masked_lm_loss, ner_loss, total_loss = None, None, None
if labels is not None:
loss_fct = CrossEntropyLoss() # -100 index = padding token
masked_lm_loss = loss_fct(prediction_scores.view(-1, self.config.vocab_size), labels.view(-1))
if ner_labels is not None:
loss_fct = CrossEntropyLoss()
active_logits = ner_logits.reshape(-1, self.config.num_ner_labels)
# padding 的label是-100
ner_loss = loss_fct(active_logits, ner_labels.view(-1))
if masked_lm_loss:
total_loss = masked_lm_loss
if ner_loss:
total_loss = total_loss + ner_loss
# 对比cls loss
# cls_hidden = outputs.pooler_output
cls_hidden = sequence_output[:, 0]
simcse_loss = self.simcse_unsup_loss2(cls_hidden)
if simcse_loss:
total_loss = total_loss + simcse_loss*10
ner_out = ner_logits.argmax(3)
return OrderedDict([
("loss", total_loss),
("mlm_loss", masked_lm_loss.unsqueeze(0)),
("ner_loss", ner_loss.unsqueeze(0)),
("logits", prediction_scores.argmax(2)),
("ner_logits", ner_out.view(ner_out.shape[0], -1)),
("simcse_loss", simcse_loss.unsqueeze(0))
])
def simcse_unsup_loss2(self, pooler_output):
pooler_output = pooler_output.view((-1, 2, pooler_output.size(-1)))
pooler_output = self.mlp(pooler_output)
z1, z2 = pooler_output[:, 0], pooler_output[:, 1]
cos_sim = self.sim(z1.unsqueeze(1), z2.unsqueeze(0))
labels = torch.arange(cos_sim.size(0)).long().to(pooler_output.device)
loss_fct = nn.CrossEntropyLoss()
loss = loss_fct(cos_sim, labels)
return loss
@staticmethod
def simcse_unsup_loss(y_pred: "tensor") -> "tensor":
# 得到y_pred对应的label, [1, 0, 3, 2, ..., batch_size-1, batch_size-2]
y_true = torch.arange(y_pred.shape[0], device=y_pred.device)
y_true = (y_true - y_true % 2 * 2) + 1
# batch内两两计算相似度, 得到相似度矩阵(对角矩阵)
sim = F.cosine_similarity(y_pred.unsqueeze(1), y_pred.unsqueeze(0), dim=-1)
# sim = torch.mm(y_pred, y_pred.transpose(0, 1))
# 将相似度矩阵对角线置为很小的值, 消除自身的影响
sim = sim - torch.eye(y_pred.shape[0], device=y_pred.device) * 1e12
# 相似度矩阵除以温度系数
sim = sim/0.05
# 计算相似度矩阵与y_true的交叉熵损失
loss = F.cross_entropy(sim, y_true)
print(loss)
return loss