[AI学习笔记]隐私保护方案：DeepSeek 联邦学习落地细节

# Python 版本要求
python >= 3.6

# 必要的库安装
pip install tensorflow federated tensorflow-privacy ABY3

import tensorflow as tf
from tensorflow.keras.utils import to_categorical

def data_preprocessing(client_data_path):
    """
    数据预处理函数
    :param client_data_path: 客户端数据文件路径
    :return: 预处理后的数据
    """
    # 读取客户端数据
    client_data = pd.read_csv(client_data_path)
    
    # 数据清洗和特征工程
    # 假设数据包含标签列'label'
    X = client_data.drop(columns=['label']).values
    y = client_data['label'].values
    
    # 数据归一化
    X = (X - X.mean(axis=0)) / X.std(axis=0)
    
    # 标签one-hot编码
    y = to_categorical(y)
    
    return X, y

import tensorflow_federated as tff

@tff.tf_computation
def client_update(model_weights, client_data):
    """
    客户端模型更新函数
    :param model_weights: 服务器端模型权重
    :param client_data: 客户端预处理后的数据
    :return: 客户端更新后的模型权重和训练指标
    """
    # 构建模型
    model = tf.keras.Sequential([
        tf.keras.layers.Dense(128, activation='relu', input_shape=(client_data[0].shape[1],)),
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(client_data[1].shape[1], activation='softmax')
    ])
    
    # 加载服务器端模型权重
    model.set_weights(model_weights)
    
    # 模型编译
    model.compile(optimizer='adam', loss='categorical_crossentropy', metrics=['accuracy'])
    
    # 客户端本地训练
    history = model.fit(client_data[0], client_data[1], epochs=5, batch_size=32, verbose=0)
    
    return model.get_weights(), history.history['loss'][-1], history.history['accuracy'][-1]

@tff.federated_computation(tff.type_at_server(tf.float32))
def server_aggregation(server_state, client_outputs):
    """
    服务器端模型聚合函数
    :param server_state: 服务器端当前状态
    :param client_outputs: 客户端更新结果
    :return: 服务器端新的状态和全局模型评估指标
    """
    # 聚合客户端模型权重
    new_model_weights = tff.federated_mean(client_outputs[0])
    
    # 更新服务器端模型状态
    new_server_state = tff.federated_map(lambda s, w: s.update(w), (server_state, new_model_weights))
    
    # 计算全局模型评估指标
    global_loss = tff.federated_mean(client_outputs[1])
    global_accuracy = tff.federated_mean(client_outputs[2])
    
    return new_server_state, global_loss, global_accuracy

def federated_learning_process(client_data_paths, num_rounds=10):
    """
    联邦学习流程控制函数
    :param client_data_paths: 各客户端数据文件路径列表
    :param num_rounds: 联邦学习轮数
    :return: 最终全局模型和训练过程指标
    """
    # 初始化服务器端模型
    initial_model = create_initial_model()
    server_state = tff.federated_value(initial_model, tff.SERVER)
    
    # 联邦学习循环
    for round in range(num_rounds):
        # 客户端数据预处理
        client_data_list = []
        for path in client_data_paths:
            X, y = data_preprocessing(path)
            client_data_list.append((X, y))
        
        # 客户端更新
        client_outputs = tff.federated_map(client_update, (tff.federated_broadcast(server_state.model_weights, client_data_list), client_data_list))
        
        # 服务器端聚合
        server_state, global_loss, global_accuracy = server_aggregation(server_state, client_outputs)
        
        print(f"Round {round+1}: Global Loss = {global_loss}, Global Accuracy = {global_accuracy}")
    
    return server_state.model_weights, global_loss, global_accuracy