(self, x: Tensor, vec: Tensor, pe: Tensor, mask: Tensor = None)
| 331 | self.backend = backend |
| 332 | |
| 333 | def forward(self, x: Tensor, vec: Tensor, pe: Tensor, mask: Tensor = None) -> Tensor: |
| 334 | mod, _ = self.modulation(vec) |
| 335 | x_mod = (1 + mod.scale) * self.pre_norm(x) + mod.shift |
| 336 | qkv, mlp = torch.split(self.linear1(x_mod), [3 * self.hidden_size, self.mlp_hidden_dim], dim=-1) |
| 337 | |
| 338 | q, k, v = rearrange(qkv, "B L (K H D) -> K B H L D", K=3, H=self.num_heads) |
| 339 | q, k = self.norm(q, k, v) |
| 340 | if mask is not None: |
| 341 | mask = repeat(mask, 'B L S-> B H L S', H=self.num_heads) |
| 342 | # compute attention |
| 343 | attn = attention(q, k, v, pe=pe, mask = mask, backend=self.backend) |
| 344 | # compute activation in mlp stream, cat again and run second linear layer |
| 345 | output = self.linear2(torch.cat((attn, self.mlp_act(mlp)), 2)) |
| 346 | return x + mod.gate * output |
| 347 | |
| 348 | |
| 349 | class DoubleStreamBlockC(DoubleStreamBlock): |
nothing calls this directly
no test coverage detected