diff --git a/engine/50e8005addb244c88d30f9daccaabf8a.png b/engine/50e8005addb244c88d30f9daccaabf8a.png new file mode 100644 index 000000000..3faf8a260 Binary files /dev/null and b/engine/50e8005addb244c88d30f9daccaabf8a.png differ diff --git a/engine/BlockManager.py b/engine/BlockManager.py new file mode 100644 index 000000000..28ceed333 --- /dev/null +++ b/engine/BlockManager.py @@ -0,0 +1,56 @@ +from mindspore import tensor, Parameter, float32 +from mindspore.ops import ScatterNdUpdate, gather, stack, MatMul +from mindspore.common.initializer import initializer, Zero +import torch + +class BlockAllocator: + def __init__(self, num_blocks, block_size, head_dim): + self.num_blocks = num_blocks + self.pool = [] + for _ in range(num_blocks): + self.pool.append(Parameter(initializer(Zero(), [block_size, head_dim], float32))) + + def alloc(self, indices, A): + for index, a in list(zip(indices, A)): + self.pool[index[0]] = ScatterNdUpdate()(self.pool[index[0]], tensor([[index[1]]]), tensor(a.detach().numpy()).unsqueeze(0)) + + def get(self, indices): + A = [] + for index in indices: + A.append(gather(self.pool[index[0]], tensor(index[1]), 0)) + return stack(A) + +class MetadataEngine: + def __init__(self, num_blocks, num_heads, block_size): + self.num_blocks = num_blocks + self.num_heads = num_heads + self.block_size = block_size + self.block_table = [] + self.num_token = 0 + self.i = 0 + self.j = 0 + + def alloc(self): + list = [] + self.num_token = self.num_token + 1 + for _ in range(self.num_heads): + if self.i == self.num_blocks: + raise ValueError("Out of Memory") + list.append([self.i, self.j]) + self.j = self.j + 1 + if self.j == self.block_size: + self.i = self.i + 1 + self.j = 0 + self.block_table.append(list) + return list + + def get(self, x): + list = [] + for k in range(self.num_token): + list.append(self.block_table[k][x]) + return list + +class Multiplication: + def matmul(x, A, trans): + matmul = MatMul(transpose_b = trans) + return torch.from_numpy(matmul(tensor([x.detach().numpy()]), A).asnumpy()) \ No newline at end of file diff --git a/engine/Design and test.md b/engine/Design and test.md new file mode 100644 index 000000000..0a5e7ebf9 --- /dev/null +++ b/engine/Design and test.md @@ -0,0 +1,87 @@ +#### 一、技术调研 + +KV Cache 是一种用于加速 Transformer 等大型语言模型(LLM)自回归推理的关键优化技术,通常用于加速 Transformer 的 Decoder 部分,本次任务在 Decoder-only 结构的 Transformer 的基础之上,合理设计单机八卡 KV Cache 管理代码,对 Transformer 进行加速。 + +##### 1. Transformer 的多头注意力机制 + +在 Transformer 被提出的论文 《Attention is All You Need》 中详细介绍了多头注意力机制(Multi-Head Attention)层的基本结构。 + +一般而言,Transformer 可以对多批(batch)数据进行处理。在这里,我们认为多头注意力机制层的输入是 $b$ 个 $l$ 行 $d_m$ 列矩阵组成的矩阵组 $(\textbf X_0,\textbf X_1,\dots,\textbf X_{b-1})$,每个矩阵代表一批数据,矩阵的每一行代表数据的一个词元(token)。 + +首先,代码会对输入的数据进行“分头”操作。具体而言,是训练三组矩阵 $(\textbf W_{Q,0},\textbf W_{Q,1},\dots,\textbf W_{Q,h-1}),(\textbf W_{K,0},\textbf W_{K,1},\dots,\textbf W_{K,h-1}),(\textbf W_{V,0},\textbf W_{V,1},\dots,\textbf W_{V,h-1})$,其中第一组和第二组是 $d_m$ 行 $d_k$ 列矩阵,第三组是 $d_m$ 行 $d_v$ 列矩阵,并对每一个在 $[0,b)$ 内的自然数 $i$ 和 $[0,h)$ 的自然数 $j$,计算 + +$$\textbf Q_{ib+j}=\textbf X_i\textbf W_{Q,j}\\\textbf K_{ib+j}=\textbf X_i\textbf W_{K,j}\\\textbf V_{ib+j}=\textbf X_i\textbf W_{V,j}$$ + +接着,对每一个在 $[0,b)$ 内的自然数 $i$ 和 $[0,h)$ 的自然数 $j$,计算 $\textbf C_{i,j}=Att(\textbf Q_{ib+j},\textbf K_{ib+j},\textbf V_{ib+j},\textbf M)$,其中 $\textbf M$ 被称为掩码矩阵,对于掩码多头注意力层而言, $\textbf M$ 是主对角线上方(不包括主对角线)全为 $-\infty$,其余部分全为 $0$ 的矩阵,而函数 $Att(\textbf Q,\textbf K,\textbf V,\textbf M)$ 的表达式为 + +$$Att(\textbf Q,\textbf K,\textbf V,\textbf M)=softmax\left(\frac{\textbf Q\textbf K^T}{\sqrt{d_k}}+\textbf M\right)\textbf V$$ + +其中 $softmax(\textbf A)$ 的第 $i$ 行是行向量 $softmax(\textbf a_i)$,其中 $\textbf a_i$ 是 $\textbf A$ 的第 $i$ 行向量,而 + +$$softmax(a_0,a_1,\dots,a_m)=\frac{(e^{a_0},e^{a_1},\dots,e^{a_{m-1}})}{\overset{m-1}{\underset{i=0}\sum}e^{a_i}}$$ + +然后,生成 $b$ 个 $l$ 行 $h\times d_v$ 列矩阵 $\textbf O_{0},\textbf O_{1},\dots,\textbf O_{b-1}$,其中,矩阵 $\textbf O_{i}$ 的第 $j$ 行是分块矩阵 $[\textbf C_{i,0}\vdots \textbf C_{i,1}\vdots \cdots\vdots \textbf C_{i,h-1}]$ 的第 $j$ 行 + +最后,训练 $h\times d_v$ 行 $d$ 列矩阵 $\textbf W$ 并输出 $b$ 个 $l$ 行 $d$ 列矩阵组成的矩阵组 $(\textbf Y_0,\textbf Y_1,\dots,\textbf Y_{b-1})$,其中,对每一个在 $[0,b)$ 内的自然数 $i$,都有 $\textbf Y_i=\textbf O_i\textbf W$ + +##### 2. Decoder-only 结构下 Transformer 的优化可行性 + +《Attention is All You Need》 的 Transformer 是 Encoder-Decoder 的,而 Encoder 的掩码矩阵无法屏蔽未来的信息,所以这里直接使用 Decoder-only 结构下的 Transformer + +我们认为,一个句子是由若干词元(token)组成的,每个 token 可以赋予一个编码,句子就变成了一个以编码为字符的字符串。Decoder-only 结构下的 Transformer 以这样的 $b$ 批字符串组 $(s_0,s_1,\dots,s_{b-1})$ 为输入,其中每个字符串的长度都是 $l$,输出是 $b$ 个 $l$ 行 $v$ 列矩阵组成的矩阵组,第 $i(0\leq i + +其中 Input Embedding 层是用于将 token 编码转化为向量,该层训练定义域为 $\{0,1,\dots,v\}$,值域为 $d_m$ 维向量的函数 $Emb(x)$,其时间复杂度是线性的。Input Embedding 层输出 $b$ 个 $l$ 行 $d_m$ 列矩阵组成的矩阵组,矩阵组的第 $i(0\leq i -1: + x = x + self.pe[pos][:x.size(0), :] # x: [batch_size, d_model] + else: + x = x + self.pe[:x.size(0), :] # x: [tgt_len + 1, batch_size, d_model] + return self.dropout(x) + +class ScaledDotProductAttention(nn.Module): + def __init__(self, d_k): + super(ScaledDotProductAttention, self).__init__() + self.d_k=d_k + + def forward(self, Q, K, V, attn_mask): + scores = torch.matmul(Q, K.transpose(-1, -2)) / np.sqrt(self.d_k) # scores : [batch_size x n_heads x len_q(=len_k) x len_k(=len_q)] + scores.masked_fill_(attn_mask, -1e9) # Fills elements of self tensor with value where mask is one. + attn = nn.Softmax(dim=-1)(scores) + context = torch.matmul(attn, V) + return context + +class MultiHeadAttention(nn.Module): + def __init__(self, d_model, d_k, d_v, n_heads): + super(MultiHeadAttention, self).__init__() + self.W_Q = nn.Linear(d_model, d_k * n_heads) + self.W_K = nn.Linear(d_model, d_k * n_heads) + self.W_V = nn.Linear(d_model, d_v * n_heads) + self.linear = nn.Linear(n_heads * d_v, d_model) + self.layer_norm = nn.LayerNorm(d_model) + self.d_model = d_model + self.d_k = d_k + self.d_v = d_v + self.n_heads = n_heads + + def forward(self, Q, batch_size, attn_mask, k_cache = None, v_cache = None): + residual = Q + if attn_mask is None: + q_s = self.W_Q(Q).view(batch_size, self.n_heads, self.d_k).flatten(0,1) # q_s: [batch_size x n_heads, d_k] + k_s = self.W_K(Q).view(batch_size, self.n_heads, self.d_k).flatten(0,1) # k_s: [batch_size x n_heads, d_k] + v_s = self.W_V(Q).view(batch_size, self.n_heads, self.d_v).flatten(0,1) # v_s: [batch_size x n_heads, d_v] + k_cache.write(k_s) + v_cache.write(v_s) + context = k_cache.transmatmul(q_s) + context = v_cache.matmul(nn.Softmax(dim=-1)(context)) # context: [batch_size x n_heads, d_v] + context = context.contiguous().view(batch_size, -1) # context: [batch_size, n_heads x d_v] + else: + q_s = self.W_Q(Q).view(batch_size, -1, self.n_heads, self.d_k).transpose(1,2) # q_s: [batch_size, n_heads, len_q, d_k] + k_s = self.W_K(Q).view(batch_size, -1, self.n_heads, self.d_k).transpose(1,2) # k_s: [batch_size, n_heads, len_q, d_k] + v_s = self.W_V(Q).view(batch_size, -1, self.n_heads, self.d_v).transpose(1,2) # v_s: [batch_size, n_heads, len_q, d_v] + + attn_mask = attn_mask.unsqueeze(1).repeat(1, self.n_heads, 1, 1) # attn_mask : [batch_size, n_heads, len_q, len_k] + + context = ScaledDotProductAttention(self.d_k)(q_s, k_s, v_s, attn_mask) # context: [batch_size x n_heads x len_q x d_v] + context = context.transpose(1, 2).contiguous().view(batch_size, -1, self.n_heads * self.d_v) # context: [batch_size, len_q, n_heads x d_v] + output = self.linear(context) + return self.layer_norm(output + residual) # output: [batch_size, len_q, d_model] + +class PoswiseFeedForwardNet(nn.Module): + def __init__(self, d_model, d_ff): + super(PoswiseFeedForwardNet, self).__init__() + self.conv1 = nn.Conv1d(in_channels=d_model, out_channels=d_ff, kernel_size=1) + self.conv2 = nn.Conv1d(in_channels=d_ff, out_channels=d_model, kernel_size=1) + self.layer_norm = nn.LayerNorm(d_model) + + def forward(self, inputs): + residual = inputs # inputs : [batch_size, len_q, d_model] + output = nn.ReLU()(self.conv1(inputs.transpose(len(inputs.size()) - 2, len(inputs.size()) - 1))) + output = self.conv2(output).transpose(len(inputs.size()) - 2, len(inputs.size()) - 1) + return self.layer_norm(output + residual) + +class DecoderLayer(nn.Module): + def __init__(self, tgt_len, d_model, d_ff, d_k, d_v, n_heads): + super(DecoderLayer, self).__init__() + self.dec_enc_attn = MultiHeadAttention(d_model, d_k, d_v, n_heads) + self.pos_ffn = PoswiseFeedForwardNet(d_model, d_ff) + self.n_heads = n_heads + self.tgt_len = tgt_len + self.d_k = d_k + self.d_v = d_v + + def forward(self, dec_inputs, batch_size, dec_enc_attn_mask, block_size, k_cache, v_cache): + if block_size > 0: + dec_outputs = self.dec_enc_attn(dec_inputs, batch_size, None, k_cache, v_cache) + else: + dec_outputs = self.dec_enc_attn(dec_inputs, batch_size, dec_enc_attn_mask) + dec_outputs = self.pos_ffn(dec_outputs) + return dec_outputs + +class Decoder(nn.Module): + def __init__(self, tgt_vocab_size, tgt_len, d_model, d_ff, d_k, d_v, n_layers, n_heads): + super(Decoder, self).__init__() + self.tgt_emb = nn.Embedding(tgt_vocab_size, d_model) + self.pos_emb = PositionalEncoding(tgt_len + 1, d_model) + self.layers = nn.ModuleList([DecoderLayer(tgt_len, d_model, d_ff, d_k, d_v, n_heads) for _ in range(n_layers)]) + + def forward(self, dec_inputs, batch_size, k_caches, v_caches, block_size = 0, pos = -1): # dec_inputs : [batch_size] or [batch_size, tgt_len] + dec_outputs = self.tgt_emb(dec_inputs) + if block_size > 0: + dec_outputs = self.pos_emb(dec_outputs, pos) # [batch_size, d_model] + dec_enc_attn_mask = None + else: + dec_outputs = self.pos_emb(dec_outputs.transpose(0, 1), pos).transpose(0, 1) # [batch_size, tgt_len, d_model] + dec_enc_attn_mask = get_attn_pad_mask(dec_inputs) + + for layer, k_cache, v_cache in list(zip(self.layers, k_caches, v_caches)): + dec_outputs = layer(dec_outputs, batch_size, dec_enc_attn_mask, block_size, k_cache, v_cache) + return dec_outputs + +class Transformer(nn.Module): + def __init__(self, tgt_vocab_size, tgt_len = 10, d_model = 512, d_ff = 2048, d_k = 64, d_v = 64, n_layers = 6, n_heads = 8): + super(Transformer, self).__init__() + self.decoder = Decoder(tgt_vocab_size, tgt_len, d_model, d_ff, d_k, d_v, n_layers, n_heads) + self.projection = nn.Linear(d_model, tgt_vocab_size, bias=False) + self.tgt_vocab_size = tgt_vocab_size + self.d_model = d_model + self.tgt_len = tgt_len + self.n_layers = n_layers + self.n_heads = n_heads + self.d_k = d_k + self.d_v = d_v + def forward(self, dec_inputs, block_size = 0): + batch_size = dec_inputs.size()[0] + if block_size > 0: + k_caches = [] + v_caches = [] + dec_inputs = dec_inputs.transpose(0, 1) + for i in range(self.n_layers): + k_caches.append(Cache(math.ceil(batch_size * len(dec_inputs) * self.n_heads / block_size), batch_size * self.n_heads, block_size, self.d_k)) + v_caches.append(Cache(math.ceil(batch_size * len(dec_inputs) * self.n_heads / block_size), batch_size * self.n_heads, block_size, self.d_v)) + dec_output = torch.empty(self.tgt_vocab_size, self.d_model) + dec_outputs = torch.empty(batch_size, 0, self.d_model) + pos = 0 + dec_logits = torch.empty(batch_size, 0, self.tgt_vocab_size) + for dec_input in dec_inputs: + for i in range(0, dec_input.shape[0]): + if dec_input[i] == 0: + dec_input[i] = dec_logit[:, :self.tgt_vocab_size - 1].data.max(1, keepdim=True).indices[i] + dec_output = self.decoder(dec_input, batch_size, k_caches, v_caches, block_size, pos) + dec_logit = self.projection(dec_output) # dec_logits : [batch_size, tgt_vocab_size] + pos = pos + 1 + dec_logits = torch.cat([dec_logits, dec_logit.unsqueeze(1)], 1) + for i in range(self.n_layers): + k_caches[i].delete() + v_caches[i].delete() + else: + k_caches = [None] * self.n_layers + v_caches = [None] * self.n_layers + dec_outputs = self.decoder(dec_inputs, batch_size, k_caches, v_caches) + dec_logits = self.projection(dec_outputs) # dec_logits : [batch_size, tgt_len, tgt_vocab_size] + return dec_logits \ No newline at end of file diff --git a/engine/test_kv_cache.py b/engine/test_kv_cache.py new file mode 100644 index 000000000..a90c2315a --- /dev/null +++ b/engine/test_kv_cache.py @@ -0,0 +1,51 @@ +from torch import nn, optim, LongTensor +from TransformerDecoderOnly import Transformer + +def make_training_batch(sentences): + output_batch = [] + target_batch = [] + for sentence in sentences: + output_batch.append([tgt_vocab[n] for n in sentence[0].split()]) + target_batch.append([tgt_vocab[n] for n in sentence[1].split()]) + return LongTensor(output_batch), LongTensor(target_batch) + +def make_learning_batch(sentences): + output_batch = [] + for sentence in sentences: + output_batch.append([tgt_vocab[n] for n in sentence.split()]) + return LongTensor(output_batch) + +if __name__ == '__main__': + + sentences = [['S i want a beer', 'i want a beer E'], ['S i want a beer', 'i want a beer E']] + + # Transformer Parameters + # Padding Should be 0 + # Unknown Should be tgt_vocab_size - 1 + tgt_vocab = {'P': 0, 'i': 1, 'want': 2, 'a': 3, 'beer': 4, 'S': 5, 'E': 6, 'U': 7} + + number_dict = {i: w for i, w in enumerate(tgt_vocab)} + + model = Transformer(len(tgt_vocab)) + + criterion = nn.CrossEntropyLoss() + optimizer = optim.Adam(model.parameters(), lr=0.001) + + dec_inputs, target_batch = make_training_batch(sentences) + + for epoch in range(20): + optimizer.zero_grad() + outputs = model(dec_inputs) + loss = criterion(outputs.view(-1, outputs.size(-1)), target_batch.contiguous().view(-1)) + print('Epoch:', '%04d' % (epoch + 1), 'cost =', '{:.6f}'.format(loss)) + loss.backward() + optimizer.step() + + # Test + sentences = ['S P P P P'] * 10 + dec_inputs = make_learning_batch(sentences) + for epoch in range(1): + predict = model(dec_inputs, 16) + predict = predict[:, :, :len(tgt_vocab) - 1].data.max(2, keepdim=True)[1] + for sentence, output in list(zip(sentences, predict)): + print(sentence, '->', [number_dict[n.item()] for n in output.squeeze()]) \ No newline at end of file