按照阿光的项目做出了学习笔记，pytorch深度学习实战项目100例

基于LSTM实现春联上联对下联

基于LSTM（长短期记忆网络）实现春联上联对下联是一种有趣且具有挑战性的任务，它涉及到自然语言处理（NLP）中的序列到序列（seq2seq）模型。LSTM是处理序列数据的理想选择，因为它能够记住长期的依赖信息，这对于生成符合语境和文化习俗的春联下联至关重要。

数据

https://github.com/wb14123/couplet-dataset

感谢大佬的分享的对联数据集

对数据集的处理

def data_generator(data):
    # 计算每个对联长度的权重
    data_probability = [float(len(x)) for wordcount, [x, y] in data.items()]  # [每个字数key对应对联list中上联数据的个数]
    data_probability = np.array(data_probability) / sum(data_probability)  # 标准化至[0,1]，这是每个字数的权重

    # 随机选择字数，然后随机选择字数对应的上联样本，生成batch
    for idx in range(15):
        # 随机选字数id，概率为上面计算的字数权重
        idx = idx + 1

        size = min(batch_size, len(data[idx][0]))  # batch_size=64，len(data[idx][0])随机选择的字数key对应的上联个数

        # 从上联列表下标list中随机选出大小为size的list
        idxs = np.random.choice(len(data[idx][0]), size=size)

        # 返回选出的上联X与下联y, 将原本1-d array维度扩展为(row,col,1)
        yield data[idx][0][idxs], np.expand_dims(data[idx][1][idxs], axis=2)

# 加载文本数据
def load_data(input_path, output_path):
    # 数据读取与切分
    def read_data(file_path):
        txt = codecs.open(file_path, encoding='utf-8').readlines()
        txt = [line.strip().split(' ') for line in txt]  # 每行按空格切分
        txt = [line for line in txt if len(line) < 16]  # 过滤掉字数超过maxlen的对联
        return txt

    # 产生数据字典
    def generate_count_dict(result_dict, x, y):
        for i, idx in enumerate(x):
            j = len(idx)
            if j not in result_dict:
                result_dict[j] = [[], []]  # [样本数据list,类别标记list]
            result_dict[j][0].append(idx)
            result_dict[j][1].append(y[i])
        return result_dict

    # 将字典数据转为numpy
    def to_numpy_array(dict):
        for count, [x, y] in dict.items():
            dict[count][0] = np.array(x)
            dict[count][1] = np.array(y)

        return dict

    x = read_data(input_path)
    y = read_data(output_path)

    # 获取词表
    vocabulary = x + y

    # 构造字符级别的特征
    string = ''
    for words in vocabulary:
        for word in words:
            string += word

    # 所有的词汇表
    vocabulary = set(string)

    word2idx = {word: i for i, word in enumerate(vocabulary)}
    idx2word = {i: word for i, word in enumerate(vocabulary)}

    # 训练数据中所有词的个数
    vocab_size = len(word2idx.keys())  # 词汇表大小

    # 将x和y转为数值
    x = [[word2idx[word] for word in sent] for sent in x]
    y = [[word2idx[word] for word in sent] for sent in y]

    train_dict = {}

    train_dict = generate_count_dict(train_dict, x, y)

    train_dict = to_numpy_array(train_dict)

    return train_dict, vocab_size, idx2word, word2idx

基本想法：
这种场景是典型的 Encoder-Decoder 框架应用问题。

在这个框架中：

• Encoder 负责读取输入序列（上联）并将其转换成一个固定长度的内部表示形式，通常是最后一个时间步的隐藏状态。这个内部表示被视为输入序列的“上下文”或“意义”，包含了生成输出序列所需的所有信息。
• Decoder 接收这个内部表示并开始生成输出序列（下联），一步一步地生成，直到产生序列结束标记或达到特定长度。

构建模型

模型架构：使用seq2seq模型，该模型一般包括一个编码器（encoder）和一个解码器（decoder），两者都可以是LSTM网络。编码器负责处理上联，而解码器则生成下联。
嵌入层：通常在模型的第一层使用嵌入层，将每个字符或词转换为固定大小的向量，这有助于模型更好地理解语言中的语义信息。

# 定义网络结构
class LSTM(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, num_layers):
        super(LSTM, self).__init__()
        self.hidden_dim = hidden_dim
        self.embeddings = nn.Embedding(vocab_size + 1, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, num_layers)
        self.linear = nn.Linear(hidden_dim, vocab_size)

    def forward(self, x):
        time_step, batch_size = x.size()  # 124, 16
        embeds = self.embeddings(x)
        output, (h_n, c_n) = self.lstm(embeds)
        output = self.linear(output.reshape(time_step * batch_size, -1))
        # 要返回所有时间点的数据，每个时间点对应一个字，也就是vocab_size维度的向量
        return output

训练模型

# 加载数据
train_dict, vocab_size, idx2word, word2idx = load_data(input_path, output_path)

# 模型训练
model = LSTM(vocab_size=vocab_size, hidden_dim=hidden_dim,
             embedding_dim=embedding_dim, num_layers=num_layers)

Configimizer = optim.Adam(model.parameters(), lr=lr)  # 优化器
criterion = nn.CrossEntropyLoss()  # 多分类损失函数

model.to(device)
loss_meter = meter.AverageValueMeter()

best_loss = 999  # 保存loss
best_model = None  # 保存对应最好准确率的模型参数

