# coding=utf-8
# Copyright 2024 The HuggingFace Inc. team, The Hugging Face Team.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tokenization classes for FLMR."""
from typing import List, Optional, Union
from transformers.utils import TensorType, logging
from transformers.models.bert.tokenization_bert import BertTokenizer
logger = logging.get_logger(__name__)
VOCAB_FILES_NAMES = {"vocab_file": "vocab.txt", "tokenizer_file": "tokenizer_config.json"}
CONTEXT_ENCODER_PRETRAINED_VOCAB_FILES_MAP = {
"vocab_file": {
"LinWeizheDragon/PreFLMR_ViT-L": (
"https://huggingface.co/LinWeizheDragon/PreFLMR_ViT-L/resolve/main/context_tokenizer/vocab.txt"
),
"LinWeizheDragon/FLMR": (
"https://huggingface.co/LinWeizheDragon/FLMR/resolve/main/context_tokenizer/vocab.txt"
),
},
"tokenizer_file": {
"LinWeizheDragon/PreFLMR_ViT-L": (
"https://huggingface.co/LinWeizheDragon/PreFLMR_ViT-L/resolve/main/context_tokenizer/tokenizer_config.json"
),
"LinWeizheDragon/FLMR": (
"https://huggingface.co/LinWeizheDragon/FLMR/resolve/main/context_tokenizer/tokenizer_config.json"
),
},
}
QUESTION_ENCODER_PRETRAINED_VOCAB_FILES_MAP = {
"vocab_file": {
"LinWeizheDragon/PreFLMR_ViT-L": (
"https://huggingface.co/LinWeizheDragon/PreFLMR_ViT-L/resolve/main/query_tokenizer/vocab.txt"
),
"LinWeizheDragon/FLMR": ("https://huggingface.co/LinWeizheDragon/FLMR/resolve/main/query_tokenizer/vocab.txt"),
},
"tokenizer_file": {
"LinWeizheDragon/PreFLMR_ViT-L": (
"https://huggingface.co/LinWeizheDragon/PreFLMR_ViT-L/resolve/main/query_tokenizer/tokenizer_config.json"
),
"LinWeizheDragon/FLMR": (
"https://huggingface.co/LinWeizheDragon/FLMR/resolve/main/query_tokenizer/tokenizer_config.json"
),
},
}
CONTEXT_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES = {
"LinWeizheDragon/PreFLMR_ViT-L": 512,
"LinWeizheDragon/FLMR": 512,
}
QUESTION_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES = {
"LinWeizheDragon/PreFLMR_ViT-L": 512,
"LinWeizheDragon/FLMR": 512,
}
CONTEXT_ENCODER_PRETRAINED_INIT_CONFIGURATION = {
"LinWeizheDragon/PreFLMR_ViT-L": {"do_lower_case": True},
"LinWeizheDragon/FLMR": {"do_lower_case": True},
}
QUESTION_ENCODER_PRETRAINED_INIT_CONFIGURATION = {
"LinWeizheDragon/PreFLMR_ViT-L": {"do_lower_case": True},
"LinWeizheDragon/FLMR": {"do_lower_case": True},
}
# Modified from colbert.modeling.tokenization
class FLMRContextEncoderTokenizer(BertTokenizer):
r"""
Construct a FLMRContextEncoder tokenizer.
[`FLMRContextEncoderTokenizer`] is identical to [`BertTokenizer`] and runs end-to-end tokenization: punctuation
splitting and wordpiece.
Refer to superclass [`BertTokenizer`] for usage examples and documentation concerning parameters.
"""
vocab_files_names = VOCAB_FILES_NAMES
pretrained_vocab_files_map = CONTEXT_ENCODER_PRETRAINED_VOCAB_FILES_MAP
max_model_input_sizes = CONTEXT_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
pretrained_init_configuration = CONTEXT_ENCODER_PRETRAINED_INIT_CONFIGURATION
def __init__(
self,
doc_maxlen: Optional[int] = 512,
**kwargs,
):
super().__init__(
doc_maxlen=doc_maxlen,
**kwargs,
)
self.doc_maxlen = doc_maxlen
self.D_marker_token, self.D_marker_token_id = "[D]", self.convert_tokens_to_ids("[unused1]")
def __call__(
self,
text: List[str],
padding: Optional[Union[str, bool]] = "max_length",
truncation: Optional[Union[bool, str]] = "longest_first",
max_length: Optional[int] = 512,
return_tensors: Optional[Union[str, TensorType]] = "pt",
**kwargs,
):
# add placehold for the [D] marker
text = [". " + x for x in text]
if max_length > self.doc_maxlen:
# can not exceed the pre-set length
max_length = self.doc_maxlen
encoding = super().__call__(
text,
padding=padding,
truncation=truncation,
return_tensors=return_tensors,
max_length=max_length,
**kwargs,
)
ids, mask = encoding["input_ids"], encoding["attention_mask"]
# postprocess for the [D] marker
ids[:, 1] = self.D_marker_token_id
# if bsize:
# # This bsize function is used in the original ColBERT codebase to split inputs into multiple batches
# if image_features is not None:
# ids, mask, image_features, reverse_indices = _sort_by_length(ids, mask, bsize, image_features=image_features)
# batches = _split_into_batches(ids, mask, bsize, image_features=image_features)
# else:
# ids, mask, reverse_indices = _sort_by_length(ids, mask, bsize)
# batches = _split_into_batches(ids, mask, bsize)
# return batches, reverse_indices
encoding["input_ids"] = ids
encoding["attention_mask"] = mask
return encoding
# Modified from colbert.modeling.tokenization
class FLMRQueryEncoderTokenizer(BertTokenizer):
r"""
Constructs a FLMRQueryEncoder tokenizer.
[`FLMRQueryEncoder`] is identical to [`BertTokenizer`] and runs end-to-end tokenization: punctuation
splitting and wordpiece.
Refer to superclass [`BertTokenizer`] for usage examples and documentation concerning parameters.
"""
vocab_files_names = VOCAB_FILES_NAMES
pretrained_vocab_files_map = QUESTION_ENCODER_PRETRAINED_VOCAB_FILES_MAP
max_model_input_sizes = QUESTION_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
pretrained_init_configuration = QUESTION_ENCODER_PRETRAINED_INIT_CONFIGURATION
def __init__(
self,
*args,
query_maxlen: Optional[int] = 32,
attend_to_mask_tokens: Optional[bool] = False,
**kwargs,
):
super().__init__(
*args,
query_maxlen=query_maxlen,
attend_to_mask_tokens=attend_to_mask_tokens,
**kwargs,
)
self.query_maxlen = query_maxlen
self.background_maxlen = 512 - self.query_maxlen + 1 # FIXME: Make this configurable
self.attend_to_mask_tokens = attend_to_mask_tokens
self.Q_marker_token, self.Q_marker_token_id = "[Q]", self.convert_tokens_to_ids("[unused0]")
def __call__(
self,
text: Union[str, List[str]],
padding: Optional[Union[str, bool]] = "max_length",
truncation: Optional[Union[bool, str]] = True,
max_length: Optional[int] = None,
return_tensors: Optional[Union[str, TensorType]] = "pt",
**kwargs,
):
if isinstance(text, str):
# convert to list if input is a single string
text = [text]
# add placehold for the [Q] marker
text = [". " + x for x in text]
if max_length is not None:
# use user specified max_length
pass
else:
# use default max length
max_length = self.query_maxlen
encoding = super().__call__(
text,
padding=padding,
truncation=truncation,
return_tensors=return_tensors,
max_length=max_length,
**kwargs,
)
ids, mask = encoding["input_ids"], encoding["attention_mask"]
# postprocess for the [Q] marker and the [MASK] augmentation
ids[:, 1] = self.Q_marker_token_id
ids[ids == self.pad_token_id] = self.mask_token_id
if self.attend_to_mask_tokens:
# When attend_to_mask_tokens is True, we want to attend to the [MASK] tokens
mask[ids == self.mask_token_id] = 1
assert mask.sum().item() == mask.size(0) * mask.size(1), mask
return {"input_ids": ids, "attention_mask": mask}