Bert4torch 快速入門實戰(zhàn)

作者簡介

作者：Bo仔很忙??

原文：https://zhuanlan.zhihu.com/p/486329434

轉(zhuǎn)載者：楊夕

推薦系統(tǒng) 百面百搭地址：

https://github.com/km1994/RES-Interview-Notes

NLP 百面百搭地址：

https://github.com/km1994/NLP-Interview-Notes

個人筆記：

https://github.com/km1994/nlp_paper_study

背景

本人經(jīng)常會閱讀蘇神的科學空間網(wǎng)站，里面有很多對前言paper淺顯易懂的解釋，以及很多蘇神自己的創(chuàng)新實踐；并且基于bert4keras框架都有了相應(yīng)的代碼實現(xiàn)。但是由于本人主要用pytorch開發(fā)，因此參考bert4keras開發(fā)了bert4torch項目，實現(xiàn)了bert4keras的主要功能。

簡介

bert4torch是一個基于pytorch的訓(xùn)練框架，前期以效仿和實現(xiàn)bert4keras的主要功能為主，方便加載多類預(yù)訓(xùn)練模型進行finetune，提供了中文注釋方便用戶理解模型結(jié)構(gòu)。主要是期望應(yīng)對新項目時，可以直接調(diào)用不同的預(yù)訓(xùn)練模型直接finetune，或方便用戶基于bert進行魔改，快速驗證自己的idea；節(jié)省在github上clone各種項目耗時耗力，且本地文件各種copy的問題。

• pip安裝

pip install bert4torch

github鏈接https://github.com/Tongjilibo/bert4torch

主要功能

1、加載預(yù)訓(xùn)練權(quán)重(bert、roberta、albert、nezha、bart、RoFormer、ELECTRA、GPT、GPT2、T5)繼續(xù)進行finetune

目前支持的預(yù)訓(xùn)練模型一覽

2、在bert基礎(chǔ)上靈活定義自己模型：主要是可以接在bert的[btz, seq_len, hdsz]的隱含層向量后做各種魔改

3、調(diào)用方式和bert4keras基本一致，簡潔高效

    model.fit(        train_dataloader,        steps_per_epoch=1000,        epochs=epochs,        callbacks=[evaluator]    )

4、實現(xiàn)基于keras的訓(xùn)練進度條動態(tài)展示

仿照keras的模型訓(xùn)練進度條

5、配合torchinfo，實現(xiàn)打印各層參數(shù)量功能

打印參數(shù)

6、結(jié)合logger，或者tensorboard可以在后臺打印日志

支持在訓(xùn)練開始/結(jié)束，batch開始/結(jié)束，epoch的開始/結(jié)束，記錄日志，寫tensorboard等

class Callback(object):    '''Callback基類    '''    def __init__(self):        pass    def on_train_begin(self, logs=None):        pass    def on_train_end(self, logs=None):        pass    def on_epoch_begin(self, global_step, epoch, logs=None):        pass    def on_epoch_end(self, global_step, epoch, logs=None):        pass    def on_batch_begin(self, global_step, batch, logs=None):        pass    def on_batch_end(self, global_step, batch, logs=None):        pass

7、集成多個example，可以作為自己的訓(xùn)練框架，方便在同一個數(shù)據(jù)集上嘗試多種解決方案

實現(xiàn)多個example可供參考

支持的預(yù)訓(xùn)練權(quán)重(bert4torch)

實戰(zhàn)

1. 建模流程示例

