自然语言处理（三）：RNN与序列建模

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

import torch
import torch.nn as nn
import numpy as np

text = """Deep learning is a subset of machine learning that uses neural
networks with many layers. These networks can learn hierarchical
representations of data, making them powerful for tasks like image
recognition, natural language processing, and speech recognition."""

chars = sorted(set(text))
char_to_idx = {ch: i for i, ch in enumerate(chars)}
idx_to_char = {i: ch for i, ch in enumerate(chars)}
vocab_size = len(chars)
print(f"词表大小: {vocab_size} 个字符")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class CharRNN(nn.Module):
    def __init__(self, vocab_size, hidden_size, num_layers):
        super().__init__()
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.embedding = nn.Embedding(vocab_size, hidden_size)
        self.lstm = nn.LSTM(hidden_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, vocab_size)

    def forward(self, x, hidden):
        embedded = self.embedding(x)
        out, hidden = self.lstm(embedded, hidden)
        out = self.fc(out.reshape(-1, self.hidden_size))
        return out, hidden

    def init_hidden(self, batch_size, device):
        h0 = torch.zeros(self.num_layers, batch_size, self.hidden_size).to(device)
        c0 = torch.zeros(self.num_layers, batch_size, self.hidden_size).to(device)
        return (h0, c0)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

def train(model, text_data, epochs=100, seq_length=50, batch_size=16, lr=0.002):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = model.to(device)
    criterion = nn.CrossEntropyLoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=lr)
    data = [char_to_idx[ch] for ch in text_data]

    for epoch in range(epochs):
        model.train()
        hidden = model.init_hidden(batch_size, device)
        total_loss = 0
        n_batches = len(data) // (seq_length * batch_size)

        for _ in range(n_batches):
            starts = np.random.randint(0, len(data) - seq_length - 1, batch_size)
            inputs = torch.LongTensor([data[s:s+seq_length] for s in starts]).to(device)
            targets = torch.LongTensor([data[s+1:s+seq_length+1] for s in starts]).to(device)

            hidden = tuple(h.detach() for h in hidden)        # 截断 BPTT
            outputs, hidden = model(inputs, hidden)
            loss = criterion(outputs, targets.reshape(-1))

            optimizer.zero_grad()
            loss.backward()
            torch.nn.utils.clip_grad_norm_(model.parameters(), 5.0)   # 防梯度爆炸
            optimizer.step()
            total_loss += loss.item()

        if (epoch + 1) % 20 == 0:
            print(f"Epoch {epoch+1}/{epochs}, Loss: {total_loss/n_batches:.4f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

def generate(model, start_str, length=200, temperature=0.8):
    device = next(model.parameters()).device
    model.eval()
    hidden = model.init_hidden(1, device)
    input_seq = [char_to_idx[ch] for ch in start_str]
    generated = start_str

    with torch.no_grad():
        for idx in input_seq[:-1]:                     # 先把隐藏状态"喂热"
            x = torch.LongTensor([[idx]]).to(device)
            _, hidden = model(x, hidden)

        x = torch.LongTensor([[input_seq[-1]]]).to(device)
        for _ in range(length):
            output, hidden = model(x, hidden)
            probs = torch.softmax(output.squeeze() / temperature, dim=0).cpu().numpy()
            char_idx = np.random.choice(len(probs), p=probs)
            generated += idx_to_char[char_idx]
            x = torch.LongTensor([[char_idx]]).to(device)

    return generated

model = CharRNN(vocab_size, hidden_size=128, num_layers=2)
train(model, text, epochs=100)
print(generate(model, "Deep learning", length=200))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

import torch
import torch.nn as nn
import random

pairs = [
    ("hello", "bonjour"), ("good morning", "bon matin"),
    ("thank you", "merci"), ("goodbye", "au revoir"),
    ("how are you", "comment allez vous"),
    ("i love you", "je t aime"), ("welcome", "bienvenue"),
]

SOS, EOS = 0, 1

class Vocab:
    def __init__(self):
        self.word2idx = {"<SOS>": SOS, "<EOS>": EOS}
        self.idx2word = {SOS: "<SOS>", EOS: "<EOS>"}
        self.n_words = 2

    def add_sentence(self, sentence):
        for word in sentence.split():
            if word not in self.word2idx:
                self.word2idx[word] = self.n_words
                self.idx2word[self.n_words] = word
                self.n_words += 1

src_vocab, tgt_vocab = Vocab(), Vocab()
for en, fr in pairs:
    src_vocab.add_sentence(en)
    tgt_vocab.add_sentence(fr)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

class Encoder(nn.Module):
    def __init__(self, vocab_size, hidden_size):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, hidden_size)
        self.lstm = nn.LSTM(hidden_size, hidden_size, batch_first=True)

    def forward(self, x):
        return self.lstm(self.embedding(x))

