
from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxConfigWithPast, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


A__  :  Optional[int]       = logging.get_logger(__name__)

A__  :  Union[str, Any]       = {
    '''google/bigbird-pegasus-large-arxiv''': (
        '''https://huggingface.co/google/bigbird-pegasus-large-arxiv/resolve/main/config.json'''
    ),
    '''google/bigbird-pegasus-large-pubmed''': (
        '''https://huggingface.co/google/bigbird-pegasus-large-pubmed/resolve/main/config.json'''
    ),
    '''google/bigbird-pegasus-large-bigpatent''': (
        '''https://huggingface.co/google/bigbird-pegasus-large-bigpatent/resolve/main/config.json'''
    ),
    # See all BigBirdPegasus models at https://huggingface.co/models?filter=bigbird_pegasus
}



class     __snake_case						(					UpperCamelCase_						):
							_a      				=			'bigbird_pegasus'
							_a      				=			['past_key_values']
							_a      				=			{
							    'num_attention_heads': 'encoder_attention_heads',
							    'hidden_size': 'd_model',
							    'attention_probs_dropout_prob': 'attention_dropout',
							}




							def __init__(	self					:	Tuple    ,    A_					:	Any=9_6_1_0_3    ,    A_					:	int=4_0_9_6    ,    A_					:	List[Any]=1_6    ,    A_					:	Optional[Any]=4_0_9_6    ,    A_					:	int=1_6    ,    A_					:	Union[str, Any]=1_6    ,    A_					:	List[Any]=4_0_9_6    ,    A_					:	Dict=1_6    ,    A_					:	List[str]=0.0    ,    A_					:	Dict=0.0    ,    A_					:	str=True    ,    A_					:	Optional[int]=True    ,    A_					:	Dict="gelu_new"    ,    A_					:	Optional[int]=1_0_2_4    ,    A_					:	List[str]=0.1    ,    A_					:	Optional[int]=0.0    ,    A_					:	Union[str, Any]=0.0    ,    A_					:	Dict=0.02    ,    A_					:	List[str]=2    ,    A_					:	Tuple=0.0    ,    A_					:	List[str]=True    ,    A_					:	Optional[int]=0    ,    A_					:	List[Any]=2    ,    A_					:	Any=1    ,    A_					:	int="block_sparse"    ,    A_					:	Optional[int]=6_4    ,    A_					:	str=3    ,    A_					:	str=False    ,    **A_					:	str    ,    ):
									lowerCAmelCase_      :					Tuple     =			vocab_size
									lowerCAmelCase_      :					Any     =			max_position_embeddings
									lowerCAmelCase_      :					List[Any]     =			d_model
									lowerCAmelCase_      :					str     =			encoder_ffn_dim
									lowerCAmelCase_      :					Dict     =			encoder_layers
									lowerCAmelCase_      :					Dict     =			encoder_attention_heads
									lowerCAmelCase_      :					str     =			decoder_ffn_dim
									lowerCAmelCase_      :					Optional[Any]     =			decoder_layers
									lowerCAmelCase_      :					Optional[Any]     =			decoder_attention_heads
									lowerCAmelCase_      :					str     =			dropout
									lowerCAmelCase_      :					Tuple     =			attention_dropout
									lowerCAmelCase_      :					Dict     =			activation_dropout
									lowerCAmelCase_      :					List[Any]     =			activation_function
									lowerCAmelCase_      :					Optional[int]     =			init_std
									lowerCAmelCase_      :					List[str]     =			encoder_layerdrop
									lowerCAmelCase_      :					int     =			decoder_layerdrop
									lowerCAmelCase_      :					List[Any]     =			classifier_dropout
									lowerCAmelCase_      :					Tuple     =			use_cache
									lowerCAmelCase_      :					Optional[Any]     =			encoder_layers
									lowerCAmelCase_      :					str     =			scale_embedding  # scale factor will be sqrt(d_model) if True

									# extra config
									lowerCAmelCase_      :					List[Any]     =			attention_type
									lowerCAmelCase_      :					str     =			block_size
									lowerCAmelCase_      :					str     =			num_random_blocks
									lowerCAmelCase_      :					Tuple     =			use_bias

									super().__init__(
									    pad_token_id=A_    ,    bos_token_id=A_    ,    eos_token_id=A_    ,    is_encoder_decoder=A_    ,    decoder_start_token_id=A_    ,    **A_    ,    )





