迁移学习与FastText精简版

迁移学习是一种方法论,它的实现方式通常是:拿一个预训练模型,对它进行微调,从而完成自己的任务。

fasttext

基本概念

fasttext 优势:

  • 网络结构非常简单
  • 训练词向量时使用层次 softmax,提升超多类别下的模型性能
  • 因模型简单无法捕捉词序特征,通过 n-gram 特征提取弥补缺陷

负采样优势:

  • 提高训练速度:只选择部分数据计算损失,损失计算更简单
  • 改进效果:增加部分负样本模拟真实噪声,让模型更稳健

补充:训练集用于模型开发,测试集用于评估效果,验证集用于调参(相当于项目经理把关)。

安装:

pip install fasttext -i https://mirrors.aliyun.com/pypi/simple/
# 或
pip install fasttext-wheel -i https://mirrors.aliyun.com/pypi/simple/

若运行报错,通常是 numpy 版本问题:先 pip uninstall -y numpy,再 pip install numpy==1.26.4

fasttext 文本分类

核心 API:train_supervised(input, epoch, lr, wordNgrams, loss) 训练;model.test(path) 返回 (样本数, 精确率, 召回率)model.predict(text, k, threshold) 预测。

关键调参路径:

步骤 参数 作用
数据预处理 统一大小写、标点前加空格 提升基础效果
增加训练轮次 epoch=20 显著提升精确率
调整学习率 lr=1 进一步提升效果
n-gram 特征 wordNgrams=2 弥补词序缺失
层次 softmax loss="hs" 多类别加速
自动超参调优 autotuneValidationFileautotuneDuration 自动搜索最优组合
多标签分类 loss="ova"(one vs all) 拆解为多个二分类,注意 lr 不可过大
import fasttext

# 1- 基础训练 + 预测 + 测试
model = fasttext.train_supervised(input="data/cooking_train.txt")
print(model.predict("Which baking dish is best to bake a banana bread ?"))
print(model.test(path="data/cooking_valid.txt"))  # (样本数, 精确率, 召回率)

# 2- 综合调参训练
model = fasttext.train_supervised(
    input="data/cooking.pre.train",
    epoch=20, lr=1, wordNgrams=2, loss="hs"
)

# 3- 自动超参调优
model = fasttext.train_supervised(
    input="data/cooking.pre.train",
    autotuneValidationFile="data/cooking.pre.valid",
    autotuneDuration=60 * 2
)

# 4- 多标签分类(ova)
model = fasttext.train_supervised(
    input="data/cooking.pre.train",
    epoch=20, lr=0.1, wordNgrams=2, loss="ova"
)
# k:最多展示目标值数量,-1 表示全部;threshold:概率阈值
print(model.predict("Which baking dish is best to bake a banana bread ?", k=3, threshold=0.5))

# 5- 保存与加载
model.save_model("model/cooking_model.pkl")
model2 = fasttext.load_model("model/cooking_model.pkl")

训练词向量

参考 NLP 课程第一天,使用 fasttext 实现 word2vec 代码。

迁移学习的概念

魔搭社区:https://www.modelscope.cn/models?page=1&tabKey=task

  • 预训练模型:别人训练好的模型,一般具备复杂的网络结构,在大量语料下训练完成。
  • 微调:对预训练模型进行垂直领域数据微调,参数可全部微调、部分微调或不微调;一般在预训练模型后加入自定义网络,自定义网络的参数必须训练。

迁移学习的两种方式:

  • 开箱即用:预训练模型任务与目标任务相似时,直接使用。
  • 微调:加入自定义网络进行垂直领域训练。

预训练语言模型类别(以 transformer 架构划分):

  • Encoder-Only:仅编码器,代表 BERT
  • Decoder-Only:仅解码器,代表 GPT
  • Encoder-Decoder:完整 transformer,代表 T5

NLP 常用预训练模型

当下流行:BERT、GPT、GPT-2、Transformer-XL、XLNet、XLM、RoBERTa、DistilBERT、ALBERT、T5、XLM-RoBERTa。

BERT 及其变体关键参数:

模型 隐层数 输出维度 注意力头 参数量 训练语料
bert-base-uncased 12 768 12 110M 小写英文
bert-large-uncased 24 1024 16 340M 小写英文
bert-base-chinese 12 768 12 110M 简繁中文

面试高频:bert-base-chinese 有 12 个自注意力头

Transformers 库使用

三层应用结构

  • Pipeline(管道):高度集成的极简方式,几行代码完成一个 NLP 任务。优点是开发简单,缺点是可调超参数少,适合快速验证模型。
  • AutoModel(自动模型):可载入 BERTology 系列模型,代码相对简单,可调超参数适中。
  • SpecificModel(具体模型):明确指定模型并按特定参数调用,灵活度高,可调参数多,适合业务场景要求高的情况。后续可结合 LoRA、QLoRA。

环境搭建:

