Upload 8 files

bert_tokenizer.py

@@ -0,0 +1,85 @@
+import json
+import os
+from pathlib import Path
+from typing import List
+
+import tokenizers
+import torch
+from pypinyin import pinyin, Style
+
+try:
+    from tokenizers import BertWordPieceTokenizer
+except:
+    from tokenizers.implementations import BertWordPieceTokenizer
+
+from transformers import BertTokenizerFast
+
+
+class ChineseBertTokenizer(BertTokenizerFast):
+
+    def __init__(self, **kwargs):
+        super(ChineseBertTokenizer, self).__init__(**kwargs)
+
+        bert_path = Path(os.path.abspath(__file__)).parent
+        print("bert_path", bert_path)
+        vocab_file = os.path.join(bert_path, 'vocab.txt')
+        config_path = os.path.join(bert_path, 'config')
+        self.max_length = 512
+        self.tokenizer = BertWordPieceTokenizer(vocab_file)
+
+        # load pinyin map dict
+        with open(os.path.join(config_path, 'pinyin_map.json'), encoding='utf8') as fin:
+            self.pinyin_dict = json.load(fin)
+        # load char id map tensor
+        with open(os.path.join(config_path, 'id2pinyin.json'), encoding='utf8') as fin:
+            self.id2pinyin = json.load(fin)
+        # load pinyin map tensor
+        with open(os.path.join(config_path, 'pinyin2tensor.json'), encoding='utf8') as fin:
+            self.pinyin2tensor = json.load(fin)
+
+    def tokenize_sentence(self, sentence):
+        # convert sentence to ids
+        tokenizer_output = self.tokenizer.encode(sentence)
+        bert_tokens = tokenizer_output.ids
+        pinyin_tokens = self.convert_sentence_to_pinyin_ids(sentence, tokenizer_output)
+        # assert，token nums should be same as pinyin token nums
+        assert len(bert_tokens) <= self.max_length
+        assert len(bert_tokens) == len(pinyin_tokens)
+        # convert list to tensor
+        input_ids = torch.LongTensor(bert_tokens)
+        pinyin_ids = torch.LongTensor(pinyin_tokens).view(-1)
+        return input_ids, pinyin_ids
+
+    def convert_sentence_to_pinyin_ids(self, sentence: str, tokenizer_output: tokenizers.Encoding) -> List[List[int]]:
+        # get pinyin of a sentence
+        pinyin_list = pinyin(sentence, style=Style.TONE3, heteronym=True, errors=lambda x: [['not chinese'] for _ in x])
+        pinyin_locs = {}
+        # get pinyin of each location
+        for index, item in enumerate(pinyin_list):
+            pinyin_string = item[0]
+            # not a Chinese character, pass
+            if pinyin_string == "not chinese":
+                continue
+            if pinyin_string in self.pinyin2tensor:
+                pinyin_locs[index] = self.pinyin2tensor[pinyin_string]
+            else:
+                ids = [0] * 8
+                for i, p in enumerate(pinyin_string):
+                    if p not in self.pinyin_dict["char2idx"]:
+                        ids = [0] * 8
+                        break
+                    ids[i] = self.pinyin_dict["char2idx"][p]
+                pinyin_locs[index] = ids
+
+        # find chinese character location, and generate pinyin ids
+        pinyin_ids = []
+        for idx, (token, offset) in enumerate(zip(tokenizer_output.tokens, tokenizer_output.offsets)):
+            if offset[1] - offset[0] != 1:
+                pinyin_ids.append([0] * 8)
+                continue
+            if offset[0] in pinyin_locs:
+                pinyin_ids.append(pinyin_locs[offset[0]])
+            else:
+                pinyin_ids.append([0] * 8)
+
+        return pinyin_ids

config.json

