Update app.py
Browse files
app.py
CHANGED
|
@@ -8,10 +8,12 @@ import torch.optim as optim
|
|
| 8 |
from PIL import Image
|
| 9 |
|
| 10 |
# CIFAR-10 labels
|
| 11 |
-
cifar10_classes = [
|
| 12 |
-
|
|
|
|
|
|
|
| 13 |
|
| 14 |
-
#
|
| 15 |
transform = transforms.Compose([
|
| 16 |
transforms.Resize((32, 32)),
|
| 17 |
transforms.ToTensor(),
|
|
@@ -19,15 +21,30 @@ transform = transforms.Compose([
|
|
| 19 |
])
|
| 20 |
|
| 21 |
# Load CIFAR-10 datasets
|
| 22 |
-
trainset = torchvision.datasets.CIFAR10(
|
| 23 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
| 24 |
testloader = torch.utils.data.DataLoader(testset, batch_size=64, shuffle=False)
|
| 25 |
|
| 26 |
def predict(model, image_tensor):
|
|
|
|
|
|
|
|
|
|
|
|
|
| 27 |
model.eval()
|
| 28 |
with torch.no_grad():
|
| 29 |
outputs = model(image_tensor.unsqueeze(0))
|
| 30 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 31 |
if torch.isnan(probs).any():
|
| 32 |
print("β οΈ Warning: NaN detected in prediction probabilities")
|
| 33 |
probs = torch.zeros_like(probs)
|
|
@@ -36,8 +53,8 @@ def predict(model, image_tensor):
|
|
| 36 |
|
| 37 |
def unlearn(model, image_tensor, label_idx, learning_rate, steps=20):
|
| 38 |
"""
|
| 39 |
-
Performs targeted unlearning by updating only the final fully connected layer
|
| 40 |
-
|
| 41 |
"""
|
| 42 |
model.train()
|
| 43 |
# Freeze all layers except the final fully connected layer (fc)
|
|
@@ -45,37 +62,41 @@ def unlearn(model, image_tensor, label_idx, learning_rate, steps=20):
|
|
| 45 |
if "fc" not in name:
|
| 46 |
param.requires_grad = False
|
| 47 |
|
| 48 |
-
# Set BatchNorm layers to
|
| 49 |
for m in model.modules():
|
| 50 |
if isinstance(m, nn.BatchNorm2d):
|
| 51 |
m.eval()
|
| 52 |
|
| 53 |
criterion = nn.CrossEntropyLoss()
|
| 54 |
-
# Use Adam optimizer for parameters that require gradients (i.e. only the fc layer)
|
| 55 |
optimizer = optim.Adam(filter(lambda p: p.requires_grad, model.parameters()), lr=learning_rate)
|
| 56 |
-
|
| 57 |
-
# Ensure label tensor is on the same device as the image_tensor
|
| 58 |
device = image_tensor.device
|
| 59 |
label_tensor = torch.tensor([label_idx], device=device)
|
| 60 |
|
| 61 |
for i in range(steps):
|
| 62 |
output = model(image_tensor.unsqueeze(0))
|
| 63 |
-
loss
|
|
|
|
|
|
|
| 64 |
if torch.isnan(loss):
|
| 65 |
print(f"β NaN detected in loss at step {i}. Stopping unlearning.")
|
| 66 |
break
|
| 67 |
-
print(f"π§ Step {i+1}/{steps} - Unlearning Loss: {loss.item():.4f}")
|
| 68 |
|
|
|
|
| 69 |
optimizer.zero_grad()
|
| 70 |
loss.backward()
|
| 71 |
-
# Clip gradients to
|
| 72 |
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
|
| 73 |
optimizer.step()
|
| 74 |
|
| 75 |
def evaluate_model(model, testloader):
|
|
|
|
|
|
|
|
|
|
| 76 |
model.eval()
|
| 77 |
total, correct, loss_total = 0, 0, 0.0
|
| 78 |
criterion = nn.CrossEntropyLoss()
|
|
|
|
| 79 |
with torch.no_grad():
|
| 80 |
for images, labels in testloader:
|
| 81 |
outputs = model(images)
|
|
@@ -84,72 +105,76 @@ def evaluate_model(model, testloader):
|
|
| 84 |
total += labels.size(0)
|
| 85 |
correct += (preds == labels).sum().item()
|
| 86 |
loss_total += loss.item() * labels.size(0)
|
| 87 |
-
|
|
|
|
|
|
|
|
|
|
| 88 |
|
| 89 |
def run_unlearning(index_to_unlearn, learning_rate):
|
| 90 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
| 91 |
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
|
| 92 |
-
|
| 93 |
-
# Load the original pre-trained model
|
| 94 |
original_model = models.resnet18(weights=None)
|
| 95 |
original_model.fc = nn.Linear(original_model.fc.in_features, 10)
|
| 96 |
original_model.load_state_dict(torch.load("resnet18.pth", map_location=device))
|
| 97 |
original_model.to(device)
|
| 98 |
original_model.eval()
|
| 99 |
-
|
| 100 |
-
# Duplicate the model for unlearning experiment
|
| 101 |
unlearned_model = models.resnet18(weights=None)
|
| 102 |
unlearned_model.fc = nn.Linear(unlearned_model.fc.in_features, 10)
|
| 103 |
unlearned_model.load_state_dict(torch.load("resnet18.pth", map_location=device))
|
| 104 |
unlearned_model.to(device)
|
| 105 |
-
|
| 106 |
# Get the sample to unlearn from the training set
|
| 107 |
image_tensor, label_idx = trainset[index_to_unlearn]
|
| 108 |
image_tensor = image_tensor.to(device)
|
| 109 |
label_name = cifar10_classes[label_idx]
|
| 110 |
print(f"ποΈ Actual Label Index: {label_idx} | Label Name: {label_name}")
|
| 111 |
-
|
| 112 |
# Prediction before unlearning
|
| 113 |
probs_before, pred_before = predict(original_model, image_tensor)
|
| 114 |
conf_before = probs_before[label_idx].item()
|
| 115 |
-
|
| 116 |
-
# Perform unlearning on the duplicated model
|
| 117 |
unlearn(unlearned_model, image_tensor, label_idx, learning_rate)
|
| 118 |
-
|
| 119 |
# Prediction after unlearning
|
| 120 |
probs_after, pred_after = predict(unlearned_model, image_tensor)
|
| 121 |
-
print(probs_after)
|
| 122 |
conf_after = probs_after[label_idx].item()
|
| 123 |
-
|
| 124 |
-
|
|
|
|
|
|
|
| 125 |
orig_acc, orig_loss = evaluate_model(original_model, testloader)
|
| 126 |
unlearn_acc, unlearn_loss = evaluate_model(unlearned_model, testloader)
|
| 127 |
-
|
| 128 |
result = f"""
|
| 129 |
π Index Unlearned: {index_to_unlearn}
|
| 130 |
ποΈ Actual Label: {label_name} (Index: {label_idx})
|
| 131 |
-
|
| 132 |
π BEFORE Unlearning:
|
| 133 |
- Prediction: {cifar10_classes[pred_before]}
|
| 134 |
- Confidence: {conf_before:.10f}
|
| 135 |
-
|
| 136 |
π§½ AFTER Unlearning:
|
| 137 |
- Prediction: {cifar10_classes[pred_after]}
|
| 138 |
- Confidence: {conf_after:.10f}
|
| 139 |
-
|
| 140 |
π Confidence Drop: {conf_before - conf_after:.6f}
|
| 141 |
-
|
| 142 |
π§ͺ Test Set Performance:
|
| 143 |
- Original Model: {orig_acc:.2f}% accuracy, Loss: {orig_loss:.4f}
|
| 144 |
- Unlearned Model: {unlearn_acc:.2f}% accuracy, Loss: {unlearn_loss:.4f}
|
| 145 |
"""
|
| 146 |
return result
|
| 147 |
|
| 148 |
-
# Gradio
|
| 149 |
demo = gr.Interface(
|
| 150 |
fn=run_unlearning,
|
| 151 |
inputs=[
|
| 152 |
-
gr.Slider(0, len(trainset)
|
| 153 |
gr.Slider(0.0001, 0.01, step=0.0001, value=0.005, label="Learning Rate (for Unlearning)")
|
| 154 |
],
|
| 155 |
outputs="text",
|
|
|
|
| 8 |
from PIL import Image
|
| 9 |
|
| 10 |
# CIFAR-10 labels
|
| 11 |
+
cifar10_classes = [
|
| 12 |
+
'airplane', 'automobile', 'bird', 'cat', 'deer',
|
| 13 |
+
'dog', 'frog', 'horse', 'ship', 'truck'
|
| 14 |
+
]
|
| 15 |
|
| 16 |
+
# Define transformations with proper normalization for 3 channels
|
| 17 |
transform = transforms.Compose([
|
| 18 |
transforms.Resize((32, 32)),
|
| 19 |
transforms.ToTensor(),
|
|
|
|
| 21 |
])
|
| 22 |
|
| 23 |
# Load CIFAR-10 datasets
|
| 24 |
+
trainset = torchvision.datasets.CIFAR10(
|
| 25 |
+
root='./data', train=True, download=True, transform=transform
|
| 26 |
+
)
|
| 27 |
+
testset = torchvision.datasets.CIFAR10(
|
| 28 |
+
root='./data', train=False, download=True, transform=transform
|
| 29 |
+
)
|
| 30 |
testloader = torch.utils.data.DataLoader(testset, batch_size=64, shuffle=False)
|
| 31 |
|
| 32 |
def predict(model, image_tensor):
|
| 33 |
+
"""
|
| 34 |
+
Performs a forward pass through the model and computes softmax probabilities
|
| 35 |
+
and predicted class using a numerically stable approach.
|
| 36 |
+
"""
|
| 37 |
model.eval()
|
| 38 |
with torch.no_grad():
|
| 39 |
outputs = model(image_tensor.unsqueeze(0))
|
| 40 |
+
logits = outputs[0]
|
| 41 |
+
# Use a numerically stable softmax: subtract max logit
|
| 42 |
+
max_logit = logits.max()
|
| 43 |
+
stable_logits = logits - max_logit
|
| 44 |
+
exp_logits = torch.exp(stable_logits)
|
| 45 |
+
probs = exp_logits / exp_logits.sum()
|
| 46 |
+
|
| 47 |
+
# Check for numerical issues (if probability is exactly 0 or NaN)
|
| 48 |
if torch.isnan(probs).any():
|
| 49 |
print("β οΈ Warning: NaN detected in prediction probabilities")
|
| 50 |
probs = torch.zeros_like(probs)
|
|
|
|
| 53 |
|
| 54 |
def unlearn(model, image_tensor, label_idx, learning_rate, steps=20):
|
| 55 |
"""
|
| 56 |
+
Performs targeted unlearning by updating only the final fully connected layer.
|
| 57 |
+
The negative cross-entropy loss drives the confidence for the target class down.
|
| 58 |
"""
|
| 59 |
model.train()
|
| 60 |
# Freeze all layers except the final fully connected layer (fc)
|
|
|
|
| 62 |
if "fc" not in name:
|
| 63 |
param.requires_grad = False
|
| 64 |
|
| 65 |
+
# Set BatchNorm layers to evaluation mode to avoid updating running stats
|
| 66 |
for m in model.modules():
|
| 67 |
if isinstance(m, nn.BatchNorm2d):
|
| 68 |
m.eval()
|
| 69 |
|
| 70 |
criterion = nn.CrossEntropyLoss()
|
|
|
|
| 71 |
optimizer = optim.Adam(filter(lambda p: p.requires_grad, model.parameters()), lr=learning_rate)
|
| 72 |
+
|
|
|
|
| 73 |
device = image_tensor.device
|
| 74 |
label_tensor = torch.tensor([label_idx], device=device)
|
| 75 |
|
| 76 |
for i in range(steps):
|
| 77 |
output = model(image_tensor.unsqueeze(0))
|
| 78 |
+
# Negative loss to reduce confidence on the target label
|
| 79 |
+
loss = -criterion(output, label_tensor)
|
| 80 |
+
|
| 81 |
if torch.isnan(loss):
|
| 82 |
print(f"β NaN detected in loss at step {i}. Stopping unlearning.")
|
| 83 |
break
|
|
|
|
| 84 |
|
| 85 |
+
print(f"π§ Step {i+1}/{steps} - Unlearning Loss: {loss.item():.4f}")
|
| 86 |
optimizer.zero_grad()
|
| 87 |
loss.backward()
|
| 88 |
+
# Clip gradients to maintain stability
|
| 89 |
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
|
| 90 |
optimizer.step()
|
| 91 |
|
| 92 |
def evaluate_model(model, testloader):
|
| 93 |
+
"""
|
| 94 |
+
Evaluates the model's accuracy and average loss on the test set.
|
| 95 |
+
"""
|
| 96 |
model.eval()
|
| 97 |
total, correct, loss_total = 0, 0, 0.0
|
| 98 |
criterion = nn.CrossEntropyLoss()
|
| 99 |
+
|
| 100 |
with torch.no_grad():
|
| 101 |
for images, labels in testloader:
|
| 102 |
outputs = model(images)
|
|
|
|
| 105 |
total += labels.size(0)
|
| 106 |
correct += (preds == labels).sum().item()
|
| 107 |
loss_total += loss.item() * labels.size(0)
|
| 108 |
+
|
| 109 |
+
accuracy = round(100 * correct / total, 2)
|
| 110 |
+
avg_loss = round(loss_total / total, 4)
|
| 111 |
+
return accuracy, avg_loss
|
| 112 |
|
| 113 |
def run_unlearning(index_to_unlearn, learning_rate):
|
| 114 |
+
"""
|
| 115 |
+
Loads a pre-trained ResNet18 model, performs unlearning on a single training example,
|
| 116 |
+
and compares model performance before and after unlearning.
|
| 117 |
+
"""
|
| 118 |
+
# Set device (GPU if available, else CPU)
|
| 119 |
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
|
| 120 |
+
|
| 121 |
+
# Load the original pre-trained model and adjust for 10 classes
|
| 122 |
original_model = models.resnet18(weights=None)
|
| 123 |
original_model.fc = nn.Linear(original_model.fc.in_features, 10)
|
| 124 |
original_model.load_state_dict(torch.load("resnet18.pth", map_location=device))
|
| 125 |
original_model.to(device)
|
| 126 |
original_model.eval()
|
| 127 |
+
|
| 128 |
+
# Duplicate the model for the unlearning experiment
|
| 129 |
unlearned_model = models.resnet18(weights=None)
|
| 130 |
unlearned_model.fc = nn.Linear(unlearned_model.fc.in_features, 10)
|
| 131 |
unlearned_model.load_state_dict(torch.load("resnet18.pth", map_location=device))
|
| 132 |
unlearned_model.to(device)
|
| 133 |
+
|
| 134 |
# Get the sample to unlearn from the training set
|
| 135 |
image_tensor, label_idx = trainset[index_to_unlearn]
|
| 136 |
image_tensor = image_tensor.to(device)
|
| 137 |
label_name = cifar10_classes[label_idx]
|
| 138 |
print(f"ποΈ Actual Label Index: {label_idx} | Label Name: {label_name}")
|
| 139 |
+
|
| 140 |
# Prediction before unlearning
|
| 141 |
probs_before, pred_before = predict(original_model, image_tensor)
|
| 142 |
conf_before = probs_before[label_idx].item()
|
| 143 |
+
|
| 144 |
+
# Perform the unlearning process on the duplicated model
|
| 145 |
unlearn(unlearned_model, image_tensor, label_idx, learning_rate)
|
| 146 |
+
|
| 147 |
# Prediction after unlearning
|
| 148 |
probs_after, pred_after = predict(unlearned_model, image_tensor)
|
|
|
|
| 149 |
conf_after = probs_after[label_idx].item()
|
| 150 |
+
|
| 151 |
+
print("Post-unlearning probabilities:", probs_after)
|
| 152 |
+
|
| 153 |
+
# Evaluate the full test set performance for both models
|
| 154 |
orig_acc, orig_loss = evaluate_model(original_model, testloader)
|
| 155 |
unlearn_acc, unlearn_loss = evaluate_model(unlearned_model, testloader)
|
| 156 |
+
|
| 157 |
result = f"""
|
| 158 |
π Index Unlearned: {index_to_unlearn}
|
| 159 |
ποΈ Actual Label: {label_name} (Index: {label_idx})
|
|
|
|
| 160 |
π BEFORE Unlearning:
|
| 161 |
- Prediction: {cifar10_classes[pred_before]}
|
| 162 |
- Confidence: {conf_before:.10f}
|
|
|
|
| 163 |
π§½ AFTER Unlearning:
|
| 164 |
- Prediction: {cifar10_classes[pred_after]}
|
| 165 |
- Confidence: {conf_after:.10f}
|
|
|
|
| 166 |
π Confidence Drop: {conf_before - conf_after:.6f}
|
|
|
|
| 167 |
π§ͺ Test Set Performance:
|
| 168 |
- Original Model: {orig_acc:.2f}% accuracy, Loss: {orig_loss:.4f}
|
| 169 |
- Unlearned Model: {unlearn_acc:.2f}% accuracy, Loss: {unlearn_loss:.4f}
|
| 170 |
"""
|
| 171 |
return result
|
| 172 |
|
| 173 |
+
# Gradio interface for interactive unlearning demonstration
|
| 174 |
demo = gr.Interface(
|
| 175 |
fn=run_unlearning,
|
| 176 |
inputs=[
|
| 177 |
+
gr.Slider(0, len(trainset)-1, step=1, label="Select Index to Unlearn"),
|
| 178 |
gr.Slider(0.0001, 0.01, step=0.0001, value=0.005, label="Learning Rate (for Unlearning)")
|
| 179 |
],
|
| 180 |
outputs="text",
|