AI Research Scientist Interview Questions for Research Roles

import numpy as np

# Simple neural network to demonstrate backpropagation
class SimpleNN:
    def __init__(self, input_size, hidden_size, output_size):
        # Initialize weights
        self.W1 = np.random.randn(input_size, hidden_size) * 0.01
        self.b1 = np.zeros((1, hidden_size))
        self.W2 = np.random.randn(hidden_size, output_size) * 0.01
        self.b2 = np.zeros((1, output_size))
    
    def sigmoid(self, x):
        return 1 / (1 + np.exp(-x))
    
    def sigmoid_derivative(self, x):
        return x * (1 - x)
    
    def forward(self, X):
        # Layer 1
        self.z1 = np.dot(X, self.W1) + self.b1
        self.a1 = self.sigmoid(self.z1)
        
        # Layer 2
        self.z2 = np.dot(self.a1, self.W2) + self.b2
        self.a2 = self.sigmoid(self.z2)
        
        return self.a2
    
    def backward(self, X, y, output, learning_rate=0.01):
        m = X.shape[0]
        
        # Output layer gradients
        # dL/da2 = a2 - y (for binary cross-entropy)
        # dL/dz2 = dL/da2 * da2/dz2 = (a2 - y) * sigmoid'(z2)
        dz2 = output - y
        dW2 = (1/m) * np.dot(self.a1.T, dz2)
        db2 = (1/m) * np.sum(dz2, axis=0, keepdims=True)
        
        # Hidden layer gradients (chain rule)
        # dL/da1 = dL/dz2 * dz2/da1 = dz2 * W2.T
        # dL/dz1 = dL/da1 * da1/dz1 = dL/da1 * sigmoid'(z1)
        da1 = np.dot(dz2, self.W2.T)
        dz1 = da1 * self.sigmoid_derivative(self.a1)
        dW1 = (1/m) * np.dot(X.T, dz1)
        db1 = (1/m) * np.sum(dz1, axis=0, keepdims=True)
        
        # Update weights
        self.W2 -= learning_rate * dW2
        self.b2 -= learning_rate * db2
        self.W1 -= learning_rate * dW1
        self.b1 -= learning_rate * db1
    
    def train(self, X, y, epochs=1000):
        for epoch in range(epochs):
            # Forward pass
            output = self.forward(X)
            
            # Backward pass
            self.backward(X, y, output)
            
            if epoch % 100 == 0:
                loss = -np.mean(y * np.log(output) + (1-y) * np.log(1-output))
                print(f'Epoch {epoch}, Loss: {loss:.4f}')

# Example usage
X = np.array([[0, 0], [0, 1], [1, 0], [1, 1]])
y = np.array([[0], [1], [1], [0]])  # XOR

nn = SimpleNN(input_size=2, hidden_size=4, output_size=1)
nn.train(X, y, epochs=5000)

import torch
import torch.nn as nn

# Problem: Deep network with sigmoid
class VanishingGradientNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.layers = nn.Sequential(*[
            nn.Sequential(nn.Linear(100, 100), nn.Sigmoid())
            for _ in range(20)  # 20 layers
        ])
    
    def forward(self, x):
        return self.layers(x)

# Solution 1: ReLU activation
class ReLUNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.layers = nn.Sequential(*[
            nn.Sequential(nn.Linear(100, 100), nn.ReLU())
            for _ in range(20)
        ])
    
    def forward(self, x):
        return self.layers(x)

# Solution 2: Residual connections
class ResidualBlock(nn.Module):
    def __init__(self, dim):
        super().__init__()
        self.layers = nn.Sequential(
            nn.Linear(dim, dim),
            nn.ReLU(),
            nn.Linear(dim, dim)
        )
    
    def forward(self, x):
        return x + self.layers(x)  # Skip connection

class ResNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.blocks = nn.Sequential(*[
            ResidualBlock(100) for _ in range(20)
        ])
    
    def forward(self, x):
        return self.blocks(x)

# Solution 3: Batch Normalization
class BatchNormNet(nn.Module):
    def __init__(self):
        super().__init__()
        self.layers = nn.Sequential(*[
            nn.Sequential(
                nn.Linear(100, 100),
                nn.BatchNorm1d(100),
                nn.ReLU()
            )
            for _ in range(20)
        ])
    
    def forward(self, x):
        return self.layers(x)

# Gradient flow analysis
def analyze_gradients(model, x, y):
    model.zero_grad()
    output = model(x)
    loss = nn.MSELoss()(output, y)
    loss.backward()
    
    # Check gradient magnitudes
    for name, param in model.named_parameters():
        if param.grad is not None:
            grad_norm = param.grad.norm().item()
            print(f"{name}: {grad_norm:.6f}")

import torch
import torch.nn as nn
import torch.nn.functional as F
import math

