【LLM】第三章：项目实操案例：智能输入法项目

【LLM】第三章项目实操案例智能输入法项目说明本篇是根据 https://www.bilibili.com/video/BV1k44LzPEhU?spm_id_from333.788.player.switchvd_sourceb6780e06031ac609460f6fbf017bbb39p38 视频中的案例爆改重构而成的很多细节地方加入了自己的想法和操作。anyway,感谢并致敬原作者一、项目需求和项目思路1、本项目的任务是模型需要根据用户已经输入的文本预测用户下一个可能要输入的词语。2、数据来源是一些真实场景的对话语料有助于模型学习用户的输入习惯和上下文关系。具体获取途径详见大标题二。3、数据处理为了构造适用于下一词预测任务的训练样本首先需要对原始语料进行分词--采用滑动窗口的方式从分词后的序列中提取连续的上下文词片段--暂且规定以6个分词单位为一个窗口把这些窗口片段作为训练集。以每个窗口的前5个词为特征、以最后1个词为标签构建训练集。4、使用什么模型全连接神经网络搭建模型也是可以的。比如像word2vec算法的思想用滑窗的前五个词当作输入最后一个词当作标签来训练模型。当然我们和word2vec算法的输入构造是不同的、目的也是不同的。所以虽然都是全连接架构但是输入输出是不一样的。此外word2vec是训练词向量的所以它训练完毕后是只取模型参数当作词向量即可。而我们这里却是要用训练完毕的模型进行预测的。就是训练模型的目的是不同的。所以这个任务是不能直接用word2vec算法来实现的就是我们得自己搭建网络自己训练模型不能使用已有得成熟算法或者说就是没有现成的轮子供我们直接调用得自己造轮子。考虑到文本预测中的输入输出一般都是序列数据所以使用序列模型应该效果会更好一些所以考虑使用RNN这个传统序列模型架构来搭建网络。原计划是为了对比再搭建一个transformer架构中的编码器作为备选模型架构的但是考虑到transformer的输入要求和rnn不一样rnn不限制输入的序列长度但transformer是严格要求所有序列的长度都要相等不等的要截断或者padding所以这两个架构在数据处理、训练、测试时对输入数据的要求是不同的。而RNN更适合本任务所以本篇只搭建rnn不考虑transformer了。5、项目架构这个案例不再像从前那样我都是写成单文件的形式本项目采取多文件架构。所以这次我用Visual Studio Code作为IDE项目架构和功能实现的设计如下左图具体项目架构搭建如下右图所以本项目的开发工作就是上右图中的8个.py文件,也就是8个模块。具体开发过程详见大标题三。二、训练数据集1、训练数据集来源https://huggingface.co/datasets/Jax-dan/HundredCV-Chathuggingface是预训练模型的平台、预训练模型的权重、使用模型的工具、数据集等。如果huggingface打不开就用hf-mirror.com代替Jax-dan/HundredCV-Chat · Datasets at HF MirrorJax-dan/HundredCV-Chat · Datasets at HF Mirror 具体操作见下图2、查看训练数据集synthesized_.jsonl文件下载完毕后就放到项目架构中的data-raw文件夹中以后我们读取原始数据就从这里读取。1json主要是用于前后端交互的字符串格式。一个{}字典就是一个json对象。2从文件名上看这个文件叫.jsonl,l表示是Line的意思就是说这个这个文件中的每一行就是一个json字符串不是整个文件是一个json对象。3对本项目来说文件中的topic\user1\user2的数据都没用本项目只用dialog中的文本。三、项目开发一编写config.py文件config.py文件的作用有1让src文件夹中的其他模块都可以简单快捷的找到保存在其他文件夹中的文件。所以其他文件夹的路径要写到config.py中。2数据处理过程中、搭建模型过程中、训练模型过程中的参数、超参数也都写到config.py文件中这样我们方便调参。所以config.py文件是边开发别的模块边添加的。下面是config.py文件的全部代码# config.py模块的具体代码如下 # 这是项目文件中其他文件夹的路径 from pathlib import Path PROJECT_ROOT_DIR Path(__file__).parent.parent #项目目录 动态获取项目的根目录 RAW_DATA_DIR PROJECT_ROOT_DIR/data/raw #原始语料目录 语料文件所在的文件夹 PROCESSED_DATA_DIR PROJECT_ROOT_DIR/data/processed #经处理的语料,可以直接喂入模型的样本和标签 LOGS_DIR PROJECT_ROOT_DIR/logs #日志文件的保存目录 MODELS_DIR PROJECT_ROOT_DIR/models #词表、模型等保存的目录 #这是数据处理过程中的超参数 ----- 编写 process.py 模块时定义的超参数 SLID_WINDOW_SIZE 6 #滑窗的大小 #定义喂入模型的小批次 ---- 编写 dataset.py模块时定义 BATCH_SIZE 64 #定义模型搭建的参数 --- 编写 model.py模块时定义的超参数 EMBEDDING_DIM 128 HIDDEN_SIZE 256 #定义模型训练的参数 --- 编写 train.py模块时定义的超参数 LEARNING_RATE 1e-3 EPOCHS 2下面讲解一下上面代码中细节和重点注意点A我们尽量用python自带的pathlib库这样你的项目开发完毕后-打包-部署到其他平台上就不会因为路径问题比如斜杠还是反斜杠、相对路径还是绝对路径等问题而跑不通了。BPath(file)返回的是config.py文件的绝对路径。.parent表示config.py文件的上一级目录(也就是src文件夹的目录)。所以两个.parent就是input_method的目录也就是我们项目的根目录。此后不管找项目中的任何文件都从这个根目录开始寻找。这种操作就是我们软编码了项目的根目录或者说我们动态生成了项目的根目录。项目中的其他文件都以此目录为起始点拼接需要的相对路径即可。这样以后不管项目部署在什么平台还是部署在云端只要pathlib库找到config.py文件在那个平台上的绝对路径就可以生成这个项目的根目录这样所有的相对路径就可以顺利拼接正确了项目才可以正常跑通。否则你会被斜杠、反斜杠、转义符等弄得晕头转向。C这个目录是IDE的工作目录也就是我们是在这个目录下打开项目的。此时我们会配置项目的虚拟环境所以在这个目录下我们可以调用虚拟环境中的python解释器。下面我项目的实际存储地址二编写process.py文件和tokenizer.py文件process.py模块的作用是对原始文本文件进行处理的主要步骤有读synthesized_.jsonl文件--提取文件中的对话句子--将所有句子划分训练句子和测试句子--用训练集的句子构建词表、保存词表到model文件夹--用滑窗构建训练集和测试集并保存。下面用图示说明一下这些步骤细节考虑一是为什么要划分训练句子和测试句子因为模型训练完毕进入使用阶段肯定会遇到没有被训练过的句子如果这里我们划出少量的句子作为测试句子就可以适当的评估模型了。尽管我们是希望模型能见到更多的句子是最好的但终有模型没见过的句子所以还是得留测试集。二是词表是为了生成词和id之间的映射构建词表是必须的因为我们在测试阶段根据用户的输入分词后需要用词表将用户的输入转化为数字编码。三是词表是根据训练句子构建还是根据全部句子构建根据训练句子构建词表因为如果把测试句子的词也放入词表但是在模型训练过程中测试句子的一些词只要它不在训练集中那它也是无法被模型训练的所以词表根据训练句子构建即可其他所有模型没见过的词都用unk标识代替。这也是我们此后处理所有未登录词的处理方法。四是词表的相关信息比如添加未登录词标识符、词表的大小、词和index之间的对应关系等信息我们不仅在process.py文件中用到后面我们训练模型、预测阶段、评估阶段都要用到所以我们要把和词表相关的数据和操作封装到tokenizer.py模块中。# process.py模块的具体代码如下 import pandas as pd import config from sklearn.model_selection import train_test_split from tqdm import tqdm from pathlib import Path import tokenizer # 这是单独提出来的、下面的process函数中的一段可以复用的逻辑用滑窗 从前往后 切训练集的样本和标签 def window_build_dataset(indexed_sentence, desc, window_size): dataset [] for s in tqdm(indexed_sentence, descdesc): if len(s) window_size: slid_window [0] * (window_size-len(s)) s input_sample slid_window[:-1] target_sample slid_window[-1] dataset.append({input:input_sample, target:target_sample}) else: for i in range(len(s)-window_size1): slid_window s[i:iwindow_size] input_sample slid_window[:-1] target_sample slid_window[-1] dataset.append({input:input_sample, target:target_sample}) return dataset def process(): #df pd.read_json(path_or_bufconfig.RAW_DATA_DIR/synthesized_.jsonl, linesTrue, orientrecords) #1、读jsonl文件你的内存大可以这样读 df pd.read_json(path_or_bufconfig.RAW_DATA_DIR/synthesized_.jsonl, linesTrue, orientrecord).sample(frac0.01, random_state0) sentence [sentence.split()[1] for dialog in df[dialog] for sentence in dialog] #2、提取文件中的所有对话句子 train_sentences, test_sentences train_test_split(sentence, test_size0.2, random_state0) #3、将句子划分为训练句子和测试句子,固定随机性方便复现 #4、对训练句子切词-去重-加unk标识构造词表保存词表 tokenizer.JiebaTokenizer.build_vocab(train_sentences, config.MODELS_DIR/vocab.txt) #生成词表 #5、用滑窗构建训练集和测试集并保存 my_tokenizer tokenizer.JiebaTokenizer.from_vocab(config.MODELS_DIR/vocab.txt) #通过词表实例化一个tokenizer对象 #5.1、训练集 indexed_train_sentence [my_tokenizer.encode(s) for s in train_sentences] train_dataset window_build_dataset(indexed_sentenceindexed_train_sentence, desc生成训练集中..., window_sizeconfig.SLID_WINDOW_SIZE) pd.DataFrame(train_dataset).to_json(config.PROCESSED_DATA_DIR/train.jsonl, orientrecords, linesTrue) train_file config.PROCESSED_DATA_DIR/train.jsonl print(训练集保存成功! if (Path(train_file).exists() and Path(train_file).stat().st_size ! 0) else 训练集保存失败!) #5.2 测试集 indexed_test_sentence [my_tokenizer.encode(s) for s in test_sentences] test_dataset window_build_dataset(indexed_sentenceindexed_test_sentence, desc生成测试集中..., window_sizeconfig.SLID_WINDOW_SIZE) pd.DataFrame(test_dataset).to_json(config.PROCESSED_DATA_DIR/test.jsonl, orientrecords, linesTrue) #保存测试集 test_file config.PROCESSED_DATA_DIR/test.jsonl print(测试集保存成功! if (Path(test_file).exists() and Path(test_file).stat().st_size ! 0) else 测试集保存失败!) if __name__ __main__: process()#本模块的功能封装和词表相关的数据和操作 import jieba from tqdm import tqdm class JiebaTokenizer: unk_token unk #这是类的属性 def __init__(self, vocab_list): self.vocab_list vocab_list self.vocab_size len(vocab_list) self.index2word {index:word for index, word in enumerate(vocab_list)} self.word2index {word:index for index, word in enumerate(vocab_list)} self.unk_token_index self.word2index[self.unk_token] staticmethod #静态方法可以用JiebaTokenizer.tokenize()调用这个方法也可以用类实例调用 def tokenize(text): #从代码逻辑讲这个方法没调用类的任何属性和类方法所以它是可以单独写成一个独立的函数 return jieba.lcut(text) #但是这里我们不想单独写就想写到类里面因为从功能逻辑上讲这个功能属于这个类所以要扣个staticmethod帽子 def encode(self, text): tokens self.tokenize(text) #类方法也可以通过类实例调用类的静态方法 return [self.word2index.get(token, self.unk_token_index) for token in tokens] classmethod #类方法和类绑定的方法只能通过类来调用可以访问类的属性和类方法 def build_vocab(cls, sentences, vocab_path): vocab_set set() for sentence in tqdm(sentences, desc构建词表): vocab_set.update(jieba.lcut(sentence)) vocab_list [cls.unk_token] list(vocab_set) print(f词表大小{len(vocab_list)}) with open(vocab_path, w, encodingutf-8) as f: save f.write(\n.join(vocab_list)) print(词表保存成功。。。。。 if save!0 else 词表保存失败.......) classmethod def from_vocab(cls, vocab_path): #根据外部文件构建一个JiebaTokenizer对象 with open(vocab_path, r, encodingutf-8) as f: vocab_list [line.strip() for line in f.readlines()] return cls(vocab_list)okenizer.py模块是把process.py模块中的和词表相关的信息和操作部分的逻辑单独拿出来而写成的有很多python语法方面的内容。下面图是运行process.py文件的结果下面解读几个细节点1、读取synthesized_.jsonl文件json文件是一种非常灵活的标准化数据当我们用pd.read_json()函数去读取json文件时我们得先知道这个json文件是如何生成的也就是知道这个json文件是如何编码自己的数据格式的然后我们才能知道如何读取这个json文件也就是知道如何解码这个json文件所以synthesized_.jsonl文件其实就是从二维dataframe数据结构编码而来的只要照着上面的解码方式就可以顺利解码了2、提取文件中的所有对话句子读出的synthesized_.jsonl文件其实就是dataframe对象我们只要[dialog]就可以切到我们想要的对话文本了然后用split根据冒号切割成一个个句子3、将所有句子打乱划分为训练句子和测试句子4、用训练句子构建词表并保存词表这里仅仅示例如何保存至于要保存到model文件夹中见最前面的代码。5、用滑窗构建训练集和测试集并保存下面只展示这个过程中的重点环节上面展示的代码仅仅是对训练句子的处理过程。测试句子的处理同理。三编写dataset.py文件dataset.py模块的功能是封装数据分小批次 可以直接训练模型的数据。我们后面要搭建模型架构、训练模型这些操作都在pytorch框架下所以喂入模型的数据除了必须是tensor类型外还得把数据的特征和标签打包到一起然后再分小批次batch才能一个一个batch地喂入模型进行模型训练。为什么如此繁琐因为深度学习中的数据一般都是海量的就是样本量非常多比如10万以上的样本量。训练过程也不像机器学习中的算法模型一样先把数据全部加载到内存然后学习出一个模型。深度学习都是分批次batch加载数据的分批次batch学习和迭代的也正是这种机制才使得深度学习可以处理海量数据而避免内存不足的限制。所以pytorch给我们提供了Dataset类来封装数据也提供了DataLoader函数来对训练集和测试集进行分小批次。我们直接拿来用即可。# dataset.py模块的具体代码如下 import torch from torch.utils.data import Dataset from torch.utils.data import DataLoader import pandas as pd import config #封装数据 class ReadData(Dataset): def __init__(self, path): super(ReadData, self).__init__() self.data pd.read_json(path, linesTrue, orientrecords) def __getitem__(self, index): input_tensor torch.tensor(self.data.iloc[index][input], dtypetorch.long) target_tensor torch.tensor(self.data.iloc[index][target], dtypetorch.long) return input_tensor, target_tensor def __len__(self): return len(self.data) #分小批次batch def get_batchdata(trainTrue): path config.PROCESSED_DATA_DIR / (train.jsonl if train else test.jsonl) dataset ReadData(path) batchdata DataLoader(dataset, batch_sizeconfig.BATCH_SIZE, shuffleTrue, drop_lastTrue) return batchdata说明pytorch在对数据进行生成、打包、shuffle、切分、分小批次以及数据预处理比如转化数据类型、数据归一化等操作pytorch都是仅仅存储着数据转化的逻辑关系不是真正的去新生成一些数据转化结果数据而是生成一些映射式或者迭代式的对象在使用的时候也是迭代查询或者递归查询这些对象这种底层的巧妙设计机制主要就是为了适应海量数据而设计的。这些操作中的细节非常非常多这里不可能一一说明想了解更多的细节可参考我以前的博文深度学习入门数据准备与环境设置,-CSDN博客【深度视觉】第二章卷积网络的数据_卷积神经网络 数据分析-CSDN博客【深度视觉】第十一章语义分隔_语义分割-CSDN博客【深度视觉】第十三章生成网络1——PixelRNN/CNN、VAE-CSDN博客上面后两篇博文中都有案例通过案例你可以对pytorch的使用流程了然于胸。下图是这部分代码的效果四编写model.py文件# model.py模块的具体代码如下 from torch import nn import config class ModelRnn(nn.Module): def __init__(self, vocab_size): super().__init__() self.embedding nn.Embedding(num_embeddingsvocab_size, embedding_dimconfig.EMBEDDING_DIM) self.rnn nn.RNN(input_sizeconfig.EMBEDDING_DIM, hidden_sizeconfig.HIDDEN_SIZE) #一定要变换数据结构(sequence, batch, dim) self.linear nn.Linear(config.HIDDEN_SIZE, vocab_size) def forward(self, x): #这里的x是(batch, sequence, dim) embed self.embedding(x) embed embed.transpose(1, 0) #变换数据结构sequence, batch output, hn self.rnn(embed) last_hidden_state output[-1, :, :] # (batch, hidden_size_dim) yhat self.linear(last_hidden_state) # (batch, vocab_size_dim) return yhat1、这篇博文中有embedding层的详细讲解【NLP】第五章注意力机制Attention_qkt公式-CSDN博客2、可不可以不要embedding层使用embedding层表示我们是自己从零开始训练的。所以如果你有训练好的词向量的话你可以不要这个层用nn.Embedding.from_pretrained()代替。nn.Embedding.from_pretrained()是PyTorch中用于加载预训练词向量的类方法‌可直接创建Embedding层并初始化权重无需手动实例化后再赋值。‌‌‌3、RNN层的数据结构和普通的线性层有些不一样。下面是RNN层数据流的展示从中可知数据流动过程的数据结构五编写train.py文件train.py模块的功能是训练模型。而训练模型的步骤是确定设备-实例化模型-加载训练数据-确定损失函数计算训练损失-优化器梯度下降更新模型参数。1、确定设备就是你打算是在cpu上还是gpu上还是云服务器上训练模型。2、实例化模型就是实例化一个模型对象。也就是我们前面写的模型架构。3、加载训练数据就是加载可以直接喂入模型的训练数据。也就是我们前面的dataloader。4、确定损失函数本项目的任务是多分类任务所以用交叉熵损失函数。5、优化器就是根据损失函数的损失值反向传播链式求导也就是求梯度然后用梯度下降法更新模型参数。上述步骤是深度学习中训练模型的常规步骤如果这都不清楚的同学请参考【深度学习】第三章搭建架构-正向传播-计算损失_dnn网络架构-CSDN博客【深度学习】第四章反向传播-梯度计算-更新参数_反向传播参数更新-CSDN博客6、安装TensorBoard我们在训练模型的过程中一般都需要可视化损失函数的下降过程以此来调整模型的超参数所以训练过程的可视化也非常重要。TensorBoard原本是TensorFlow的可视化工具。pytorch原生的可视化工具是在utils模块下的tensorboard但不太好用。所以我们得借助TensorBoardX工具在pytorch框架下使用TensorBoard进行可视化建模。所以我们必须要先pip安装tensorboardX和tensorboard才能使用。# train.py模块的具体代码如下 import torch from dataset import get_batchdata import config import model from tqdm import tqdm from torch.utils.tensorboard import SummaryWriter import time import tokenizer def train(): #准备训练的--设备、数据、模型、损失函数、优化器 device torch.device(cuda if torch.cuda.is_available() else cpu) #设备 train_dataloader get_batchdata() #数据 my_tokenizer tokenizer.JiebaTokenizer.from_vocab(config.MODELS_DIR/vocab.txt) #通过词表实例化一个tokenizer对象 Model model.ModelRnn(vocab_size my_tokenizer.vocab_size).to(device) #模型 criterion torch.nn.CrossEntropyLoss(reductionsum) #损失函数 optimizer torch.optim.Adam(Model.parameters(), lrconfig.LEARNING_RATE) #优化器 writer SummaryWriter(log_dirconfig.LOGS_DIR/time.strftime(%Y-%M-%D_%H.%M.%S)) #可视化 #开始训练 Model.train() #训练模式 epoch_loss [] #保存损失值 best_loss float(inf) #保存模型时使用的阈值 for epoch in range(config.EPOCHS): print(*10, f Epoch: {epoch1} , *10) total_loss 0 for x, y in tqdm(train_dataloader, desc训练中): x x.to(device) # x.shape--torch.Size([64, 5]) y y.to(device) # y.shape--torch.Size([64]) yhat Model.forward(x) #yhat.shape--torch.Size([64, 21155]) loss criterion(yhat, y) loss.backward() optimizer.step() optimizer.zero_grad() total_loss loss.item() epoch_loss_temp total_loss/len(train_dataloader) print(f本次epoch的平均损失{epoch_loss_temp}) epoch_loss.append(epoch_loss_temp) writer.add_scalar(epoch_loss, epoch_loss_temp, epoch) if epoch_loss_temp best_loss: best_loss epoch_loss_temp torch.save(Model.state_dict(), config.MODELS_DIR/best_model.pt) #pt就是pytorch的简写是pytorch自定义的文件格式 print(模型保存成功) writer.close() if __name__ __main__: train()这是train.py文件的训练过程这是可视化的效果六编写predict.py文件预测环节就是用户输入一个字或者一个词就开始预测下一步用户可能要输入的字或词。所以一般情况下预测部分都是被封装成http接口让客户用但是这里我们简化一下用input函数承接用户的输入。# predict.py模块的具体代码如下 import torch import config import model import tokenizer def predict(text, tokenizer, Model, device): tokens tokenizer.encode(text) input_tensor torch.tensor([tokens], dtypetorch.long).to(device) Model.eval() with torch.no_grad(): output Model(input_tensor) top5_idx torch.topk(output, k5).indices top5_list top5_idx.tolist()[0] top5_tokens [tokenizer.index2word[index] for index in top5_list] return top5_tokens def run_predic(): print(欢迎使用输入法模型(输入q或者quit退出)) device torch.device(cuda if torch.cuda.is_available() else cpu) #设备 my_tokenizer tokenizer.JiebaTokenizer.from_vocab(config.MODELS_DIR/vocab.txt) #通过词表实例化一个tokenizer对象 Model model.ModelRnn(vocab_size my_tokenizer.vocab_size).to(device) #模型 Model.load_state_dict(torch.load(config.MODELS_DIR/best_model.pt)) input_history #存储客户的输入历史 while True: user_input input(请输入 ) if user_input in [q, quit]: print(欢迎下次再来) break if user_input.strip() : print(请输入内容) continue input_history user_input print(f历史输入{input_history}) top5_tokens predict(input_history, my_tokenizer, Model, device) print(f预测结果{top5_tokens}) if __name____main__: run_predic()这是pedict.py脚本的运行效果七编写evaluate.py脚本模型评估我们使用top1准确率和top5准确率两个指标来衡量。# evaluate.py模块的具体代码如下 import torch import config import model import dataset import tokenizer def run_evaluate(): device torch.device(cuda if torch.cuda.is_available() else cpu) #设备 my_tokenizer tokenizer.JiebaTokenizer.from_vocab(config.MODELS_DIR/vocab.txt) print(词表加载成功 if my_tokenizer ! None else 词表加载失败) Model model.ModelRnn(vocab_sizemy_tokenizer.vocab_size).to(device) #模型 Model.load_state_dict(torch.load(config.MODELS_DIR/best_model.pt)) print(模型加载成功) test_dataloader dataset.get_batchdata(trainFalse) #测试数据 top1_acc_count 0 #top1准确率 top5_acc_count 0 #top5准确率 total_count 0 Model.eval() with torch.no_grad(): for x, y in test_dataloader: output Model(x) #正向传播 top5_index torch.topk(output, k5).indices.tolist() for y, top5_index in zip(y, top5_index): total_count 1 if y top5_index[0]: top1_acc_count 1 if y in top5_index: top5_acc_count 1 top1_acc top1_acc_count/total_count top5_acc top5_acc_count/total_count print(ftop1_acc: {top1_acc}) print(ftop5_acc: {top5_acc}) if __name__ __main__: run_evaluate()这是evaluate.py脚本的运行结果至此项目开发完毕。