本项目为书籍《ChatGPT原理与实战：大型语言模型的算法、技术和私有化》中第5章《提示学习与大模型的涌现》实战部分代码-基于Prompt的文本情感分析实战。

针对酒店评论数据集，利用BERT模型在小样本数据下进行模型训练及测试，更深入地了解Prompt任务进行下游任务的流程。

项目主要结构如下：

• data 存放数据的文件夹
  • ChnSentiCorp_htl_all.csv 原始酒店评论情感数据 【数据量3MB不到】
  • sample.json 处理后的语料样例
• prompt_model 已训练好的模型路径
  • config.json
  • pytorch_model.bin
  • vocab.txt
• pretrain_model 预训练文件路径
  • config.json
  • pytorch_model.bin 【虽然提示学习是在预训练中结合下游任务一起训练，但这里演示的还是有增量训练的，不过模型参数没有更新，见后】
  • vocab.txt
• data_helper.py 数据预处理文件
• data_set.py 模型所需数据类文件
• model.py 模型文件
• train.py 模型训练文件
• predict.py 模型推理文件

注意：由于GitHub不方便放模型文件，因此prompt_model文件夹和pretrain_model文件夹中的模型bin文件，请从百度云盘中下载。

模型训练或推理所需环境，请参考requirements.txt文件。

数据预处理需要运行data_helper.py文件，会在data文件夹中生成训练集和测试集文件。

命令如下：

python3 data_helper.py

注意：如果需要修改数据生成路径或名称，请修改data_helper.py文件66-68行，自行定义。

模型训练需要运行train.py文件，会自动生成output_dir文件夹，存放每个epoch保存的模型文件。

命令如下：

python3 train.py --device 0 \
                 --data_dir "data/" \
                 --train_file_path "data/train.json" \
                 --test_file_path "data/test.json" \
                 --pretrained_model_path "pretrain_model/" \
                 --max_len 256 \
                 --train_batch_size 4 \
                 --test_batch_size 16 \
                 --num_train_epochs 10 \
                 --token_handler "mean" 

注意：当服务器资源不同或读者更换数据等时，可以在模型训练时修改响应参数，详细参数说明见代码或阅读书5.4.4小节。

模型训练示例如下：

模型训练阶段损失值及验证集准确率变化如下：

模型训练需要运行predict.py文件，可以采用项目中以提供的模型，也可以采用自己训练后的模型。

命令如下：

python3 predict.py --device 0 --max_len 256

注意：如果修改模型路径，请修改–model_path参数。

模型推理示例如下：

样例1：
输入的评论数据：这家酒店是我在携程定的酒店里面是最差的，房间设施太小气，环境也不好，特别是我住的那天先是第一晚停了一会儿电，第二天停水，没法洗漱，就连厕所也没法上，糟糕头顶。
情感极性：负向
样例2：
输入的评论数据：这个宾馆的接待人员没有丝毫的职业道德可言。我以前定过几次这个宾馆，通常情况下因为入住客人少，因此未发生与他们的冲突，此于他们说要接待一个团，为了腾房，就要强迫已入住的客人退房，而且态度恶劣，言语嚣张，还采用欺骗手段说有其他的房间。
情感极性：负向
样例3：
输入的评论数据：香港马可最吸引人的地方当然是她便利的条件啦；附近的美心酒楼早茶很不错（就在文化中心里头），挺有特色的；
情感极性：正向
样例4：
输入的评论数据：绝对是天津最好的五星级酒店，无愧于万豪的品牌！我住过两次，感觉都非常好。非常喜欢酒店配备的CD机，遥控窗帘，卫生间电动百页窗。早餐也非常好，品种多品质好。能在早餐吃到寿司的酒店不多，我喜欢这里的大堂，很有三亚万豪的风范！
情感极性：正向  
  
  
我自己测试的效果：  

开始对评论数据进行情感分析，输入CTRL + C，则退出
输入的评论数据为：这家酒店是我在携程定的酒店里面是有点问题的。
情感极性为：负向
输入的评论数据为：好吧，就这样吧。
情感极性为：负向
输入的评论数据为：好吧
情感极性为：负向
输入的评论数据为：好
情感极性为：负向
输入的评论数据为：豪酒店啊！
情感极性为：负向
输入的评论数据为：好酒店啊！
情感极性为：正向
输入的评论数据为：豪华酒店！
情感极性为：正向
输入的评论数据为：华酒店！
情感极性为：负向
输入的评论数据为：华丽酒店！
情感极性为：正向

