AI 模型隐私实践笔记

AI安全 差分隐私 联邦学习 模型安全
作者: 醉月思 发布日期: 2023-06-25 分类: 技术实践

做 AI 模型隐私，纸面上是"保护训练数据"，落到生产里才发现攻击面宽得多：有人推断某条记录是否在训练集里，有人从模型参数反推用户行为，有人盯推理结果做侧信道分析。

我们遇到过医疗模型被成员推断、金融模型被反推交易记录、推荐逻辑被竞争对手抄走——合规是底线，业务安全才是逼着你动手的理由。

为什么这事儿值得折腾

最开始只是觉得要把训练数据保护起来，后来发现事情没那么简单。模型攻击者不只想要你的训练数据，还想要模型参数、推理结果，甚至是单个用户的具体行为。

我们遇到过这样的场景：

  • 医疗模型被攻击，推断出某个病人是否在训练集中
  • 金融模型被反推，猜出用户的具体交易记录
  • 推荐系统被窃取，竞争对手直接复制了我们的业务逻辑

所以这事儿不只是合规要求，是业务安全。

差分隐私实战

差分隐私听起来很玄，但做起来就是加噪声。

基础差分隐私

import numpy as np

def gaussian_mechanism(data, sensitivity, epsilon, delta):
    """
    高斯机制实现差分隐私
    :param data: 原始数据
    :param sensitivity: 数据敏感度
    :param epsilon: 隐私预算
    :param delta: 失败概率
    """
    # 计算噪声标准差
    sigma = np.sqrt(2 * np.log(1.25 / delta)) * sensitivity / epsilon

    # 添加高斯噪声
    noise = np.random.normal(0, sigma)
    return data + noise

# 使用示例
original_value = 42.5
sensitivity = 1.0  # L2 敏感度
epsilon = 1.0
delta = 1e-5

private_value = gaussian_mechanism(original_value, sensitivity, epsilon, delta)
print(f"原始值: {original_value}, 加噪声后: {private_value}")

训练时的差分隐私

用 PyTorch 实现差分隐私训练：

import torch
import torch.nn as nn
from opacus import PrivacyEngine

# 定义模型
model = nn.Sequential(
    nn.Linear(784, 256),
    nn.ReLU(),
    nn.Linear(256, 128),
    nn.ReLU(),
    nn.Linear(128, 10)
)

# 定义优化器
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

# 创建隐私引擎
privacy_engine = PrivacyEngine()

# 附着到模型和优化器
model, optimizer, train_loader = privacy_engine.make_private_with_epsilon(
    module=model,
    optimizer=optimizer,
    data_loader=train_loader,
    epochs=10,
    target_epsilon=3.0,
    target_delta=1e-5,
    max_grad_norm=1.0,
)

# 正常训练
for epoch in range(10):
    for data, target in train_loader:
        optimizer.zero_grad()
        output = model(data)
        loss = nn.functional.cross_entropy(output, target)
        loss.backward()
        optimizer.step()

    # 检查隐私预算消耗
    epsilon = privacy_engine.get_epsilon(delta=1e-5)
    print(f"Epoch {epoch}, Epsilon: {epsilon:.2f}")

坑一：精度损失太明显

刚开始直接加噪声，模型精度掉得太厉害。从 95% 掉到 70%，业务方直接翻脸。

解决方法：

  • 调整隐私预算，epsilon 从 1.0 提升到 3.0
  • 降低敏感度估计，使用 clip_by_norm 限制梯度
  • 增加模型容量，用更强的模型抗噪声
# 调整后的参数
target_epsilon = 3.0  # 从 1.0 提升到 3.0
max_grad_norm = 0.5   # 从 1.0 降低到 0.5

# 增强模型
model = nn.Sequential(
    nn.Linear(784, 512),  # 增加层数和宽度
    nn.ReLU(),
    nn.Linear(512, 256),
    nn.ReLU(),
    nn.Linear(256, 128),
    nn.ReLU(),
    nn.Linear(128, 10)
)

坑二：训练时间暴涨

差分隐私训练比普通训练慢 3-5 倍，主要花在梯度裁剪和噪声计算上。

解决方法：

  • 批大一点，减少梯度裁剪频率
  • 用混合精度训练
  • 简化模型结构
# 使用混合精度训练
from torch.cuda.amp import autocast, GradScaler