class     __snake_case						(					UpperCamelCase_						):




							@property
							def        UpperCAmelCase__      (	self					:	Optional[int]):



									if self.task in ["default", "seq2seq-lm"]:
											lowerCAmelCase_      :					str     =			OrderedDict(
											    [
											        ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
											        ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
											    ])

											if self.use_past:
													lowerCAmelCase_      :					Optional[int]     =			{0: '''batch'''}
													lowerCAmelCase_      :					Tuple     =			{0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
											else:
													lowerCAmelCase_      :					Optional[int]     =			{0: '''batch''', 1: '''decoder_sequence'''}
													lowerCAmelCase_      :					Optional[int]     =			{0: '''batch''', 1: '''decoder_sequence'''}

											if self.use_past:
													self.fill_with_past_key_values_(A_    ,    direction='''inputs''')
									elif self.task == "causal-lm":
											# TODO: figure this case out.
											lowerCAmelCase_      :					Tuple     =			OrderedDict(
											    [
											        ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
											        ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
											    ])
											if self.use_past:
													lowerCAmelCase_ ,  lowerCAmelCase_      :					Union[str, Any]     =			self.num_layers
													for i in range(A_):
															lowerCAmelCase_      :					Optional[Any]     =			{0: '''batch''', 2: '''past_sequence + sequence'''}
															lowerCAmelCase_      :					Optional[Any]     =			{0: '''batch''', 2: '''past_sequence + sequence'''}
									else:
											lowerCAmelCase_      :					Any     =			OrderedDict(
											    [
											        ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
											        ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
											        ('''decoder_input_ids''', {0: '''batch''', 1: '''decoder_sequence'''}),
											        ('''decoder_attention_mask''', {0: '''batch''', 1: '''decoder_sequence'''}),
											    ])

									return common_inputs




							@property
							def        UpperCAmelCase__      (	self					:	str):



									if self.task in ["default", "seq2seq-lm"]:
											lowerCAmelCase_      :					str     =			super().outputs
									else:
											lowerCAmelCase_      :					int     =			super(A_    ,    self).outputs
											if self.use_past:
													lowerCAmelCase_ ,  lowerCAmelCase_      :					Optional[int]     =			self.num_layers
													for i in range(A_):
															lowerCAmelCase_      :					int     =			{0: '''batch''', 2: '''past_sequence + sequence'''}
															lowerCAmelCase_      :					Optional[Any]     =			{0: '''batch''', 2: '''past_sequence + sequence'''}
									return common_outputs




							def        UpperCAmelCase__      (	self					:	int    ,    A_					:	List[str]    ,    A_					:	Dict = -1    ,    A_					:	str = -1    ,    A_					:	List[str] = False    ,    A_					:	Dict = None    ,    ):
									lowerCAmelCase_      :					Tuple     =			self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
									    A_    ,    A_    ,    A_    ,    A_    ,    A_)

									# Generate decoder inputs
									lowerCAmelCase_      :					Union[str, Any]     =			seq_length if not self.use_past else 1
									lowerCAmelCase_      :					List[Any]     =			self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
									    A_    ,    A_    ,    A_    ,    A_    ,    A_)
									lowerCAmelCase_      :					Optional[int]     =			{F"""decoder_{name}""": tensor for name, tensor in decoder_inputs.items()}
									lowerCAmelCase_      :					Any     =			dict(**A_    ,    **A_)

									if self.use_past:
											if not is_torch_available():
													raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.''')
											else:
													import torch
											lowerCAmelCase_ ,  lowerCAmelCase_      :					Tuple     =			common_inputs['''input_ids'''].shape
											lowerCAmelCase_      :					Dict     =			common_inputs['''decoder_input_ids'''].shape[1]
											lowerCAmelCase_ ,  lowerCAmelCase_      :					Optional[Any]     =			self.num_attention_heads
											lowerCAmelCase_      :					Any     =			(
											    batch,
											    num_encoder_attention_heads,
											    encoder_seq_length,
											    self._config.hidden_size // num_encoder_attention_heads,
											)
											lowerCAmelCase_      :					Tuple     =			decoder_seq_length + 3
											lowerCAmelCase_      :					Union[str, Any]     =			(
											    batch,
											    num_decoder_attention_heads,
											    decoder_past_length,
											    self._config.hidden_size // num_decoder_attention_heads,
											)

											lowerCAmelCase_      :					Tuple     =			torch.cat(
											    [common_inputs['''decoder_attention_mask'''], torch.ones(A_    ,    A_)]    ,    dim=1)

											lowerCAmelCase_      :					List[str]     =			[]
											# If the number of encoder and decoder layers are present in the model configuration, both are considered
											lowerCAmelCase_ ,  lowerCAmelCase_      :					Tuple     =			self.num_layers
											lowerCAmelCase_      :					Optional[int]     =			min(A_    ,    A_)
											lowerCAmelCase_      :					int     =			max(A_    ,    A_) - min_num_layers
											lowerCAmelCase_      :					Any     =			'''encoder''' if num_encoder_layers > num_decoder_layers else '''decoder'''

											for _ in range(A_):
													common_inputs["past_key_values"].append(
													    (
													        torch.zeros(A_),
													        torch.zeros(A_),
													        torch.zeros(A_),
													        torch.zeros(A_),
													    ))
											# TODO: test this.
											lowerCAmelCase_      :					List[str]     =			encoder_shape if remaining_side_name == '''encoder''' else decoder_shape
											for _ in range(A_    ,    A_):
													common_inputs["past_key_values"].append((torch.zeros(A_), torch.zeros(A_)))
									return common_inputs




							def        UpperCAmelCase__      (	self					:	List[Any]    ,    A_					:	Union[str, Any]    ,    A_					:	Any = -1    ,    A_					:	Dict = -1    ,    A_					:	List[str] = False    ,    A_					:	Optional[int] = None    ,    ):
									lowerCAmelCase_      :					int     =			self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
									    A_    ,    A_    ,    A_    ,    A_    ,    A_)

									if self.use_past:
											if not is_torch_available():
													raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.''')
											else:
													import torch
											lowerCAmelCase_ ,  lowerCAmelCase_      :					Tuple     =			common_inputs['''input_ids'''].shape
											# Not using the same length for past_key_values
											lowerCAmelCase_      :					str     =			seqlen + 2
											lowerCAmelCase_ ,  lowerCAmelCase_      :					List[Any]     =			self.num_layers
											lowerCAmelCase_ ,  lowerCAmelCase_      :					Any     =			self.num_attention_heads
											lowerCAmelCase_      :					Optional[int]     =			(
											    batch,
											    num_encoder_attention_heads,
											    past_key_values_length,
											    self._config.hidden_size // num_encoder_attention_heads,
											)

											lowerCAmelCase_      :					Dict     =			common_inputs['''attention_mask'''].dtype
											lowerCAmelCase_      :					List[str]     =			torch.cat(
											    [common_inputs['''attention_mask'''], torch.ones(A_    ,    A_    ,    dtype=A_)]    ,    dim=1)
											lowerCAmelCase_      :					int     =			[
											    (torch.zeros(A_), torch.zeros(A_)) for _ in range(A_)
											]
									return common_inputs




							def        UpperCAmelCase__      (	self					:	Union[str, Any]    ,    A_					:	Optional[int]    ,    A_					:	List[str] = -1    ,    A_					:	int = -1    ,    A_					:	Union[str, Any] = False    ,    A_					:	int = None    ,    ):
									lowerCAmelCase_      :					Any     =			compute_effective_axis_dimension(
									    A_    ,    fixed_dimension=OnnxConfig.default_fixed_batch    ,    num_token_to_add=0)

									# If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
									lowerCAmelCase_      :					Optional[int]     =			tokenizer.num_special_tokens_to_add(A_)
									lowerCAmelCase_      :					Any     =			compute_effective_axis_dimension(
									    A_    ,    fixed_dimension=OnnxConfig.default_fixed_sequence    ,    num_token_to_add=A_)

									# Generate dummy inputs according to compute batch and sequence
									lowerCAmelCase_      :					List[Any]     =			[''' '''.join([tokenizer.unk_token]) * seq_length] * batch_size
									lowerCAmelCase_      :					List[Any]     =			dict(tokenizer(A_    ,    return_tensors=A_))
									return common_inputs




							def        UpperCAmelCase__      (	self					:	Optional[int]    ,    A_					:	Optional[int]    ,    A_					:	List[str] = -1    ,    A_					:	Any = -1    ,    A_					:	Tuple = False    ,    A_					:	List[str] = None    ,    ):



									if self.task in ["default", "seq2seq-lm"]:
											lowerCAmelCase_      :					Optional[int]     =			self._generate_dummy_inputs_for_default_and_seqaseq_lm(
											    A_    ,    batch_size=A_    ,    seq_length=A_    ,    is_pair=A_    ,    framework=A_)

									elif self.task == "causal-lm":
											lowerCAmelCase_      :					Dict     =			self._generate_dummy_inputs_for_causal_lm(
											    A_    ,    batch_size=A_    ,    seq_length=A_    ,    is_pair=A_    ,    framework=A_)
									else:
											lowerCAmelCase_      :					str     =			self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
											    A_    ,    batch_size=A_    ,    seq_length=A_    ,    is_pair=A_    ,    framework=A_)

									return common_inputs







							def        UpperCAmelCase__      (	self					:	str    ,    A_					:	str    ,    A_					:	Union[str, Any]    ,    A_					:	Any    ,    A_					:	Optional[Any]):



									if self.task in ["default", "seq2seq-lm"]:
											lowerCAmelCase_      :					Optional[Any]     =			super()._flatten_past_key_values_(A_    ,    A_    ,    A_    ,    A_)
									else:
											lowerCAmelCase_      :					List[str]     =			super(A_    ,    self)._flatten_past_key_values_(
											    A_    ,    A_    ,    A_    ,    A_)










