基于 Transfomer 的预训练模型 | GPT

GPT（Generative Pre-trained Transformer）是一个由 OpenAI 开发的先进的自然语言处理（NLP）模型，专门用于处理各种语言任务。它基于 Transformer 架构，一种在 NLP 领域非常有效的深度学习模型结构。GPT 模型在多种语言任务上表现出色，包括但不限于：文本生成、问答系统、机器翻译、文本摘要、感情分析等。

要注意的是，不同于 BERT，GPT 是一个自回归模型，使用了监督学习的方式，进行从左到右的语言建模，因此 GPT 在生成时只能查看之前的单词（单向上下文）。

基本结构

由于 GPT 任务的特殊性，其只保留了 Transformer 的解码器（Decoder），舍弃了编码器（Encoder）。所以如果能够理解 Transformer，那么理解 GPT 模型自然也就不难了。

GPT Decoder Layer 结构，删除了编码器-解码器注意力模块

Decoder 层面的自注意力机制允许模型在生成每个新词时看到之前的所有词，这对于生成连贯的文本至关重要。而 Encoder 部分是为了编码输入序列到一个固定长度的连续表征中，这在 GPT 的预训练目标中并不是必要的。此外，省略 Encoder 可以简化模型的结构，并专注于提高文本生成能力。

两种文本生成策略

  1. 贪心解码：
    • 在每一步，模型选择概率最高的单词作为下一个单词。
    • 这种方法速度快，计算成本低，但可能不是最佳选择，因为它不考虑整体句子的最佳组合，可能导致局部最优解。
  2. 集束搜索解码：
    • 在每一步保持多个可能的候选序列（称为 “集束” 或 “beam”）。
    • 集束的宽度（Beam width）决定了在每一步有多少候选序列被考虑。
    • 集束搜索会在每个时间步骤考虑多个可能的继承候选，并在序列结束时选择整体得分最高的序列。
    • 这种方法更能找到质量高的序列，但计算成本更高，速度也较慢。

代码实现

GPT Decoder
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
34
35
36
37
38
39
# ...
class DecoderLayer(nn.Module):
    def __init__(self):
        super(DecoderLayer, self).__init__()
        self.self_attn = MultiHeadAttention()  # 多头自注意力层
        self.feed_forward = FFN()  # 位置前馈神经网络层
        self.norm1 = nn.LayerNorm(d_embedding)  # 第一个层归一化
        self.norm2 = nn.LayerNorm(d_embedding)  # 第二个层归一化

    def forward(self, dec_inputs, attn_mask=None):
        # 使用多头自注意力处理输入
        attn_output, _ = self.self_attn(dec_inputs, dec_inputs, dec_inputs, attn_mask)
        # 将注意力输出与输入相加并进行第一个层归一化
        norm1_outputs = self.norm1(dec_inputs + attn_output)
        # 将归一化后的输出输入到位置前馈神经网络
        ff_outputs = self.feed_forward(norm1_outputs)
        # 将前馈神经网络输出与第一次归一化后的输出相加并进行第二个层归一化
        dec_outputs = self.norm2(norm1_outputs + ff_outputs)
        return dec_outputs

n_layers = 6  # 设置 Decoder 的层数
class Decoder(nn.Module):
    def __init__(self, corpus):
        super(Decoder, self).__init__()
        self.src_emb = nn.Embedding(corpus.vocab_size, d_embedding)  # 词嵌入层（参数为词典维度）
        self.pos_emb = nn.Embedding(corpus.seq_len, d_embedding)  # 位置编码层（参数为序列长度）        
        self.layers = nn.ModuleList([DecoderLayer() for _ in range(n_layers)])

    def forward(self, dec_inputs):        
        positions = torch.arange(len(dec_inputs), device=dec_inputs.device).unsqueeze(-1) # 位置信息        
        inputs_embedding = self.src_emb(dec_inputs) + self.pos_emb(positions) # 词嵌入与位置编码相加        
        attn_mask = get_attn_subsequent_mask(inputs_embedding).to(dec_inputs.device) # 生成自注意力掩码
        dec_outputs =  inputs_embedding # 初始化解码器输入，这是第一层解码器层的输入      
        for layer in self.layers:
            # 每个解码器层接收前一层的输出作为输入，并生成新的输出
            # 对于第一层解码器层，其输入是 dec_outputs，即词嵌入和位置编码的和
            # 对于后续的解码器层，其输入是前一层解码器层的输出            
            dec_outputs = layer(dec_outputs, attn_mask) # 将输入数据传递给解码器层
        return dec_outputs # 返回最后一个解码器层的输出，作为整个解码器的输出
文本生成
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
from nltk.tokenize import word_tokenize # 分词工具

