⚖️ 03 - 联邦平均算法FedAvg

阶段：联邦学习基础 第3课

📖 联邦平均算法

FedAvg是联邦学习最经典的算法，由McMahan等人于2017年提出。其核心公式为加权平均：wₜ₊₁ = Σₖ(nₖ/n)·wₖᵗ⁺¹。关键超参数包括采样率C、本地epoch数E和批大小B。FedAvg在IID数据上表现良好，但在Non-IID场景下可能出现客户端漂移。收敛上界为O(1/(TCK)) + O(E²σ²) + O(Eζ²)。

本课将深入探讨FedAvg算法的核心原理和实现方法，包括算法推导、Python实现和隐私分析。通过完整的代码实验，你将掌握该主题的关键技术和实践方法。

理解本课内容对于构建完整的联邦学习系统至关重要。FedAvg算法是联邦学习技术栈中的核心组件，掌握其原理和实现是进阶学习的基础。

核心原理

关键概念

技术细节

性能分析

应用场景

与其他技术的关系

🐍 Python实现

np.random.seed(42)
n_clients, n_features = 10, 20
clients = []
for k in range(n_clients):
    shift = k * 0.3
    X = np.random.randn(200, n_features) + shift
    tw = np.random.randn(n_features)
    y = (1.0/(1.0+np.exp(-(X @ tw + shift))) > 0.5).astype(int)
    clients.append({'X': X, 'y': y, 'n': 200})

class FedAvgServer:
    def __init__(self, n_features, lr=0.05):
        self.global_w = np.random.randn(n_features) * 0.01
        self.global_b = 0.0
        self.lr = lr
        self.history = []

    def sigmoid(self, z):
        return 1.0 / (1.0 + np.exp(-np.clip(z, -500, 500)))

    def select_clients(self, n_total, fraction=0.5):
        n = max(1, int(n_total * fraction))
        return np.random.choice(n_total, n, replace=False)

    def train(self, clients, n_rounds=30, C=0.5, local_epochs=5):
        import copy
        for r in range(n_rounds):
            selected = self.select_clients(len(clients), C)
            updates, counts = [], []
            for idx in selected:
                c = clients[idx]
                w, b = copy.deepcopy(self.global_w), self.global_b
                for _ in range(local_epochs):
                    pred = self.sigmoid(c['X'] @ w + b)
                    err = pred - c['y']
                    w -= self.lr * (c['X'].T @ err) / c['n']
                    b -= self.lr * np.mean(err)
                updates.append((w, b))
                counts.append(c['n'])
            total_n = sum(counts)
            self.global_w = sum(n/total_n*w for n,(w,_) in zip(counts, updates))
            self.global_b = sum(n/total_n*b for n,(_,b) in zip(counts, updates))
            self.history.append(self._acc(clients))
        return self.history

    def _acc(self, clients):
        total, correct = 0, 0
        for c in clients:
            pred = (self.sigmoid(c['X'] @ self.global_w + self.global_b) >= 0.5).astype(int)
            correct += np.sum(pred == c['y'])
            total += c['n']
        return correct / total

print("=== Sampling Rate ===")
for C in [0.2, 0.5, 1.0]:
    s = FedAvgServer(n_features, lr=0.05)
    print("  C=%.1f: Acc=%.4f" % (C, s.train(clients, 30, C=C)[-1]))

print("\n=== Local Epochs ===")
for E in [1, 3, 5, 10, 20]:
    s = FedAvgServer(n_features, lr=0.05)
    print("  E=%2d: Acc=%.4f" % (E, s.train(clients, 30, C=0.5, local_epochs=E)[-1]))
print("Verified - FedAvg experiments complete")

🔒 隐私分析

📝 练习

📚 扩展阅读

📐 数学基础

🔧 实现细节

🚀 进阶主题

📋 案例分析

📅 发展历程

📝 关键要点

📚 参考文献

❓ 常见问题