import math
from typing import Optional

import torch
import torch.utils.checkpoint as checkpoint
from torch import nn

from ....modeling_utils import PreTrainedModel
from ....utils import add_start_docstrings, logging
from ...bert.modeling_bert import BertModel
from .configuration_retribert import RetriBertConfig


A__  :  Tuple       = logging.get_logger(__name__)

A__  :  List[str]       = [
    '''yjernite/retribert-base-uncased''',
    # See all RetriBert models at https://huggingface.co/models?filter=retribert
]



class     __snake_case						(					UpperCamelCase_						):
							_a      				=			RetriBertConfig
							_a      				=			None
							_a      				=			'''retribert'''




							def        UpperCAmelCase__      (	self					:	Optional[int]    ,    A_					:	Union[str, Any]):
									if isinstance(A_    ,    nn.Linear):
											module.weight.data.normal_(mean=0.0    ,    std=self.config.initializer_range)
											if module.bias is not None:
													module.bias.data.zero_()
									elif isinstance(A_    ,    nn.Embedding):
											module.weight.data.normal_(mean=0.0    ,    std=self.config.initializer_range)
											if module.padding_idx is not None:
													module.weight.data[module.padding_idx].zero_()
									elif isinstance(A_    ,    nn.LayerNorm):
											module.bias.data.zero_()
											module.weight.data.fill_(1.0)





