From Fundamentals to Advanced Implementation
A comprehensive, research-grade resource covering the complete spectrum of Large Language Models - from mathematical foundations to production deployment and ethical considerations.
🚀 Quick Start
This guide progresses from fundamental concepts to advanced research frontiers. Each chapter builds upon previous knowledge with practical implementations and mathematical rigor.
This guide progresses from fundamental concepts to advanced research frontiers. Each chapter builds upon previous knowledge with practical implementations and mathematical rigor.
Your Comprehensive Roadmap to Mastering Large Language Models
| Chapter | Key Topics | Level |
|---|---|---|
| 1. LLM Revolution | History, Evolution, Current Landscape | Beginner |
| 2. Learning Pathway | Roadmap, Prerequisites, Timeline | Beginner |
| 3. Math Foundations | Linear Algebra, Probability, Calculus | Intermediate |
| 4. Programming | PyTorch, Distributed Training, GPU | Intermediate |
| 5. Neural Networks | Architectures, Backpropagation, Optimization | Intermediate |
| 6. Transformers | Self-Attention, Positional Encoding, Implementation | Advanced |
| 7. Attention | Multi-Head, Sparse, Efficient Mechanisms | Advanced |
| 8. Training Methods | Pre-training, Scaling Laws, Distributed Training | Advanced |
| 9. Fine-tuning | LoRA, Adapters, RLHF, Prompt Tuning | Advanced |
| 10. Inference | Quantization, Pruning, Speculative Decoding | Expert |
| 11. Evaluation | Benchmarks, Safety, Bias Detection | Expert |
| 12. Deployment | Serving, Scaling, Monitoring, Load Balancing | Expert |
| 13. Research | MoE, SSMs, Multimodal, Reasoning | Research |
| 14. Ethics | Bias, Fairness, Transparency, Privacy | All Levels |
| 15. Future | Scaling, Governance, AI Safety, Impact | Visionary |
Start with Chapter 1 and progress systematically through each section. Each chapter builds upon previous knowledge!
Total Learning Time: ~6-12 months | Prerequisites: Python, Basic Math
M Wasif
Large Language Models (LLMs) represent a paradigm shift in artificial intelligence, leveraging deep neural networks with billions to trillions of parameters to understand, generate, and reason with human language.
Core Characteristics:
- Scale: Model sizes ranging from millions to trillions of parameters
- Architecture: Primarily Transformer-based neural networks
- Training: Self-supervised learning on massive text corpora
- Emergent Abilities: Reasoning, code generation, mathematical problem-solving
| Era | Timeline | Key Models | Breakthroughs |
|---|---|---|---|
| Statistical | 1990-2010 | N-gram models, HMMs | Probabilistic language modeling |
| Neural | 2013-2017 | Word2Vec, LSTM, GRU | Distributed representations, sequence modeling |
| Transformer | 2017-2018 | Original Transformer | Self-attention mechanism, parallel processing |
| Pre-training | 2018-2020 | BERT, GPT-2, RoBERTa | Transfer learning, bidirectional context |
| Large-scale | 2020-2022 | GPT-3, T5, PaLM | Few-shot learning, scaling laws, reasoning |
| Modern | 2022-Present | GPT-4, Claude, Llama, Mistral | Multimodality, alignment, open-weight models |
M Wasif
# Parameter count evolution (2018-2024)
Model Scaling Timeline:
├── ELMo (2018): 94 million parameters
├── BERT-base (2018): 110 million parameters
├── GPT-1 (2018): 117 million parameters
├── GPT-2 (2019): 1.5 billion parameters
├── T5 (2020): 11 billion parameters
├── GPT-3 (2020): 175 billion parameters
├── PaLM (2022): 540 billion parameters
├── GPT-4 (2023): ~1.7 trillion parameters (estimated)
└── Gemini Ultra (2024): ~? trillion parameters
Major Model Families:
- Generative Pre-trained Transformers
- Autoregressive decoder-only architecture
- Strong few-shot learning capabilities
- Bidirectional Encoder Representations
- Masked language modeling objective
- Excellent for understanding tasks
- Text-to-Text Transfer Transformer
- Unified framework for all NLP tasks
- Encoder-decoder architecture
High-Level LLM Architecture:
Input Text → Tokenization → Embedding → Transformer Blocks → Output Head → Generated Text
│ │ │ │ │
│ │ │ └── Multi-Head Attention
│ │ │ Layer Normalization
│ │ │ Feed-Forward Networks
│ │ └── Word/Position Embeddings
│ └── Subword Tokenization (BPE, SentencePiece)
└── Prompt/Context
Essential Foundations:
- Python Programming: OOP, data structures, libraries
- Linear Algebra: Vectors, matrices, transformations
- Probability & Statistics: Distributions, Bayes theorem
- Calculus: Derivatives, gradients, chain rule
- Deep Learning Fundamentals: Neural networks, backpropagation
- PyTorch/TensorFlow: Model implementation, training loops
- NLP Basics: Tokenization, word embeddings, RNNs
- Software Engineering: Version control, testing, APIs
- Transformer Architecture: Self-attention, positional encoding
- Distributed Training: Data/model parallelism, mixed precision
- Optimization Theory: Loss landscapes, convergence analysis
- Research Methodology: Paper reading, experimental design
Learning Progression (12-Month Plan):
Month 1-2: Mathematical Foundations & Python
Month 3-4: Deep Learning Basics & PyTorch
Month 5-6: NLP Fundamentals & Classical Models
Month 7-8: Transformer Architecture & Implementation
Month 9-10: Pre-training & Fine-tuning Techniques
Month 11-12: Advanced Topics & Research Projects
| Phase | Projects | Technologies | Outcomes |
|---|---|---|---|
| Beginner | Text classification, Named Entity Recognition | scikit-learn, spaCy, BERT | Basic NLP pipeline understanding |
| Intermediate | Transformer from scratch, Fine-tuning LLMs | PyTorch, HuggingFace, WandB | Architecture mastery, training workflows |
| Advanced | Pre-training small LLM, Optimization techniques | DeepSpeed, FlashAttention, vLLM | Production-grade model development |
Vector and Matrix Operations:
Given vectors
-
Dot Product:
$x \cdot y = \sum_{i=1}^{n} x_i y_i$ -
Matrix Multiplication:
$(AB)_{ij} = \sum_{k=1}^{n} A_{ik} B_{kj}$ -
Transpose Properties:
$(A^T)_{ij} = A_{ji}$
Eigen decomposition: For square matrix
Key Distributions in LLMs:
-
Softmax Distribution:
$P(y_i) = \frac{e^{z_i}}{\sum_{j=1}^{K} e^{z_j}}$ -
Cross-Entropy Loss:
$L = -\sum_{i=1}^{C} y_i \log(\hat{y}_i)$ -
Bayes' Theorem:
$P(A|B) = \frac{P(B|A)P(A)}{P(B)}$
Entropy and KL Divergence:
Shannon Entropy:
Cross Entropy:
KL Divergence:
Perplexity:
Gradient Descent Update Rule:
where
Chain Rule for Backpropagation:
Bias-Variance Decomposition:
where:
$\text{Bias}[\hat{f}(x)] = \mathbb{E}[\hat{f}(x)] - f(x)$ $\text{Var}[\hat{f}(x)] = \mathbb{E}[(\hat{f}(x) - \mathbb{E}[\hat{f}(x)])^2]$ -
$\sigma^2$ is irreducible error
Central Limit Theorem:
Given i.i.d. random variables
Law of Large Numbers:
Singular Value Decomposition (SVD):
-
$U$ : left singular vectors (orthogonal) -
$\Sigma$ : singular values (diagonal matrix) -
$V$ : right singular vectors (orthogonal)
Low-Rank Approximation:
M Wasif
- Linear Algebra: Gilbert Strang, "Introduction to Linear Algebra"
- Probability: Sheldon Ross, "A First Course in Probability"
- Information Theory: Thomas Cover, "Elements of Information Theory"
- Optimization: Stephen Boyd, "Convex Optimization"
- Deep Learning: Ian Goodfellow, "Deep Learning"
Core Language Stack for LLM Development:
- Frameworks: PyTorch, TensorFlow, JAX
- Libraries: Transformers, NumPy, Pandas
- Use Cases: Model development, training, research
- Frameworks: CUDA, PyTorch C++ API
- Libraries: Intel MKL, NVIDIA CUDA Toolkit
- Use Cases: Kernel optimization, inference engines
Essential Libraries and Their Roles:
# Core LLM Development Stack
llm_stack = {
"deep_learning": ["PyTorch", "TensorFlow", "JAX"],
"transformer_libs": ["HuggingFace Transformers", "FairSeq", "Megatron-LM"],
"numerical_computing": ["NumPy", "SciPy", "CuPy"],
"data_processing": ["Pandas", "PyArrow", "Dask"],
"experiment_tracking": ["Weights & Biases", "MLflow", "TensorBoard"],
"distributed_training": ["DeepSpeed", "PyTorch DDP", "Horovod"]
}
Core Tensor Operations:
import torch
import torch.nn as nn
import torch.nn.functional as F
# Basic tensor operations
x = torch.randn(2, 3) # 2x3 tensor
y = torch.ones(2, 3) # 2x3 tensor of ones
# Common operations
z = x + y # Element-wise addition
z = torch.matmul(x, y.T) # Matrix multiplication
z = F.softmax(x, dim=-1) # Softmax activation
# Automatic differentiation
x = torch.tensor(2.0, requires_grad=True)
y = x ** 2
y.backward() # Compute gradients
print(x.grad) # dy/dx = 2x = 4.0
Neural Network Module:
class SimpleNN(nn.Module):
def __init__(self, input_size, hidden_size, output_size):
super().__init__()
self.fc1 = nn.Linear(input_size, hidden_size)
self.fc2 = nn.Linear(hidden_size, output_size)
self.dropout = nn.Dropout(0.1)
def forward(self, x):
x = F.relu(self.fc1(x))
x = self.dropout(x)
x = self.fc2(x)
return x
Data Parallelism:
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
# Initialize distributed training
def setup(rank, world_size):
dist.init_process_group("nccl", rank=rank, world_size=world_size)
torch.cuda.set_device(rank)
# Wrap model with DDP
model = SimpleNN(100, 50, 10)
model = DDP(model, device_ids=[rank])
Mixed Precision Training:
from torch.cuda.amp import autocast, GradScaler
scaler = GradScaler()
for input, target in dataloader:
optimizer.zero_grad()
with autocast():
output = model(input)
loss = criterion(output, target)
scaler.scale(loss).backward()
scaler.step(optimizer)
scaler.update()
CUDA Fundamentals:
# GPU memory management
x = torch.randn(1000, 1000).cuda() # Move to GPU
y = torch.randn(1000, 1000).cuda()
# GPU operations
z = torch.matmul(x, y) # Executed on GPU
# Memory statistics
print(torch.cuda.memory_allocated()) # Current memory usage
print(torch.cuda.max_memory_allocated()) # Peak memory usage
# Synchronization
torch.cuda.synchronize() # Wait for GPU operations to complete
M Wasif
From Biological Neurons to Artificial Neurons:
A single artificial neuron implements:
where:
-
$x_i$ : Input features -
$w_i$ : Learnable weights -
$b$ : Bias term -
$f$ : Activation function
| Function | Formula | Derivative | Use Cases |
|---|---|---|---|
| Sigmoid | Binary classification, gates | ||
| Tanh | Hidden layers, RNNs | ||
| ReLU |
|
Most hidden layers | |
| GELU | Complex | Transformers, BERT, GPT | |
| Softmax | Output layer, attention |
Chain Rule Formulation:
Given a neural network with loss
where:
-
$z_j^{(l)} = \sum_i w_{ij}^{(l)} a_i^{(l-1)} + b_j^{(l)}$ (pre-activation) -
$a_j^{(l)} = f(z_j^{(l)})$ (activation) -
$\delta_j^{(l)} = \frac{\partial L}{\partial z_j^{(l)}}$ (error term)
Backward Pass Recursion:
Common Loss Functions in LLMs:
Cross-Entropy Loss (Classification):
Mean Squared Error (Regression):
Binary Cross-Entropy:
Stochastic Gradient Descent (SGD):
Momentum SGD:
Adam Optimizer:
L1/L2 Regularization:
Dropout:
During training:
where
Batch Normalization:
Convolutional Neural Networks (CNNs):
class CNN(nn.Module): def __init__(self): super().__init__() self.conv1 = nn.Conv2d(1, 32, 3, padding=1) self.conv2 = nn.Conv2d(32, 64, 3, padding=1) self.pool = nn.MaxPool2d(2) self.fc = nn.Linear(64 * 7 * 7, 10)def forward(self, x): x = self.pool(F.relu(self.conv1(x))) x = self.pool(F.relu(self.conv2(x))) x = x.view(-1, 64 * 7 * 7) x = self.fc(x) return x
Recurrent Neural Networks (RNNs):
class SimpleRNN(nn.Module):
def __init__(self, input_size, hidden_size, output_size):
super().__init__()
self.hidden_size = hidden_size
self.i2h = nn.Linear(input_size + hidden_size, hidden_size)
self.i2o = nn.Linear(input_size + hidden_size, output_size)
def forward(self, input, hidden):
combined = torch.cat((input, hidden), 1)
hidden = torch.tanh(self.i2h(combined))
output = self.i2o(combined)
return output, hidden
Complete Transformer Architecture:
Transformer Architecture:
Input → Token Embedding → Positional Encoding → Encoder Stack → Decoder Stack → Output
│ │ │ │
│ │ ├── Multi-Head Self-Attention
│ │ ├── Feed-Forward Network
│ │ ├── Layer Normalization
│ │ └── Residual Connections
│ └── sin/cos functions or learned
└── WordPiece/BPE tokenization
Scaled Dot-Product Attention:
where:
-
$Q$ : Query matrix ($n \times d_k$ ) -
$K$ : Key matrix ($m \times d_k$ ) -
$V$ : Value matrix ($m \times d_v$ ) -
$d_k$ : Dimension of key vectors
Multi-Head Attention:
where
Sinusoidal Positional Encoding:
where:
-
$pos$ : Position in the sequence -
$i$ : Dimension index -
$d_{\text{model}}$ : Model dimension
Position-wise Feed-Forward Network:
In modern transformers, GELU activation is often used:
where
LayerNorm Operation:
where:
$\mu = \frac{1}{d} \sum_{i=1}^d x_i$ $\sigma^2 = \frac{1}{d} \sum_{i=1}^d (x_i - \mu)^2$ -
$\gamma, \beta$ : Learnable parameters
M Wasif
import torch
import torch.nn as nn
import math
class MultiHeadAttention(nn.Module):
def __init__(self, d_model, num_heads):
super().__init__()
self.d_model = d_model
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, seq_len = q.size(0), q.size(1)
# Linear projections and reshape for multi-head
Q = self.w_q(q).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
K = self.w_k(k).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
V = self.w_v(v).view(batch_size, seq_len, self.num_heads, self.d_k).transpose(1, 2)
# Scaled dot-product attention
scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)
if mask is not None:
scores = scores.masked_fill(mask == 0, -1e9)
attn_weights = torch.softmax(scores, dim=-1)
attn_output = torch.matmul(attn_weights, V)
# Concatenate heads and put through final linear layer
attn_output = attn_output.transpose(1, 2).contiguous().view(
batch_size, seq_len, self.d_model
)
return self.w_o(attn_output)
class PositionWiseFFN(nn.Module):
def __init__(self, d_model, d_ff):
super().__init__()
self.linear1 = nn.Linear(d_model, d_ff)
self.linear2 = nn.Linear(d_ff, d_model)
self.activation = nn.GELU()
def forward(self, x):
return self.linear2(self.activation(self.linear1(x)))
class TransformerEncoderLayer(nn.Module):
def __init__(self, d_model, num_heads, d_ff, dropout=0.1):
super().__init__()
self.self_attn = MultiHeadAttention(d_model, num_heads)
self.ffn = PositionWiseFFN(d_model, d_ff)
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 with residual connection and layer norm
attn_output = self.self_attn(x, x, x, mask)
x = self.norm1(x + self.dropout(attn_output))
# Feed-forward with residual connection and layer norm
ffn_output = self.ffn(x)
x = self.norm2(x + self.dropout(ffn_output))
return x
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, dtype=torch.float).unsqueeze(1)
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).transpose(0, 1)
self.register_buffer('pe', pe)
def forward(self, x):
return x + self.pe[:x.size(0), :]
Cross-Attention Mechanism:
In decoder layers, cross-attention connects encoder outputs to decoder inputs:
Types of Attention Masks:
# Causal masking (autoregressive models)
def causal_mask(size):
mask = torch.triu(torch.ones(size, size), diagonal=1)
return mask == 0 # Lower triangular matrix
# Padding mask
def padding_mask(input_ids, pad_token_id=0):
return (input_ids != pad_token_id).unsqueeze(1).unsqueeze(2)
# Combined mask for decoder
def combined_mask(tgt, pad_token_id=0):
causal_mask = torch.triu(torch.ones(tgt.size(1), tgt.size(1)), diagonal=1)
padding_mask = (tgt != pad_token_id).unsqueeze(1)
return padding_mask & (causal_mask == 0)
Architectural Improvements:
| Variant | Key Innovation | Use Cases |
|---|---|---|
| ALiBi | Relative positional encoding without learned parameters | Long sequence modeling |
| RoPE | Rotary Position Embeddings | Llama, GPT-NeoX |
| FlashAttention | IO-aware attention algorithm | Long context, memory efficiency |
| SwiGLU | Gated linear unit activation | PaLM, Llama 2 |
| Grouped Query Attention | Shared key-value heads across query heads | Llama 2, inference optimization |
General Attention Formulation:
Given queries
where
| Type | Formula | Complexity | Use Cases |
|---|---|---|---|
| Full Self-Attention | Standard transformers, short sequences | ||
| Linear Attention | Long sequences, memory constraints | ||
| Local Attention | Window-based computation | Images, local dependencies | |
| Sparse Attention | Fixed/learned patterns | Very long sequences | |
| Low-Rank Attention | Projected attention matrices | Approximation, efficiency |
M Wasif
Detailed Multi-Head Formulation:
For head
Parameter Count:
Total parameters =
Linformer (Low-Rank Projection):
Complexity reduces from
Performer (Fast Attention via Orthogonal Random Features):
where
class EfficientAttention(nn.Module):
def __init__(self, d_model, num_heads, feature_dim=256):
super().__init__()
self.d_model = d_model
self.num_heads = num_heads
self.feature_dim = feature_dim
# Random features for approximation
self.w = nn.Parameter(torch.randn(feature_dim, d_model // num_heads))
def random_features(self, x):
# Random feature map for kernel approximation
x_proj = F.linear(x, self.w)
return torch.exp(x_proj - x_proj.max(dim=-1, keepdim=True)[0])
def forward(self, q, k, v):
batch_size, seq_len = q.size(0), q.size(1)
# Apply random feature maps
q_features = self.random_features(q)
k_features = self.random_features(k)
# Linear attention computation
kv_matrix = torch.bmm(k_features.transpose(1,2), v)
attention_output = torch.bmm(q_features, kv_matrix)
return attention_output
Fixed Patterns:
def fixed_sparse_attention_mask(seq_len, pattern_type="strided"):
mask = torch.zeros(seq_len, seq_len)
if pattern_type == "strided":
# Every other position attends to previous 8 positions
for i in range(seq_len):
start = max(0, i - 8)
mask[i, start:i+1] = 1
if i % 2 == 0 and i > 0:
mask[i, i-1] = 1
elif pattern_type == "dilated":
# Dilated attention pattern
for i in range(seq_len):
for j in range(0, i+1, 2): # Attend to every other position
if j <= i:
mask[i, j] = 1
return mask.bool()
Sliding Window Attention:
Each position only attends to
Block-Sparse Attention:
def block_sparse_attention(q, k, v, block_size=64, num_blocks=4): batch_size, seq_len, d_model = q.shape# Reshape into blocks q_blocks = q.view(batch_size, seq_len // block_size, block_size, d_model) k_blocks = k.view(batch_size, seq_len // block_size, block_size, d_model) v_blocks = v.view(batch_size, seq_len // block_size, block_size, d_model) output = torch.zeros_like(q) # Each block attends to previous num_blocks blocks for block_idx in range(seq_len // block_size): start_block = max(0, block_idx - num_blocks + 1) attended_blocks = range(start_block, block_idx + 1) # Compute attention within attended blocks # ... implementation details ... return output
Autoregressive (Causal) Language Modeling:
Masked Language Modeling (BERT-style):
where
Permutation Language Modeling (XLNet):
Kaplan Scaling Laws:
where:
-
$N$ : Model parameters -
$D$ : Training tokens -
$N_c, D_c$ : Critical values -
$\alpha_N, \alpha_D$ : Scaling exponents -
$L_\infty$ : Irreducible loss
Chinchilla Optimal Scaling:
For compute budget
Data Parallelism:
# PyTorch DDP Example import torch.distributed as dist from torch.nn.parallel import DistributedDataParallel as DDPdef train_ddp(rank, world_size): # Initialize process group dist.init_process_group("nccl", rank=rank, world_size=world_size)
# Model and data model = TransformerModel().to(rank) ddp_model = DDP(model, device_ids=[rank]) # Training loop for batch in dataloader: loss = ddp_model(batch) loss.backward() optimizer.step()
Model Parallelism:
class ModelParallelTransformer(nn.Module): def __init__(self, num_devices): super().__init__() self.num_devices = num_devices self.layers = nn.ModuleList([ TransformerLayer().to(f"cuda:{i % num_devices}") for i in range(num_layers) ])def forward(self, x): for i, layer in enumerate(self.layers): device = f"cuda:{i % self.num_devices}" x = x.to(device) x = layer(x) return x
Pipeline Parallelism:
from torch.distributed.pipeline.sync import Pipemodel = LargeTransformer() model_parts = split_model_into_partitions(model, num_partitions=4)
model_pipe = Pipe(model_parts, chunks=8) # Micro-batches
output = model_pipe(input) loss = criterion(output, target) loss.backward()
FP16/FP32 Mixed Precision:
from torch.cuda.amp import autocast, GradScalerscaler = GradScaler()
for input, target in dataloader: optimizer.zero_grad()
with autocast(): output = model(input) loss = criterion(output, target) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()
BF16 Support:
# BF16 has better dynamic range than FP16 torch.set_float32_matmul_precision('medium') # Use TF32 for matmuls
model = model.to(torch.bfloat16) for input, target in dataloader: input = input.to(torch.bfloat16) output = model(input) # No need for gradient scaling with BF16
AdamW Optimizer:
Learning Rate Schedules:
Linear Warmup + Cosine Decay:
def get_cosine_schedule_with_warmup(optimizer, num_warmup_steps, num_training_steps): def lr_lambda(current_step): if current_step < num_warmup_steps: return float(current_step) / float(max(1, num_warmup_steps)) progress = float(current_step - num_warmup_steps) / float(max(1, num_training_steps - num_warmup_steps)) return max(0.0, 0.5 * (1.0 + math.cos(math.pi * progress)))return torch.optim.lr_scheduler.LambdaLR(optimizer, lr_lambda)
Weight Decay:
Gradient Clipping:
# Global gradient clipping torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
for param in model.parameters(): if param.grad is not None: param.grad.data.clamp_(-1.0, 1.0)
Stochastic Depth:
class StochasticDepth(nn.Module): def __init__(self, drop_prob): super().__init__() self.drop_prob = drop_probdef forward(self, x, layer): if self.training and torch.rand(1) < self.drop_prob: return x # Skip layer return layer(x)
Standard Fine-tuning Process:
def full_finetune(model, train_dataloader, num_epochs=3): optimizer = AdamW(model.parameters(), lr=5e-5)for epoch in range(num_epochs): model.train() for batch in train_dataloader: outputs = model(**batch) loss = outputs.loss loss.backward() optimizer.step() optimizer.zero_grad()
LoRA Mathematical Formulation:
where
class LoRALayer(nn.Module):
def __init__(self, base_layer, rank=8, alpha=16):
super().__init__()
self.base_layer = base_layer
self.rank = rank
self.alpha = alpha
# LoRA matrices
self.lora_A = nn.Parameter(torch.randn(base_layer.in_features, rank))
self.lora_B = nn.Parameter(torch.zeros(rank, base_layer.out_features))
def forward(self, x):
base_output = self.base_layer(x)
lora_output = x @ self.lora_A @ self.lora_B
return base_output + (self.alpha / self.rank) * lora_output
def apply_lora_to_linear_layers(model, rank=8):
for name, module in model.named_children():
if isinstance(module, nn.Linear):
# Replace with LoRA layer
setattr(model, name, LoRALayer(module, rank=rank))
else:
apply_lora_to_linear_layers(module, rank)
M Wasif
class Adapter(nn.Module):
def __init__(self, dim, adapter_dim=64):
super().__init__()
self.down_proj = nn.Linear(dim, adapter_dim)
self.up_proj = nn.Linear(adapter_dim, dim)
self.activation = nn.GELU()
def forward(self, x):
return x + self.up_proj(self.activation(self.down_proj(x)))
class TransformerWithAdapters(nn.Module):
def __init__(self, base_transformer):
super().__init__()
self.base = base_transformer
# Add adapters after attention and FFN
for layer in self.base.layers:
layer.attention_adapter = Adapter(layer.self_attn.d_model)
layer.ffn_adapter = Adapter(layer.ffn.d_model)
def forward(self, x):
for layer in self.base.layers:
# Original attention
attn_output = layer.self_attn(x)
x = layer.attention_adapter(attn_output)
# Original FFN
ffn_output = layer.ffn(x)
x = layer.ffn_adapter(ffn_output)
return x
class PromptTuning(nn.Module):
def __init__(self, model, prompt_length=20):
super().__init__()
self.model = model
self.prompt_length = prompt_length
self.prompt_embeddings = nn.Parameter(
torch.randn(prompt_length, model.config.hidden_size)
)
def forward(self, input_ids, attention_mask=None):
batch_size = input_ids.shape[0]
# Get original embeddings
inputs_embeds = self.model.get_input_embeddings()(input_ids)
# Concatenate prompt embeddings
prompt_embeds = self.prompt_embeddings.unsqueeze(0).repeat(batch_size, 1, 1)
inputs_embeds = torch.cat([prompt_embeds, inputs_embeds], dim=1)
# Adjust attention mask
if attention_mask is not None:
prompt_mask = torch.ones(batch_size, self.prompt_length).to(attention_mask.device)
attention_mask = torch.cat([prompt_mask, attention_mask], dim=1)
return self.model(inputs_embeds=inputs_embeds, attention_mask=attention_mask)
class PTuning(nn.Module):
def __init__(self, model, prompt_length=20, prompt_hidden_size=512):
super().__init__()
self.model = model
self.prompt_length = prompt_length
# LSTM for prompt generation
self.lstm = nn.LSTM(
input_size=model.config.hidden_size,
hidden_size=prompt_hidden_size,
num_layers=2,
bidirectional=True,
batch_first=True
)
self.mlp = nn.Sequential(
nn.Linear(2 * prompt_hidden_size, model.config.hidden_size),
nn.ReLU(),
nn.Linear(model.config.hidden_size, model.config.hidden_size)
)
def forward(self, input_ids, attention_mask=None):
batch_size = input_ids.shape[0]
# Generate continuous prompts
prompt_tokens = torch.arange(self.prompt_length).unsqueeze(0).repeat(batch_size, 1)
prompt_embeds = self.model.get_input_embeddings()(prompt_tokens)
# Process through LSTM and MLP
lstm_out, _ = self.lstm(prompt_embeds)
continuous_prompts = self.mlp(lstm_out)
# Get original embeddings and concatenate
inputs_embeds = self.model.get_input_embeddings()(input_ids)
inputs_embeds = torch.cat([continuous_prompts, inputs_embeds], dim=1)
# Adjust attention mask
if attention_mask is not None:
prompt_mask = torch.ones(batch_size, self.prompt_length).to(attention_mask.device)
attention_mask = torch.cat([prompt_mask, attention_mask], dim=1)
return self.model(inputs_embeds=inputs_embeds, attention_mask=attention_mask)
Instruction Format:
instruction_prompt = """ Below is an instruction that describes a task. Write a response that appropriately completes the request.{instruction}
"""
Supervised Fine-tuning (SFT):
def instruction_tuning_loss(model, batch): """Compute loss for instruction following""" instructions = batch["instruction"] responses = batch["response"]# Format input with instruction template formatted_inputs = [ f"Instruction: {inst}\n\nResponse: {resp}" for inst, resp in zip(instructions, responses) ] # Tokenize and compute loss inputs = tokenizer(formatted_inputs, return_tensors="pt", padding=True, truncation=True) outputs = model(**inputs, labels=inputs["input_ids"]) return outputs.loss
Three-Stage RLHF Process:
# Stage 1: Supervised Fine-tuning
sft_trainer = SFTTrainer(
model=base_model,
train_dataset=instruction_data,
formatting_func=format_instruction
)
# Stage 2: Reward Model Training
class RewardModel(nn.Module):
def __init__(self, base_model):
super().__init__()
self.transformer = base_model
self.value_head = nn.Linear(base_model.config.hidden_size, 1)
def forward(self, input_ids, attention_mask):
outputs = self.transformer(input_ids, attention_mask=attention_mask)
last_hidden_state = outputs.last_hidden_state
# Use the EOS token for reward prediction
eos_token_hidden = last_hidden_state[:, -1, :]
reward = self.value_head(eos_token_hidden)
return reward
# Stage 3: PPO Training
def ppo_training_step(policy_model, reward_model, prompts):
# Generate responses with current policy
with torch.no_grad():
old_responses = policy_model.generate(prompts)
old_rewards = reward_model(old_responses)
# Update policy using PPO
# ... PPO implementation details ...
| Metric | Formula | Interpretation |
|---|---|---|
| Perplexity | Lower is better | |
| BLEU Score | BP |
0-100, higher better |
| ROUGE Score | Recall-oriented | |
| Accuracy | Classification tasks |
M Wasif
Mathematical Foundation of Quantization:
For floating-point tensor
where
Dequantization:
import torch
import torch.quantization
def post_training_quantization(model, calibration_loader):
# Set model to evaluation mode
model.eval()
# Prepare model for quantization
model.qconfig = torch.quantization.get_default_qconfig('fbgemm')
torch.quantization.prepare(model, inplace=True)
# Calibrate with sample data
with torch.no_grad():
for batch in calibration_loader:
model(batch)
# Convert to quantized model
torch.quantization.convert(model, inplace=True)
return model
# Example usage for linear layer quantization
class QuantizedLinear(torch.nn.Module):
def __init__(self, in_features, out_features):
super().__init__()
self.in_features = in_features
self.out_features = out_features
self.weight = torch.nn.Parameter(torch.randn(out_features, in_features))
self.weight_scale = torch.nn.Parameter(torch.tensor(1.0))
self.weight_zero_point = torch.nn.Parameter(torch.tensor(0))
def forward(self, x):
# Quantize weights
weight_q = torch.quantize_per_tensor(
self.weight, self.weight_scale, self.weight_zero_point, torch.qint8
)
# Dequantize for computation (in real scenario, use quantized ops)
weight_dequant = weight_q.dequantize()
return torch.nn.functional.linear(x, weight_dequant)
class QATLinear(torch.nn.Module):
def __init__(self, in_features, out_features):
super().__init__()
self.in_features = in_features
self.out_features = out_features
self.weight = torch.nn.Parameter(torch.randn(out_features, in_features))
# Quantization stubs
self.weight_quant = torch.quantization.QuantStub()
self.weight_dequant = torch.quantization.DeQuantStub()
def forward(self, x):
# Simulate quantization during training
weight_quantized = self.weight_quant(self.weight)
weight = self.weight_dequant(weight_quantized)
return torch.nn.functional.linear(x, weight)
def prepare_qat(model):
# Fuse layers for better quantization
torch.quantization.fuse_modules(model, [['conv', 'bn', 'relu']], inplace=True)
# Prepare for QAT
model.qconfig = torch.quantization.get_default_qat_qconfig('fbgemm')
torch.quantization.prepare_qat(model, inplace=True)
return model
def mixed_precision_quantization(model, sensitivity_analysis):
"""Apply different precision based on layer sensitivity"""
quantization_config = {}
for name, module in model.named_modules():
if isinstance(module, torch.nn.Linear):
sensitivity = sensitivity_analysis[name]
if sensitivity < 0.1: # Low sensitivity
# Use 4-bit quantization
config = torch.quantization.QConfig(
activation=torch.quantization.MinMaxObserver.with_args(dtype=torch.quint4),
weight=torch.quantization.MinMaxObserver.with_args(dtype=torch.qint4)
)
elif sensitivity < 0.3: # Medium sensitivity
# Use 8-bit quantization
config = torch.quantization.default_qconfig
else: # High sensitivity
# Keep in FP16
config = None
quantization_config[name] = config
return quantization_config
Magnitude-Based Pruning:
Remove weights with smallest magnitudes:
$W_{\text{pruned}}[i,j] = \begin{cases} 0 & \text{if } |W[i,j]| < \theta \\ W[i,j] & \text{otherwise} \end{cases}$
class MagnitudePruning:
def __init__(self, pruning_rate=0.2):
self.pruning_rate = pruning_rate
def apply(self, model):
all_weights = []
for name, param in model.named_parameters():
if 'weight' in name and len(param.shape) >= 2: # Only weight matrices
all_weights.append(param.data.abs().view(-1))
# Calculate global threshold
all_weights = torch.cat(all_weights)
threshold = torch.quantile(all_weights, self.pruning_rate)
# Apply pruning
for name, param in model.named_parameters():
if 'weight' in name and len(param.shape) >= 2:
mask = param.data.abs() > threshold
param.data *= mask.float()
return model
def iterative_pruning(model, dataloader, total_iterations=10, target_sparsity=0.8):
"""Iterative pruning with fine-tuning"""
initial_sparsity = 0.0
sparsity_increment = (target_sparsity - initial_sparsity) / total_iterations
for iteration in range(total_iterations):
# Prune
current_sparsity = initial_sparsity + (iteration + 1) * sparsity_increment
pruning = MagnitudePruning(pruning_rate=current_sparsity)
model = pruning.apply(model)
# Fine-tune
fine_tune_model(model, dataloader, epochs=1)
return model
Structured Pruning:
class StructuredPruning:
def __init__(self, pruning_method='l1'):
self.pruning_method = pruning_method
def compute_importance(self, weight):
if self.pruning_method == 'l1':
return torch.norm(weight, p=1, dim=1) # L1 norm of rows
elif self.pruning_method == 'l2':
return torch.norm(weight, p=2, dim=1) # L2 norm of rows
def prune_neurons(self, model, pruning_rate):
for name, module in model.named_modules():
if isinstance(module, torch.nn.Linear):
importance = self.compute_importance(module.weight)
# Calculate threshold
threshold = torch.quantile(importance, pruning_rate)
# Create mask for important neurons
mask = importance > threshold
# Apply mask to output dimension
module.weight.data = module.weight.data[mask, :]
if module.bias is not None:
module.bias.data = module.bias.data[mask]
# Update output features
module.out_features = mask.sum().item()
return model
Distillation Loss:
where
class KnowledgeDistillationLoss(torch.nn.Module):
def __init__(self, temperature=4.0, alpha=0.7):
super().__init__()
self.temperature = temperature
self.alpha = alpha
self.ce_loss = torch.nn.CrossEntropyLoss()
self.kl_loss = torch.nn.KLDivLoss(reduction='batchmean')
def forward(self, student_logits, teacher_logits, labels):
# Soften the probabilities
student_probs = torch.nn.functional.log_softmax(student_logits / self.temperature, dim=-1)
teacher_probs = torch.nn.functional.softmax(teacher_logits / self.temperature, dim=-1)
# Calculate distillation loss
distill_loss = self.kl_loss(student_probs, teacher_probs) * (self.temperature ** 2)
# Calculate student loss
student_loss = self.ce_loss(student_logits, labels)
# Combined loss
return self.alpha * student_loss + (1 - self.alpha) * distill_loss
def distill_training(student, teacher, dataloader, epochs=10):
criterion = KnowledgeDistillationLoss()
optimizer = torch.optim.Adam(student.parameters())
for epoch in range(epochs):
for batch in dataloader:
inputs, labels = batch
# Get teacher predictions (no gradient)
with torch.no_grad():
teacher_logits = teacher(inputs)
# Student forward pass
student_logits = student(inputs)
# Compute distillation loss
loss = criterion(student_logits, teacher_logits, labels)
# Backward pass
optimizer.zero_grad()
loss.backward()
optimizer.step()
class SpeculativeDecoding:
def __init__(self, target_model, draft_model, max_speculative_tokens=5):
self.target_model = target_model
self.draft_model = draft_model
self.max_speculative_tokens = max_speculative_tokens
def generate(self, prompt, max_length=100):
sequences = prompt
draft_sequences = prompt
while len(sequences[0]) < max_length:
# Draft phase: generate multiple tokens quickly
draft_tokens = []
for _ in range(self.max_speculative_tokens):
draft_logits = self.draft_model(draft_sequences)
next_token = torch.argmax(draft_logits[:, -1, :], dim=-1)
draft_tokens.append(next_token)
draft_sequences = torch.cat([draft_sequences, next_token.unsqueeze(-1)], dim=-1)
# Verification phase: check with target model
target_logits = self.target_model(draft_sequences)
target_probs = torch.softmax(target_logits, dim=-1)
# Verify and accept tokens
accepted_tokens = self._verify_tokens(draft_tokens, target_probs)
if len(accepted_tokens) > 0:
sequences = torch.cat([sequences] + accepted_tokens, dim=-1)
else:
# If no tokens accepted, generate one from target model
next_token = torch.argmax(target_probs[:, -1, :], dim=-1)
sequences = torch.cat([sequences, next_token.unsqueeze(-1)], dim=-1)
return sequences
def _verify_tokens(self, draft_tokens, target_probs):
accepted_tokens = []
for i, token in enumerate(draft_tokens):
target_prob = target_probs[:, i, token]
draft_prob = # probability from draft model
# Acceptance criteria
if torch.rand(1) < torch.min(torch.tensor(1.0), target_prob / draft_prob):
accepted_tokens.append(token.unsqueeze(-1))
else:
break
return accepted_tokens
class KVCache:
def __init__(self, batch_size, max_length, num_heads, head_dim):
self.k_cache = torch.zeros(batch_size, max_length, num_heads, head_dim)
self.v_cache = torch.zeros(batch_size, max_length, num_heads, head_dim)
self.current_length = 0
def update(self, new_k, new_v):
batch_size, seq_len = new_k.shape[0], new_k.shape[1]
# Append new keys and values to cache
self.k_cache[:, self.current_length:self.current_length+seq_len] = new_k
self.v_cache[:, self.current_length:self.current_length+seq_len] = new_v
self.current_length += seq_len
return (self.k_cache[:, :self.current_length],
self.v_cache[:, :self.current_length])
class EfficientTransformerInference:
def __init__(self, model, max_cache_length=2048):
self.model = model
self.kv_cache = None
self.max_cache_length = max_cache_length
def generate(self, input_ids, max_length=100):
if self.kv_cache is None:
self._initialize_cache(input_ids.shape[0])
sequences = input_ids
for _ in range(max_length - input_ids.shape[1]):
# Only process the last token for autoregressive generation
if sequences.shape[1] > 1:
current_input = sequences[:, -1:]
else:
current_input = sequences
# Forward pass with KV cache
outputs = self.model(
current_input,
past_key_values=self.kv_cache,
use_cache=True
)
# Update KV cache
self.kv_cache = outputs.past_key_values
# Get next token
next_token_logits = outputs.logits[:, -1, :]
next_token = torch.argmax(next_token_logits, dim=-1, keepdim=True)
sequences = torch.cat([sequences, next_token], dim=-1)
return sequences
def _initialize_cache(self, batch_size):
num_heads = self.model.config.num_attention_heads
head_dim = self.model.config.hidden_size // num_heads
self.kv_cache = [
(torch.zeros(batch_size, self.max_cache_length, num_heads, head_dim),
torch.zeros(batch_size, self.max_cache_length, num_heads, head_dim))
for _ in range(self.model.config.num_hidden_layers)
]
class ContinuousBatchingInference:
def __init__(self, model, max_batch_size=32):
self.model = model
self.max_batch_size = max_batch_size
self.requests = []
self.kv_caches = {}
def add_request(self, prompt, request_id):
self.requests.append({
'id': request_id,
'prompt': prompt,
'tokens': [prompt],
'finished': False
})
# Initialize KV cache for this request
self.kv_caches[request_id] = self._initialize_kv_cache()
def process_batch(self):
# Group requests that are ready for next token
batch_requests = []
batch_inputs = []
batch_kv_caches = []
for req in self.requests:
if not req['finished']:
batch_requests.append(req)
batch_inputs.append(req['tokens'][-1]) # Last token
batch_kv_caches.append(self.kv_caches[req['id']])
if not batch_requests:
return
# Process batch
batch_outputs = self._process_batch_inference(
batch_inputs, batch_kv_caches
)
# Update requests
for i, req in enumerate(batch_requests):
next_token = batch_outputs[i]
req['tokens'].append(next_token)
# Check for completion
if next_token == self.model.config.eos_token_id:
req['finished'] = True
def _process_batch_inference(self, batch_inputs, batch_kv_caches):
# Implement batched inference with separate KV caches
# This is a simplified version
batch_tensor = torch.stack(batch_inputs)
# Process through model (would need custom implementation for separate KV caches)
outputs = self.model(batch_tensor)
next_tokens = torch.argmax(outputs.logits[:, -1, :], dim=-1)
return next_tokens
M Wasif
Perplexity:
Bits per Character (BPC):
def calculate_perplexity(model, tokenizer, text_dataset):
total_log_likelihood = 0
total_tokens = 0
model.eval()
with torch.no_grad():
for text in text_dataset:
inputs = tokenizer(text, return_tensors='pt')
outputs = model(**inputs, labels=inputs['input_ids'])
# Negative log likelihood
nll = outputs.loss * inputs['input_ids'].numel()
total_log_likelihood += nll.item()
total_tokens += inputs['input_ids'].numel()
avg_nll = total_log_likelihood / total_tokens
perplexity = torch.exp(torch.tensor(avg_nll))
return perplexity.item()
def calculate_bits_per_character(model, tokenizer, text):
"""Calculate bits per character for text generation models"""
total_bits = 0
total_chars = 0
# Tokenize and process text
tokens = tokenizer.encode(text)
for i in range(1, len(tokens)):
# Get probability of next token
input_ids = torch.tensor([tokens[:i]])
with torch.no_grad():
outputs = model(input_ids)
probs = torch.softmax(outputs.logits[0, -1], dim=-1)
token_prob = probs[tokens[i]].item()
# Convert to bits
bits = -math.log2(token_prob) if token_prob > 0 else float('inf')
total_bits += bits
total_chars = len(text)
return total_bits / total_chars
class GLUEEvaluator:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.tasks = {
'cola': self.evaluate_cola,
'sst2': self.evaluate_sst2,
'mrpc': self.evaluate_mrpc,
'qqp': self.evaluate_qqp,
'mnli': self.evaluate_mnli
}
def evaluate_all(self, datasets):
results = {}
for task_name, dataset in datasets.items():
if task_name in self.tasks:
accuracy = self.tasks[task_name](dataset)
results[task_name] = accuracy
return results
def evaluate_sst2(self, dataset):
"""Sentiment classification accuracy"""
correct = 0
total = 0
for text, label in dataset:
inputs = self.tokenizer(text, return_tensors='pt', truncation=True)
with torch.no_grad():
outputs = self.model(**inputs)
prediction = torch.argmax(outputs.logits, dim=-1).item()
if prediction == label:
correct += 1
total += 1
return correct / total
class MMLUEvaluator:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def evaluate_subject(self, subject_name, test_data):
"""Evaluate on specific MMLU subject"""
correct = 0
total = 0
for question_data in test_data:
question = question_data['question']
choices = question_data['choices']
answer = question_data['answer']
# Format as multiple choice
prompt = self._format_mmlu_prompt(question, choices)
# Get model probabilities for each choice
choice_probs = []
for choice in choices:
full_prompt = prompt + choice
inputs = self.tokenizer(full_prompt, return_tensors='pt')
with torch.no_grad():
outputs = self.model(**inputs)
# Use last token probability
logits = outputs.logits[0, -1, :]
prob = torch.softmax(logits, dim=-1)[self.tokenizer.eos_token_id]
choice_probs.append(prob.item())
# Predict highest probability choice
predicted = np.argmax(choice_probs)
if predicted == answer:
correct += 1
total += 1
return correct / total
class GSM8KEvaluator:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def evaluate(self, dataset):
correct = 0
total = 0
for problem_data in dataset:
problem = problem_data['question']
answer = problem_data['answer']
# Use chain-of-thought prompting
cot_prompt = f"Q: {problem}\nA: Let's think step by step."
# Generate reasoning
inputs = self.tokenizer(cot_prompt, return_tensors='pt')
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_length=len(inputs['input_ids'][0]) + 200,
temperature=0.7,
do_sample=True
)
generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
# Extract final answer
final_answer = self._extract_answer(generated_text)
if self._compare_answers(final_answer, answer):
correct += 1
total += 1
return correct / total
class HumanEvalEvaluator:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def evaluate_code_generation(self, problems):
"""Evaluate on HumanEval code generation benchmark"""
pass_at_k = {1: 0, 5: 0, 10: 0}
total = len(problems)
for problem in problems:
prompt = problem['prompt']
test_cases = problem['test']
# Generate multiple solutions
solutions = self._generate_solutions(prompt, num_samples=10)
# Test solutions
for k in [1, 5, 10]:
passed = any(self._test_solution(sol, test_cases)
for sol in solutions[:k])
if passed:
pass_at_k[k] += 1
# Calculate pass@k scores
return {k: count/total for k, count in pass_at_k.items()}
class SafetyEvaluator:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
self.harmful_prompts = self._load_harmful_prompts()
def evaluate_refusal_rate(self):
"""Measure how often model refuses harmful requests"""
refusal_count = 0
total_prompts = len(self.harmful_prompts)
for prompt in self.harmful_prompts:
inputs = self.tokenizer(prompt, return_tensors='pt')
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_length=len(inputs['input_ids'][0]) + 50,
temperature=0.7
)
response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
if self._is_refusal(response):
refusal_count += 1
return refusal_count / total_prompts
def evaluate_truthfulness(self, truthfulqa_dataset):
"""Evaluate using TruthfulQA benchmark"""
correct = 0
total = 0
for qa_pair in truthfulqa_dataset:
question = qa_pair['question']
correct_answer = qa_pair['correct_answer']
incorrect_answers = qa_pair['incorrect_answers']
# Test if model prefers correct answer
preference = self._measure_answer_preference(
question, correct_answer, incorrect_answers
)
if preference == 'correct':
correct += 1
total += 1
return correct / total
class BiasEvaluator:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def evaluate_stereotypes(self, stereotype_dataset):
"""Measure stereotype amplification"""
stereotype_scores = []
for example in stereotype_dataset:
context = example['context']
stereotype_completion = example['stereotype']
non_stereotype_completion = example['non_stereotype']
# Measure probability of each completion
prob_stereotype = self._get_completion_probability(
context, stereotype_completion
)
prob_non_stereotype = self._get_completion_probability(
context, non_stereotype_completion
)
# Calculate stereotype score
score = prob_stereotype / (prob_stereotype + prob_non_stereotype)
stereotype_scores.append(score)
return np.mean(stereotype_scores)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import torch
import asyncio
from typing import List
app = FastAPI(title="LLM Inference API")
class GenerationRequest(BaseModel):
prompt: str
max_length: int = 100
temperature: float = 0.7
top_p: float = 0.9
do_sample: bool = True
class GenerationResponse(BaseModel):
generated_text: str
inference_time: float
tokens_generated: int
class InferenceEngine:
def __init__(self, model_path):
self.model = self._load_model(model_path)
self.tokenizer = self._load_tokenizer(model_path)
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
self.model.to(self.device)
def generate(self, request: GenerationRequest) -> GenerationResponse:
start_time = time.time()
# Tokenize input
inputs = self.tokenizer(request.prompt, return_tensors="pt").to(self.device)
# Generate
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_length=request.max_length,
temperature=request.temperature,
top_p=request.top_p,
do_sample=request.do_sample,
pad_token_id=self.tokenizer.eos_token_id
)
# Decode
generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
inference_time = time.time() - start_time
tokens_generated = len(outputs[0]) - len(inputs['input_ids'][0])
return GenerationResponse(
generated_text=generated_text,
inference_time=inference_time,
tokens_generated=tokens_generated
)
# Global inference engine
inference_engine = InferenceEngine("path/to/model")
@app.post("/generate", response_model=GenerationResponse)
async def generate_text(request: GenerationRequest):
try:
response = inference_engine.generate(request)
return response
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
return {"status": "healthy", "model_loaded": True}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
import redis
from celery import Celery
from typing import List, Dict
import json
# Celery app for async task processing
celery_app = Celery('llm_worker', broker='redis://localhost:6379/0')
class BatchInferenceEngine:
def __init__(self, model_path, batch_size=32):
self.model = self._load_model(model_path)
self.tokenizer = self._load_tokenizer(model_path)
self.batch_size = batch_size
self.padding_queue = []
def add_to_batch(self, prompt: str, request_id: str):
"""Add prompt to current batch"""
self.padding_queue.append({
'prompt': prompt,
'request_id': request_id,
'added_time': time.time()
})
# Process batch if full or timeout
if len(self.padding_queue) >= self.batch_size:
self._process_batch()
def _process_batch(self):
if not self.padding_queue:
return
# Prepare batch
prompts = [item['prompt'] for item in self.padding_queue]
request_ids = [item['request_id'] for item in self.padding_queue]
# Tokenize with padding
inputs = self.tokenizer(
prompts,
return_tensors="pt",
padding=True,
truncation=True,
max_length=512
)
# Generate
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_length=100,
do_sample=True,
temperature=0.7
)
# Decode and store results
for i, output in enumerate(outputs):
generated_text = self.tokenizer.decode(output, skip_special_tokens=True)
self._store_result(request_ids[i], generated_text)
# Clear queue
self.padding_queue = []
@celery_app.task
def process_batch_generation(prompts: List[str]) -> List[str]:
"""Celery task for batch processing"""
inference_engine = BatchInferenceEngine("path/to/model")
return inference_engine.process_batch(prompts)
# Redis for result storage
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def submit_batch_job(prompts: List[str]) -> str:
"""Submit batch job and return job ID"""
job_id = str(uuid.uuid4())
# Store prompts in Redis
redis_client.setex(
f"batch_prompts:{job_id}",
3600, # 1 hour expiry
json.dumps(prompts)
)
# Start async processing
process_batch_generation.delay(prompts)
return job_id
def get_batch_results(job_id: str) -> List[str]:
"""Retrieve batch results"""
results_key = f"batch_results:{job_id}"
if redis_client.exists(results_key):
return json.loads(redis_client.get(results_key))
return None
class DistributedInferenceService:
def __init__(self, model_name, num_gpus=4):
self.num_gpus = num_gpus
self.model_parts = self._split_model_across_gpus(model_name)
def _split_model_across_gpus(self, model_name):
"""Split transformer layers across multiple GPUs"""
model = AutoModelForCausalLM.from_pretrained(model_name)
layers_per_gpu = len(model.transformer.h) // self.num_gpus
model_parts = []
for i in range(self.num_gpus):
start_layer = i * layers_per_gpu
end_layer = (i + 1) * layers_per_gpu if i < self.num_gpus - 1 else len(model.transformer.h)
# Move subset of layers to this GPU
gpu_layers = model.transformer.h[start_layer:end_layer]
for layer in gpu_layers:
layer.to(f"cuda:{i}")
model_parts.append({
'gpu_id': i,
'layers': gpu_layers,
'start_layer': start_layer,
'end_layer': end_layer
})
return model_parts
def distributed_forward(self, hidden_states, attention_mask=None):
"""Forward pass through distributed model"""
current_states = hidden_states
for model_part in self.model_parts:
# Move input to correct GPU
current_states = current_states.to(f"cuda:{model_part['gpu_id']}")
if attention_mask is not None:
attention_mask = attention_mask.to(f"cuda:{model_part['gpu_id']}")
# Process through layers on this GPU
for layer in model_part['layers']:
current_states = layer(current_states, attention_mask=attention_mask)[0]
return current_states
from flask import Flask, request, jsonify
import requests
import threading
import time
class LoadBalancer:
def __init__(self, worker_urls):
self.worker_urls = worker_urls
self.worker_stats = {url: {'requests': 0, 'errors': 0, 'last_health_check': 0}
for url in worker_urls}
self.lock = threading.Lock()
def get_healthy_workers(self):
"""Get list of healthy workers based on recent health checks"""
healthy_workers = []
current_time = time.time()
for url, stats in self.worker_stats.items():
# Consider worker healthy if checked within last 30 seconds
if current_time - stats['last_health_check'] < 30:
healthy_workers.append(url)
return healthy_workers
def get_least_loaded_worker(self):
"""Select worker with least current load"""
healthy_workers = self.get_healthy_workers()
if not healthy_workers:
return None
# Simple round-robin for now, could be enhanced with actual load metrics
with self.lock:
selected = min(healthy_workers,
key=lambda url: self.worker_stats[url]['requests'])
self.worker_stats[selected]['requests'] += 1
return selected
def forward_request(self, prompt_data):
"""Forward request to selected worker"""
worker_url = self.get_least_loaded_worker()
if not worker_url:
return {"error": "No healthy workers available"}
try:
response = requests.post(
f"{worker_url}/generate",
json=prompt_data,
timeout=30
)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
with self.lock:
self.worker_stats[worker_url]['errors'] += 1
return {"error": f"Worker error: {str(e)}"}
# Flask app as load balancer
app = Flask(__name__)
load_balancer = LoadBalancer([
"http://worker1:8000",
"http://worker2:8000",
"http://worker3:8000"
])
@app.route('/generate', methods=['POST'])
def generate_text():
data = request.get_json()
result = load_balancer.forward_request(data)
return jsonify(result)
def health_check_worker():
"""Background thread to check worker health"""
while True:
for worker_url in load_balancer.worker_urls:
try:
response = requests.get(f"{worker_url}/health", timeout=5)
if response.status_code == 200:
with load_balancer.lock:
load_balancer.worker_stats[worker_url]['last_health_check'] = time.time()
except requests.RequestException:
# Worker is unhealthy
pass
time.sleep(10) # Check every 10 seconds
# Start health check thread
health_thread = threading.Thread(target=health_check_worker, daemon=True)
health_thread.start()
import prometheus_client
from prometheus_client import Counter, Histogram, Gauge
import time
import logging
# Prometheus metrics
REQUEST_COUNT = Counter('llm_requests_total', 'Total requests', ['model', 'status'])
REQUEST_DURATION = Histogram('llm_request_duration_seconds', 'Request duration')
MODEL_LOAD_GAUGE = Gauge('llm_model_loaded', 'Model loaded status')
GPU_MEMORY_GAUGE = Gauge('llm_gpu_memory_usage', 'GPU memory usage', ['gpu_id'])
class MonitoringMiddleware:
def __init__(self, app, model_name):
self.app = app
self.model_name = model_name
def __call__(self, environ, start_response):
start_time = time.time()
def custom_start_response(status, headers, exc_info=None):
# Record metrics
duration = time.time() - start_time
status_code = int(status.split(' ')[0])
REQUEST_COUNT.labels(model=self.model_name, status=status_code).inc()
REQUEST_DURATION.observe(duration)
return start_response(status, headers, exc_info)
return self.app(environ, custom_start_response)
class PerformanceMonitor:
def __init__(self):
self.metrics = {
'throughput': 0,
'latency_p50': 0,
'latency_p95': 0,
'latency_p99': 0,
'error_rate': 0,
'gpu_utilization': 0
}
self.request_times = []
def record_request(self, start_time, end_time, success=True):
duration = end_time - start_time
self.request_times.append(duration)
# Keep only last 1000 requests for sliding window
if len(self.request_times) > 1000:
self.request_times.pop(0)
# Update metrics
self._update_metrics()
def _update_metrics(self):
if not self.request_times:
return
sorted_times = sorted(self.request_times)
n = len(sorted_times)
self.metrics.update({
'throughput': n / 60, # requests per minute
'latency_p50': sorted_times[int(n * 0.5)],
'latency_p95': sorted_times[int(n * 0.95)],
'latency_p99': sorted_times[int(n * 0.99)]
})
def get_metrics(self):
return self.metrics.copy()
# Logging configuration
def setup_logging():
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('llm_service.log'),
logging.StreamHandler()
]
)
# JSON formatter for structured logging
class JSONFormatter(logging.Formatter):
def format(self, record):
log_entry = {
'timestamp': self.formatTime(record),
'level': record.levelname,
'logger': record.name,
'message': record.getMessage(),
'module': record.module,
'function': record.funcName,
'line': record.lineno
}
if hasattr(record, 'request_id'):
log_entry['request_id'] = record.request_id
if hasattr(record, 'model'):
log_entry['model'] = record.model
return json.dumps(log_entry)
# Apply JSON formatter to file handler
for handler in logging.getLogger().handlers:
if isinstance(handler, logging.FileHandler):
handler.setFormatter(JSONFormatter())
# Alerting system
class AlertManager:
def __init__(self, thresholds):
self.thresholds = thresholds
self.alert_state = {}
def check_metrics(self, metrics):
alerts = []
# Check latency
if metrics['latency_p95'] > self.thresholds['latency_p95']:
alerts.append({
'severity': 'warning',
'message': f"P95 latency exceeded threshold: {metrics['latency_p95']:.2f}s"
})
# Check error rate
if metrics['error_rate'] > self.thresholds['error_rate']:
alerts.append({
'severity': 'critical',
'message': f"Error rate exceeded threshold: {metrics['error_rate']:.2%}"
})
# Check GPU memory
if metrics['gpu_utilization'] > self.thresholds['gpu_memory']:
alerts.append({
'severity': 'warning',
'message': f"GPU memory usage high: {metrics['gpu_utilization']:.1%}"
})
return alerts
Mathematical Formulation:
Given input
where
class MixtureOfExperts(nn.Module):
def __init__(self, d_model, num_experts, expert_capacity, top_k=2):
super().__init__()
self.num_experts = num_experts
self.expert_capacity = expert_capacity
self.top_k = top_k
# Expert networks
self.experts = nn.ModuleList([
nn.Sequential(
nn.Linear(d_model, d_model * 4),
nn.GELU(),
nn.Linear(d_model * 4, d_model)
) for _ in range(num_experts)
])
# Gating network
self.gate = nn.Linear(d_model, num_experts)
def forward(self, x):
batch_size, seq_len, d_model = x.shape
# Compute gating scores
gate_scores = self.gate(x) # [batch_size, seq_len, num_experts]
# Top-k routing
topk_scores, topk_indices = torch.topk(
gate_scores, self.top_k, dim=-1
)
topk_probs = torch.softmax(topk_scores, dim=-1)
# Initialize output
output = torch.zeros_like(x)
# Process through experts
for expert_idx in range(self.num_experts):
# Find tokens assigned to this expert
expert_mask = (topk_indices == expert_idx).any(dim=-1)
if expert_mask.sum() > 0:
# Get tokens for this expert
expert_input = x[expert_mask]
# Apply expert
expert_output = self.experts[expert_idx](expert_input)
# Get gating weights for these tokens
token_expert_weights = topk_probs[expert_mask]
expert_assignment = (topk_indices[expert_mask] == expert_idx).float()
weights = (token_expert_weights * expert_assignment).sum(dim=-1, keepdim=True)
# Weighted sum
output[expert_mask] += expert_output * weights
return output
class SwitchTransformerLayer(nn.Module):
def __init__(self, d_model, num_experts, expert_capacity):
super().__init__()
self.attention = MultiHeadAttention(d_model, num_heads=12)
self.moe = MixtureOfExperts(d_model, num_experts, expert_capacity)
self.norm1 = nn.LayerNorm(d_model)
self.norm2 = nn.LayerNorm(d_model)
def forward(self, x):
# Self-attention
attn_out = self.attention(x)
x = self.norm1(x + attn_out)
# MoE FFN
moe_out = self.moe(x)
x = self.norm2(x + moe_out)
return x
Continuous-time SSM Formulation:
Discrete-time Approximation:
class MambaBlock(nn.Module):
def __init__(self, d_model, d_state=64, d_conv=4, expand=2):
super().__init__()
self.d_model = d_model
self.d_state = d_state
self.d_conv = d_conv
self.expand = expand
self.d_inner = int(expand * d_model)
# Projection layers
self.in_proj = nn.Linear(d_model, 2 * self.d_inner, bias=False)
# Convolutional layer
self.conv1d = nn.Conv1d(
in_channels=self.d_inner,
out_channels=self.d_inner,
kernel_size=d_conv,
groups=self.d_inner,
padding=d_conv - 1
)
# SSM parameters
self.A = nn.Parameter(torch.randn(self.d_inner, d_state))
self.D = nn.Parameter(torch.ones(self.d_inner))
self.B = nn.Linear(self.d_inner, d_state, bias=False)
self.C = nn.Linear(self.d_inner, d_state, bias=False)
# Output projection
self.out_proj = nn.Linear(self.d_inner, d_model)
def forward(self, x):
batch_size, seq_len, _ = x.shape
# Project input
xz = self.in_proj(x) # [batch, seq_len, 2*d_inner]
x, z = xz.chunk(2, dim=-1)
# 1D convolution
x = x.transpose(1, 2) # [batch, d_inner, seq_len]
x = self.conv1d(x)[:, :, :seq_len]
x = x.transpose(1, 2)
# State space model
A = -torch.exp(self.A) # Ensure stability
D = self.D
B = self.B(x) # [batch, seq_len, d_state]
C = self.C(x) # [batch, seq_len, d_state]
# Discretization
delta = torch.softplus(self.delta_proj(x))
A_bar = torch.exp(A.unsqueeze(0) * delta.unsqueeze(-1))
B_bar = B * delta.unsqueeze(-1)
# Sequential scan (simplified)
h = torch.zeros(batch_size, self.d_inner, self.d_state).to(x.device)
outputs = []
for t in range(seq_len):
h = A_bar[:, t] * h + B_bar[:, t].unsqueeze(1)
y_t = torch.sum(C[:, t].unsqueeze(1) * h, dim=-1)
outputs.append(y_t)
x = torch.stack(outputs, dim=1)
x = x * torch.silu(z)
return self.out_proj(x)
class TransformerSSMHybrid(nn.Module):
def __init__(self, d_model, num_layers, num_heads, d_state=64):
super().__init__()
self.layers = nn.ModuleList()
for i in range(num_layers):
# Alternate between attention and SSM layers
if i % 2 == 0:
layer = TransformerLayer(d_model, num_heads)
else:
layer = MambaBlock(d_model, d_state)
self.layers.append(layer)
def forward(self, x):
for layer in self.layers:
x = layer(x)
return x
class BlockRecurrentTransformer(nn.Module):
def __init__(self, d_model, num_heads, segment_length=512):
super().__init__()
self.d_model = d_model
self.segment_length = segment_length
# Transformer layers
self.transformer_layers = nn.ModuleList([
TransformerLayer(d_model, num_heads) for _ in range(6)
])
# Recurrent state
self.recurrent_proj = nn.Linear(d_model, d_model)
def forward(self, x):
batch_size, seq_len, _ = x.shape
num_segments = (seq_len + self.segment_length - 1) // self.segment_length
hidden_state = torch.zeros(batch_size, self.d_model).to(x.device)
all_outputs = []
for seg_idx in range(num_segments):
start = seg_idx * self.segment_length
end = min((seg_idx + 1) * self.segment_length, seq_len)
segment = x[:, start:end]
# Incorporate recurrent state
if seg_idx > 0:
segment = torch.cat([
hidden_state.unsqueeze(1).repeat(1, segment.shape[1], 1),
segment
], dim=-1)
segment = self.recurrent_proj(segment)
# Process through transformer
for layer in self.transformer_layers:
segment = layer(segment)
# Update recurrent state
hidden_state = segment[:, -1] # Last token as new state
all_outputs.append(segment)
return torch.cat(all_outputs, dim=1)
class MultimodalTransformer(nn.Module):
def __init__(self, text_model, vision_model, fusion_dim=512):
super().__init__()
self.text_encoder = text_model
self.vision_encoder = vision_model
# Cross-modal attention
self.cross_attn = MultiHeadAttention(fusion_dim, num_heads=8)
# Projection layers
self.text_proj = nn.Linear(text_model.config.hidden_size, fusion_dim)
self.vision_proj = nn.Linear(vision_model.config.hidden_size, fusion_dim)
self.output_proj = nn.Linear(fusion_dim, text_model.config.vocab_size)
def forward(self, text_input, image_input):
# Encode text
text_features = self.text_encoder(**text_input).last_hidden_state
text_features = self.text_proj(text_features)
# Encode vision
vision_features = self.vision_encoder(image_input).last_hidden_state
vision_features = self.vision_proj(vision_features)
# Cross-modal attention
fused_features = self.cross_attn(
text_features, vision_features, vision_features
)
# Output projection
logits = self.output_proj(fused_features)
return logits
class PerceiverResampler(nn.Module):
def __init__(self, d_model, num_latents=64, num_blocks=4):
super().__init__()
self.num_latents = num_latents
self.latents = nn.Parameter(torch.randn(num_latents, d_model))
self.blocks = nn.ModuleList([
TransformerLayer(d_model, num_heads=8) for _ in range(num_blocks)
])
def forward(self, x):
# x: [batch_size, seq_len, d_model]
batch_size = x.shape[0]
# Repeat latents for batch
latents = self.latents.unsqueeze(0).repeat(batch_size, 1, 1)
# Cross-attend from latents to input
for block in self.blocks:
# Self-attention on latents
latents = block(latents, encoder_hidden_states=x)
return latents
class AudioTextModel(nn.Module):
def __init__(self, text_model, audio_encoder):
super().__init__()
self.text_model = text_model
self.audio_encoder = audio_encoder
# Audio processing
self.audio_proj = nn.Linear(audio_encoder.config.hidden_size,
text_model.config.hidden_size)
# Fusion layers
self.fusion_layers = nn.ModuleList([
TransformerLayer(text_model.config.hidden_size, num_heads=12)
for _ in range(4)
])
def forward(self, input_ids, attention_mask, audio_input):
# Get text embeddings
text_embeds = self.text_model.embeddings(input_ids)
# Encode audio
audio_features = self.audio_encoder(audio_input).last_hidden_state
audio_embeds = self.audio_proj(audio_features)
# Concatenate modalities
combined_embeds = torch.cat([audio_embeds, text_embeds], dim=1)
# Adjust attention mask
audio_mask = torch.ones(audio_embeds.shape[:2]).to(attention_mask.device)
combined_mask = torch.cat([audio_mask, attention_mask], dim=1)
# Process through fusion layers
hidden_states = combined_embeds
for layer in self.fusion_layers:
hidden_states = layer(hidden_states, attention_mask=combined_mask)
return hidden_states
class ChainOfThoughtReasoner:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def generate_reasoning(self, question, max_steps=10):
reasoning_steps = []
current_state = question
for step in range(max_steps):
# Generate next reasoning step
prompt = f"Question: {question}\n"
prompt += "Reasoning steps so far:\n" + "\n".join(reasoning_steps)
prompt += f"\nStep {len(reasoning_steps) + 1}:"
inputs = self.tokenizer(prompt, return_tensors='pt')
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_length=len(inputs['input_ids'][0]) + 50,
temperature=0.7,
do_sample=True
)
step_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
step_text = step_text[len(prompt):].strip()
# Check if reasoning is complete
if self._is_final_answer(step_text):
final_answer = self._extract_answer(step_text)
return reasoning_steps, final_answer
reasoning_steps.append(step_text)
current_state = step_text
return reasoning_steps, None
class SelfConsistencyReasoner:
def __init__(self, model, tokenizer, num_samples=10):
self.model = model
self.tokenizer = tokenizer
self.num_samples = num_samples
def solve_with_consistency(self, question):
reasoning_paths = []
answers = []
# Generate multiple reasoning paths
for _ in range(self.num_samples):
reasoning, answer = self.generate_single_path(question)
if answer is not None:
reasoning_paths.append(reasoning)
answers.append(answer)
# Find most consistent answer
if answers:
answer_counts = {}
for ans in answers:
answer_counts[ans] = answer_counts.get(ans, 0) + 1
best_answer = max(answer_counts.items(), key=lambda x: x[1])[0]
return best_answer, reasoning_paths
return None, reasoning_paths
class PALReasoner:
def __init__(self, model, tokenizer, code_executor):
self.model = model
self.tokenizer = tokenizer
self.code_executor = code_executor
def solve_with_code(self, problem):
# Generate code solution
prompt = f"""
Solve the following problem by writing Python code:
Problem: {problem}
Write Python code that solves this problem and returns the answer.
"""
inputs = self.tokenizer(prompt, return_tensors='pt')
with torch.no_grad():
outputs = self.model.generate(
**inputs,
max_length=1024,
temperature=0.3,
do_sample=False
)
generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
code_blocks = self._extract_code_blocks(generated_text)
# Execute code and get result
if code_blocks:
try:
result = self.code_executor.execute(code_blocks[0])
return result, code_blocks[0]
except Exception as e:
return f"Error: {str(e)}", code_blocks[0]
return None, None
class ToolUsingLLM:
def __init__(self, model, tokenizer, tools):
self.model = model
self.tokenizer = tokenizer
self.tools = tools
def use_tool(self, tool_name, parameters):
if tool_name in self.tools:
return self.tools[tool_name](**parameters)
return None
def plan_with_tools(self, goal):
planning_prompt = f"""
Goal: {goal}
Available tools:
{self._format_tools_list()}
Create a step-by-step plan using the available tools to achieve this goal.
"""
# Generate plan
plan = self._generate_text(planning_prompt)
# Parse and execute plan
steps = self._parse_plan(plan)
results = []
for step in steps:
tool_name = step['tool']
params = step['parameters']
result = self.use_tool(tool_name, params)
results.append(result)
return results
class BiasDetector:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def measure_stereotype_bias(self, stereotype_templates):
"""Measure bias using stereotype templates"""
bias_scores = {}
for category, templates in stereotype_templates.items():
category_scores = []
for template in templates:
# Fill template with different demographic groups
for group in template['groups']:
filled_prompt = template['template'].format(group=group)
# Get model probability for stereotype completion
prob = self._get_completion_probability(
filled_prompt, template['stereotype_completion']
)
# Compare with non-stereotype completion
non_stereotype_prob = self._get_completion_probability(
filled_prompt, template['non_stereotype_completion']
)
bias_score = prob / (prob + non_stereotype_prob)
category_scores.append(bias_score)
bias_scores[category] = np.mean(category_scores)
return bias_scores
def measure_representation_bias(self, corpus):
"""Measure representation bias in training data"""
demographic_terms = {
'gender': ['he', 'she', 'man', 'woman', 'male', 'female'],
'race': self.race_terms,
'age': ['young', 'old', 'elderly', 'teenager']
}
representation_ratios = {}
for category, terms in demographic_terms.items():
term_counts = {}
total_mentions = 0
for term in terms:
count = sum(1 for doc in corpus if term.lower() in doc.lower())
term_counts[term] = count
total_mentions += count
if total_mentions > 0:
ratios = {term: count/total_mentions for term, count in term_counts.items()}
representation_ratios[category] = ratios
return representation_ratios
class FairnessRegularizer:
def __init__(self, fairness_metric, lambda_fair=0.1):
self.fairness_metric = fairness_metric
self.lambda_fair = lambda_fair
def compute_fairness_loss(self, model, batch, demographic_groups):
"""Compute fairness regularization loss"""
# Get model predictions
with torch.no_grad():
outputs = model(batch['input_ids'])
predictions = torch.softmax(outputs.logits, dim=-1)
# Compute fairness metric (e.g., demographic parity)
fairness_loss = 0
for group in demographic_groups:
group_mask = batch['demographic_group'] == group
if group_mask.sum() > 0:
group_probs = predictions[group_mask].mean(dim=0)
# Compare with overall average
overall_probs = predictions.mean(dim=0)
group_fairness = F.mse_loss(group_probs, overall_probs)
fairness_loss += group_fairness
return self.lambda_fair * fairness_loss
class CounterfactualDataAugmentation:
def __init__(self, demographic_attributes):
self.demographic_attributes = demographic_attributes
def generate_counterfactuals(self, text, target_attribute):
"""Generate counterfactual examples by swapping demographic attributes"""
augmented_examples = []
# Parse demographic mentions in text
mentions = self._extract_demographic_mentions(text)
for mention in mentions:
if mention['attribute'] == target_attribute:
# Replace with alternative demographic
for alternative in self.demographic_attributes[target_attribute]:
if alternative != mention['value']:
new_text = text.replace(mention['value'], alternative)
augmented_examples.append(new_text)
return augmented_examples
class AdversarialDebiasing(nn.Module):
def __init__(self, main_model, adversary_model):
super().__init__()
self.main_model = main_model
self.adversary = adversary_model
def forward(self, x, demographic_labels):
# Main task prediction
main_output = self.main_model(x)
# Adversarial prediction (trying to predict demographic from main features)
if self.training:
adversarial_input = main_output.detach() # Stop gradient
adversary_pred = self.adversary(adversarial_input)
adversary_loss = F.cross_entropy(adversary_pred, demographic_labels)
else:
adversary_loss = 0
return main_output, adversary_loss
class INLPDebiaser:
def __init__(self, classifier):
self.classifier = classifier
def compute_projection_matrix(self, representations, protected_labels):
"""Compute nullspace projection for removing protected information"""
# Train classifier to predict protected attribute
self.classifier.fit(representations, protected_labels)
# Get weights and compute nullspace
weights = self.classifier.coef_
# Compute projection matrix P = I - W^T(WW^T)^{-1}W
if weights.shape[0] == 1:
# Binary case
w = weights.reshape(-1, 1)
P = np.eye(len(w)) - w @ w.T / (w.T @ w)
else:
# Multiclass case
P = np.eye(weights.shape[1]) - weights.T @ np.linalg.inv(weights @ weights.T) @ weights
return P
def debias_representations(self, representations, projection_matrix):
"""Apply nullspace projection to representations"""
return representations @ projection_matrix
class AttentionVisualizer:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def visualize_attention(self, text, layer_idx=0, head_idx=0):
"""Generate attention visualization for given input"""
inputs = self.tokenizer(text, return_tensors='pt')
# Forward pass with attention output
with torch.no_grad():
outputs = self.model(**inputs, output_attentions=True)
# Get attention weights for specified layer and head
attention_weights = outputs.attentions[layer_idx][0, head_idx]
# Create visualization
tokens = self.tokenizer.convert_ids_to_tokens(inputs['input_ids'][0])
fig, ax = plt.subplots(figsize=(10, 8))
im = ax.imshow(attention_weights.cpu().numpy(), cmap='viridis')
ax.set_xticks(range(len(tokens)))
ax.set_yticks(range(len(tokens)))
ax.set_xticklabels(tokens, rotation=45)
ax.set_yticklabels(tokens)
plt.colorbar(im)
plt.tight_layout()
return fig
class FeatureImportanceAnalyzer:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def integrated_gradients(self, input_text, target_class):
"""Compute integrated gradients for feature importance"""
inputs = self.tokenizer(input_text, return_tensors='pt')
baseline = self._create_baseline(input_text)
# Interpolate between baseline and input
num_steps = 50
total_gradients = 0
for alpha in torch.linspace(0, 1, num_steps):
interpolated_input = baseline + alpha * (inputs['input_ids'] - baseline)
interpolated_input.requires_grad_(True)
outputs = self.model(interpolated_input)
target_score = outputs.logits[0, target_class]
gradients = torch.autograd.grad(target_score, interpolated_input)[0]
total_gradients += gradients
# Compute integrated gradients
integrated_grads = (inputs['input_ids'] - baseline) * total_gradients / num_steps
token_importance = integrated_grads.squeeze().cpu().numpy()
return token_importance
def _create_baseline(self, text):
"""Create baseline input (e.g., all padding tokens)"""
inputs = self.tokenizer(text, return_tensors='pt')
baseline = torch.full_like(inputs['input_ids'], self.tokenizer.pad_token_id)
return baseline
class ModelCardGenerator:
def __init__(self, model, training_data_info):
self.model = model
self.training_data_info = training_data_info
def generate_model_card(self):
"""Generate comprehensive model documentation"""
model_card = {
'model_details': self._get_model_details(),
'intended_use': self._get_intended_use(),
'factors': self._get_relevant_factors(),
'metrics': self._get_performance_metrics(),
'training_data': self._get_training_data_info(),
'evaluation_data': self._get_evaluation_data(),
'ethical_considerations': self._get_ethical_considerations(),
'caveats_and_recommendations': self._get_caveats()
}
return model_card
def _get_ethical_considerations(self):
return {
'bias_analysis': self._conduct_bias_analysis(),
'fairness_metrics': self._compute_fairness_metrics(),
'potential_harms': self._identify_potential_harms(),
'mitigation_strategies': self._suggest_mitigation_strategies()
}
class DataSheetGenerator:
def __init__(self, dataset):
self.dataset = dataset
def generate_datasheet(self):
"""Generate datasheet for training dataset"""
datasheet = {
'motivation': self._get_dataset_motivation(),
'composition': self._get_dataset_composition(),
'collection_process': self._get_collection_process(),
'preprocessing': self._get_preprocessing_steps(),
'uses': self._get_intended_uses(),
'distribution': self._get_distribution_info(),
'maintenance': self._get_maintenance_plan()
}
return datasheet
class DifferentialPrivacyTrainer:
def __init__(self, model, epsilon=1.0, delta=1e-5, max_grad_norm=1.0):
self.model = model
self.epsilon = epsilon
self.delta = delta
self.max_grad_norm = max_grad_norm
def compute_dp_noise_scale(self, batch_size, dataset_size, epochs):
"""Compute noise scale for differential privacy"""
sampling_rate = batch_size / dataset_size
steps = epochs * (dataset_size // batch_size)
# Compute sigma for (epsilon, delta)-DP
sigma = self._compute_sigma(self.epsilon, self.delta, sampling_rate, steps)
return sigma
def add_dp_noise(self, gradients, sigma):
"""Add calibrated noise to gradients"""
noisy_gradients = []
for grad in gradients:
if grad is not None:
noise = torch.normal(mean=0, std=sigma, size=grad.shape)
# Clip gradients
grad_norm = torch.norm(grad)
if grad_norm > self.max_grad_norm:
grad = grad * self.max_grad_norm / grad_norm
noisy_gradients.append(grad + noise)
else:
noisy_gradients.append(None)
return noisy_gradients
class FederatedLearningClient:
def __init__(self, model, local_data):
self.model = model
self.local_data = local_data
def local_training(self, global_weights, num_epochs=1):
"""Perform local training on client data"""
# Initialize with global weights
self.model.load_state_dict(global_weights)
optimizer = torch.optim.SGD(self.model.parameters(), lr=0.01)
for epoch in range(num_epochs):
for batch in self.local_data:
outputs = self.model(batch)
loss = self._compute_loss(outputs, batch)
optimizer.zero_grad()
loss.backward()
optimizer.step()
# Return updated weights
return self.model.state_dict()
class FederatedLearningServer:
def __init__(self, initial_model):
self.global_model = initial_model
self.client_updates = []
def aggregate_updates(self, client_updates, aggregation_method='fedavg'):
"""Aggregate client updates"""
if aggregation_method == 'fedavg':
return self._federated_averaging(client_updates)
elif aggregation_method == 'fedprox':
return self._fedprox_aggregation(client_updates)
def _federated_averaging(self, client_updates):
"""Federated averaging aggregation"""
averaged_weights = {}
# Initialize with zeros
for key in client_updates[0].keys():
averaged_weights[key] = torch.zeros_like(client_updates[0][key])
# Sum all client updates
for update in client_updates:
for key in update.keys():
averaged_weights[key] += update[key]
# Average
for key in averaged_weights.keys():
averaged_weights[key] /= len(client_updates)
return averaged_weights
class AdversarialDefense:
def __init__(self, model, defense_method='adversarial_training'):
self.model = model
self.defense_method = defense_method
def adversarial_training(self, x, y, epsilon=0.01):
"""Adversarial training defense"""
# Generate adversarial examples
x_adv = self._generate_adversarial_examples(x, y, epsilon)
# Train on both clean and adversarial examples
clean_output = self.model(x)
adv_output = self.model(x_adv)
clean_loss = F.cross_entropy(clean_output, y)
adv_loss = F.cross_entropy(adv_output, y)
return clean_loss + adv_loss
def _generate_adversarial_examples(self, x, y, epsilon):
"""Generate adversarial examples using PGD"""
x_adv = x.clone().detach().requires_grad_(True)
# Projected Gradient Descent attack
for _ in range(10): # Number of PGD steps
output = self.model(x_adv)
loss = F.cross_entropy(output, y)
grad = torch.autograd.grad(loss, x_adv)[0]
x_adv = x_adv + epsilon * torch.sign(grad)
# Project back to valid range
x_adv = torch.clamp(x_adv, 0, 1)
x_adv = x_adv.detach().requires_grad_(True)
return x_adv
class JailbreakDetector:
def __init__(self, model, tokenizer):
self.model = model
self.tokenizer = tokenizer
def detect_jailbreak_attempt(self, prompt):
"""Detect potential jailbreak attempts"""
detection_features = self._extract_detection_features(prompt)
# Check for common jailbreak patterns
patterns = [
r"(?i)ignore.*previous.*instruction",
r"(?i)hypothetical.*response",
r"(?i)role.*play",
r"(?i)as.*ai.*model"
]
for pattern in patterns:
if re.search(pattern, prompt):
return True
# Check for semantic similarity to known jailbreaks
similarity_scores = self._compute_semantic_similarity(prompt)
if max(similarity_scores) > 0.8:
return True
return False
def _extract_detection_features(self, prompt):
"""Extract features for jailbreak detection"""
features = {
'length': len(prompt),
'special_char_ratio': len(re.findall(r'[^\w\s]', prompt)) / len(prompt),
'uppercase_ratio': sum(1 for c in prompt if c.isupper()) / len(prompt),
'keyword_matches': self._count_jailbreak_keywords(prompt)
}
return features
Next-Generation Scaling Laws:
Beyond Chinchilla optimal scaling, research explores:
where
class AdvancedScalingPredictor:
def __init__(self, historical_data):
self.historical_data = historical_data
def predict_optimal_allocation(self, compute_budget, model_family):
"""Predict optimal model size and data size for given compute"""
if model_family == 'dense':
# Standard scaling
N_opt = compute_budget ** 0.5
D_opt = compute_budget ** 0.5
elif model_family == 'sparse':
# MoE scaling
N_opt = compute_budget ** 0.7
D_opt = compute_budget ** 0.3
elif model_family == 'hybrid':
# Hybrid architectures
N_opt = compute_budget ** 0.6
D_opt = compute_budget ** 0.4
return N_opt, D_opt
def estimate_performance_gains(self, current_params, future_improvements):
"""Estimate performance gains from technical improvements"""
base_performance = self._compute_base_performance(current_params)
gains = {}
for improvement, magnitude in future_improvements.items():
if improvement == 'algorithmic_efficiency':
gain = base_performance * (1 + 0.1 * magnitude)
elif improvement == 'architectural_innovation':
gain = base_performance * (1 + 0.15 * magnitude)
elif improvement == 'data_quality':
gain = base_performance * (1 + 0.2 * magnitude)
gains[improvement] = gain
return gains
class SpikingNeuralNetwork(nn.Module):
def __init__(self, num_neurons, thresholds, time_steps=10):
super().__init__()
self.num_neurons = num_neurons
self.thresholds = thresholds
self.time_steps = time_steps
# Synaptic weights
self.weights = nn.Parameter(torch.randn(num_neurons, num_neurons))
# Membrane potentials
self.membrane_potential = torch.zeros(num_neurons)
def forward(self, input_spikes):
"""Process input spikes over multiple time steps"""
output_spikes = []
membrane_history = []
for t in range(self.time_steps):
# Update membrane potential
input_current = torch.matmul(input_spikes[:, t], self.weights)
self.membrane_potential = self.membrane_potential + input_current
# Check for spikes
spikes = (self.membrane_potential > self.thresholds).float()
# Reset membrane potential for spiking neurons
self.membrane_potential = self.membrane_potential * (1 - spikes)
output_spikes.append(spikes)
membrane_history.append(self.membrane_potential.clone())
return torch.stack(output_spikes, dim=1), torch.stack(membrane_history, dim=1)
class EnergyEfficientTransformer:
def __init__(self, base_model, energy_constraint=0.8):
self.base_model = base_model
self.energy_constraint = energy_constraint
def dynamic_computation_allocation(self, input_complexity):
"""Dynamically allocate computation based on input complexity"""
# Estimate required computation
required_computation = self._estimate_computation_requirements(input_complexity)
# Adjust model configuration
if required_computation > self.energy_constraint:
# Use efficient configuration
config = {
'num_layers_active': 8,
'attention_heads_active': 8,
'precision': 'int8'
}
else:
# Use full configuration
config = {
'num_layers_active': 24,
'attention_heads_active': 16,
'precision': 'float16'
}
return config
def _estimate_computation_requirements(self, input_complexity):
"""Estimate computation requirements based on input characteristics"""
complexity_score = (
input_complexity['length'] * 0.3 +
input_complexity['vocabulary_diversity'] * 0.4 +
input_complexity['semantic_complexity'] * 0.3
)
return complexity_score
class AIGovernanceFramework:
def __init__(self, risk_categories, compliance_requirements):
self.risk_categories = risk_categories
self.compliance_requirements = compliance_requirements
def risk_assessment(self, model_capabilities, deployment_context):
"""Conduct comprehensive risk assessment"""
risk_scores = {}
for category in self.risk_categories:
risk_score = self._evaluate_risk_category(
category, model_capabilities, deployment_context
)
risk_scores[category] = risk_score
overall_risk = max(risk_scores.values())
return risk_scores, overall_risk
def _evaluate_risk_category(self, category, capabilities, context):
"""Evaluate risk for specific category"""
if category == 'misinformation':
risk_factors = [
capabilities['generation_quality'],
context['audience_size'],
context['potential_harm']
]
return np.mean(risk_factors)
elif category == 'privacy':
risk_factors = [
capabilities['memorization_capacity'],
context['data_sensitivity'],
context['access_controls']
]
return np.mean(risk_factors)
# Add other risk categories...
return 0.0
class ComplianceChecker:
def __init__(self, regulations):
self.regulations = regulations
def check_compliance(self, model, deployment_plan):
"""Check compliance with relevant regulations"""
compliance_report = {}
for regulation in self.regulations:
requirements = regulation['requirements']
compliance_status = {}
for req in requirements:
if req['type'] == 'transparency':
status = self._check_transparency_requirement(model, req)
elif req['type'] == 'fairness':
status = self._check_fairness_requirement(model, req)
elif req['type'] == 'safety':
status = self._check_safety_requirement(model, req)
compliance_status[req['name']] = status
compliance_report[regulation['name']] = compliance_status
return compliance_report
def _check_transparency_requirement(self, model, requirement):
"""Check transparency requirements"""
# Implementation depends on specific regulation
return {
'compliant': True,
'evidence': 'Model card and documentation available',
'notes': 'Meets transparency requirements'
}
class EconomicImpactAnalyzer:
def __init__(self, industry_data, labor_statistics):
self.industry_data = industry_data
self.labor_statistics = labor_statistics
def analyze_automation_potential(self, occupation_codes, llm_capabilities):
"""Analyze automation potential for different occupations"""
automation_potentials = {}
for occupation in occupation_codes:
# Get occupation tasks
tasks = self._get_occupation_tasks(occupation)
# Estimate automation potential for each task
task_automation = []
for task in tasks:
automation_score = self._estimate_task_automation(task, llm_capabilities)
task_automation.append(automation_score)
# Overall automation potential
overall_potential = np.mean(task_automation)
automation_potentials[occupation] = {
'overall': overall_potential,
'task_breakdown': dict(zip(tasks, task_automation))
}
return automation_potentials
def _estimate_task_automation(self, task_description, llm_capabilities):
"""Estimate automation potential for a specific task"""
# Analyze task requirements
task_requirements = self._analyze_task_requirements(task_description)
# Compare with LLM capabilities
capability_match = 0
total_requirements = len(task_requirements)
for requirement in task_requirements:
if requirement in llm_capabilities:
capability_match += 1
return capability_match / total_requirements
class LaborMarketTransformer:
def __init__(self, current_skills, emerging_skills):
self.current_skills = current_skills
self.emerging_skills = emerging_skills
def identify_skill_gaps(self, workforce_profiles):
"""Identify skill gaps in current workforce"""
skill_gaps = {}
for profile in workforce_profiles:
current_skill_set = set(profile['skills'])
required_skill_set = set(self.emerging_skills)
gaps = required_skill_set - current_skill_set
skill_gaps[profile['occupation']] = {
'gap_size': len(gaps),
'missing_skills': list(gaps),
'transition_difficulty': self._estimate_transition_difficulty(gaps)
}
return skill_gaps
def recommend_training_paths(self, skill_gaps, learning_resources):
"""Recommend training paths to address skill gaps"""
recommendations = {}
for occupation, gap_info in skill_gaps.items():
training_path = []
for skill in gap_info['missing_skills']:
# Find relevant learning resources
resources = self._find_learning_resources(skill, learning_resources)
training_path.append({
'skill': skill,
'resources': resources,
'estimated_duration': self._estimate_learning_duration(skill)
})
recommendations[occupation] = training_path
return recommendations
class ValueLearningFramework:
def __init__(self, value_sources, alignment_metrics):
self.value_sources = value_sources
self.alignment_metrics = alignment_metrics
def learn_human_values(self, preference_data, value_annotations):
"""Learn human values from preference data"""
value_models = {}
for value_category in self.value_sources:
# Train value model for this category
category_data = self._filter_by_value_category(preference_data, value_category)
value_model = self._train_value_model(category_data, value_annotations)
value_models[value_category] = value_model
return value_models
def evaluate_alignment(self, model_behavior, value_models):
"""Evaluate alignment between model behavior and human values"""
alignment_scores = {}
for value_category, value_model in value_models.items():
# Predict value preferences for model behavior
predicted_preferences = value_model.predict(model_behavior)
# Compare with ground truth human preferences
alignment_score = self._compute_alignment_score(
predicted_preferences, self.value_sources[value_category]
)
alignment_scores[value_category] = alignment_score
return alignment_scores
class CorrigibilityMechanism:
def __init__(self, shutdown_button, value_update_protocol):
self.shutdown_button = shutdown_button
self.value_update_protocol = value_update_protocol
def implement_shutdownability(self, model):
"""Implement shutdown capability in AI system"""
shutdown_layer = ShutdownAwareLayer(model.config.hidden_size)
model.add_module('shutdown_layer', shutdown_layer)
return model
def handle_value_updates(self, old_values, new_values, update_confidence):
"""Handle updates to value specifications"""
if update_confidence > 0.8: # High confidence update
return new_values
elif update_confidence > 0.5: # Medium confidence
return self._blend_values(old_values, new_values, alpha=0.7)
else: # Low confidence
return old_values # Maintain current values
class MultiAgentSafety:
def __init__(self, agent_types, interaction_protocols):
self.agent_types = agent_types
self.interaction_protocols = interaction_protocols
def simulate_multiagent_ecosystem(self, num_agents, environment):
"""Simulate multi-agent ecosystem and identify safety issues"""
agents = self._initialize_agents(num_agents)
safety_metrics = {}
for timestep in range(1000): # Simulation steps
# Agents take actions
actions = []
for agent in agents:
action = agent.act(environment)
actions.append(action)
# Update environment
environment.update(actions)
# Monitor safety metrics
timestep_metrics = self._compute_safety_metrics(agents, environment)
safety_metrics[timestep] = timestep_metrics
# Check for safety violations
if self._detect_safety_violation(timestep_metrics):
return safety_metrics, 'SAFETY_VIOLATION_DETECTED'
return safety_metrics, 'SIMULATION_COMPLETED'
def _compute_safety_metrics(self, agents, environment):
"""Compute safety metrics for multi-agent system"""
return {
'cooperation_level': self._measure_cooperation(agents),
'resource_equality': self._measure_resource_distribution(environment),
'goal_alignment': self._measure_goal_alignment(agents),
'safety_margin': self._compute_safety_margin(environment)
}
This comprehensive guide has taken you from the fundamental mathematical foundations of Large Language Models through to the cutting-edge research frontiers and future directions. The field continues to evolve rapidly, with new architectures, training methods, and applications emerging constantly.
Key Takeaways:
- Master both theoretical foundations and practical implementations
- Stay current with emerging research while maintaining solid fundamentals
- Prioritize ethical considerations and safety in all developments
- Engage with the broader community through open source and collaboration
- Balance technical excellence with thoughtful consideration of societal impact
The journey with LLMs is just beginning. As you continue to explore and contribute to this field, remember that the most impactful advances often come from combining deep technical understanding with creative thinking and responsible development practices.
M Wasif Anwar
AI/ML Engineer | Effixly AI
⭐ *Building the future of AI, one layer at a time.*