# 贪心搜索
def generate_text_greedy_search(model, input_str, max_len=5):
    # 将模型设置为评估（测试）模式，关闭 dropout 和 batch normalization 等训练相关的层
    model.eval()
    # 使用 NLTK 工具进行词汇切分
    input_str = word_tokenize(input_str)
    # 将输入字符串中的每个 token 转换为其在词汇表中的索引, 如果输入的词不在词表里面，就忽略这个词
    input_tokens = [model.corpus.vocab[token] for token in input_str if token in model.corpus.vocab]
    # 检查输入的有意义的词汇长度是否为 0
    if len(input_tokens) == 0:
        return         
    # 创建一个列表，用于存储生成的词汇
    output_tokens = input_tokens
    # 禁用梯度计算，以节省内存并加速测试过程
    with torch.no_grad():
        # 生成最多 max_len 个 tokens
        for _ in range(max_len):
            # 将当前生成的 tokens 转换为 torch 张量并将其传递给模型
            device = "cuda" if torch.cuda.is_available() else "cpu"
            inputs = torch.LongTensor(output_tokens).unsqueeze(0).to(device)
            outputs = model(inputs)                
            # 只关心最后一个时间步（即最新生成的 token）的 logits
            logits = outputs[:, -1, :]
            # 找到具有最高分数的 token
            _, next_token = torch.topk(logits, 1, dim=-1)
            # 如果生成的 token 是 EOS（结束符），则停止生成
            if next_token.item() == model.corpus.vocab["<eos>"]:
                break
            # 否则，将生成的 token 添加到生成的词汇列表中
            output_tokens.append(next_token.item())
    # 将输出 tokens 转换回文本字符串
    output_str = " ".join([model.corpus.idx2word[token] for token in output_tokens])
    return output_str

# 集束搜索
def generate_text_beam_search(model, input_str, max_len=5, beam_width=5, repetition_penalty=1.2):
    # 将模型设置为评估（测试）模式，关闭 dropout 和 batch normalization 等训练相关的层
    model.eval()
    # 分词
    input_str = word_tokenize(input_str)
    # 将输入字符串中的每个 token 转换为其在词汇表中的索引, 如果输入的词不再词表里面，就忽略这个词
    input_tokens = [model.corpus.vocab[token] for token in input_str if token in model.corpus.vocab]
    # 检查输入的有意义的词汇长度是否为0
    if len(input_tokens) == 0:
        return         
    # 创建一个列表，用于存储候选序列，初始候选序列只包含输入 tokens
    candidates = [(input_tokens, 0.0)]    
    # 创建一个列表，用于存储所有生成的序列及其得分
    final_results = []
    # 禁用梯度计算，以节省内存并加速测试过程
    with torch.no_grad():
        # 生成最多max_len个tokens
        for _ in range(max_len):
            # 创建一个新的候选列表，用于存储当前时间步生成的候选序列
            new_candidates = []            
            # 遍历当前候选序列
            for candidate, candidate_score in candidates:
                # 将当前候选序列转换为 torch 张量并将其传递给模型
                device = "cuda" if torch.cuda.is_available() else "cpu"
                inputs = torch.LongTensor(candidate).unsqueeze(0).to(device)
                outputs = model(inputs)                
                # 只关心最后一个时间步（即最新生成的token）的logits
                logits = outputs[:, -1, :]
                # 应用重复惩罚：为已经生成的词汇应用惩罚，降低它们再次被选择的概率
                for token in set(candidate):
                    logits[0, token] /= repetition_penalty
                # 将 <pad> 标记的得分设置为一个很大的负数，以避免选择它
                logits[0, model.corpus.vocab["<pad>"]] = -1e9                
                # 找到具有最高分数的前 beam_width 个 tokens
                scores, next_tokens = torch.topk(logits, beam_width, dim=-1)
                # 遍历生成的 tokens 及其得分
                for score, next_token in zip(scores.squeeze(), next_tokens.squeeze()):
                    # 将生成的 token 添加到当前候选序列
                    new_candidate = candidate + [next_token.item()]                    
                    # 更新候选序列得分
                    new_score = candidate_score - score.item()                    
                    # 如果生成的 token 是 EOS（结束符），将其添加到最终结果中
                    if next_token.item() == model.corpus.vocab["<eos>"]:
                        final_results.append((new_candidate, new_score))
                    else:
                        # 将新生成的候选序列添加到新候选列表中
                        new_candidates.append((new_candidate, new_score))
            # 从新候选列表中选择得分最高的 beam_width 个序列
            candidates = sorted(new_candidates, key=lambda x: x[1], reverse=True)[:beam_width]
    # 选择得分最高的候选序列，如果 final_results 为空，选择当前得分最高的候选序列
    if final_results:
        best_candidate, _ = sorted(final_results, key=lambda x: x[1])[0]
    else:
        best_candidate, _ = sorted(candidates, key=lambda x: x[1])[0]
    # 将输出 token 转换回文本字符串
    output_str = " ".join([model.corpus.idx2word[token] for token in best_candidate])
    return output_str
