attention_pool.py 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116
  1. from typing import Optional, Type
  2. import torch
  3. import torch.nn as nn
  4. import torch.nn.functional as F
  5. from .attention import maybe_add_mask
  6. from .config import use_fused_attn
  7. from .mlp import Mlp
  8. from .weight_init import trunc_normal_tf_
  9. class AttentionPoolLatent(nn.Module):
  10. """ Attention pooling w/ latent query
  11. """
  12. fused_attn: torch.jit.Final[bool]
  13. def __init__(
  14. self,
  15. in_features: int,
  16. out_features: int = None,
  17. embed_dim: int = None,
  18. num_heads: int = 8,
  19. feat_size: Optional[int] = None,
  20. mlp_ratio: float = 4.0,
  21. qkv_bias: bool = True,
  22. qk_norm: bool = False,
  23. latent_len: int = 1,
  24. latent_dim: int = None,
  25. pos_embed: str = '',
  26. pool_type: str = 'token',
  27. norm_layer: Optional[Type[nn.Module]] = None,
  28. act_layer: Optional[Type[nn.Module]] = nn.GELU,
  29. drop: float = 0.0,
  30. device = None,
  31. dtype = None
  32. ):
  33. dd = {'device': device, 'dtype': dtype}
  34. super().__init__()
  35. embed_dim = embed_dim or in_features
  36. out_features = out_features or in_features
  37. assert embed_dim % num_heads == 0
  38. self.num_heads = num_heads
  39. self.head_dim = embed_dim // num_heads
  40. self.feat_size = feat_size
  41. self.scale = self.head_dim ** -0.5
  42. self.pool = pool_type
  43. self.fused_attn = use_fused_attn()
  44. if pos_embed == 'abs':
  45. assert feat_size is not None
  46. self.pos_embed = nn.Parameter(torch.zeros(feat_size, in_features, **dd))
  47. else:
  48. self.pos_embed = None
  49. self.latent_dim = latent_dim or embed_dim
  50. self.latent_len = latent_len
  51. self.latent = nn.Parameter(torch.zeros(1, self.latent_len, embed_dim, **dd))
  52. self.q = nn.Linear(embed_dim, embed_dim, bias=qkv_bias, **dd)
  53. self.kv = nn.Linear(embed_dim, embed_dim * 2, bias=qkv_bias, **dd)
  54. if qk_norm:
  55. qk_norm_layer = norm_layer or nn.LayerNorm
  56. self.q_norm = qk_norm_layer(self.head_dim, **dd)
  57. self.k_norm = qk_norm_layer(self.head_dim, **dd)
  58. else:
  59. self.q_norm = nn.Identity()
  60. self.k_norm = nn.Identity()
  61. self.proj = nn.Linear(embed_dim, embed_dim, **dd)
  62. self.proj_drop = nn.Dropout(drop)
  63. self.norm = norm_layer(out_features, **dd) if norm_layer is not None else nn.Identity()
  64. self.mlp = Mlp(embed_dim, int(embed_dim * mlp_ratio), act_layer=act_layer, **dd)
  65. self.init_weights()
  66. def init_weights(self):
  67. if self.pos_embed is not None:
  68. trunc_normal_tf_(self.pos_embed, std=self.pos_embed.shape[1] ** -0.5)
  69. trunc_normal_tf_(self.latent, std=self.latent_dim ** -0.5)
  70. def forward(self, x, attn_mask: Optional[torch.Tensor] = None):
  71. B, N, C = x.shape
  72. if self.pos_embed is not None:
  73. # FIXME interpolate
  74. x = x + self.pos_embed.unsqueeze(0).to(x.dtype)
  75. q_latent = self.latent.expand(B, -1, -1)
  76. q = self.q(q_latent).reshape(B, self.latent_len, self.num_heads, self.head_dim).transpose(1, 2)
  77. kv = self.kv(x).reshape(B, N, 2, self.num_heads, self.head_dim).permute(2, 0, 3, 1, 4)
  78. k, v = kv.unbind(0)
  79. q, k = self.q_norm(q), self.k_norm(k)
  80. if self.fused_attn:
  81. x = F.scaled_dot_product_attention(q, k, v, attn_mask=attn_mask)
  82. else:
  83. q = q * self.scale
  84. attn = q @ k.transpose(-2, -1)
  85. attn = maybe_add_mask(attn, attn_mask)
  86. attn = attn.softmax(dim=-1)
  87. x = attn @ v
  88. x = x.transpose(1, 2).reshape(B, self.latent_len, C)
  89. x = self.proj(x)
  90. x = self.proj_drop(x)
  91. x = x + self.mlp(self.norm(x))
  92. # optional pool if latent seq_len > 1 and pooled output is desired
  93. if self.pool == 'token':
  94. x = x[:, 0]
  95. elif self.pool == 'avg':
  96. x = x.mean(1)
  97. return x