大模型与数据隐私保护_02

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, Dataset
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import matplotlib.pyplot as plt
import time
import copy

# 差分隐私噪声添加函数
class DPPrivacy:
    def __init__(self, epsilon=1.0, delta=1e-5):
        """初始化差分隐私参数
        Args:
            epsilon: 隐私预算，值越小隐私保护越强
            delta: 失败概率，通常设置为1e-5
        """
        self.epsilon = epsilon
        self.delta = delta
        self.mechanism = "gaussian"
        
    def add_noise(self, data, sensitivity=1.0):
        """向数据添加差分隐私噪声
        Args:
            data: 输入数据
            sensitivity: 数据敏感度
        Returns:
            添加噪声后的数据
        """
        if self.mechanism == "laplace":
            # 拉普拉斯机制
            scale = sensitivity / self.epsilon
            noise = np.random.laplace(loc=0, scale=scale, size=data.shape)
        else:
            # 高斯机制
            sigma = np.sqrt(2 * np.log(1.25 / self.delta)) * sensitivity / self.epsilon
            noise = np.random.normal(loc=0, scale=sigma, size=data.shape)
            
        return data + noise

# 自定义数据集
class CustomDataset(Dataset):
    def __init__(self, data, labels, apply_dp=False, dp_epsilon=1.0):
        self.data = data
        self.labels = labels
        self.apply_dp = apply_dp
        self.dp = DPPrivacy(epsilon=dp_epsilon)
        
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        sample = self.data[idx]
        label = self.labels[idx]
        
        # 如果启用差分隐私，则添加噪声
        if self.apply_dp:
            sample = self.dp.add_noise(sample)
            
        return torch.tensor(sample, dtype=torch.float32), torch.tensor(label, dtype=torch.long)

# 简单的分类模型
class SimpleClassifier(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim):
        super(SimpleClassifier, self).__init__()
        self.layers = nn.Sequential(
            nn.Linear(input_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, output_dim)
        )
        
    def forward(self, x):
        return self.layers(x)

# 训练函数
def train_model(model, train_loader, val_loader, criterion, optimizer, num_epochs=10):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.to(device)
    
    best_model_weights = copy.deepcopy(model.state_dict())
    best_acc = 0.0
    
    # 记录训练过程
    history = {
        'train_loss': [],
        'train_acc': [],
        'val_loss': [],
        'val_acc': []
    }
    
    for epoch in range(num_epochs):
        print(f'Epoch {epoch+1}/{num_epochs}')
        print('-' * 50)
        
        # 训练阶段
        model.train()
        running_loss = 0.0
        running_corrects = 0
        
        for inputs, labels in train_loader:
            inputs = inputs.to(device)
            labels = labels.to(device)
            
            # 清零梯度
            optimizer.zero_grad()
            
            # 前向传播
            outputs = model(inputs)
            _, preds = torch.max(outputs, 1)
            loss = criterion(outputs, labels)
            
            # 反向传播和优化
            loss.backward()
            optimizer.step()
            
            # 统计
            running_loss += loss.item() * inputs.size(0)
            running_corrects += torch.sum(preds == labels.data)
        
        # 计算训练集损失和准确率
        epoch_loss = running_loss / len(train_loader.dataset)
        epoch_acc = running_corrects.double() / len(train_loader.dataset)
        
        history['train_loss'].append(epoch_loss)
        history['train_acc'].append(epoch_acc.item())
        
        print(f'Train Loss: {epoch_loss:.4f} Acc: {epoch_acc:.4f}')
        
        # 验证阶段
        model.eval()
        running_loss = 0.0
        running_corrects = 0
        
        with torch.no_grad():
            for inputs, labels in val_loader:
                inputs = inputs.to(device)
                labels = labels.to(device)
                
                # 前向传播
                outputs = model(inputs)
                _, preds = torch.max(outputs, 1)
                loss = criterion(outputs, labels)
                
                # 统计
                running_loss += loss.item() * inputs.size(0)
                running_corrects += torch.sum(preds == labels.data)
        
        # 计算验证集损失和准确率
        epoch_loss = running_loss / len(val_loader.dataset)
        epoch_acc = running_corrects.double() / len(val_loader.dataset)
        
        history['val_loss'].append(epoch_loss)
        history['val_acc'].append(epoch_acc.item())
        
        print(f'Val Loss: {epoch_loss:.4f} Acc: {epoch_acc:.4f}')
        print()
        
        # 保存最佳模型
        if epoch_acc > best_acc:
            best_acc = epoch_acc
            best_model_weights = copy.deepcopy(model.state_dict())
    
    # 加载最佳模型权重
    model.load_state_dict(best_model_weights)
    return model, history

# 绘制训练过程
def plot_history(history, title):
    plt.figure(figsize=(12, 4))
    
    # 绘制损失曲线
    plt.subplot(1, 2, 1)
    plt.plot(history['train_loss'], label='Train Loss')
    plt.plot(history['val_loss'], label='Val Loss')
    plt.title(f'{title} - Loss')
    plt.xlabel('Epoch')
    plt.ylabel('Loss')
    plt.legend()
    
    # 绘制准确率曲线
    plt.subplot(1, 2, 2)
    plt.plot(history['train_acc'], label='Train Acc')
    plt.plot(history['val_acc'], label='Val Acc')
    plt.title(f'{title} - Accuracy')
    plt.xlabel('Epoch')
    plt.ylabel('Accuracy')
    plt.legend()
    
    plt.tight_layout()
    plt.savefig(f'{title}_history.svg')
    plt.close()

