数据遗忘权实现：从训练集中删除特定用户数据

在机器学习成为现代应用核心的时代，用户数据的隐私保护与合规要求日益严格。数据遗忘权（Right to be Forgotten） 要求系统不仅能删除存储的原始数据，还要消除这些数据对已训练模型的影响。本教程面向开发者与数据工程师，系统讲解如何从训练集中删除特定用户数据，并确保模型不再携带该用户的“记忆”。

理解数据遗忘权在机器学习中的挑战

为什么简单的删除不够？

在传统数据库中，删除一行记录是直接且彻底的。但在机器学习中，模型参数是训练数据的一种压缩表示。即使用户数据被删除，模型权重中依然保留着该数据的统计贡献。例如：

• 用户A的浏览记录参与了推荐模型训练，即使从日志库删除，推荐模型仍会输出与A偏好相似的推荐。
• 在联邦学习或预训练模型中，数据混合在大量参数中，难以回溯。

从训练集删除数据，意味着必须消除该数据对模型参数的影响，而不仅仅是删除文件。

重新训练：基准但昂贵的方法

最彻底的方法是从原始训练集中移除指定用户数据，然后重新训练整个模型。优点是无偏差，缺点是计算成本极高，对于大模型几乎不可行。

主流实现技术概览

精确遗忘 vs. 近似遗忘

• 精确遗忘：要求遗忘后的模型与从未使用该数据训练的模型在功能上完全一致。
• 近似遗忘：仅保证输出不可被攻击者用于推断被遗忘的数据点，且模型性能保持在可接受范围。由于计算开销，实际系统多采用近似遗忘。

核心技术分类

• 训练数据拆分与重训（SISA）：通过分片与检查点将重训练成本控制在小范围内。
• 梯度影响移除：直接修正模型参数，抵消被遗忘数据在训练中的贡献。
• 差分隐私保证：在训练时注入噪声，使单个数据点对模型影响受限，遗忘更容易。

SISA 方法：分片、隔离、切片与聚合

SISA（Sharded, Isolated, Sliced, and Aggregated） 是目前实用性最强的遗忘权实现框架，由论文 Eternal Sunshine of the Spotless Net 提出。其核心思想是 将训练数据划分成多个互不重叠的分片，每个分片独立训练一个子模型，最后用聚合模型做推理。

架构设计

1. 分片（Sharding）：将完整训练集随机且均匀地分成 S 个分片。
2. 隔离训练（Isolated Training）：每个分片独立训练一个模型副本，完全不接触其他分片数据。
3. 切片（Slicing）：在每个分片内部，训练过程再切分为多个时间切片，保存中间检查点。
4. 聚合（Aggregation）：推理时将各个子模型的输出进行聚合（例如投票、平均 logits），得到最终预测。

遗忘操作流程

当需要删除某个用户 u 的数据时：

1. 定位数据所在分片：由于数据划分是确定性的，u 的数据只属于某一个分片（比如分片 k）。
2. 回退到干净检查点：在该分片的训练时间线上，找到该用户数据被首次使用之前的检查点（依靠数据记录可查）。
3. 从该检查点继续训练：移除用户 u 的所有样本后，利用该分片中剩余数据从检查点恢复训练至收敛。
4. 更新聚合模型：用重新训练后的子模型替换旧子模型，无须改动其他分片。

这种方式将重训练成本从全数据集降低到 单个分片的部分训练周期，极大提升了效率。

代码示例（PyTorch 伪实现）

import torch
from torch.utils.data import DataLoader, Subset
import numpy as np

