|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
""" TF 2.0 DeBERTa model.""" |
|
|
|
|
|
from __future__ import annotations |
|
|
|
import math |
|
from typing import Dict, Optional, Sequence, Tuple, Union |
|
|
|
import numpy as np |
|
import tensorflow as tf |
|
|
|
from ...activations_tf import get_tf_activation |
|
from ...modeling_tf_outputs import ( |
|
TFBaseModelOutput, |
|
TFMaskedLMOutput, |
|
TFQuestionAnsweringModelOutput, |
|
TFSequenceClassifierOutput, |
|
TFTokenClassifierOutput, |
|
) |
|
from ...modeling_tf_utils import ( |
|
TFMaskedLanguageModelingLoss, |
|
TFModelInputType, |
|
TFPreTrainedModel, |
|
TFQuestionAnsweringLoss, |
|
TFSequenceClassificationLoss, |
|
TFTokenClassificationLoss, |
|
get_initializer, |
|
unpack_inputs, |
|
) |
|
from ...tf_utils import check_embeddings_within_bounds, shape_list, stable_softmax |
|
from ...utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward, logging |
|
from .configuration_deberta import DebertaConfig |
|
|
|
|
|
logger = logging.get_logger(__name__) |
|
|
|
|
|
_CONFIG_FOR_DOC = "DebertaConfig" |
|
_CHECKPOINT_FOR_DOC = "kamalkraj/deberta-base" |
|
|
|
TF_DEBERTA_PRETRAINED_MODEL_ARCHIVE_LIST = [ |
|
"kamalkraj/deberta-base", |
|
|
|
] |
|
|
|
|
|
class TFDebertaContextPooler(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
self.dense = tf.keras.layers.Dense(config.pooler_hidden_size, name="dense") |
|
self.dropout = TFDebertaStableDropout(config.pooler_dropout, name="dropout") |
|
self.config = config |
|
|
|
def call(self, hidden_states, training: bool = False): |
|
|
|
|
|
context_token = hidden_states[:, 0] |
|
context_token = self.dropout(context_token, training=training) |
|
pooled_output = self.dense(context_token) |
|
pooled_output = get_tf_activation(self.config.pooler_hidden_act)(pooled_output) |
|
return pooled_output |
|
|
|
@property |
|
def output_dim(self) -> int: |
|
return self.config.hidden_size |
|
|
|
|
|
class TFDebertaXSoftmax(tf.keras.layers.Layer): |
|
""" |
|
Masked Softmax which is optimized for saving memory |
|
|
|
Args: |
|
input (`tf.Tensor`): The input tensor that will apply softmax. |
|
mask (`tf.Tensor`): The mask matrix where 0 indicate that element will be ignored in the softmax calculation. |
|
dim (int): The dimension that will apply softmax |
|
""" |
|
|
|
def __init__(self, axis=-1, **kwargs): |
|
super().__init__(**kwargs) |
|
self.axis = axis |
|
|
|
def call(self, inputs: tf.Tensor, mask: tf.Tensor): |
|
rmask = tf.logical_not(tf.cast(mask, tf.bool)) |
|
output = tf.where(rmask, float("-inf"), inputs) |
|
output = stable_softmax(output, self.axis) |
|
output = tf.where(rmask, 0.0, output) |
|
return output |
|
|
|
|
|
class TFDebertaStableDropout(tf.keras.layers.Layer): |
|
""" |
|
Optimized dropout module for stabilizing the training |
|
|
|
Args: |
|
drop_prob (float): the dropout probabilities |
|
""" |
|
|
|
def __init__(self, drop_prob, **kwargs): |
|
super().__init__(**kwargs) |
|
self.drop_prob = drop_prob |
|
|
|
@tf.custom_gradient |
|
def xdropout(self, inputs): |
|
""" |
|
Applies dropout to the inputs, as vanilla dropout, but also scales the remaining elements up by 1/drop_prob. |
|
""" |
|
mask = tf.cast( |
|
1 |
|
- tf.compat.v1.distributions.Bernoulli(probs=1.0 - self.drop_prob).sample(sample_shape=shape_list(inputs)), |
|
tf.bool, |
|
) |
|
scale = tf.convert_to_tensor(1.0 / (1 - self.drop_prob), dtype=tf.float32) |
|
if self.drop_prob > 0: |
|
inputs = tf.where(mask, 0.0, inputs) * scale |
|
|
|
def grad(upstream): |
|
if self.drop_prob > 0: |
|
return tf.where(mask, 0.0, upstream) * scale |
|
else: |
|
return upstream |
|
|
|
return inputs, grad |
|
|
|
def call(self, inputs: tf.Tensor, training: tf.Tensor = False): |
|
if training: |
|
return self.xdropout(inputs) |
|
return inputs |
|
|
|
|
|
class TFDebertaLayerNorm(tf.keras.layers.Layer): |
|
"""LayerNorm module in the TF style (epsilon inside the square root).""" |
|
|
|
def __init__(self, size, eps=1e-12, **kwargs): |
|
super().__init__(**kwargs) |
|
self.size = size |
|
self.eps = eps |
|
|
|
def build(self, input_shape): |
|
self.gamma = self.add_weight(shape=[self.size], initializer=tf.ones_initializer(), name="weight") |
|
self.beta = self.add_weight(shape=[self.size], initializer=tf.zeros_initializer(), name="bias") |
|
return super().build(input_shape) |
|
|
|
def call(self, x: tf.Tensor) -> tf.Tensor: |
|
mean = tf.reduce_mean(x, axis=[-1], keepdims=True) |
|
variance = tf.reduce_mean(tf.square(x - mean), axis=[-1], keepdims=True) |
|
std = tf.math.sqrt(variance + self.eps) |
|
return self.gamma * (x - mean) / std + self.beta |
|
|
|
|
|
class TFDebertaSelfOutput(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
self.dense = tf.keras.layers.Dense(config.hidden_size, name="dense") |
|
self.LayerNorm = tf.keras.layers.LayerNormalization(epsilon=config.layer_norm_eps, name="LayerNorm") |
|
self.dropout = TFDebertaStableDropout(config.hidden_dropout_prob, name="dropout") |
|
|
|
def call(self, hidden_states, input_tensor, training: bool = False): |
|
hidden_states = self.dense(hidden_states) |
|
hidden_states = self.dropout(hidden_states, training=training) |
|
hidden_states = self.LayerNorm(hidden_states + input_tensor) |
|
return hidden_states |
|
|
|
|
|
class TFDebertaAttention(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
self.self = TFDebertaDisentangledSelfAttention(config, name="self") |
|
self.dense_output = TFDebertaSelfOutput(config, name="output") |
|
self.config = config |
|
|
|
def call( |
|
self, |
|
input_tensor: tf.Tensor, |
|
attention_mask: tf.Tensor, |
|
query_states: tf.Tensor = None, |
|
relative_pos: tf.Tensor = None, |
|
rel_embeddings: tf.Tensor = None, |
|
output_attentions: bool = False, |
|
training: bool = False, |
|
) -> Tuple[tf.Tensor]: |
|
self_outputs = self.self( |
|
hidden_states=input_tensor, |
|
attention_mask=attention_mask, |
|
query_states=query_states, |
|
relative_pos=relative_pos, |
|
rel_embeddings=rel_embeddings, |
|
output_attentions=output_attentions, |
|
training=training, |
|
) |
|
if query_states is None: |
|
query_states = input_tensor |
|
attention_output = self.dense_output( |
|
hidden_states=self_outputs[0], input_tensor=query_states, training=training |
|
) |
|
|
|
output = (attention_output,) + self_outputs[1:] |
|
|
|
return output |
|
|
|
|
|
class TFDebertaIntermediate(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
|
|
self.dense = tf.keras.layers.Dense( |
|
units=config.intermediate_size, kernel_initializer=get_initializer(config.initializer_range), name="dense" |
|
) |
|
|
|
if isinstance(config.hidden_act, str): |
|
self.intermediate_act_fn = get_tf_activation(config.hidden_act) |
|
else: |
|
self.intermediate_act_fn = config.hidden_act |
|
|
|
def call(self, hidden_states: tf.Tensor) -> tf.Tensor: |
|
hidden_states = self.dense(inputs=hidden_states) |
|
hidden_states = self.intermediate_act_fn(hidden_states) |
|
|
|
return hidden_states |
|
|
|
|
|
class TFDebertaOutput(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
|
|
self.dense = tf.keras.layers.Dense( |
|
units=config.hidden_size, kernel_initializer=get_initializer(config.initializer_range), name="dense" |
|
) |
|
self.LayerNorm = tf.keras.layers.LayerNormalization(epsilon=config.layer_norm_eps, name="LayerNorm") |
|
self.dropout = TFDebertaStableDropout(config.hidden_dropout_prob, name="dropout") |
|
|
|
def call(self, hidden_states: tf.Tensor, input_tensor: tf.Tensor, training: bool = False) -> tf.Tensor: |
|
hidden_states = self.dense(inputs=hidden_states) |
|
hidden_states = self.dropout(hidden_states, training=training) |
|
hidden_states = self.LayerNorm(hidden_states + input_tensor) |
|
|
|
return hidden_states |
|
|
|
|
|
class TFDebertaLayer(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
|
|
self.attention = TFDebertaAttention(config, name="attention") |
|
self.intermediate = TFDebertaIntermediate(config, name="intermediate") |
|
self.bert_output = TFDebertaOutput(config, name="output") |
|
|
|
def call( |
|
self, |
|
hidden_states: tf.Tensor, |
|
attention_mask: tf.Tensor, |
|
query_states: tf.Tensor = None, |
|
relative_pos: tf.Tensor = None, |
|
rel_embeddings: tf.Tensor = None, |
|
output_attentions: bool = False, |
|
training: bool = False, |
|
) -> Tuple[tf.Tensor]: |
|
attention_outputs = self.attention( |
|
input_tensor=hidden_states, |
|
attention_mask=attention_mask, |
|
query_states=query_states, |
|
relative_pos=relative_pos, |
|
rel_embeddings=rel_embeddings, |
|
output_attentions=output_attentions, |
|
training=training, |
|
) |
|
attention_output = attention_outputs[0] |
|
intermediate_output = self.intermediate(hidden_states=attention_output) |
|
layer_output = self.bert_output( |
|
hidden_states=intermediate_output, input_tensor=attention_output, training=training |
|
) |
|
outputs = (layer_output,) + attention_outputs[1:] |
|
|
|
return outputs |
|
|
|
|
|
class TFDebertaEncoder(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
|
|
self.layer = [TFDebertaLayer(config, name=f"layer_._{i}") for i in range(config.num_hidden_layers)] |
|
self.relative_attention = getattr(config, "relative_attention", False) |
|
self.config = config |
|
if self.relative_attention: |
|
self.max_relative_positions = getattr(config, "max_relative_positions", -1) |
|
if self.max_relative_positions < 1: |
|
self.max_relative_positions = config.max_position_embeddings |
|
|
|
def build(self, input_shape): |
|
if self.relative_attention: |
|
self.rel_embeddings = self.add_weight( |
|
name="rel_embeddings.weight", |
|
shape=[self.max_relative_positions * 2, self.config.hidden_size], |
|
initializer=get_initializer(self.config.initializer_range), |
|
) |
|
return super().build(input_shape) |
|
|
|
def get_rel_embedding(self): |
|
rel_embeddings = self.rel_embeddings if self.relative_attention else None |
|
return rel_embeddings |
|
|
|
def get_attention_mask(self, attention_mask): |
|
if len(shape_list(attention_mask)) <= 2: |
|
extended_attention_mask = tf.expand_dims(tf.expand_dims(attention_mask, 1), 2) |
|
attention_mask = extended_attention_mask * tf.expand_dims(tf.squeeze(extended_attention_mask, -2), -1) |
|
attention_mask = tf.cast(attention_mask, tf.uint8) |
|
elif len(shape_list(attention_mask)) == 3: |
|
attention_mask = tf.expand_dims(attention_mask, 1) |
|
|
|
return attention_mask |
|
|
|
def get_rel_pos(self, hidden_states, query_states=None, relative_pos=None): |
|
if self.relative_attention and relative_pos is None: |
|
q = shape_list(query_states)[-2] if query_states is not None else shape_list(hidden_states)[-2] |
|
relative_pos = build_relative_position(q, shape_list(hidden_states)[-2]) |
|
return relative_pos |
|
|
|
def call( |
|
self, |
|
hidden_states: tf.Tensor, |
|
attention_mask: tf.Tensor, |
|
query_states: tf.Tensor = None, |
|
relative_pos: tf.Tensor = None, |
|
output_attentions: bool = False, |
|
output_hidden_states: bool = False, |
|
return_dict: bool = True, |
|
training: bool = False, |
|
) -> Union[TFBaseModelOutput, Tuple[tf.Tensor]]: |
|
all_hidden_states = () if output_hidden_states else None |
|
all_attentions = () if output_attentions else None |
|
|
|
attention_mask = self.get_attention_mask(attention_mask) |
|
relative_pos = self.get_rel_pos(hidden_states, query_states, relative_pos) |
|
|
|
if isinstance(hidden_states, Sequence): |
|
next_kv = hidden_states[0] |
|
else: |
|
next_kv = hidden_states |
|
|
|
rel_embeddings = self.get_rel_embedding() |
|
|
|
for i, layer_module in enumerate(self.layer): |
|
if output_hidden_states: |
|
all_hidden_states = all_hidden_states + (hidden_states,) |
|
|
|
layer_outputs = layer_module( |
|
hidden_states=next_kv, |
|
attention_mask=attention_mask, |
|
query_states=query_states, |
|
relative_pos=relative_pos, |
|
rel_embeddings=rel_embeddings, |
|
output_attentions=output_attentions, |
|
training=training, |
|
) |
|
hidden_states = layer_outputs[0] |
|
|
|
if query_states is not None: |
|
query_states = hidden_states |
|
if isinstance(hidden_states, Sequence): |
|
next_kv = hidden_states[i + 1] if i + 1 < len(self.layer) else None |
|
else: |
|
next_kv = hidden_states |
|
|
|
if output_attentions: |
|
all_attentions = all_attentions + (layer_outputs[1],) |
|
|
|
|
|
if output_hidden_states: |
|
all_hidden_states = all_hidden_states + (hidden_states,) |
|
|
|
if not return_dict: |
|
return tuple(v for v in [hidden_states, all_hidden_states, all_attentions] if v is not None) |
|
|
|
return TFBaseModelOutput( |
|
last_hidden_state=hidden_states, hidden_states=all_hidden_states, attentions=all_attentions |
|
) |
|
|
|
|
|
def build_relative_position(query_size, key_size): |
|
""" |
|
Build relative position according to the query and key |
|
|
|
We assume the absolute position of query \\(P_q\\) is range from (0, query_size) and the absolute position of key |
|
\\(P_k\\) is range from (0, key_size), The relative positions from query to key is \\(R_{q \\rightarrow k} = P_q - |
|
P_k\\) |
|
|
|
Args: |
|
query_size (int): the length of query |
|
key_size (int): the length of key |
|
|
|
Return: |
|
`tf.Tensor`: A tensor with shape [1, query_size, key_size] |
|
|
|
""" |
|
q_ids = tf.range(query_size, dtype=tf.int32) |
|
k_ids = tf.range(key_size, dtype=tf.int32) |
|
rel_pos_ids = q_ids[:, None] - tf.tile(tf.reshape(k_ids, [1, -1]), [query_size, 1]) |
|
rel_pos_ids = rel_pos_ids[:query_size, :] |
|
rel_pos_ids = tf.expand_dims(rel_pos_ids, axis=0) |
|
return tf.cast(rel_pos_ids, tf.int64) |
|
|
|
|
|
def c2p_dynamic_expand(c2p_pos, query_layer, relative_pos): |
|
shapes = [ |
|
shape_list(query_layer)[0], |
|
shape_list(query_layer)[1], |
|
shape_list(query_layer)[2], |
|
shape_list(relative_pos)[-1], |
|
] |
|
return tf.broadcast_to(c2p_pos, shapes) |
|
|
|
|
|
def p2c_dynamic_expand(c2p_pos, query_layer, key_layer): |
|
shapes = [ |
|
shape_list(query_layer)[0], |
|
shape_list(query_layer)[1], |
|
shape_list(key_layer)[-2], |
|
shape_list(key_layer)[-2], |
|
] |
|
return tf.broadcast_to(c2p_pos, shapes) |
|
|
|
|
|
def pos_dynamic_expand(pos_index, p2c_att, key_layer): |
|
shapes = shape_list(p2c_att)[:2] + [shape_list(pos_index)[-2], shape_list(key_layer)[-2]] |
|
return tf.broadcast_to(pos_index, shapes) |
|
|
|
|
|
def torch_gather(x, indices, gather_axis): |
|
if gather_axis < 0: |
|
gather_axis = tf.rank(x) + gather_axis |
|
|
|
if gather_axis != tf.rank(x) - 1: |
|
pre_roll = tf.rank(x) - 1 - gather_axis |
|
permutation = tf.roll(tf.range(tf.rank(x)), pre_roll, axis=0) |
|
x = tf.transpose(x, perm=permutation) |
|
indices = tf.transpose(indices, perm=permutation) |
|
else: |
|
pre_roll = 0 |
|
|
|
flat_x = tf.reshape(x, (-1, tf.shape(x)[-1])) |
|
flat_indices = tf.reshape(indices, (-1, tf.shape(indices)[-1])) |
|
gathered = tf.gather(flat_x, flat_indices, batch_dims=1) |
|
gathered = tf.reshape(gathered, tf.shape(indices)) |
|
|
|
if pre_roll != 0: |
|
permutation = tf.roll(tf.range(tf.rank(x)), -pre_roll, axis=0) |
|
gathered = tf.transpose(gathered, perm=permutation) |
|
|
|
return gathered |
|
|
|
|
|
class TFDebertaDisentangledSelfAttention(tf.keras.layers.Layer): |
|
""" |
|
Disentangled self-attention module |
|
|
|
Parameters: |
|
config (`str`): |
|
A model config class instance with the configuration to build a new model. The schema is similar to |
|
*BertConfig*, for more details, please refer [`DebertaConfig`] |
|
|
|
""" |
|
|
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
if config.hidden_size % config.num_attention_heads != 0: |
|
raise ValueError( |
|
f"The hidden size ({config.hidden_size}) is not a multiple of the number of attention " |
|
f"heads ({config.num_attention_heads})" |
|
) |
|
self.num_attention_heads = config.num_attention_heads |
|
self.attention_head_size = int(config.hidden_size / config.num_attention_heads) |
|
self.all_head_size = self.num_attention_heads * self.attention_head_size |
|
self.in_proj = tf.keras.layers.Dense( |
|
self.all_head_size * 3, |
|
kernel_initializer=get_initializer(config.initializer_range), |
|
name="in_proj", |
|
use_bias=False, |
|
) |
|
self.pos_att_type = config.pos_att_type if config.pos_att_type is not None else [] |
|
|
|
self.relative_attention = getattr(config, "relative_attention", False) |
|
self.talking_head = getattr(config, "talking_head", False) |
|
|
|
if self.talking_head: |
|
self.head_logits_proj = tf.keras.layers.Dense( |
|
self.num_attention_heads, |
|
kernel_initializer=get_initializer(config.initializer_range), |
|
name="head_logits_proj", |
|
use_bias=False, |
|
) |
|
self.head_weights_proj = tf.keras.layers.Dense( |
|
self.num_attention_heads, |
|
kernel_initializer=get_initializer(config.initializer_range), |
|
name="head_weights_proj", |
|
use_bias=False, |
|
) |
|
|
|
self.softmax = TFDebertaXSoftmax(axis=-1) |
|
|
|
if self.relative_attention: |
|
self.max_relative_positions = getattr(config, "max_relative_positions", -1) |
|
if self.max_relative_positions < 1: |
|
self.max_relative_positions = config.max_position_embeddings |
|
self.pos_dropout = TFDebertaStableDropout(config.hidden_dropout_prob, name="pos_dropout") |
|
if "c2p" in self.pos_att_type: |
|
self.pos_proj = tf.keras.layers.Dense( |
|
self.all_head_size, |
|
kernel_initializer=get_initializer(config.initializer_range), |
|
name="pos_proj", |
|
use_bias=False, |
|
) |
|
if "p2c" in self.pos_att_type: |
|
self.pos_q_proj = tf.keras.layers.Dense( |
|
self.all_head_size, kernel_initializer=get_initializer(config.initializer_range), name="pos_q_proj" |
|
) |
|
|
|
self.dropout = TFDebertaStableDropout(config.attention_probs_dropout_prob, name="dropout") |
|
|
|
def build(self, input_shape): |
|
self.q_bias = self.add_weight( |
|
name="q_bias", shape=(self.all_head_size), initializer=tf.keras.initializers.Zeros() |
|
) |
|
self.v_bias = self.add_weight( |
|
name="v_bias", shape=(self.all_head_size), initializer=tf.keras.initializers.Zeros() |
|
) |
|
return super().build(input_shape) |
|
|
|
def transpose_for_scores(self, tensor: tf.Tensor) -> tf.Tensor: |
|
shape = shape_list(tensor)[:-1] + [self.num_attention_heads, -1] |
|
|
|
tensor = tf.reshape(tensor=tensor, shape=shape) |
|
|
|
|
|
return tf.transpose(tensor, perm=[0, 2, 1, 3]) |
|
|
|
def call( |
|
self, |
|
hidden_states: tf.Tensor, |
|
attention_mask: tf.Tensor, |
|
query_states: tf.Tensor = None, |
|
relative_pos: tf.Tensor = None, |
|
rel_embeddings: tf.Tensor = None, |
|
output_attentions: bool = False, |
|
training: bool = False, |
|
) -> Tuple[tf.Tensor]: |
|
""" |
|
Call the module |
|
|
|
Args: |
|
hidden_states (`tf.Tensor`): |
|
Input states to the module usually the output from previous layer, it will be the Q,K and V in |
|
*Attention(Q,K,V)* |
|
|
|
attention_mask (`tf.Tensor`): |
|
An attention mask matrix of shape [*B*, *N*, *N*] where *B* is the batch size, *N* is the maximum |
|
sequence length in which element [i,j] = *1* means the *i* th token in the input can attend to the *j* |
|
th token. |
|
|
|
return_att (`bool`, optional): |
|
Whether return the attention matrix. |
|
|
|
query_states (`tf.Tensor`, optional): |
|
The *Q* state in *Attention(Q,K,V)*. |
|
|
|
relative_pos (`tf.Tensor`): |
|
The relative position encoding between the tokens in the sequence. It's of shape [*B*, *N*, *N*] with |
|
values ranging in [*-max_relative_positions*, *max_relative_positions*]. |
|
|
|
rel_embeddings (`tf.Tensor`): |
|
The embedding of relative distances. It's a tensor of shape [\\(2 \\times |
|
\\text{max_relative_positions}\\), *hidden_size*]. |
|
|
|
|
|
""" |
|
if query_states is None: |
|
qp = self.in_proj(hidden_states) |
|
query_layer, key_layer, value_layer = tf.split( |
|
self.transpose_for_scores(qp), num_or_size_splits=3, axis=-1 |
|
) |
|
else: |
|
|
|
def linear(w, b, x): |
|
out = tf.matmul(x, w, transpose_b=True) |
|
if b is not None: |
|
out += tf.transpose(b) |
|
return out |
|
|
|
ws = tf.split( |
|
tf.transpose(self.in_proj.weight[0]), num_or_size_splits=self.num_attention_heads * 3, axis=0 |
|
) |
|
qkvw = tf.TensorArray(dtype=tf.float32, size=3) |
|
for k in tf.range(3): |
|
qkvw_inside = tf.TensorArray(dtype=tf.float32, size=self.num_attention_heads) |
|
for i in tf.range(self.num_attention_heads): |
|
qkvw_inside = qkvw_inside.write(i, ws[i * 3 + k]) |
|
qkvw = qkvw.write(k, qkvw_inside.concat()) |
|
qkvb = [None] * 3 |
|
|
|
q = linear(qkvw[0], qkvb[0], query_states) |
|
k = linear(qkvw[1], qkvb[1], hidden_states) |
|
v = linear(qkvw[2], qkvb[2], hidden_states) |
|
query_layer = self.transpose_for_scores(q) |
|
key_layer = self.transpose_for_scores(k) |
|
value_layer = self.transpose_for_scores(v) |
|
|
|
query_layer = query_layer + self.transpose_for_scores(self.q_bias[None, None, :]) |
|
value_layer = value_layer + self.transpose_for_scores(self.v_bias[None, None, :]) |
|
|
|
rel_att = None |
|
|
|
scale_factor = 1 + len(self.pos_att_type) |
|
scale = math.sqrt(shape_list(query_layer)[-1] * scale_factor) |
|
query_layer = query_layer / scale |
|
|
|
attention_scores = tf.matmul(query_layer, tf.transpose(key_layer, [0, 1, 3, 2])) |
|
if self.relative_attention: |
|
rel_embeddings = self.pos_dropout(rel_embeddings, training=training) |
|
rel_att = self.disentangled_att_bias(query_layer, key_layer, relative_pos, rel_embeddings, scale_factor) |
|
|
|
if rel_att is not None: |
|
attention_scores = attention_scores + rel_att |
|
|
|
if self.talking_head: |
|
attention_scores = tf.transpose( |
|
self.head_logits_proj(tf.transpose(attention_scores, [0, 2, 3, 1])), [0, 3, 1, 2] |
|
) |
|
|
|
attention_probs = self.softmax(attention_scores, attention_mask) |
|
attention_probs = self.dropout(attention_probs, training=training) |
|
if self.talking_head: |
|
attention_probs = tf.transpose( |
|
self.head_weights_proj(tf.transpose(attention_probs, [0, 2, 3, 1])), [0, 3, 1, 2] |
|
) |
|
|
|
context_layer = tf.matmul(attention_probs, value_layer) |
|
context_layer = tf.transpose(context_layer, [0, 2, 1, 3]) |
|
context_layer_shape = shape_list(context_layer) |
|
|
|
|
|
|
|
|
|
new_context_layer_shape = context_layer_shape[:-2] + [context_layer_shape[-2] * context_layer_shape[-1]] |
|
context_layer = tf.reshape(context_layer, new_context_layer_shape) |
|
outputs = (context_layer, attention_probs) if output_attentions else (context_layer,) |
|
return outputs |
|
|
|
def disentangled_att_bias(self, query_layer, key_layer, relative_pos, rel_embeddings, scale_factor): |
|
if relative_pos is None: |
|
q = shape_list(query_layer)[-2] |
|
relative_pos = build_relative_position(q, shape_list(key_layer)[-2]) |
|
shape_list_pos = shape_list(relative_pos) |
|
if len(shape_list_pos) == 2: |
|
relative_pos = tf.expand_dims(tf.expand_dims(relative_pos, 0), 0) |
|
elif len(shape_list_pos) == 3: |
|
relative_pos = tf.expand_dims(relative_pos, 1) |
|
|
|
elif len(shape_list_pos) != 4: |
|
raise ValueError(f"Relative position ids must be of dim 2 or 3 or 4. {len(shape_list_pos)}") |
|
|
|
att_span = tf.cast( |
|
tf.minimum( |
|
tf.maximum(shape_list(query_layer)[-2], shape_list(key_layer)[-2]), self.max_relative_positions |
|
), |
|
tf.int64, |
|
) |
|
rel_embeddings = tf.expand_dims( |
|
rel_embeddings[self.max_relative_positions - att_span : self.max_relative_positions + att_span, :], 0 |
|
) |
|
|
|
score = 0 |
|
|
|
|
|
if "c2p" in self.pos_att_type: |
|
pos_key_layer = self.pos_proj(rel_embeddings) |
|
pos_key_layer = self.transpose_for_scores(pos_key_layer) |
|
c2p_att = tf.matmul(query_layer, tf.transpose(pos_key_layer, [0, 1, 3, 2])) |
|
c2p_pos = tf.clip_by_value(relative_pos + att_span, 0, att_span * 2 - 1) |
|
c2p_att = torch_gather(c2p_att, c2p_dynamic_expand(c2p_pos, query_layer, relative_pos), -1) |
|
score += c2p_att |
|
|
|
|
|
if "p2c" in self.pos_att_type: |
|
pos_query_layer = self.pos_q_proj(rel_embeddings) |
|
pos_query_layer = self.transpose_for_scores(pos_query_layer) |
|
pos_query_layer /= tf.math.sqrt(tf.cast(shape_list(pos_query_layer)[-1] * scale_factor, dtype=tf.float32)) |
|
if shape_list(query_layer)[-2] != shape_list(key_layer)[-2]: |
|
r_pos = build_relative_position(shape_list(key_layer)[-2], shape_list(key_layer)[-2]) |
|
else: |
|
r_pos = relative_pos |
|
p2c_pos = tf.clip_by_value(-r_pos + att_span, 0, att_span * 2 - 1) |
|
p2c_att = tf.matmul(key_layer, tf.transpose(pos_query_layer, [0, 1, 3, 2])) |
|
p2c_att = tf.transpose( |
|
torch_gather(p2c_att, p2c_dynamic_expand(p2c_pos, query_layer, key_layer), -1), [0, 1, 3, 2] |
|
) |
|
if shape_list(query_layer)[-2] != shape_list(key_layer)[-2]: |
|
pos_index = tf.expand_dims(relative_pos[:, :, :, 0], -1) |
|
p2c_att = torch_gather(p2c_att, pos_dynamic_expand(pos_index, p2c_att, key_layer), -2) |
|
score += p2c_att |
|
|
|
return score |
|
|
|
|
|
class TFDebertaEmbeddings(tf.keras.layers.Layer): |
|
"""Construct the embeddings from word, position and token_type embeddings.""" |
|
|
|
def __init__(self, config, **kwargs): |
|
super().__init__(**kwargs) |
|
|
|
self.config = config |
|
self.embedding_size = getattr(config, "embedding_size", config.hidden_size) |
|
self.hidden_size = config.hidden_size |
|
self.max_position_embeddings = config.max_position_embeddings |
|
self.position_biased_input = getattr(config, "position_biased_input", True) |
|
self.initializer_range = config.initializer_range |
|
if self.embedding_size != config.hidden_size: |
|
self.embed_proj = tf.keras.layers.Dense( |
|
config.hidden_size, |
|
kernel_initializer=get_initializer(config.initializer_range), |
|
name="embed_proj", |
|
use_bias=False, |
|
) |
|
self.LayerNorm = tf.keras.layers.LayerNormalization(epsilon=config.layer_norm_eps, name="LayerNorm") |
|
self.dropout = TFDebertaStableDropout(config.hidden_dropout_prob, name="dropout") |
|
|
|
def build(self, input_shape: tf.TensorShape): |
|
with tf.name_scope("word_embeddings"): |
|
self.weight = self.add_weight( |
|
name="weight", |
|
shape=[self.config.vocab_size, self.embedding_size], |
|
initializer=get_initializer(self.initializer_range), |
|
) |
|
|
|
with tf.name_scope("token_type_embeddings"): |
|
if self.config.type_vocab_size > 0: |
|
self.token_type_embeddings = self.add_weight( |
|
name="embeddings", |
|
shape=[self.config.type_vocab_size, self.embedding_size], |
|
initializer=get_initializer(self.initializer_range), |
|
) |
|
else: |
|
self.token_type_embeddings = None |
|
|
|
with tf.name_scope("position_embeddings"): |
|
if self.position_biased_input: |
|
self.position_embeddings = self.add_weight( |
|
name="embeddings", |
|
shape=[self.max_position_embeddings, self.hidden_size], |
|
initializer=get_initializer(self.initializer_range), |
|
) |
|
else: |
|
self.position_embeddings = None |
|
|
|
super().build(input_shape) |
|
|
|
def call( |
|
self, |
|
input_ids: tf.Tensor = None, |
|
position_ids: tf.Tensor = None, |
|
token_type_ids: tf.Tensor = None, |
|
inputs_embeds: tf.Tensor = None, |
|
mask: tf.Tensor = None, |
|
training: bool = False, |
|
) -> tf.Tensor: |
|
""" |
|
Applies embedding based on inputs tensor. |
|
|
|
Returns: |
|
final_embeddings (`tf.Tensor`): output embedding tensor. |
|
""" |
|
if input_ids is None and inputs_embeds is None: |
|
raise ValueError("Need to provide either `input_ids` or `input_embeds`.") |
|
|
|
if input_ids is not None: |
|
check_embeddings_within_bounds(input_ids, self.config.vocab_size) |
|
inputs_embeds = tf.gather(params=self.weight, indices=input_ids) |
|
|
|
input_shape = shape_list(inputs_embeds)[:-1] |
|
|
|
if token_type_ids is None: |
|
token_type_ids = tf.fill(dims=input_shape, value=0) |
|
|
|
if position_ids is None: |
|
position_ids = tf.expand_dims(tf.range(start=0, limit=input_shape[-1]), axis=0) |
|
|
|
final_embeddings = inputs_embeds |
|
if self.position_biased_input: |
|
position_embeds = tf.gather(params=self.position_embeddings, indices=position_ids) |
|
final_embeddings += position_embeds |
|
if self.config.type_vocab_size > 0: |
|
token_type_embeds = tf.gather(params=self.token_type_embeddings, indices=token_type_ids) |
|
final_embeddings += token_type_embeds |
|
|
|
if self.embedding_size != self.hidden_size: |
|
final_embeddings = self.embed_proj(final_embeddings) |
|
|
|
final_embeddings = self.LayerNorm(final_embeddings) |
|
|
|
if mask is not None: |
|
if len(shape_list(mask)) != len(shape_list(final_embeddings)): |
|
if len(shape_list(mask)) == 4: |
|
mask = tf.squeeze(tf.squeeze(mask, axis=1), axis=1) |
|
mask = tf.cast(tf.expand_dims(mask, axis=2), tf.float32) |
|
|
|
final_embeddings = final_embeddings * mask |
|
|
|
final_embeddings = self.dropout(final_embeddings, training=training) |
|
|
|
return final_embeddings |
|
|
|
|
|
class TFDebertaPredictionHeadTransform(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
|
|
self.embedding_size = getattr(config, "embedding_size", config.hidden_size) |
|
|
|
self.dense = tf.keras.layers.Dense( |
|
units=self.embedding_size, |
|
kernel_initializer=get_initializer(config.initializer_range), |
|
name="dense", |
|
) |
|
|
|
if isinstance(config.hidden_act, str): |
|
self.transform_act_fn = get_tf_activation(config.hidden_act) |
|
else: |
|
self.transform_act_fn = config.hidden_act |
|
self.LayerNorm = tf.keras.layers.LayerNormalization(epsilon=config.layer_norm_eps, name="LayerNorm") |
|
|
|
def call(self, hidden_states: tf.Tensor) -> tf.Tensor: |
|
hidden_states = self.dense(inputs=hidden_states) |
|
hidden_states = self.transform_act_fn(hidden_states) |
|
hidden_states = self.LayerNorm(hidden_states) |
|
|
|
return hidden_states |
|
|
|
|
|
class TFDebertaLMPredictionHead(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, input_embeddings: tf.keras.layers.Layer, **kwargs): |
|
super().__init__(**kwargs) |
|
|
|
self.config = config |
|
self.embedding_size = getattr(config, "embedding_size", config.hidden_size) |
|
|
|
self.transform = TFDebertaPredictionHeadTransform(config, name="transform") |
|
|
|
|
|
|
|
self.input_embeddings = input_embeddings |
|
|
|
def build(self, input_shape: tf.TensorShape): |
|
self.bias = self.add_weight(shape=(self.config.vocab_size,), initializer="zeros", trainable=True, name="bias") |
|
|
|
super().build(input_shape) |
|
|
|
def get_output_embeddings(self) -> tf.keras.layers.Layer: |
|
return self.input_embeddings |
|
|
|
def set_output_embeddings(self, value: tf.Variable): |
|
self.input_embeddings.weight = value |
|
self.input_embeddings.vocab_size = shape_list(value)[0] |
|
|
|
def get_bias(self) -> Dict[str, tf.Variable]: |
|
return {"bias": self.bias} |
|
|
|
def set_bias(self, value: tf.Variable): |
|
self.bias = value["bias"] |
|
self.config.vocab_size = shape_list(value["bias"])[0] |
|
|
|
def call(self, hidden_states: tf.Tensor) -> tf.Tensor: |
|
hidden_states = self.transform(hidden_states=hidden_states) |
|
seq_length = shape_list(hidden_states)[1] |
|
hidden_states = tf.reshape(tensor=hidden_states, shape=[-1, self.embedding_size]) |
|
hidden_states = tf.matmul(a=hidden_states, b=self.input_embeddings.weight, transpose_b=True) |
|
hidden_states = tf.reshape(tensor=hidden_states, shape=[-1, seq_length, self.config.vocab_size]) |
|
hidden_states = tf.nn.bias_add(value=hidden_states, bias=self.bias) |
|
|
|
return hidden_states |
|
|
|
|
|
class TFDebertaOnlyMLMHead(tf.keras.layers.Layer): |
|
def __init__(self, config: DebertaConfig, input_embeddings: tf.keras.layers.Layer, **kwargs): |
|
super().__init__(**kwargs) |
|
self.predictions = TFDebertaLMPredictionHead(config, input_embeddings, name="predictions") |
|
|
|
def call(self, sequence_output: tf.Tensor) -> tf.Tensor: |
|
prediction_scores = self.predictions(hidden_states=sequence_output) |
|
|
|
return prediction_scores |
|
|
|
|
|
|
|
class TFDebertaMainLayer(tf.keras.layers.Layer): |
|
config_class = DebertaConfig |
|
|
|
def __init__(self, config: DebertaConfig, **kwargs): |
|
super().__init__(**kwargs) |
|
|
|
self.config = config |
|
|
|
self.embeddings = TFDebertaEmbeddings(config, name="embeddings") |
|
self.encoder = TFDebertaEncoder(config, name="encoder") |
|
|
|
def get_input_embeddings(self) -> tf.keras.layers.Layer: |
|
return self.embeddings |
|
|
|
def set_input_embeddings(self, value: tf.Variable): |
|
self.embeddings.weight = value |
|
self.embeddings.vocab_size = shape_list(value)[0] |
|
|
|
def _prune_heads(self, heads_to_prune): |
|
""" |
|
Prunes heads of the model. heads_to_prune: dict of {layer_num: list of heads to prune in this layer} See base |
|
class PreTrainedModel |
|
""" |
|
raise NotImplementedError |
|
|
|
@unpack_inputs |
|
def call( |
|
self, |
|
input_ids: TFModelInputType | None = None, |
|
attention_mask: np.ndarray | tf.Tensor | None = None, |
|
token_type_ids: np.ndarray | tf.Tensor | None = None, |
|
position_ids: np.ndarray | tf.Tensor | None = None, |
|
inputs_embeds: np.ndarray | tf.Tensor | None = None, |
|
output_attentions: Optional[bool] = None, |
|
output_hidden_states: Optional[bool] = None, |
|
return_dict: Optional[bool] = None, |
|
training: bool = False, |
|
) -> Union[TFBaseModelOutput, Tuple[tf.Tensor]]: |
|
if input_ids is not None and inputs_embeds is not None: |
|
raise ValueError("You cannot specify both input_ids and inputs_embeds at the same time") |
|
elif input_ids is not None: |
|
input_shape = shape_list(input_ids) |
|
elif inputs_embeds is not None: |
|
input_shape = shape_list(inputs_embeds)[:-1] |
|
else: |
|
raise ValueError("You have to specify either input_ids or inputs_embeds") |
|
|
|
if attention_mask is None: |
|
attention_mask = tf.fill(dims=input_shape, value=1) |
|
|
|
if token_type_ids is None: |
|
token_type_ids = tf.fill(dims=input_shape, value=0) |
|
|
|
embedding_output = self.embeddings( |
|
input_ids=input_ids, |
|
position_ids=position_ids, |
|
token_type_ids=token_type_ids, |
|
inputs_embeds=inputs_embeds, |
|
mask=attention_mask, |
|
training=training, |
|
) |
|
|
|
encoder_outputs = self.encoder( |
|
hidden_states=embedding_output, |
|
attention_mask=attention_mask, |
|
output_attentions=output_attentions, |
|
output_hidden_states=output_hidden_states, |
|
return_dict=return_dict, |
|
training=training, |
|
) |
|
|
|
sequence_output = encoder_outputs[0] |
|
|
|
if not return_dict: |
|
return (sequence_output,) + encoder_outputs[1:] |
|
|
|
return TFBaseModelOutput( |
|
last_hidden_state=sequence_output, |
|
hidden_states=encoder_outputs.hidden_states, |
|
attentions=encoder_outputs.attentions, |
|
) |
|
|
|
|
|
class TFDebertaPreTrainedModel(TFPreTrainedModel): |
|
""" |
|
An abstract class to handle weights initialization and a simple interface for downloading and loading pretrained |
|
models. |
|
""" |
|
|
|
config_class = DebertaConfig |
|
base_model_prefix = "deberta" |
|
|
|
|
|
DEBERTA_START_DOCSTRING = r""" |
|
The DeBERTa model was proposed in [DeBERTa: Decoding-enhanced BERT with Disentangled |
|
Attention](https://arxiv.org/abs/2006.03654) by Pengcheng He, Xiaodong Liu, Jianfeng Gao, Weizhu Chen. It's build |
|
on top of BERT/RoBERTa with two improvements, i.e. disentangled attention and enhanced mask decoder. With those two |
|
improvements, it out perform BERT/RoBERTa on a majority of tasks with 80GB pretraining data. |
|
|
|
This model is also a [tf.keras.Model](https://www.tensorflow.org/api_docs/python/tf/keras/Model) subclass. Use it |
|
as a regular TF 2.0 Keras Model and refer to the TF 2.0 documentation for all matter related to general usage and |
|
behavior. |
|
|
|
<Tip> |
|
|
|
TensorFlow models and layers in `transformers` accept two formats as input: |
|
|
|
- having all inputs as keyword arguments (like PyTorch models), or |
|
- having all inputs as a list, tuple or dict in the first positional argument. |
|
|
|
The reason the second format is supported is that Keras methods prefer this format when passing inputs to models |
|
and layers. Because of this support, when using methods like `model.fit()` things should "just work" for you - just |
|
pass your inputs and labels in any format that `model.fit()` supports! If, however, you want to use the second |
|
format outside of Keras methods like `fit()` and `predict()`, such as when creating your own layers or models with |
|
the Keras `Functional` API, there are three possibilities you can use to gather all the input Tensors in the first |
|
positional argument: |
|
|
|
- a single Tensor with `input_ids` only and nothing else: `model(input_ids)` |
|
- a list of varying length with one or several input Tensors IN THE ORDER given in the docstring: |
|
`model([input_ids, attention_mask])` or `model([input_ids, attention_mask, token_type_ids])` |
|
- a dictionary with one or several input Tensors associated to the input names given in the docstring: |
|
`model({"input_ids": input_ids, "token_type_ids": token_type_ids})` |
|
|
|
Note that when creating models and layers with |
|
[subclassing](https://keras.io/guides/making_new_layers_and_models_via_subclassing/) then you don't need to worry |
|
about any of this, as you can just pass inputs like you would to any other Python function! |
|
|
|
</Tip> |
|
|
|
Parameters: |
|
config ([`DebertaConfig`]): Model configuration class with all the parameters of the model. |
|
Initializing with a config file does not load the weights associated with the model, only the |
|
configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights. |
|
""" |
|
|
|
DEBERTA_INPUTS_DOCSTRING = r""" |
|
Args: |
|
input_ids (`np.ndarray`, `tf.Tensor`, `List[tf.Tensor]` ``Dict[str, tf.Tensor]` or `Dict[str, np.ndarray]` and each example must have the shape `({0})`): |
|
Indices of input sequence tokens in the vocabulary. |
|
|
|
Indices can be obtained using [`AutoTokenizer`]. See [`PreTrainedTokenizer.encode`] and |
|
[`PreTrainedTokenizer.__call__`] for details. |
|
|
|
[What are input IDs?](../glossary#input-ids) |
|
attention_mask (`np.ndarray` or `tf.Tensor` of shape `({0})`, *optional*): |
|
Mask to avoid performing attention on padding token indices. Mask values selected in `[0, 1]`: |
|
|
|
- 1 for tokens that are **not masked**, |
|
- 0 for tokens that are **masked**. |
|
|
|
[What are attention masks?](../glossary#attention-mask) |
|
token_type_ids (`np.ndarray` or `tf.Tensor` of shape `({0})`, *optional*): |
|
Segment token indices to indicate first and second portions of the inputs. Indices are selected in `[0, |
|
1]`: |
|
|
|
- 0 corresponds to a *sentence A* token, |
|
- 1 corresponds to a *sentence B* token. |
|
|
|
[What are token type IDs?](../glossary#token-type-ids) |
|
position_ids (`np.ndarray` or `tf.Tensor` of shape `({0})`, *optional*): |
|
Indices of positions of each input sequence tokens in the position embeddings. Selected in the range `[0, |
|
config.max_position_embeddings - 1]`. |
|
|
|
[What are position IDs?](../glossary#position-ids) |
|
inputs_embeds (`np.ndarray` or `tf.Tensor` of shape `({0}, hidden_size)`, *optional*): |
|
Optionally, instead of passing `input_ids` you can choose to directly pass an embedded representation. This |
|
is useful if you want more control over how to convert *input_ids* indices into associated vectors than the |
|
model's internal embedding lookup matrix. |
|
output_attentions (`bool`, *optional*): |
|
Whether or not to return the attentions tensors of all attention layers. See `attentions` under returned |
|
tensors for more detail. |
|
output_hidden_states (`bool`, *optional*): |
|
Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for |
|
more detail. |
|
return_dict (`bool`, *optional*): |
|
Whether or not to return a [`~utils.ModelOutput``] instead of a plain tuple. |
|
""" |
|
|
|
|
|
@add_start_docstrings( |
|
"The bare DeBERTa Model transformer outputting raw hidden-states without any specific head on top.", |
|
DEBERTA_START_DOCSTRING, |
|
) |
|
class TFDebertaModel(TFDebertaPreTrainedModel): |
|
def __init__(self, config: DebertaConfig, *inputs, **kwargs): |
|
super().__init__(config, *inputs, **kwargs) |
|
|
|
self.deberta = TFDebertaMainLayer(config, name="deberta") |
|
|
|
@unpack_inputs |
|
@add_start_docstrings_to_model_forward(DEBERTA_INPUTS_DOCSTRING.format("batch_size, sequence_length")) |
|
@add_code_sample_docstrings( |
|
checkpoint=_CHECKPOINT_FOR_DOC, |
|
output_type=TFBaseModelOutput, |
|
config_class=_CONFIG_FOR_DOC, |
|
) |
|
def call( |
|
self, |
|
input_ids: TFModelInputType | None = None, |
|
attention_mask: np.ndarray | tf.Tensor | None = None, |
|
token_type_ids: np.ndarray | tf.Tensor | None = None, |
|
position_ids: np.ndarray | tf.Tensor | None = None, |
|
inputs_embeds: np.ndarray | tf.Tensor | None = None, |
|
output_attentions: Optional[bool] = None, |
|
output_hidden_states: Optional[bool] = None, |
|
return_dict: Optional[bool] = None, |
|
training: Optional[bool] = False, |
|
) -> Union[TFBaseModelOutput, Tuple[tf.Tensor]]: |
|
outputs = self.deberta( |
|
input_ids=input_ids, |
|
attention_mask=attention_mask, |
|
token_type_ids=token_type_ids, |
|
position_ids=position_ids, |
|
inputs_embeds=inputs_embeds, |
|
output_attentions=output_attentions, |
|
output_hidden_states=output_hidden_states, |
|
return_dict=return_dict, |
|
training=training, |
|
) |
|
|
|
return outputs |
|
|
|
|
|
@add_start_docstrings("""DeBERTa Model with a `language modeling` head on top.""", DEBERTA_START_DOCSTRING) |
|
class TFDebertaForMaskedLM(TFDebertaPreTrainedModel, TFMaskedLanguageModelingLoss): |
|
def __init__(self, config: DebertaConfig, *inputs, **kwargs): |
|
super().__init__(config, *inputs, **kwargs) |
|
|
|
if config.is_decoder: |
|
logger.warning( |
|
"If you want to use `TFDebertaForMaskedLM` make sure `config.is_decoder=False` for " |
|
"bi-directional self-attention." |
|
) |
|
|
|
self.deberta = TFDebertaMainLayer(config, name="deberta") |
|
self.mlm = TFDebertaOnlyMLMHead(config, input_embeddings=self.deberta.embeddings, name="cls") |
|
|
|
def get_lm_head(self) -> tf.keras.layers.Layer: |
|
return self.mlm.predictions |
|
|
|
@unpack_inputs |
|
@add_start_docstrings_to_model_forward(DEBERTA_INPUTS_DOCSTRING.format("batch_size, sequence_length")) |
|
@add_code_sample_docstrings( |
|
checkpoint=_CHECKPOINT_FOR_DOC, |
|
output_type=TFMaskedLMOutput, |
|
config_class=_CONFIG_FOR_DOC, |
|
) |
|
def call( |
|
self, |
|
input_ids: TFModelInputType | None = None, |
|
attention_mask: np.ndarray | tf.Tensor | None = None, |
|
token_type_ids: np.ndarray | tf.Tensor | None = None, |
|
position_ids: np.ndarray | tf.Tensor | None = None, |
|
inputs_embeds: np.ndarray | tf.Tensor | None = None, |
|
output_attentions: Optional[bool] = None, |
|
output_hidden_states: Optional[bool] = None, |
|
return_dict: Optional[bool] = None, |
|
labels: np.ndarray | tf.Tensor | None = None, |
|
training: Optional[bool] = False, |
|
) -> Union[TFMaskedLMOutput, Tuple[tf.Tensor]]: |
|
r""" |
|
labels (`tf.Tensor` or `np.ndarray` of shape `(batch_size, sequence_length)`, *optional*): |
|
Labels for computing the masked language modeling loss. Indices should be in `[-100, 0, ..., |
|
config.vocab_size]` (see `input_ids` docstring) Tokens with indices set to `-100` are ignored (masked), the |
|
loss is only computed for the tokens with labels in `[0, ..., config.vocab_size]` |
|
""" |
|
outputs = self.deberta( |
|
input_ids=input_ids, |
|
attention_mask=attention_mask, |
|
token_type_ids=token_type_ids, |
|
position_ids=position_ids, |
|
inputs_embeds=inputs_embeds, |
|
output_attentions=output_attentions, |
|
output_hidden_states=output_hidden_states, |
|
return_dict=return_dict, |
|
training=training, |
|
) |
|
sequence_output = outputs[0] |
|
prediction_scores = self.mlm(sequence_output=sequence_output, training=training) |
|
loss = None if labels is None else self.hf_compute_loss(labels=labels, logits=prediction_scores) |
|
|
|
if not return_dict: |
|
output = (prediction_scores,) + outputs[2:] |
|
return ((loss,) + output) if loss is not None else output |
|
|
|
return TFMaskedLMOutput( |
|
loss=loss, |
|
logits=prediction_scores, |
|
hidden_states=outputs.hidden_states, |
|
attentions=outputs.attentions, |
|
) |
|
|
|
|
|
@add_start_docstrings( |
|
""" |
|
DeBERTa Model transformer with a sequence classification/regression head on top (a linear layer on top of the |
|
pooled output) e.g. for GLUE tasks. |
|
""", |
|
DEBERTA_START_DOCSTRING, |
|
) |
|
class TFDebertaForSequenceClassification(TFDebertaPreTrainedModel, TFSequenceClassificationLoss): |
|
def __init__(self, config: DebertaConfig, *inputs, **kwargs): |
|
super().__init__(config, *inputs, **kwargs) |
|
|
|
self.num_labels = config.num_labels |
|
|
|
self.deberta = TFDebertaMainLayer(config, name="deberta") |
|
self.pooler = TFDebertaContextPooler(config, name="pooler") |
|
|
|
drop_out = getattr(config, "cls_dropout", None) |
|
drop_out = self.config.hidden_dropout_prob if drop_out is None else drop_out |
|
self.dropout = TFDebertaStableDropout(drop_out, name="cls_dropout") |
|
self.classifier = tf.keras.layers.Dense( |
|
units=config.num_labels, |
|
kernel_initializer=get_initializer(config.initializer_range), |
|
name="classifier", |
|
) |
|
|
|
@unpack_inputs |
|
@add_start_docstrings_to_model_forward(DEBERTA_INPUTS_DOCSTRING.format("batch_size, sequence_length")) |
|
@add_code_sample_docstrings( |
|
checkpoint=_CHECKPOINT_FOR_DOC, |
|
output_type=TFSequenceClassifierOutput, |
|
config_class=_CONFIG_FOR_DOC, |
|
) |
|
def call( |
|
self, |
|
input_ids: TFModelInputType | None = None, |
|
attention_mask: np.ndarray | tf.Tensor | None = None, |
|
token_type_ids: np.ndarray | tf.Tensor | None = None, |
|
position_ids: np.ndarray | tf.Tensor | None = None, |
|
inputs_embeds: np.ndarray | tf.Tensor | None = None, |
|
output_attentions: Optional[bool] = None, |
|
output_hidden_states: Optional[bool] = None, |
|
return_dict: Optional[bool] = None, |
|
labels: np.ndarray | tf.Tensor | None = None, |
|
training: Optional[bool] = False, |
|
) -> Union[TFSequenceClassifierOutput, Tuple[tf.Tensor]]: |
|
r""" |
|
labels (`tf.Tensor` or `np.ndarray` of shape `(batch_size,)`, *optional*): |
|
Labels for computing the sequence classification/regression loss. Indices should be in `[0, ..., |
|
config.num_labels - 1]`. If `config.num_labels == 1` a regression loss is computed (Mean-Square loss), If |
|
`config.num_labels > 1` a classification loss is computed (Cross-Entropy). |
|
""" |
|
outputs = self.deberta( |
|
input_ids=input_ids, |
|
attention_mask=attention_mask, |
|
token_type_ids=token_type_ids, |
|
position_ids=position_ids, |
|
inputs_embeds=inputs_embeds, |
|
output_attentions=output_attentions, |
|
output_hidden_states=output_hidden_states, |
|
return_dict=return_dict, |
|
training=training, |
|
) |
|
sequence_output = outputs[0] |
|
pooled_output = self.pooler(sequence_output, training=training) |
|
pooled_output = self.dropout(pooled_output, training=training) |
|
logits = self.classifier(pooled_output) |
|
loss = None if labels is None else self.hf_compute_loss(labels=labels, logits=logits) |
|
|
|
if not return_dict: |
|
output = (logits,) + outputs[1:] |
|
|
|
return ((loss,) + output) if loss is not None else output |
|
|
|
return TFSequenceClassifierOutput( |
|
loss=loss, |
|
logits=logits, |
|
hidden_states=outputs.hidden_states, |
|
attentions=outputs.attentions, |
|
) |
|
|
|
|
|
@add_start_docstrings( |
|
""" |
|
DeBERTa Model with a token classification head on top (a linear layer on top of the hidden-states output) e.g. for |
|
Named-Entity-Recognition (NER) tasks. |
|
""", |
|
DEBERTA_START_DOCSTRING, |
|
) |
|
class TFDebertaForTokenClassification(TFDebertaPreTrainedModel, TFTokenClassificationLoss): |
|
def __init__(self, config: DebertaConfig, *inputs, **kwargs): |
|
super().__init__(config, *inputs, **kwargs) |
|
|
|
self.num_labels = config.num_labels |
|
|
|
self.deberta = TFDebertaMainLayer(config, name="deberta") |
|
self.dropout = tf.keras.layers.Dropout(rate=config.hidden_dropout_prob) |
|
self.classifier = tf.keras.layers.Dense( |
|
units=config.num_labels, kernel_initializer=get_initializer(config.initializer_range), name="classifier" |
|
) |
|
|
|
@unpack_inputs |
|
@add_start_docstrings_to_model_forward(DEBERTA_INPUTS_DOCSTRING.format("batch_size, sequence_length")) |
|
@add_code_sample_docstrings( |
|
checkpoint=_CHECKPOINT_FOR_DOC, |
|
output_type=TFTokenClassifierOutput, |
|
config_class=_CONFIG_FOR_DOC, |
|
) |
|
def call( |
|
self, |
|
input_ids: TFModelInputType | None = None, |
|
attention_mask: np.ndarray | tf.Tensor | None = None, |
|
token_type_ids: np.ndarray | tf.Tensor | None = None, |
|
position_ids: np.ndarray | tf.Tensor | None = None, |
|
inputs_embeds: np.ndarray | tf.Tensor | None = None, |
|
output_attentions: Optional[bool] = None, |
|
output_hidden_states: Optional[bool] = None, |
|
return_dict: Optional[bool] = None, |
|
labels: np.ndarray | tf.Tensor | None = None, |
|
training: Optional[bool] = False, |
|
) -> Union[TFTokenClassifierOutput, Tuple[tf.Tensor]]: |
|
r""" |
|
labels (`tf.Tensor` or `np.ndarray` of shape `(batch_size, sequence_length)`, *optional*): |
|
Labels for computing the token classification loss. Indices should be in `[0, ..., config.num_labels - 1]`. |
|
""" |
|
outputs = self.deberta( |
|
input_ids=input_ids, |
|
attention_mask=attention_mask, |
|
token_type_ids=token_type_ids, |
|
position_ids=position_ids, |
|
inputs_embeds=inputs_embeds, |
|
output_attentions=output_attentions, |
|
output_hidden_states=output_hidden_states, |
|
return_dict=return_dict, |
|
training=training, |
|
) |
|
sequence_output = outputs[0] |
|
sequence_output = self.dropout(sequence_output, training=training) |
|
logits = self.classifier(inputs=sequence_output) |
|
loss = None if labels is None else self.hf_compute_loss(labels=labels, logits=logits) |
|
|
|
if not return_dict: |
|
output = (logits,) + outputs[1:] |
|
return ((loss,) + output) if loss is not None else output |
|
|
|
return TFTokenClassifierOutput( |
|
loss=loss, |
|
logits=logits, |
|
hidden_states=outputs.hidden_states, |
|
attentions=outputs.attentions, |
|
) |
|
|
|
|
|
@add_start_docstrings( |
|
""" |
|
DeBERTa Model with a span classification head on top for extractive question-answering tasks like SQuAD (a linear |
|
layers on top of the hidden-states output to compute `span start logits` and `span end logits`). |
|
""", |
|
DEBERTA_START_DOCSTRING, |
|
) |
|
class TFDebertaForQuestionAnswering(TFDebertaPreTrainedModel, TFQuestionAnsweringLoss): |
|
def __init__(self, config: DebertaConfig, *inputs, **kwargs): |
|
super().__init__(config, *inputs, **kwargs) |
|
|
|
self.num_labels = config.num_labels |
|
|
|
self.deberta = TFDebertaMainLayer(config, name="deberta") |
|
self.qa_outputs = tf.keras.layers.Dense( |
|
units=config.num_labels, kernel_initializer=get_initializer(config.initializer_range), name="qa_outputs" |
|
) |
|
|
|
@unpack_inputs |
|
@add_start_docstrings_to_model_forward(DEBERTA_INPUTS_DOCSTRING.format("batch_size, sequence_length")) |
|
@add_code_sample_docstrings( |
|
checkpoint=_CHECKPOINT_FOR_DOC, |
|
output_type=TFQuestionAnsweringModelOutput, |
|
config_class=_CONFIG_FOR_DOC, |
|
) |
|
def call( |
|
self, |
|
input_ids: TFModelInputType | None = None, |
|
attention_mask: np.ndarray | tf.Tensor | None = None, |
|
token_type_ids: np.ndarray | tf.Tensor | None = None, |
|
position_ids: np.ndarray | tf.Tensor | None = None, |
|
inputs_embeds: np.ndarray | tf.Tensor | None = None, |
|
output_attentions: Optional[bool] = None, |
|
output_hidden_states: Optional[bool] = None, |
|
return_dict: Optional[bool] = None, |
|
start_positions: np.ndarray | tf.Tensor | None = None, |
|
end_positions: np.ndarray | tf.Tensor | None = None, |
|
training: Optional[bool] = False, |
|
) -> Union[TFQuestionAnsweringModelOutput, Tuple[tf.Tensor]]: |
|
r""" |
|
start_positions (`tf.Tensor` or `np.ndarray` of shape `(batch_size,)`, *optional*): |
|
Labels for position (index) of the start of the labelled span for computing the token classification loss. |
|
Positions are clamped to the length of the sequence (`sequence_length`). Position outside of the sequence |
|
are not taken into account for computing the loss. |
|
end_positions (`tf.Tensor` or `np.ndarray` of shape `(batch_size,)`, *optional*): |
|
Labels for position (index) of the end of the labelled span for computing the token classification loss. |
|
Positions are clamped to the length of the sequence (`sequence_length`). Position outside of the sequence |
|
are not taken into account for computing the loss. |
|
""" |
|
outputs = self.deberta( |
|
input_ids=input_ids, |
|
attention_mask=attention_mask, |
|
token_type_ids=token_type_ids, |
|
position_ids=position_ids, |
|
inputs_embeds=inputs_embeds, |
|
output_attentions=output_attentions, |
|
output_hidden_states=output_hidden_states, |
|
return_dict=return_dict, |
|
training=training, |
|
) |
|
sequence_output = outputs[0] |
|
logits = self.qa_outputs(inputs=sequence_output) |
|
start_logits, end_logits = tf.split(value=logits, num_or_size_splits=2, axis=-1) |
|
start_logits = tf.squeeze(input=start_logits, axis=-1) |
|
end_logits = tf.squeeze(input=end_logits, axis=-1) |
|
loss = None |
|
|
|
if start_positions is not None and end_positions is not None: |
|
labels = {"start_position": start_positions} |
|
labels["end_position"] = end_positions |
|
loss = self.hf_compute_loss(labels=labels, logits=(start_logits, end_logits)) |
|
|
|
if not return_dict: |
|
output = (start_logits, end_logits) + outputs[2:] |
|
return ((loss,) + output) if loss is not None else output |
|
|
|
return TFQuestionAnsweringModelOutput( |
|
loss=loss, |
|
start_logits=start_logits, |
|
end_logits=end_logits, |
|
hidden_states=outputs.hidden_states, |
|
attentions=outputs.attentions, |
|
) |
|
|