双向Transformer：BERT（Bidirectional Encoder Representations from Transformers）

基于Transformer架构，通过双向上下文建模训练，提高完成任务的性能。

一 BERT的核心理念

1.1双向上下文建模依赖

之前讲的双向递归是用两个RNN进行，而BERT是通过Transformer的自注意力机制同时捕捉上下文信息。

1.1.1掩码语言模型（MLM）

在预训练时，随机遮盖（mask） 输入句子中15%的Token，这样模型就需要预测被遮盖的Token，同时利用被遮盖位置左右两侧的上下文信息。

（Token：文本处理的基本单元，表示经过分词或子词划分后的最小语义单位，它是模型输入的最小组成部分。）

1.1.2Transformer编码器的双向支持：

由于是基于Transformer编码器，之前发过Transformer是基于自注意力机制，允许每个元素与其他所有元素都有交互，所以可以直接进行上下文的感知。

并行计算：不同于RNN/LSTM的逐词处理，Transformer可一次性处理全部输入序列。

全局上下文感知：每个Token的表示由其自身和所有其他Token的加权组合决定，天然支持双向信息融合。

二 BERT架构

BERT基于 Transformer Encoder 堆叠而成，核心组件如下：

2.1输入表示（Input Embeddings）

Token Embeddings：将单词映射为向量。

Segment Embeddings：区分句子A和句子B（用于句间任务如问答）。

Position Embeddings：编码词的位置信息（取代RNN的时序处理）。

eg：

原句：[CLS] I love deep learning [SEP]

分词后：["[CLS]", "I", "love", "deep", "learning", "[SEP]"]         

序列长度：6（需填充至模型最大长度，如512，但此处简化示例）。

为什么要填充到最大长度，包括下面的列子也进行了填充，原因如下：

（1）批量计算的统一性

并行计算需求：GPU/TPU等硬件通过批量（Batch）处理数据加速训练，但同一批次内的样本必须具有相同的维度。若序列长度不一，无法直接堆叠成矩阵。

填充实现方法：将短序列末尾添加[PAD]标记，使同一批次内所有样本长度一致。

# 原始序列（长度不一）
["[CLS] I love NLP [SEP]", "[CLS] Hello [SEP]"]

# 填充后（统一长度为6）
[
  ["[CLS]", "I", "love", "NLP", "[SEP]", "[PAD]"],  
  ["[CLS]", "Hello", "[SEP]", "[PAD]", "[PAD]", "[PAD]"]
]

（2）模型架构的固定输入维度

（3）内存与计算资源优化

填充的副作用与解决方案

Token Embeddings：

Segment Embeddings：

同属于一个句子的Segment ID相同，比如：

输入句子：[CLS] How old are you? [SEP] I am 20. [SEP]

Segment IDs：[0, 0, 0, 0, 0, 0, 1, 1, 1, 1]

Position Embeddings：

最终输入表示： 

每个 Token 的最终输入向量是三者之和：

2.2Multi-Head Self-Attention（多头自注意力）

之前Transformer文章讲过

2.3前馈神经网络（FFN）

之前Transformer文章讲过

三 预训练任务

BERT通过两个无监督任务预训练模型：

3.1Masked Language Model（MLM）

操作：随机遮盖输入中15%的词（如将“机器学习”变为“机器[MASK]”），模型预测被遮盖的词。

改进：部分遮盖词替换为随机词，防止模型过度依赖局部信息。

3.2Next Sentence Prediction（NSP）

目标：判断两个句子是否为上下句关系。

输入格式：50%正样本（连续句子），50%负样本（随机采样）。

四 BERT vs 传统模型

五 BERT的应用场景

（1）文本分类：直接取 [CLS] 向量输入分类器。

（2）命名实体识别（NER）：对每个词的输出向量分类。

（3）问答系统：输入问题与文本，输出答案位置（如SQuAD数据集）。

（4）语义相似度：两文本拼接后通过 [CLS] 判断相似度。

六  BERT的限制

计算资源需求大：预训练需数千GPU小时。

文本生成长度受限：最大输入长度（通常512词）限制长文本处理。