# 建立分詞器tokenizer = Tokenizer(dict_path, do_lower_case=True)# 加載數(shù)據(jù)集，可以自己繼承Dataset來定義class MyDataset(ListDataset):    @staticmethod    def load_data(filenames):        """讀取文本文件，整理成需要的格式        """        D = []        return Ddef collate_fn(batch):    '''處理上述load_data得到的batch數(shù)據(jù)，整理成對應(yīng)device上的Tensor    注意：返回值分為feature和label, feature可整理成list或tuple    '''    batch_token_ids, batch_segment_ids, batch_labels = [], [], []    return [batch_token_ids, batch_segment_ids], batch_labels.flatten()# 加載數(shù)據(jù)集train_dataloader = DataLoader(MyDataset('file_path'), batch_size=batch_size, shuffle=True, collate_fn=collate_fn) # 定義bert上的模型結(jié)構(gòu)，以文本二分類為例class Model(BaseModel):    def __init__(self) -> None:        super().__init__()        self.bert = build_transformer_model(config_path, checkpoint_path, with_pool=True)        self.dropout = nn.Dropout(0.1)        self.dense = nn.Linear(768, 2)
    def forward(self, token_ids, segment_ids):        # build_transformer_model得到的模型僅接受list/tuple傳參，因此入?yún)⒅挥幸粋€時候包裝成[token_ids]        hidden_states, pooled_output = self.bert([token_ids, segment_ids])        output = self.dropout(pooled_output)        output = self.dense(output)        return outputmodel = Model().to(device)# 定義使用的loss和optimizer，這里支持自定義model.compile(    loss=nn.CrossEntropyLoss(), # 可以自定義Loss    optimizer=optim.Adam(model.parameters(), lr=2e-5),  # 可以自定義優(yōu)化器    scheduler=None, # 可以自定義scheduler    metrics=['accuracy'])# 定義評價函數(shù)def evaluate(data):    total, right = 0., 0.    for x_true, y_true in data:        y_pred = model.predict(x_true).argmax(axis=1)        total += len(y_true)        right += (y_true == y_pred).sum().item()    return right / totalclass Evaluator(Callback):    """評估與保存，這里定義僅在epoch結(jié)束后調(diào)用    """    def __init__(self):        self.best_val_acc = 0.
    def on_epoch_end(self, global_step, epoch, logs=None):        val_acc = evaluate(valid_dataloader)        if val_acc > self.best_val_acc:            self.best_val_acc = val_acc            model.save_weights('best_model.pt')        print(f'val_acc: {val_acc:.5f}, best_val_acc: {self.best_val_acc:.5f}\n')if?__name__?==?'__main__':    evaluator = Evaluator()    model.fit(train_dataloader, epochs=20, steps_per_epoch=100, grad_accumulation_steps=2, callbacks=[evaluator])

2. 主要模塊講解

1) 數(shù)據(jù)處理部分

a. 精簡詞表，并建立分詞器

token_dict, keep_tokens = load_vocab(    dict_path=dict_path,  # 詞典文件路徑    simplified=True,  # 過濾冗余部分token，如[unused1]    startswith=['[PAD]', '[UNK]', '[CLS]', '[SEP]'],  # 指定起始的token，如[UNK]從bert默認的103位置調(diào)整到1)tokenizer?=?Tokenizer(token_dict,?do_lower_case=True)??#?若無需精簡，僅使用當前行定義tokenizer即可

b. 好用的小函數(shù)

• text_segmentate(): 截斷總長度至不超過maxlen, 接受多個sequence輸入，每次截斷最長的句子，indices表示刪除的token位置

• tokenizer.encode(): 把text轉(zhuǎn)成token_ids，默認句首添加[CLS]，句尾添加[SEP]，返回token_ids和segment_ids，相當于同時調(diào)用tokenizer.tokenize()和tokenizer.tokens_to_ids()

• tokenizer.decode(): 把token_ids轉(zhuǎn)成text，默認會刪除[CLS], [SEP], [UNK]等特殊字符，相當于調(diào)用tokenizer.ids_to_tokens()并做了一些后處理

• sequence_padding: 將序列padding到同一長度, 傳入一個元素為list, ndarray, tensor的list，返回ndarry或tensor

2) 模型定義部分

• 模型創(chuàng)建

