| 332 | |
| 333 | |
| 334 | class WanI2VCrossAttention(WanSelfAttention): |
| 335 | |
| 336 | def __init__(self, |
| 337 | dim, |
| 338 | num_heads, |
| 339 | window_size=(-1, -1), |
| 340 | qk_norm=True, |
| 341 | eps=1e-6): |
| 342 | super().__init__(dim, num_heads, window_size, qk_norm, eps) |
| 343 | |
| 344 | self.k_img = nn.Linear(dim, dim) |
| 345 | self.v_img = nn.Linear(dim, dim) |
| 346 | # self.alpha = nn.Parameter(torch.zeros((1, ))) |
| 347 | self.norm_k_img = WanRMSNorm( |
| 348 | dim, eps=eps) if qk_norm else nn.Identity() |
| 349 | |
| 350 | def forward(self, x, context, context_lens): |
| 351 | """ |
| 352 | x: [B, L1, C]. |
| 353 | context: [B, L2, C]. |
| 354 | context_lens: [B]. |
| 355 | """ |
| 356 | context_img = context[:, :257] |
| 357 | context = context[:, 257:] |
| 358 | b, n, d = x.size(0), self.num_heads, self.head_dim |
| 359 | |
| 360 | # compute query, key, value |
| 361 | q = self.norm_q(self.q(x)).view(b, -1, n, d) |
| 362 | k = self.norm_k(self.k(context)).view(b, -1, n, d) |
| 363 | v = self.v(context).view(b, -1, n, d) |
| 364 | k_img = self.norm_k_img(self.k_img(context_img)).view(b, -1, n, d) |
| 365 | v_img = self.v_img(context_img).view(b, -1, n, d) |
| 366 | img_x = flash_attention(q, k_img, v_img, k_lens=None) |
| 367 | # compute attention |
| 368 | x = flash_attention(q, k, v, k_lens=context_lens) |
| 369 | |
| 370 | # output |
| 371 | x = x.flatten(2) |
| 372 | img_x = img_x.flatten(2) |
| 373 | x = x + img_x |
| 374 | x = self.o(x) |
| 375 | return x |
| 376 | |
| 377 | |
| 378 | WANX_CROSSATTENTION_CLASSES = { |
nothing calls this directly
no outgoing calls
no test coverage detected