大模型应用：联邦学习融合本地大模型：隐私合规推荐的核心流程与实践.62

import torch
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from modelscope import AutoModelForCausalLM, AutoTokenizer
from sklearn.metrics import accuracy_score
from cryptography.fernet import Fernet

# 全局配置
DEVICE = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
MODEL_NAME = "qwen/Qwen1.5-1.8B-Chat"
CACHE_DIR = "D:\\modelscope\\hub"
NUM_CLIENTS = 5   # 模拟5家门店
NUM_ROUNDS = 5    # 联邦迭代轮次

# 1. 模拟门店本地数据（仅含标签，无敏感信息）
def generate_client_data():
    client_datasets = []
    for _ in range(NUM_CLIENTS):
        data = {
            "user_intent_label": np.random.randint(0, 8, 500),  # 8类用户意图
            "product_id": np.random.randint(1001, 1010, 500),   # 商品ID（9种）
            "feedback": np.random.randint(0, 2, 500)            # 推荐反馈（0/1）
        }
        client_datasets.append(pd.DataFrame(data))
    return client_datasets

# 2. Qwen1.5 初始化（INT4 量化节省显存）
def init_qwen_model():
    tokenizer = AutoTokenizer.from_pretrained(
        MODEL_NAME, cache_dir=CACHE_DIR, trust_remote_code=True
    )
    model = AutoModelForCausalLM.from_pretrained(
        MODEL_NAME,
        cache_dir=CACHE_DIR,
        trust_remote_code=True,
        torch_dtype=torch.float16,
        device_map=DEVICE,
        load_in_4bit=True  # 启用 INT4 量化
    )
    return tokenizer, model

# 3. 参数加密工具（模拟安全传输）
class ParameterEncryptor:
    def __init__(self):
        self.cipher = Fernet(Fernet.generate_key())
    
    def encrypt(self, params):
        """将参数列表扁平化为字节并加密"""
        flat_params = torch.cat([p.flatten().cpu() for p in params]).numpy()
        return self.cipher.encrypt(flat_params.tobytes())
    
    def decrypt(self, encrypted_params, param_shapes):
        """解密并按原始形状还原参数"""
        decrypted_bytes = self.cipher.decrypt(encrypted_params)
        decrypted_array = np.frombuffer(decrypted_bytes, dtype=np.float16)
        params = []
        idx = 0
        for shape in param_shapes:
            size = int(np.prod(shape))
            tensor = torch.tensor(decrypted_array[idx:idx+size], dtype=torch.float16)
            params.append(tensor.reshape(shape).to(DEVICE))
            idx += size
        return params

# 4. 门店本地训练（基于提示微调）
def client_train(local_data, model, tokenizer):
    model.train()
    optimizer = torch.optim.Adam(model.parameters(), lr=2e-5)
    
    for epoch in range(5):  # 本地训练5个epoch
        for i in range(0, len(local_data), 32):
            batch = local_data.iloc[i:i+32]
            # 构造提示文本
            prompts = [
                f"用户意图：{l}，商品ID：{p}，推荐是否满意？"
                for l, p in zip(batch["user_intent_label"], batch["product_id"])
            ]
            inputs = tokenizer(
                prompts, return_tensors="pt", padding=True,
                truncation=True, max_length=64
            ).to(DEVICE)
            
            # 构造目标标签（将 feedback 映射为 token）
            labels_text = ["满意" if f == 1 else "不满意" for f in batch["feedback"]]
            labels = tokenizer(labels_text, return_tensors="pt", padding=True).input_ids.to(DEVICE)
            
            # 前向传播（注意：labels 应为 decoder 输入，此处简化）
            outputs = model(**inputs, labels=labels)
            loss = outputs.loss
            
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
    
    # 评估本地模型
    val_acc = evaluate_model(model, tokenizer, local_data.sample(frac=0.2))
    return val_acc

