Datawhale------Tiny-universe学习笔记—

1. Qwen整体介绍

对于一个完全没接触过大模型的小白来说，猛一听这个名字首先会一懵：Qwen是啥。这里首先解答一下这个问题。下面是官网给出介绍：Qwen是阿里巴巴集团Qwen团队研发的大语言模型和大型多模态模型系列。其实随着大模型领域的发展，这类产品已经有很多了例如：由百度开发的ERNIE，由清华大学开发的Zhuiyi等等。

目前，Qwen已升级至Qwen2版本。无论是语言模型还是多模态模型，均在大规模多语言和多模态数据上进行预训练，并通过高质量数据进行后期微调以贴近人类偏好。Qwen具备自然语言理解、文本生成、视觉理解、音频理解、工具使用、角色扮演、作为AI Agent进行互动等多种能力。

废话不多说，我们可以先看一下Qwen的整体架构。Qwen的整体架构与Llama2类似，如下图所示:

接下来我们顺着整体架构图学习，对于输入问题Text，首先会经过Tokenizer。在这里，对于没有了解过NLP的友友们又开始疑惑了：Tokenizer是啥？其实Tokenizer就是一个分词器，在这里的作用就是将问题中句子分成各个词，每一个词都对应着词表的索引，每个索引对应着一个词向量。

接着之后生成一个input_ids，由此输入Qwen2的主干部分。

1.1 模型初始化

第一部首先进行模型初始化:

class Qwen2Model(Qwen2PreTrainedModel):def __init__(self, config: Qwen2Config):super().__init__(config)self.padding_idx = config.pad_token_idself.vocab_size = config.vocab_sizeself.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 processingself.post_init()

下面我们来解释一下这段代码：

1.def __init__(self, config: Qwen2Config):
super().__init__(config)

Qwen2Model继承自Qwen2PreTrainedModel，Qwen2PreTrainedModel继承自PreTrainedModel。PretrainedConfig是transformers框架中所有配置类的基类。

Qwen2PreTrainedModel是已经预训练好的模型，具体代码如下：

引自：transformers/src/transformers/models/qwen2/modeling_qwen2.py at main · huggingface/transformers · GitHub

class Qwen2PreTrainedModel(PreTrainedModel):config_class = Qwen2Configbase_model_prefix = "model"supports_gradient_checkpointing = True_no_split_modules = ["Qwen2DecoderLayer"]_skip_keys_device_placement = "past_key_values"_supports_flash_attn_2 = True_supports_sdpa = True_supports_cache_class = True_supports_quantized_cache = True_supports_static_cache = Truedef _init_weights(self, module):std = self.config.initializer_rangeif isinstance(module, nn.Linear):module.weight.data.normal_(mean=0.0, std=std)if module.bias is not None:module.bias.data.zero_()elif isinstance(module, nn.Embedding):module.weight.data.normal_(mean=0.0, std=std)if module.padding_idx is not None:module.weight.data[module.padding_idx].zero_()

2. self.padding_idx = config.pad_token_id
self.vocab_size = config.vocab_size

这里设置了模型的两个属性:padding_idx（用于指定填充标记的索引），vocab_size（词汇表的大小，即模型能够处理的不同token的数量）。

3. 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)

初始化模型的嵌入层、解码器层、归一化层：

