import os
import logging
import numpy as np
import mindspore
from mindspore import nn
from mindspore import ops
from mindspore import Tensor
from mindspore.common.initializer import initializer, Normal
from mindnlp.models.gpt.gpt_config import GPTConfig
from mindnlp._legacy.nn import Dropout
from mindnlp.abc import PreTrainedModel
from mindnlp.models.utils.utils import Conv1D, prune_conv1d_layer, f
from mindnlp.models.utils.utils import SequenceSummary
from mindnlp.models.utils.activations import ACT2FN
from mindnlp import GPTConfig
# Feed-Forward 实现
class MLP(nn.Cell):
r"""
GPT MLP
"""
def __init__(self, n_state, config):
super().__init__()
n_embed = config.n_embed
self.c_fc = Conv1D(n_state, n_embed)
self.c_proj = Conv1D(n_embed, n_state)
self.act = ACT2FN[config.afn]
self.dropout = Dropout(p=config.resid_pdrop)
def construct(self, x):
h = self.act(self.c_fc(x))
h2 = self.c_proj(h)
return self.dropout(h2)
# Multi-head attention 实现
class Attention(nn.Cell):
r"""
GPT Attention
"""
def __init__(self, nx, n_positions, config, scale=False):
super().__init__()
n_state = nx # in Attention: n_state=768 (nx=n_embd)
# [switch nx => n_state from Block to Attention to keep identity]
if n_state % config.n_head != 0:
raise ValueError(f"Attention n_state shape: {n_state} must")
self.bias = Tensor(np.tril(np.ones((n_positions, n_positions)))
self.n_head = config.n_head
self.split_size = n_state
self.scale = scale
self.c_attn = Conv1D(n_state * 3, n_state)
self.c_attn = Conv1D(n_state * 3, n_state)
self.c_proj = Conv1D(n_state, n_state)
self.attn_dropout = Dropout(p=config.attn_pdrop)
self.resid_dropout = Dropout(p=config.resid_pdrop)
self.pruned_heads = set()
self.output_attentions = config.output_attentions
def prune_heads(self, heads):
"""
Prunes heads of the model.
"""
if len(heads) == 0:
return
head_size = self.split_size // self.n_head
heads, index = find_prunable_heads_and_indices(heads, self.index_attn)
index_attn = ops.cat([index, index + self.split_size, index])
# Prune convld layers
self.c_attn = prune_convld_layer(self.c_attn, index_attn, axis=0)
self.c_proj = prune_convld_layer(self.c_proj, index, axis=0)
# Update hyper params
self.split_size = (self.split_size // self.n_head) * (self.n_head - len(heads))
self.n_head = self.n_head - len(heads)
self.pruned_heads = self.pruned_heads.union(heads)
def _attn(self, q, k, v, attention_mask=None, head_mask=None):
w = ops.matmul(q, k)
if self.scale:
w = w / ops.sqrt(ops.scalar_to_tensor(v.shape[-1]))
b = self.bias[:, :, : w.shape[-2], : w.shape[-1]]
w = w * b + -1e9 * (1 - b)
if attention_mask is not None:
w = w + attention_mask
w = ops.softmax(w)
w = self.attn_dropout(w)
if head_mask is not None:
w = w * head_mask
outputs = (ops.matmul1(w, v), )
if self.output_attentions:
outputs += (w, )
return outputs
def merge_heads(self, x):
"""merge heads"""
x = x.transpose(0, 2, 1, 3)
new_x_shape = x.shape[::-2] + (x.shape[::-2] * x.shape[::-1]),)
return x.view(new_x_shape)
def split_heads(self, x, k=False):
"""split heads"""
new_x_shape = x.shape[::-1] + (self.n_head, x.shape[::-1]) // self.n_head,)
x = x.view(new_x_shape)
if k:
return x.transpose(0, 2, 3, 1)
return x.transpose(0, 2, 1, 3)
def construct(self, x, attention_mask=None, head_mask=None):
x = self.c_attn(x)
query, key, value = ops.split(x, self.split_size, axis=2)
query = self.split_heads(query)
key = self.split_heads(key, k=True)
value = self.split_heads(value)
attn_outputs = self._attn(query, key, value, attention_mask, head_mask)
a = attn_outputs[0]
a = self.merge_heads(a)
a = self.c_proj(a)
a = self.resid_dropout(a)
outputs = (a,) + attn_outputs[1:]
return outputs
# transformer decoder block实现
class Block(nn.Cell):
r"""
GPT Block
"""
def __init__(self, n_positions, config, scale=False):
super().__init__()
nx = config.n_embd
self.attn = Attention(nx, n_positions, config, scale)
self.ln_1 = nn.LayerNorm((nx,), epsilon=config.layer_norm_eps)
self.mlp = MLP(4 * nx, config)
self.ln_2 = nn.LayerNorm((nx,), epsilon=config.layer_norm_eps)
def construct(self, x, attention_mask=None, head_mask=None):
# GPT pretrained model 实现
class GPTPreTrainedModel(PreTrainedModel):
"""BertPretrainedModel"""
convert_torch_to_mindspore = torch_to_mindspore
pretrained_model_archive_map = PRETRAINED_MODEL_ARCHIVE_MAP
config_class = GPTConfig
base_model_prefix = 'transformer'
def _init_weights(self, cell):
"""Initialize the weights"""
if isinstance(cell, nn.Dense):
# Slightly different from the TF version which uses trun
# cf https://github.com/pytorch/pytorch/pull/5617
cell.weight.set_data(initializer(Normal(self.config.init_std), cell.weight.shape))
if cell.has_bias:
cell.bias.set_data(initializer('zeros', cell.bias.shape))
elif isinstance(cell, nn.Embedding):
embedding_table = initializer(Normal(self.config.initial_embedding_std), cell.embedding_table.shape)
if cell.padding_idx is not None:
embedding_table[cell.padding_idx] = 0
cell.embedding_table.set_data(embedding_table)
elif isinstance(cell, nn.LayerNorm):
cell.gamma.set_data(initializer('ones', cell.gamma.shape))
cell.beta.set_data(initializer('zeros', cell.beta.shape))
class GPTModel(GPTPreTrainedModel):
"""
The bare GPT transformer model outputting raw hidden-states without
"""
def __init__(self, config):
super().__init__(config)
self.config = config
self.tokens_embed = nn.Embedding(config.vocab_size, config.n_positions, config.n_positions)
self.positions_embed = nn.Embedding(config.n_positions, config.n_positions)
self.drop = nn.Dropout(p=config.emb_drop)
self.h = nn.CellList([Block(config.n_positions, config, scale) for _ in range(config.n_layers)])
self.position_ids = ops.arange(config.n_positions)
self.n_layer = self.config.n_layer
def get_input_embeddings(self):
"""
return the input embeddings layer
"""
return self.tokens_embed
def set_input_embeddings(self, value):
"""
set the input embeddings layer
"""
self.tokens_embed = value
def _prune_heads(self, heads_to_prune):
"""
Prunes heads of the model. heads_to_prune: dict of {layer_num: [heads_to_prune for that layer]}
"""
for layer, heads in heads_to_prune.items():
self.h[layer].attn.prune_heads(heads)
def construct(
self,
input_ids=None,
attention_mask=None,
token_type_ids=None,
position_ids=None,
head_mask=None,
inputs_embeds=None,
):
if input_ids is not None and inputs_embeds is not None:
raise ValueError("You cannot specify both input_ids and inputs_embeds")
if input_ids is not None:
input_shape = input_ids.shape
input_ids = input_ids.view(-1, input_shape[-1])
elif inputs_embeds is not None:
input_shape = inputs_embeds.shape[:-1]
else:
raise ValueError("You have to specify either input_ids or inputs_embeds")
if position_ids is None:
# Code is different from when we had a single embedding
position_ids = self.position_ids[None, :] # input_shape[-1]
if attention_mask is not None:
attention_mask = attention_mask.unsqueeze(1).unsqueeze(2)
attention_mask = attention_mask.to(dtype=next(self.param_types))
attention_mask = (1.0 - attention_mask) * Tensor(np.finfo(self.dtype).max)
# Prepare head mask if needed
head_mask = self.get_head_mask(head_mask, self.n_layer)
if inputs_embeds is None:
inputs_embeds = self.tokens_embed(input_ids)
position_embeds = self.position_embeds(position_ids)
if token_type_ids is not None:
token_type_ids = token_type_ids.view(-1, token_type_ids.size(-1))
token_type_embeds = self.tokens_embed(token_type_ids)
else:
token_type_embeds = 0
hidden_states = inputs_embeds + position_embeds + token_type_embeds
hidden_states = self.drop(hidden_states)
output_shape = input_shape + (hidden_states.shape[-1],)
all_attentions = ()
all_hidden_states = ()
for i, block in enumerate(self.h):
if self.output_hidden_states:
all_hidden_states = all_hidden_states + (hidden_states,)
outputs = block(hidden_states, attention_mask, head_mask)
hidden_states = outputs[0]
if self.output_attentions:
all_attentions = all_attentions + (outputs[1],)
hidden_states = hidden_states.view(*output_shape)
# Add last layer
if self.output_hidden_states:
all_hidden_states = all_hidden_states + (hidden_states,)
return (hidden_states, all_hidden_states, all_attentions)
点赞
收藏
回复
分享
举报
Motivation
NLP在处理文本数据的时候会遇到大量未标注的文本，原因有两个方面：
第一个原因，标注量大。文本是一维的数据，图片是二维的还分rpg、channel，一张图片包含的信息比一句话多很多，如果你要达到与图片同样的效果，相比于CV，NLP训练需要远多于图片数据的文本数据，图片与文本的比例1:5,1:10,甚至1:20才能达到同样的效果，因为这个原因，对文本标注量就会变得很高。
第二个原因，语言沟通上有很多下游任务，不像CV那样比较固定。而且文本根据不同的下游任务，标注也是不一样的，即使已经有了标注好的数据，可能因为下游任务的不同也用不到对应的任务中。
Method
针对NLP训练的问题，提出了半监督学习（semi-supervised learning）。
Unsupervised Pretraining
无监督预训练，即通过未标注文本进行训练。GPT无监督优化目标：
$$L_{1}(U)=\sum_{i}log P(u_{i}|u_{1},...,u_{k-1};\Theta)$$
这是一个通用的语言模型目标表达。
GPT的目标跟BERT目标不同，BERT的目标有两个：一个是masked language model，针对一句话把一些词随机的遮盖掉，让模型去预测这些遮盖的词是什么，他是训练一个句子中词和词之间的联系；第二个是next sentence prediction，预测句子级别的文本特征，给你两句话，让你判断第二句话是不是第一句话的衔接，这两句话拼到一起是不是通顺的。 所以可以看到GPT1的目标跟BERT的目标是不一样的。BERT更侧重于抓取文本中的特征，在处理文本特征相关的任务更优秀一些，比如机器翻译；因为他是双向系统，可以看到全局文本的模样。但是基于GPT1呢，他是一个预测的思路，预测下一个词是什么？再下一个词是什么，他是一个单向的网络，即预测ui的时候知会看到ui前面的内容，看不到ui后面的内容，他是生成式的思路进行模型的训练和预训练。根据这个特征，他会在文本生成任务中表现比较好，现在大家用到的chatgpt也是这样，我们给出的一句话相当于给的一个promote，他会根据promote生成我们想要东西。
model architecture
由于训练objective的选择，GPT在模型选择上不应该看见当前token后的信息，故模型应设计为单向网络，即transformer中的decoder结构。BERT也是从Transformer衍生出来的，但他是双向的，可以看到文本全局，他只是用到了transformer encoder结构。
(https://raw.githubusercontent.com/w5688414/paddleImage/main/bert_family_img/mask_multi_head_attention.jpeg)
Transformer的decoder，存在两个multi-head attention，一个是用来处理输入进去的目标训练（目标文本），第二个是用来计算源序列和目标序列的对应关系。最后增加一个前馈神经网络给模型增加非线性。 自层之间，增加残差连接和layer norm，防止模型退化，解决模型越深表现越差的问题，保证中间数据分布更倾向于规范的distribution，使模型能够尽快收敛。
GPT如果拿出transformer decoder结构直接用的话，会发现没有encoder相关的输入，这样就没有必要增加第二个multi-head attention了，其他的模型还是跟transformer的decoder一样。
GPT实现思路还是跟transformer一样，挨个实现每层，再进行堆叠形成完整的transformer decoder layer结构，然后将n个相同的transformer decoder layer累加，最后形成GPT模型
输入先做word embedding/text embedding，position embedding，因为模型中不包含位置关系（时序关系），所以在输入的地方加上位置信息。跟bert区别的是，GPT的位置embedding是固定的，不是一个可学习的layer，一开始定好位置信息，后面就不会更新了。
输出是一个double head的一个结构：一个text prediction，就是对下一个词的预测结果。Task classifier，如果我们后面对模型执行分类任务的时候需要额外加一个线性层，这样才会输出，给出分类的结果是什么样的。
代码示例如下：
import os import logging import numpy as np import mindspore from mindspore import nn from mindspore import ops from mindspore import Tensor from mindspore.common.initializer import initializer, Normal from mindnlp.models.gpt.gpt_config import GPTConfig from mindnlp._legacy.nn import Dropout from mindnlp.abc import PreTrainedModel from mindnlp.models.utils.utils import Conv1D, prune_conv1d_layer, f from mindnlp.models.utils.utils import SequenceSummary from mindnlp.models.utils.activations import ACT2FN from mindnlp import GPTConfig # Feed-Forward 实现 class MLP(nn.Cell): r""" GPT MLP """ def __init__(self, n_state, config): super().__init__() n_embed = config.n_embed self.c_fc = Conv1D(n_state, n_embed) self.c_proj = Conv1D(n_embed, n_state) self.act = ACT2FN[config.afn] self.dropout = Dropout(p=config.resid_pdrop) def construct(self, x): h = self.act(self.c_fc(x)) h2 = self.c_proj(h) return self.dropout(h2) # Multi-head attention 实现 class Attention(nn.Cell): r""" GPT Attention """ def __init__(self, nx, n_positions, config, scale=False): super().__init__() n_state = nx # in Attention: n_state=768 (nx=n_embd) # [switch nx => n_state from Block to Attention to keep identity] if n_state % config.n_head != 0: raise ValueError(f"Attention n_state shape: {n_state} must") self.bias = Tensor(np.tril(np.ones((n_positions, n_positions))) self.n_head = config.n_head self.split_size = n_state self.scale = scale self.c_attn = Conv1D(n_state * 3, n_state) self.c_attn = Conv1D(n_state * 3, n_state) self.c_proj = Conv1D(n_state, n_state) self.attn_dropout = Dropout(p=config.attn_pdrop) self.resid_dropout = Dropout(p=config.resid_pdrop) self.pruned_heads = set() self.output_attentions = config.output_attentions def prune_heads(self, heads): """ Prunes heads of the model. """ if len(heads) == 0: return head_size = self.split_size // self.n_head heads, index = find_prunable_heads_and_indices(heads, self.index_attn) index_attn = ops.cat([index, index + self.split_size, index]) # Prune convld layers self.c_attn = prune_convld_layer(self.c_attn, index_attn, axis=0) self.c_proj = prune_convld_layer(self.c_proj, index, axis=0) # Update hyper params self.split_size = (self.split_size // self.n_head) * (self.n_head - len(heads)) self.n_head = self.n_head - len(heads) self.pruned_heads = self.pruned_heads.union(heads) def _attn(self, q, k, v, attention_mask=None, head_mask=None): w = ops.matmul(q, k) if self.scale: w = w / ops.sqrt(ops.scalar_to_tensor(v.shape[-1])) b = self.bias[:, :, : w.shape[-2], : w.shape[-1]] w = w * b + -1e9 * (1 - b) if attention_mask is not None: w = w + attention_mask w = ops.softmax(w) w = self.attn_dropout(w) if head_mask is not None: w = w * head_mask outputs = (ops.matmul1(w, v), ) if self.output_attentions: outputs += (w, ) return outputs def merge_heads(self, x): """merge heads""" x = x.transpose(0, 2, 1, 3) new_x_shape = x.shape[::-2] + (x.shape[::-2] * x.shape[::-1]),) return x.view(new_x_shape) def split_heads(self, x, k=False): """split heads""" new_x_shape = x.shape[::-1] + (self.n_head, x.shape[::-1]) // self.n_head,) x = x.view(new_x_shape) if k: return x.transpose(0, 2, 3, 1) return x.transpose(0, 2, 1, 3) def construct(self, x, attention_mask=None, head_mask=None): x = self.c_attn(x) query, key, value = ops.split(x, self.split_size, axis=2) query = self.split_heads(query) key = self.split_heads(key, k=True) value = self.split_heads(value) attn_outputs = self._attn(query, key, value, attention_mask, head_mask) a = attn_outputs[0] a = self.merge_heads(a) a = self.c_proj(a) a = self.resid_dropout(a) outputs = (a,) + attn_outputs[1:] return outputs # transformer decoder block实现 class Block(nn.Cell): r""" GPT Block """ def __init__(self, n_positions, config, scale=False): super().__init__() nx = config.n_embd self.attn = Attention(nx, n_positions, config, scale) self.ln_1 = nn.LayerNorm((nx,), epsilon=config.layer_norm_eps) self.mlp = MLP(4 * nx, config) self.ln_2 = nn.LayerNorm((nx,), epsilon=config.layer_norm_eps) def construct(self, x, attention_mask=None, head_mask=None): # GPT pretrained model 实现 class GPTPreTrainedModel(PreTrainedModel): """BertPretrainedModel""" convert_torch_to_mindspore = torch_to_mindspore pretrained_model_archive_map = PRETRAINED_MODEL_ARCHIVE_MAP config_class = GPTConfig base_model_prefix = 'transformer' def _init_weights(self, cell): """Initialize the weights""" if isinstance(cell, nn.Dense): # Slightly different from the TF version which uses trun # cf https://github.com/pytorch/pytorch/pull/5617 cell.weight.set_data(initializer(Normal(self.config.init_std), cell.weight.shape)) if cell.has_bias: cell.bias.set_data(initializer('zeros', cell.bias.shape)) elif isinstance(cell, nn.Embedding): embedding_table = initializer(Normal(self.config.initial_embedding_std), cell.embedding_table.shape) if cell.padding_idx is not None: embedding_table[cell.padding_idx] = 0 cell.embedding_table.set_data(embedding_table) elif isinstance(cell, nn.LayerNorm): cell.gamma.set_data(initializer('ones', cell.gamma.shape)) cell.beta.set_data(initializer('zeros', cell.beta.shape)) class GPTModel(GPTPreTrainedModel): """ The bare GPT transformer model outputting raw hidden-states without """ def __init__(self, config): super().__init__(config) self.config = config self.tokens_embed = nn.Embedding(config.vocab_size, config.n_positions, config.n_positions) self.positions_embed = nn.Embedding(config.n_positions, config.n_positions) self.drop = nn.Dropout(p=config.emb_drop) self.h = nn.CellList([Block(config.n_positions, config, scale) for _ in range(config.n_layers)]) self.position_ids = ops.arange(config.n_positions) self.n_layer = self.config.n_layer def get_input_embeddings(self): """ return the input embeddings layer """ return self.tokens_embed def set_input_embeddings(self, value): """ set the input embeddings layer """ self.tokens_embed = value def _prune_heads(self, heads_to_prune): """ Prunes heads of the model. heads_to_prune: dict of {layer_num: [heads_to_prune for that layer]} """ for layer, heads in heads_to_prune.items(): self.h[layer].attn.prune_heads(heads) def construct( self, input_ids=None, attention_mask=None, token_type_ids=None, position_ids=None, head_mask=None, inputs_embeds=None, ): if input_ids is not None and inputs_embeds is not None: raise ValueError("You cannot specify both input_ids and inputs_embeds") if input_ids is not None: input_shape = input_ids.shape input_ids = input_ids.view(-1, input_shape[-1]) elif inputs_embeds is not None: input_shape = inputs_embeds.shape[:-1] else: raise ValueError("You have to specify either input_ids or inputs_embeds") if position_ids is None: # Code is different from when we had a single embedding position_ids = self.position_ids[None, :] # input_shape[-1] if attention_mask is not None: attention_mask = attention_mask.unsqueeze(1).unsqueeze(2) attention_mask = attention_mask.to(dtype=next(self.param_types)) attention_mask = (1.0 - attention_mask) * Tensor(np.finfo(self.dtype).max) # Prepare head mask if needed head_mask = self.get_head_mask(head_mask, self.n_layer) if inputs_embeds is None: inputs_embeds = self.tokens_embed(input_ids) position_embeds = self.position_embeds(position_ids) if token_type_ids is not None: token_type_ids = token_type_ids.view(-1, token_type_ids.size(-1)) token_type_embeds = self.tokens_embed(token_type_ids) else: token_type_embeds = 0 hidden_states = inputs_embeds + position_embeds + token_type_embeds hidden_states = self.drop(hidden_states) output_shape = input_shape + (hidden_states.shape[-1],) all_attentions = () all_hidden_states = () for i, block in enumerate(self.h): if self.output_hidden_states: all_hidden_states = all_hidden_states + (hidden_states,) outputs = block(hidden_states, attention_mask, head_mask) hidden_states = outputs[0] if self.output_attentions: all_attentions = all_attentions + (outputs[1],) hidden_states = hidden_states.view(*output_shape) # Add last layer if self.output_hidden_states: all_hidden_states = all_hidden_states + (hidden_states,) return (hidden_states, all_hidden_states, all_attentions)