class ScaledDotProductAttention(nn.Module):
    def __init__(self, temperature):
        super().__init__()
        self.temperature = temperature
    
    def forward(self, q, k, v, mask=None):
        """
        q: (batch, seq_len, d_k)
        k: (batch, seq_len, d_k)
        v: (batch, seq_len, d_v)
        """
        # Compute attention scores
        attn = torch.matmul(q, k.transpose(-2, -1)) / self.temperature
        
        # Apply mask (for padding or causal attention)
        if mask is not None:
            attn = attn.masked_fill(mask == 0, -1e9)
        
        # Softmax to get attention weights
        attn_weights = F.softmax(attn, dim=-1)
        
        # Apply attention to values
        output = torch.matmul(attn_weights, v)
        
        return output, attn_weights

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, n_heads, dropout=0.1):
        super().__init__()
        assert d_model % n_heads == 0
        
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads
        
        # Linear projections
        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)
        
        self.attention = ScaledDotProductAttention(temperature=math.sqrt(self.d_k))
        self.dropout = nn.Dropout(dropout)
    
    def forward(self, q, k, v, mask=None):
        batch_size = q.size(0)
        
        # Linear projections and split into heads
        q = self.w_q(q).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        k = self.w_k(k).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        v = self.w_v(v).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        
        # Apply attention
        output, attn_weights = self.attention(q, k, v, mask)
        
        # Concatenate heads
        output = output.transpose(1, 2).contiguous().view(batch_size, -1, self.d_model)
        
        # Final linear projection
        output = self.w_o(output)
        
        return output, attn_weights

# Example usage
d_model = 512
n_heads = 8
seq_len = 10
batch_size = 2

mha = MultiHeadAttention(d_model, n_heads)
x = torch.randn(batch_size, seq_len, d_model)

# Self-attention (q, k, v all from x)
output, attn = mha(x, x, x)
print(f"Output shape: {output.shape}")
print(f"Attention weights shape: {attn.shape}")

import torch
import torch.nn as nn

# Batch Normalization
class BatchNormExample(nn.Module):
    def __init__(self, num_features):
        super().__init__()
        self.bn = nn.BatchNorm1d(num_features)
    
    def forward(self, x):
        # x: (batch_size, num_features)
        # Normalizes across batch dimension for each feature
        return self.bn(x)

# Layer Normalization
class LayerNormExample(nn.Module):
    def __init__(self, normalized_shape):
        super().__init__()
        self.ln = nn.LayerNorm(normalized_shape)
    
    def forward(self, x):
        # x: (batch_size, seq_len, d_model)
        # Normalizes across feature dimension for each sample
        return self.ln(x)

# Manual implementation
class ManualLayerNorm(nn.Module):
    def __init__(self, normalized_shape, eps=1e-5):
        super().__init__()
        self.eps = eps
        self.gamma = nn.Parameter(torch.ones(normalized_shape))
        self.beta = nn.Parameter(torch.zeros(normalized_shape))
    
    def forward(self, x):
        # Calculate mean and variance across last dimension
        mean = x.mean(dim=-1, keepdim=True)
        var = x.var(dim=-1, keepdim=True, unbiased=False)
        
        # Normalize
        x_norm = (x - mean) / torch.sqrt(var + self.eps)
        
        # Scale and shift
        return self.gamma * x_norm + self.beta

# Comparison
batch_size, seq_len, d_model = 2, 10, 512

# Batch Norm (for CNN)
x_cnn = torch.randn(batch_size, d_model, 28, 28)
bn = nn.BatchNorm2d(d_model)
out_bn = bn(x_cnn)

# Layer Norm (for Transformer)
x_transformer = torch.randn(batch_size, seq_len, d_model)
ln = nn.LayerNorm(d_model)
out_ln = ln(x_transformer)

print(f"Batch Norm output: {out_bn.shape}")
print(f"Layer Norm output: {out_ln.shape}")

import torch
import torch.nn as nn
import math

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        
        # Create positional encoding matrix
        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)
        self.register_buffer('pe', pe)
    
    def forward(self, x):
        return x + self.pe[:, :x.size(1)]

class TransformerEncoderLayer(nn.Module):
    def __init__(self, d_model, n_heads, d_ff, dropout=0.1):
        super().__init__()
        
        # Multi-head attention
        self.self_attn = nn.MultiheadAttention(d_model, n_heads, dropout=dropout)
        
        # Feed-forward network
        self.ffn = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_ff, d_model)
        )
        
        # Layer normalization
        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
        attn_output, _ = self.self_attn(x, x, x, attn_mask=mask)
        x = self.norm1(x + self.dropout(attn_output))
        
        # Feed-forward with residual connection
        ffn_output = self.ffn(x)
        x = self.norm2(x + self.dropout(ffn_output))
        
        return x

