Generative AI for Text

# Language Model: predict next word
# P("sat" | "The cat") = 0.6
# P("on" | "The cat sat") = 0.9
# P("mat" | "The cat sat on the") = 0.8

# Probability chain rule
P(A, B, C) = P(A) * P(B|A) * P(C|A,B)

# Language model
P(w1, w2, ..., wn) = P(w1) * P(w2|w1) * ... * P(wn|w1..wn-1)

# Perplexity: lower = better
# PP = 2^(-1/N * sum(log2(P(wi))))

# Generative process
# 1. Sample w1 ~ P(w1)
# 2. Sample w2 ~ P(w2|w1)
# 3. Repeat until <EOS>

class RNNLM(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.rnn = nn.RNN(embed_dim, hidden_dim, batch_first=True)
        self.fc = nn.Linear(hidden_dim, vocab_size)

    def forward(self, x, hidden=None):
        embeds = self.embedding(x)
        out, hidden = self.rnn(embeds, hidden)
        logits = self.fc(out)
        return logits, hidden

# Training: teacher forcing
# Input:  ["The", "cat", "sat"]
# Target: ["cat", "sat", "on"]

model = RNNLM(25000, 128, 256)
criterion = nn.CrossEntropyLoss()

for epoch in range(10):
    for batch in loader:
        input_seq = batch[:, :-1]   # "The cat sat"
        target_seq = batch[:, 1:]   # "cat sat on"

        logits, _ = model(input_seq)
        loss = criterion(logits.reshape(-1, vocab_size),
                        target_seq.reshape(-1))
        loss.backward()
        optimizer.step()

# RNN: sequential, slow
h1 = f(x1, h0)
h2 = f(x2, h1)
h3 = f(x3, h2)
...
hn = f(xn, hn-1)

# Cannot parallelize!
# Must compute h1 before h2, h2 before h3...

# Attention: look at all positions
context = sum(alpha_i * h_i)  # weighted sum
# alpha_i = attention weight for position i

# Transformer: fully parallel
# All positions computed simultaneously
Q, K, V = Linear(X)  # parallel
attn = softmax(Q @ K.T / sqrt(d)) @ V

import torch
import torch.nn as nn
import math

def scaled_dot_product_attention(Q, K, V, mask=None):
    d_k = Q.size(-1)
    scores = torch.matmul(Q, K.transpose(-2, -1))
    scores = scores / math.sqrt(d_k)

    if mask is not None:
        scores = scores.masked_fill(mask == 0, -1e9)

    attn_weights = torch.softmax(scores, dim=-1)
    output = torch.matmul(attn_weights, V)
    return output, attn_weights

# Example
seq_len, d_k = 4, 64
Q = torch.randn(1, seq_len, d_k)
K = torch.randn(1, seq_len, d_k)
V = torch.randn(1, seq_len, d_k)

output, weights = scaled_dot_product_attention(Q, K, V)
print(weights.shape)  # (1, 4, 4) - attention matrix

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super().__init__()
        self.num_heads = num_heads
        self.d_k = d_model // num_heads

        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.W_o = nn.Linear(d_model, d_model)

    def forward(self, Q, K, V, mask=None):
        batch_size = Q.size(0)

        # Linear projections
        Q = self.W_q(Q).view(batch_size, -1,
                             self.num_heads, self.d_k)
        K = self.W_k(K).view(batch_size, -1,
                             self.num_heads, self.d_k)
        V = self.W_v(V).view(batch_size, -1,
                             self.num_heads, self.d_k)

        # Transpose: (batch, heads, seq_len, d_k)
        Q = Q.transpose(1, 2)
        K = K.transpose(1, 2)
        V = V.transpose(1, 2)

        # Attention
        output, weights = scaled_dot_product_attention(Q, K, V, mask)

        # Concat and project
        output = output.transpose(1, 2).contiguous()
        output = output.view(batch_size, -1,
                            self.num_heads * self.d_k)
        return self.W_o(output)

class TransformerEncoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff, dropout):
        super().__init__()
        self.self_attn = MultiHeadAttention(d_model, num_heads)
        self.feed_forward = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Linear(d_ff, d_model)
        )
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x, mask=None):
        # Self-attention + residual + norm
        attn_output = self.self_attn(x, x, x, mask)
        x = self.norm1(x + self.dropout(attn_output))

        # Feed-forward + residual + norm
        ff_output = self.feed_forward(x)
        x = self.norm2(x + self.dropout(ff_output))
        return x

# Stacking N layers
class TransformerEncoder(nn.Module):
    def __init__(self, N=6, d_model=512, heads=8, d_ff=2048):
        super().__init__()
        self.layers = nn.ModuleList([
            TransformerEncoderLayer(d_model, heads, d_ff, 0.1)
            for _ in range(N)
        ])

    def forward(self, x, mask=None):
        for layer in self.layers:
            x = layer(x, mask)
        return x