实时性不足：推理速度较慢，需针对性优化。

# -*- coding: utf-8 -*-
# @FileName: bert_text_classification.py
import torch
from torch.utils.data import DataLoader
from transformers import BertTokenizer, BertForSequenceClassification, AdamW
from datasets import load_dataset
from tqdm import tqdm

# 参数设置
MODEL_NAME = "bert-base-uncased"
NUM_LABELS = 2
MAX_LENGTH = 256
BATCH_SIZE = 16
EPOCHS = 3
LEARNING_RATE = 2e-5

# 1. 加载模型和分词器
tokenizer = BertTokenizer.from_pretrained(MODEL_NAME)
model = BertForSequenceClassification.from_pretrained(MODEL_NAME, num_labels=NUM_LABELS)

# 2. 加载并预处理数据集
def preprocess_data():
    dataset = load_dataset('imdb')
    
    # 编码函数
    def tokenize_func(batch):
        return tokenizer(
            batch["text"], 
            padding="max_length", 
            truncation=True, 
            max_length=MAX_LENGTH,
            return_tensors="pt"
        )

    # 应用分词
    dataset = dataset.map(tokenize_func, batched=True)
    dataset = dataset.rename_column("label", "labels")
    
    # 设置Tensor格式
    for split in ['train', 'test']:
        dataset[split].set_format(
            type='torch', 
            columns=['input_ids', 'attention_mask', 'labels']
        )
    
    return dataset

dataset = preprocess_data()
train_loader = DataLoader(dataset['train'], batch_size=BATCH_SIZE, shuffle=True)
test_loader = DataLoader(dataset['test'], batch_size=BATCH_SIZE)

# 3. 训练配置
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)
optimizer = AdamW(model.parameters(), lr=LEARNING_RATE)

# 4. 训练循环
def train_model():
    model.train()
    for epoch in range(EPOCHS):
        total_loss = 0
        progress_bar = tqdm(train_loader, desc=f'Epoch {epoch+1}/{EPOCHS}')
        
        for batch in progress_bar:
            optimizer.zero_grad()
            
            inputs = {
                "input_ids": batch["input_ids"].to(device),
                "attention_mask": batch["attention_mask"].to(device),
                "labels": batch["labels"].to(device)
            }
            
            outputs = model(**inputs)
            loss = outputs.loss
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
            progress_bar.set_postfix({'loss': loss.item()})
        
        print(f"Epoch {epoch+1} Average Loss: {total_loss/len(train_loader):.4f}")

# 运行训练
train_model()

# 5. 保存模型
model.save_pretrained("./bert_imdb_sentiment")

# 6. 评估函数
def evaluate_model():
    model.eval()
    total_correct = 0
    
    with torch.no_grad():
        for batch in tqdm(test_loader, desc="Evaluating"):
            inputs = {
                "input_ids": batch["input_ids"].to(device),
                "attention_mask": batch["attention_mask"].to(device)
            }
            labels = batch["labels"].to(device)
            
            outputs = model(**inputs)
            predictions = torch.argmax(outputs.logits, dim=1)
            total_correct += (predictions == labels).sum().item()
    
    accuracy = total_correct / len(dataset['test'])
    print(f"\nTest Accuracy: {accuracy*100:.2f}%\n")

evaluate_model()

# 7. 预测函数
def predict_sentiment(text):
    model.eval()
    encoding = tokenizer(
        text, 
        max_length=MAX_LENGTH, 
        padding='max_length', 
        truncation=True, 
        return_tensors='pt'
    ).to(device)
    
    with torch.no_grad():
        outputs = model(**encoding)
        prob = torch.nn.functional.softmax(outputs.logits, dim=1)
    
    label = torch.argmax(prob).item()
    return "Positive" if label == 1 else "Negative", prob[0][label].item()

# 测试样例
sample_texts = [
    "This movie is fantastic! The acting is brilliant.",
    "A terrible waste of time. The plot makes no sense."
]

for text in sample_texts:
    label, confidence = predict_sentiment(text)
    print(f"Text: {text[:60]}...")
    print(f"=> Predicted: {label} (Confidence: {confidence:.4f})\n")