PoliBa-DeepLearning/NanoSocrates
2
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'dev.embedder' of https://repositories.communitynotfound.work/PoliBa-DeepLearning/NanoSocrates into dev.embedder

Browse Source
This commit is contained in:
GassiGiuseppe
2025-10-07 17:42:21 +02:00
parent b4ee8362a2 3021a51961
commit f801afe0e4
13 changed files with 668 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
Project_Model/Libs/BPE/Enums/SpecialToken.py
View File

@@ -10,7 +10,6 @@ class SpecialToken(Enum):
RELATIONSHIP = "<PRED>"
OBJECT = "<OBJ>"
ABSTRACT = "<ABS>"
CORPUS_END = "<END>"
## Tasks' Token
RDF_TO_TEXT = "<RDF2TXT>"
@@ -20,4 +19,6 @@ class SpecialToken(Enum):
# BPE Training:
# NanoSocrates
START = "<START>"
START = "<START>"
CORPUS_END = "<END>"
PAD = "<PAD>"

19
Project_Model/Libs/Transformer/Classes/DeToken.py Normal file
View File

@@ -0,0 +1,19 @@
import torch
class DeToken(torch.nn.Module):
def __init__(self, embedding_size: int, vocabulary_size: int) -> None:
super().__init__()
self.__linear = torch.nn.Linear(embedding_size, vocabulary_size)
def forward(self, x: torch.Tensor) -> torch.Tensor:
# 1) Go from latent space to vocabularu space
x = self.__linear(x)
# 2) Go to logits
x = torch.softmax(x, 2)
return x

26
Project_Model/Libs/Transformer/Classes/Decoder.py
View File

@@ -35,22 +35,28 @@ class Decoder(nn.Module):
)
self.__layer_norm_3 = nn.LayerNorm(embedding_dimension)
def forward(self, x, k_x, v_x, padding_mask = None): #-> list[torch.Tensor]: # k_x = v_x . While x_q = x
def forward(
self,
args: tuple[
torch.Tensor,
torch.Tensor,
torch.Tensor,
torch.Tensor
]
): # -> list[torch.Tensor]: # k_x = v_x . While x_q = x
# WARNING: args is needed to have sequential
x, k_x, v_x, padding_mask = args
# build of attention mask
attention_mask = get_causal_attention_mask(x.size(1))
# 1) Masked Attention
MASKED_ATTENTION = self.__masked_attention(
x, x, x, key_padding_mask=padding_mask, attn_mask=attention_mask
x, x, x, key_padding_mask=padding_mask, attention_mask=attention_mask
)
# 2) Dropout
DROPPED_MASKED_ATTENTION = self.__dropout(
MASKED_ATTENTION
)
DROPPED_MASKED_ATTENTION = self.__dropout(MASKED_ATTENTION)
del MASKED_ATTENTION
# 3) Residual Connection
@@ -61,7 +67,9 @@ class Decoder(nn.Module):
x = self.__layer_norm_1(x)
# 5) Encoder–decoder (cross) attention
CROSS_ATTENTION = self.__cross_attention(x, k_x, v_x, key_padding_mask=padding_mask)
CROSS_ATTENTION = self.__cross_attention(
x, k_x, v_x, key_padding_mask=padding_mask
)
# 6) Dropout
DROPPED_CROSS_ATTENTION = self.__dropout(CROSS_ATTENTION)
@@ -88,7 +96,7 @@ class Decoder(nn.Module):
# 12) Layer Normalization
x = self.__layer_norm_3(x)
return x, k_x, v_x, padding_mask
return (x, k_x, v_x, padding_mask)
# use eval to disable dropout ecc

12
Project_Model/Libs/Transformer/Classes/Encoder.py
View File

