作者：昇腾实战派 * Ming-L
关注公众号：AI模力圈

简介

Qwen3-VL 是阿里 Qwen 团队继 Qwen2.5-VL 之后推出的新一代多模态大模型，其最大的变革在于引入了 “DeepStack” (深层融合) 架构与 MoE (Mixture-of-Experts) 的全面结合，旨在解决超长视觉上下文（如长视频、极高分辨率图像）的推理效率与深度推理能力。该模型经过昇腾适配，当前具有优秀的表现。

昇腾适配仓：MindSpeed-MM/mindspeed_mm/models/transformers/qwen3vl/modeling_qwen3_vl_moe.py · Ascend/MindSpeed-MM - AtomGit | GitCode

开源代码链接：transformers/src/transformers/models/qwen3_vl_moe at main · huggingface/transformers ·GitHub

整体架构

模型架构图如下，qwen3-vl 也使用了典型的多模态三段式架构 VIT-Projector-LLM

image

架构主要分为 3 个核心部分：

  • Vision Encoder （VIT）
    • 支持 Native Resolution Input，即原生分辨率输入，不需要强制缩放或者 padding 后输入。
    • token 动态化：可以看到 encoder 后的 每个样本token 也是动态的，比如 picture为 11427tokens，picture2 为 8tokens，该设计最大化的保留了信息，也避免在简单图像中浪费计算资源。
  • Projector
    • 传统架构的 Projector对于视觉 Token 只在 LLM 的 layer0 侧注入，相当于提供了视觉 token 的总结报告，会导致忽略细节。
    • qwen3-vl 的 projector 使用了 DeepStack 策略，从图的右侧可以看到视觉信息会注入到 LLM 的不同层中，而非仅在 layer0 注入，增强了模型对图像细节的长期记忆和细粒度理解。（Projector 理解为组件，DeepStack 理解为策略，使用 Projector 实现 DeepStack）
  • LLM
    • 可以看到架构图，Qwen3-vl 在 LLM 处提供了 MOE（Qwen3-vl-235B-A22B）和 Dense 两个版本。
    • 使用了 Interleaved-MRoPE：将旋转位置编码分解为时间 (t)、高度 (h)、宽度 (w) 三个维度。这使得模型能完美理解视频的时序关系和图像的空间关系。

模型代码整体架构

image

qwen3-vl-moe主要有以下几个类：

  1. Qwen3VLMoeForConditionalGeneration
    • 封装面向用户的接口。
  2. Qwen3VLMoeModel
    • Qwen3-vl-Moe 整体模型定义，协调视觉，文本模块，构建完整多模态 MOE 功能。
  3. Qwen3VLMoeVisionModel
    • 即 VIT 视觉模块，主要进行视觉信息特征提取
  4. Qwen3VLMoeTextModel
    • LLM，进行 DeepStack 把视觉信息融入到 LLM 中；并且进行 MOE 路由，计算。

多模态融合

此处讲解 qwen3-vl的多模态融合做法，其主要分为以下几个阶段，特征提取，特征拼接，多模态特征融入 LLM。

  1. 特征提取 与 特征拼接：这一部分主要是提取视觉文本信息特征，最后进行主特征拼接。（这里面有主特征和 deepstack 特征，主特征就是模型的最后一层输出，deepstack 记录了模型的中间状态）

      • 代码如下，这里比较关键的函数：get_image_features(图像特征提取)，里面会遍历vit 每一 layer 的计算结果，把设定要保存的中间过程保存到 deepstack_list，最后一层输出作为主特征输出。

        • if pixel_values is not None:  
            image_embeds, deepstack_image_embeds = self.get_image_features(pixel_values, image_grid_thw)  
            image_embeds = torch.cat(image_embeds, dim=0).to(inputs_embeds.device, inputs_embeds.dtype)  
            image_mask, _ = self.get_placeholder_mask(  
                input_ids, inputs_embeds=inputs_embeds, image_features=image_embeds  
            )  
            inputs_embeds = inputs_embeds.masked_scatter(image_mask, image_embeds)  
                  #  涉及 video 的逻辑  
        if pixel_values_videos is not None:  
            video_embeds, deepstack_video_embeds = self.get_video_features(pixel_values_videos, video_grid_thw)  
            video_embeds = torch.cat(video_embeds, dim=0).to(inputs_embeds.device, inputs_embeds.dtype)  
            _, video_mask = self.get_placeholder_mask(  
                input_ids, inputs_embeds=inputs_embeds, video_features=video_embeds  
            )  
            inputs_embeds = inputs_embeds.masked_scatter(video_mask, video_embeds)

      • 流程图如下

        • image

    • 为了更清晰的了解其中的细节，将展示关键的 tensor.shape 计算过程。

      • image
        1. 假设输入一批图片，这一批图片有两张，imageA =224*224；imageB =448 * 448 ，config.patch_size =14；imageA.patch = 224/14 * 224/14 =256 ；imageB.patch = 448/14 * 448/14 = 1024
        2. 第一步是两个关键的入参
          • pixel_values：图片被打散后的原始像素数据集合，shape = [patch_size总和【256+1024】， vit 的输入维度【1152】]
          • image_grid_thw：批次图片结构元数据，shape = [批次图像数【2】, 结构数据 t，h，w【3】]
        3. 通过 get_image_features 进行图像信息特征提取，以及图像信息压缩（假设 spatail_merge_size2），256 个 token 被压缩成 64 个 token，1024 被压缩成 256 个 token
          • 输出 image_embeds[list]，每个 tensor 为[（压缩后的 token 数），（LLM 的 embedding 维度（维度对齐））]
        4. 通过 torch.cat 把处理后的视觉 token 放一起
          • 输出 tensor.shape=[视觉 token 总和【64 + 256】， LLM embedding 维度]

  2. LLM 前向与 DeepStack 融合

    • 代码如下：

      • #   Qwen3VLMoeTextModel.forward  
