Transformers and Attention Mechanism

import numpy as np

def softmax(x, axis=-1):
    """Numerically stable softmax."""
    e_x = np.exp(x - np.max(x, axis=axis, keepdims=True))
    return e_x / e_x.sum(axis=axis, keepdims=True)

def scaled_dot_product_attention(Q, K, V):
    """Compute scaled dot-product attention.
    
    Args:
        Q: Queries (seq_len, d_k)
        K: Keys (seq_len, d_k)
        V: Values (seq_len, d_v)
    Returns:
        output: Weighted sum of values (seq_len, d_v)
        weights: Attention weights (seq_len, seq_len)
    """
    d_k = K.shape[-1]
    
    # Step 1: Compute attention scores
    scores = np.matmul(Q, K.T)  # (seq_len, seq_len)
    
    # Step 2: Scale by sqrt(d_k)
    scores = scores / np.sqrt(d_k)
    
    # Step 3: Apply softmax to get attention weights
    weights = softmax(scores, axis=-1)  # Each row sums to 1
    
    # Step 4: Weighted sum of values
    output = np.matmul(weights, V)  # (seq_len, d_v)
    
    return output, weights

# Example: "The cat sat on the mat"
np.random.seed(42)
seq_len = 6  # 6 words
d_model = 8  # Embedding dimension

# Simulated word embeddings
words = ['The', 'cat', 'sat', 'on', 'the', 'mat']
embeddings = np.random.randn(seq_len, d_model)

# In self-attention, Q, K, V are all derived from the same input
# Using random weight matrices for demonstration
W_Q = np.random.randn(d_model, d_model)
W_K = np.random.randn(d_model, d_model)
W_V = np.random.randn(d_model, d_model)

Q = np.matmul(embeddings, W_Q)
K = np.matmul(embeddings, W_K)
V = np.matmul(embeddings, W_V)

output, attention_weights = scaled_dot_product_attention(Q, K, V)

print(f"Input shape: {embeddings.shape}")
print(f"Q, K, V shapes: {Q.shape}, {K.shape}, {V.shape}")
print(f"Output shape: {output.shape}")
print(f"Attention weights shape: {attention_weights.shape}")

# Display attention weights
print("\nAttention weights (each row shows what that word attends to):")
print(f"{'':>8}", end='')
for w in words:
    print(f"{w:>8}", end='')
print()

for i, word in enumerate(words):
    print(f"{word:>8}", end='')
    for j in range(seq_len):
        print(f"{attention_weights[i, j]:>8.3f}", end='')
    print()

print("\nEach row sums to 1.0 (probability distribution).")
print("Higher values mean that word pays more attention to the column word.")

import numpy as np
import matplotlib.pyplot as plt

def positional_encoding(max_length, d_model):
    """Generate sinusoidal positional encoding."""
    pe = np.zeros((max_length, d_model))
    position = np.arange(max_length)[:, np.newaxis]  # (max_length, 1)
    div_term = np.exp(np.arange(0, d_model, 2) * -(np.log(10000.0) / d_model))
    
    pe[:, 0::2] = np.sin(position * div_term)  # Even indices
    pe[:, 1::2] = np.cos(position * div_term)  # Odd indices
    return pe

# Generate positional encoding
max_len = 100
d_model = 128
pe = positional_encoding(max_len, d_model)

print(f"Positional encoding shape: {pe.shape}")
print(f"\nEncoding for position 0 (first 8 dims): {pe[0, :8].round(3)}")
print(f"Encoding for position 1 (first 8 dims): {pe[1, :8].round(3)}")
print(f"Encoding for position 50 (first 8 dims): {pe[50, :8].round(3)}")

# Visualize
plt.figure(figsize=(14, 6))
plt.imshow(pe, aspect='auto', cmap='RdBu')
plt.colorbar(label='Value')
plt.xlabel('Embedding Dimension')
plt.ylabel('Position in Sequence')
plt.title('Sinusoidal Positional Encoding')
plt.savefig('positional_encoding.png', dpi=100)
plt.show()

# Show that relative positions have consistent relationships
print("\nRelative position encoding property:")
print("Distance between position 10 and 11:")
dist_10_11 = np.linalg.norm(pe[10] - pe[11])
print(f"  ||PE(10) - PE(11)|| = {dist_10_11:.4f}")

print("Distance between position 50 and 51:")
dist_50_51 = np.linalg.norm(pe[50] - pe[51])
print(f"  ||PE(50) - PE(51)|| = {dist_50_51:.4f}")

print(f"\nAdjacentpositions have similar distances regardless of absolute position.")
print("This helps the model learn relative positioning.")

from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
import torch

# === Method 1: Quick Pipeline ===
print("=== Method 1: Quick Pipeline ===")
classifier = pipeline('sentiment-analysis')

texts = [
    "This product exceeded all my expectations, truly remarkable!",
    "Terrible experience, the item arrived damaged and support was unhelpful.",
    "The movie had great visuals but the plot was somewhat predictable.",
    "I absolutely love this restaurant, best biryani in the city!",
    "Not worth the price, there are much better alternatives available."
]