'''調(diào)用模型后，若設(shè)置with_pool,?with_nsp,?with_mlm，則返回值依次為[hidden_states,?pool_emb/nsp_emb,?mlm_scores],否則只返回hidden_states'''build_transformer_model(    config_path=config_path, # 模型的config文件地址    checkpoint_path=checkpoint_path, # 模型文件地址，默認值None表示不加載預(yù)訓(xùn)練模型    model='bert', # 加載的模型結(jié)構(gòu)，這里Model也可以基于nn.Module自定義后傳入    application='encoder',  # 模型應(yīng)用，支持encoder，lm和unilm格式    segment_vocab_size=2,  # type_token_ids數(shù)量，默認為2，如不傳入segment_ids則需設(shè)置為0    with_pool=False,  # 是否包含Pool部分    with_nsp=False,  # 是否包含NSP部分    with_mlm=False,  # 是否包含MLM部分    return_model_config=False,  # 是否返回模型配置參數(shù)    output_all_encoded_layers=False,  # 是否返回所有hidden_state層)

• 定義loss，optimizer，scheduler等

'''定義使用的loss和optimizer，這里支持自定義'''model.compile(    loss=nn.CrossEntropyLoss(), # 可以自定義Loss    optimizer=optim.Adam(model.parameters(), lr=2e-5),  # 可以自定義優(yōu)化器    scheduler=None, # 可以自定義scheduler    adversarial_train={'name': 'fgm'},  # 訓(xùn)練trick方案設(shè)置，支持fgm, pgd, gradient_penalty, vat    metrics=['accuracy']  # loss等默認打印的字段無需設(shè)置)

• 自定義模型

'''基于bert上層的各類魔改，如last2layer_average, token_first_last_average'''class Model(BaseModel):    # 需要繼承BaseModel    def __init__(self):        super().__init__()        self.bert = build_transformer_model(config_path, checkpoint_path)    def forward(self):        pass

• 自定義訓(xùn)練過程

'''自定義fit過程，適用于自帶fit()不滿足需求時'''class Model(BaseModel):    def fit(self, train_dataloader, steps_per_epoch, epochs):           train_dataloader = cycle(train_dataloader)        self.train()        for epoch in range(epochs):            for bti in range(steps_per_epoch):                train_X, train_y = next(train_dataloader)                output = self.forward(*train_X)                loss = self.criterion(output, train_y)                loss.backward()                self.optimizer.step()                self.optimizer.zero_grad()

• 模型保存和加載

'''prefix: 是否以原始的key來保存，如word_embedding原始key為bert.embeddings.word_embeddings.weight默認為None表示不啟用, 若基于BaseModel自定義模型，需指定為bert模型對應(yīng)的成員變量名，直接使用設(shè)置為''主要是為了別的訓(xùn)練框架容易加載'''model.save_weights(save_path, prefix=None)model.load_weights(load_path, strict=True, prefix=None)

• 加載transformers模型進行訓(xùn)練

from transformers import AutoModelForSequenceClassificationclass Model(BaseModel):    def __init__(self):        super().__init__()        self.bert = AutoModelForSequenceClassification.from_pretrained("file_path", num_labels=2)    def forward(self, token_ids, attention_mask, segment_ids):        output = self.bert(input_ids=token_ids, attention_mask=attention_mask, token_type_ids=segment_ids)        return output.logits

3) 模型評估部分

'''支持在多個位置執(zhí)行'''class Evaluator(Callback):    """評估與保存    """    def __init__(self):        self.best_val_acc = 0.    def on_train_begin(self, logs=None):  # 訓(xùn)練開始時候        pass    def on_train_end(self, logs=None):  # 訓(xùn)練結(jié)束時候        pass    def on_batch_begin(self, global_step, batch, logs=None):  # batch開始時候        pass    def on_batch_end(self, global_step, batch, logs=None):  # batch結(jié)束時候        # 可以設(shè)置每隔多少個step，后臺記錄log，寫tensorboard等        # 盡量不要在batch_begin和batch_end中print，防止打斷進度條功能        pass    def on_epoch_begin(self, global_step, epoch, logs=None):  # epoch開始時候        pass    def on_epoch_end(self, global_step, epoch, logs=None):  # epoch結(jié)束時候        val_acc = evaluate(valid_dataloader)        if val_acc > self.best_val_acc:            self.best_val_acc = val_acc            model.save_weights('best_model.pt')        print(f'val_acc: {val_acc:.5f}, best_val_acc: {self.best_val_acc:.5f}\n')

3. 其他特性講解

1) 單機多卡訓(xùn)練

a. 使用DataParallel

'''DP有兩種方式，第一種是forward只計算logit，第二種是forward直接計算loss建議使用第二種，可以部分緩解負載不均衡的問題'''from bert4torch.models import BaseModelDP# ===========處理數(shù)據(jù)和定義model===========model = BaseModelDP(model)  # 指定DP模式使用多gpumodel.compile(    loss=lambda x, _: x.mean(),  # 多個gpu計算的loss的均值    optimizer=optim.Adam(model.parameters(), lr=2e-5),  # 用足夠小的學習率)

b. 使用DistributedDataParallel

'''DDP使用torch.distributed.launch，從命令行啟動'''# 需要定義命令行參數(shù)parser = argparse.ArgumentParser()parser.add_argument("--local_rank", type=int, default=-1)args = parser.parse_args()torch.cuda.set_device(args.local_rank)device = torch.device('cuda', args.local_rank)torch.distributed.init_process_group(backend='nccl')# ===========處理數(shù)據(jù)和定義model===========# 指定DDP模型使用多gpu, master_rank為指定用于打印訓(xùn)練過程的local_rankmodel = BaseModelDDP(    model, master_rank=0,     device_ids=[args.local_rank],     output_device=args.local_rank,     find_unused_parameters=False??)#?定義使用的loss和optimizer，這里支持自定義model.compile(loss=lambda x, _: x,  # 直接把forward計算的loss傳出來optimizer=optim.Adam(model.parameters(), lr=2e-5),  # 用足夠小的學習率)

2) tensorboard保存訓(xùn)練過程

from tensorboardX import SummaryWriterclass Evaluator(Callback):    """每隔多少個step評估并記錄tensorboard    """    def on_batch_end(self, global_step, batch, logs=None):        if global_step % 100 == 0:            writer.add_scalar(f"train/loss", logs['loss'], global_step)            val_acc = evaluate(valid_dataloader)            writer.add_scalar(f"valid/acc", val_acc, global_step)

3) 打印訓(xùn)練參數(shù)

from torchinfo import summarysummary(model, input_data=next(iter(train_dataloader))[0])