for layer_idx, decoder_layer in enumerate(self.layers):  
    #   A. 先跑当前层 (Attention + MoE)  
    #   注意：这里的 MoE 接收的是混合序列  
    hidden_states = decoder_layer(hidden_states, ...)  
    #   B. 【DeepStack 融合点】  
    #   如果这一层配置了注入  
    if deepstack_visual_embeds is not None and layer_idx in ...:  
        #   调用 _deepstack_process 进行 "残差连接"  
        hidden_states = self._deepstack_process(  
            hidden_states,   
            visual_pos_masks, #   告诉模型哪里是图片Token  
            deepstack_visual_embeds[layer_idx] #   对应的中间层视觉特征  
        )

    • 流程图如下

      • image

MOE 设计

Qwen3-VL的MoE采用了 **Fine-grained Top-K（细粒度专家，专家数量多）**机制，同时为了解决显存和速度的矛盾，代码在底层计算上区分了训练和推理路径。

这篇文章主要是想讲多模态 MOE，因此会主要讲一下和 LLM MOE 的区别，实际上两者的 MOE 实现方式没有本质上的区别，都是将 LLM 的 mlp 部分由 dense 改成 moe，而且代码结构也是几乎一样的。这里列出 qwen3-vl-moe 的几个特点，然后再对其 moe 代码进行讲解。

特点

  1. Router 处理的特征为混合特征，混合文本+图像。

    • qwen3-vl-moe处理方式：其 Router 并未引入针对特定模态的定制化逻辑，而是将文本与图像 Token 拼接为单一序列直接输入。依靠文本与图像在特征空间中天然存在的分布差异，模型能够自动区分不同模态的信息，因此无需进行显式的隔离或独立处理。

      •     # # # image和text的信息都合并到inputs_embeds一同输入到模型  
            #  1. 获取图像特征  
            if pixel_values is not None:  
                image_embeds, deepstack_image_embeds = self.get_image_features(pixel_values, image_grid_thw)  
                image_embeds = torch.cat(image_embeds, dim=0).to(inputs_embeds.device, inputs_embeds.dtype)  
                #  获取掩码  
                image_mask, _ = self.get_placeholder_mask(  
                    input_ids, inputs_embeds=inputs_embeds, image_features=image_embeds  
                )  
                #  2. 直接将图像特征“散射”覆盖到输入嵌入中  
                inputs_embeds = inputs_embeds.masked_scatter(image_mask, image_embeds)

  2. 视觉 patch 块过大问题

    • 视觉 token 是成 patch 出现的（如一次 256 个，或者几千个），在这种情况下有load_balancing_loss_func 函数，约束 router把视觉token打散分给不同专家，避免整块数据丢给一个专家导致分配不均，显存溢出等问题。

    • qwen3-vl-moe处理方式：解决这个问题的核心机制是 Token-Level Routing（Token 级路由） 配合 辅助负载均衡损失（Auxiliary Load Balancing Loss）。

      •     # # token级路由  
    def forward(self, hidden_states):  
        #  1. 压扁序列：这里把所有图片、所有文本的 Token 混在一起变成了二维矩阵  
        #  (batch_size, seq_len, dim) -> (batch_size * seq_len, dim)  
        hidden_states = hidden_states.reshape(-1, self.hidden_dim)  

        #  2. 计算每个 Token 去往哪个专家  
        #  router_logits: (total_tokens, num_experts)  
        router_logits = F.linear(hidden_states, self.weight)  

        #  3. 选出 Top-K  
        #  这里的 router_indices 决定了每个 Token 去哪里  
        router_top_value, router_indices = torch.topk(router_logits, self.top_k, dim=-1)  

        #  ...  
        return router_logits, router_scores, router_indices

      •     # # # # **只提取分配给该专家的 Token** 进行计算。这从物理上避免了显存溢出  
    def forward(self, hidden_states, top_k_index, top_k_weights):  
        #  ...  
        #  遍历每一个被选中的专家  
        for expert_idx in expert_hit:  
            expert_idx = expert_idx[0]  

            #  1. 找出有哪些 Token 属于当前这个专家  
            #  top_k_pos: 属于第几优选（1st or 2nd choice）  
            #  token_idx: 原序列中 Token 的索引位置（比如第 5 个和第 1000 个 token）  
            top_k_pos, token_idx = torch.where(expert_mask[expert_idx])  

            #  2. 只把这些 Token 拿出来 (Gather)  
            #  current_state 的长度远小于 visual token 的总数  
            current_state = hidden_states[token_idx]  

            #  3. 专家只计算这一小部分 Token  
            gate, up = nn.functional.linear(current_state, self.gate_up_proj[expert_idx]).chunk(2, dim=-1)  
            current_hidden_states = self.act_fn(gate) * up  
            current_hidden_states = nn.functional.linear(current_hidden_states, self.down_proj[expert_idx])  

            #  ...  
            #  4. 把计算结果放回去 (Scatter Add)  
            final_hidden_states.index_add_(0, token_idx, current_hidden_states.to(final_hidden_states.dtype))

      • # # Router 发现 Expert 1 接收了过多的 Token，Loss 就会变高，梯度下降会强迫 Router 修改权重，把 Token 赶到其他空闲的 Expert 去。  