conda activate nlp_cuda
pip install transformers
pip install datasets
pip install tf-keras

Pipeline 方式应用预训练模型

通过 pipeline(task, model) 加载模型后直接调用文本即可。

任务 task 参数 说明
文本分类 text-classification 判断文本类别,如好评/差评
特征抽取 feature-extraction 返回文本的词向量表示
完形填空 fill-mask 预测 [MASK] 位置的词
阅读理解 question-answering 根据 context 回答 question
文本摘要 summarization 生成文档概括总结
命名实体识别 ner 序列标注,常用 BIO/BIOES 标识
import os
os.environ["TF_ENABLE_ONEDNN_OPTS"] = "0"
from transformers import pipeline

# 文本分类
model = pipeline(task="text-classification", model=r"D:\soft\PretrainedModel\chinese_sentiment")
print(model("这家餐馆的卫生太差了,吃了拉稀,非常不推荐"))

# 完形填空:必须写 [MASK]
model = pipeline(task="fill-mask", model=r"D:\soft\PretrainedModel\chinese-bert-wwm")
print(model("我想明天去[MASK]家吃饭。"))

# 命名实体识别:B 表示实体开始,I 表示实体中间
model = pipeline(task="ner", model=r"D:\soft\PretrainedModel\roberta-base-finetuned-cluener2020-chinese")
print(model('鲁迅原名周树人,代表作有朝花夕拾,在商务部上班,今天他去故宫游览'))

AutoModel 方式应用预训练模型

通过 AutoTokenizer 处理数据、AutoModelForXxx 加载模型,比 Pipeline 灵活。

核心组件:

  • AutoConfig:加载配置
  • AutoTokenizer:加载分词器
  • AutoModel:加载基础模型
  • AutoModelForSequenceClassification:文本分类
  • AutoModelForMaskedLM:完形填空
  • AutoModelForQuestionAnswering:阅读理解
  • AutoModelForSeq2SeqLM:文本摘要
  • AutoModelForTokenClassification:NER