嵌入层（nn.Embedding）：模型使用嵌入层将输入的标记映射成密集的向量表示。config.vocab_size是词汇表的大小，config.hidden_size是嵌入向量的维度，self.padding_idx是padding token的索引。
解码器层（nn.ModuleList()）：模型包含多个解码器层，这些层都是由 `Qwen2DecoderLayer`` 定义。每个解码器层都是根据配置对象中的参数构建的，并且有一个索引layer_idx，它表示层在模型中的位置。
归一化层 Qwen2RMSNorm：归一化层使用的是 Root Mean Square Layer Normalization

4. self.gradient_checkpointing = False

设置了是否使用 gradient_checkpoint 主要是用来节省显存。它用于控制是否使用梯度检查点技术。这是一种节省内存的技术，通过在正向传播中丢弃一些中间梯度来实现。

5. self.post_init()

调用 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'smodules properly initialized (such as weight initialization)."""self.init_weights()self._backward_compatibility_gradient_checkpointing()

1.2 forward方法

第二步实现Qwen2Model的forward方法。在实现 Qwen2Model 的 forward 方法时，我们需要分成三个主要部分：Embedding、Hidden States 和 Decoder Layers。这一过程展示了模型的前向传播行为，即在接收输入数据后，如何计算输出结果。

1. Embedding

首先，对于输入的 input_ids，我们将使用 torch.nn.Embedding 进行嵌入处理。这一步负责将每个输入标识符映射到一个高维向量空间中，以便后续的处理。

2. Hidden States

接下来，经过嵌入处理后的向量将转化为 Hidden States。这些状态代表了输入数据的内部表示，将作为输入提供给模型的 Decoder Layers。

3. Decoder Layers

最后，经过前两步处理的 Hidden States 会传递到多层的 Decoder Layers 进行进一步的处理。Decoder Layers 是模型的核心部分，它们通过自注意力机制和前馈神经网络对输入进行深入处理，以生成最终的输出。

代码如下：

引自：transformers/src/transformers/models/qwen2/modeling_qwen2.py at main · huggingface/transformers · GitHub

    @add_start_docstrings_to_model_forward(QWEN2_INPUTS_DOCSTRING)def forward(self,input_ids: torch.LongTensor = None,attention_mask: Optional[torch.Tensor] = None,position_ids: Optional[torch.LongTensor] = None,past_key_values: Optional[List[torch.FloatTensor]] = None,inputs_embeds: Optional[torch.FloatTensor] = None,use_cache: Optional[bool] = None,output_attentions: Optional[bool] = None,output_hidden_states: Optional[bool] = None,return_dict: Optional[bool] = None,cache_position: Optional[torch.LongTensor] = None,) -> Union[Tuple, BaseModelOutputWithPast]:output_attentions = output_attentions if output_attentions is not None else self.config.output_attentionsoutput_hidden_states = (output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states)use_cache = use_cache if use_cache is not None else self.config.use_cachereturn_dict = return_dict if return_dict is not None else self.config.use_return_dictif (input_ids is None) ^ (inputs_embeds is not None):raise ValueError("You cannot specify both input_ids and inputs_embeds at the same time, and must specify either one")if self.gradient_checkpointing and self.training:if use_cache:logger.warning_once("`use_cache=True` is incompatible with gradient checkpointing. Setting `use_cache=False`...")use_cache = Falseuse_legacy_cache = Falseif use_cache and not isinstance(past_key_values, Cache) and not self.training:use_legacy_cache = Truepast_key_values = DynamicCache.from_legacy_cache(past_key_values)logger.warning_once("We detected that you are passing `past_key_values` as a tuple and this is deprecated and will be removed in v4.46. ""Please use an appropriate `Cache` class (https://huggingface.co/docs/transformers/internal/generation_utils#transformers.Cache)")if inputs_embeds is None:inputs_embeds = self.embed_tokens(input_ids)if cache_position is None:past_seen_tokens = past_key_values.get_seq_length() if past_key_values is not None else 0cache_position = torch.arange(past_seen_tokens, past_seen_tokens + inputs_embeds.shape[1], device=inputs_embeds.device)if position_ids is None:position_ids = cache_position.unsqueeze(0)causal_mask = self._update_causal_mask(attention_mask, inputs_embeds, cache_position, past_key_values, output_attentions)hidden_states = inputs_embeds# create position embeddings to be shared across the decoder layersposition_embeddings = self.rotary_emb(hidden_states, position_ids)# decoder layersall_hidden_states = () if output_hidden_states else Noneall_self_attns = () if output_attentions else Nonenext_decoder_cache = Nonefor decoder_layer in self.layers:if output_hidden_states:all_hidden_states += (hidden_states,)if self.gradient_checkpointing and self.training:layer_outputs = self._gradient_checkpointing_func(decoder_layer.__call__,hidden_states,causal_mask,position_ids,past_key_values,output_attentions,use_cache,cache_position,position_embeddings,)else:layer_outputs = decoder_layer(hidden_states,attention_mask=causal_mask,position_ids=position_ids,past_key_value=past_key_values,output_attentions=output_attentions,use_cache=use_cache,cache_position=cache_position,position_embeddings=position_embeddings,)hidden_states = layer_outputs[0]if use_cache:next_decoder_cache = layer_outputs[2 if output_attentions else 1]if output_attentions:all_self_attns += (layer_outputs[1],)hidden_states = self.norm(hidden_states)# add hidden states from the last decoder layerif output_hidden_states:all_hidden_states += (hidden_states,)next_cache = Noneif use_cache:next_cache = next_decoder_cache.to_legacy_cache() if use_legacy_cache else next_decoder_cacheif not return_dict:return tuple(v for v in [hidden_states, next_cache, all_hidden_states, all_self_attns] if v is not None)return BaseModelOutputWithPast(last_hidden_state=hidden_states,past_key_values=next_cache,hidden_states=all_hidden_states,attentions=all_self_attns,)# Copied from transformers.models.llama.modeling_llama.LlamaModel._update_causal_mask

这里内容有点多，我们看核心：

inputs_embeds = self.embed_tokens(input_ids)
# embed positions
hidden_states = inputs_embedsfor idx, decoder_layer in enumerate(self.layers):# 将所有的hidden_states保存成tupleif output_hidden_states:all_hidden_states += (hidden_states,)# 将hs送入每一层decoder_layerlayer_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的一个类，然后进入下一个layerhidden_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的形式输出。

1.3 RMSNorm

计算公式：

其中:

x是层的输入的hidden_state
$eq?w_%7Bi%7D$ 表示的是hidden_state的最后一个维度的值
n 表示上面输入的最后一个维度的数量。
ϵ 表示是很小的数，防止除0。

class Qwen2RMSNorm(nn.Module):  # 标准化层def __init__(self, hidden_size, eps=1e-6):"""Qwen2RMSNorm is equivalent to T5LayerNorm"""super().__init__()self.weight = nn.Parameter(torch.ones(hidden_size))self.variance_epsilon = epsdef forward(self, hidden_states):input_dtype = hidden_states.dtypehidden_states = hidden_states.to(torch.float32)variance = hidden_states.pow(2).mean(-1, keepdim=True)hidden_states = hidden_states * torch.rsqrt(variance + self.variance_epsilon)return self.weight * hidden_states.to(input_dtype)