scaler = GradScaler()

for data, target in train_loader:
    optimizer.zero_grad()

    with autocast():
        output = model(data)
        loss = nn.functional.cross_entropy(output, target)

    scaler.scale(loss).backward()
    scaler.step(optimizer)
    scaler.update()

联邦学习实践

联邦学习让数据不出本地，但能共用模型。

基础联邦学习

from flwr.client import NumPyClient, ClientApp
from flwr.common import ndarrays_to_parameters

class FlowerClient(NumPyClient):
    def __init__(self, model, train_loader, test_loader):
        self.model = model
        self.train_loader = train_loader
        self.test_loader = test_loader

    def get_parameters(self, config):
        return [val.cpu().numpy() for _, val in self.model.state_dict().items()]

    def set_parameters(self, parameters):
        params_dict = zip(self.model.state_dict().keys(), parameters)
        state_dict = {k: torch.tensor(v) for k, v in params_dict}
        self.model.load_state_dict(state_dict, strict=True)

    def fit(self, parameters, config):
        self.set_parameters(parameters)

        # 本地训练
        optimizer = torch.optim.SGD(self.model.parameters(), lr=0.01)
        for _ in range(5):  # 本地训练轮数
            for data, target in self.train_loader:
                optimizer.zero_grad()
                output = self.model(data)
                loss = nn.functional.cross_entropy(output, target)
                loss.backward()
                optimizer.step()

        return self.get_parameters({}), len(self.train_loader.dataset), {}

    def evaluate(self, parameters, config):
        self.set_parameters(parameters)
        correct = 0
        total = 0
        with torch.no_grad():
            for data, target in self.test_loader:
                output = self.model(data)
                _, predicted = torch.max(output.data, 1)
                total += target.size(0)
                correct += (predicted == target).sum().item()
        accuracy = correct / total
        return float(accuracy), len(self.test_loader.dataset), {}

# 创建客户端应用
def client_fn(cid):
    model = create_model()
    train_loader = get_train_loader(int(cid))
    test_loader = get_test_loader(int(cid))
    return FlowerClient(model, train_loader, test_loader)

app = ClientApp(client_fn)

服务器端聚合

from flwr.server import ServerApp, ServerConfig
from flwr.server.strategy import FedAvg

# 定义聚合策略
strategy = FedAvg(
    fraction_fit=0.1,  # 每轮选择 10% 的客户端
    fraction_evaluate=0.05,
    min_fit_clients=10,
    min_evaluate_clients=5,
    min_available_clients=100,
)

def server_fn(context):
    return ServerConfig(num_rounds=100)

server = ServerApp(server_fn=server_fn)

坑三：客户端掉线严重

联邦学习最头疼的是客户端掉线。100 个客户端，经常只有 60-70 个在线。

解决方法：

  • 增加客户端基数，从 100 增加到 200
  • 降低每轮要求的客户端数量
  • 实现客户端容错机制
# 调整聚合策略
strategy = FedAvg(
    fraction_fit=0.05,  # 从 0.1 降低到 0.05
    fraction_evaluate=0.02,
    min_fit_clients=5,   # 从 10 降低到 5
    min_evaluate_clients=3,
    min_available_clients=200,  # 从 100 增加到 200
    # 添加容错
    drop_out_clients=True,
)

坑四：模型更新太大

有些客户端训练质量很差，推送的参数更新会破坏模型。

解决方法：

  • 限制本地训练轮数
  • 实现安全聚合，检测异常更新
  • 增加模型更新的过滤机制
from flwr.server.strategy import FedAvg
import numpy as np

class SafeFedAvg(FedAvg):
    def aggregate_fit(self, server_round, results, failures):
        if not results:
            return None, {}

        # 提取所有更新
        weights_results = [(parameters, num_examples) for parameters, num_examples, _ in results]

        # 计算平均值和标准差
        all_updates = []
        for params, _ in weights_results:
            all_updates.append(np.concatenate([p.flatten() for p in params]))
        all_updates = np.array(all_updates)

        mean = np.mean(all_updates, axis=0)
        std = np.std(all_updates, axis=0)

        # 过滤异常值
        filtered_results = []
        for (parameters, num_examples, _), update in zip(results, all_updates):
            z_score = np.abs((update - mean) / (std + 1e-8))
            if np.max(z_score) < 3:  # 3 sigma 原则
                filtered_results.append((parameters, num_examples, _))

        return super().aggregate_fit(server_round, filtered_results, failures)