我们看下model.py内容：

\# -\*- coding:utf-8 -\*-
from torch.nn import CrossEntropyLoss
import torch.nn as nn
import torch
from transformers.models.bert.modeling\_bert import BertModel, BertOnlyMLMHead, BertPreTrainedModel


class PromptModel(BertPreTrainedModel):
    """Prompt分类模型"""

    def \_\_init\_\_(self, config):
        super().\_\_init\_\_(config)
        """
        初始化函数
        Args:
            config: 配置参数
        """
        self.bert = BertModel(config, add\_pooling\_layer=False)
        self.cls = BertOnlyMLMHead(config)

    def forward(self, input\_ids, attention\_mask, mask\_index, token\_handler, words\_ids, words\_ids\_mask,
                label=None):
        """
        前向函数，计算Prompt模型预测结果
        Args:
            input\_ids:
            attention\_mask:
            mask\_index:
            token\_handler:
            words\_ids:
            words\_ids\_mask:
            label:

        Returns:

        """
        # 获取BERT模型的输出结果
        sequence\_output = self.bert(input\_ids=input\_ids, attention\_mask=attention\_mask)\[0\]
        # 经过一个全连接层，获取隐层节点状态中的每一个位置的词表
        logits = self.cls(sequence\_output)
        # 获取批次数据中每个样本内容对应mask位置标记
        logits\_shapes = logits.shape
        mask = mask\_index + torch.range(0, logits\_shapes\[0\] - 1, dtype=torch.long, device=logits.device) \* \\
               logits\_shapes\[1\]
        mask = mask.reshape(\[-1, 1\]).repeat(\[1, logits\_shapes\[2\]\])
        # 获取每个mask标记对应的logits内容
        mask\_logits = logits.reshape(\[-1, logits\_shapes\[2\]\]).gather(0, mask).reshape(-1, logits\_shapes\[2\])
        # 获取答案空间映射的标签向量
        label\_words\_logits = self.process\_logits(mask\_logits, token\_handler, words\_ids, words\_ids\_mask)
        # 将其进行归一化以及获取对应标签
        score = torch.nn.functional.softmax(label\_words\_logits, dim=-1)
        pre\_label = torch.argmax(label\_words\_logits, dim=1)
        outputs = (score, pre\_label)
        # 当label不为空时，计算损失值
        if label is not None:
            loss\_fct = CrossEntropyLoss()
            loss = loss\_fct(label\_words\_logits, label)
            outputs = (loss,) + outputs
        return outputs

    def process\_logits(self, mask\_logits, token\_handler, words\_ids, words\_ids\_mask):
        """
        获取答案空间映射的标签向量，用于分类判断
        Args:
            mask\_logits: mask位置信息
            token\_handler: 多token操作策略，包含first、mask和mean
            words\_ids: 标签词id矩阵
            words\_ids\_mask: 标签词id掩码矩阵

        Returns:

        """
        # 获取标签词id及掩码矩阵
        label\_words\_ids = nn.Parameter(words\_ids, requires\_grad=False)
        label\_words\_mask = nn.Parameter(torch.clamp(words\_ids\_mask.sum(dim=-1), max=1), requires\_grad=False)
        # 获取mask位置上标签词向量
        label\_words\_logits = mask\_logits\[:, label\_words\_ids\]
        # 根据多token操作策略进行标签词向量构建
        if token\_handler == "first":
            label\_words\_logits = label\_words\_logits.select(dim=-1, index=0)
        elif token\_handler == "max":
            label\_words\_logits = label\_words\_logits - 1000 \* (1 - words\_ids\_mask.unsqueeze(0))
            label\_words\_logits = label\_words\_logits.max(dim=-1).values
        elif token\_handler == "mean":
            label\_words\_logits = (label\_words\_logits \* words\_ids\_mask.unsqueeze(0)).sum(dim=-1) / (
                    words\_ids\_mask.unsqueeze(0).sum(dim=-1) + 1e-15)
        # 将填充的位置进行掩码
        label\_words\_logits -= 10000 \* (1 - label\_words\_mask)
        # 最终获取mask标记对应的答案空间映射向量
        label\_words\_logits = (label\_words\_logits \* label\_words\_mask).sum(-1) / label\_words\_mask.sum(-1)
        return label\_words\_logits

