(self, inputs)
| 148 | # Must NOT use autocast here or model output is so degraded it can't gen a coherent image. |
| 149 | @torch.compiler.disable |
| 150 | def forward(self, inputs): |
| 151 | for item in inputs: |
| 152 | if torch.is_floating_point(item): |
| 153 | item.requires_grad_(True) |
| 154 | x_chunk, timesteps, context_chunk, attn_mask_chunk = inputs |
| 155 | t_chunk = 1.0 - timesteps |
| 156 | bs, c, gh, gw = x_chunk.shape |
| 157 | |
| 158 | # This is only the conditional pathway |
| 159 | B = x_chunk.shape[0] |
| 160 | device = x_chunk.device |
| 161 | img_tokens = self._img_to_tokens(x_chunk) |
| 162 | L_img = img_tokens.shape[1] |
| 163 | L_text = context_chunk.shape[1] |
| 164 | L = L_text + L_img |
| 165 | latent_dim = img_tokens.shape[-1] |
| 166 | |
| 167 | x_full = torch.zeros(B, L, latent_dim, dtype=img_tokens.dtype, device=device) |
| 168 | x_full[:, L_text:] = img_tokens |
| 169 | |
| 170 | text_pos = torch.arange(L_text, device=device).view(-1, 1).expand(L_text, 3) |
| 171 | img_pos = self._image_position_ids(gh, gw, device) |
| 172 | position_ids = torch.cat([text_pos, img_pos], dim=0).unsqueeze(0).expand(B, L, 3) |
| 173 | |
| 174 | indicator = torch.empty(B, L, dtype=torch.long, device=device) |
| 175 | indicator[:, :L_text] = LLM_TOKEN_INDICATOR |
| 176 | indicator[:, L_text:] = OUTPUT_IMAGE_INDICATOR |
| 177 | |
| 178 | segment_ids = torch.ones(B, L, dtype=torch.long, device=device) |
| 179 | pad = (attn_mask_chunk == 0) |
| 180 | segment_ids[:, :L_text][pad] = SEQUENCE_PADDING_INDICATOR |
| 181 | indicator[:, :L_text][pad] = 0 |
| 182 | # Block-diagonal mask from segment ids: (B, 1, L, L), True = attend. |
| 183 | attn_mask = (segment_ids.unsqueeze(2) == segment_ids.unsqueeze(1)).unsqueeze(1) |
| 184 | |
| 185 | # backbone |
| 186 | llm_features = context_chunk |
| 187 | x = x_full |
| 188 | t = t_chunk |
| 189 | |
| 190 | indicator = indicator.to(torch.long) |
| 191 | output_image_mask = (indicator == OUTPUT_IMAGE_INDICATOR).to(x.dtype).unsqueeze(-1) |
| 192 | |
| 193 | x = x * output_image_mask |
| 194 | h = self.input_proj(x) * output_image_mask |
| 195 | |
| 196 | t_cond = self.t_embedding(t, dtype=x.dtype) |
| 197 | if t.dim() == 1: |
| 198 | t_cond = t_cond.unsqueeze(1) |
| 199 | adaln_input = F.silu(self.adaln_proj(t_cond)) |
| 200 | |
| 201 | # h is zero on the text rows (content lives only on image rows), add writes the text features in place |
| 202 | if llm_features is not None: |
| 203 | L_text = llm_features.shape[1] |
| 204 | text_mask = (indicator[:, :L_text] == LLM_TOKEN_INDICATOR).to(x.dtype).unsqueeze(-1) |
| 205 | llm = self.llm_cond_norm(llm_features * text_mask) |
| 206 | llm = self.llm_cond_proj(llm) * text_mask |
| 207 | h[:, :L_text] = h[:, :L_text] + llm |
nothing calls this directly
no test coverage detected