MBCDN

Das Sequential-Modell ist eine einfache Möglichkeit, sequenzielle Modelle in PyTorch zu definieren. Es erlaubt uns, eine sequentielle Kette von Schichten zu erstellen, in der die Ausgabe einer Schicht als Eingabe für die nächste Schicht verwendet wird.

import torch
import torch.nn as nn
 
model = nn.Sequential(
    nn.Linear(784, 256),
    nn.ReLU(),
    nn.Linear(256, 10),
    nn.Softmax(dim=1)
)

In diesem Beispiel wird ein Sequential-Modell mit zwei linearen Schichten, einer ReLU-Aktivierungsfunktion und einer Softmax-Aktivierungsfunktion erstellt. Die Eingabe des Modells hat eine Größe von 784, und die Ausgabe hat eine Größe von 10.

Das Module-Modell ist eine flexiblere Art, Modelle in PyTorch zu definieren. Es ermöglicht uns, eigene Schichten und komplexe Modelle zu erstellen, indem wir die nn.Module-Klasse erweitern.

import torch
import torch.nn as nn
 
class MyModel(nn.Module):
    def __init__(self):
        super(MyModel, self).__init__()
        self.fc1 = nn.Linear(784, 256)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(256, 10)
        self.softmax = nn.Softmax(dim=1)
 
    def forward(self, x):
        x = self.fc1(x)
        x = self.relu(x)
        x = self.fc2(x)
        x = self.softmax(x)
        return x
 
model = MyModel()

In diesem Beispiel wird ein Module-Modell erstellt, das den gleichen Aufbau wie das Sequential-Modell hat. Die Schichten werden jedoch als Attribute der Modellklasse definiert, und die Vorwärtsfunktion wird durch die Methode forward festgelegt. Dadurch haben wir mehr Flexibilität bei der Gestaltung und Steuerung des Modells.

Das Functional-Modell ist eine weitere Möglichkeit, Modelle in PyTorch zu definieren. Es basiert auf der Verwendung von Funktionen aus dem torch.nn.functional-Modul, anstatt Schichtenobjekte zu erstellen.

import torch
import torch.nn.functional as F
 
def my_model(x):
    x = F.linear(x, 784, 256)
    x = F.relu(x)
    x = F.linear(x, 256, 10)
    x = F.softmax(x, dim=1)
    return x
 
input = torch.randn(1, 784)
output = my_model(input)

In diesem Beispiel wird das Modell als eine Funktion definiert, die die Funktionen aus dem torch.nn.functional-Modul verwendet. Das Functional-Modell ist nützlich, wenn Sie ein einfaches Modell ohne interne Zustände oder trainierbare Parameter erstellen möchten.

Convolutional Neural Networks sind besonders gut geeignet für die Verarbeitung von Bildern. Sie verwenden spezielle Schichten wie Convolutional Layers und Pooling Layers, um die räumliche Struktur der Eingabedaten zu berücksichtigen.

import torch
import torch.nn as nn
 
class MyCNN(nn.Module):
    def __init__(self):
        super(MyCNN, self).__init__()
        self.conv1 = nn.Conv2d(3, 16, kernel_size=3, stride=1, padding=1)
        self.relu = nn.ReLU()
        self.pool = nn.MaxPool2d(kernel_size=2, stride=2)
        self.fc = nn.Linear(16 * 14 * 14, 10)
 
    def forward(self, x):
        x = self.conv1(x)
        x = self.relu(x)
        x = self.pool(x)
        x = x.view(-1, 16 * 14 * 14)
        x = self.fc(x)
        return x
 
model = MyCNN()

In diesem Beispiel wird ein CNN-Modell erstellt, das eine Convolutional Layer, eine ReLU-Aktivierungsfunktion, eine Max-Pooling-Layer und eine Fully Connected Layer enthält. Das Modell wird für die Klassifikation von Bildern verwendet.

Recurrent Neural Networks sind gut geeignet für die Verarbeitung von sequenziellen Daten wie Texten oder Zeitreihen. Sie haben die Fähigkeit, Informationen über vergangene Zustände beizubehalten.

import torch
import torch.nn as nn
 