for text in texts:
    result = classifier(text)[0]
    print(f"[{result['label']:>8}] ({result['score']:.3f}) {text[:60]}...")

# === Method 2: Using Specific Model ===
print("\n=== Method 2: Specific Model (distilbert) ===")
model_name = 'distilbert-base-uncased-finetuned-sst-2-english'
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(model_name)

text = "Transformers have completely revolutionized natural language processing"

# Tokenize
inputs = tokenizer(text, return_tensors='pt', truncation=True, padding=True)
print(f"\nInput text: '{text}'")
print(f"Token IDs: {inputs['input_ids'][0][:10].tolist()}... (first 10)")
print(f"Attention mask: {inputs['attention_mask'][0][:10].tolist()}... (first 10)")

# Predict
with torch.no_grad():
    outputs = model(**inputs)
    logits = outputs.logits
    probabilities = torch.nn.functional.softmax(logits, dim=-1)

print(f"\nLogits: {logits[0].tolist()}")
print(f"Probabilities: Negative={probabilities[0][0]:.4f}, Positive={probabilities[0][1]:.4f}")
predicted = 'POSITIVE' if probabilities[0][1] > 0.5 else 'NEGATIVE'
print(f"Prediction: {predicted}")

# === Method 3: Multiple Tasks ===
print("\n=== Method 3: Multiple NLP Tasks ===")

# Named Entity Recognition
ner = pipeline('ner', grouped_entities=True)
ner_text = "Rahul Sharma works at Infosys in Bangalore"
entities = ner(ner_text)
print(f"\nNER: '{ner_text}'")
for ent in entities:
    print(f"  {ent['word']:>20} -> {ent['entity_group']} ({ent['score']:.3f})")

# Question Answering
qa = pipeline('question-answering')
result = qa(
    question="What is the Transformer architecture based on?",
    context="The Transformer architecture, introduced in 2017, is based entirely on attention mechanisms. It does not use recurrence or convolution."
)
print(f"\nQA: Answer = '{result['answer']}' (confidence: {result['score']:.3f})")

# Zero-Shot Classification
zero_shot = pipeline('zero-shot-classification')
result = zero_shot(
    "The new policy will increase taxes on luxury goods",
    candidate_labels=['politics', 'sports', 'technology', 'finance']
)
print(f"\nZero-Shot Classification:")
for label, score in zip(result['labels'], result['scores']):
    print(f"  {label:>12}: {score:.3f}")

import numpy as np

def softmax(x, axis=-1):
    e_x = np.exp(x - np.max(x, axis=axis, keepdims=True))
    return e_x / e_x.sum(axis=axis, keepdims=True)

def single_head_attention(Q, K, V):
    """Single attention head."""
    d_k = K.shape[-1]
    scores = np.matmul(Q, K.T) / np.sqrt(d_k)
    weights = softmax(scores)
    return np.matmul(weights, V), weights

def multi_head_attention(X, num_heads, d_model):
    """Multi-head attention.
    
    Args:
        X: Input (seq_len, d_model)
        num_heads: Number of attention heads
        d_model: Total model dimension
    """
    assert d_model % num_heads == 0
    d_k = d_model // num_heads  # Dimension per head
    seq_len = X.shape[0]
    
    # Generate random weight matrices for each head
    np.random.seed(42)
    heads_output = []
    all_weights = []
    
    for h in range(num_heads):
        # Each head has its own projection matrices
        W_Q = np.random.randn(d_model, d_k) * 0.1
        W_K = np.random.randn(d_model, d_k) * 0.1
        W_V = np.random.randn(d_model, d_k) * 0.1
        
        Q = np.matmul(X, W_Q)
        K = np.matmul(X, W_K)
        V = np.matmul(X, W_V)
        
        head_out, weights = single_head_attention(Q, K, V)
        heads_output.append(head_out)
        all_weights.append(weights)
    
    # Concatenate all heads
    concatenated = np.concatenate(heads_output, axis=-1)  # (seq_len, d_model)
    
    # Final linear projection
    W_O = np.random.randn(d_model, d_model) * 0.1
    output = np.matmul(concatenated, W_O)
    
    return output, all_weights

# Example
seq_len = 5
d_model = 16
num_heads = 4

words = ['I', 'love', 'machine', 'learning', 'today']
X = np.random.randn(seq_len, d_model)

output, head_weights = multi_head_attention(X, num_heads, d_model)

print(f"Input shape: {X.shape}")
print(f"Output shape: {output.shape}")
print(f"Number of heads: {num_heads}")
print(f"Dimension per head: {d_model // num_heads}")

# Show how different heads attend differently
print("\nAttention patterns per head (for word 'love'):")
for h in range(num_heads):
    weights_for_love = head_weights[h][1]  # Word index 1 = 'love'
    top_word_idx = np.argmax(weights_for_love)
    print(f"  Head {h+1}: strongest attention to '{words[top_word_idx]}' "
          f"(weight: {weights_for_love[top_word_idx]:.3f})")

print("\nDifferent heads capture different relationships.")
print("In a trained model, heads specialize in syntax, semantics, position, etc.")

import numpy as np

