多项选择任务与迁移学习

import keras_nlp
import keras
import tensorflow as tf  # 仅用于 tf.data。

import numpy as np
import pandas as pd

import matplotlib.pyplot as plt

!wget "https://github.com/rowanz/swagaf/archive/refs/heads/master.zip" -O swag.zip
!unzip -q swag.zip

!ls swagaf-master/data

class CFG:
    preset = "deberta_v3_extra_small_en"  # 预训练模型名称
    sequence_length = 200  # 输入序列长度
    seed = 42  # 随机种子
    epochs = 5  # 训练轮数
    batch_size = 8  # 批量大小
    augment = True  # 增强（随机选项）

keras.utils.set_random_seed(CFG.seed)

# 训练数据
train_df = pd.read_csv(
    "swagaf-master/data/train.csv", index_col=0
)  # 将 CSV 文件读取到 DataFrame
train_df = train_df.sample(frac=0.02)
print("# 训练数据: {:,}".format(len(train_df)))

# 验证数据
valid_df = pd.read_csv(
    "swagaf-master/data/val.csv", index_col=0
)  # 将 CSV 文件读取到 DataFrame
valid_df = valid_df.sample(frac=0.02)
print("# 验证数据: {:,}".format(len(valid_df)))

# 定义一个函数，根据提示和选项创建选项
def make_options(row):
    row["options"] = [
        f"{row.startphrase}\n{row.ending0}",  # 选项 0
        f"{row.startphrase}\n{row.ending1}",  # 选项 1
        f"{row.startphrase}\n{row.ending2}",  # 选项 2
        f"{row.startphrase}\n{row.ending3}",
    ]  # 选项 3
    return row

train_df = train_df.apply(make_options, axis=1)
valid_df = valid_df.apply(make_options, axis=1)

preprocessor = keras_nlp.models.DebertaV3Preprocessor.from_preset(
    preset=CFG.preset,  # 模型名称
    sequence_length=CFG.sequence_length,  # 最大序列长度，较短时将进行填充
)

outs = preprocessor(train_df.options.iloc[0])  # 处理第一行的选项

# 显示每个处理后输出的形状
for k, v in outs.items():
    print(k, ":", v.shape)

def preprocess_fn(text, label=None):
    text = preprocessor(text)  # 预处理文本
    return (
        (text, label) if label is not None else text
    )  # 返回处理后的文本和标签（如果可用）

def option_shuffle(options, labels, prob=0.50, seed=None):
    if tf.random.uniform([]) > prob:  # 洗牌概率检查
        return options, labels
    # 按相同顺序洗牌选项和标签的索引
    indices = tf.random.shuffle(tf.range(tf.shape(options)[0]), seed=seed)
    # 洗牌选项和标签
    options = tf.gather(options, indices)
    labels = tf.gather(labels, indices)
    return options, labels

def augment_fn(text, label=None):
    text, label = option_shuffle(text, label, prob=0.5)  # 洗牌选项
    return (text, label) if label is not None else text

def build_dataset(
    texts,
    labels=None,
    batch_size=32,
    cache=False,
    augment=False,
    repeat=False,
    shuffle=1024,
):
    AUTO = tf.data.AUTOTUNE  # AUTOTUNE 选项
    slices = (
        (texts,)
        if labels is None
        else (texts, keras.utils.to_categorical(labels, num_classes=4))
    )  # 创建切片
    ds = tf.data.Dataset.from_tensor_slices(slices)  # 从切片创建数据集
    ds = ds.cache() if cache else ds  # 如果启用，缓存数据集
    if augment:  # 如果启用，应用增强
        ds = ds.map(augment_fn, num_parallel_calls=AUTO)
    ds = ds.map(preprocess_fn, num_parallel_calls=AUTO)  # 映射预处理函数
    ds = ds.repeat() if repeat else ds  # 如果启用，重复数据集
    opt = tf.data.Options()  # 创建数据集选项
    if shuffle:
        ds = ds.shuffle(shuffle, seed=CFG.seed)  # 如果启用，打乱数据集
        opt.experimental_deterministic = False
    ds = ds.with_options(opt)  # 设置数据集选项
    ds = ds.batch(batch_size, drop_remainder=True)  # 批量处理数据集
    ds = ds.prefetch(AUTO)  # 预取下一个批次
    return ds  # 返回构建的数据集

# 构建训练数据加载器
train_texts = train_df.options.tolist()  # 提取训练文本
train_labels = train_df.label.tolist()  # 提取训练标签
train_ds = build_dataset(
    train_texts,
    train_labels,
    batch_size=CFG.batch_size,
    cache=True,
    shuffle=True,
    repeat=True,
    augment=CFG.augment,
)

# 构建验证数据加载器
valid_texts = valid_df.options.tolist()  # 提取验证文本
valid_labels = valid_df.label.tolist()  # 提取验证标签
valid_ds = build_dataset(
    valid_texts,
    valid_labels,
    batch_size=CFG.batch_size,
    cache=True,
    shuffle=False,
    repeat=False,
    augment=False,
)

import math


def get_lr_callback(batch_size=8, mode="cos", epochs=10, plot=False):
    lr_start, lr_max, lr_min = 1.0e-6, 0.6e-6 * batch_size, 1e-6
    lr_ramp_ep, lr_sus_ep = 2, 0

    def lrfn(epoch):  # 学习率更新函数
        if epoch < lr_ramp_ep:
            lr = (lr_max - lr_start) / lr_ramp_ep * epoch + lr_start
        elif epoch < lr_ramp_ep + lr_sus_ep:
            lr = lr_max
        else:
            decay_total_epochs, decay_epoch_index = (
                epochs - lr_ramp_ep - lr_sus_ep + 3,
                epoch - lr_ramp_ep - lr_sus_ep,
            )
            phase = math.pi * decay_epoch_index / decay_total_epochs
            lr = (lr_max - lr_min) * 0.5 * (1 + math.cos(phase)) + lr_min
        return lr

    if plot:  # 如果 plot 为 True，则绘制学习率曲线
        plt.figure(figsize=(10, 5))
        plt.plot(
            np.arange(epochs),
            [lrfn(epoch) for epoch in np.arange(epochs)],
            marker="o",
        )
        plt.xlabel("epoch")
        plt.ylabel("lr")
        plt.title("学习率调度器")
        plt.show()

    return keras.callbacks.LearningRateScheduler(
        lrfn, verbose=False
    )  # 创建学习率回调


_ = get_lr_callback(CFG.batch_size, plot=True)

def get_callbacks():
    callbacks = []
    lr_cb = get_lr_callback(CFG.batch_size)  # 获取学习率回调
    ckpt_cb = keras.callbacks.ModelCheckpoint(
        f"best.keras",
        monitor="val_accuracy",
        save_best_only=True,
        save_weights_only=False,
        mode="max",
    )  # 获取模型检查点回调
    callbacks.extend([lr_cb, ckpt_cb])  # 添加学习率和检查点回调
    return callbacks  # 返回回调列表


callbacks = get_callbacks()