class MyRNN(nn.Module):
    def __init__(self):
        super(MyRNN, self).__init__()
        self.rnn = nn.RNN(input_size=100, hidden_size=128, num_layers=2, batch_first=True)
        self.fc = nn.Linear(128, 10)
 
    def forward(self, x):
        _, h = self.rnn(x)
        x = self.fc(h[-1])
        return x
 
model = MyRNN()

In diesem Beispiel wird ein RNN-Modell erstellt, das aus einer RNN-Schicht und einer Fully Connected Layer besteht. Das Modell nimmt Eingaben mit einer Sequenzlänge von 100 und gibt eine Ausgabe der Größe 10 zurück.

Transformer-Modelle haben in den letzten Jahren an Beliebtheit gewonnen und sind besonders gut für die Verarbeitung von natürlicher Sprache geeignet. Sie verwenden Aufmerksamkeitsmechanismen, um die Beziehungen zwischen den Elementen einer Sequenz zu modellieren.

import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.nn import TransformerEncoder, TransformerEncoderLayer
 
class MyTransformer(nn.Module):
    def __init__(self):
        super(MyTransformer, self).__init__()
        self.embedding = nn.Embedding(10000, 256)
        self.transformer_layer = TransformerEncoderLayer(256, 8)
        self.transformer = TransformerEncoder(self.transformer_layer, num_layers=4)
        self.fc = nn.Linear(256, 10)
 
    def forward(self, x):
        x = self.embedding(x)
        x = x.permute(1, 0, 2)
        x = self.transformer(x)
        x = x.mean(dim=0)
        x = self.fc(x)
        return x
 
model = MyTransformer()

In diesem Beispiel wird ein Transformer-Modell erstellt, das eine Embedding-Schicht, einen TransformerEncoder und eine Fully Connected Layer enthält. Das Modell nimmt Sequenzeingaben, z.B. Token-IDs, und gibt eine Ausgabe der Größe 10 zurück.

# Pattern Recognition
 
import torch
import torch.nn as nn
import torch.optim as optim
 
# Daten vorbereiten (Beispiel)
sequences = [
    [1,0,1,1,0],
    [0,1,1,0,0],
    [0,1,0,0,0],
    [1,0,1,0,1],
    [1,0,0,0,1],
    [0,1,0,1,0],
    [1,1,0,1,1],
    [0,0,1,0,0]
]
labels = [1,1,0,1,0,0,0,1]  # Beispieldaten fur die Musterklassifizierung
 
# Hyperparameter
input_size = 1
hidden_size = 5*5
output_size = 2
num_layers = 3
num_epochs = 100
learning_rate = 0.1
 
# Daten in Tensoren umwandeln
input_sequences = torch.tensor(sequences, dtype=torch.float32).unsqueeze(-1)
target_labels = torch.tensor(labels, dtype=torch.long)
 
# Modell definieren
class RNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size, num_layers):
        super(RNN, self).__init__()
        self.hidden_size = hidden_size
        self.rnn = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)
 
    def forward(self, x):
        h0 = torch.zeros(num_layers, x.size(0), self.hidden_size).to(x.device)
        c0 = torch.zeros(num_layers, x.size(0), self.hidden_size).to(x.device)
        out, _ = self.rnn(x, (h0, c0))
        out = self.fc(out[:, -1, :])
        return out
 
model = RNN(input_size, hidden_size, output_size, num_layers)
 
# Verlustfunktion und Optimierungsalgorithmus definieren
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)
 
# Modell trainieren
for epoch in range(num_epochs):
    optimizer.zero_grad()
    outputs = model(input_sequences)
    loss = criterion(outputs, target_labels)
    loss.backward()
    optimizer.step()
    print(f'Epoch: {epoch+1}/{num_epochs}, Loss: {loss.item()}')
 
# Modell evaluieren
with torch.no_grad():
    test_input = torch.tensor([[1,0,0,1,1]], dtype=torch.float32).unsqueeze(-1)
    predicted = model(test_input)
    _, predicted_labels = torch.max(predicted.data, 1)
    print(f'Predicted Label: {predicted_labels.item()} {predicted.data}')
 
 
'''
torch.save(model.state_dict(), "test.pt")
model.load_state_dict(torch.load("test.pt"))
'''

import torch
import torch.nn as nn
from torch.nn import functional as F
 