class TransformerEncoder(nn.Module):
    def __init__(self, vocab_size, d_model, n_heads, d_ff, n_layers, dropout=0.1):
        super().__init__()
        
        self.embedding = nn.Embedding(vocab_size, d_model)
        self.pos_encoding = PositionalEncoding(d_model)
        
        self.layers = nn.ModuleList([
            TransformerEncoderLayer(d_model, n_heads, d_ff, dropout)
            for _ in range(n_layers)
        ])
        
        self.dropout = nn.Dropout(dropout)
    
    def forward(self, x, mask=None):
        # Embedding + positional encoding
        x = self.embedding(x) * math.sqrt(self.embedding.embedding_dim)
        x = self.pos_encoding(x)
        x = self.dropout(x)
        
        # Apply encoder layers
        for layer in self.layers:
            x = layer(x, mask)
        
        return x

# Example usage
vocab_size = 10000
d_model = 512
n_heads = 8
d_ff = 2048
n_layers = 6

encoder = TransformerEncoder(vocab_size, d_model, n_heads, d_ff, n_layers)

# Input: (batch_size, seq_len)
x = torch.randint(0, vocab_size, (2, 10))
output = encoder(x)
print(f"Output shape: {output.shape}")  # (2, 10, 512)

# Ablation study example
class ModelWithAblations:
    def __init__(self, use_attention=True, use_residual=True, use_dropout=True):
        self.use_attention = use_attention
        self.use_residual = use_residual
        self.use_dropout = use_dropout
    
    def build_model(self):
        layers = []
        
        if self.use_attention:
            layers.append(AttentionLayer())
        
        layers.append(FFNLayer())
        
        if self.use_dropout:
            layers.append(nn.Dropout(0.1))
        
        if self.use_residual:
            return ResidualWrapper(nn.Sequential(*layers))
        else:
            return nn.Sequential(*layers)

# Run ablation experiments
configs = [
    {'use_attention': True, 'use_residual': True, 'use_dropout': True},   # Full model
    {'use_attention': False, 'use_residual': True, 'use_dropout': True},  # No attention
    {'use_attention': True, 'use_residual': False, 'use_dropout': True},  # No residual
    {'use_attention': True, 'use_residual': True, 'use_dropout': False},  # No dropout
]

results = []
for config in configs:
    model = ModelWithAblations(**config)
    accuracy = train_and_evaluate(model, seed=42)
    results.append({**config, 'accuracy': accuracy})

# Analyze results
import pandas as pd
df = pd.DataFrame(results)
print(df)

import random
import numpy as np
import torch
import os

def set_all_seeds(seed=42):
    """Set seeds for reproducibility"""
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    os.environ['PYTHONHASHSEED'] = str(seed)
    
    # Deterministic operations
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False

# Log everything
import logging
import json

def log_experiment(config, results):
    experiment_log = {
        'timestamp': datetime.now().isoformat(),
        'config': config,
        'results': results,
        'environment': {
            'python_version': sys.version,
            'torch_version': torch.__version__,
            'cuda_version': torch.version.cuda,
            'gpu': torch.cuda.get_device_name(0) if torch.cuda.is_available() else 'CPU'
        }
    }
    
    with open('experiment_log.json', 'w') as f:
        json.dump(experiment_log, f, indent=2)

