AI基本结构10-mlp实现简单nlp

前置程序词嵌入import torch import torch.nn as nn import torch.nn.functional as F import string # 定义字典 char2indx {s: i for i, s in enumerate(string.ascii_lowercase)} print(char2indx) example love idx [] for i in example: idx.append(char2indx[i]) idx torch.tensor(idx) print(idx) # 使用独热编码将文本转换为二维张量 num_claz len(char2indx.keys()) x F.one_hot(idx, num_classesnum_claz).float() print(x,x.shape) dims 5 num_claz len(char2indx.keys()) x F.one_hot(idx, num_classesnum_claz).float() w torch.randn(num_claz, dims)对26个小写字符进行26维向量化映射并输出示例对应的独热向量再尝试对26维进行压缩即使用自定义矩阵 对 通过向量化长度为n的字符串得到的n*26的矩阵 进行矩阵的空间变换。若所需压缩维度为m则进行矩阵乘法xw[n*26][26*m]。{a: 0, b: 1, c: 2, d: 3, e: 4, f: 5, g: 6, h: 7, i: 8, j: 9, k: 10, l: 11, m: 12, n: 13, o: 14, p: 15, q: 16, r: 17, s: 18, t: 19, u: 20, v: 21, w: 22, x: 23, y: 24, z: 25} tensor([11, 14, 21, 4]) tensor([[0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 1., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0.], [0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 1., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0.], [0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 1., 0., 0., 0., 0.], [0., 0., 0., 0., 1., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0., 0.]]) torch.Size([4, 26]) tensor([[ 0.8121, 1.7843, -1.8114, -0.1955, -0.1218], [ 0.7272, -0.1841, -1.0490, 0.5851, -0.5069], [-1.1661, 1.7609, -0.5384, 1.6000, -0.6927], [ 0.5573, -0.7052, 0.6820, 0.8804, -0.9916]]) torch.Size([4, 5])这里在python环境下可以简化代码的计算流程w[idx], w[idx].shape对于pytorch可以将用于制作独热编码长度为n的idx直接传给形状为w.shapedim*m的变换矩阵x自动生成独热编码后降维为目标矩阵。故可以以此编写实现上述嵌入功能的函数class Embedding: def __init__(self, num_embeddings, embedding_dims): self.weight torch.randn((num_embeddings, embedding_dims), requires_gradTrue) def __call__(self, x): self.out self.weight[x] return self.out def parameters(self): return [self.weight]编解码class CharTokenizer: def __init__(self, data, begin_ind0, end_ind1): # data: list[str] # 得到所有的字符 chars sorted(list(set(.join(data)))) self.char2ind {s: i 2 for i, s in enumerate(chars)} self.char2ind[|b|] begin_ind self.char2ind[|e|] end_ind self.ind2char {v: k for k, v in self.char2ind.items()} self.begin_ind begin_ind self.end_ind end_ind def encode(self, x): # x: str return [self.char2ind[i] for i in x] def decode(self, x): # x: int or list[x] if isinstance(x, int): return self.ind2char[x] return [self.ind2char[i] for i in x] #测试 tokenizer CharTokenizer(datasets[whole_func_string]) test_str def f(x): re tokenizer.encode(test_str) print(re) print(.join(tokenizer.decode(re)))单文本训练数据生成def autoregressive_trans(text, tokenizer, context_length10): # text: str inputs, labels [], [] bind tokenizer.begin_ind eind tokenizer.end_ind enc tokenizer.encode(text) # 增加特殊字符 data [bind] * context_length enc [eind] for i in range(len(data) - context_length): inputs.append(data[i: i context_length]) labels.append(data[i context_length]) return inputs, labels #测试 inputs, labels autoregressive_trans(test_str, tokenizer, 3) for a, b in zip(inputs, labels): print(f{.join(tokenizer.decode(a))} ----- {tokenizer.decode(b)})autoregressive_trans 函数用于将一段文本转换为自回归语言模型Autoregressive Language Model的训练样本。函数首先利用 tokenizer.encode() 将输入文本编码为对应的 token 序列并在序列前添加 context_length 个起始标记begin_ind末尾添加结束标记end_ind以便模型学习文本的起始和结束信息。随后采用滑动窗口Sliding Window方式以长度为 context_length 的连续 token 序列作为模型输入 inputs将紧随其后的下一个 token 作为预测目标 labels不断向后滑动直到遍历完整个序列从而构造出大量“前文预测下一个词”的训练样本。这种处理方式符合自回归模型“根据历史上下文预测下一个 token”的训练机制使模型能够学习文本的时序依赖关系。最后的测试代码通过设置较小的上下文长度context_length3并利用 tokenizer.decode() 将 token 序列还原为文本直观展示了每个输入序列与对应预测目标之间的映射关系例如“前三个字符 → 预测第四个字符”便于验证数据预处理过程是否正确。基于数据集的数据生成#需要处理的数据可能是多列文本所以需要分类处理 def process(data, tokenizer): text data[whole_func_string] # text: str if isinstance(text, str): inputs, labels autoregressive_trans(text, tokenizer) return {inputs: inputs, labels: labels} # text: list[str] inputs, labels [], [] for t in text: i, l autoregressive_trans(t, tokenizer) inputs i labels l return {inputs: inputs, labels: labels} #测试 process(datasets[8], tokenizer) process(datasets[8: 9], tokenizer)该函数 process 用于对数据集中的文本进行统一预处理并兼容单条样本和批量样本两种输入形式。首先从数据中读取 whole_func_string 字段并通过 isinstance(text, str) 判断当前输入是否为单条文本若为字符串则直接调用 autoregressive_trans() 将文本转换为自回归模型所需的输入序列 inputs 和训练标签 labels若输入为文本列表即 batchedTrue 时传入的一个批次数据则依次遍历列表中的每条文本对每条文本执行相同的转换操作并将所有样本的 inputs 和 labels 合并到两个列表中返回。最终该函数统一输出包含 inputs 和 labels 的字典便于后续使用 Dataset.map() 对整个数据集进行批量预处理。最后两行测试代码分别验证了函数对单个样本datasets[8]和单样本批次datasets[8:9]两种输入格式均能正确完成处理说明该函数具有良好的通用性和兼容性。数据准备# 将数据分为训练集和测试集 tokenized datasets.train_test_split(test_size0.1, seed1024, shuffleTrue) f lambda x: process(x, tokenizer) tokenized tokenized.map(f, batchedTrue, remove_columnsdatasets.column_names) tokenized.set_format(typetorch, devicedevice) #测试 print(tokenized[train][inputs].shape, tokenized[train][labels].shape) train_loader DataLoader(tokenized[train], batch_sizebatch_size, shuffleTrue) test_loader DataLoader(tokenized[test], batch_sizebatch_size, shuffleTrue) #测试 print(next(iter(train_loader)))代码作用f lambda x: process(x, tokenizer)定义一个匿名函数将数据批次交给process()并传入tokenizer。tokenized.map(f, batchedTrue, remove_columns...)对整个数据集批量执行分词/预处理并删除原始列生成新的特征列。tokenized.set_format(typetorch, devicedevice)将数据集输出格式设为 PyTorchTensor可直接放到指定设备方便后续DataLoader和模型训练。模型定义class CharMLP(nn.Module): def __init__(self, vs): # vs 字典大小 super().__init__() self.emb nn.Embedding(vs, 30) self.hidden1 nn.Linear(10 * 30, 200) self.hidden2 nn.Linear(200, 100) self.lm nn.Linear(100, vs) def forward(self, x): # x: (B, 10) 10为窗口大小 B x.shape[0] emb self.emb(x) # (B, 10, 30) h emb.view(B, -1) # (B, 300) h F.relu(self.hidden1(h)) # (B, 200) h F.relu(self.hidden2(h)) # (B, 100) out self.lm(h) # (B, vs) return out #测试 model CharMLP(len(tokenizer.char2ind)).to(device)CharMLP( (emb): Embedding(99, 30) (hidden1): Linear(in_features300, out_features200, biasTrue) (hidden2): Linear(in_features200, out_features100, biasTrue) (lm): Linear(in_features100, out_features99, biasTrue) )这部分和之前的差不多不说了评估函数定义def estimate_loss(model): re {} # 将模型切换至评估模式 model.eval() re[train] _loss(model, train_loader) re[test] _loss(model, test_loader) # 将模型切换至训练模式 model.train() return re torch.no_grad() def _loss(model, data_loader): 计算模型在不同数据集下面的评估指标 loss [] data_iter iter(data_loader) # 随机使用多个批量数据来预估模型效果 for k in range(eval_iters): data next(data_iter, None) if data is None: data_iter iter(data_loader) data next(data_iter, None) inputs, labels data[inputs], data[labels] logits model(inputs) loss.append(F.cross_entropy(logits, labels).item()) return torch.tensor(loss).mean().item() estimate_loss(model)1estimate_loss 函数作用estimate_loss 用于评估模型在训练集和测试集上的损失情况。函数首先将模型切换到评估模式model.eval()分别调用 _loss 计算训练集和测试集的平均损失并将结果保存到字典中最后再调用 model.train() 恢复训练模式确保后续模型能够继续正常训练。2torch.no_grad() 装饰器torch.no_grad() 表示在评估过程中关闭梯度计算不会构建计算图也不会更新模型参数从而减少显存占用并提高推理速度。这是模型验证和测试阶段的标准做法。3_loss 函数作用_loss 用于计算指定数据集上的平均损失值。函数首先创建一个空列表用于保存每个批次的损失然后从 DataLoader 中依次读取数据。当数据遍历结束时会重新创建迭代器继续取数据保证能够连续计算 eval_iters 个批次的损失。4损失计算过程对于每个批次的数据函数读取输入 inputs 和标签 labels将输入送入模型得到预测结果 logits随后利用交叉熵损失函数 F.cross_entropy() 计算预测结果与真实标签之间的误差并将每个批次的损失值保存到列表中。5返回评估结果所有批次计算完成后函数对损失列表求平均值并返回一个标量作为该数据集的平均损失。最终调用 estimate_loss(model) 后将返回形如 {train: train_loss, test: test_loss} 的字典用于监控模型在训练集和测试集上的性能变化判断模型是否收敛以及是否存在过拟合现象。模型训练def train_model(model, optimizer, epochs10): # 记录模型在训练集上的模型损失 lossi [] for epoch in range(epochs): for i, data in enumerate(train_loader, 0): inputs, labels data[inputs], data[labels] optimizer.zero_grad() logits model(inputs) loss F.cross_entropy(logits, labels) lossi.append(loss.item()) loss.backward() optimizer.step() # 评估模型并输出结果 stats estimate_loss(model) train_loss ftrain loss {stats[train]:.4f} test_loss ftest loss {stats[test]:.4f} print(fepoch {epoch:2}: {train_loss}, {test_loss}) return lossi训练老三样损失优化循环epoch 0: train loss 1.3786, test loss 1.5191 epoch 1: train loss 1.2663, test loss 1.4948 epoch 2: train loss 1.2127, test loss 1.4368 epoch 3: train loss 1.1642, test loss 1.3986 epoch 4: train loss 1.1281, test loss 1.3736 epoch 5: train loss 1.1588, test loss 1.3578 epoch 6: train loss 1.0906, test loss 1.4026 epoch 7: train loss 1.0988, test loss 1.3691 epoch 8: train loss 1.0670, test loss 1.3782 epoch 9: train loss 1.0861, test loss 1.3579测试模型torch.no_grad() def generate(model, context, tokenizer, max_new_tokens300): # context: (1, 10) out [] model.eval() for _ in range(max_new_tokens): logits model(context) # (1, 99) probs F.softmax(logits, dim-1) # (1, 99) # 随机生成文本 ix torch.multinomial(probs, num_samples1) # (1, 1) # 更新背景 context torch.concat((context[:, 1:], ix), dim-1) out.append(ix.item()) if out[-1] tokenizer.end_ind: break model.train() return out #测试 context torch.zeros((1, 10), dtypetorch.long, devicedevice) print(.join(tokenizer.decode(generate(model, context, tokenizer)))) l train_model(model, optim.Adam(model.parameters(), lrlearning_rate)) #测试训练效果 context torch.zeros((1, 10), dtypetorch.long, devicedevice) print(.join(tokenizer.decode(generate(model, context, tokenizer))))def galuary).defastist(): it for a ry, types.Mapords((axt3rBL( sc.basc.secop.copy(sermine, fields, pretator) return [sc.tist(Func, start/sparsoc: repy. Fils all_freqect_ReplaceAPMudtil.IEj)[.]).collect() .statter()ramefter;s[).sPart samples to 1 r None by.__sewts._1generate 函数作用generate 用于利用训练好的自回归语言模型生成文本。函数以初始上下文 context 为输入通过不断预测下一个 token并将预测结果加入上下文实现逐词逐 token生成文本直到达到最大生成长度或遇到结束标记。2关闭梯度与评估模式函数使用 torch.no_grad() 装饰器关闭梯度计算并调用 model.eval() 将模型切换到评估模式。这能够减少显存占用提高推理速度同时避免 Dropout 等训练层影响生成结果。3预测下一个 Token在每一次循环中将当前上下文输入模型得到预测结果 logits再利用 F.softmax() 将其转换为概率分布 probs表示每个 token 作为下一个字符或单词的生成概率。4随机采样生成文本函数使用 torch.multinomial() 根据概率分布随机采样一个 token而不是直接选择概率最大的 token。这样生成的文本具有一定随机性和多样性避免每次生成完全相同的结果。5更新上下文窗口生成的新 token 会追加到当前上下文末尾同时丢弃最前面的一个 token即通过 context torch.concat((context[:, 1:], ix), dim-1) 构造新的固定长度上下文。这种滑动窗口机制保证模型始终使用最近的 context_length 个 token 进行预测。6结束条件与结果返回每生成一个 token都会保存到输出列表 out 中若生成的 token 为结束标记 tokenizer.end_ind则提前结束生成否则持续生成直到达到 max_new_tokens。最后恢复模型训练模式model.train()并返回生成的 token 序列。缺陷1无法捕捉窗口外的关联关系多层感知器MLP通常采用固定长度的上下文窗口作为输入只能利用窗口内的历史信息预测下一个词。当文本中存在跨越较长距离的语义依赖时如代词指代、长句中的前后呼应窗口之外的信息无法被模型获取因此难以学习文本中的长期依赖关系。2模型结构固定无法随数据动态调整MLP 的网络结构由输入层、隐藏层和输出层组成输入维度在模型设计时已经固定因此只能处理固定长度的输入序列。当输入内容发生变化时模型无法根据不同位置之间的重要程度动态分配关注权重也无法灵活适应不同长度或不同语义复杂度的文本这限制了模型对复杂语言特征的表达能力。3运算效率低下随着上下文窗口长度的增加MLP 输入层的维度也会线性增大导致网络参数数量迅速增加计算量和存储开销显著提高。同时每个输入位置都需要与全连接层中的所有神经元进行计算使模型训练和推理效率较低难以扩展到大规模文本和长序列任务。相比之下循环神经网络RNN和 Transformer 等模型能够更加高效地处理序列数据并具有更好的扩展性。