A__  :  str       = R'''

    This model inherits from [`PreTrainedModel`]. Check the superclass documentation for the generic methods the
    library implements for all its model (such as downloading or saving, resizing the input embeddings, pruning heads
    etc.)

    This model is also a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass.
    Use it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage
    and behavior.

    Parameters:
        config ([`RetriBertConfig`]): 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.
'''



@add_start_docstrings(
    '''Bert Based model to embed queries or document for document retrieval.'''						,UpperCamelCase_						,)
class     __snake_case						(					UpperCamelCase_						):




							def __init__(	self					:	Optional[int]    ,    A_					:	RetriBertConfig):
									super().__init__(A_)
									lowerCAmelCase_      :					Optional[int]     =			config.projection_dim

									lowerCAmelCase_      :					int     =			BertModel(A_)
									lowerCAmelCase_      :					int     =			None if config.share_encoders else BertModel(A_)
									lowerCAmelCase_      :					str     =			nn.Dropout(config.hidden_dropout_prob)
									lowerCAmelCase_      :					Dict     =			nn.Linear(config.hidden_size    ,    config.projection_dim    ,    bias=A_)
									lowerCAmelCase_      :					Any     =			nn.Linear(config.hidden_size    ,    config.projection_dim    ,    bias=A_)

									lowerCAmelCase_      :					int     =			nn.CrossEntropyLoss(reduction='''mean''')

									# Initialize weights and apply final processing
									self.post_init()




							def        UpperCAmelCase__      (	self					:	str    ,    A_					:	Any    ,    A_					:	Optional[int]    ,    A_					:	Optional[int]    ,    A_					:	Optional[Any]=-1    ,    ):
									# reproduces BERT forward pass with checkpointing
									if checkpoint_batch_size < 0 or input_ids.shape[0] < checkpoint_batch_size:
											return sent_encoder(A_    ,    attention_mask=A_)[1]
									else:
											# prepare implicit variables
											lowerCAmelCase_      :					Any     =			input_ids.device
											lowerCAmelCase_      :					str     =			input_ids.size()
											lowerCAmelCase_      :					Dict     =			torch.zeros(A_    ,    dtype=torch.long    ,    device=A_)
											lowerCAmelCase_      :					List[str]     =			[None] * sent_encoder.config.num_hidden_layers
											lowerCAmelCase_      :					torch.Tensor     =			sent_encoder.get_extended_attention_mask(
											    A_    ,    A_)

											# define function for checkpointing
											def partial_encode(*A_					:	List[Any]):
													lowerCAmelCase_      :					Optional[Any]     =			sent_encoder.encoder(
													    inputs[0]    ,    attention_mask=inputs[1]    ,    head_mask=A_    ,    )
													lowerCAmelCase_      :					Any     =			encoder_outputs[0]
													lowerCAmelCase_      :					Optional[Any]     =			sent_encoder.pooler(A_)
													return pooled_output

											# run embedding layer on everything at once
											lowerCAmelCase_      :					str     =			sent_encoder.embeddings(
											    input_ids=A_    ,    position_ids=A_    ,    token_type_ids=A_    ,    inputs_embeds=A_)
											# run encoding and pooling on one mini-batch at a time
											lowerCAmelCase_      :					str     =			[]
											for b in range(math.ceil(input_ids.shape[0] / checkpoint_batch_size)):
													lowerCAmelCase_      :					Optional[Any]     =			embedding_output[b * checkpoint_batch_size : (b + 1) * checkpoint_batch_size]
													lowerCAmelCase_      :					int     =			extended_attention_mask[b * checkpoint_batch_size : (b + 1) * checkpoint_batch_size]
													lowerCAmelCase_      :					str     =			checkpoint.checkpoint(A_    ,    A_    ,    A_)
													pooled_output_list.append(A_)
											return torch.cat(A_    ,    dim=0)




							def        UpperCAmelCase__      (	self					:	List[str]    ,    A_					:	List[Any]    ,    A_					:	Union[str, Any]=None    ,    A_					:	Dict=-1    ,    ):
									lowerCAmelCase_      :					Optional[int]     =			self.embed_sentences_checkpointed(
									    A_    ,    A_    ,    self.bert_query    ,    A_    ,    )
									return self.project_query(A_)




							def        UpperCAmelCase__      (	self					:	Optional[Any]    ,    A_					:	str    ,    A_					:	int=None    ,    A_					:	Any=-1    ,    ):
									lowerCAmelCase_      :					Union[str, Any]     =			self.embed_sentences_checkpointed(
									    A_    ,    A_    ,    self.bert_query if self.bert_doc is None else self.bert_doc    ,    A_    ,    )
									return self.project_doc(A_)







							def        UpperCAmelCase__      (	self					:	List[Any]    ,    A_					:	torch.LongTensor    ,    A_					:	Optional[torch.FloatTensor]    ,    A_					:	torch.LongTensor    ,    A_					:	Optional[torch.FloatTensor]    ,    A_					:	int = -1    ,    ):
									lowerCAmelCase_      :					List[Any]     =			input_ids_query.device
									lowerCAmelCase_      :					List[str]     =			self.embed_questions(A_    ,    A_    ,    A_)
									lowerCAmelCase_      :					Union[str, Any]     =			self.embed_answers(A_    ,    A_    ,    A_)
									lowerCAmelCase_      :					Any     =			torch.mm(A_    ,    a_reps.t())
									lowerCAmelCase_      :					int     =			self.ce_loss(A_    ,    torch.arange(compare_scores.shape[1]).to(A_))
									lowerCAmelCase_      :					str     =			self.ce_loss(compare_scores.t()    ,    torch.arange(compare_scores.shape[0]).to(A_))
									lowerCAmelCase_      :					List[Any]     =			(loss_qa + loss_aq) / 2
									return loss





