Transformer架構與Transfer Learning在自然語言處理的深度應用

from transformers import BertTokenizer, BertForSequenceClassification, AdamW
from transformers import get_linear_schedule_with_warmup
from torch.utils.data import Dataset, DataLoader
import torch
import torch.nn as nn
from sklearn.model_selection import train_test_split
import numpy as np

class TextClassificationDataset(Dataset):
    """
    自訂文本分類資料集類別
    負責處理文本的編碼與批次化
    """
    def __init__(self, texts, labels, tokenizer, max_length=128):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_length = max_length
    
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = str(self.texts[idx])
        label = self.labels[idx]
        
        # 使用tokenizer進行編碼
        encoding = self.tokenizer.encode_plus(
            text,
            add_special_tokens=True,
            max_length=self.max_length,
            padding='max_length',
            truncation=True,
            return_attention_mask=True,
            return_tensors='pt'
        )
        
        return {
            'input_ids': encoding['input_ids'].flatten(),
            'attention_mask': encoding['attention_mask'].flatten(),
            'labels': torch.tensor(label, dtype=torch.long)
        }

def create_data_loader(texts, labels, tokenizer, max_length, batch_size):
    """
    建立資料載入器
    """
    dataset = TextClassificationDataset(
        texts=texts,
        labels=labels,
        tokenizer=tokenizer,
        max_length=max_length
    )
    
    return DataLoader(
        dataset,
        batch_size=batch_size,
        shuffle=True,
        num_workers=2
    )

def train_epoch(model, data_loader, optimizer, scheduler, device):
    """
    執行一個訓練週期
    """
    model.train()
    total_loss = 0
    correct_predictions = 0
    
    for batch in data_loader:
        input_ids = batch['input_ids'].to(device)
        attention_mask = batch['attention_mask'].to(device)
        labels = batch['labels'].to(device)
        
        # 清除梯度
        optimizer.zero_grad()
        
        # 前向傳播
        outputs = model(
            input_ids=input_ids,
            attention_mask=attention_mask,
            labels=labels
        )
        
        loss = outputs.loss
        logits = outputs.logits
        
        # 計算準確率
        predictions = torch.argmax(logits, dim=1)
        correct_predictions += torch.sum(predictions == labels)
        
        # 反向傳播
        loss.backward()
        
        # 梯度裁剪防止梯度爆炸
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        
        optimizer.step()
        scheduler.step()
        
        total_loss += loss.item()
    
    return correct_predictions.double() / len(data_loader.dataset), total_loss / len(data_loader)

def evaluate_model(model, data_loader, device):
    """
    評估模型效能
    """
    model.eval()
    total_loss = 0
    correct_predictions = 0
    
    with torch.no_grad():
        for batch in data_loader:
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['labels'].to(device)
            
            outputs = model(
                input_ids=input_ids,
                attention_mask=attention_mask,
                labels=labels
            )
            
            loss = outputs.loss
            logits = outputs.logits
            
            predictions = torch.argmax(logits, dim=1)
            correct_predictions += torch.sum(predictions == labels)
            total_loss += loss.item()
    
    return correct_predictions.double() / len(data_loader.dataset), total_loss / len(data_loader)