分词器关键参数:

  • return_tensors:返回张量类型,pt(PyTorch)或 np(NumPy)
  • padding:是否填充到 max_length
  • truncation:超过 max_length 是否截断
  • max_length:句子最大长度
  • encode_plusencode 返回更丰富的信息(含 input_idstoken_type_idsattention_mask

文本分类示例:

from transformers import AutoTokenizer, AutoModelForSequenceClassification

model_path = r"D:\PretrainedModel\chinese_sentiment"
tokenizer = AutoTokenizer.from_pretrained(model_path)
model = AutoModelForSequenceClassification.from_pretrained(model_path)

data_tensor = tokenizer.encode(
    text="我爱黑马",
    return_tensors="pt",
    padding="max_length",
    truncation=True,
    max_length=10
)

model.eval()
result = model(data_tensor)
print(result[0].argmax(dim=-1))  # 预测分类 ID

特征抽取示例:

返回 last_hidden_state(最后一层隐藏状态)和 pooler_output(池化层输出,实际是线性层)。

from transformers import AutoTokenizer, AutoModel

model_path = r"D:\soft\PretrainedModel\bert-base-chinese"
tokenizer = AutoTokenizer.from_pretrained(model_path)
model = AutoModel.from_pretrained(model_path)

data_tensor = tokenizer.encode_plus(
    text=['你是谁', '人生该如何起头'],
    return_tensors="pt",
    padding="max_length",
    truncation=True,
    max_length=30
)

model.eval()
result = model(**data_tensor)  # 字典解包
print("最后一个隐藏状态信息", result.last_hidden_state.shape)
print("池化层信息", result.pooler_output.shape)

扩展:BERT 中的"池化层"与 CNN 中的池化不是同一个东西,BERT 的池化层实际是一个线性层。“Pooling"在机器学习中泛指将多个值聚合为一个值的操作。

完形填空要点:

  • 不要设置 paddingtruncation 参数
  • 结果 logits 形状为 [1, 词数, 词汇表大小]
  • 通过 tokenizer.convert_ids_to_tokens() 将索引转为词

阅读理解要点:

  • 每次让模型回答一个问题(与 Pipeline 主要区别)
  • 通过 start_logitsend_logits 的 argmax 得到答案在 context 中的切片

文本摘要要点:

  • 使用 model.generate(**data_tensor) 生成文本
  • 通过 tokenizer.decode()convert_ids_to_tokens() 解码结果

NER 要点:

  • logits 形状 [1, 词数, 实体类别数]
  • 通过 config.id2label 将索引映射为实体名称
  • 过滤 tokenizer.all_special_tokens 中的特殊符号

迁移学习实践(文本分类案例)

实现流程

  1. 获取数据集
  2. 数据预处理:实例化 dataset、dataloader,在 dataloader 中用自定义函数进行文本张量化
  3. 搭建模型:用 BERT 预训练模型得到文本特征表示,再经过自定义网络实现分类
  4. 模型训练:不训练 BERT 参数,只更新自定义网络参数
  5. 模型测试:GPU 训练的模型在 CPU 上使用时,model.load_state_dict(torch.load(path, map_location="cpu"))

数据预处理

批处理函数 collate_fn 在 Dataloader 中自动调用,对每个批次的数据进行张量化处理。

import torch
import torch.nn as nn
from datasets import load_dataset
from torch.utils.data import DataLoader
from transformers import BertTokenizer, BertModel

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

model_path = r"D:\PretrainedModel\bert-base-chinese"
# 分词器不涉及张量计算,不需要发送到 GPU
bert_tokenizer = BertTokenizer.from_pretrained(model_path)
bert_model = BertModel.from_pretrained(model_path).to(device)

def collate_fn(data):
    sents = [item["text"] for item in data]
    labels = [item["label"] for item in data]

    data_tensor = bert_tokenizer.batch_encode_plus(
        sents,
        return_tensors="pt",
        padding=True,
        truncation=True,
        max_length=300
    )
    labels = torch.LongTensor(labels)
    return data_tensor["input_ids"], data_tensor["token_type_ids"], data_tensor["attention_mask"], labels

def get_dataloader(file_path):
    dataset = load_dataset(path="data", data_files=file_path, split="train")
    # batch_size 只能为 1 时,说明样本长度不一致
    return DataLoader(dataset, batch_size=8, shuffle=True, drop_last=True, collate_fn=collate_fn)

搭建模型

迁移学习思路:BERT 预训练模型特征处理 + 自定义分类网络。

关键点:

  • 自定义线性层输入维度必须是 768(BERT 词向量维度),输出维度为分类数(二分类为 2)
  • 使用 torch.no_grad() 冻结 BERT 参数
  • 分类任务固定取 bert_output.last_hidden_state[:, 0],即 [CLS] 的隐藏状态
  • BERT 编码器是自注意力,[CLS] 能聚合整句信息作为分类表示
class AiModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.linear = nn.Linear(in_features=768, out_features=2)

    def forward(self, input_ids, token_type_ids, attention_mask):
        with torch.no_grad():
            bert_output = bert_model(
                input_ids=input_ids,
                token_type_ids=token_type_ids,
                attention_mask=attention_mask
            )
        # 取 [CLS] 的隐藏状态作为句子表示
        return self.linear(bert_output.last_hidden_state[:, 0])

模型训练

要点:

  1. model.train() 切换训练模式
  2. param.requires_grad_(False) 禁用 BERT 梯度
  3. GPU 训练需将预训练模型、自定义模型、输入数据都放到 GPU
from tqdm import tqdm

def train():
    dataloader = get_dataloader("train.csv")

    # 禁用 BERT 梯度
    for param in bert_model.parameters():
        param.requires_grad_(False)

    model = AiModel().to(device=device)
    optimizer = torch.optim.AdamW(params=model.parameters(), lr=1e-3)
    loss = nn.CrossEntropyLoss()

    model.train()
    for epoch in range(1):
        for i, (input_ids, token_type_ids, attention_mask, labels) in enumerate(tqdm(dataloader), start=1):
            input_ids = input_ids.to(device)
            token_type_ids = token_type_ids.to(device)
            attention_mask = attention_mask.to(device)
            labels = labels.to(device)

            pred_result = model(input_ids, token_type_ids, attention_mask)
            loss_value = loss(pred_result, labels)

            optimizer.zero_grad()
            loss_value.sum().backward()
            optimizer.step()  # 只更新自定义线性层

            if i % 20 == 0:
                pred_index = torch.argmax(pred_result, dim=-1)
                acc = (pred_index == labels).sum().item() / len(labels)
                print(f"第{epoch+1}轮次,批次{i},准确率{round(acc, 4)}")

    torch.save(model.state_dict(), "model/bert.pkl")

模型预测

要点:model.eval() + with torch.no_grad(),GPU 训练的模型在 CPU 使用需 map_location="cpu"

def predict():
    dataloader = get_dataloader("test.csv")

    model = AiModel().to(device)
    model.load_state_dict(torch.load("model/bert.pkl"))

    correct_count = 0
    total_sample_count = 0

    model.eval()
    with torch.no_grad():
        for i, (input_ids, token_type_ids, attention_mask, labels) in enumerate(tqdm(dataloader), start=1):
            input_ids = input_ids.to(device)
            token_type_ids = token_type_ids.to(device)
            attention_mask = attention_mask.to(device)
            labels = labels.to(device)

            pred_result = model(input_ids, token_type_ids, attention_mask)
            pred_index = torch.argmax(pred_result, dim=-1)
            correct_count += (pred_index == labels).sum().item()
            total_sample_count += len(labels)

            if i % 20 == 0:
                acc = correct_count / total_sample_count
                print(f"已预测批次{i},累计平均准确率{round(acc, 4)}")