模型推理保护

训练完成后的推理阶段也不能放松。

模型加密

import tenseal as ts

# 创建 TenSEAL 上下文
context = ts.context(
    ts.SCHEME_TYPE.CKKS,
    poly_modulus_degree=8192,
    coeff_mod_bit_sizes=[60, 40, 40, 60]
)
context.global_scale = 2**40
context.generate_galois_keys()

# 加密模型权重
model_weights = model.state_dict()
encrypted_weights = {}

for name, param in model_weights.items():
    encrypted_weights[name] = ts.ckks_tensor(context, param.cpu().numpy())

# 保存加密模型
torch.save(encrypted_weights, 'encrypted_model.pth')

安全推理

def secure_inference(encrypted_weights, input_data):
    """
    在加密数据上进行推理
    """
    context = encrypted_weights[list(encrypted_weights.keys())[0]].context

    # 加密输入
    encrypted_input = ts.ckks_tensor(context, input_data)

    # 简单的线性层计算
    encrypted_output = encrypted_input.mm(encrypted_weights['fc1.weight'].data)
    encrypted_output = encrypted_output + encrypted_weights['fc1.bias'].data

    # 解密结果
    decrypted_output = encrypted_output.decrypt()
    return decrypted_output

坑五：加密开销太大

加密后的推理速度慢 10-20 倍，延迟高得离谱。

解决方法：

  • 只加密敏感层，其他层保持明文
  • 用更轻量的加密方案
  • 批量推理，摊薄加密成本
# 只加密输出层
class PartiallyEncryptedModel(nn.Module):
    def __init__(self, base_model, context):
        super().__init__()
        self.features = nn.Sequential(*list(base_model.children())[:-2])
        self.context = context

        # 加密最后两层
        last_layer_weight = list(base_model.children())[-2].weight.data
        last_layer_bias = list(base_model.children())[-2].bias.data
        output_layer_weight = list(base_model.children())[-1].weight.data
        output_layer_bias = list(base_model.children())[-1].bias.data

        self.encrypted_weight = ts.ckks_tensor(context, last_layer_weight)
        self.encrypted_bias = ts.ckks_tensor(context, last_layer_bias)
        self.encrypted_output_weight = ts.ckks_tensor(context, output_layer_weight)
        self.encrypted_output_bias = ts.ckks_tensor(context, output_layer_bias)

    def forward(self, x):
        # 前几层正常推理
        x = self.features(x)
        x = x.view(x.size(0), -1)

        # 最后两层加密计算
        x_encrypted = ts.ckks_tensor(self.context, x.cpu().numpy())
        x_encrypted = x_encrypted.mm(self.encrypted_weight.data) + self.encrypted_bias.data
        x_encrypted = x_encrypted.mm(self.encrypted_output_weight.data) + self.encrypted_output_bias.data

        # 解密结果
        output = x_encrypted.decrypt()
        return torch.tensor(output)

模型攻击与防御

知道攻击方式才能更好地防御。

成员推断攻击

import torch.nn.functional as F

class MembershipInferenceAttack:
    """
    成员推断攻击：判断样本是否在训练集中
    """
    def __init__(self, target_model, shadow_models):
        self.target_model = target_model
        self.shadow_models = shadow_models
        self.attack_model = None

    def train_attack_model(self, member_data, non_member_data):
        """
        训练攻击模型
        """
        # 收集影子模型的预测置信度
        attack_data = []
        attack_labels = []

        for shadow_model in self.shadow_models:
            for data, _ in member_data:
                output = shadow_model(data)
                confidence = F.softmax(output, dim=1)
                attack_data.append(confidence.detach().numpy())
                attack_labels.append(1)  # 成员

            for data, _ in non_member_data:
                output = shadow_model(data)
                confidence = F.softmax(output, dim=1)
                attack_data.append(confidence.detach().numpy())
                attack_labels.append(0)  # 非成员

        # 训练攻击模型
        from sklearn.ensemble import RandomForestClassifier
        self.attack_model = RandomForestClassifier()
        self.attack_model.fit(attack_data, attack_labels)

    def attack(self, data):
        """
        对目标模型进行攻击
        """
        output = self.target_model(data)
        confidence = F.softmax(output, dim=1)
        return self.attack_model.predict(confidence.detach().numpy())