# hyperparameters
batch_size = 16 # how many independent sequences will we process in parallel?
block_size = 32 # what is the maximum context length for predictions?
max_iters = 15000
eval_interval = 1000
learning_rate = 1e-3
device = 'cuda' if torch.cuda.is_available() else 'cpu'
eval_iters = 200
n_embd = 64
n_head = 4
n_layer = 4
dropout = 0.0
# ------------
 
torch.manual_seed(1337)
 
!wget https://raw.githubusercontent.com/karpathy/char-rnn/master/data/tinyshakespeare/input.txt
with open('input.txt', 'r', encoding='utf-8') as f:
    text = f.read()
 
# here are all the unique characters that occur in this text
chars = sorted(list(set(text)))
vocab_size = len(chars)
# create a mapping from characters to integers
stoi = { ch:i for i,ch in enumerate(chars) }
itos = { i:ch for i,ch in enumerate(chars) }
encode = lambda s: [stoi[c] for c in s] # encoder: take a string, output a list of integers
decode = lambda l: ''.join([itos[i] for i in l]) # decoder: take a list of integers, output a string
 
# Train and test splits
data = torch.tensor(encode(text), dtype=torch.long)
n = int(0.9*len(data)) # first 90% will be train, rest val
train_data = data[:n]
val_data = data[n:]
 
# data loading
def get_batch(split):
    # generate a small batch of data of inputs x and targets y
    data = train_data if split == 'train' else val_data
    ix = torch.randint(len(data) - block_size, (batch_size,))
    x = torch.stack([data[i:i+block_size] for i in ix])
    y = torch.stack([data[i+1:i+block_size+1] for i in ix])
    x, y = x.to(device), y.to(device)
    return x, y
 
@torch.no_grad()
def estimate_loss():
    out = {}
    model.eval()
    for split in ['train', 'val']:
        losses = torch.zeros(eval_iters)
        for k in range(eval_iters):
            X, Y = get_batch(split)
            logits, loss = model(X, Y)
            losses[k] = loss.item()
        out[split] = losses.mean()
    model.train()
    return out
 
class Head(nn.Module):
    """ one head of self-attention """
 
    def __init__(self, head_size):
        super().__init__()
        self.key = nn.Linear(n_embd, head_size, bias=False)
        self.query = nn.Linear(n_embd, head_size, bias=False)
        self.value = nn.Linear(n_embd, head_size, bias=False)
        self.register_buffer('tril', torch.tril(torch.ones(block_size, block_size)))
 
        self.dropout = nn.Dropout(dropout)
 
    def forward(self, x):
        B,T,C = x.shape
        k = self.key(x)   # (B,T,C)
        q = self.query(x) # (B,T,C)
        # compute attention scores ("affinities")
        wei = q @ k.transpose(-2,-1) * C**-0.5 # (B, T, C) @ (B, C, T) -> (B, T, T)
        wei = wei.masked_fill(self.tril[:T, :T] == 0, float('-inf')) # (B, T, T)
        wei = F.softmax(wei, dim=-1) # (B, T, T)
        wei = self.dropout(wei)
        # perform the weighted aggregation of the values
        v = self.value(x) # (B,T,C)
        out = wei @ v # (B, T, T) @ (B, T, C) -> (B, T, C)
        return out
 
class MultiHeadAttention(nn.Module):
    """ multiple heads of self-attention in parallel """
 
    def __init__(self, num_heads, head_size):
        super().__init__()
        self.heads = nn.ModuleList([Head(head_size) for _ in range(num_heads)])
        self.proj = nn.Linear(n_embd, n_embd)
        self.dropout = nn.Dropout(dropout)
 
    def forward(self, x):
        out = torch.cat([h(x) for h in self.heads], dim=-1)
        out = self.dropout(self.proj(out))
        return out
 
class FeedFoward(nn.Module):
    """ a simple linear layer followed by a non-linearity """
 
    def __init__(self, n_embd):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(n_embd, 4 * n_embd),
            nn.ReLU(),
            nn.Linear(4 * n_embd, n_embd),
            nn.Dropout(dropout),
        )
 
    def forward(self, x):
        return self.net(x)
 