@@ -1,3 +1,4 @@
import torch
import torch.nn as nn
from Project_Model.Libs.Transformer.Classes.FeedForwardNetwork import FeedForwardNetwork
from Project_Model.Libs.Transformer.Classes.TorchMultiHeadAttention import (
@@ -29,14 +30,17 @@ class Encoder(
embedding_dimension
) # norm of second "Add and Normalize"
self.__dropout = nn.Dropout(0.1) # ...
pass
def forward(self, x, padding_mask = None):
def forward(self, args: tuple[torch.Tensor, torch.Tensor]):
# WARNING: args is needed to have sequential
x, padding_mask = args
# -> ATTENTION -> dropout -> add and normalize -> FF -> dropout -> add and normalize ->
# Attention with Residual Connection [ input + self-attention]
# 1) Multi Head Attention
ATTENTION = self.__attention(x, x, x,key_padding_mask= padding_mask)
ATTENTION = self.__attention(x, x, x, key_padding_mask=padding_mask)
# 2) Dropout
DROPPED_ATTENTION = self.__dropout(ATTENTION)
@@ -62,7 +66,7 @@ class Encoder(
# 8) Layer Normalization
x = self.__layer_norm_2(x)
return x,padding_mask
return (x, padding_mask)
# use eval to disable dropout ecc

24
Project_Model/Libs/Transformer/Classes/MultiHeadAttention.py
View File

@@ -1,24 +0,0 @@
# multi-head attention -> (then to) ff
# attention: qkv -> score = qk -> divide -> softamx
# multihead -> QKV diferent in each head ( built by : X*[WQ/QK/WV])
# z = soft(Q*K'/sqr(d))*V
# recombine Z: 1) concatenate. 2) [z01234] * W = Z
# we expect later to have padding token
########################
# WIP
########################
import torch.nn as nn
embed_dim = 256
num_heads = 8
multihead_attn = nn.MultiheadAttention(embed_dim, num_heads)
class MultiheadAttention:
def __init__(
self,
num_heads=8,
) -> None:
pass

129
Project_Model/Libs/Transformer/Classes/NanoSocratesCore.py
View File

@@ -4,55 +4,108 @@ from .Encoder import Encoder
from ....Libs.Embedder import NanoSocratesEmbedder
import torch
class NanoSocratesCore(torch.nn.Module):
def __init__(self,
embedded_size: int,
feed_forward_dim: int,
encoder_layers: int,
decoder_layers:int,
attention_heads: int,
vocab_size: int) -> None:
def __init__(
self,
sentence_length: int,
vocab_size: int,
embedding_size: int = 256,
feed_forward_multiplier: int = 4,
num_encoder_layers: int = 2,
num_decoder_layers: int = 2,
num_attention_heads: int = 4,
) -> None:
feed_forward_dim = embedding_size * feed_forward_multiplier
self.__sentence_length = sentence_length
self.__encoder_sequence = torch.nn.Sequential(
*[Encoder(embedded_size, feed_forward_dim, attention_heads) for _ in range(encoder_layers)]
)
#* unpack the list so that each encoder has its own weights
*[
Encoder(embedding_size, feed_forward_dim, num_attention_heads)
for _ in range(num_encoder_layers)
]
)
# * unpack the list so that each encoder has its own weights
self.__decoder_sequence = torch.nn.Sequential(
*[Decoder(embedded_size, feed_forward_dim, attention_heads) for _ in range(decoder_layers)]
*[
Decoder(embedding_size, feed_forward_dim, num_attention_heads)
for _ in range(num_decoder_layers)
]
)
self.__linear = torch.nn.Linear(embedding_size, vocab_size)
self.__input_embeder = NanoSocratesEmbedder(vocab_size, embedding_size)
self.__output_embedder = NanoSocratesEmbedder(vocab_size, embedding_size)
def forward(
self,
encoder_input: list[list[int]],
decoder_input: list[list[int]],
encoder_padding_mask: list[list[int]],
):
if len(encoder_padding_mask) != len(encoder_input):
raise Exception("Mismatch in received_dimensions")
# TODO: check for tensor in input to embedder
# 1) Embed User-Input for encoders
ENCODER_INPUT = self.__input_embeder(encoder_input)
# 2) Encode User-Input
ENCODER_OUTPUT, _ = self.__encoder_sequence(ENCODER_INPUT, encoder_padding_mask)
del ENCODER_INPUT
exit_loop = False
decoder_token_list = decoder_input[:]
decoder_phase = 0
LOGITS_HISTORY: list[torch.Tensor] = []
# 3) Autoregressive Output
while not exit_loop:
# 3.0) Increment Counter
decoder_phase += 1
# 3.1) Embed Decoder Input
decoder_input = self.__output_embedder(decoder_token_list)
# 3.2) Decode Decoder Input
DECODER_OUTPUT, _, _, _ = self.__decoder_sequence(
decoder_input, ENCODER_OUTPUT, ENCODER_OUTPUT
)
self.__linear = torch.nn.Linear(embedded_size, vocab_size, bias=False)
self.__input_embeder = NanoSocratesEmbedder(vocab_size,embedded_size)
self.__output_embedder = NanoSocratesEmbedder(vocab_size,embedded_size)
# 3.3) Go back to Token space
# TODO: change name
LOGITS = self.__linear(DECODER_OUTPUT)
del DECODER_OUTPUT
# 3.4) Transform in probabilities
# TODO: change name
TOKEN_PROBABILITIES = torch.softmax(LOGITS, dim=-1)
del LOGITS
def forward(self, token_list, padding_mask = None):
x = self.__input_embeder(token_list)
x = self.__encoder_sequence(x, padding_mask)[0]
LOGITS_HISTORY.append(TOKEN_PROBABILITIES)
# 3.5) Take most probable tokens
TOKEN_IDS = torch.argmax(TOKEN_PROBABILITIES, -1)
# do while
x = self.__decoder_sequence(x,x,x, padding_mask)[0]
logits = self.__linear(x)
log_prob = torch.softmax(logits, dim=-1)
output = torch.argmax(log_prob)
while self.keep_going(log_prob):
# from log_prob again into x
# TODO: check for dimensions and for efficiency
DECODER_TOKEN_TENSOR = torch.tensor(decoder_token_list)
DECODER_TOKEN_TENSOR[:, decoder_phase] = TOKEN_IDS
decoder_token_list = DECODER_TOKEN_TENSOR.tolist()
del TOKEN_IDS
del DECODER_TOKEN_TENSOR
x = self.__decoder_sequence(x,x,x, padding_mask)[0]
logits = self.__linear(x)
log_prob = torch.softmax(logits, dim=-1)
# argmax
return log_prob
def keep_going(self, x: ) -> bool:
# 3.6) Check if we generated all tokens
if decoder_phase == self.__sentence_length - 1:
exit_loop = True
return LOGITS_HISTORY

7
Project_Model/Libs/Transformer/Classes/TorchMultiHeadAttention.py
View File

@@ -8,17 +8,16 @@ class TorchMultiHeadAttention(nn.Module):
self,
embedding_dimension: int,
number_of_attention_heads: int,
dropout: float = 0.0,
dropout: float = 0.0
):
super().__init__()
self.attention = nn.MultiheadAttention(
self.attention = torch.nn.MultiheadAttention(
embedding_dimension,
number_of_attention_heads,
num_heads=number_of_attention_heads,
dropout=dropout,
batch_first=True,
)
def forward(
self,
x_q: torch.Tensor,

7
Project_Model/Libs/Transformer/Classes/__init__.py
View File

@@ -1,15 +1,16 @@
from .Decoder import Decoder
from .Encoder import Encoder
from .FeedForwardNetwork import FeedForwardNetwork
from .MultiHeadAttention import MultiheadAttention
# from .MultiHeadAttention import MultiheadAttention
from .TorchMultiHeadAttention import TorchMultiHeadAttention
from .SpannedMasker import SpannedMasker
from .DeToken import DeToken
__all__ = [
"Decoder",
"Encoder",
"FeedForwardNetwork",
"MultiheadAttention",
"TorchMultiHeadAttention",
"SpannedMasker"
"SpannedMasker",
"DeToken"
]

4
Project_Model/Libs/Transformer/Utils/post_tokenization.py
View File

@@ -47,7 +47,7 @@ def normalize_sequence(
) -> tuple[list[int], list[bool]]:
new_sequence = pad_sequence(sequence, max_length, pad_token)
new_sequence = truncate_sequence(sequence, max_length, end_token)
PADDING_MASK = create_padding_mask(sequence, pad_token)
new_sequence = truncate_sequence(new_sequence, max_length, end_token)
PADDING_MASK = create_padding_mask(new_sequence, pad_token)
return (new_sequence, PADDING_MASK)