|
23 | 23 |
|
24 | 24 | from mobius.components._common import Linear |
25 | 25 | from mobius.components._rms_norm import RMSNorm |
| 26 | +from mobius.components._vision import VisionLayerNorm |
26 | 27 |
|
27 | 28 | if TYPE_CHECKING: |
28 | 29 | import onnx_ir as ir |
@@ -117,6 +118,81 @@ def forward(self, op: OpBuilder, vision_features: ir.Value): |
117 | 118 | return hidden |
118 | 119 |
|
119 | 120 |
|
| 121 | +class Cosmos3EdgeMultiModalProjector(nn.Module): |
| 122 | + """Cosmos3-Edge pixel-shuffle merger projector. |
| 123 | +
|
| 124 | + ``LayerNorm → spatial 2x2 pixel-shuffle → Linear(fc1) → GELU → Linear(fc2)`` |
| 125 | +
|
| 126 | + The SigLIP vision encoder emits a fixed ``grid x grid`` patch grid |
| 127 | + (``num_patches`` patches, e.g. 16x16 = 256). ``use_postshuffle_norm=false`` |
| 128 | + means the ``LayerNorm`` is applied on the raw ``vision_hidden_size`` (1152) |
| 129 | + features **before** the spatial merge. The merge concatenates each |
| 130 | + ``spatial_merge_size x spatial_merge_size`` block of adjacent patches into a |
| 131 | + single ``spatial_merge_size**2 * vision_hidden_size`` (4608) vector, which |
| 132 | + ``linear_fc1`` maps to ``intermediate_size`` (11520) and ``linear_fc2`` maps |
| 133 | + to ``text_hidden_size`` (2048). |
| 134 | +
|
| 135 | + HF weights (``model.projector.*``): |
| 136 | + - ``norm.{weight,bias}`` (pre-shuffle LayerNorm) |
| 137 | + - ``linear_fc1.{weight,bias}`` |
| 138 | + - ``linear_fc2.{weight,bias}`` |
| 139 | + """ |
| 140 | + |
| 141 | + def __init__( |
| 142 | + self, |
| 143 | + vision_hidden_size: int, |
| 144 | + text_hidden_size: int, |
| 145 | + intermediate_size: int, |
| 146 | + grid_size: int, |
| 147 | + spatial_merge_size: int = 2, |
| 148 | + norm_eps: float = 1e-6, |
| 149 | + ): |
| 150 | + super().__init__() |
| 151 | + self._grid = grid_size |
| 152 | + self._ms = spatial_merge_size |
| 153 | + self._vision_hidden = vision_hidden_size |
| 154 | + merged_dim = vision_hidden_size * spatial_merge_size * spatial_merge_size |
| 155 | + # Pre-shuffle LayerNorm over the raw vision hidden size. |
| 156 | + self.norm = VisionLayerNorm(vision_hidden_size, eps=norm_eps) |
| 157 | + self.linear_fc1 = Linear(merged_dim, intermediate_size, bias=True) |
| 158 | + self.linear_fc2 = Linear(intermediate_size, text_hidden_size, bias=True) |
| 159 | + |
| 160 | + def forward(self, op: OpBuilder, vision_features: ir.Value): |
| 161 | + # vision_features: [batch, grid*grid, vision_hidden] |
| 162 | + ms = self._ms |
| 163 | + g = self._grid |
| 164 | + gm = g // ms |
| 165 | + d = self._vision_hidden |
| 166 | + |
| 167 | + # Pre-shuffle LayerNorm (use_postshuffle_norm=false). |
| 168 | + x = self.norm(op, vision_features) |
| 169 | + |
| 170 | + batch = op.Shape(vision_features, start=0, end=1) # dynamic [1] |
| 171 | + |
| 172 | + # [B, g*g, D] -> [B, g/ms, ms, g/ms, ms, D] |
| 173 | + shape_6d = op.Concat( |
| 174 | + batch, |
| 175 | + op.Constant(value_ints=[gm, ms, gm, ms, d]), |
| 176 | + axis=0, |
| 177 | + ) |
| 178 | + x = op.Reshape(x, shape_6d) |
| 179 | + # Group hidden dim outermost per merged block (HF F.unfold ordering): |
| 180 | + # [B, g/ms, ms, g/ms, ms, D] -> [B, g/ms, g/ms, D, ms, ms] |
| 181 | + x = op.Transpose(x, perm=[0, 1, 3, 5, 2, 4]) |
| 182 | + # Flatten to [B, (g/ms)^2, D*ms*ms] |
| 183 | + shape_3d = op.Concat( |
| 184 | + batch, |
| 185 | + op.Constant(value_ints=[gm * gm, d * ms * ms]), |
| 186 | + axis=0, |
| 187 | + ) |
| 188 | + x = op.Reshape(x, shape_3d) |
| 189 | + |
| 190 | + x = self.linear_fc1(op, x) |
| 191 | + x = op.Gelu(x) |
| 192 | + x = self.linear_fc2(op, x) |
| 193 | + return x |
| 194 | + |
| 195 | + |
120 | 196 | class LinearMultiModalProjector(nn.Module): |
121 | 197 | """Single linear projection (PaliGemma, Qwen2-Audio). |
122 | 198 |
|
|
0 commit comments