使用循环神经网络学习汉语拼音拼写：数据准备和模型构建

本项目使用循环神经网络 (RNN) 学习汉语拼音拼写，重点介绍数据准备和模型构建过程。

数据来源:

拼音数据（无声调）：https://www.jianguoyun.com/p/DQ3els0Q-rqYBhi3pIgFIAA

数据准备:

定义数据集: 采用字符模型，因此一个字符为一个样本。每个样本采用 one-hot 编码。
样本时间相关性: 分别实现序列的随机采样和序列的顺序划分。
标签: 标签 Y 与 X 同形状，但时间超前 1。
数据形状: 一次梯度更新使用的数据形状为：（时间步，Batch，类别数）。

模型构建:

基本循环神经网络模型:
- 循环单元为 nn.RNN 或 GRU
- 输出层的全连接使用 RNN 所有时间步的输出
- 隐状态初始值为 0
测试前向传播:
- 如果采用顺序划分，需进行梯度截断。

训练:

损失函数为平均交叉熵。

预测:

给定一个前缀，进行单步预测和 K 步预测。

代码实现:

import torch
from torch import nn

# 定义模型
class RNNModel(nn.Module):
    def __init__(self, input_size, hidden_size, output_size, num_layers, rnn_type='rnn'):
        super(RNNModel, self).__init__()
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.rnn_type = rnn_type
        
        self.encoder = nn.Embedding(input_size, hidden_size)
        if rnn_type == 'rnn':
            self.rnn = nn.RNN(hidden_size, hidden_size, num_layers, batch_first=True)
        elif rnn_type == 'gru':
            self.rnn = nn.GRU(hidden_size, hidden_size, num_layers, batch_first=True)
        self.decoder = nn.Linear(hidden_size, output_size)
        
    def forward(self, input, hidden):
        input = self.encoder(input)
        output, hidden = self.rnn(input, hidden)
        output = self.decoder(output)
        return output, hidden
    
    def init_hidden(self, batch_size):
        if self.rnn_type == 'gru':
            return torch.zeros(self.num_layers, batch_size, self.hidden_size)
        else:
            return torch.zeros(self.num_layers, batch_size, self.hidden_size)

# 定义数据集类
class PinyinDataset(torch.utils.data.Dataset):
    def __init__(self, data_path):
        self.data = []
        with open(data_path, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if line:
                    self.data.append(line)
                    
    def __getitem__(self, index):
        sample = self.data[index]
        x = torch.zeros(len(sample), input_size)
        y = torch.zeros(len(sample), output_size)
        for i, c in enumerate(sample):
            x[i][char_to_index[c]] = 1
            if i > 0:
                y[i-1][char_to_index[c]] = 1
        return x, y
    
    def __len__(self):
        return len(self.data)

# 准备数据
input_size = output_size = 35  # 总共有35个拼音字符
hidden_size = 128
num_layers = 2
batch_size = 128
num_epochs = 10
learning_rate = 0.01
clip = 5  # 梯度截断

# 构建字符到索引的映射
chars = 'abcdefghijklmnopqrstuvwxyz0123456789'
char_to_index = {c: i for i, c in enumerate(chars)}

# 加载数据集
dataset = PinyinDataset('pinyin.txt')
dataloader = torch.utils.data.DataLoader(dataset, batch_size=batch_size, shuffle=True)

# 创建模型
model = RNNModel(input_size, hidden_size, output_size, num_layers, rnn_type='rnn')

# 定义损失函数和优化器
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

# 训练模型
for epoch in range(num_epochs):
    total_loss = 0
    for i, (inputs, labels) in enumerate(dataloader):
        batch_size = inputs.shape[0]
        hidden = model.init_hidden(batch_size)
        inputs = inputs.permute(1, 0, 2)
        labels = labels.permute(1, 0, 2)
        
        # 前向传播和计算损失
        outputs, hidden = model(inputs, hidden)
        outputs = outputs.view(-1, output_size)
        labels = labels.view(-1).long()
        loss = criterion(outputs, labels)
        
        # 反向传播和更新参数
        optimizer.zero_grad()
        loss.backward()
        nn.utils.clip_grad_norm_(model.parameters(), clip)
        optimizer.step()
        
        total_loss += loss.item()
        
    print('Epoch [{}/{}], Loss: {:.4f}'.format(epoch+1, num_epochs, total_loss/len(dataset)))
    
# 测试模型
prefix = 'ni'  # 给定一个前缀
num_steps = 5  # 进行5步预测
hidden = model.init_hidden(1)
input = torch.zeros(1, 1, input_size)
input[0][0][char_to_index[prefix[0]]] = 1  # 设置第一个字符的one-hot编码
output = prefix
for i in range(num_steps):
    output_prob, hidden = model(input, hidden)
    output_prob = output_prob.view(-1, output_size)
    output_prob = nn.functional.softmax(output_prob, dim=1)
    topk_prob, topk_index = torch.topk(output_prob, k=5)  # 取前5个概率最大的字符
    topk_index = topk_index.squeeze().tolist()
    next_char = chars[topk_index[0]]
    output += next_char
    input = torch.zeros(1, 1, input_size)
    input[0][0][char_to_index[next_char]] = 1  # 设置下一个字符的one-hot编码
print(output)