@@ -1,5 +1,5 @@
 {
-  "_name_or_path": "./chinesebert-base",
+  "_name_or_path": "../ChineseBERT-base",
   "architectures": [
     "GlyceBertForMaskedLM"
   ],

modeling_glycebert.py

@@ -8,21 +8,26 @@
 @version: 1.0
 @desc  : ChineseBert Model
 """
+import json
+import os
 import warnings
+from typing import List
 
+import numpy as np
 import torch
 from torch import nn
 from torch.nn import CrossEntropyLoss, MSELoss
+from torch.nn import functional as F
+
 try:
     from transformers.modeling_bert import BertEncoder, BertPooler, BertOnlyMLMHead, BertPreTrainedModel, BertModel
 except:
-    from transformers.models.bert.modeling_bert import BertEncoder, BertPooler, BertOnlyMLMHead, BertPreTrainedModel, BertModel
+    from transformers.models.bert.modeling_bert import BertEncoder, BertPooler, BertOnlyMLMHead, BertPreTrainedModel, \
+        BertModel
 
 from transformers.modeling_outputs import BaseModelOutputWithPooling, MaskedLMOutput, SequenceClassifierOutput, \
     QuestionAnsweringModelOutput, TokenClassifierOutput
 
-from models.fusion_embedding import FusionBertEmbeddings
-from models.classifier import BertMLP
 
 class GlyceBertModel(BertModel):
     r"""
@@ -65,19 +70,19 @@ class GlyceBertModel(BertModel):
         self.init_weights()
 
     def forward(
-        self,
-        input_ids=None,
-        pinyin_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,
-        output_attentions=None,
-        output_hidden_states=None,
-        return_dict=None,
+            self,
+            input_ids=None,
+            pinyin_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,
+            output_attentions=None,
+            output_hidden_states=None,
+            return_dict=None,
     ):
         r"""
         encoder_hidden_states  (:obj:`torch.FloatTensor` of shape :obj:`(batch_size, sequence_length, hidden_size)`, `optional`):
@@ -176,21 +181,21 @@ class GlyceBertForMaskedLM(BertPreTrainedModel):
         return self.cls.predictions.decoder
 
     def forward(
-        self,
-        input_ids=None,
-        pinyin_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,
-        output_attentions=None,
-        output_hidden_states=None,
-        return_dict=None,
-        **kwargs
+            self,
+            input_ids=None,
+            pinyin_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,
+            output_attentions=None,
+            output_hidden_states=None,
+            return_dict=None,
+            **kwargs
     ):
         r"""
         labels (:obj:`torch.LongTensor` of shape :obj:`(batch_size, sequence_length)`, `optional`):
@@ -259,18 +264,18 @@ class GlyceBertForSequenceClassification(BertPreTrainedModel):
         self.init_weights()
 
     def forward(
-        self,
-        input_ids=None,
-        pinyin_ids=None,
-        attention_mask=None,
-        token_type_ids=None,
-        position_ids=None,
-        head_mask=None,
-        inputs_embeds=None,
-        labels=None,
-        output_attentions=None,
-        output_hidden_states=None,
-        return_dict=None,
+            self,
+            input_ids=None,
+            pinyin_ids=None,
+            attention_mask=None,
+            token_type_ids=None,
+            position_ids=None,
+            head_mask=None,
+            inputs_embeds=None,
+            labels=None,
+            output_attentions=None,
+            output_hidden_states=None,
+            return_dict=None,
     ):
         r"""
         labels (:obj:`torch.LongTensor` of shape :obj:`(batch_size,)`, `optional`):
@@ -379,19 +384,19 @@ class GlyceBertForQuestionAnswering(BertPreTrainedModel):
         self.init_weights()
 
     def forward(
-        self,
-        input_ids=None,
-        pinyin_ids=None,
-        attention_mask=None,
-        token_type_ids=None,
-        position_ids=None,
-        head_mask=None,
-        inputs_embeds=None,
-        start_positions=None,
-        end_positions=None,
-        output_attentions=None,
-        output_hidden_states=None,
-        return_dict=None,
+            self,
+            input_ids=None,
+            pinyin_ids=None,
+            attention_mask=None,
+            token_type_ids=None,
+            position_ids=None,
+            head_mask=None,
+            inputs_embeds=None,
+            start_positions=None,
+            end_positions=None,
+            output_attentions=None,
+            output_hidden_states=None,
+            return_dict=None,
     ):
         r"""
         start_positions (:obj:`torch.LongTensor` of shape :obj:`(batch_size,)`, `optional`):
@@ -454,6 +459,7 @@ class GlyceBertForQuestionAnswering(BertPreTrainedModel):
             attentions=outputs.attentions,
         )
 