# 5. 模型评估函数
def evaluate_model(model, tokenizer, val_data):
    model.eval()
    predictions, true_labels = [], []
    with torch.no_grad():
        for i in range(0, len(val_data), 32):
            batch = val_data.iloc[i:i+32]
            prompts = [
                f"用户意图：{l}，商品ID：{p}，推荐是否满意？"
                for l, p in zip(batch["user_intent_label"], batch["product_id"])
            ]
            inputs = tokenizer(prompts, return_tensors="pt", padding=True, truncation=True, max_length=64).to(DEVICE)
            
            # 生成预测
            generated = model.generate(**inputs, max_new_tokens=5)
            decoded = tokenizer.batch_decode(generated, skip_special_tokens=True)
            
            # 解析预测结果
            for text in decoded:
                pred = 1 if "满意" in text else 0
                predictions.append(pred)
            true_labels.extend(batch["feedback"].tolist())
    
    return accuracy_score(true_labels, predictions)

# 6. 总部参数聚合（联邦平均）
def server_aggregate(client_models, client_weights):
    """聚合多个客户端模型参数（FedAvg）"""
    aggregated_state_dict = {}
    total_weight = sum(client_weights)
    
    # 初始化聚合字典
    for key in client_models[0].state_dict().keys():
        aggregated_state_dict[key] = torch.zeros_like(client_models[0].state_dict()[key])
    
    # 加权平均
    for model, weight in zip(client_models, client_weights):
        w = weight / total_weight
        for key in aggregated_state_dict.keys():
            aggregated_state_dict[key] += w * model.state_dict()[key]
    
    return aggregated_state_dict

# 7. 主联邦学习流程
def federated_pipeline():
    client_datasets = generate_client_data()
    encryptor = ParameterEncryptor()
    global_tokenizer, global_model = init_qwen_model()
    local_accs = [[] for _ in range(NUM_CLIENTS)]
    federated_accs = []
    client_weights = [len(data) for data in client_datasets]  # 按数据量加权

    for round_idx in range(NUM_ROUNDS):
        print(f"=== 第 {round_idx + 1} 轮联邦优化 ===")
        client_models = []
        
        # 各门店本地训练
        for cid in range(NUM_CLIENTS):
            # 深拷贝全局模型到本地
            local_model = AutoModelForCausalLM.from_pretrained(
                MODEL_NAME, cache_dir=CACHE_DIR, trust_remote_code=True,
                torch_dtype=torch.float16, device_map=DEVICE, load_in_4bit=True
            )
            local_model.load_state_dict(global_model.state_dict())
            
            acc = client_train(client_datasets[cid], local_model, global_tokenizer)
            local_accs[cid].append(acc)
            client_models.append(local_model)
            print(f"门店 {cid + 1}：本地准确率 = {acc:.4f}")
        
        # 服务器聚合（模拟加密传输：本地加密 → 服务器解密 → 聚合）
        # 此处为简化，直接聚合；实际可加入 encryptor.encrypt/decrypt
        aggregated_state = server_aggregate(client_models, client_weights)
        global_model.load_state_dict(aggregated_state)
        
        # 评估联邦模型（在所有客户端数据上）
        fed_acc = np.mean([
            evaluate_model(global_model, global_tokenizer, d.sample(frac=0.2))
            for d in client_datasets
        ])
        federated_accs.append(fed_acc)
        print(f"本轮联邦模型平均准确率：{fed_acc:.4f}\n")
    
    # 可视化结果
    plot_comparison(local_accs, federated_accs)

# 8. 效果对比可视化
def plot_comparison(local_accs, federated_accs):
    plt.rcParams['font.sans-serif'] = ['SimHei']  # 支持中文
    plt.figure(figsize=(12, 6))
    rounds = list(range(1, NUM_ROUNDS + 1))
    
    # 绘制各门店本地模型准确率（虚线）
    for i, accs in enumerate(local_accs):
        plt.plot(rounds, accs, "--o", label=f"门店{i+1} 本地模型")
    
    # 绘制联邦模型准确率（实线，红色）
    plt.plot(rounds, federated_accs, "-s", linewidth=2, color="red", label="联邦优化模型（平均）")
    
    plt.xlabel("联邦迭代轮次")
    plt.ylabel("推荐准确率")
    plt.title("联邦学习 vs 单节点本地模型效果对比")
    plt.legend()
    plt.grid(alpha=0.3)
    plt.savefig("federated_comparison.png", dpi=300, bbox_inches="tight")
    plt.show()

# 启动主流程
if __name__ == "__main__":
    federated_pipeline()