提示学习的核心还是在里面：

 # 获取每个mask标记对应的logits内容
 mask\_logits = logits.reshape(\[-1, logits\_shapes\[2\]\]).gather(0, mask).reshape(-1, logits\_shapes\[2\])
 # 获取答案空间映射的标签向量
 label\_words\_logits = self.process\_logits(mask\_logits, token\_handler, words\_ids, words\_ids\_mask)
 # 将其进行归一化以及获取对应标签

上述模型是一个基于BERT的Prompt分类模型，主要由两部分组成：

1. self.bert：这是BERT模型的主体部分，用于提取输入文本的特征表示。

2. self.cls：这是一个全连接层，用于从BERT模型的输出中获取每个位置的词表。

模型的功能主要是通过BERT模型提取输入文本的特征，然后通过全连接层获取每个位置的词表，最后通过处理logits和计算损失值来进行分类预测。

在前向传播过程中，模型首先获取BERT模型的输出结果，然后经过全连接层获取每个位置的词表，然后获取每个mask标记对应的logits内容，然后获取答案空间映射的标签向量，最后计算损失值。

在处理logits的过程中，模型首先获取标签词id及掩码矩阵，然后获取mask位置上标签词向量，然后根据多token操作策略进行标签词向量构建，然后将填充的位置进行掩码，最后获取mask标记对应的答案空间映射向量。

总的来说，这个模型的主要功能是进行文本分类预测。

【模型训练】

