1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69
| import torch import torch.nn as nn import torch.nn.functional as F
class MultiHeadLatentAttention(nn.Module): def __init__(self, dim, n_heads, n_kv_heads, low_rank_dim): super().__init__() self.dim = dim self.n_heads = n_heads self.n_kv_heads = n_kv_heads self.head_dim = dim // n_heads self.low_rank_dim = low_rank_dim self.wq = nn.Linear(dim, dim, bias=False) self.wkv = nn.Linear(dim, low_rank_dim * 2, bias=False) self.wk = nn.Linear(low_rank_dim, n_kv_heads * self.head_dim, bias=False) self.wv = nn.Linear(low_rank_dim, n_kv_heads * self.head_dim, bias=False) self.wo = nn.Linear(dim, dim, bias=False) self.scale = self.head_dim ** -0.5 def forward(self, x, mask=None): batch_size, seq_len, _ = x.shape q = self.wq(x).view(batch_size, seq_len, self.n_heads, self.head_dim) q = q.transpose(1, 2) kv_low = self.wkv(x) k_low = kv_low[:, :, :self.low_rank_dim] v_low = kv_low[:, :, self.low_rank_dim:] k = self.wk(k_low).view(batch_size, seq_len, self.n_kv_heads, self.head_dim) v = self.wv(v_low).view(batch_size, seq_len, self.n_kv_heads, self.head_dim) k = k.transpose(1, 2) v = v.transpose(1, 2) if self.n_kv_heads != self.n_heads: n_rep = self.n_heads // self.n_kv_heads k = k.repeat_interleave(n_rep, dim=1) v = v.repeat_interleave(n_rep, dim=1) attn = torch.matmul(q, k.transpose(-2, -1)) * self.scale if mask is not None: attn = attn.masked_fill(mask == 0, float("-inf")) attn = F.softmax(attn, dim=-1) output = torch.matmul(attn, v) output = output.transpose(1, 2).contiguous() output = output.view(batch_size, seq_len, -1) return self.wo(output)
|