# 主要訓練流程
def train_bert_classifier(train_texts, train_labels, val_texts, val_labels, num_classes, epochs=3):
    """
    完整的BERT文本分類器訓練流程
    """
    # 設定裝置
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    print(f'使用裝置: {device}')
    
    # 載入預訓練模型與分詞器
    model_name = 'bert-base-chinese'  # 使用中文BERT模型
    tokenizer = BertTokenizer.from_pretrained(model_name)
    model = BertForSequenceClassification.from_pretrained(
        model_name,
        num_labels=num_classes
    )
    model = model.to(device)
    
    # 建立資料載入器
    train_loader = create_data_loader(
        train_texts,
        train_labels,
        tokenizer,
        max_length=128,
        batch_size=16
    )
    
    val_loader = create_data_loader(
        val_texts,
        val_labels,
        tokenizer,
        max_length=128,
        batch_size=16
    )
    
    # 設定優化器與學習率調度器
    optimizer = AdamW(model.parameters(), lr=2e-5, eps=1e-8)
    total_steps = len(train_loader) * epochs
    scheduler = get_linear_schedule_with_warmup(
        optimizer,
        num_warmup_steps=0,
        num_training_steps=total_steps
    )
    
    # 訓練迴圈
    best_accuracy = 0
    
    for epoch in range(epochs):
        print(f'\n第 {epoch + 1}/{epochs} 個訓練週期')
        print('-' * 50)
        
        train_acc, train_loss = train_epoch(
            model,
            train_loader,
            optimizer,
            scheduler,
            device
        )
        
        print(f'訓練損失: {train_loss:.4f} | 訓練準確率: {train_acc:.4f}')
        
        val_acc, val_loss = evaluate_model(model, val_loader, device)
        
        print(f'驗證損失: {val_loss:.4f} | 驗證準確率: {val_acc:.4f}')
        
        if val_acc > best_accuracy:
            torch.save(model.state_dict(), 'best_model.pth')
            best_accuracy = val_acc
            print('儲存最佳模型')
    
    return model, tokenizer

# 推論函數
def predict_text(text, model, tokenizer, device, max_length=128):
    """
    對單一文本進行預測
    """
    model.eval()
    
    encoding = tokenizer.encode_plus(
        text,
        add_special_tokens=True,
        max_length=max_length,
        padding='max_length',
        truncation=True,
        return_attention_mask=True,
        return_tensors='pt'
    )
    
    input_ids = encoding['input_ids'].to(device)
    attention_mask = encoding['attention_mask'].to(device)
    
    with torch.no_grad():
        outputs = model(input_ids=input_ids, attention_mask=attention_mask)
        logits = outputs.logits
        prediction = torch.argmax(logits, dim=1).item()
        probabilities = torch.softmax(logits, dim=1).squeeze().cpu().numpy()
    
    return prediction, probabilities

from fastai.text.all import *
import pandas as pd
from pathlib import Path