class SISAModelHub:
    def __init__(self, base_model, dataset, S=10, checkpoint_interval=100):
        self.S = S
        self.shards = self._shard_dataset(dataset, S)
        self.models = [copy.deepcopy(base_model) for _ in range(S)]
        self.optimizers = [torch.optim.Adam(m.parameters()) for m in self.models]
        self.checkpoints = {i: [] for i in range(S)}  # 保存各分片检查点
        self.checkpoint_interval = checkpoint_interval
        self.data_map = {}  # user_id -> (shard_idx, sample_indices)

    def _shard_dataset(self, dataset, S):
        # 按用户哈希分片，保证同一用户的所有数据在同一个分片
        indices_per_shard = [[] for _ in range(S)]
        for idx, (data, user_id) in enumerate(dataset):
            shard_id = hash(user_id) % S
            indices_per_shard[shard_id].append(idx)
        return [Subset(dataset, indices) for indices in indices_per_shard]

    def train(self, epochs=5):
        for shard_id in range(self.S):
            loader = DataLoader(self.shards[shard_id], batch_size=32, shuffle=True)
            for epoch in range(epochs):
                for batch_idx, batch in enumerate(loader):
                    # 标准训练步骤
                    loss = self._train_step(batch, shard_id)
                    # 定期保存检查点
                    if batch_idx % self.checkpoint_interval == 0:
                        self.checkpoints[shard_id].append({
                            'model_state': deepcopy(self.models[shard_id].state_dict()),
                            'batch_id': batch_idx,
                            'epoch': epoch
                        })

    def unlearn_user(self, user_id):
        # 1. 确定分片
        shard_id = hash(user_id) % self.S
        # 2. 找到该用户数据首次加入训练的时间点（假定记录在 data_map）
        if user_id not in self.data_map:
            return  # 数据不存在
        first_batch_used = self.data_map[user_id]['first_batch']
        # 3. 找到 first_batch_used 之前的最近检查点
        restore_checkpoint = None
        for cp in reversed(self.checkpoints[shard_id]):
            if cp['batch_id'] < first_batch_used:
                restore_checkpoint = cp
                break
        if restore_checkpoint is None:
            # 没有合适检查点，则从头开始训练该分片
            self.models[shard_id].apply(weight_reset)
        else:
            self.models[shard_id].load_state_dict(restore_checkpoint['model_state'])
        # 4. 从该分片数据集中移除该用户所有样本
        self._remove_user_from_shard(shard_id, user_id)
        # 5. 继续训练至收敛（简化示例，仅训练若干 epoch）
        loader = DataLoader(self.shards[shard_id], batch_size=32, shuffle=True)
        for epoch in range(5):  # 实际应使用早停
            for batch in loader:
                self._train_step(batch, shard_id)
        # 6. 清除过期检查点，更新聚合逻辑即可

    def predict(self, x):
        # 简单平均聚合
        outputs = [model(x) for model in self.models]
        return torch.mean(torch.stack(outputs), dim=0)

说明：该示例重点展示遗忘流程，省略了具体训练步骤与用户数据时间戳记录。实际部署时需配合数据管道记录每个样本参与训练的批次时间。

SISA 的权衡与优化

• 分片数：分片越多，单次遗忘成本越低，但聚合模型的精度可能因分片数据量减少而下降，推理时需要多次前向传播，增加延迟。
• 检查点粒度：检查点保存越频繁，回退成本越低，但存储开销增大。可采用增量快照或只保存参数变化。
• 数据不均衡：采用哈希分片时，可能出现某些分片样本过多或过少，可结合数据均衡策略。

其他实现路径

基于梯度更新修正

对于深度学习模型，可通过牛顿法或影响函数近似移除特定样本对参数的影响。公式上：

[ \theta_{\text{unlearn}} \approx \theta - H^{-1} \nabla_{\theta} L(z_{\text{forget}}, \theta) ]

其中 ( H ) 是海森矩阵，( L ) 为损失。该方法计算成本高，适用于少量样本删除，不适用于大型模型。近期工作如 SCRUB 或 Bad Teaching 通过对抗训练让模型“忘记”特定分布，实现近似遗忘。

差分隐私训练驱动的“免费”遗忘

如果在训练时使用了 DP-SGD，单一样本对模型梯度的影响被严格限制。此时遗忘可通过简单的模型参数加法性修正实现，甚至只需宣布“该用户已不在训练集”即可满足法律意义上的遗忘要求，因为模型天然满足移除某个用户后不可区分的性质。但差分隐私会引入效能损失，需在隐私预算与精度间权衡。

实践检查清单：从训练集中安全删除用户数据

1. 数据溯源：确保可以追溯每一条训练数据对应的用户标识，以及其在训练流程中的使用时间点。
2. 架构选型：根据数据量、遗忘频率、延迟要求选择 SISA、影响函数或差分隐私路径。
3. 日志与审计：记录每次遗忘操作的发起、完成时间、涉及的分片和检查点，满足合规审查。
4. 效果验证：通过成员推理攻击（Membership Inference Attack）测试遗忘后模型是否仍泄露已删除用户的信息。
5. 性能回归测试：遗忘后必须通过自动化测试套件，确保模型在主任务上的指标（准确率、F1等）未出现不可接受的下降。
6. 持续集成：将遗忘请求作为系统 API 的一部分，支持批量遗忘与增量遗忘，确保处理时效。

总结

数据遗忘权的机器学习实现已从理论走向工程化。SISA 架构提供了高效、可审计的分片重训练方案，是当前落地的主流选择。开发者应根据系统规模与具体需求，在精确性与效率间做出权衡。无论选择何种技术，做到透明的数据血缘追踪和持续的遗忘效果验证，才是真正完成了“从训练集中删除用户数据”这一使命。

开始设计你的遗忘系统时，不妨先在小规模分片上实现 SISA 原型，再逐步扩展到生产环境。隐私保护，从可控的遗忘开始。