class Block(nn.Module):
    """ Transformer block: communication followed by computation """
 
    def __init__(self, n_embd, n_head):
        # n_embd: embedding dimension, n_head: the number of heads we'd like
        super().__init__()
        head_size = n_embd // n_head
        self.sa = MultiHeadAttention(n_head, head_size)
        self.ffwd = FeedFoward(n_embd)
        self.ln1 = nn.LayerNorm(n_embd)
        self.ln2 = nn.LayerNorm(n_embd)
 
    def forward(self, x):
        x = x + self.sa(self.ln1(x))
        x = x + self.ffwd(self.ln2(x))
        return x
 
# super simple bigram model
class BigramLanguageModel(nn.Module):
 
    def __init__(self):
        super().__init__()
        # each token directly reads off the logits for the next token from a lookup table
        self.token_embedding_table = nn.Embedding(vocab_size, n_embd)
        self.position_embedding_table = nn.Embedding(block_size, n_embd)
        self.blocks = nn.Sequential(*[Block(n_embd, n_head=n_head) for _ in range(n_layer)])
        self.ln_f = nn.LayerNorm(n_embd) # final layer norm
        self.lm_head = nn.Linear(n_embd, vocab_size)
 
    def forward(self, idx, targets=None):
        B, T = idx.shape
 
        # idx and targets are both (B,T) tensor of integers
        tok_emb = self.token_embedding_table(idx) # (B,T,C)
        pos_emb = self.position_embedding_table(torch.arange(T, device=device)) # (T,C)
        x = tok_emb + pos_emb # (B,T,C)
        x = self.blocks(x) # (B,T,C)
        x = self.ln_f(x) # (B,T,C)
        logits = self.lm_head(x) # (B,T,vocab_size)
 
        if targets is None:
            loss = None
        else:
            B, T, C = logits.shape
            logits = logits.view(B*T, C)
            targets = targets.view(B*T)
            loss = F.cross_entropy(logits, targets)
 
        return logits, loss
 
    def generate(self, idx, max_new_tokens):
        # idx is (B, T) array of indices in the current context
        for _ in range(max_new_tokens):
            # crop idx to the last block_size tokens
            idx_cond = idx[:, -block_size:]
            # get the predictions
            logits, loss = self(idx_cond)
            # focus only on the last time step
            logits = logits[:, -1, :] # becomes (B, C)
            # apply softmax to get probabilities
            probs = F.softmax(logits, dim=-1) # (B, C)
            # sample from the distribution
            idx_next = torch.multinomial(probs, num_samples=1) # (B, 1)
            # append sampled index to the running sequence
            idx = torch.cat((idx, idx_next), dim=1) # (B, T+1)
        return idx
 
 
model = BigramLanguageModel()
m = model.to(device)
# print the number of parameters in the model
print(sum(p.numel() for p in m.parameters())/1e6, 'M parameters')
 
# create a PyTorch optimizer
optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate)
 
for iter in range(max_iters):
 
    # every once in a while evaluate the loss on train and val sets
    if iter % eval_interval == 0 or iter == max_iters - 1:
        losses = estimate_loss()
        print(f"step {iter}: train loss {losses['train']:.4f}, val loss {losses['val']:.4f}")
 
    # sample a batch of data
    xb, yb = get_batch('train')
 
    # evaluate the loss
    logits, loss = model(xb, yb)
    optimizer.zero_grad(set_to_none=True)
    loss.backward()
    optimizer.step()
 
 
# random
#context = torch.zeros((1, 1), dtype=torch.long, device=device)
#print(decode(m.generate(context, max_new_tokens=100)[0].tolist()))
 
# prompt
#context = torch.tensor([encode("Who is the king?")], dtype=torch.long, device=device)
#print(decode(m.generate(context, max_new_tokens=100)[0].tolist()))
 
# completion
def complete(prompt, max_new_tokens=200, temperature=1.0):
    idx = torch.tensor([encode(prompt)], dtype=torch.long, device=device)
    with torch.no_grad():
        for _ in range(max_new_tokens):
            idx_cond = idx[:, -block_size:]
            logits, _ = m(idx_cond)
            logits = logits[:, -1, :] / temperature
            probs = F.softmax(logits, dim=-1)
            idx_next = torch.multinomial(probs, num_samples=1)
            idx = torch.cat((idx, idx_next), dim=1)
 
    full = decode(idx[0].tolist())
    return full # full[len(prompt):] 
print(complete("To be, or not to be: ", max_new_tokens=100, temperature=0.8))