GPT 完整实现
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
class GPT(nn.Module):
    def __init__(self, corpus):
        super(GPT, self).__init__()
        self.corpus = corpus
        self.decoder = Decoder(corpus) # 解码器，用于学习文本生成能力
        self.projection = nn.Linear(d_embedding, corpus.vocab_size)  # 全连接层，输出预测结果

    def forward(self, dec_inputs):        
        dec_outputs = self.decoder(dec_inputs) # 将输入数据传递给解码器
        logits = self.projection(dec_outputs) # 传递给全连接层以生成预测
        return logits # 返回预测结果
    
    def decode(self, input_str, strategy='greedy', **kwargs):
        if strategy == 'greedy': # 贪心解码函数
            return generate_text_greedy_search(self, input_str, **kwargs)
        elif strategy == 'beam_search': # 集束解码函数
            return generate_text_beam_search(self, input_str, **kwargs)
        else:
            raise ValueError(f"Unknown decoding strategy: {strategy}")

预训练一个轻量 GPT

定义语料库类
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
34
35
36
37
38
39
# WikiCorpus语料库类
class WikiCorpus:
    def __init__(self, sentences, max_seq_len=256):
        self.sentences = sentences
        self.seq_len = max_seq_len
        self.vocab = self.create_vocabularies()
        self.vocab_size = len(self.vocab)
        self.idx2word = {v: k for k, v in self.vocab.items()}

    def create_vocabularies(self):
        # counter = Counter(word for sentence in self.sentences for word in sentence.split())
        # vocab = {'<pad>': 0, '<sos>': 1, '<eos>': 2, **{word: i+3 for i, word in enumerate(counter)}}
        with open("shared_vocab.txt", "r") as f:
            vocab = {line.split()[0]: int(line.split()[1]) for line in f}
        return vocab   


    def make_batch(self, batch_size):
        input_batch, target_batch = [], []

        # 随机选择句子索引
        sentence_indices = torch.randperm(len(self.sentences))[:batch_size]
        for index in sentence_indices:
            sentence = self.sentences[index]
            words = sentence.split()[:self.seq_len - 2]  # 截断句子,确保长度不超过 max_seq_len - 2（为了留出 <sos> 和 <eos>）
            seq = [self.vocab['<sos>']] + [self.vocab[word] for word in words] + [self.vocab['<eos>']]

            # 对序列进行填充
            seq += [self.vocab['<pad>']] * (self.seq_len - len(seq))

            # 将处理好的序列添加到批次中
            input_batch.append(seq[:-1])
            target_batch.append(seq[1:])

        # 将批次转换为LongTensor类型
        input_batch = torch.LongTensor(input_batch)
        target_batch = torch.LongTensor(target_batch)

        return input_batch, target_batch
训练类
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
import torch
import torch.nn as nn
import torch.optim as optim

class Trainer:
    def __init__(self, model, corpus, batch_size=24, learning_rate=0.01, epochs=10, device=None):
        self.model = model
        self.corpus = corpus
        self.vocab_size = corpus.vocab_size
        self.batch_size = batch_size
        self.lr = learning_rate
        self.epochs = epochs
        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")        
        self.criterion = nn.CrossEntropyLoss(ignore_index=corpus.vocab["<pad>"])
        self.optimizer = optim.Adam(self.model.parameters(), self.lr)
    
    def train(self):
        self.model.to(self.device)
        for epoch in range(self.epochs):
            self.optimizer.zero_grad()
            dec_inputs, target_batch = self.corpus.make_batch(self.batch_size)
            dec_inputs, target_batch = dec_inputs.to(self.device), target_batch.to(self.device)
            outputs = self.model(dec_inputs)
            loss = self.criterion(outputs.view(-1, self.corpus.vocab_size), target_batch.view(-1))
            loss.backward()
            self.optimizer.step()
训练过程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from Utilities import read_data

# 导入数据集    
from CorpusLoader import WikiCorpus
corpus = WikiCorpus(read_data('wikitext-103/wiki.train.txt')) 

# 导入 GPT 模型
from GPT_Model_with_Decode import GPT
WikiGPT = GPT(corpus)
print(WikiGPT) # 打印模型架构

# 训练 GPT 模型
from ModelTrainer import Trainer
trainer = Trainer(WikiGPT, corpus, learning_rate=0.01, epochs = 200)
trainer.train()  

import datetime
# 获取当前时间戳
now = datetime.datetime.now()
timestamp = now.strftime("%Y%m%d_%H%M%S")

# 保存模型
import torch
model_save_path = f'99_TrainedModel/WikiGPT_{trainer.lr}_{trainer.epochs}_{timestamp}.pth'

从 GPT 到 ChatGPT

GPT 预训练模型是完整语言模型的基础。预训练模型通过在大量文本上学习语言的统计规律，来获得对语言的一般理解。这个阶段模型不专注于任何特定任务，只是学习如何预测文本中下一个单词的出现。

完成预训练后，模型可以用于各种特定的语言任务，如文本生成、翻译、问答等。为了在这些任务上表现得更好，模型通常会进行微调（Fine-tuning），在特定任务的数据集上进一步训练，以适应特定的语言使用场景。预训练模型为这个过程提供了一个强大的起点，微调则根据特定任务进一步优化模型。

机器学习
#机器学习 #NLP #Transformer #深度学习 #GPT #LLM
基于 Transfomer 的预训练模型 | GPT
https://goer17.github.io/2023/12/03/Transfomer-家族之-GPT/
作者
Captain_Lee
发布于
2023年12月3日
许可协议