# Share code and models
"""
# README.md
## Reproducibility

### Environment
```bash
conda create -n research python=3.9
conda activate research
pip install -r requirements.txt

python train.py --config configs/experiment1.yaml --seed 42

python evaluate.py --checkpoint checkpoints/best_model.pt

import torch
import torch.nn as nn
import torch.nn.functional as F

class ContrastiveLoss(nn.Module):
    def __init__(self, temperature=0.5):
        super().__init__()
        self.temperature = temperature
    
    def forward(self, features):
        """
        features: (2*batch_size, dim) - pairs of augmented samples
        """
        batch_size = features.shape[0] // 2
        
        # Normalize features
        features = F.normalize(features, dim=1)
        
        # Compute similarity matrix
        similarity_matrix = torch.matmul(features, features.T)
        
        # Create labels (positive pairs)
        labels = torch.cat([torch.arange(batch_size) + batch_size,
                           torch.arange(batch_size)]).to(features.device)
        
        # Mask to remove self-similarity
        mask = torch.eye(2 * batch_size, dtype=torch.bool).to(features.device)
        similarity_matrix = similarity_matrix.masked_fill(mask, -9e15)
        
        # Compute loss
        similarity_matrix = similarity_matrix / self.temperature
        loss = F.cross_entropy(similarity_matrix, labels)
        
        return loss

class SimCLR(nn.Module):
    def __init__(self, encoder, projection_dim=128):
        super().__init__()
        self.encoder = encoder
        self.projection = nn.Sequential(
            nn.Linear(encoder.output_dim, 512),
            nn.ReLU(),
            nn.Linear(512, projection_dim)
        )
    
    def forward(self, x1, x2):
        # Encode both augmented views
        h1 = self.encoder(x1)
        h2 = self.encoder(x2)
        
        # Project to contrastive space
        z1 = self.projection(h1)
        z2 = self.projection(h2)
        
        # Concatenate for contrastive loss
        features = torch.cat([z1, z2], dim=0)
        
        return features

# Training loop
model = SimCLR(encoder, projection_dim=128)
criterion = ContrastiveLoss(temperature=0.5)
optimizer = torch.optim.Adam(model.parameters())

for epoch in range(100):
    for batch in dataloader:
        # Get two augmented views
        x1, x2 = augment(batch)
        
        # Forward pass
        features = model(x1, x2)
        loss = criterion(features)
        
        # Backward pass
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

import torch
import torch.nn as nn

class PatchEmbedding(nn.Module):
    def __init__(self, img_size=224, patch_size=16, in_channels=3, embed_dim=768):
        super().__init__()
        self.img_size = img_size
        self.patch_size = patch_size
        self.n_patches = (img_size // patch_size) ** 2
        
        # Convolution to extract patches and embed
        self.projection = nn.Conv2d(
            in_channels, embed_dim,
            kernel_size=patch_size,
            stride=patch_size
        )
    
    def forward(self, x):
        # x: (batch, channels, height, width)
        x = self.projection(x)  # (batch, embed_dim, n_patches_h, n_patches_w)
        x = x.flatten(2)  # (batch, embed_dim, n_patches)
        x = x.transpose(1, 2)  # (batch, n_patches, embed_dim)
        return x

class VisionTransformer(nn.Module):
    def __init__(self, img_size=224, patch_size=16, in_channels=3,
                 embed_dim=768, n_heads=12, n_layers=12, num_classes=1000):
        super().__init__()
        
        # Patch embedding
        self.patch_embed = PatchEmbedding(img_size, patch_size, in_channels, embed_dim)
        n_patches = self.patch_embed.n_patches
        
        # Class token
        self.cls_token = nn.Parameter(torch.zeros(1, 1, embed_dim))
        
        # Positional embedding
        self.pos_embed = nn.Parameter(torch.zeros(1, n_patches + 1, embed_dim))
        
        # Transformer encoder
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=embed_dim,
            nhead=n_heads,
            dim_feedforward=4*embed_dim,
            dropout=0.1
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, n_layers)
        
        # Classification head
        self.head = nn.Linear(embed_dim, num_classes)
    
    def forward(self, x):
        batch_size = x.shape[0]
        
        # Patch embedding
        x = self.patch_embed(x)  # (batch, n_patches, embed_dim)
        
        # Add class token
        cls_tokens = self.cls_token.expand(batch_size, -1, -1)
        x = torch.cat([cls_tokens, x], dim=1)  # (batch, n_patches+1, embed_dim)
        
        # Add positional embedding
        x = x + self.pos_embed
        
        # Transformer
        x = self.transformer(x)
        
        # Classification (use class token)
        cls_output = x[:, 0]
        logits = self.head(cls_output)
        
        return logits

# Example usage
model = VisionTransformer(
    img_size=224,
    patch_size=16,
    embed_dim=768,
    n_heads=12,
    n_layers=12,
    num_classes=1000
)

x = torch.randn(2, 3, 224, 224)
output = model(x)
print(f"Output shape: {output.shape}")  # (2, 1000)

import torch
import torch.nn as nn
import torch.nn.functional as F

class DiffusionModel:
    def __init__(self, model, timesteps=1000, beta_start=0.0001, beta_end=0.02):
        self.model = model
        self.timesteps = timesteps
        
        # Variance schedule
        self.betas = torch.linspace(beta_start, beta_end, timesteps)
        self.alphas = 1 - self.betas
        self.alphas_cumprod = torch.cumprod(self.alphas, dim=0)
    
    def forward_diffusion(self, x0, t):
        """Add noise to x0 at timestep t"""
        noise = torch.randn_like(x0)
        
        sqrt_alphas_cumprod_t = self.alphas_cumprod[t].sqrt()
        sqrt_one_minus_alphas_cumprod_t = (1 - self.alphas_cumprod[t]).sqrt()
        
        # x_t = sqrt(alpha_cumprod_t) * x_0 + sqrt(1 - alpha_cumprod_t) * noise
        xt = sqrt_alphas_cumprod_t * x0 + sqrt_one_minus_alphas_cumprod_t * noise
        
        return xt, noise
    
    def train_step(self, x0):
        """Training step: predict noise"""
        batch_size = x0.shape[0]
        
        # Sample random timesteps
        t = torch.randint(0, self.timesteps, (batch_size,))
        
        # Add noise
        xt, noise = self.forward_diffusion(x0, t)
        
        # Predict noise
        predicted_noise = self.model(xt, t)
        
        # Loss: MSE between predicted and actual noise
        loss = F.mse_loss(predicted_noise, noise)
        
        return loss
    
    @torch.no_grad()
    def sample(self, shape):
        """Generate samples by reversing diffusion"""
        # Start from pure noise
        x = torch.randn(shape)
        
        # Iteratively denoise
        for t in reversed(range(self.timesteps)):
            # Predict noise
            t_batch = torch.full((shape[0],), t, dtype=torch.long)
            predicted_noise = self.model(x, t_batch)
            
            # Compute denoising step
            alpha_t = self.alphas[t]
            alpha_cumprod_t = self.alphas_cumprod[t]
            beta_t = self.betas[t]
            
            # Denoise
            x = (1 / alpha_t.sqrt()) * (
                x - (beta_t / (1 - alpha_cumprod_t).sqrt()) * predicted_noise
            )
            
            # Add noise (except last step)
            if t > 0:
                noise = torch.randn_like(x)
                x = x + beta_t.sqrt() * noise
        
        return x

# Simple U-Net for noise prediction
class SimpleUNet(nn.Module):
    def __init__(self, in_channels=3, out_channels=3):
        super().__init__()
        # Simplified U-Net architecture
        self.encoder = nn.Sequential(
            nn.Conv2d(in_channels, 64, 3, padding=1),
            nn.ReLU(),
            nn.Conv2d(64, 128, 3, padding=1),
            nn.ReLU()
        )
        
        self.decoder = nn.Sequential(
            nn.Conv2d(128, 64, 3, padding=1),
            nn.ReLU(),
            nn.Conv2d(64, out_channels, 3, padding=1)
        )
    
    def forward(self, x, t):
        # Encode timestep (simplified)
        x = self.encoder(x)
        x = self.decoder(x)
        return x

# Training
model = SimpleUNet()
diffusion = DiffusionModel(model)
optimizer = torch.optim.Adam(model.parameters())

for epoch in range(100):
    for batch in dataloader:
        loss = diffusion.train_step(batch)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

# Sampling
samples = diffusion.sample((16, 3, 32, 32))

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from collections import deque
import random

# Q-Network
class DQN(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()
        self.network = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim)
        )
    
    def forward(self, state):
        return self.network(state)

# Experience Replay Buffer
class ReplayBuffer:
    def __init__(self, capacity=10000):
        self.buffer = deque(maxlen=capacity)
    
    def push(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))
    
    def sample(self, batch_size):
        batch = random.sample(self.buffer, batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)
        return (
            torch.FloatTensor(states),
            torch.LongTensor(actions),
            torch.FloatTensor(rewards),
            torch.FloatTensor(next_states),
            torch.FloatTensor(dones)
        )
    
    def __len__(self):
        return len(self.buffer)

# DQN Agent
class DQNAgent:
    def __init__(self, state_dim, action_dim, lr=1e-3, gamma=0.99, epsilon=1.0):
        self.action_dim = action_dim
        self.gamma = gamma
        self.epsilon = epsilon
        self.epsilon_min = 0.01
        self.epsilon_decay = 0.995
        
        # Q-networks
        self.q_network = DQN(state_dim, action_dim)
        self.target_network = DQN(state_dim, action_dim)
        self.target_network.load_state_dict(self.q_network.state_dict())
        
        self.optimizer = optim.Adam(self.q_network.parameters(), lr=lr)
        self.replay_buffer = ReplayBuffer()
    
    def select_action(self, state):
        """Epsilon-greedy action selection"""
        if random.random() < self.epsilon:
            return random.randrange(self.action_dim)
        else:
            with torch.no_grad():
                state = torch.FloatTensor(state).unsqueeze(0)
                q_values = self.q_network(state)
                return q_values.argmax().item()
    
    def train(self, batch_size=64):
        """Train on batch from replay buffer"""
        if len(self.replay_buffer) < batch_size:
            return
        
        states, actions, rewards, next_states, dones = self.replay_buffer.sample(batch_size)
        
        # Current Q values
        current_q = self.q_network(states).gather(1, actions.unsqueeze(1))
        
        # Target Q values
        with torch.no_grad():
            next_q = self.target_network(next_states).max(1)[0]
            target_q = rewards + (1 - dones) * self.gamma * next_q
        
        # Loss
        loss = nn.MSELoss()(current_q.squeeze(), target_q)
        
        # Optimize
        self.optimizer.zero_grad()
        loss.backward()
        self.optimizer.step()
        
        # Decay epsilon
        self.epsilon = max(self.epsilon_min, self.epsilon * self.epsilon_decay)
        
        return loss.item()
    
    def update_target_network(self):
        """Copy weights from Q-network to target network"""
        self.target_network.load_state_dict(self.q_network.state_dict())

# Training loop
def train_dqn(env, agent, episodes=1000, max_steps=500):
    scores = []
    
    for episode in range(episodes):
        state = env.reset()
        total_reward = 0
        
        for step in range(max_steps):
            # Select and perform action
            action = agent.select_action(state)
            next_state, reward, done, _ = env.step(action)
            
            # Store transition
            agent.replay_buffer.push(state, action, reward, next_state, done)
            
            # Train
            loss = agent.train()
            
            total_reward += reward
            state = next_state
            
            if done:
                break
        
        # Update target network periodically
        if episode % 10 == 0:
            agent.update_target_network()
        
        scores.append(total_reward)
        
        if episode % 100 == 0:
            print(f"Episode {episode}, Score: {total_reward:.2f}, Epsilon: {agent.epsilon:.3f}")
    
    return scores

# Example usage
# env = gym.make('CartPole-v1')
# agent = DQNAgent(state_dim=4, action_dim=2)
# scores = train_dqn(env, agent)

# Reduces overestimation bias
with torch.no_grad():
    # Use Q-network to select action
    next_actions = self.q_network(next_states).argmax(1)
    # Use target network to evaluate
    next_q = self.target_network(next_states).gather(1, next_actions.unsqueeze(1))
    target_q = rewards + (1 - dones) * self.gamma * next_q.squeeze()

class DuelingDQN(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dim=128):
        super().__init__()
        
        self.feature = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU()
        )
        
        # Value stream
        self.value = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, 1)
        )
        
        # Advantage stream
        self.advantage = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, action_dim)
        )
    
    def forward(self, state):
        features = self.feature(state)
        value = self.value(features)
        advantage = self.advantage(features)
        
        # Q(s,a) = V(s) + (A(s,a) - mean(A(s,a)))
        q_values = value + (advantage - advantage.mean(dim=1, keepdim=True))
        return q_values

class PrioritizedReplayBuffer:
    def __init__(self, capacity, alpha=0.6):
        self.capacity = capacity
        self.alpha = alpha
        self.buffer = []
        self.priorities = np.zeros(capacity)
        self.position = 0
    
    def push(self, transition):
        max_priority = self.priorities.max() if self.buffer else 1.0
        
        if len(self.buffer) < self.capacity:
            self.buffer.append(transition)
        else:
            self.buffer[self.position] = transition
        
        self.priorities[self.position] = max_priority
        self.position = (self.position + 1) % self.capacity
    
    def sample(self, batch_size, beta=0.4):
        priorities = self.priorities[:len(self.buffer)]
        probabilities = priorities ** self.alpha
        probabilities /= probabilities.sum()
        
        indices = np.random.choice(len(self.buffer), batch_size, p=probabilities)
        samples = [self.buffer[idx] for idx in indices]
        
        # Importance sampling weights
        weights = (len(self.buffer) * probabilities[indices]) ** (-beta)
        weights /= weights.max()
        
        return samples, indices, torch.FloatTensor(weights)
    
    def update_priorities(self, indices, priorities):
        for idx, priority in zip(indices, priorities):
            self.priorities[idx] = priority

import torch
import torch.nn as nn
import torch.nn.functional as F

class GCNLayer(nn.Module):
    def __init__(self, in_features, out_features):
        super().__init__()
        self.linear = nn.Linear(in_features, out_features)
    
    def forward(self, X, A):
        """
        X: Node features (N x in_features)
        A: Adjacency matrix (N x N)
        """
        # Add self-loops
        A_hat = A + torch.eye(A.size(0)).to(A.device)
        
        # Degree matrix
        D = torch.diag(A_hat.sum(1))
        
        # Symmetric normalization: D^(-1/2) A D^(-1/2)
        D_inv_sqrt = torch.diag(1.0 / torch.sqrt(D.diag()))
        A_norm = D_inv_sqrt @ A_hat @ D_inv_sqrt
        
        # Propagate
        return F.relu(self.linear(A_norm @ X))

class GCN(nn.Module):
    def __init__(self, in_features, hidden_features, out_features, num_layers=2):
        super().__init__()
        
        self.layers = nn.ModuleList()
        self.layers.append(GCNLayer(in_features, hidden_features))
        
        for _ in range(num_layers - 2):
            self.layers.append(GCNLayer(hidden_features, hidden_features))
        
        self.layers.append(GCNLayer(hidden_features, out_features))
    
    def forward(self, X, A):
        for layer in self.layers[:-1]:
            X = layer(X, A)
        
        # Last layer without activation
        X = self.layers[-1].linear(X)
        return X

# Example: Node classification
num_nodes = 100
in_features = 16
hidden_features = 32
out_features = 7  # Number of classes

# Random graph
X = torch.randn(num_nodes, in_features)
A = torch.randint(0, 2, (num_nodes, num_nodes)).float()
A = (A + A.T) / 2  # Make symmetric

model = GCN(in_features, hidden_features, out_features)
output = model(X, A)
print(f"Output shape: {output.shape}")  # (100, 7)

class GATLayer(nn.Module):
    def __init__(self, in_features, out_features, num_heads=8, dropout=0.6):
        super().__init__()
        self.num_heads = num_heads
        self.out_features = out_features
        
        # Linear transformations
        self.W = nn.Linear(in_features, num_heads * out_features)
        self.a = nn.Parameter(torch.zeros(size=(2 * out_features, num_heads)))
        
        self.leakyrelu = nn.LeakyReLU(0.2)
        self.dropout = nn.Dropout(dropout)
    
    def forward(self, X, A):
        """
        X: (N, in_features)
        A: (N, N) adjacency matrix
        """
        N = X.size(0)
        
        # Linear transformation
        h = self.W(X).view(N, self.num_heads, self.out_features)
        
        # Attention mechanism
        # Concatenate for all pairs
        h_i = h.repeat(1, N, 1).view(N * N, self.num_heads, self.out_features)
        h_j = h.repeat(N, 1, 1)
        
        # Attention coefficients
        concat = torch.cat([h_i, h_j], dim=2)
        e = self.leakyrelu(torch.matmul(concat, self.a).squeeze(2))
        e = e.view(N, N, self.num_heads)
        
        # Mask attention for non-neighbors
        zero_vec = -9e15 * torch.ones_like(e)
        attention = torch.where(A.unsqueeze(2) > 0, e, zero_vec)
        
        # Softmax
        attention = F.softmax(attention, dim=1)
        attention = self.dropout(attention)
        
        # Aggregate
        h_prime = torch.matmul(attention.transpose(1, 2), h)
        
        return h_prime.mean(dim=1)  # Average over heads

class GAT(nn.Module):
    def __init__(self, in_features, hidden_features, out_features, num_heads=8):
        super().__init__()
        
        self.gat1 = GATLayer(in_features, hidden_features, num_heads)
        self.gat2 = GATLayer(hidden_features * num_heads, out_features, 1)
    
    def forward(self, X, A):
        X = F.elu(self.gat1(X, A))
        X = self.gat2(X, A)
        return F.log_softmax(X, dim=1)

class SAGELayer(nn.Module):
    def __init__(self, in_features, out_features, aggregator='mean'):
        super().__init__()
        self.aggregator = aggregator
        
        if aggregator == 'mean':
            self.linear = nn.Linear(in_features, out_features)
        elif aggregator == 'pool':
            self.mlp = nn.Sequential(
                nn.Linear(in_features, in_features),
                nn.ReLU()
            )
            self.linear = nn.Linear(in_features, out_features)
    
    def forward(self, X, A, num_samples=10):
        """
        Sample neighbors and aggregate
        """
        N = X.size(0)
        aggregated = []
        
        for i in range(N):
            # Get neighbors
            neighbors = A[i].nonzero().squeeze()
            
            # Sample neighbors
            if len(neighbors) > num_samples:
                sampled = neighbors[torch.randperm(len(neighbors))[:num_samples]]
            else:
                sampled = neighbors
            
            # Aggregate neighbor features
            if self.aggregator == 'mean':
                neighbor_features = X[sampled].mean(dim=0)
            elif self.aggregator == 'pool':
                neighbor_features = self.mlp(X[sampled]).max(dim=0)[0]
            
            # Concatenate with self features
            combined = torch.cat([X[i], neighbor_features])
            aggregated.append(combined)
        
        aggregated = torch.stack(aggregated)
        return F.relu(self.linear(aggregated))

import torch
import torch.nn as nn
import numpy as np
import matplotlib.pyplot as plt

class IntegratedGradients:
    def __init__(self, model):
        self.model = model
    
    def compute_gradients(self, inputs, target_class):
        """Compute gradients of output w.r.t. input"""
        inputs.requires_grad = True
        
        outputs = self.model(inputs)
        
        # Gradient of target class score
        self.model.zero_grad()
        outputs[0, target_class].backward()
        
        return inputs.grad
    
    def integrated_gradients(self, inputs, target_class, baseline=None, steps=50):
        """
        Integrated Gradients: Integrate gradients along path from baseline to input
        """
        if baseline is None:
            baseline = torch.zeros_like(inputs)
        
        # Generate interpolated inputs
        alphas = torch.linspace(0, 1, steps)
        interpolated_inputs = []
        
        for alpha in alphas:
            interpolated = baseline + alpha * (inputs - baseline)
            interpolated_inputs.append(interpolated)
        
        interpolated_inputs = torch.cat(interpolated_inputs, dim=0)
        
        # Compute gradients
        gradients = []
        for i in range(steps):
            grad = self.compute_gradients(
                interpolated_inputs[i:i+1],
                target_class
            )
            gradients.append(grad)
        
        gradients = torch.stack(gradients)
        
        # Integrate
        avg_gradients = gradients.mean(dim=0)
        integrated_grads = (inputs - baseline) * avg_gradients
        
        return integrated_grads

# Example usage
model = nn.Sequential(
    nn.Linear(10, 20),
    nn.ReLU(),
    nn.Linear(20, 5)
)

ig = IntegratedGradients(model)
inputs = torch.randn(1, 10)
attributions = ig.integrated_gradients(inputs, target_class=0)
print(f"Feature attributions: {attributions}")

class SaliencyMap:
    def __init__(self, model):
        self.model = model
    
    def generate(self, image, target_class):
        """Generate saliency map"""
        image.requires_grad = True
        
        # Forward pass
        output = self.model(image)
        
        # Backward pass
        self.model.zero_grad()
        output[0, target_class].backward()
        
        # Saliency is absolute value of gradient
        saliency = image.grad.abs()
        
        return saliency
    
    def visualize(self, image, saliency):
        """Visualize saliency map"""
        fig, axes = plt.subplots(1, 2, figsize=(10, 5))
        
        # Original image
        axes[0].imshow(image.squeeze().permute(1, 2, 0).detach())
        axes[0].set_title('Original Image')
        axes[0].axis('off')
        
        # Saliency map
        axes[1].imshow(saliency.squeeze().max(dim=0)[0].detach(), cmap='hot')
        axes[1].set_title('Saliency Map')
        axes[1].axis('off')
        
        plt.show()

class GradCAM:
    def __init__(self, model, target_layer):
        self.model = model
        self.target_layer = target_layer
        self.gradients = None
        self.activations = None
        
        # Register hooks
        target_layer.register_forward_hook(self.save_activation)
        target_layer.register_backward_hook(self.save_gradient)
    
    def save_activation(self, module, input, output):
        self.activations = output
    
    def save_gradient(self, module, grad_input, grad_output):
        self.gradients = grad_output[0]
    
    def generate(self, image, target_class):
        """Generate GradCAM heatmap"""
        # Forward pass
        output = self.model(image)
        
        # Backward pass
        self.model.zero_grad()
        output[0, target_class].backward()
        
        # Pool gradients
        pooled_gradients = self.gradients.mean(dim=[2, 3], keepdim=True)
        
        # Weight activations
        weighted_activations = self.activations * pooled_gradients
        
        # Sum across channels
        heatmap = weighted_activations.sum(dim=1).squeeze()
        
        # ReLU and normalize
        heatmap = F.relu(heatmap)
        heatmap /= heatmap.max()
        
        return heatmap

# Example with ResNet
import torchvision.models as models

model = models.resnet50(pretrained=True)
gradcam = GradCAM(model, model.layer4[-1])

image = torch.randn(1, 3, 224, 224)
heatmap = gradcam.generate(image, target_class=281)  # Cat class

import shap

class SHAPExplainer:
    def __init__(self, model, background_data):
        """
        model: PyTorch model
        background_data: Representative dataset for background distribution
        """
        self.model = model
        self.explainer = shap.DeepExplainer(model, background_data)
    
    def explain(self, inputs):
        """Generate SHAP values"""
        shap_values = self.explainer.shap_values(inputs)
        return shap_values
    
    def visualize(self, inputs, shap_values, feature_names=None):
        """Visualize SHAP values"""
        shap.summary_plot(shap_values, inputs, feature_names=feature_names)

# Example usage
# background = torch.randn(100, 10)  # Background dataset
# explainer = SHAPExplainer(model, background)
# shap_values = explainer.explain(test_inputs)
# explainer.visualize(test_inputs, shap_values)

from lime import lime_tabular

class LIMEExplainer:
    def __init__(self, model, training_data, feature_names):
        self.model = model
        self.explainer = lime_tabular.LimeTabularExplainer(
            training_data,
            feature_names=feature_names,
            mode='classification'
        )
    
    def predict_fn(self, X):
        """Wrapper for model prediction"""
        X_tensor = torch.FloatTensor(X)
        with torch.no_grad():
            outputs = self.model(X_tensor)
            probs = F.softmax(outputs, dim=1)
        return probs.numpy()
    
    def explain_instance(self, instance, num_features=10):
        """Explain single prediction"""
        explanation = self.explainer.explain_instance(
            instance,
            self.predict_fn,
            num_features=num_features
        )
        return explanation
    
    def visualize(self, explanation):
        """Visualize explanation"""
        explanation.show_in_notebook()

# Example
# training_data = np.random.randn(1000, 10)
# feature_names = [f'feature_{i}' for i in range(10)]
# lime_explainer = LIMEExplainer(model, training_data, feature_names)
# explanation = lime_explainer.explain_instance(test_instance)

class AttentionVisualizer:
    def __init__(self, model):
        self.model = model
        self.attention_weights = []
        
        # Register hooks to capture attention
        for layer in model.transformer.layers:
            layer.self_attn.register_forward_hook(self.save_attention)
    
    def save_attention(self, module, input, output):
        # output[1] contains attention weights
        self.attention_weights.append(output[1].detach())
    
    def visualize_attention(self, tokens, layer=0, head=0):
        """Visualize attention weights"""
        attention = self.attention_weights[layer][0, head].cpu().numpy()
        
        fig, ax = plt.subplots(figsize=(10, 10))
        im = ax.imshow(attention, cmap='viridis')
        
        ax.set_xticks(range(len(tokens)))
        ax.set_yticks(range(len(tokens)))
        ax.set_xticklabels(tokens, rotation=90)
        ax.set_yticklabels(tokens)
        
        plt.colorbar(im)
        plt.title(f'Attention Weights - Layer {layer}, Head {head}')
        plt.tight_layout()
        plt.show()