Qwen,作为一款备受瞩目的大语言模型,其架构设计与Llama2有着异曲同工之妙。本文将深入剖析Qwen模型的内部构造,着重解读其配置、核心层以及关键组件,旨在为读者呈现一幅清晰而详尽的技术蓝图。
Qwen2Config:模型配置的核心
Qwen2Config
类是Qwen模型配置的基石,它承载着模型初始化所需的各项参数。从词汇表大小到嵌入维度,再到注意力机制的细节设置,都囊括其中。
1.1 Model:模型构建的蓝图
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`,每个层对应模型的一个解码器层
[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:
前向传播是自然语言处理模型的核心,它定义了数据在模型中的流动方式。在Qwen模型中,前向传播过程涉及到嵌入、解码器层以及归一化等多个步骤。
inputs_embeds = self.embed_tokens(input_ids)#这行代码使用 `embed_tokens` 方法将输入的 `input_ids`转换为嵌入向量
hidden_states = inputs_embeds#初始化 `hidden_states` 为输入嵌入,它将在后续的解码器层中被更新
for idx, decoder_layer in enumerate(self.layers):
# 将所有的hidden_states保存成tuple
if output_hidden_states: #判断是否需要输出所有层的隐藏状态
all_hidden_states += (hidden_states,)
# 将hs送入每一层decoder_layer
# 调用当前解码器层的前向传播方法,传入当前的 `hidden_states` 和其他必要的参数
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]
hidden_states = self.norm(hidden_states)
if output_hidden_states:
all_hidden_states += (hidden_states,)
具体来说,输入的input_ids
首先通过嵌入层转换为嵌入向量。然后,这些嵌入向量被作为初始的hidden_states
,依次传入各个解码器层。每一层解码器都对hidden_states
进行处理,并输出新的hidden_states
,这些新的hidden_states
又被作为下一层的输入。最后,经过所有解码器层的处理后,hidden_states
通过归一化层进行标准化,并最终输出。
1.2 Qwen2DecoderLayer:解码器层的奥秘
Qwen2DecoderLayer
是Qwen模型的核心组成部分,它负责对输入进行解码和变换。一个Qwen2DecoderLayer
主要由自注意力机制(self-attention)、多层感知机(MLP)以及两种归一化层组成。
1.2.1 初始化:
在Qwen2DecoderLayer
的初始化过程中,首先确定了隐藏层的大小。然后,根据配置中的_attn_implementation
参数,选择合适的自注意力实现方式。接着,初始化多层感知机(MLP)和两个归一化层,分别用于自注意力之前和之后。
QWEN2_ATTENTION_CLASSES = {
"eager": Qwen2Attention, # 默认的注意力实现,一般情况下是这个
"flash_attention_2": Qwen2FlashAttention2, # 一个优化的注意力实现
"sdpa": Qwen2SdpaAttention, # 一种特殊的注意力实现
}
class Qwen2DecoderLayer(nn.Module):
def __init__(self, config: Qwen2Config):
super().__init__()
self.hidden_size = config.hidden_size # 设置隐藏层的大小
self.self_attn = QWEN2_ATTENTION_CLASSES[config._attn_implementation](config, layer_idx)# 根据配置中的 `_attn_implementation` 键来选择使用哪种自注意力实现,并初始化它
self.mlp = Qwen2MLP(config)# 初始化一个多层感知机(MLP),用于在自注意力之后处理隐藏状态
self.input_layernorm = Qwen2RMSNorm(config.hidden_size, eps=config.rms_norm_eps)# 初始化两个归一化层,分别用于自注意力之前和之后。这两个层都是 RMS 归一化层,使用配置中的 `hidden_size` 和 `rms_norm_eps` 参数
self.post_attention_layernorm = Qwen2RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
1.2.2 Forward:
在前向传播过程中,Qwen2DecoderLayer
首先将输入的hidden_states
进行归一化,然后通过自注意力机制进行处理。自注意力的输出与原始的hidden_states
进行残差连接,以缓解梯度消失问题。接着,将残差连接后的结果再次进行归一化,并通过多层感知机(MLP)进行处理。最后,MLP的输出再次与之前的hidden_states
进行残差连接,得到最终的输出。
residual = hidden_states# 保存原始的 `hidden_states` 到 `residual` 变量中,用于后面的残差连接
hidden_states = self.input_layernorm(hidden_states) # 将 `hidden_states` 通过输入归一化层(`input_layernorm`),RMSNorm标准化
hidden_states, self_attn_weights, present_key_value = self.self_attn(
hidden_states=hidden_states,
attention_mask=attention_mask,
position_ids=position_ids,
past_key_value=past_key_value,
output_attentions=output_attentions,
use_cache=use_cache,
**kwargs,
)
hidden_states = residual + hidden_states
residual = hidden_states
hidden_states = self.post_attention_layernorm(hidden_states)
hidden_states = self.mlp(hidden_states)
hidden_states = residual + hidden_states
outputs = (hidden_states,)
return outputs
1.3 Qwen2Attention:自注意力机制的实现
Qwen2Attention
类实现了多头自注意力机制,这是Transformer架构中的一个核心组件。它允许模型在处理序列数据时考虑到不同位置之间的关系。
1.3.1 初始化:
在Qwen2Attention
的初始化过程中,首先保存了传入的配置对象,并计算了每个注意力头的维度大小。然后,初始化了用于计算查询(Query)、键(Key)和值(Value)的线性层。此外,还初始化了一个旋转嵌入层,用于增强模型对序列顺序的感知能力。
class Qwen2Attention(nn.Module):
"""Multi-headed attention from 'Attention Is All You Need' paper"""
def __init__(self, config: Qwen2Config):
super().__init__()
self.config = config# 保存传入的配置对象,它包含了自注意力层所需的所有配置参数。
self.layer_idx = layer_idx# 保存索引
self.hidden_size = config.hidden_size
self.num_heads = config.num_attention_heads
self.head_dim = self.hidden_size // self.num_heads# 计算每个注意力头的维度大小,它是隐藏层大小除以头的数量
self.num_key_value_heads = config.num_key_value_heads# 设置键值对头的数量
self.num_key_value_groups = self.num_heads // self.num_key_value_heads
self.max_position_embeddings = config.max_position_embeddings# 设置最大位置嵌入的大小,这通常用于位置编码
self.rope_theta = config.rope_theta# 设置旋转嵌入(Rotary Positional Embedding)的参数
self.is_causal = True# 指示是否使用因果自注意力(即在生成下一个 token 时只能使用之前的 token)
self.attention_dropout = config.attention_dropout# 设置注意力权重的dropout率
if (self.head_dim * self.num_heads) != self.hidden_size:
raise ValueError(
f"hidden_size must be divisible by num_heads (got `hidden_size`: {self.hidden_size}"
f" and `num_heads`: {self.num_heads})."
)
self.q_proj = nn.Linear(self.hidden_size, self.num_heads * self.head_dim, bias=config.attention_bias)# Query
self.k_proj = nn.Linear(self.hidden_size, self.num_key_value_heads * self.head_dim, bias=config.attention_bias)# Key
self.v_proj = nn.Linear(self.hidden_size, self.num_key_value_heads * self.head_dim, bias=config.attention_bias)# Value
self.o_proj = nn.Linear(self.num_heads * self.head_dim, self.hidden_size, bias=config.attention_bias)# 计算最终输出的投影
# 初始化一个旋转嵌入层,这是一种特殊的位置编码,可以增强模型对序列顺序的感知能力
self.rotary_emb = Qwen2RotaryEmbedding(
self.head_dim,
max_position_embeddings=self.max_position_embeddings,
base=self.rope_theta,
)
1.3.2 Forward:
在前向传播过程中,Qwen2Attention
首先将输入的hidden_states
通过线性层转换为查询(Query)、键(Key)和值(Value)张量。然后,将旋转位置嵌入应用于查询和键张量,以增强模型对序列顺序的感知能力。接着,计算查询和键的点积,得到注意力权重。注意力权重经过softmax归一化后,与值张量进行加权求和,得到最终的注意力输出。
bsz, q_len, _ = hidden_states.size()
query_states = self.q_proj(hidden_states)
key_states = self.k_proj(hidden_states)
value_states = self.v_proj(hidden_states)
# reshape多头处理--分块--(bs,T,heads,hd_d),为了将输入的隐藏状态转换为适合多头自注意力计算的形式,每个头可以独立地处理序列的一部分,从而实现并行处理和更细粒度的表示学习
# `ranspose` 函数用于交换张量的两个维度
query_states = query_states.view(bsz, q_len, self.num_heads, self.head_dim).transpose(1, 2)
key_states = key_states.view(bsz, q_len, self.num_key_value_heads, self.head_dim).transpose(1, 2)
value_states = value_states.view(bsz, q_len, self.num_key_value_heads, self.head_dim).transpose(1, 2)
cos, sin = self.rotary_emb(value_states, seq_len=kv_seq_len)
query_states, key_states = apply_rotary_pos_emb(query_states, key_states, cos, sin, position_ids)
key_states = repeat_kv(key_states, self.num_key_value_groups)
value_states = repeat_kv(value_states, self.num_key_value_groups)
attn_weights = torch.matmul(query_states, key_states.transpose(2, 3)) / math.sqrt(self.head_dim)
attn_weights = attn_weights + attention_mask
attn_weights = nn.functional.softmax(attn_weights, dim=-1, dtype=torch.float32).to(query_states.dtype)
attn_weights = nn.functional.dropout(attn_weights, p=self.attention_dropout, training=self.training)
attn_output = torch.matmul(attn_weights, value_states)
attn_output = attn_output.transpose(1, 2).contiguous()
attn_output = attn_output.reshape(bsz, q_len, self.hidden_size)
attn_output = self.o_proj(attn_output)
return attn_output, attn_weights, past_key_value
1.3.3 细节Debug:
1.3.3.1 GQA:
GQA(分组查询注意力)是一种优化注意力机制,旨在减少推理过程中对KV Cache的显存占用。通过共享键和值表示,GQA可以在不显著降低模型性能的前提下,大幅降低显存需求,从而提高LLM Serving的请求处理能力。
1.3.3.2 apply_rotary_pos_emb & 1.3.3.3 attention_mask:
旋转位置嵌入(Rotary Positional Embedding)和注意力掩码(Attention Mask)是自注意力机制中的两个重要组成部分。旋转位置嵌入用于将位置信息融入到查询和键中,从而提高模型对序列顺序的感知能力。注意力掩码用于控制模型在计算注意力权重时可以访问的位置,例如,在因果语言模型中,只能访问当前位置之前的位置。
1.4 Qwen2MLP:多层感知机的应用
Qwen2MLP
类实现了一个多层感知机(MLP)结构,通常用于Transformer模型中的前馈网络(Feed-Forward Network,FFN)。MLP通过引入非线性激活函数和中间层来增加模型的表达能力。在Qwen2MLP
中,使用了一种特殊的结构,其中gate_proj
的输出与up_proj
的输出相乘,这种结构有助于模型学习输入数据的复杂特征。
class Qwen2MLP(nn.Module):
def __init__(self, config):
super().__init__()
# 这俩不必多说
self.config = config
self.hidden_size = config.hidden_size
self.intermediate_size = config.intermediate_size
# 三个全连接层
self.gate_proj = nn.Linear(self.hidden_size, self.intermediate_size, bias=False)# 定义一个线性层,用于将输入投影到中间层的大小。这个层的权重在训练过程中是不变的(`bias=False`)
self.up_proj = nn.Linear(self.hidden_size, self.intermediate_size, bias=False)# 将输入投影到中间层的大小
self.down_proj = nn.Linear(self.intermediate_size, self.hidden_size, bias=False)# 将中间层的输出投影回隐藏层的大小
self.act_fn = ACT2FN[config.hidden_act]# 根据配置中的激活函数类型,选择相应的激活函数。`ACT2FN` 是一个将激活函数名称映射到 PyTorch 激活函数的字典
# 定义了前向传播函数,它是模型的输入数据流经网络的路径
def forward(self, x):
down_proj = self.down_proj(self.act_fn(self.gate_proj(x)) * self.up_proj(x))
return down_proj
在前向传播过程中,Qwen2MLP
首先将输入x
通过gate_proj
和up_proj
两个线性层进行投影。然后,将gate_proj
的输出通过激活函数进行非线性变换,并与up_proj
的输出相乘。最后,将相乘的结果通过down_proj
线性层投影回原始的隐藏层维度,得到最终的输出。
1.5 Qwen2RMSNorm:RMS归一化的应用
Qwen2RMSNorm
类实现了 RMS(Root Mean Square)归一化。RMS归一化与传统的LayerNorm不同,它只使用方差(不包括均值)来进行归一化,这使得它在某些情况下可能更简单或更有效。在Qwen2RMSNorm
中,通过对方差进行归一化,然后将结果乘以一个可学习的权重,这有助于模型调整归一化后的特征表示。
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 = eps
def forward(self, hidden_states):
input_dtype = hidden_states.dtype
hidden_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)
在前向传播过程中,Qwen2RMSNorm
首先将输入的hidden_states
转换为float32类型,并计算其方差。然后,通过对方差进行归一化,并将结果乘以一个可学习的权重。最后,将加权后的结果转换回原始的数据类型,并返回。
通过对Qwen模型的架构、核心层以及关键组件的深入剖析,我们可以更清晰地理解其工作原理和技术特点。这对于我们更好地应用Qwen模型,以及进行相关研究和开发具有重要意义。