def load_balancing_loss_func(gate_logits, num_experts, top_k=2, attention_mask=None):  
    #  ...  
    #  1. 计算路由概率  
    routing_weights = torch.nn.functional.softmax(concatenated_gate_logits, dim=-1)  

    #  ...  
    #  2. 计算每个专家被选中的实际比例 (tokens_per_expert)  
    #  expert_mask 是 one-hot 编码，mean 之后就是占比  
    tokens_per_expert = torch.mean(expert_mask.float(), dim=0)  

    #  3. 计算 Router 对每个专家的平均打分 (router_prob_per_expert)  
    router_prob_per_expert = torch.mean(routing_weights, dim=0)  

    #  4. Switch Transformer Loss:   
    #  如果分布极度不均匀（比如某一项是 1.0，其他是 0），这个点积结果会很大  
    #  只有当分布接近均匀分布时，这个 Loss 最小  
    overall_loss = torch.sum(tokens_per_expert * router_prob_per_expert.unsqueeze(0))  

    return overall_loss * num_experts

  3. 模态不平衡问题方案

    • 上述通过load_balancing_loss_func，分摊视觉 token 给多个专家。
    • 使用 fine-grained moe，即细粒度专家，同时也对应代表着具有较多的专家，可以处理大的 token patch；（可能会催生较多的 experts 擅长处理视觉信息，类似视觉专家组）
    • 一份样本的文本和视觉 token 数差异较大，视觉的信息通常会多的多，因此训练时可能会控制输入的数据成分。

路由专家

  1. 专家初始化：一次性申请所有专家的权重空间。
    • #   所有的专家参数并没有被存成 List[MLP]，而是存成了巨大的 Parameter Tensor  
#   形状为 (Num_Experts, Hidden, Dim)，方便进行并行矩阵运算  
self.gate_up_proj = nn.Parameter(torch.zeros(self.num_experts, self.hidden_size, 2 * self.expert_dim))  
self.down_proj = nn.Parameter(torch.empty((self.num_experts, self.expert_dim, self.hidden_size)))
  1. 路由环节：决定token的分配，核心类：Qwen3VLMoeTextSparseMoeBlock；计算每个 token 对每个专家的匹配分数，最后选出 Top-K
    • #   引用自提供的代码 Qwen3VLMoeTextSparseMoeBlock.forward  