import torch
import math

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len).unsqueeze(1).float()
        div_term = torch.exp(
            torch.arange(0, d_model, 2).float() *
            -(math.log(10000.0) / d_model))

        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        pe = pe.unsqueeze(0)  # (1, max_len, d_model)
        self.register_buffer('pe', pe)

    def forward(self, x):
        # x: (batch, seq_len, d_model)
        return x + self.pe[:, :x.size(1), :]

# Usage
pe = PositionalEncoding(d_model=512)
x = torch.randn(1, 10, 512)  # (batch=1, len=10, dim=512)
output = pe(x)  # add positional info

class TransformerDecoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff, dropout):
        super().__init__()
        # Masked self-attention
        self.self_attn = MultiHeadAttention(d_model, num_heads)
        # Cross-attention (to encoder)
        self.cross_attn = MultiHeadAttention(d_model, num_heads)
        # Feed-forward
        self.feed_forward = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Linear(d_ff, d_model)
        )
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.norm3 = nn.LayerNorm(d_model)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x, enc_output, src_mask, tgt_mask):
        # Masked self-attention
        attn1 = self.self_attn(x, x, x, tgt_mask)
        x = self.norm1(x + self.dropout(attn1))

        # Cross-attention
        attn2 = self.cross_attn(x, enc_output, enc_output, src_mask)
        x = self.norm2(x + self.dropout(attn2))

        # Feed-forward
        ff = self.feed_forward(x)
        x = self.norm3(x + self.dropout(ff))
        return x

class Transformer(nn.Module):
    def __init__(self, src_vocab, tgt_vocab, d_model=512,
                 num_heads=8, num_layers=6, d_ff=2048):
        super().__init__()
        self.encoder_embed = nn.Embedding(src_vocab, d_model)
        self.decoder_embed = nn.Embedding(tgt_vocab, d_model)
        self.pos_encoding = PositionalEncoding(d_model)

        self.encoder = TransformerEncoder(
            num_layers, d_model, num_heads, d_ff)
        self.decoder = TransformerDecoder(
            num_layers, d_model, num_heads, d_ff)

        self.output_proj = nn.Linear(d_model, tgt_vocab)

    def forward(self, src, tgt, src_mask, tgt_mask):
        # Encode
        enc_output = self.encoder(
            self.pos_encoding(self.encoder_embed(src)),
            src_mask)

        # Decode
        dec_output = self.decoder(
            self.pos_encoding(self.decoder_embed(tgt)),
            enc_output, src_mask, tgt_mask)

        return self.output_proj(dec_output)

# Original "Attention Is All You Need"
# Encoder: 6 layers, d_model=512, 8 heads
# Decoder: 6 layers, d_model=512, 8 heads

from transformers import BertTokenizer, BertModel

# Load pre-trained BERT
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
model = BertModel.from_pretrained('bert-base-uncased')

# Tokenize
text = "The cat sat on the mat"
inputs = tokenizer(text, return_tensors='pt')
print(inputs.keys())
# dict_keys(['input_ids', 'token_type_ids',
#            'attention_mask'])

# Get embeddings
with torch.no_grad():
    outputs = model(**inputs)

last_hidden = outputs.last_hidden_state
print(last_hidden.shape)  # (1, 8, 768)

# Fine-tuning for classification
class BertClassifier(nn.Module):
    def __init__(self):
        super().__init__()
        self.bert = BertModel.from_pretrained('bert-base-uncased')
        self.classifier = nn.Linear(768, 2)

    def forward(self, input_ids, attention_mask):
        outputs = self.bert(input_ids, attention_mask)
        pooled = outputs.pooler_output
        return self.classifier(pooled)

from transformers import GPT2LMHeadModel, GPT2Tokenizer

# Load GPT-2
tokenizer = GPT2Tokenizer.from_pretrained('gpt2')
model = GPT2LMHeadModel.from_pretrained('gpt2')

# Generate text
prompt = "The future of AI is"
input_ids = tokenizer.encode(prompt, return_tensors='pt')

# Generate
output = model.generate(
    input_ids,
    max_length=50,
    num_return_sequences=1,
    temperature=0.7,
    do_sample=True
)

print(tokenizer.decode(output[0]))
# "The future of AI is bright. We are seeing more..."

# Temperature: higher = more random
# Top-k sampling: only consider top k tokens
# Top-p (nucleus): sample from smallest set with p% mass

# BERT: Encoder (bidirectional)
Input: "The [MASK] sat on the mat"
       <-- both directions -->
Output: "cat" (classification)

# GPT: Decoder (left-to-right)
Input: "The future of AI is"
       <-- left to right only -->
Output: "bright and promising..."

# Use cases:
# BERT:
#   - Sentiment analysis
#   - Named entity recognition
#   - Question answering
#   - Text classification

