En este tutorial, exploramos cómo se comporta el aprendizaje federado cuando el servidor de agregación centralizado tradicional se elimina y se reemplaza con un mecanismo de chismes de igual a igual totalmente descentralizado. Implementamos FedAvg centralizado y Gossip Federated Learning descentralizado desde cero e introducimos privacidad diferencial del lado del cliente inyectando ruido calibrado en las actualizaciones del modelo local. Al realizar experimentos controlados con datos MNIST que no son IID, examinamos cómo la fortaleza de la privacidad, medida por diferentes valores épsilon, afecta directamente la velocidad de convergencia, la estabilidad y la precisión del modelo final. Además, estudiamos las compensaciones prácticas entre las garantías de privacidad y la eficiencia del aprendizaje en sistemas de aprendizaje descentralizados del mundo real. Consulte los códigos completos aquí.
importar sistema operativo, matemáticas, aleatorio, tiempo desde clases de datos importar clase de datos desde escribir importar Dict, Lista, Tupla importar subproceso, sys def pip_install(pkgs): subprocess.check_call([sys.executable, "-m", "pip", "install", "-q"] + pkgs) pip_install(["torch", "torchvision", "numpy", "matplotlib", "networkx", "tqdm"]) importar numpy como np importar antorcha importar antorcha.nn como nn importar antorcha.nn.funcional como F desde torch.utils.data importar DataLoader, subconjunto de torchvision importar conjuntos de datos, transforma importar matplotlib.pyplot como plt importar networkx como nx desde tqdm importar trange SEED = 7 random.seed(SEED) np.random.seed(SEED) torch.manual_seed(SEED) torch.cuda.manual_seed_all(SEED) torch.backends.cudnn.deterministic = False torch.backends.cudnn.benchmark = True transform = transforms.Compose([transforms.ToTensor()]) train_ds = datasets.MNIST(root="/content/data", train=True, download=True, transform=transform) test_ds = conjuntos de datos.MNIST(root="/content/data", train=False, download=True, transform=transform)
Configuramos el entorno de ejecución e instalamos todas las dependencias necesarias. Inicializamos semillas aleatorias y configuraciones de dispositivos para mantener la reproducibilidad en todos los experimentos. También cargamos el conjunto de datos MNIST, que sirve como punto de referencia ligero pero eficaz para experimentos de aprendizaje federados. Consulte los códigos completos aquí.
def make_noniid_clients(conjunto de datos, num_clients=20, shards_per_client=2, semilla=SEED): rng = np.random.default_rng(semilla) y = np.array([conjunto de datos[i][1]para i en rango(len(conjunto de datos))]) idx = np.arange(len(conjunto de datos)) idx_sorted = idx[np.argsort(y)] num_shards = num_clients * shards_per_client shard_size = len(dataset) // num_shards shards = [idx_sorted[i*shard_size:(i+1)*shard_size] para i en el rango(num_shards)] rng.shuffle(shards) client_indices =[]para c en rango(num_clientes): take = shards[c*shards_per_client:(c+1)*shards_per_client] client_indices.append(np.concatenate(take)) return client_indices NUM_CLIENTES = 20 client_indices = make_noniid_clients(train_ds, num_clients=NUM_CLIENTES, shards_per_client=2) test_loader = DataLoader(test_ds, lote_size=1024, shuffle=False, num_workers=2, pin_memory=True) clase MLP(nn.Module): def __init__(self): super().__init__() self.fc1 = nn.Linear(28*28, 256) self.fc2 = nn.Linear(256, 128) self.fc3 = nn.Linear(128, 10) def adelante(self, x): x = x.view(x.size(0), -1) x = F.relu(self.fc1(x)) x = F.relu(self.fc2(x)) return self.fc3(x)
Construimos una distribución de datos que no es IID dividiendo el conjunto de datos de entrenamiento en fragmentos basados en etiquetas en varios clientes. Definimos un modelo de red neuronal compacto que equilibra la expresividad y la eficiencia computacional. Nos permite simular de manera realista la heterogeneidad de los datos, un desafío crítico en los sistemas de aprendizaje federados. Consulte los códigos completos aquí.
def get_model_params(modelo): regresa {k: v.detach().clone() para k, v en model.state_dict().items()} def set_model_params(modelo, params): model.load_state_dict(params, estricto=True) def add_params(a, b): regresa {k: a[k] + b[k] para k en a.keys()} def sub_params(a, b): regresa {k: a[k] – b[k] para k en a.keys()} def scale_params(a, s): regresa {k: a[k] * s para k en a.keys()} def mean_params(params_list): out = {k: torch.zeros_like(params_list)[0][k]) para k en params_list[0].keys()} para p en params_list: para k en out.keys(): out[k] += p[k] para k en out.keys(): out[k] /= len(params_list) devuelve def l2_norm_params(delta): sq = 0.0 para v en delta.values(): sq += float(torch.sum(v.float() * v.float()).item()) devuelve math.sqrt(sq) def dp_sanitize_update(delta, clip_norm, epsilon, delta_dp, rng): norma = l2_norm_params(delta) escala = min(1.0, clip_norm / (norma + 1e-12)) recortado = scale_params(delta, escala) si épsilon es Ninguno o math.isinf(epsilon) o épsilon <= 0: devuelve sigma recortado = clip_norm * math.sqrt(2.0 * math.log(1.25 / delta_dp)) / épsilon ruidoso = {} para k, v en clipped.items(): ruido = torch.normal(media=0.0, std=sigma, tamaño=v.shape, generador=rng, dispositivo=v.device, dtype=v.dtype) ruidoso[k] = v + retorno de ruido ruidoso
Implementamos utilidades de manipulación de parámetros que permiten sumar, restar, escalar y promediar pesos de modelos entre clientes. Introducimos privacidad diferencial recortando actualizaciones locales e inyectando ruido gaussiano, ambos determinados por el presupuesto de privacidad elegido. Sirve como el mecanismo central de privacidad que nos permite estudiar el equilibrio entre privacidad y utilidad tanto en entornos centralizados como descentralizados. Consulte los códigos completos aquí.
def local_train_one_client(base_params, client_id, epochs, lr, lote_size, Weight_decay=0.0): modelo = MLP().to(dispositivo) set_model_params(modelo, base_params) model.train() cargador = DataLoader( Subset(train_ds, client_indices[client_id].tolist() if hasattr(client_indices[client_id], "tolist") else client_indices[client_id]), lote_size=batch_size, shuffle=True, num_workers=2, pin_memory=True) opt = torch.optim.SGD(model.parameters(), lr=lr, momentum=0.9, Weight_decay=weight_decay) para _ en rango (épocas): para xb, yb en el cargador: xb, yb = xb.to(dispositivo), yb.to(dispositivo) opt.zero_grad(set_to_none=True) logits = model(xb) loss = F.cross_entropy(logits, yb) loss.backward() opt.step() return get_model_params(modelo) @torch.no_grad() def evaluar(params): modelo = MLP().to(dispositivo) set_model_params(modelo, params) model.eval() total, correcto = 0, 0 loss_sum = 0.0 para xb, yb en test_loader: xb, yb = xb.to(dispositivo), yb.to(dispositivo) logits = model(xb) pérdida = F.cross_entropy(logits, yb, reducción="suma") loss_sum += float(loss.item()) pred = torch.argmax(logits, dim=1) correcto += int((pred == yb).sum().item()) total += int(yb.numel()) return loss_sum / total, correcto / total
Definimos el bucle de entrenamiento local que cada cliente ejecuta de forma independiente sobre sus datos privados. También implementamos una rutina de evaluación unificada para medir la pérdida y la precisión de las pruebas para cualquier estado del modelo. Juntas, estas funciones simulan un comportamiento de aprendizaje federado realista donde la capacitación y la evaluación están completamente desvinculadas de la propiedad de los datos. Consulte los códigos completos aquí.
@dataclass clase FedAvgConfig: rondas: int = 25 clientes_por_ronda: int = 10 épocas_locales: int = 1 lr: float = 0.06 tamaño_lote: int = 64 clip_norm: float = 2.0 épsilon: float = math.inf delta_dp: float = 1e-5 def run_fedavg(cfg): global_params = get_model_params(MLP().to(dispositivo)) historial = {"test_loss":[], "prueba_acc":[]} para r en trange(cfg.rounds): elegido = random.sample(range(NUM_CLIENTS), k=cfg.clients_per_round) start_params = global_params actualizaciones =[]para cid elegido: local_params = local_train_one_client(start_params, cid, cfg.local_epochs, cfg.lr, cfg.batch_size) delta = sub_params(local_params, start_params) rng = torch.Generator(dispositivo=dispositivo) rng.manual_seed(SEED * 10000 + r * 100 + cid) delta_dp = dp_sanitize_update(delta, cfg.clip_norm, cfg.epsilon, cfg.delta_dp, rng) actualizaciones.append(delta_dp) avg_update = mean_params(actualizaciones) global_params = add_params(start_params, avg_update) tl, ta = evaluar(global_params) historial["test_loss"].append(tl) historial["test_acc"].append(ta) historial de retorno, global_params
Implementamos el algoritmo centralizado FedAvg, donde un subconjunto de clientes se entrena localmente y envía actualizaciones privadas diferenciales a un agregador central. Realizamos un seguimiento del rendimiento del modelo en las rondas de comunicación para observar el comportamiento de convergencia bajo diferentes presupuestos de privacidad. Esto sirve como punto de referencia con el que se compara el aprendizaje descentralizado basado en chismes. Consulte los códigos completos aquí.
@dataclass clase GossipConfig: rondas: int = 25 local_epochs: int = 1 lr: float = 0.06 tamaño_lote: int = 64 clip_norm: float = 2.0 épsilon: float = math.inf delta_dp: float = 1e-5 topología: str = "ring" p: float = 0.2 gossip_pairs_per_round: int = 10 def build_topology(cfg): if cfg.topology == "ring": G = nx.cycle_graph(NUM_CLIENTS) elif cfg.topology == "erdos_renyi": G = nx.erdos_renyi_graph(NUM_CLIENTS, cfg.p, seed=SEED) si no nx.is_connected(G): comps = list(nx.connected_components(G)) para i en el rango(len(comps) – 1): a = next(iter(comps[i])) b = next(iter(comps[i+1])) G.add_edge(a, b) else: elevar ValueError return G def run_gossip(cfg): node_params = [get_model_params(MLP().to(device)) para _ en el rango (NUM_CLIENTES)] G = build_topology(cfg) historial = {"avg_test_loss":[], "avg_test_acc":[]} para r en extraño(cfg.rounds): new_params =[]para cid en el rango (NUM_CLIENTES): p0 = node_params[cid] p_local = local_train_one_client(p0, cid, cfg.local_epochs, cfg.lr, cfg.batch_size) delta = sub_params(p_local, p0) rng = torch.Generator(dispositivo=dispositivo) rng.manual_seed(SEED * 10000 + r * 100 + cid) delta_dp = dp_sanitize_update(delta, cfg.clip_norm, cfg.epsilon, cfg.delta_dp, rng) p_local_dp = add_params(p0, delta_dp) new_params.append(p_local_dp) node_params = new_params bordes = lista(G.edges()) para _ en rango(cfg.gossip_pairs_per_round): i, j = random.choice(edges) avg = mean_params([node_params[i], node_params[j]]) node_params[i] = promedio node_params[j] = pérdidas promedio, accs =[],[]para cid en rango (NUM_CLIENTES): tl, ta = evaluar(node_params[cid]) pérdidas.append(tl) accs.append(ta) historial["avg_test_loss"].append(float(np.mean(losses))) historial["avg_test_acc"].append(float(np.mean(accs))) historial de retorno, nodo_params
Implementamos el aprendizaje federado de Gossip descentralizado utilizando un modelo de igual a igual que intercambia a través de una topología de red predefinida. Simulamos un entrenamiento local repetido y un promedio de parámetros por pares sin depender de un servidor central. Nos permite analizar cómo el ruido de la privacidad se propaga a través de patrones de comunicación descentralizados y afecta la convergencia. Consulte los códigos completos aquí.
eps_sweep = [math.inf, 8.0, 4.0, 2.0, 1.0] ROUNDS = 20 fedavg_results = {} gossip_results = {} common_local_epochs = 1 common_lr = 0.06 common_bs = 64 common_clip = 2.0 common_delta = 1e-5 para eps en eps_sweep: fcfg = FedAvgConfig( rondas=RONDAS, clientes_por_ronda=10, local_epochs=common_local_epochs, lr=common_lr, lote_size=common_bs, clip_norm=common_clip, epsilon=eps, delta_dp=common_delta ) hist_f, _ = run_fedavg(fcfg) fedavg_results[eps] = hist_f gcfg = GossipConfig( rondas=RONDAS, local_epochs=common_local_epochs, lr=common_lr, lote_size=common_bs, clip_norm=common_clip, epsilon=eps, delta_dp=common_delta, topology="ring", gossip_pairs_per_round=10 ) hist_g, _ = run_gossip(gcfg) gossip_results[eps] = hist_g plt.figure(figsize=(10, 5)) para eps en eps_sweep: plt.plot(fedavg_results[eps]["test_acc"], label=f"FedAvg eps={eps}") plt.xlabel("Ronda") plt.ylabel("Precisión") plt.legend() plt.grid(True) plt.show() plt.figure(figsize=(10, 5)) para eps en eps_sweep: plt.plot(gossip_results[eps]["avg_test_acc"], label=f"Gossip eps={eps}") plt.xlabel("Round") plt.ylabel("Precisión promedio") plt.legend() plt.grid(True) plt.show() final_fed = [fedavg_results[eps]["test_acc"][-1] para eps en eps_sweep] final_gos = [gossip_results[eps]["avg_test_acc"][-1] para eps en eps_sweep] x = [100.0 if math.isinf(eps) else eps para eps en eps_sweep] plt.figure(figsize=(8, 5)) plt.plot(x, final_fed, marcador="o", label="FedAvg") plt.plot(x, final_gos, marcador="o", label="Gossip") plt.xlabel("Epsilon") plt.ylabel("Precisión final") plt.legend() plt.grid(True) plt.show() def rounds_to_threshold(acc_curve, umbral): para i, a en enumerar(acc_curve): si a >= umbral: devolver i + 1 devolver Ninguno best_f = fedavg_results[math.inf]["test_acc"][-1] best_g = gossip_results[math.inf]["avg_test_acc"][-1] th_f = 0.9 * best_f th_g = 0.9 * best_g para eps en eps_sweep: rf = rounds_to_threshold(fedavg_results[eps]["test_acc"], th_f) rg = rounds_to_threshold(gossip_results[eps]["avg_test_acc"], th_g) print(eps, rf, rg)
Realizamos experimentos controlados en múltiples niveles de privacidad y recopilamos resultados para estrategias de capacitación tanto centralizadas como descentralizadas. Visualizamos las tendencias de convergencia y la precisión final para exponer claramente el equilibrio entre privacidad y utilidad. También calculamos métricas de velocidad de convergencia para comparar cuantitativamente cómo responden los diferentes esquemas de agregación a las crecientes restricciones de privacidad.
En conclusión, demostramos que la descentralización cambia fundamentalmente cómo se propaga el ruido diferencial de la privacidad a través de un sistema federado. Observamos que, si bien el FedAvg centralizado normalmente converge más rápido bajo restricciones de privacidad débiles, el aprendizaje federado basado en chismes es más resistente a las actualizaciones ruidosas a costa de una convergencia más lenta. Nuestros experimentos resaltaron que una mayor privacidad garantiza un aprendizaje significativamente lento en ambos entornos, pero el efecto se amplifica en topologías descentralizadas debido al retraso en la mezcla de información. En general, demostramos que el diseño de sistemas federados que preservan la privacidad requiere un razonamiento conjunto sobre la topología de agregación, los patrones de comunicación y los presupuestos de privacidad en lugar de tratarlos como opciones independientes.
Consulte los códigos completos aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