def forward(self, hidden_states: torch.Tensor) -> torch.Tensor:  
    batch_size = hidden_states.shape[0]  
    #   [逻辑 1] 扁平化处理  
    #   MoE 路由是对每个 Token 独立进行的，所以将 (Batch, Seq, Dim) 展平成 (Batch*Seq, Dim)  
    hidden_states = hidden_states.reshape(-1, self.hidden_size)  

    #   [逻辑 2] 计算路由分值 (Router Logits)  
    #   self.gate 是一个 Linear 层: (Hidden_Size -> Num_Experts)  
    #   这一步算出每个 Token 对所有 60+ 个专家的原始打分  
    router_logits = self.gate(hidden_states)  

    #   [逻辑 3] 概率归一化  
    #   使用 Softmax 将打分转换为概率分布 (0~1 之间)  
    routing_weights = torch.nn.functional.softmax(router_logits, dim=-1, dtype=torch.float)  

    #   [逻辑 4] Top-K 选择 (核心稀疏逻辑)  
    #   比如 K=4，选出概率最高的 4 个专家的 权重(routing_weights) 和 索引(router_indices)  
    #   这一步决定了稀疏性，只有这几个专家会被激活  
    routing_weights, router_indices = torch.topk(routing_weights, self.top_k, dim=-1)  

    #   [逻辑 5] 权重再归一化 (Re-normalization)  
    #   将选出的 K 个专家的概率之和重新归一化为 1。  
    #   例如原概率是 [0.1, 0.2]，归一化后变成 [0.33, 0.66]  
    routing_weights = routing_weights / routing_weights.sum(dim=-1, keepdim=True)  
    routing_weights = routing_weights.to(router_logits.dtype)  

    #   [逻辑 6] 构建稀疏矩阵  
    #   创建一个全 0 矩阵，只在选中的 indices 位置填入归一化后的权重  
    router_weights = torch.zeros_like(router_logits).scatter_(1, router_indices, routing_weights)  

    #   [逻辑 7] 恢复形状并进入专家计算  
    hidden_states = hidden_states.reshape(batch_size, -1, self.hidden_size)  
    #   调用 Experts 层进行具体计算  
    routed_out = self.experts(hidden_states, router_weights, router_indices)  

    #   [整体逻辑总结]  
    #   这一段不仅做出了选择，还直接返回了 routed_out。  
    #   注意：此处没有 + shared_output 的操作。证明了它是 Pure MoE 架构。  
    return routed_out

    • 代码流程

      • image
      • image
      • tensor.shape 变化示例
        • image
      • image

  1. 专家计算环节：执行计算。核心类： Qwen3VLMoeTextExperts。根据路由结果，执行 MLP 计算

    • 前向传播

      • 训练模式：训练时 Batch Size 很大，如果把每个 Token 复制 K 份来做并行，显存会不够。所以采用 Loop (循环) 方式，一次只计算一个专家。

        • 其执行流程简单概括为找到非空闲专家（分配 token>0） --> 遍历非空闲专家 -> 取当前专家参数进行计算 -> 加权
        • if self.training:

      next_states = torch.zeros_like(hidden_states, dtype=hidden_states.dtype, device=hidden_states.device)

      #   [逻辑 1] 确定活跃专家

      #   并不是所有专家在当前 Batch 都会被用到，先找出有任务的专家 (expert_hit)

      #   避免对空任务的专家进行无效计算

      with torch.no_grad():

          expert_mask = torch.nn.functional.one_hot(router_indices, num_classes=self.num_experts)

          expert_mask = expert_mask.permute(2, 1, 0)

          expert_hit = torch.greater(expert_mask.sum(dim=(-1, -2)), 0).nonzero()

      #   [逻辑 2] 循环计算 (Serial Execution)

      for expert_idx in expert_hit[:]:

          #   [逻辑 2.1] 索引筛选

          #   找出当前专家 expert_idx 需要处理哪些 token (token_idx)

          with torch.no_grad():

              _, token_idx = torch.where(expert_mask[expert_idx[0]])

          #   只取出相关的 Token 数据

          current_state = hidden_states[token_idx]

          #   [逻辑 2.2] 专家参数提取与计算

          #   self.gate_up_proj[expert_idx] 利用切片取出当前专家的权重

          #   执行标准的 SwiGLU: (Gate * Sigmoid(Gate)) * Up

          gate_up = current_state @ self.gate_up_proj[expert_idx]

          gate, up = gate_up.chunk(2, dim=-1)

          gated_output = up * self.act_fn(gate)

          out = gated_output @ self.down_proj[expert_idx]

          #   [逻辑 2.3] 加权累加

          #   将计算结果乘以 Router 给的权重，加回到 next_states 对应的位置

          weighted_output = out[0] * routing_weights[token_idx, expert_idx, None]

          next_states.index_add_(0, token_idx, weighted_output.to(hidden_states.dtype))

    • 推理模式：推理时 Batch 小，延迟敏感。使用 torch.bmm (Batch Matrix Multiply) 实现全并行。

      • else:  
            #   [Step B1: 数据复制]  
            #   将输入 Token 复制 Num_Experts 份。  
            #   这一步显存开销大，但允许所有专家同时计算。  
            hidden_states = hidden_states.repeat(self.num_experts, 1)  
            hidden_states = hidden_states.view(self.num_experts, -1, self.hidden_size)  
            #   [Step B2: 并行矩阵乘法 (Batch MatMul)]  
            #   torch.bmm: [Experts, N, Hidden] x [Experts, Hidden, Inter]  
            #   一次性算出所有专家对所有数据的 Gate+Up 结果。  
            gate_up = torch.bmm(hidden_states, self.gate_up_proj)  
            gate, up = gate_up.chunk(2, dim=-1)  
            #   [Step B3: 激活与下投影]  
            next_states = torch.bmm((up * self.act_fn(gate)), self.down_proj)  
            #   [Step B4: 结果聚合]  
            #   此时 next_states 包含了所有专家对所有 Token 的结果。  
            #   我们只需要保留 Router 选中的那 K 个。  
            #   通过 routing_weights (包含大量的 0 和少量的 TopK 权重) 进行乘法。  
            next_states = next_states.reshape(self.num_experts, batch_size, -1, self.hidden_size)  
            next_states = (  
                next_states * routing_weights.transpose(0, 1).view(self.num_experts, batch_size, -1)[..., None]  
            )  
            #   Sum dim=0: 将所有专家的结果加起来 (没选中的专家权重是0，所以没影响)  
            next_states = next_states.sum(dim=0)  
        return next_states

  2. 专家负载均衡Loss实现

    • def load_balancing_loss_func(  
    gate_logits: Union[torch.Tensor, tuple[torch.Tensor], None],  
    num_experts: Optional[int] = None,  
    top_k=2,  
    attention_mask: Optional[torch.Tensor] = None,  
) -> Union[torch.Tensor, int]:  
    #   [Step 1: 算概率]  
    routing_weights = torch.nn.functional.softmax(concatenated_gate_logits, dim=-1)  

    #   [Step 2: 算实际分配情况 (Tokens per Expert)]  
    #   expert_mask 是 One-Hot，mean 后代表每个专家实际接了百分之多少的单  
    _, selected_experts = torch.topk(routing_weights, top_k, dim=-1)  
    expert_mask = torch.nn.functional.one_hot(selected_experts, num_experts)  
    tokens_per_expert = torch.mean(expert_mask.float(), dim=0)  

    #   [Step 3: 算预期分配情况 (Router Probability)]  
    #   Router 的输出概率均值，代表 Router "想" 给每个专家分多少单  
    router_prob_per_expert = torch.mean(routing_weights, dim=0)  

    #   [Step 4: 计算点积 Loss]  
    #   这是一个经典的 Aux Loss 公式 (来自 Switch Transformer 论文)  
    #   目标是让 tokens_per_expert 和 router_prob_per_expert 都是均匀分布 (1/N)  
    #   如果分布极度不均（有的专家极其忙，有的闲），这个乘积和会变大  
    overall_loss = torch.sum(tokens_per_expert * router_prob_per_expert.unsqueeze(0))  

    return overall_loss * num_experts  

