文章目录
- 一、引言
- 二、数据预处理与词汇表构建
- 2.1 文本预处理简介
- 2.2 构建词汇表
- 2.3 文本编码与数据集构造
- 2.4 数据加载与 DataLoader 构建
- 三、Transformer 文本生成模型构建
- 3.1 位置编码模块
- 3.2 Transformer 文本生成模型
- 四、模型训练与优化
- 4.1 损失函数与优化器
- 4.2 训练函数定义
- 4.3 训练过程
- 五、文本生成与评估
- 5.1 文本生成函数设计
- 5.2 生成示例
- 六、扩展与优化方向
- 6.1 数据扩展与预处理
- 6.2 模型结构扩展
- 6.3 文本生成策略
- 6.4 部署与应用
- 七、结论
- 一键复制代码
一、引言
在当今自然语言处理领域,文本生成模型已经广泛应用于对话系统、文章撰写、内容创作等场景。本文将以从零开始开发一个中文文本生成模型为例,详细讲解如何构建数据预处理流程、设计词汇表、构造 Transformer 模型、训练模型以及实现文本生成。整个案例使用简化的中文数据,仅作教学演示,但同时也介绍了实际开发中可扩展的方向。
二、数据预处理与词汇表构建
在任何自然语言处理项目中,数据预处理都是基础且关键的环节。下面介绍如何对中文文本进行预处理,并构建一个基于字符级的词汇表。
2.1 文本预处理简介
中文文本处理较为特殊,一种常用方法是逐字分词(即把每个汉字看作一个基本单元)。在本例中,我们使用简单的正则替换去除空格,并对文本进行逐字切分。
def chinese_tokenizer(text):"""对中文文本进行分词(逐字分词),去除空格"""text = text.replace(" ", "")return list(text)
2.2 构建词汇表
词汇表是将文本中的字符映射为数字索引的关键数据结构。本例中,我们构建一个简单的词汇表,并预留特殊标记:
<pad>
用于填充,<unk>
用于处理词汇表中未出现的字符。
def build_vocab(texts):"""根据文本数据构建词汇表,保留所有字符特殊标记:"<pad>" 用于填充,"<unk>" 用于未知字符"""vocab = {"<pad>": 0, "<unk>": 1}idx = 2for text in texts:for char in chinese_tokenizer(text):if char not in vocab:vocab[char] = idxidx += 1return vocab
2.3 文本编码与数据集构造
将文本转换为字符索引序列,并利用滑动窗口方法生成输入和目标序列,是构造训练样本的关键。这里我们设定输入序列长度为 10,目标序列为右移一位的结果。
def encode_text(text, vocab):"""将中文文本转换为字符索引序列,未在词汇表中的字符映射为 <unk>"""tokens = chinese_tokenizer(text)return [vocab.get(token, vocab["<unk>"]) for token in tokens]class TextGenerationDataset(Dataset):def __init__(self, texts, vocab, seq_len=10):"""利用滑动窗口生成样本,每个样本长度为 seq_len+1输入为前 seq_len 个字符,目标为右移一位后的字符序列"""self.vocab = vocabself.seq_len = seq_lenself.data = []for text in texts:encoded = encode_text(text, vocab)if len(encoded) < seq_len + 1:continuefor i in range(len(encoded) - seq_len):self.data.append(encoded[i:i+seq_len+1])def __len__(self):return len(self.data)def __getitem__(self, idx):seq = self.data[idx]return torch.tensor(seq[:-1], dtype=torch.long), torch.tensor(seq[1:], dtype=torch.long)
2.4 数据加载与 DataLoader 构建
我们以四条示例中文数据构造数据集,并使用 PyTorch 的 DataLoader 进行批量数据加载。
# 示例中文数据
texts = ["今天天气很好,我们一起出去玩吧!","我喜欢学习人工智能和深度学习。","自然语言处理是计算机科学的重要方向。","数据科学改变世界,未来无限可能。"
]vocab = build_vocab(texts)
print("词汇表大小:", len(vocab))# 训练时输入序列长度设为 10
seq_len = 10
dataset = TextGenerationDataset(texts, vocab, seq_len=seq_len)
dataloader = DataLoader(dataset, batch_size=4, shuffle=True)# 生成逆向词汇表,用于将索引转换为字符
inv_vocab = {idx: char for char, idx in vocab.items()}
三、Transformer 文本生成模型构建
Transformer 模型在文本生成任务中表现出色。本文将介绍如何设计位置编码模块、构造 Transformer 编码器以及最终的输出层。
3.1 位置编码模块
由于 Transformer 没有循环结构,必须通过位置编码注入序列中每个元素的位置信息。我们采用经典的正弦和余弦函数方式构造位置编码,并设置最大长度为 50,以便生成过程中支持更长序列。
class PositionalEncoding(nn.Module):def __init__(self, d_model, max_len=50):"""初始化位置编码,max_len 设置为 50,确保生成时支持更长序列"""super(PositionalEncoding, self).__init__()pe = torch.zeros(max_len, d_model) # [max_len, d_model]position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) # [max_len, 1]div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))pe[:, 0::2] = torch.sin(position * div_term)pe[:, 1::2] = torch.cos(position * div_term)pe = pe.unsqueeze(0) # [1, max_len, d_model]self.register_buffer('pe', pe)def forward(self, x):"""输入 x: [batch_size, seq_len, d_model]将位置编码加到输入上,要求 x.size(1) <= max_len"""x = x + self.pe[:, :x.size(1)]return x
3.2 Transformer 文本生成模型
模型由词嵌入层、位置编码模块、Transformer 编码器和全连接输出层构成。训练时输入序列长度为 10,但模型位置编码最大支持 50,因此生成时可以逐步扩展序列。
class TransformerLM(nn.Module):def __init__(self, vocab_size, embed_dim, num_heads, num_layers, dropout=0.1, max_len=50):""":param vocab_size: 词汇表大小:param embed_dim: 嵌入向量维度:param num_heads: 注意力头数:param num_layers: Transformer 编码器层数:param dropout: Dropout 概率:param max_len: 位置编码最大长度(生成时支持的最大序列长度)"""super(TransformerLM, self).__init__()self.embed = nn.Embedding(vocab_size, embed_dim)self.pos_encoder = PositionalEncoding(embed_dim, max_len=max_len)encoder_layer = nn.TransformerEncoderLayer(d_model=embed_dim, nhead=num_heads, dropout=dropout)self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)self.fc = nn.Linear(embed_dim, vocab_size)self.embed_dim = embed_dimdef forward(self, x):""":param x: 输入序列 [batch_size, seq_len]:return: logits [batch_size, seq_len, vocab_size]"""x = self.embed(x) * math.sqrt(self.embed_dim)x = self.pos_encoder(x)# TransformerEncoder 要求输入形状为 [seq_len, batch_size, embed_dim]x = x.transpose(0, 1)x = self.transformer(x)x = x.transpose(0, 1) # 恢复为 [batch_size, seq_len, embed_dim]logits = self.fc(x)return logitsvocab_size = len(vocab)
embed_dim = 128
num_heads = 4
num_layers = 2# 注意:训练时输入序列长度为 10,但生成时模型支持最长 50
model = TransformerLM(vocab_size, embed_dim, num_heads, num_layers, dropout=0.1, max_len=50)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)
四、模型训练与优化
训练过程中,我们使用交叉熵损失函数来比较模型输出与真实目标,并使用 Adam 优化器调整模型参数。下面给出训练函数以及训练过程中的监控方法。
4.1 损失函数与优化器
criterion = nn.CrossEntropyLoss(ignore_index=vocab["<pad>"])
optimizer = optim.Adam(model.parameters(), lr=0.001)
4.2 训练函数定义
逐批次读取数据,对模型进行前向传播、损失计算、反向传播和参数更新。
def train_epoch(model, dataloader, criterion, optimizer, device):model.train()total_loss = 0for inputs, targets in dataloader:inputs, targets = inputs.to(device), targets.to(device)optimizer.zero_grad()outputs = model(inputs) # [batch_size, seq_len, vocab_size]loss = criterion(outputs.view(-1, vocab_size), targets.view(-1))loss.backward()optimizer.step()total_loss += loss.item() * inputs.size(0)return total_loss / len(dataloader.dataset)
4.3 训练过程
我们设置 5 个 epoch,并在每个 epoch 后输出训练损失,帮助监控训练进程。
num_epochs = 5
for epoch in range(num_epochs):loss = train_epoch(model, dataloader, criterion, optimizer, device)print(f"Epoch {epoch+1}/{num_epochs}, Loss: {loss:.4f}")
五、文本生成与评估
文本生成是模型应用的核心功能。本部分介绍如何从一个起始提示开始,逐步生成后续字符,并将生成序列转换为文本字符串。
5.1 文本生成函数设计
生成函数采用逐步生成策略,即将生成的字符追加到输入序列中,再利用更新后的序列继续预测下一个字符。
def generate_text(model, start_text, vocab, inv_vocab, max_len=30):"""根据起始文本生成后续文本:param start_text: 起始文本字符串:param vocab: 字符到索引映射:param inv_vocab: 索引到字符映射:param max_len: 生成的额外字符数量:return: 生成的完整文本字符串"""model.eval()# 将起始文本转换为索引序列input_ids = [vocab.get(char, vocab["<unk>"]) for char in chinese_tokenizer(start_text)]input_tensor = torch.tensor(input_ids, dtype=torch.long).unsqueeze(0).to(device)# 逐步生成字符for _ in range(max_len):with torch.no_grad():logits = model(input_tensor) # [1, seq_len, vocab_size]# 获取最后一个位置的 logits,并选取概率最大的字符next_logits = logits[0, -1, :]next_id = torch.argmax(next_logits).item()input_ids.append(next_id)input_tensor = torch.tensor(input_ids, dtype=torch.long).unsqueeze(0).to(device)generated_text = "".join([inv_vocab.get(idx, "<unk>") for idx in input_ids])return generated_text
5.2 生成示例
利用起始提示“今天天气”,生成一段文本以验证模型效果。
start_prompt = "今天天气"
generated_text = generate_text(model, start_prompt, vocab, inv_vocab, max_len=30)
print("生成的文本:", generated_text)
六、扩展与优化方向
在实际开发中,本文示例仅为入门级演示,其目的是帮助初学者快速上手文本生成模型的基本流程。然而,生产环境中的项目往往需要更高的性能、更强的鲁棒性以及更精细的控制。下面我们详细介绍一些扩展与优化方向,从数据、模型结构、生成策略到部署应用,均涉及专业工具和技术细节,并对专业术语进行解释。
6.1 数据扩展与预处理
数据是模型质量的基石,丰富而高质量的语料可以显著提升生成效果。
-
数据规模扩充
- 描述:在演示中我们使用了少量示例数据,但实际应用应采用大规模中文语料库,例如新闻数据、小说、社交媒体文本等。
- 工具推荐:可以使用开源数据集(如中文维基百科、搜狗新闻数据等),或利用网络爬虫工具(如 Scrapy)收集数据。
- 优势:更大数据量能提高模型泛化能力,减少生成文本的重复和逻辑错误。
-
预处理改进
- 描述:简单的逐字分词有时难以捕捉复杂语义。
- 工具推荐:
- 中文分词工具:jieba、THULAC、HanLP 等能够进行基于词的分词;
- 预训练分词器:BERT 分词器能够利用子词单元(subword units)捕捉更多语义细节。
- 优势:精细的预处理可以保留上下文信息,避免因分词粗糙而导致的信息丢失,从而提升生成文本的连贯性和语义准确性。
6.2 模型结构扩展
改进模型结构是提升文本生成质量的重要手段。
-
增加模型深度与宽度
- 描述:增加 Transformer 层数(深度)和嵌入维度(宽度)可以使模型具备更强的表达能力。
- 技术细节:
- 深度:例如将层数从 2 增加到 6 或 12;
- 宽度:嵌入维度从 128 提升到 256 或更高。
- 注意:更大的模型通常需要更多的训练数据和算力支持,同时也可能引入过拟合风险,需要配合正则化技术(如 Dropout)。
-
架构创新
- 描述:采用更先进的模型架构能进一步提升生成质量。
- 推荐架构:
- Transformer-XL:引入循环机制以处理长依赖;
- GPT 系列:自回归生成模型,已在多项任务中表现优异;
- T5:将生成和理解任务统一框架,适用于多任务场景。
- 优势:这些架构在捕捉长距离依赖和上下文信息方面更具优势,适合复杂的生成任务。
-
参数高效化
- 描述:在保持模型性能的前提下,降低参数量以减少计算资源占用。
- 技术方法:
- 知识蒸馏:将大模型的知识传递给较小模型;
- 量化:如 8-bit 量化技术,将浮点数权重转换为低精度表示;
- 剪枝:移除冗余参数。
- 优势:这些方法可以在不显著损失性能的情况下,降低模型部署成本,提高推理速度。
6.3 文本生成策略
生成策略直接决定生成文本的多样性与连贯性。
-
采样策略
- 贪心采样:每一步选择概率最高的 token,简单但容易陷入局部最优。
- 温度采样:通过温度参数调节概率分布,增加随机性,使生成结果更加多样。
- Top-k/Top-p(核采样):仅在概率前 k 个 token 或累计概率达到阈值的 token 中采样,平衡多样性和合理性。
- 技术细节:这些方法可以通过修改预测阶段的代码实现,比如对 logits 应用 softmax 后乘以温度因子,然后使用
torch.multinomial
进行采样。
-
控制生成
- 描述:在某些应用中,需要生成符合特定主题、情感或关键词的文本。
- 技术方法:
- 条件生成:在模型输入中添加控制信息,例如主题标签;
- 约束采样:在采样过程中对候选 token 进行过滤。
- 优势:这种方法能使生成结果更符合预期,并增强用户交互体验。
6.4 部署与应用
模型开发完成后,如何高效部署并提供在线服务也是关键环节。
-
模型压缩与加速
- 工具推荐:
- ONNX:将模型导出为中间表示,并利用 ONNX Runtime 加速推理;
- TensorRT:针对 NVIDIA GPU 进行优化,显著降低延迟。
- 优势:这些工具可在不改变模型结构的前提下,实现模型的加速与资源优化。
- 工具推荐:
-
服务化部署
- 描述:将训练好的模型封装为 API 服务,供前端应用调用。
- 工具推荐:
- FastAPI 或 Flask:轻量级 Web 框架,快速搭建 RESTful API;
- Docker:将应用容器化,便于跨平台部署。
- 优势:服务化部署实现实时文本生成,能满足在线用户请求,并且易于维护和扩展。
-
监控与反馈
- 描述:线上部署后需要对模型性能、生成质量和系统资源进行监控。
- 技术细节:
- 日志监控:记录生成结果、响应时间、错误日志等;
- 用户反馈:通过 A/B 测试或用户调研不断改进生成策略;
- 自动化告警:使用 Prometheus、Grafana 等监控工具对关键指标进行监控。
- 优势:及时发现问题,快速响应,保证系统稳定性和用户体验。
七、结论
本文从数据预处理、模型构建、训练与生成等方面详细介绍了如何从零开始开发一个中文文本生成模型。通过逐步讲解每个模块的设计思路和实现细节,读者可以全面理解 Transformer 模型在文本生成任务中的应用。尽管示例数据与模型规模较小,但这些技术在实际生产中均有重要应用。未来,随着数据规模扩大和模型架构的不断演进,文本生成技术将继续提升,为智能创作和对话系统提供更强支持。希望本文能为读者提供清晰的学习路径和实践指南,同时激发更多创新思考。
封面图:
一键复制代码
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
import math
import re# ----------------------------
# 1. 数据预处理与词汇表构建
# ----------------------------def chinese_tokenizer(text):"""对中文文本进行分词(逐字分词),去除空格"""text = text.replace(" ", "")return list(text)def build_vocab(texts):"""根据文本数据构建词汇表,保留所有字符特殊标记:"<pad>" 用于填充,"<unk>" 用于未知字符"""vocab = {"<pad>": 0, "<unk>": 1}idx = 2for text in texts:for char in chinese_tokenizer(text):if char not in vocab:vocab[char] = idxidx += 1return vocabdef encode_text(text, vocab):"""将中文文本转换为字符索引序列,未在词汇表中的字符映射为 <unk>"""tokens = chinese_tokenizer(text)return [vocab.get(token, vocab["<unk>"]) for token in tokens]class TextGenerationDataset(Dataset):def __init__(self, texts, vocab, seq_len=10):"""利用滑动窗口生成样本,每个样本长度为 seq_len+1输入为前 seq_len 个字符,目标为右移一位后的字符序列"""self.vocab = vocabself.seq_len = seq_lenself.data = []for text in texts:encoded = encode_text(text, vocab)if len(encoded) < seq_len + 1:continuefor i in range(len(encoded) - seq_len):self.data.append(encoded[i:i+seq_len+1])def __len__(self):return len(self.data)def __getitem__(self, idx):seq = self.data[idx]return torch.tensor(seq[:-1], dtype=torch.long), torch.tensor(seq[1:], dtype=torch.long)# 示例中文数据
texts = ["今天天气很好,我们一起出去玩吧!","我喜欢学习人工智能和深度学习。","自然语言处理是计算机科学的重要方向。","数据科学改变世界,未来无限可能。"
]vocab = build_vocab(texts)
print("词汇表大小:", len(vocab))# 构造数据集与 DataLoader(训练时输入序列长度设为 10)
seq_len = 10
dataset = TextGenerationDataset(texts, vocab, seq_len=seq_len)
dataloader = DataLoader(dataset, batch_size=4, shuffle=True)# 生成逆向词汇表,用于将索引转换为字符
inv_vocab = {idx: char for char, idx in vocab.items()}# ----------------------------
# 2. 定义位置编码模块
# ----------------------------
class PositionalEncoding(nn.Module):def __init__(self, d_model, max_len=50):"""初始化位置编码,max_len 设置为 50,确保生成时支持更长序列"""super(PositionalEncoding, self).__init__()pe = torch.zeros(max_len, d_model) # [max_len, d_model]position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) # [max_len, 1]div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))pe[:, 0::2] = torch.sin(position * div_term)pe[:, 1::2] = torch.cos(position * div_term)pe = pe.unsqueeze(0) # [1, max_len, d_model]self.register_buffer('pe', pe)def forward(self, x):"""输入 x: [batch_size, seq_len, d_model]将位置编码加到输入上,要求 x.size(1) <= max_len"""x = x + self.pe[:, :x.size(1)]return x# ----------------------------
# 3. 定义 Transformer 文本生成模型
# ----------------------------
class TransformerLM(nn.Module):def __init__(self, vocab_size, embed_dim, num_heads, num_layers, dropout=0.1, max_len=50):""":param vocab_size: 词汇表大小:param embed_dim: 嵌入向量维度:param num_heads: 注意力头数:param num_layers: Transformer 编码器层数:param dropout: Dropout 概率:param max_len: 位置编码最大长度(生成时支持的最大序列长度)"""super(TransformerLM, self).__init__()self.embed = nn.Embedding(vocab_size, embed_dim)self.pos_encoder = PositionalEncoding(embed_dim, max_len=max_len)encoder_layer = nn.TransformerEncoderLayer(d_model=embed_dim, nhead=num_heads, dropout=dropout)self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)self.fc = nn.Linear(embed_dim, vocab_size)self.embed_dim = embed_dimdef forward(self, x):""":param x: 输入序列 [batch_size, seq_len]:return: logits [batch_size, seq_len, vocab_size]"""x = self.embed(x) * math.sqrt(self.embed_dim)x = self.pos_encoder(x)# TransformerEncoder 要求输入形状为 [seq_len, batch_size, embed_dim]x = x.transpose(0, 1)x = self.transformer(x)x = x.transpose(0, 1) # 恢复为 [batch_size, seq_len, embed_dim]logits = self.fc(x)return logitsvocab_size = len(vocab)
embed_dim = 128
num_heads = 4
num_layers = 2# 注意:训练时输入序列长度为 10,但生成时模型可支持最长 50(由位置编码决定)
model = TransformerLM(vocab_size, embed_dim, num_heads, num_layers, dropout=0.1, max_len=50)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)# ----------------------------
# 4. 定义损失函数与优化器
# ----------------------------
criterion = nn.CrossEntropyLoss(ignore_index=vocab["<pad>"])
optimizer = optim.Adam(model.parameters(), lr=0.001)# ----------------------------
# 5. 训练模型
# ----------------------------
def train_epoch(model, dataloader, criterion, optimizer, device):model.train()total_loss = 0for inputs, targets in dataloader:inputs, targets = inputs.to(device), targets.to(device)optimizer.zero_grad()outputs = model(inputs) # [batch_size, seq_len, vocab_size]loss = criterion(outputs.view(-1, vocab_size), targets.view(-1))loss.backward()optimizer.step()total_loss += loss.item() * inputs.size(0)return total_loss / len(dataloader.dataset)num_epochs = 5
for epoch in range(num_epochs):loss = train_epoch(model, dataloader, criterion, optimizer, device)print(f"Epoch {epoch+1}/{num_epochs}, Loss: {loss:.4f}")# ----------------------------
# 6. 定义文本生成函数
# ----------------------------
def generate_text(model, start_text, vocab, inv_vocab, max_len=30):"""根据起始文本生成后续文本:param start_text: 起始文本字符串:param vocab: 字符到索引映射:param inv_vocab: 索引到字符映射:param max_len: 生成的额外字符数量:return: 生成的完整文本字符串"""model.eval()# 将起始文本转换为索引序列input_ids = [vocab.get(char, vocab["<unk>"]) for char in chinese_tokenizer(start_text)]input_tensor = torch.tensor(input_ids, dtype=torch.long).unsqueeze(0).to(device)# 逐步生成字符for _ in range(max_len):with torch.no_grad():logits = model(input_tensor) # [1, seq_len, vocab_size]# 获取最后一个位置的 logits,并选取概率最大的字符next_logits = logits[0, -1, :]next_id = torch.argmax(next_logits).item()input_ids.append(next_id)input_tensor = torch.tensor(input_ids, dtype=torch.long).unsqueeze(0).to(device)generated_text = "".join([inv_vocab.get(idx, "<unk>") for idx in input_ids])return generated_text# ----------------------------
# 7. 测试文本生成
# ----------------------------
start_prompt = "今天天气"
generated_text = generate_text(model, start_prompt, vocab, inv_vocab, max_len=30)
print("生成的文本:", generated_text)