# GPT:
#   - Story generation
#   - Code completion
#   - Summarization
#   - Chatbots

pip install transformers torch

from transformers import pipeline

# Sentiment analysis
classifier = pipeline('sentiment-analysis')
result = classifier('I love this product!')
print(result)
# [{'label': 'POSITIVE', 'score': 0.9998}]

# Text generation
generator = pipeline('text-generation', model='gpt2')
result = generator('The future of AI is', max_length=50)
print(result[0]['generated_text'])

# Translation
translator = pipeline('translation_en_to_fr')
result = translator('Hello, how are you?')
print(result[0]['translation_text'])

# Summarization
summarizer = pipeline('summarization')
text = "Long article about AI..."
result = summarizer(text, max_length=50)
print(result[0]['summary_text'])

from transformers import (BertForSequenceClassification,
                           Trainer, TrainingArguments)

# Load model
model = BertForSequenceClassification.from_pretrained(
    'bert-base-uncased', num_labels=2)

# Training arguments
training_args = TrainingArguments(
    output_dir='./results',
    num_train_epochs=3,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=64,
    warmup_steps=500,
    weight_decay=0.01,
    logging_dir='./logs',
    evaluation_strategy='epoch'
)

# Trainer
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset
)

# Fine-tune!
trainer.train()

# Save
model.save_pretrained('./my_bert_model')

uv add django torch transformers

uv run django-admin startproject wk13 .
uv run manage.py startapp dashboard

# dashboard/views.py
import json, torch
from django.http import StreamingHttpResponse
from transformers import (BertForSequenceClassification,
                           BertTokenizer)

def train(request, model_type):
    def event_stream():
        # Load pre-trained model
        if model_type == 'bert':
            model = BertForSequenceClassification.from_pretrained(
                'bert-base-uncased', num_labels=2)
            tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')

        # Fine-tune on sentiment dataset
        optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5)

        for epoch in range(3):
            train_loss = 0
            correct = 0
            total = 0

            for batch in train_loader:
                outputs = model(**batch)
                loss = outputs.loss
                optimizer.zero_grad()
                loss.backward()
                optimizer.step()

                train_loss += loss.item()
                predictions = outputs.logits.argmax(-1)
                correct += (predictions == batch['labels']).sum().item()
                total += batch['labels'].size(0)

            accuracy = correct / total
            yield f"data: {json.dumps({'epoch': epoch, 'loss': train_loss/len(train_loader), 'accuracy': accuracy})}\n\n"

    return StreamingHttpResponse(event_stream(), content_type='text/event-stream')

from django.http import StreamingHttpResponse
import json, time

def train_stream(request, model_type):
    def event_stream():
        model = load_model(model_type)
        optimizer = torch.optim.AdamW(model.parameters(), lr=2e-5)

        for epoch in range(5):
            model.train()
            epoch_loss = 0
            correct = 0
            total = 0

            for batch in mini_loader:
                optimizer.zero_grad()
                outputs = model(**batch)
                loss = outputs.loss
                loss.backward()
                optimizer.step()

                epoch_loss += loss.item()
                preds = outputs.logits.argmax(-1)
                correct += (preds == batch['labels']).sum().item()
                total += batch['labels'].size(0)

            accuracy = correct / total
            avg_loss = epoch_loss / len(mini_loader)

            yield f"data: {json.dumps({\n"
                  f"  'epoch': {epoch},\n"
                  f"  'loss': {avg_loss:.4f},\n"
                  f"  'accuracy': {accuracy:.4f}\n"
                  f"})}\n\n"
            time.sleep(0.1)

    resp = StreamingHttpResponse(event_stream(),
        content_type='text/event-stream')
    resp['Cache-Control'] = 'no-cache'
    return resp

<!-- dashboard/templates/index.html -->
<div x-data="{ model: 'bert', training: false }">
  <h1>Transformer Training Dashboard</h1>

  <select x-model="model">
    <option value="bert">BERT (Encoder)</option>
    <option value="gpt2">GPT-2 (Decoder)</option>
    <option value="t5">T5 (Encoder-Decoder)</option>
  </select>

  <button @click="startTrain()">
    Start Fine-tuning
  </button>

  <div>
    <p>Epoch: <span x-text="epoch"></span></p>
    <p>Loss: <span x-text="loss"></span></p>
    <p>Accuracy: <span x-text="acc"></span></p>
  </div>

  <canvas id="chart" width="420" height="200"></canvas>
</div>

<script>
function startTrain() {
  const es = new EventSource('/train/' + model);
  es.onmessage = function(e) {
    const m = JSON.parse(e.data);
    epoch = m.epoch;
    loss = m.loss.toFixed(4);
    acc = m.accuracy.toFixed(4);
    drawChart(m.epoch, m.loss, m.accuracy);
  };
}
</script>