防御措施

def defend_membership_inference(model, train_loader, test_loader):
    """
    防御成员推断攻击
    """
    # 方法一：正则化
    def regularized_loss(output, target, model, lambda_reg=0.01):
        ce_loss = F.cross_entropy(output, target)

        # 添加 L2 正则化
        l2_loss = 0
        for param in model.parameters():
            l2_loss += torch.norm(param, 2)

        return ce_loss + lambda_reg * l2_loss

    # 方法二：对抗训练
    def adversarial_training(model, train_loader, epochs=5):
        optimizer = torch.optim.SGD(model.parameters(), lr=0.01)

        for epoch in range(epochs):
            for data, target in train_loader:
                data.requires_grad = True

                # 正常前向传播
                output = model(data)
                loss = F.cross_entropy(output, target)

                # 计算梯度
                loss.backward()

                # 生成对抗样本
                data_grad = data.grad.data
                perturbed_data = data + 0.01 * data_grad.sign()

                # 对抗训练
                output_adv = model(perturbed_data)
                loss_adv = F.cross_entropy(output_adv, target)

                optimizer.zero_grad()
                (loss + loss_adv).backward()
                optimizer.step()

    # 方法三：输出噪声
    def add_output_noise(output, epsilon=1.0):
        noise = torch.randn_like(output) * (epsilon / 2)
        return output + noise

    return regularized_loss, adversarial_training, add_output_noise

部署时的注意事项

配置检查清单

# privacy-checklist.yml
model_privacy:
  training:
    differential_privacy:
      enabled: true
      epsilon: 3.0
      delta: 1e-5
      max_grad_norm: 0.5

    federated_learning:
      enabled: false  # 根据项目需要选择
      min_clients: 10
      aggregation: "fedavg"

  inference:
    encryption:
      enabled: true
      scheme: "ckks"
      partial_encryption: true

    rate_limiting:
      enabled: true
      max_requests_per_minute: 1000

    input_validation:
      enabled: true
      max_input_size: 1024

  monitoring:
    privacy_budget_tracking: true
    attack_detection: true
    anomaly_detection: true

监控指标

class PrivacyMonitor:
    def __init__(self):
        self.privacy_budget = 0.0
        self.max_privacy_budget = 3.0
        self.attack_count = 0

    def log_query(self, epsilon_used):
        """
        记录查询的隐私预算消耗
        """
        self.privacy_budget += epsilon_used
        if self.privacy_budget > self.max_privacy_budget:
            raise ValueError("隐私预算耗尽")

    def detect_attack(self, query_pattern):
        """
        检测可疑查询模式
        """
        # 简单的异常检测
        if query_pattern['frequency'] > 100:  # 每分钟超过 100 次
            self.attack_count += 1
            return True
        return False

    def get_status(self):
        """
        获取隐私状态
        """
        return {
            'privacy_budget': self.privacy_budget,
            'remaining_budget': self.max_privacy_budget - self.privacy_budget,
            'attack_count': self.attack_count,
            'status': 'safe' if self.attack_count == 0 else 'under_attack'
        }

写在最后

AI 模型隐私保护这事儿，不是一次性方案，是个持续的过程。

解决了：

  • 训练数据隐私泄露
  • 模型参数被盗取
  • 推理结果被推断

带来了：

  • 训练成本增加 50%
  • 推理延迟增加 2-3 倍
  • 系统复杂度大幅提升

但有些场景就是值得：医疗、金融、用户隐私敏感的业务。不是所有模型都需要这么重的保护，但一旦需要，就得做扎实。

最后说一句，隐私保护不是要完美，是要让攻击成本高于攻击收益。能用技术手段解决的用技术，不能用技术的，至少要有监控和审计。

这次 AI 模型隐私保护实践花了三个月，从差分隐私到联邦学习，再到模型加密，总算是把整套流程跑通了。业务方的焦虑倒是缓解了不少，但我们知道这只是个开始。

版权声明: 本文首发于 指尖魔法屋-AI 模型隐私实践笔记（https://blog.thinkmoon.cn/post/258-ai-model-privacy-design-protection-practice/） 转载或引用必须申明原指尖魔法屋来源及源地址！

相关阅读

根据标签、分类和主题为你挑选的相近文章