if kwargs.get("output_router_logits", False):  
    #   计算 Aux Loss  
    aux_loss = load_balancing_loss_func(...)  
    #   加到总 Loss 里  
    if labels is not None:  
        loss += self.config.text_config.router_aux_loss_coef * aux_loss
      • Qwen3-VL 在训练时，强约束 Router 必须均衡分配任务。这对于多模态模型至关重要，否则 Router 极易把海量的视觉 Token 全部丢给少数几个专家，导致路由坍缩。

如果你对多模态大模型、强化学习、昇腾 NPU 部署、模型性能优化感兴趣，欢迎持续关注【AI模力圈】。

我们会持续更新：

  1. 多模态模型结构拆解
  2. 强化学习算法原理与实践
  3. 昇腾 NPU 迁移部署与踩坑复盘
  4. 模型训练与推理性能优化

图解版、速读版内容也会同步更新到公众号 / 小红书。

# transformer # 人工智能 # 深度学习 # AIGC
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

深入理解 AI Agent · MEMORY #02：记忆的工程机制

avatar AI Agent技术社区
cover

Windows部署OpenClaw：PowerShell安装、自定义模型与远程访问实测

avatar AI Agent技术社区
cover

DeepSeek V4 Pro正式版上线：AI Agent能力跃升背后的行业趋势

avatar AI Agent技术社区