class NLPPipeline:
    """
    使用fastai的完整NLP處理流程
    """
    def __init__(self, model_path='models'):
        self.model_path = Path(model_path)
        self.model_path.mkdir(exist_ok=True)
        self.lm_learner = None
        self.classifier_learner = None
    
    def prepare_data(self, texts, labels, valid_pct=0.2):
        """
        準備資料並建立DataLoaders
        """
        # 建立DataFrame
        df = pd.DataFrame({
            'text': texts,
            'label': labels
        })
        
        # 分割訓練集與驗證集
        splits = RandomSplitter(valid_pct=valid_pct)(range(len(df)))
        
        # 建立語言模型的DataLoaders
        dls_lm = DataBlock(
            blocks=TextBlock.from_df('text', is_lm=True),
            get_x=ColReader('text'),
            splitter=IndexSplitter(splits[1])
        ).dataloaders(df, bs=64)
        
        # 建立分類器的DataLoaders
        dls_clas = DataBlock(
            blocks=(TextBlock.from_df('text'), CategoryBlock),
            get_x=ColReader('text'),
            get_y=ColReader('label'),
            splitter=IndexSplitter(splits[1])
        ).dataloaders(df, bs=64)
        
        return dls_lm, dls_clas
    
    def train_language_model(self, dls_lm, epochs=10, base_lr=1e-2):
        """
        訓練語言模型
        """
        print("開始訓練語言模型...")
        
        # 建立語言模型學習器
        self.lm_learner = language_model_learner(
            dls_lm,
            AWD_LSTM,
            drop_mult=0.5,
            metrics=[accuracy, Perplexity()]
        )
        
        # 尋找最佳學習率
        lr_min, lr_steep = self.lm_learner.lr_find()
        print(f'建議的學習率範圍: {lr_min:.2e} 到 {lr_steep:.2e}')
        
        # 使用one-cycle訓練策略
        self.lm_learner.fit_one_cycle(epochs, base_lr)
        
        # 儲存編碼器
        self.lm_learner.save_encoder('finetuned_encoder')
        
        print("語言模型訓練完成")
        
        return self.lm_learner
    
    def train_classifier(self, dls_clas, epochs=8, base_lr=1e-2):
        """
        訓練文本分類器
        """
        print("開始訓練文本分類器...")
        
        # 建立分類器學習器
        self.classifier_learner = text_classifier_learner(
            dls_clas,
            AWD_LSTM,
            drop_mult=0.5,
            metrics=[accuracy, F1Score()]
        )
        
        # 載入預訓練的編碼器
        self.classifier_learner.load_encoder('finetuned_encoder')
        
        # 分階段訓練
        # 第一階段:凍結編碼器,只訓練分類頭
        print("第一階段:訓練分類頭...")
        self.classifier_learner.freeze()
        self.classifier_learner.fit_one_cycle(2, base_lr)
        
        # 第二階段:解凍最後幾層,進行微調
        print("第二階段:微調後層...")
        self.classifier_learner.freeze_to(-2)
        self.classifier_learner.fit_one_cycle(2, slice(base_lr/100, base_lr/10))
        
        # 第三階段:解凍所有層,整體微調
        print("第三階段:整體微調...")
        self.classifier_learner.unfreeze()
        self.classifier_learner.fit_one_cycle(
            epochs - 4,
            slice(base_lr/1000, base_lr/100)
        )
        
        # 儲存模型
        self.classifier_learner.save('text_classifier')
        
        print("分類器訓練完成")
        
        return self.classifier_learner
    
    def predict(self, text):
        """
        對新文本進行預測
        """
        if self.classifier_learner is None:
            raise ValueError("請先訓練分類器")
        
        prediction, idx, probabilities = self.classifier_learner.predict(text)
        
        return {
            'prediction': prediction,
            'class_index': idx,
            'probabilities': probabilities.numpy()
        }
    
    def evaluate(self):
        """
        評估模型效能
        """
        if self.classifier_learner is None:
            raise ValueError("請先訓練分類器")
        
        # 顯示驗證集結果
        self.classifier_learner.show_results()
        
        # 顯示混淆矩陣
        interp = ClassificationInterpretation.from_learner(self.classifier_learner)
        interp.plot_confusion_matrix(figsize=(10, 10))
        
        # 顯示分類錯誤最嚴重的樣本
        interp.plot_top_losses(9, figsize=(15, 10))
        
        return interp

# 使用範例
def run_fastai_pipeline(texts, labels):
    """
    執行完整的fastai訓練流程
    """
    # 初始化流程
    pipeline = NLPPipeline()
    
    # 準備資料
    dls_lm, dls_clas = pipeline.prepare_data(texts, labels)
    
    # 訓練語言模型
    lm_learner = pipeline.train_language_model(dls_lm, epochs=10)
    
    # 訓練分類器
    classifier_learner = pipeline.train_classifier(dls_clas, epochs=8)
    
    # 評估模型
    interpretation = pipeline.evaluate()
    
    return pipeline

# 資料增強技術
class TextAugmentation:
    """
    文本資料增強工具類別
    """
    @staticmethod
    def synonym_replacement(text, n=2):
        """
        同義詞替換
        """
        # 實作同義詞替換邏輯
        # 這裡需要整合同義詞詞典
        pass
    
    @staticmethod
    def random_insertion(text, n=2):
        """
        隨機插入
        """
        # 實作隨機插入邏輯
        pass
    
    @staticmethod
    def random_swap(text, n=2):
        """
        隨機交換
        """
        # 實作隨機交換邏輯
        pass
    
    @staticmethod
    def random_deletion(text, p=0.1):
        """
        隨機刪除
        """
        # 實作隨機刪除邏輯
        pass