Framework for efficient machine unlearning that reformulates forgetting as inverse learning. Achieves significant computational speedup by replacing expensive Hessian operations with gradient-based optimization, enabling privacy-preserving model updates.
Framework for efficient machine unlearning that reformulates forgetting as inverse learning. Achieves significant computational speedup by replacing expensive Hessian operations with gradient-based optimization, enabling privacy-preserving model updates.
Efficient Machine Unlearning via Influence Approximation
Efficient Machine Unlearning addresses the critical challenge of removing specific training data from models for privacy compliance (e.g., GDPR right to be forgotten). Rather than expensive Hessian-based approaches, this framework establishes a theoretical connection between learning and unlearning, enabling efficient gradient-based deletion.
Core Concept
The fundamental insight is that unlearning can be viewed as the inverse of incremental learning. By reformulating the problem through this lens:
Unlearning becomes optimization rather than matrix inversion
"""
Compute influence of to_forget_batch on other_batch.
Measures how much removing to_forget would change loss on other_batch.
Args:
to_forget_batch: Data to remove
other_batch: Data to evaluate influence on
loss_fn: Loss function
Returns:
Influence score (higher = more influential)
"""
# Gradient of to_forget sample
self
# Gradient of other sample
self
# Influence = dot product of gradients
# High similarity means to_forget influences other's loss
return
def
approximate_hessian_inverse_sqrt
self,
data_batch: Dict,
loss_fn: callable,
num_samples: int = 50
"""
Approximate H^-1 v using Hutchinson trace estimator.
More efficient than exact Hessian computation.
Args:
data_batch: Batch to estimate on
loss_fn: Loss function
num_samples: Number of samples for estimation
Returns:
Approximated H^-1 v vector
"""
next
self
# Random vector for trace estimation
sum
for
in
self
# Compute Hv using finite differences
self
# Approximate H^-1 v using conjugate gradient or similar
"""
Compute Hessian-vector product: H v where H is loss Hessian.
Uses reverse-mode differentiation for efficiency.
"""
# Forward pass
self
'input_ids'
'attention_mask'
'labels'
# First gradient
self
True
True
True
1
for
in
if
is
not
None
# Gradient-vector dot product
# Second gradient (Hessian-vector product)
self
True
True
1
for
in
if
is
not
None
return
Step 2: Reformulate unlearning as inverse learning
View the forgetting problem through the lens of incremental learning:
classIncrementalLearningPerspective:
"""
Reformulates unlearning as inverse of incremental learning.
Key insight: If data x was added to model M to get M', then
removing x from M' should reverse the process.
"""def__init__(self, model: nn.Module):
self.model = model
defcompute_unlearning_gradient(self,
to_forget_batch: Dict,
loss_fn: callable) -> Dict:
"""
Compute gradient direction for removing influence of batch.
Instead of computing H^-1 g (inverse problem),
directly optimize to remove influence.
Args:
to_forget_batch: Batch to remove
loss_fn: Loss function
Returns:
Direction to move parameters to unlearn data
"""# Compute loss gradient for forget batch
outputs = self.model(to_forget_batch['input_ids'])
loss = loss_fn(outputs, to_forget_batch.get('labels'))
# Gradient indicating how to fit this data
grads = torch.autograd.grad(
loss,
self.model.parameters(),
retain_graph=True,
create_graph=False,
allow_unused=True
)
# Unlearning direction: negative of learning gradient# Moving opposite direction removes the data's influence
unlearn_direction = {
name: -g if g isnotNoneelseNonefor name, g inzip(
[n for n, _ inself.model.named_parameters()],
grads
)
}
return unlearn_direction
defincremental_unlearning_update(self,
to_forget_batch: Dict,
learning_rate: float,
loss_fn: callable) -> Dict:
"""
Single unlearning step using incremental perspective.
Move parameters opposite to how they'd move if learning.
"""
unlearn_dir = self.compute_unlearning_gradient(to_forget_batch, loss_fn)
updates = {}
for name, param inself.model.named_parameters():
if name in unlearn_dir and unlearn_dir[name] isnotNone:
# Update: move opposite to learning direction
param.data = param.data - learning_rate * unlearn_dir[name]
updates[name] = -learning_rate * unlearn_dir[name]
return updates
classGradientBasedUnlearner:
"""Efficiently unlearns data through gradient optimization"""def__init__(self, model: nn.Module, learning_rate: float = 1e-4):
self.model = model
self.learning_rate = learning_rate
self.optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)
self.incremental = IncrementalLearningPerspective(model)
defunlearn_sample(self, to_forget_sample: Dict,
loss_fn: callable,
num_steps: int = 5) -> Dict:
"""
Unlearn a single sample through iterative optimization.
Args:
to_forget_sample: Single sample to remove
loss_fn: Loss function
num_steps: Number of gradient steps
Returns:
Metrics about unlearning process
"""
initial_loss = None
metrics = {
'step_losses': [],
'parameter_changes': [],
'num_steps': num_steps
}
for step inrange(num_steps):
# Compute loss on sample to forget
outputs = self.model(
to_forget_sample['input_ids'],
to_forget_sample.get('attention_mask')
)
loss = loss_fn(outputs, to_forget_sample.get('labels'))
if initial_loss isNone:
initial_loss = loss.item()
metrics['step_losses'].append(loss.item())
# Unlearning step: move opposite to learning direction
unlearn_update = self.incremental.incremental_unlearning_update(
to_forget_sample,
self.learning_rate,
loss_fn
)
param_change = sum(
torch.norm(v).item() for v in unlearn_update.values()
if v isnotNone
)
metrics['parameter_changes'].append(param_change)
metrics['loss_reduction'] = initial_loss - metrics['step_losses'][-1]
return metrics
defunlearn_batch(self, to_forget_batch: Dict,
loss_fn: callable,
batch_num_steps: int = 10) -> Dict:
"""
Unlearn all samples in a batch.
Args:
to_forget_batch: Full batch to remove
loss_fn: Loss function
batch_num_steps: Total optimization steps for batch
Returns:
Unlearning metrics
"""
metrics = {
'batch_loss_initial': None,
'batch_loss_final': None,
'num_steps': batch_num_steps,
'total_param_change': 0.0
}
for step inrange(batch_num_steps):
# Forward pass
outputs = self.model(
to_forget_batch['input_ids'],
to_forget_batch.get('attention_mask')
)
loss = loss_fn(outputs, to_forget_batch.get('labels'))
if step == 0:
metrics['batch_loss_initial'] = loss.item()
# Backward passself.optimizer.zero_grad()
# Maximize loss (adversarial: try to increase loss on forget data)# This removes the learned associations
loss.backward()
# Update with negated gradients (opposite direction)for param inself.model.parameters():
if param.grad isnotNone:
param.data -= self.learning_rate * param.grad
metrics['total_param_change'] += torch.norm(param.grad).item()
# Final evaluationwith torch.no_grad():
outputs = self.model(to_forget_batch['input_ids'])
final_loss = loss_fn(outputs, to_forget_batch.get('labels'))
metrics['batch_loss_final'] = final_loss.item()
return metrics
Step 4: Implement membership inference test
Verify that unlearning was successful:
classMembershipInferenceTest:
"""Tests whether data has been successfully unlearned"""def__init__(self, model: nn.Module):
self.model = model
defcompute_membership_score(self, sample: Dict,
loss_fn: callable) -> float:
"""
Compute membership score: model's loss on sample.
High loss = likely not a training sample (unlearned)
Low loss = likely a training sample (still learned)
Args:
sample: Sample to test membership of
loss_fn: Loss function
Returns:
Membership score (higher = more likely member)
"""with torch.no_grad():
outputs = self.model(
sample['input_ids'],
sample.get('attention_mask')
)
loss = loss_fn(outputs, sample.get('labels'))
# Inverse loss as membership score# (lower loss = higher membership probability)
membership_score = -loss.item()
return membership_score
defmembership_inference_attack(self,
train_samples: List[Dict],
test_samples: List[Dict],
loss_fn: callable) -> Dict:
"""
Perform membership inference attack to test unlearning.
Args:
train_samples: Original training samples
test_samples: Non-training samples
loss_fn: Loss function
Returns:
AUC score indicating inference accuracy
"""
train_scores = [
self.compute_membership_score(s, loss_fn)
for s in train_samples
]
test_scores = [
self.compute_membership_score(s, loss_fn)
for s in test_samples
]
# Compute AUC: can we distinguish train from test?from sklearn.metrics import roc_auc_score
y_true = [1] * len(train_scores) + [0] * len(test_scores)
y_pred = train_scores + test_scores
auc = roc_auc_score(y_true, y_pred)
return {
'auc': auc,
'train_avg_score': np.mean(train_scores),
'test_avg_score': np.mean(test_scores),
'separation': np.mean(train_scores) - np.mean(test_scores)
}
defverify_unlearning(self,
unlearned_samples: List[Dict],
remaining_samples: List[Dict],
loss_fn: callable,
threshold: float = 0.5) -> Dict:
"""
Verify that samples have been unlearned.
Args:
unlearned_samples: Samples that should be forgotten
remaining_samples: Samples that should still be known
loss_fn: Loss function
threshold: Threshold for considering unlearned
Returns:
Verification results
"""
unlearned_scores = [
self.compute_membership_score(s, loss_fn)
for s in unlearned_samples
]
remaining_scores = [
self.compute_membership_score(s, loss_fn)
for s in remaining_samples
]
# Successful unlearning: unlearned samples have much higher loss
unlearning_gap = np.mean(unlearned_scores) - np.mean(remaining_scores)
successful_unlearns = sum(
1for score in unlearned_scores
if score > threshold
)
return {
'unlearning_gap': unlearning_gap,
'successful_unlearns': successful_unlearns,
'total_samples': len(unlearned_samples),
'success_rate': successful_unlearns / len(unlearned_samples),
'verified': successful_unlearns / len(unlearned_samples) > 0.9
}
Step 5: Implement end-to-end unlearning pipeline
Integrate all components into complete unlearning workflow:
classUnlearningPipeline:
"""Complete efficient machine unlearning system"""def__init__(self, model: nn.Module, loss_fn: callable,
learning_rate: float = 1e-4):
self.model = model
self.loss_fn = loss_fn
self.unlearner = GradientBasedUnlearner(model, learning_rate)
self.verifier = MembershipInferenceTest(model)
self.influence = InfluenceApproximator(model)
defunlearn_request(self, to_forget_data: List[Dict],
num_steps: int = 10) -> Dict:
"""
Process unlearning request for data batch.
Args:
to_forget_data: Data to remove
num_steps: Optimization steps
Returns:
Unlearning report
"""
report = {
'num_samples': len(to_forget_data),
'unlearning_metrics': None,
'verification': None,
'success': False
}
# Step 1: Compute influence scores
influence_scores = []
for sample in to_forget_data:
score = self.influence.compute_influence_score(
sample,
{'input_ids': torch.zeros(1)}, # Dummyself.loss_fn
)
influence_scores.append(score)
# Step 2: Unlearn data
batch_metrics = self.unlearner.unlearn_batch(
{
'input_ids': torch.cat([s['input_ids'] for s in to_forget_data]),
'labels': torch.cat([s.get('labels', s['input_ids'])
for s in to_forget_data])
},
self.loss_fn,
batch_num_steps=num_steps
)
report['unlearning_metrics'] = batch_metrics
# Step 3: Verify unlearning
verification = self.verifier.verify_unlearning(
to_forget_data,
[], # Would have hold-out test set in practiceself.loss_fn
)
report['verification'] = verification
report['success'] = verification['verified']
return report
defevaluate_utility(self, test_data: List[Dict]) -> float:
"""
Evaluate model utility preservation after unlearning.
Args:
test_data: Test set for evaluation
Returns:
Accuracy or loss on test set
"""with torch.no_grad():
total_loss = 0.0for sample in test_data:
outputs = self.model(
sample['input_ids'],
sample.get('attention_mask')
)
loss = self.loss_fn(outputs, sample.get('labels'))
total_loss += loss.item()
avg_loss = total_loss / len(test_data)
return avg_loss