# 主函数
def main():
    # 设置随机种子，保证结果可复现
    np.random.seed(42)
    torch.manual_seed(42)
    
    # 生成模拟数据（10000个样本，50个特征，10个类别）
    X = np.random.randn(10000, 50)
    y = np.random.randint(0, 10, size=10000)
    
    # 划分训练集和测试集
    X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42)
    
    # 数据标准化
    scaler = StandardScaler()
    X_train = scaler.fit_transform(X_train)
    X_val = scaler.transform(X_val)
    
    # 创建数据集和数据加载器
    # 无隐私保护的数据集
    train_dataset_no_dp = CustomDataset(X_train, y_train, apply_dp=False)
    val_dataset_no_dp = CustomDataset(X_val, y_val, apply_dp=False)
    
    # 有差分隐私保护的数据集（不同隐私预算）
    train_dataset_dp_1 = CustomDataset(X_train, y_train, apply_dp=True, dp_epsilon=1.0)
    val_dataset_dp_1 = CustomDataset(X_val, y_val, apply_dp=True, dp_epsilon=1.0)
    
    train_dataset_dp_5 = CustomDataset(X_train, y_train, apply_dp=True, dp_epsilon=5.0)
    val_dataset_dp_5 = CustomDataset(X_val, y_val, apply_dp=True, dp_epsilon=5.0)
    
    # 创建数据加载器
    batch_size = 64
    train_loader_no_dp = DataLoader(train_dataset_no_dp, batch_size=batch_size, shuffle=True)
    val_loader_no_dp = DataLoader(val_dataset_no_dp, batch_size=batch_size, shuffle=False)
    
    train_loader_dp_1 = DataLoader(train_dataset_dp_1, batch_size=batch_size, shuffle=True)
    val_loader_dp_1 = DataLoader(val_dataset_dp_1, batch_size=batch_size, shuffle=False)
    
    train_loader_dp_5 = DataLoader(train_dataset_dp_5, batch_size=batch_size, shuffle=True)
    val_loader_dp_5 = DataLoader(val_dataset_dp_5, batch_size=batch_size, shuffle=False)
    
    # 定义模型、损失函数和优化器
    input_dim = X_train.shape[1]
    hidden_dim = 128
    output_dim = 10
    
    # 训练无隐私保护的模型
    print("\n训练无隐私保护的模型...")
    model_no_dp = SimpleClassifier(input_dim, hidden_dim, output_dim)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model_no_dp.parameters(), lr=0.001)
    
    start_time = time.time()
    model_no_dp, history_no_dp = train_model(
        model_no_dp, train_loader_no_dp, val_loader_no_dp, criterion, optimizer, num_epochs=15
    )
    time_no_dp = time.time() - start_time
    
    # 训练有差分隐私保护的模型（epsilon=1.0）
    print("\n训练有差分隐私保护的模型（epsilon=1.0）...")
    model_dp_1 = SimpleClassifier(input_dim, hidden_dim, output_dim)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model_dp_1.parameters(), lr=0.001)
    
    start_time = time.time()
    model_dp_1, history_dp_1 = train_model(
        model_dp_1, train_loader_dp_1, val_loader_dp_1, criterion, optimizer, num_epochs=15
    )
    time_dp_1 = time.time() - start_time
    
    # 训练有差分隐私保护的模型（epsilon=5.0）
    print("\n训练有差分隐私保护的模型（epsilon=5.0）...")
    model_dp_5 = SimpleClassifier(input_dim, hidden_dim, output_dim)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model_dp_5.parameters(), lr=0.001)
    
    start_time = time.time()
    model_dp_5, history_dp_5 = train_model(
        model_dp_5, train_loader_dp_5, val_loader_dp_5, criterion, optimizer, num_epochs=15
    )
    time_dp_5 = time.time() - start_time
    
    # 绘制训练历史
    plot_history(history_no_dp, 'No Privacy Protection')
    plot_history(history_dp_1, 'Differential Privacy (epsilon=1.0)')
    plot_history(history_dp_5, 'Differential Privacy (epsilon=5.0)')
    
    # 输出结果比较
    print("\n模型性能比较：")
    print(f"无隐私保护模型: 验证准确率 = {max(history_no_dp['val_acc']):.4f}, 训练时间 = {time_no_dp:.2f}秒")
    print(f"差分隐私模型 (epsilon=1.0): 验证准确率 = {max(history_dp_1['val_acc']):.4f}, 训练时间 = {time_dp_1:.2f}秒")
    print(f"差分隐私模型 (epsilon=5.0): 验证准确率 = {max(history_dp_5['val_acc']):.4f}, 训练时间 = {time_dp_5:.2f}秒")
    
    print("\n结论：差分隐私保护会略微降低模型性能，但可以有效保护数据隐私。epsilon值越大，隐私保护强度越低，模型性能越接近无隐私保护模型。")

if __name__ == '__main__':
    main()