# Demonstrate the key difference between BERT and GPT attention patterns

def create_attention_mask(seq_len, model_type):
    """Create attention mask for different Transformer variants."""
    if model_type == 'bert':
        # BERT: bidirectional -- every token attends to every other token
        mask = np.ones((seq_len, seq_len))
    elif model_type == 'gpt':
        # GPT: causal/autoregressive -- tokens only attend to previous tokens
        mask = np.tril(np.ones((seq_len, seq_len)))  # Lower triangular
    return mask

words = ['The', 'cat', 'sat', 'on', 'mat']
seq_len = len(words)

# BERT attention mask
bert_mask = create_attention_mask(seq_len, 'bert')
print("=== BERT (Encoder-Only, Bidirectional) ===")
print("Every word can attend to every other word:")
print(f"{'':>6}", end='')
for w in words:
    print(f"{w:>6}", end='')
print()
for i, w in enumerate(words):
    print(f"{w:>6}", end='')
    for j in range(seq_len):
        symbol = 'YES' if bert_mask[i, j] == 1 else ' - '
        print(f"{symbol:>6}", end='')
    print()

# GPT attention mask
gpt_mask = create_attention_mask(seq_len, 'gpt')
print("\n=== GPT (Decoder-Only, Causal/Autoregressive) ===")
print("Each word can only attend to itself and previous words:")
print(f"{'':>6}", end='')
for w in words:
    print(f"{w:>6}", end='')
print()
for i, w in enumerate(words):
    print(f"{w:>6}", end='')
    for j in range(seq_len):
        symbol = 'YES' if gpt_mask[i, j] == 1 else ' - '
        print(f"{symbol:>6}", end='')
    print()

print("\n=== Summary ===")
print("BERT: Bidirectional. Sees all words. Best for UNDERSTANDING text.")
print("      Use case: classification, NER, QA, semantic search.")
print("      Pre-training: Masked Language Modeling (predict [MASK] tokens).")
print()
print("GPT:  Autoregressive. Sees only past words. Best for GENERATING text.")
print("      Use case: text generation, chatbots, code completion.")
print("      Pre-training: Next Token Prediction (predict the next word).")
print()
print("T5:   Encoder-Decoder. Encoder sees all input, decoder generates output.")
print("      Use case: translation, summarization, any text-to-text task.")

from transformers import AutoTokenizer, AutoModelForSequenceClassification, Trainer, TrainingArguments
import torch
from torch.utils.data import Dataset
import numpy as np

# Custom dataset class
class TextDataset(Dataset):
    def __init__(self, texts, labels, tokenizer, max_length=128):
        self.encodings = tokenizer(
            texts, truncation=True, padding=True, max_length=max_length,
            return_tensors='pt'
        )
        self.labels = torch.tensor(labels, dtype=torch.long)
    
    def __len__(self):
        return len(self.labels)
    
    def __getitem__(self, idx):
        item = {key: val[idx] for key, val in self.encodings.items()}
        item['labels'] = self.labels[idx]
        return item

# Sample data
train_texts = [
    "This course is excellent and very well structured",
    "Terrible teaching, could not understand anything",
    "Great content, learned so much about machine learning",
    "Waste of time, the instructor was unprepared",
    "Brilliant explanations with practical examples",
    "Boring lectures with no real-world applications",
    "The best AI course I have taken, highly recommended",
    "Poor quality videos and outdated content"
]
train_labels = [1, 0, 1, 0, 1, 0, 1, 0]  # 1=positive, 0=negative

test_texts = [
    "Amazing course, the projects were very helpful",
    "Not worth the money, too basic for advanced learners"
]
test_labels = [1, 0]

# Load pre-trained model and tokenizer
model_name = 'distilbert-base-uncased'
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(
    model_name, num_labels=2
)

print(f"Model: {model_name}")
print(f"Parameters: {sum(p.numel() for p in model.parameters()):,}")

# Create datasets
train_dataset = TextDataset(train_texts, train_labels, tokenizer)
test_dataset = TextDataset(test_texts, test_labels, tokenizer)

# Training arguments
training_args = TrainingArguments(
    output_dir='./results',
    num_train_epochs=5,
    per_device_train_batch_size=4,
    per_device_eval_batch_size=4,
    learning_rate=2e-5,          # Very low LR for fine-tuning
    weight_decay=0.01,
    logging_steps=1,
    eval_strategy='epoch',
    save_strategy='epoch',
    load_best_model_at_end=True
)

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

trainer.train()

# Predict on new texts
new_texts = [
    "This is an outstanding program with practical projects",
    "Disappointing experience, would not recommend to anyone"
]

for text in new_texts:
    inputs = tokenizer(text, return_tensors='pt', truncation=True, padding=True)
    with torch.no_grad():
        outputs = model(**inputs)
    probs = torch.nn.functional.softmax(outputs.logits, dim=-1)
    pred = 'POSITIVE' if probs[0][1] > 0.5 else 'NEGATIVE'
    conf = probs[0][1].item() if pred == 'POSITIVE' else probs[0][0].item()
    print(f"[{pred}] ({conf:.3f}) {text}")