References: Qwen-blog
Qwen 整体架构：
其中:
tokenizer将文本转为词表里面的数值。- 数值经过
embedding得到一一对应的向量。 attention_mask是用来看见左边、右边，双向等等来设定。- 各类下游任务，
Casual,seqcls等，基本都是基础模型model后面接对应的Linear层，还有损失函数不一样。
首先从 github 下载 transformer 仓库，并命名为 transformer_213（因为我是 2.13 日期下载的），此步骤为了防止与包混淆。
定义 run_demo.ipynb 来详细了解 qwen2。
from transformers_213.src.transformers.models.qwen2 import Qwen2Config, Qwen2Model
import torch
def run_qwen2():
qwen2config = Qwen2Config(
vocab_size=151936,
hidden_size=4096//2,
num_hidden_layers=32//2,
num_attention_heads=32,
intermediate_size=2048//2
)
qwen2model = Qwen2Model(qwen2config)
input_ids = torch.randint(0, qwen2config.vocab_size, (4, 30))
res = qwen2model(input_ids)
print(res)
if __name__ == "__main__":
run_qwen2()
1 Qwen2Config#
Qwen2Config 中包含一些自定义的超参数，例如vocab_size,hidden_size,num_hidden_layers, num_attention_heads等。类似于dict可以调用里面的超参数:config.pad_token_id。
1.1 Qwen2Model#
1.1.1 初始化#
- 设置了模型的两个属性:
padding_idx（用于指定填充标记的索引），vocab_size（词汇表的大小） - 初始化了模型的嵌入层、解码器层、归一化层
- 嵌入层（
nn.Embedding）：模型使用嵌入层将输入的标记映射成密集的向量表示。 - 解码器层（
nn.ModuleList()）：模型包含多个解码器层，这些层都是由 `Qwen2DecoderLayer`` 定义 - 归一化层
Qwen2RMSNorm：归一化层使用的是 Root Mean Square Layer Normalization - 设置了是否使用
gradient_checkpoint主要是用来节省显存 - 调用
post_init()完成一些初始化和准备检查的代码
class Qwen2Model(Qwen2PreTrainedModel):
def __init__(self, config: Qwen2Config):
super().__init__(config)
self.padding_idx = config.pad_token_id
self.vocab_size = config.vocab_size
self.embed_tokens = nn.Embedding(config.vocab_size, config.hidden_size, self.padding_idx)
self.layers = nn.ModuleList(
[Qwen2DecoderLayer(config, layer_idx) for layer_idx in range(config.num_hidden_layers)]
)
self.norm = Qwen2RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
self.gradient_checkpointing = False
# Initialize weights and apply final processing
self.post_init()
对于post_init函数：
主要是对参数进行初始化，以及初始化梯度检查点作用
def post_init(self):
"""
A method executed at the end of each Transformer model initialization, to execute code that needs the model's
modules properly initialized (such as weight initialization).
"""
self.init_weights()
self._backward_compatibility_gradient_checkpointing()
1.1.2 Forward#
在此只对核心主干进行讲解:
inputs_embeds = self.embed_tokens(input_ids)
# embed positions
hidden_states = inputs_embeds
for idx, decoder_layer in enumerate(self.layers):
# 将所有的hidden_states保存成tuple
if output_hidden_states:
all_hidden_states += (hidden_states,)
# 将hs送入每一层decoder_layer
layer_outputs = decoder_layer(
hidden_states,
attention_mask=attention_mask,
position_ids=position_ids,
past_key_value=past_key_value,
output_attentions=output_attentions,
use_cache=use_cache,
)
# 取出上一层decoder_输出的hs,再传入下一个layer
# 只要第一个,第二个是cache的一个类，然后进入下一个layer
hidden_states = layer_outputs[0]
# 将最后layers输出后的hidden_states进行标准化
hidden_states = self.norm(hidden_states)
# 加上最后一层的hidden_states
if output_hidden_states:
all_hidden_states += (hidden_states,)
- 如果保存
output_hidden_states的话，就是第一个为input_ids进行emb，然后保存到n-1层的decoder_layer的输出hs，再加上最后一层layer的输出hs进行过norm后的hs. - 最后是以
BaseModelOutputWithPast的形式输出。
待续。。。。。