+
 class GlyceBertForTokenClassification(BertPreTrainedModel):
     def __init__(self, config, mlp=False):
         super().__init__(config)
@@ -530,3 +536,146 @@ class GlyceBertForTokenClassification(BertPreTrainedModel):
             hidden_states=outputs.hidden_states,
             attentions=outputs.attentions,
         )
+
+
+class FusionBertEmbeddings(nn.Module):
+    """
+    Construct the embeddings from word, position, glyph, pinyin and token_type embeddings.
+    """
+
+    def __init__(self, config):
+        super(FusionBertEmbeddings, self).__init__()
+        config_path = os.path.join(config.name_or_path, 'config')
+        font_files = []
+        for file in os.listdir(config_path):
+            if file.endswith(".npy"):
+                font_files.append(os.path.join(config_path, file))
+        self.word_embeddings = nn.Embedding(config.vocab_size, config.hidden_size, padding_idx=0)
+        self.position_embeddings = nn.Embedding(config.max_position_embeddings, config.hidden_size)
+        self.token_type_embeddings = nn.Embedding(config.type_vocab_size, config.hidden_size)
+        self.pinyin_embeddings = PinyinEmbedding(embedding_size=128, pinyin_out_dim=config.hidden_size,
+                                                 config_path=config_path)
+        self.glyph_embeddings = GlyphEmbedding(font_npy_files=font_files)
+
+        # self.LayerNorm is not snake-cased to stick with TensorFlow models variable name and be able to load
+        # any TensorFlow checkpoint file
+        self.glyph_map = nn.Linear(1728, config.hidden_size)
+        self.map_fc = nn.Linear(config.hidden_size * 3, config.hidden_size)
+        self.LayerNorm = nn.LayerNorm(config.hidden_size, eps=config.layer_norm_eps)
+        self.dropout = nn.Dropout(config.hidden_dropout_prob)
+
+        # position_ids (1, len position emb) is contiguous in memory and exported when serialized
+        self.register_buffer("position_ids", torch.arange(config.max_position_embeddings).expand((1, -1)))
+
+    def forward(self, input_ids=None, pinyin_ids=None, token_type_ids=None, position_ids=None, inputs_embeds=None):
+        if input_ids is not None:
+            input_shape = input_ids.size()
+        else:
+            input_shape = inputs_embeds.size()[:-1]
+
+        seq_length = input_shape[1]
+
+        if position_ids is None:
+            position_ids = self.position_ids[:, :seq_length]
+
+        if token_type_ids is None:
+            token_type_ids = torch.zeros(input_shape, dtype=torch.long, device=self.position_ids.device)
+
+        if inputs_embeds is None:
+            inputs_embeds = self.word_embeddings(input_ids)
+
+        # get char embedding, pinyin embedding and glyph embedding
+        word_embeddings = inputs_embeds  # [bs,l,hidden_size]
+        pinyin_embeddings = self.pinyin_embeddings(pinyin_ids)  # [bs,l,hidden_size]
+        glyph_embeddings = self.glyph_map(self.glyph_embeddings(input_ids))  # [bs,l,hidden_size]
+        # fusion layer
+        concat_embeddings = torch.cat((word_embeddings, pinyin_embeddings, glyph_embeddings), 2)
+        inputs_embeds = self.map_fc(concat_embeddings)
+
+        position_embeddings = self.position_embeddings(position_ids)
+        token_type_embeddings = self.token_type_embeddings(token_type_ids)
+
+        embeddings = inputs_embeds + position_embeddings + token_type_embeddings
+        embeddings = self.LayerNorm(embeddings)
+        embeddings = self.dropout(embeddings)
+        return embeddings
+
+
+class PinyinEmbedding(nn.Module):
+    def __init__(self, embedding_size: int, pinyin_out_dim: int, config_path):
+        """
+            Pinyin Embedding Module
+        Args:
+            embedding_size: the size of each embedding vector
+            pinyin_out_dim: kernel number of conv
+        """
+        super(PinyinEmbedding, self).__init__()
+        with open(os.path.join(config_path, 'pinyin_map.json')) as fin:
+            pinyin_dict = json.load(fin)
+        self.pinyin_out_dim = pinyin_out_dim
+        self.embedding = nn.Embedding(len(pinyin_dict['idx2char']), embedding_size)
+        self.conv = nn.Conv1d(in_channels=embedding_size, out_channels=self.pinyin_out_dim, kernel_size=2,
+                              stride=1, padding=0)
+
+    def forward(self, pinyin_ids):
+        """
+        Args:
+            pinyin_ids: (bs*sentence_length*pinyin_locs)
+
+        Returns:
+            pinyin_embed: (bs,sentence_length,pinyin_out_dim)
+        """
+        # input pinyin ids for 1-D conv
+        embed = self.embedding(pinyin_ids)  # [bs,sentence_length,pinyin_locs,embed_size]
+        bs, sentence_length, pinyin_locs, embed_size = embed.shape
+        view_embed = embed.view(-1, pinyin_locs, embed_size)  # [(bs*sentence_length),pinyin_locs,embed_size]
+        input_embed = view_embed.permute(0, 2, 1)  # [(bs*sentence_length), embed_size, pinyin_locs]
+        # conv + max_pooling
+        pinyin_conv = self.conv(input_embed)  # [(bs*sentence_length),pinyin_out_dim,H]
+        pinyin_embed = F.max_pool1d(pinyin_conv, pinyin_conv.shape[-1])  # [(bs*sentence_length),pinyin_out_dim,1]
+        return pinyin_embed.view(bs, sentence_length, self.pinyin_out_dim)  # [bs,sentence_length,pinyin_out_dim]
+
+
+class BertMLP(nn.Module):
+    def __init__(self, config, ):
+        super().__init__()
+        self.dense_layer = nn.Linear(config.hidden_size, config.hidden_size)
+        self.dense_to_labels_layer = nn.Linear(config.hidden_size, config.num_labels)
+        self.activation = nn.Tanh()
+
+    def forward(self, sequence_hidden_states):
+        sequence_output = self.dense_layer(sequence_hidden_states)
+        sequence_output = self.activation(sequence_output)
+        sequence_output = self.dense_to_labels_layer(sequence_output)
+        return sequence_output
+
+
+class GlyphEmbedding(nn.Module):
+    """Glyph2Image Embedding"""
+
+    def __init__(self, font_npy_files: List[str]):
+        super(GlyphEmbedding, self).__init__()
+        font_arrays = [
+            np.load(np_file).astype(np.float32) for np_file in font_npy_files
+        ]
+        self.vocab_size = font_arrays[0].shape[0]
+        self.font_num = len(font_arrays)
+        self.font_size = font_arrays[0].shape[-1]
+        # N, C, H, W
+        font_array = np.stack(font_arrays, axis=1)
+        self.embedding = nn.Embedding(
+            num_embeddings=self.vocab_size,
+            embedding_dim=self.font_size ** 2 * self.font_num,
+            _weight=torch.from_numpy(font_array.reshape([self.vocab_size, -1]))
+        )
+
+    def forward(self, input_ids):
+        """
+            get glyph images for batch inputs
+        Args:
+            input_ids: [batch, sentence_length]
+        Returns:
+            images: [batch, sentence_length, self.font_num*self.font_size*self.font_size]
+        """
+        # return self.embedding(input_ids).view([-1, self.font_num, self.font_size, self.font_size])
+        return self.embedding(input_ids)

tokenizer_config.json

@@ -1,7 +1,7 @@
 {
   "auto_map": {
     "AutoTokenizer": [
-      "tokenizer.ChineseBertTokenizer",
+      "bert_tokenizer.ChineseBertTokenizer",
       null
     ]
   },