class Decoder(nn.Module):
    def __init__(self, hidden_size, vocab_size):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, hidden_size)
        self.lstm = nn.LSTM(hidden_size, hidden_size, batch_first=True)
        self.fc = nn.Linear(hidden_size, vocab_size)

    def forward(self, x, hidden):
        out, hidden = self.lstm(self.embedding(x), hidden)
        return self.fc(out.squeeze(1)), hidden

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

def train_seq2seq(encoder, decoder, pairs, epochs=500, lr=0.01):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    encoder, decoder = encoder.to(device), decoder.to(device)
    enc_opt = torch.optim.Adam(encoder.parameters(), lr=lr)
    dec_opt = torch.optim.Adam(decoder.parameters(), lr=lr)
    criterion = nn.CrossEntropyLoss()

    for epoch in range(epochs):
        total_loss = 0
        random.shuffle(pairs)
        for src_sent, tgt_sent in pairs:
            src_ids = [src_vocab.word2idx[w] for w in src_sent.split()] + [EOS]
            tgt_ids = [tgt_vocab.word2idx[w] for w in tgt_sent.split()] + [EOS]

            src_t = torch.LongTensor([src_ids]).to(device)
            tgt_t = torch.LongTensor(tgt_ids).to(device)

            enc_opt.zero_grad(); dec_opt.zero_grad()

            _, hidden = encoder(src_t)                     # 上下文向量
            dec_input = torch.LongTensor([[SOS]]).to(device)
            loss = 0

            for i in range(len(tgt_ids)):
                output, hidden = decoder(dec_input, hidden)
                loss += criterion(output, tgt_t[i:i+1])
                dec_input = tgt_t[i:i+1].unsqueeze(0)      # teacher forcing
            loss.backward()
            enc_opt.step(); dec_opt.step()
            total_loss += loss.item() / len(tgt_ids)

        if (epoch + 1) % 100 == 0:
            print(f"Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(pairs):.4f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

def translate(encoder, decoder, sentence):
    device = next(encoder.parameters()).device
    encoder.eval(); decoder.eval()
    with torch.no_grad():
        src_ids = [src_vocab.word2idx.get(w, 0) for w in sentence.split()] + [EOS]
        _, hidden = encoder(torch.LongTensor([src_ids]).to(device))
        dec_input = torch.LongTensor([[SOS]]).to(device)
        words = []
        for _ in range(20):
            output, hidden = decoder(dec_input, hidden)
            token = output.argmax(dim=-1).item()
            if token == EOS:
                break
            words.append(tgt_vocab.idx2word[token])
            dec_input = torch.LongTensor([[token]]).to(device)
    return ' '.join(words)

hidden_size = 256
encoder = Encoder(src_vocab.n_words, hidden_size)
decoder = Decoder(hidden_size, tgt_vocab.n_words)
train_seq2seq(encoder, decoder, pairs)
for s in ["hello", "thank you", "good morning"]:
    print(f"{s} -> {translate(encoder, decoder, s)}")