def train(model, device, train\_data, test\_data, args, tokenizer):
    """
    训练模型
    Args:
        model: 模型
        device: 设备信息
        train\_data: 训练数据类
        test\_data: 测试数据类
        args: 训练参数配置信息
        tokenizer: 分词器
    Returns:
    """
    tb\_write = SummaryWriter()
    if args.gradient\_accumulation\_steps < 1:
        raise ValueError("gradient\_accumulation\_steps参数无效，必须大于等于1")
    # 计算真实的训练batch\_size大小
    train\_batch\_size = int(args.train\_batch\_size / args.gradient\_accumulation\_steps)
    train\_sampler = RandomSampler(train\_data)
    # 构造训练所需的data\_loader
    train\_data\_loader = DataLoader(train\_data, sampler=train\_sampler,
                                   batch\_size=train\_batch\_size, collate\_fn=collate\_func)
    total\_steps = int(len(train\_data\_loader) \* args.num\_train\_epochs / args.gradient\_accumulation\_steps)
    logger.info("总训练步数为:{}".format(total\_steps))
    model.to(device)
    # 获取模型所有参数
    param\_optimizer = list(model.named\_parameters())
    optimizer\_grouped\_parameters = \[
        {'params': \[p for n, p in param\_optimizer if not any(
            nd in n for nd in args.requires\_grad\_params)\], 'weight\_decay': 0.01},
        {'params': \[p for n, p in param\_optimizer if any(
            nd in n for nd in args.requires\_grad\_params)\], 'weight\_decay': 0.0}
    \]

    # 冻结不训练的参数
    for name, param in model.named\_parameters():
        if not any(r\_name in name for r\_name in args.requires\_grad\_params):
            param.requires\_grad = False

    # 验证是否冻结成功
    requires\_grad\_params = \[\]
    for name, param in model.named\_parameters():
        if param.requires\_grad:
            requires\_grad\_params.append(name)
            print("需要训练参数为{}，大小为{}".format(name, param.size()))
    # 设置优化器
    optimizer = AdamW(optimizer\_grouped\_parameters,
                      lr=args.learning\_rate, eps=args.adam\_epsilon)
    scheduler = get\_linear\_schedule\_with\_warmup(optimizer, num\_warmup\_steps=int(args.warmup\_proportion \* total\_steps),
                                                num\_training\_steps=total\_steps)
    # 清空cuda缓存
    torch.cuda.empty\_cache()
    # 将模型调至训练状态
    model.train()
    tr\_loss, logging\_loss, min\_loss = 0.0, 0.0, 0.0
    global\_step = 0
    words\_ids = train\_data.words\_ids.to(device)
    words\_ids\_mask = train\_data.words\_ids\_mask.to(device)
    # 开始训练模型
    for iepoch in trange(0, int(args.num\_train\_epochs), desc="Epoch", disable=False):
        iter\_bar = tqdm(train\_data\_loader, desc="Iter (loss=X.XXX)", disable=False)
        for step, batch in enumerate(iter\_bar):
            # 获取模型训练每个批次所需的输入内容，并放到对应设备上
            input\_ids = batch\["input\_ids"\].to(device)
            attention\_mask = batch\["attention\_mask"\].to(device)
            mask\_index = batch\["mask\_index"\].to(device)
            label = batch\["label"\].to(device)

            # 获取训练结果
            outputs = model.forward(input\_ids=input\_ids, attention\_mask=attention\_mask, mask\_index=mask\_index,
                                    token\_handler=args.token\_handler,
                                    words\_ids=words\_ids, words\_ids\_mask=words\_ids\_mask,
                                    label=label)
            loss = outputs\[0\]
            tr\_loss += loss.item()
            # 将损失值放到Iter中，方便观察
            iter\_bar.set\_description("Iter (loss=%5.3f)" % loss.item())
            # 判断是否进行梯度累积，如果进行，则将损失值除以累积步数
            if args.gradient\_accumulation\_steps > 1:
                loss = loss / args.gradient\_accumulation\_steps
            # 损失进行回传
            loss.backward()
            torch.nn.utils.clip\_grad\_norm\_(model.parameters(), args.max\_grad\_norm)
            # 当训练步数整除累积步数时，进行参数优化
            if (step + 1) % args.gradient\_accumulation\_steps == 0:
                optimizer.step()
                scheduler.step()
                optimizer.zero\_grad()
                global\_step += 1
                # 如果步数整除logging\_steps，则记录学习率和训练集损失值
                if args.logging\_steps > 0 and global\_step % args.logging\_steps == 0:
                    tb\_write.add\_scalar("lr", scheduler.get\_lr()\[0\], global\_step)
                    tb\_write.add\_scalar("train\_loss", (tr\_loss - logging\_loss) /
                                        (args.logging\_steps \* args.gradient\_accumulation\_steps), global\_step)
                    logging\_loss = tr\_loss

        # 每个Epoch对模型进行一次测试，记录测试集的损失
        eval\_loss, eval\_acc = evaluate(model, device, test\_data, args)
        tb\_write.add\_scalar("test\_loss", eval\_loss, global\_step)
        tb\_write.add\_scalar("test\_acc", eval\_acc, global\_step)
        print("test\_loss: {}, test\_acc:{}".format(eval\_loss, eval\_acc))
        model.train()
        # 每个epoch进行完，则保存模型
        output\_dir = os.path.join(args.output\_dir, "checkpoint-{}".format(global\_step))
        model\_to\_save = model.module if hasattr(model, "module") else model
        model\_to\_save.save\_pretrained(output\_dir)
        tokenizer.save\_pretrained(output\_dir)
        # 清空cuda缓存
        torch.cuda.empty\_cache()

关键代码：

for name, param in model.named_parameters():

    if not any(r\_name in name for r\_name in args.requires\_grad\_params):

        param.requires\_grad = False

这段代码的作用是冻结模型中的某些参数，使它们在训练过程中不会被更新。

具体来说，model.named_parameters()是一个迭代器，它返回模型中所有参数的名称（name）和值（param）。然后，对于每一个参数，它检查参数的名称是否包含在args.requires_grad_params列表中。如果参数的名称不在这个列表中，那么就将这个参数的requires_grad属性设置为False，这意味着在训练过程中，这个参数的值不会被更新。

这种技术通常用于迁移学习，当我们想要固定预训练模型的某些层，只训练模型的其他部分时，就会用到这种技术。