for epoch in range(epochs):
    model.train()  # 开启训练模式
    loss_meter.reset()

    for x, y in data_generator(train_dict):
        x = torch.from_numpy(x).long().transpose(1, 0).contiguous()
        x = x.to(device)

        y = torch.from_numpy(y).long().transpose(1, 0).contiguous()
        y = y.to(device)

        Configimizer.zero_grad()

        # 形成预测结果
        output_ = model(x)

        # 计算损失
        loss = criterion(output_, y.long().view(-1))
        loss.backward()
        Configimizer.step()

        loss_meter.add(loss.item())

    # 打印信息
    print("【EPOCH: 】%s" % str(epoch + 1))
    print("训练损失为%s" % (str(loss_meter.mean)))

    # 保存模型及相关信息
    if loss_meter.mean < best_loss:
        best_loss = loss_meter.mean
        best_model = model.state_dict()

    # 在训练结束保存最优的模型参数
    if epoch == epochs - 1:
        # 保存模型
        torch.save(best_model, './best_model.pkl')

测试

import codecs

import numpy as np
import torch
from torch import nn
from torch import optim
from torchnet import meter

# 模型输入参数，需要自己根据需要调整
input_path = 'C:\\Users\\kaai\\AppData\\Local\\Temp\\BNZ.65e95f542f0fca6f\\train\\in.txt'
output_path = 'C:\\Users\\kaai\\AppData\\Local\\Temp\\BNZ.65e95f542f0fca6f\\train\\out.txt'
num_layers = 1  # LSTM层数
hidden_dim = 100  # LSTM中的隐层大小
epochs = 50  # 迭代次数
batch_size = 128  # 每个批次样本大小
embedding_dim = 15  # 每个字形成的嵌入向量大小
lr = 0.01  # 学习率
device = 'cpu'  # 设备

# 用于生成训练数据
def data_generator(data):
    # 计算每个对联长度的权重
    data_probability = [float(len(x)) for wordcount, [x, y] in data.items()]  # [每个字数key对应对联list中上联数据的个数]
    data_probability = np.array(data_probability) / sum(data_probability)  # 标准化至[0,1]，这是每个字数的权重

    # 随机选择字数，然后随机选择字数对应的上联样本，生成batch
    for idx in range(15):
        # 随机选字数id，概率为上面计算的字数权重
        idx = idx + 1

        size = min(batch_size, len(data[idx][0]))  # batch_size=64，len(data[idx][0])随机选择的字数key对应的上联个数

        # 从上联列表下标list中随机选出大小为size的list
        idxs = np.random.choice(len(data[idx][0]), size=size)

        # 返回选出的上联X与下联y, 将原本1-d array维度扩展为(row,col,1)
        yield data[idx][0][idxs], np.expand_dims(data[idx][1][idxs], axis=2)

# 加载文本数据
def load_data(input_path, output_path):
    # 数据读取与切分
    def read_data(file_path):
        txt = codecs.open(file_path, encoding='utf-8').readlines()
        txt = [line.strip().split(' ') for line in txt]  # 每行按空格切分
        txt = [line for line in txt if len(line) < 16]  # 过滤掉字数超过maxlen的对联
        return txt

    # 产生数据字典
    def generate_count_dict(result_dict, x, y):
        for i, idx in enumerate(x):
            j = len(idx)
            if j not in result_dict:
                result_dict[j] = [[], []]  # [样本数据list,类别标记list]
            result_dict[j][0].append(idx)
            result_dict[j][1].append(y[i])
        return result_dict

    # 将字典数据转为numpy
    def to_numpy_array(dict):
        for count, [x, y] in dict.items():
            dict[count][0] = np.array(x)
            dict[count][1] = np.array(y)

        return dict

    x = read_data(input_path)
    y = read_data(output_path)

    # 获取词表
    vocabulary = x + y

    # 构造字符级别的特征
    string = ''
    for words in vocabulary:
        for word in words:
            string += word

    # 所有的词汇表
    vocabulary = set(string)

    word2idx = {word: i for i, word in enumerate(vocabulary)}
    idx2word = {i: word for i, word in enumerate(vocabulary)}

    # 训练数据中所有词的个数
    vocab_size = len(word2idx.keys())  # 词汇表大小

    # 将x和y转为数值
    x = [[word2idx[word] for word in sent] for sent in x]
    y = [[word2idx[word] for word in sent] for sent in y]

    train_dict = {}

    train_dict = generate_count_dict(train_dict, x, y)

    train_dict = to_numpy_array(train_dict)

    return train_dict, vocab_size, idx2word, word2idx

# 定义网络结构
class LSTM(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, num_layers):
        super(LSTM, self).__init__()
        self.hidden_dim = hidden_dim
        self.embeddings = nn.Embedding(vocab_size + 1, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, num_layers)
        self.linear = nn.Linear(hidden_dim, vocab_size)

    def forward(self, x):
        time_step, batch_size = x.size()  # 124, 16
        embeds = self.embeddings(x)
        output, (h_n, c_n) = self.lstm(embeds)
        output = self.linear(output.reshape(time_step * batch_size, -1))
        # 要返回所有时间点的数据，每个时间点对应一个字，也就是vocab_size维度的向量
        return output


def couplet_match(s):
    # 将字符串转为数值
    x = [word2idx[word] for word in s]

    # 将数值向量转为tensor
    x = torch.from_numpy(np.array(x).reshape(-1, 1))

    # 加载模型
    model_path = './best_model.pkl'
    model = LSTM(vocab_size=vocab_size, hidden_dim=hidden_dim,
                 embedding_dim=embedding_dim, num_layers=num_layers)

    model.load_state_dict(torch.load(model_path, 'cpu'))

    y = model(x)
    y = y.argmax(axis=1)
    r = ''.join([idx2word[idx.item()] for idx in y])

    print('上联：%s，下联：%s' % (s, r))
# 加载数据
train_dict, vocab_size, idx2word, word2idx = load_data(input_path, output_path)
# 测试
sentence = '恭喜发财'
couplet_match(sentence)