代码收藏家 Pysyft学习笔记
文章目录
以下笔记内容源于官方文档，记录的是我认为比较重要的地方
先做一下import：
import sys
import torch
from torch.nn import Parameter
import torch.nn as nn
import syft as sy
hook = sy.TorchHook(torch)
1 隐私、分布式数据科学的基础工具
1.1 张量指针
张量的指针：
bob = sy.VirtualWorker(hook, id="bob")
x = torch.tensor([1,2,3,4,5])
y = torch.tensor([1,1,1,1,1])
x_ptr = x.send(bob)
y_ptr = y.send(bob)
print(x_ptr)
以上代码输出：
(Wrapper)>[PointerTensor | me:71478333721 -> bob:43597191961]
可看出，x_ptr是一个张量指针，它将x这个张量发送给了bob
这里我们发送了两个张量给bob，让我们来看看bob是不是已经含有了这两个张量：
print(bob._objects)
输出：
{26833634777: tensor([1, 2, 3, 4, 5]), 28924365874: tensor([1, 1, 1, 1, 1])}
看来确实是有两个张量。
以上，bob相当于是个我们创建出的虚拟机，是一个VirtualWoker对象，以上的一系列操作相当于是将我们本机的张量发送给另一台机器。
另外，注意一下。当我们调用 x.send(bob)时，它返回了一个称为x_ptr的新对象。这是我们第一个指向张量的指针。张量的指针本身实际上并不保存数据。相反，它们仅包含有关存储在另一台机器上的张量（带有数据）的元数据。这些张量的目的是为我们提供一个直观的API，以告诉其他机器使用该张量来计算函数。
x_ptr指针有两个主要属性：
x_ptr.location : bob, location(位置)，对指针指向的位置的引用x_ptr.id_at_location : <random integer>, 张量存储在所在位置的id，随机分配它们以以下格式打印： <id_at_location>@<location>
还有其他更通用的属性：
x_ptr.id : <random integer>, 指针张量的ID，它是随机分配的x_ptr.owner : "me", 拥有指针张量的工作机，这里是本地机器，名为“me”（我）最后，若要bob返还指针，则使用如下代码：
x_ptr.get()
y_ptr.get()
这样，再打印一下bob._objects会发现不含张量了。
1.2 使用张量指针
使用张量指针进行异地计算的范例：
我们希望能够对远程张量执行张量操作。幸运的是，张量指针使这变得很容易！您可以像使用普通张量一样使用指针！
x = torch.tensor([1,2,3,4,5]).send(bob)
y = torch.tensor([1,1,1,1,1]).send(bob)
z = x + y
print(z)
瞧！
在背后，发生了非常有力的事情。不再是x和y在本地计算加法，而是将命令序列化并发送给Bob，由后者执行计算，创建张量z，然后将指向z的指针返回给我们！
如果我们在指针上调用.get()，那么我们将把结果返回到我们的机器上！
z.get()
这个API已经扩展到Torch的所有操作！
对于之前范例中的x、y【注意这里的x和y为张量指针】：
z = torch.add(x, y)
print(z)
输出：
(Wrapper)>[PointerTensor | me:25024527595 -> bob:5999607250]
返还z到我们的本机上，打印出相加后的结果：
z_local = z.get()
print(z_local)
输出：
tensor([2, 3, 4, 5, 6])
变量（包括反向传播）
看一个范例即可：
# x和y必须是float类型数据，因为只有该类型数据才能够进行梯度下降
x = torch.tensor([1,2,3,4,5.], requires_grad=True).send(bob)
y = torch.tensor([1,1,1,1,1.], requires_grad=True).send(bob)
z = torch.add(x, y).sum() # 类似于损失函数
z.backward() # 反向传播
x_local = x.get()
print(x_local)
print(x_local.grad)
输出：
tensor([1., 2., 3., 4., 5.], requires_grad=True)
tensor([1., 1., 1., 1., 1.])
2 联邦学习简介
在上一节中，我们了解了张量指针，它创建了隐私保护深度学习所需的基础架构。在本节中，我们将看到如何使用这些基本工具来实现我们的第一个隐私保护深度学习算法：联邦学习。
什么是联邦学习？
它是训练深度学习模型的一种简单而强大的方法。考虑一下训练数据，一般它总是某种收集过程的结果：人们（通过设备）通过记录现实世界中的事件来生成数据。通常，此数据被聚合到单个中央位置，以便您可以训练机器学习模型。而联邦学习扭转了这一局面！
你无需将训练数据带到模型（一个中央服务器），而是将模型带到训练数据（无论其位于何处）。
这个想法允许创建数据的任何人拥有数据唯一的永久副本，从而保持对有权访问该数据的人的控制。
2.1 一个联邦学习的范例
让我们从一个集中式训练的模型开始。就像得到模型一样简单。我们首先需要：
ok，这个例子的demo如下：
import torch
from torch import nn
from torch import optim
data = torch.tensor([[0,0],[0,1],[1,0],[1,1.]], requires_grad=True)
target = torch.tensor([[0],[0],[1],[1.]], requires_grad=True)
model = nn.Linear(2, 1)
def train():
opt = optim.SGD(params=model.parameters(), lr=0.1)
for iter in range(20):
# 1) 消除之前的梯度（如果存在）
opt.zero_grad()
# 2) 预测
pred = model(data)
# 3) 计算损失
loss = ((pred - target)**2).sum()
# 4) 指出那些导致损失的参数（损失回传）
loss.backward()
# 5) 更新参数
opt.step()
# 6) 打印进程
print(loss.data)
if __name__ == "__main__":
train()
以上这种训练方式，就是常规方式，我们所有的数据都汇总到我们的本地计算机中，我们可以使用它来更新我们的模型。但是，联邦学习无法以这种方式工作。 因此，让我们修改此范例以实现联邦学习方式！来看一个简单的demo：
import torch
import torch.nn as nn
import torch.optim as optim
import syft as sy
hook = sy.TorchHook(torch)
# 创建一对虚拟工作机
bob = sy.VirtualWorker(hook, id='bob')
alice = sy.VirtualWorker(hook, id='alice')
# 一个数据集以及对应的标签
data = torch.tensor([[0, 0], [0, 1], [1, 0], [1, 1.]], requires_grad=True)
target = torch.tensor([[0], [0], [1], [1.]], requires_grad=True)
# 通过以下方式获取每个工作机的训练数据的指针
# 向bob和alice发送一些训练数据
data_bob = data[0: 2]
target_bob = target[0: 2]
data_alice = data[2:]
target_alice = target[2:]
# 初始化模型
model = nn.Linear(2, 1)
# 获取数据指针和标签指针
data_bob = data_bob.send(bob)
data_alice = data_alice.send(alice)
target_bob = target_bob.send(bob)
target_alice = target_alice.send(alice)
# 将指针组织到列表中
datasets = [(data_bob, target_bob), (data_alice, target_alice)] # 相当于拥有两份数据集了
def train():
opt = optim.SGD(params=model.parameters(), lr=0.1)
for iter in range(10):
# 1) 遍历每个工作机的数据集
for data, target in datasets:
# 2) 将模型发送给对应的工作机
model.send(data.location) # data.location返回其所在的工作机id
# 3) 消除之前的梯度（如果存在）
opt.zero_grad()
# 4) 预测
pred = model(data)
# 5) 计算损失
loss = ((pred - target)**2).sum()
# 6) 指出那些导致损失的参数（损失回传）
loss.backward()
# 7) 更新参数
opt.step()
# 8) 获取模型（带梯度）:将工作机上的model返还至主机上
model.get()
# 6) 打印进程：此时loss为张量指针，所以要先返还到本机上再打印
print(loss.get().data)
if __name__ == "__main__":
train()
输出：
tensor(0.2189)
tensor(1.4988)
tensor(0.3456)
tensor(0.1696)
tensor(0.2707)
tensor(0.0996)
tensor(0.1882)
tensor(0.0702)
tensor(0.1312)
tensor(0.0506)
tensor(0.0922)
tensor(0.0368)
tensor(0.0654)
tensor(0.0270)
tensor(0.0468)
tensor(0.0200)
tensor(0.0336)
tensor(0.0148)
tensor(0.0243)
tensor(0.0110)
以上，我们将模型发送给每个工作机，生成新的梯度，然后将梯度带回我们的本地服务器，以此更新全局模型。在此过程中，我们永远不会看到或请求访问基础训练数据！我们保留Bob和Alice的隐私！
但以上范例有着缺陷：
最值得注意的是，当我们调用model.get()并从Bob或Alice接收更新的模型时，我们实际上可以通过查看Bob和Alice的梯度来学习很多关于Bob和Alice的训练数据。在某些情况下，我们可以完美地恢复他们的训练数据！所以，这里算是泄露隐私了（译者注：此处属于隐私泄露攻击）
那么，该怎么办？好吧，人们采用的第一个策略是在将多个梯度上载到中央服务器之前对多个个体进行平均【在中央服务器上做聚合】。但是，此策略将需要对张量指针对象进行更复杂的使用。因此，在下一节中，我们将花费一些时间来学习更多高级指针功能，然后我们将升级此联邦学习示例。
3 高级远程执行工具
在上一节中，我们使用联邦学习思想训练了一个模型。 为此，我们在模型上调用了.send()和.get()，将其发送到训练数据的位置【各客户机上】，对其进行了更新，最后将其恢复【返还到了本地服务器上】。但是，在示例的最后，我们意识到我们需要进一步保护人们的隐私。也就是说，我们要在调用.get()之前对梯度进行平均。这样，我们将永远不会看到任何人的确切梯度（因此更好地保护了他们的隐私！！！）【也就是FedAvg算法思想】
但是，为了做到这一点，我们还需要：
此外，当我们在这里时，我们还将学习一些更高级的张量操作，这将有助于我们使用本示例以及将来的一些示例
3.1 指向指针的指针
PointerTensor（张量指针）对象的感觉就像普通的张量一样。实际上，它们非常类似于张量，甚至我们甚至可以拥有指向这些指针的指针。来看一个例子：
import torch
import syft as sy
hook = sy.TorchHook(torch)
bob = sy.VirtualWorker(hook, id='bob')
alice = sy.VirtualWorker(hook, id='alice')
x = torch.tensor([1, 2, 3, 4]) # 本地张量
x_ptr = x.send(bob) # 将本地张量发给bob，并返回张量指针
pointer_to_x_ptr = x_ptr.send(alice) # 将指针发给alice，返回指针的指针
print(pointer_to_x_ptr)
print(bob._objects)
print(alice._objects)
输出：
(Wrapper)>[PointerTensor | me:20770727071 -> alice:39671448516]
{47690627694: tensor([1, 2, 3, 4])}
{39671448516: (Wrapper)>[PointerTensor | alice:39671448516 -> bob:47690627694]}
在这个示例中我们创建了一个名为 x 的张量然后发送给了Bob，并在本地计算机创建了一个指针(x_ptr)。
然后我们调用了 x_ptr.send(alice) ，它将指针发送给Alice。
可以看到，alice._objects为一个指针，指向bob._objects。
注意，这里没有移动数据，而是将指针移动到了数据上。
那么使用pointer_to_x_ptr.get()就可以获取x_ptr，返回给Alice，x_ptr.get()获取x返回给bob。[.get()方法相当于获取指针所指的内容]
Pointer->Pointer->Data对象上的运算
就像普通的指针一样，我们可以在这些张量上执行任何Pytorch操作，例如：
p2p2x = torch.tensor([1, 2, 3, 4, 5]).send(bob).send(alice)
print(bob._objects)
print(alice._objects)
y = torch.add(p2p2x, p2p2x)
print(y.get().get())
print(bob._objects)
print(alice._objects)
输出：
{28982941060: tensor([1, 2, 3, 4, 5])}
{2660316055: (Wrapper)>[PointerTensor | alice:2660316055 -> bob:28982941060]}
tensor([ 2, 4, 6, 8, 10])
{28982941060: tensor([1, 2, 3, 4, 5])}
{2660316055: (Wrapper)>[PointerTensor | alice:2660316055 -> bob:28982941060], 26777481313: (Wrapper)>[PointerTensor | alice:38854603139 -> bob:42427576642]}
可以看到，在进行add操作后，alice工作机又出现了新的指针y
3.2 指针链操作
在上一节中，每当我们调用.send()或.get()操作时，它都会直接在本地计算机的张量上调用该操作。但是，如果您有一连串的指针，有时您希望在该链中“最后”的指针上调用.get()或.send()之类的操作（例如，将数据直接从一个工作程序发送到另一个工作程序）。为此，您想使用专门为此隐私保护操作设计的功能。
实现这个功能的操作，就要使用.move()方法，看下面这个例子：
import torch
import syft as sy
hook = sy.TorchHook(torch)
bob = sy.VirtualWorker(hook, id='bob')
alice = sy.VirtualWorker(hook, id='alice')
x = torch.tensor([1, 2, 3, 4]).send(bob)
print('bob:{}'.format(bob._objects))
print('alice:{}'.format(alice._objects))
x = x.move(alice)
print('bob:{}'.format(bob._objects))
print('alice:{}'.format(alice._objects))
print(x)
输出：
bob:{242612995: tensor([1, 2, 3, 4])}
alice:{}
bob:{}
alice:{242612995: tensor([1, 2, 3, 4])}
(Wrapper)>[PointerTensor | me:17438136219 -> alice:242612995]
这样，我们就实现了将数据从一个工作机，给转移到另一台工作机上。现在我们就有足够的工具来在中央服务器上实现模型平均这一聚合操作。
4 模型平均的联邦学习
在本教程的第2部分中，我们使用了非常简单的联邦学习版本来训练模型。这要求每个数据所有者信任模型所有者才能看到其梯度。在本章中，我们将展示如何使用第3部分中的高级聚合工具来允许参数由可信的“安全工作机”【中央服务器】聚合，然后将最终结果模型发送回模型所有者（我们）【工作机】。
同样，在开始实验之前，先做import：
import torch
import syft as sy
import copy
hook = sy.TorchHook(torch)
from torch import nn, optim
第一步：建立数据所有者
首先，我们将创建两个数据所有者（Bob和Alice），每个数据所有者拥有少量数据。 我们还将初始化一个名为“secure_worker”的安全机器。实际上，这可以是安全的硬件（例如英特尔的SGX），也可以只是受信任的中介。
# 创建一对工作机
bob = sy.VirtualWorker(hook, id='bob')
alice = sy.VirtualWorker(hook, id='alice')
# 中央服务器
secure_worker = sy.VirtualWorker(hook, id='secure_worker')
# 数据集
data = torch.tensor([[0,0],[0,1],[1,0],[1,1.]], requires_grad=True)
target = torch.tensor([[0],[0],[1],[1.]], requires_grad=True)
# 通过以下方式获取每个工作机的训练数据的指针
# 向bob和alice发送一些训练数据
bob_data = data[0:2].send(bob)
bob_target = target[0:2].send(bob)
alice_data = data[2:].send(alice)
alice_target = target[2:].send(alice)
第二步：建立我们的模型
对于此示例，我们将使用简单的线性模型进行训练。 我们通常可以使用PyTorch的nn.Linear构造函数对其进行初始化。
model = nn.Linear(2, 1)
第三步：发送模型的拷贝给alice和bob
接下来，我们需要将当前模型的副本发送给Alice和Bob，以便他们可以对自己的数据集执行学习步骤。
# 发送模型给工作机
bob_model = model.copy().send(bob)
alice_model = model.copy().send(alice)
# 给每个工作机设置优化器
bob_opt = optim.SGD(params=bob_model.parameters(), lr=0.1)
alice_opt = optim.SGD(params=alice_model.parameters(), lr=0.1)
第四步：训练bob和alice的模型（并行）
与通过安全平均进行联邦学习的常规做法一样，每个数据所有者首先在本地对模型进行几次迭代训练，然后再对模型进行平均。
# 并行进行训练两个工作机的模型
for i in range(10):
# 训练bob的模型
bob_opt.zero_grad()
bob_pred = bob_model(bob_data)
bob_loss = ((bob_preed - bob_target) ** 2).sum()
bob_loss.backward()
bob_opt.step()
bob_loss = bob_loss.get().data
# 训练alice的模型
alice_opt.zero_grad()
alice_pred = alice_model(alice_data)
alice_loss = ((alice_pred - alice_target) ** 2).sum()
alice_loss = backward()
alice_opt.step()
alice_loss = alice_loss.get().data
第五步：将两个更新的模型发送给安全工作机
现在，每个数据所有者都拥有部分受过训练的模型，是时候以安全的方式将它们平均在一起了。我们通过指示Alice和Bob将其模型发送到安全（可信）服务器来实现这一目标。
请注意，这种使用我们的API的方式意味着每个模型都直接发送到secure_worker。我们从未见过。
# 将训练好的模型都发送到中央服务器去
bob_model.move(secure_worker)
alice_model.move(secure_worker)
第六步：模型平均
最后，此训练epoch（译者注：一个epoch表示全部训练数据完整训练一轮）的最后一步是将Bob和Alice的训练模型平均在一起，然后使用它来设置全局“模型”的值。
# 进行模型平均
with torch.no_grad():
model.weight.set_(((alice_model.weight.data + bob_model.weight.data) / 2).get())
model.bias.set_(((alice_model.bias.data + bob_model.bias.data) / 2).get())
代码综合
现在，了解了各步骤后【以上步骤为一个epoch】，就只需要对此进行迭代多次epochs。可得到综合下的代码：
import torch
from torch import optim, nn
import syft as sy
import copy
hook = sy.TorchHook(torch)
# 创建一对工作机
bob = sy.VirtualWorker(hook, id='bob')
alice = sy.VirtualWorker(hook, id='alice')
# 中央服务器
secure_worker = sy.VirtualWorker(hook, id='secure_worker')
# 数据集
data = torch.tensor([[0,0],[0,1],[1,0],[1,1.]], requires_grad=True)
target = torch.tensor([[0],[0],[1],[1.]], requires_grad=True)
# 通过以下方式获取每个工作机的训练数据的指针
# 向bob和alice发送一些训练数据
bob_data = data[0:2].send(bob)
bob_target = target[0:2].send(bob)
alice_data = data[2:].send(alice)
alice_target = target[2:].send(alice)
# 建立模型
model = nn.Linear(2, 1)
# 设置epoch和iter数目
epochs = 10
worker_iters = 5
for epoch in range(epochs):
# 发送模型给工作机
bob_model = model.copy().send(bob)
alice_model = model.copy().send(alice)
# 给每个工作机设置优化器
bob_opt = optim.SGD(params=bob_model.parameters(), lr=0.1)
alice_opt = optim.SGD(params=alice_model.parameters(), lr=0.1)
# 并行进行训练两个工作机的模型
for worker_iter in range(worker_iters):
# 训练bob的模型
bob_opt.zero_grad()
bob_pred = bob_model(bob_data)
bob_loss = ((bob_pred - bob_target) ** 2).sum()
bob_loss.backward()
bob_opt.step()
bob_loss = bob_loss.get().data
# 训练alice的模型
alice_opt.zero_grad()
alice_pred = alice_model(alice_data)
alice_loss = ((alice_pred - alice_target) ** 2).sum()
alice_loss.backward()
alice_opt.step()
alice_loss = alice_loss.get().data
# 将训练好的模型都发送到中央服务器去
bob_model.move(secure_worker)
alice_model.move(secure_worker)
# 进行模型平均
with torch.no_grad():
model.weight.set_(((alice_model.weight.data + bob_model.weight.data) / 2).get())
model.bias.set_(((alice_model.bias.data + bob_model.bias.data) / 2).get())
print("bob loss: {}".format(bob_loss))
print("alice loss: {}".format(alice_loss))
输出：
bob loss: 0.08803102374076843
alice loss: 0.027631577104330063
bob loss: 0.050551094114780426
alice loss: 0.008686726912856102
bob loss: 0.030559318140149117
alice loss: 0.0031294457148760557
bob loss: 0.019234294071793556
alice loss: 0.0011045122519135475
bob loss: 0.01255376823246479
alice loss: 0.0003567059466149658
bob loss: 0.008459877222776413
alice loss: 9.674315515439957e-05
bob loss: 0.005859040655195713
alice loss: 1.7674939954304136e-05
bob loss: 0.004151101224124432
alice loss: 6.076284080336336e-07
bob loss: 0.0029959077946841717
alice loss: 1.5105272268556291e-06
bob loss: 0.0021943310275673866
alice loss: 5.455509381135926e-06
最后，我们想确保我们得到的模型学习正确，因此我们将在测试数据集上对其进行评估。在这个玩具问题中，我们将使用原始数据，但在实践中，我们将希望使用新数据来了解模型对看不见的样本的泛化程度。
preds = model(data)
loss = ((preds - target) ** 2).sum()
print(preds)
print(target)
print(loss.data)
在这个示例中，平均模型相对于本地训练的纯文本模型表现不佳，但是我们能够在不暴露每个工作机的训练数据的情况下对其进行训练。我们还能够在可信任的聚合器上聚合每个工作机的更新模型，以防止数据泄露给模型所有者。
在未来的教程中，我们的目标是直接使用梯度进行可信聚合，以便我们可以使用更好的梯度估计来更新模型并获得更强大的模型。
5 欢迎来到沙盒
在上一教程中，我们一直在手工初始化hook和所有工作机。 当您只是在玩耍/了解接口时，这可能会有些烦人。因此，从现在开始，我们将使用特殊的便捷函数创建所有这些相同的变量。
例如：
import torch
import syft as sy
sy.create_sandbox(globals())
那我们创建出的这个沙盒能带给我们什么呢？
通过上面的代码，我们创建了几个虚拟工作机，并加载了很多测试数据集，把它们分布在了各个工作机周围，以便我们可以使用诸如联邦学习之类的隐私保护技术进行训练，输出如下：
Setting up Sandbox...
- Hooking PyTorch
- Creating Virtual Workers:
- bob
- theo
- jason
- alice
- andy
- jon
Storing hook and workers as global variables...
Loading datasets from SciKit Learn...
- Boston Housing Dataset
- Diabetes Dataset
- Breast Cancer Dataset
- Digits Dataset
- Iris Dataset
- Wine Dataset
- Linnerud Dataset
Distributing Datasets Amongst Workers...
Collecting workers into a VirtualGrid...
Done!
可以看到，通过创建沙盒，我们一共创建出了6个虚拟工作机，以及将hook和这6个工作机都作为了全局变量，以便我们使用。
5.1 工作机搜索功能
进行远程数据科学的一个重要方面是我们希望能够在远程计算机上搜索数据集。设想一个研究实验室想要向医院查询“无线电”数据集，代码如下：
import torch
import syft as sy
sy.create_sandbox(globals())
torch.tensor([1, 2, 3, 4, 6])
x = torch.tensor([1,2,3,4,5]).tag("#fun", "#boston", "#housing").describe("The input datapoints to the boston housing dataset.")
y = torch.tensor([1,2,3,4,5]).tag("#fun", "#boston", "#housing").describe("The input datapoints to the boston housing dataset.")
z = torch.tensor([1,2,3,4,5]).tag("#fun", "#mnist",).describe("The images in the MNIST training da")
x = x.send(bob)
y = y.send(bob)
z = z.send(bob)
# 这会在标签或说明中搜索完全匹配
results = bob.search(["#boston", "#housing"])
print(results)
print(results[0].description)
print(results)输出：
[tensor([[6.3200e-03, 1.8000e+01, 2.3100e+00, ..., 1.5300e+01, 3.9690e+02,
4.9800e+00],
[2.7310e-02, 0.0000e+00, 7.0700e+00, ..., 1.7800e+01, 3.9690e+02,
9.1400e+00],
[2.7290e-02, 0.0000e+00, 7.0700e+00, ..., 1.7800e+01, 3.9283e+02,
4.0300e+00],
...,
[4.4620e-02, 2.5000e+01, 4.8600e+00, ..., 1.9000e+01, 3.9563e+02,
7.2200e+00],
[3.6590e-02, 2.5000e+01, 4.8600e+00, ..., 1.9000e+01, 3.9690e+02,
6.7200e+00],
[3.5510e-02, 2.5000e+01, 4.8600e+00, ..., 1.9000e+01, 3.9064e+02,
7.5100e+00]])
Tags: #boston_housing .. #data #housing #boston _boston_dataset:
Description: .. _boston_dataset:...
Shape: torch.Size([84, 13]), tensor([1, 2, 3, 4, 5])
Tags: #boston #fun #housing
Description: The input datapoints to the boston housing dataset....
Shape: torch.Size([5]), tensor([1, 2, 3, 4, 5])
Tags: #boston #fun #housing
Description: The input datapoints to the boston housing dataset....
Shape: torch.Size([5]), tensor([24.0000, 21.6000, 34.7000, 33.4000, 36.2000, 28.7000, 22.9000, 27.1000,
16.5000, 18.9000, 15.0000, 18.9000, 21.7000, 20.4000, 18.2000, 19.9000,
23.1000, 17.5000, 20.2000, 18.2000, 13.6000, 19.6000, 15.2000, 14.5000,
15.6000, 13.9000, 16.6000, 14.8000, 18.4000, 21.0000, 12.7000, 14.5000,
13.2000, 13.1000, 13.5000, 18.9000, 20.0000, 21.0000, 24.7000, 30.8000,
34.9000, 26.6000, 25.3000, 24.7000, 21.2000, 19.3000, 20.0000, 16.6000,
14.4000, 19.4000, 19.7000, 20.5000, 25.0000, 23.4000, 18.9000, 35.4000,
24.7000, 31.6000, 23.3000, 19.6000, 18.7000, 16.0000, 22.2000, 25.0000,
33.0000, 23.5000, 19.4000, 22.0000, 17.4000, 20.9000, 24.2000, 21.7000,
22.8000, 23.4000, 24.1000, 21.4000, 20.0000, 20.8000, 21.2000, 20.3000,
28.0000, 23.9000, 24.8000, 22.9000])
Tags: #boston_housing #boston .. #housing #target _boston_dataset:
Description: .. _boston_dataset:...
Shape: torch.Size([84])]
print(results[0].description)输出：
.. _boston_dataset:
Boston house prices dataset
---------------------------
**Data Set Characteristics:**
:Number of Instances: 506
:Number of Attributes: 13 numeric/categorical predictive. Median Value (attribute 14) is usually the target.
:Attribute Information (in order):
- CRIM per capita crime rate by town
- ZN proportion of residential land zoned for lots over 25,000 sq.ft.
- INDUS proportion of non-retail business acres per town
- CHAS Charles River dummy variable (= 1 if tract bounds river; 0 otherwise)
- NOX nitric oxides concentration (parts per 10 million)
- RM average number of rooms per dwelling
- AGE proportion of owner-occupied units built prior to 1940
- DIS weighted distances to five Boston employment centres
- RAD index of accessibility to radial highways
- TAX full-value property-tax rate per $10,000
- PTRATIO pupil-teacher ratio by town
- B 1000(Bk - 0.63)^2 where Bk is the proportion of black people by town
- LSTAT % lower status of the population
- MEDV Median value of owner-occupied homes in $1000's
:Missing Attribute Values: None
:Creator: Harrison, D. and Rubinfeld, D.L.
This is a copy of UCI ML housing dataset.
https://archive.ics.uci.edu/ml/machine-learning-databases/housing/
This dataset was taken from the StatLib library which is maintained at Carnegie Mellon University.
The Boston house-price data of Harrison, D. and Rubinfeld, D.L. 'Hedonic
prices and the demand for clean air', J. Environ. Economics & Management,
vol.5, 81-102, 1978. Used in Belsley, Kuh & Welsch, 'Regression diagnostics
...', Wiley, 1980. N.B. Various transformations are used in the table on
pages 244-261 of the latter.
The Boston house-price data has been used in many machine learning papers that address regression
problems.
.. topic:: References
- Belsley, Kuh & Welsch, 'Regression diagnostics: Identifying Influential Data and Sources of Collinearity', Wiley, 1980. 244-261.
- Quinlan,R. (1993). Combining Instance-Based and Model-Based Learning. In Proceedings on the Tenth International Conference of Machine Learning, 236-243, University of Massachusetts, Amherst. Morgan Kaufmann.
5.2 虚拟网格
Grid 只是工作机的集合，当我们想要将数据集放在一起时，它为我们提供了一些方便的功能，例如在沙盒中search对应tag的dataset和target。代码如下：
import torch
import syft as sy
sy.create_sandbox(globals())
grid = sy.PrivateGridNetwork(*workers)
results = grid.search("#boston")
boston_data = grid.search("#boston", "#data")
boston_target = grid.search("#boston", "#target")
print(results)
print(boston_data)
print(boston_target)
6 使用CNN在MNIST上进行联邦学习
6.1 背景
联邦学习是一种非常令人兴奋且令人振奋的机器学习技术，旨在建立可在分散数据上学习的系统。想法是，数据保留在其工作机（也称为worker）的手中，这有助于改善隐私和所有权，并且该模型在工作机之间共享。例如，一种直接的应用程序是在编写文本时预测手机上的下一个单词：您不希望将用于训练的数据（即，短信）发送到中央服务器。
因此，联合学习的兴起与数据隐私意识的传播紧密相关，并且自2018年5月起实施数据保护的欧盟GDPR成为催化剂。为了遵循法规，苹果或谷歌等大型参与者已开始对该技术进行大量投资，特别是为了保护移动用户的隐私，但他们尚未提供其工具。在OpenMined，我们相信愿意进行机器学习项目的任何人都应该能够毫不费力地实现隐私保护工具。我们已经构建了用于单行加密数据的工具如我们的博客文章所述，现在我们发布了利用新的PyTorch 1.0版本提供了直观的界面来构建安全且可扩展的模型。
6.2 代码实现
导包以及模型规格
首先导入所需的官方包
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torchvision import datasets, transforms
然后是PySyft多出的部分，特别是要定义远程工作机alice 和 bob。
import syft as sy # <-- NEW: import the Pysyft library
hook = sy.TorchHook(torch) # <-- NEW: hook PyTorch ie add extra functionalities to support Federated Learning
bob = sy.VirtualWorker(hook, id="bob") # <-- NEW: define remote worker bob
alice = sy.VirtualWorker(hook, id="alice") # <-- NEW: and alice
我们定义学习任务的设置
# 定义学习任务的设置
def parser_args():
parser = ArgumentParser()
parser.add_argument('--batch_size', default=64)
parser.add_argument('--test_batch_size', default=1000)
parser.add_argument('--epochs', default=10)
parser.add_argument('--lr', default=0.01)
parser.add_argument('--momentum', default=0.5)
parser.add_argument('--no_cuda', default=False)
parser.add_argument('--seed', default=1)
parser.add_argument('--log_interval', default=30)
parser.add_argument('--save_model', default=False)
args = paser.parser_args()
return args
args = parser_args()
use_cuda = not args.no_cuda and torch.cuda.is_available()
torch.manual_seed(args.seed)
device = torch.device('cuda' if use_cuda else 'cpu')
# 当if-else语句只涉及一条赋值语句时，就按下面这个方式写，提高阅读性
kwargs = {'num_workers': 1, 'pin_memory': True} if use_cuda else {}
数据加载并发给工作机
我们首先加载数据，然后使用.federate方法将训练数据集转换为跨工作机的联合数据集。现在，该联合数据集已提供给FederatedDataLoader。测试数据集保持不变。
# 数据记载并发送给工作机
mnist_datasets = datasets.MNIST('../data', train=True, download=True,
transfrom=transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,)
])).federate((bob, alice))
fed_train_loader = sy.FedratedDataLoader(
mnist_datasets,
batch_size=args.batch_size,
shuffle=True,
**kwargs
)
test_loader = torch.utils.data.Dataloader(
datasets.MNIST('../data', train=False,
transform=transform.Compose([
transfroms.ToTensor(),
transforms.Normalize((0.1307,), (0.3081,))
])),
batch_size=args.test_batch_size,
shuffle=True,
**kwargs
)
CNN规格
在这里，我们使用与官方示例中完全相同的CNN
# 定义网络
class Net(nn.Module):
def __init__(self):
super(Net, self).__init__()
self.conv1 = nn.Conv2d(1, 20, 5, 1)
self.conv2 = nn.Conv2d(20, 50, 5, 1)
self.fc1 = nn.Linear(4*4*50, 500)
self.fc2 = nn.Linear(500, 10)
def forward(self, x):
x = F.relu(self.conv1(x))
x = F.max_pool2d(x, 2, 2)
x = F.relu(self.conv2(x))
x = F.max_pool2d(x, 2, 2)
x = x.view(-1, 4*4*50)
x = F.relu(self.fc1(x))
x = self.fc2(x)
return F.log_softmax(x, dim=1)
定义训练和测试函数
对于训练功能，由于数据批次分布在alice和bob之间，因此我们需要将模型发送到每个批次的正确位置。 然后，您使用相同的语法远程执行所有操作，就像执行本地PyTorch一样。完成后，您需要恢复模型更新和损失以寻求改进。
def train(args. model, device, dataloader, optimizer, epoch_num):
loss = F.nll_loss(output, target)
loss.backward()
optimizer.step()
data, target = data.to(device), target.to(device) # 将数据加载到device上
optimizer.zero_grad()
output = model(data)
loss = F.nll_loss(output, target)
loss.backward()
optimizer.step()
model.get() # 将model返还到本地
if batch_idx % args.log_interval == 0:
print('Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}'.format(
epoch_num, batch_idx * args.batch_size, len(dataloader) * args.batch_size,
100. * batch_idx / len(dataloader), loss.iten()))
def test(args, model, device, test_loader):
model.eval()
test_loss = 0
correct = 0
with torch.no_grad():
for data, target in test_loader:
data, target = data.to(device), target.to(device)
output = model(data)
test_loss += F.nll_loss(output, target, reduction='sum').item() # sum up batch loss
pred = output.argmax(1, keepdim=True) # get the index of the max log-probability
correct += pred.eq(target.view_as(pred)).sum().item()
test_loss /= len(test_loader.dataset)
print('\nTest set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\n'.format(
test_loss, correct, len(test_loader.dataset),
100. * correct / len(test_loader.dataset)))
开始训练
%%time
model = Net().to(device)
optimizer = optim.SGD(model.parameters(), lr=args.lr) # TODO momentum is not supported at the moment
for epoch in range(1, args.epochs + 1):
train(args, model, device, fed_train_loader, optimizer, epoch)
test(args, model, device, test_loader)
if (args.save_model):
torch.save(model.state_dict(), "mnist_cnn.pt")
综合代码（以下代码未进行测试）
由于torch版本和官方示例所用的torch版本不一致，所以以下示例仅体现出联邦学习的思想
"""
所有类都放在该文件中，方便查看（但后期复杂的任务就不能这么搞了）
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torchvision import datasets, transforms
from argparse import ArgumentParser
import syft as sy
hook = sy.TorchHook(torch)
# 定义工作机
bob = sy.VirtualWorker(hook, id='bob')
alice = sy.VirtualWorker(hook, id='alice')
# 定义网络
class Net(nn.Module):
def __init__(self):
super(Net, self).__init__()
self.conv1 = nn.Conv2d(1, 20, 5, 1)
self.conv2 = nn.Conv2d(20, 50, 5, 1)
self.fc1 = nn.Linear(4*4*50, 500)
self.fc2 = nn.Linear(500, 10)
def forward(self, x):
x = F.relu(self.conv1(x))
x = F.max_pool2d(x, 2, 2)
x = F.relu(self.conv2(x))
x = F.max_pool2d(x, 2, 2)
x = x.view(-1, 4*4*50)
x = F.relu(self.fc1(x))
x = self.fc2(x)
return F.log_softmax(x, dim=1)
# 定义学习任务的设置
def parser_args():
parser = ArgumentParser()
parser.add_argument('--batch_size', default=64)
parser.add_argument('--test_batch_size', default=1000)
parser.add_argument('--epochs', default=10)
parser.add_argument('--lr', default=0.01)
parser.add_argument('--momentum', default=0.5)
parser.add_argument('--no_cuda', default=False)
parser.add_argument('--seed', default=1)
parser.add_argument('--log_interval', default=1)
parser.add_argument('--save_model', default=False)
args = parser.parse_args()
return args
def train(args, model, device, dataloader, optimizer, epoch_num):
model.train()
for batch_idx, (data, target) in enumerate(dataloader):
model.send(data.location)
data, target = data.to(device), target.to(device) # 将数据加载到device上
optimizer.zero_grad()
output = model(data)
loss = F.nll_loss(output, target)
loss.backward()
optimizer.step()
model.get() # 将model返还到本地
if batch_idx % args.log_interval == 0:
print('Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}'.format(
epoch_num, batch_idx * args.batch_size, len(dataloader) * args.batch_size,
100. * batch_idx / len(dataloader), loss.iten()))
def test(args, model, device, test_loader):
model.eval()
test_loss = 0
correct = 0
with torch.no_grad():
for data, target in test_loader:
data, target = data.to(device), target.to(device)
output = model(data)
test_loss += F.nll_loss(output, target, reduction='sum').item() # sum up batch loss
pred = output.argmax(1, keepdim=True) # get the index of the max log-probability
correct += pred.eq(target.view_as(pred)).sum().item()
test_loss /= len(test_loader.dataset)
print('\nTest set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\n'.format(
test_loss, correct, len(test_loader.dataset),100. * correct / len(test_loader.dataset)))
if __name__ == "__main__":
args = parser_args()
use_cuda = not args.no_cuda and torch.cuda.is_available()
torch.manual_seed(args.seed)
device = torch.device('cuda' if use_cuda else 'cpu')
# 当if-else语句只涉及一条赋值语句时，就按下面这个方式写，提高阅读性
kwargs = {'num_workers': 1, 'pin_memory': True} if use_cuda else {}
# 数据记载并发送给工作机
mnist_datasets = datasets.MNIST('../data', train=True, download=True,
transfrom=transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.1307,), (0.3081,))
])).federate((bob, alice))
fed_train_loader = sy.FedratedDataLoader(
mnist_datasets,
batch_size=args.batch_size,
shuffle=True,
**kwargs
)
test_loader = torch.utils.data.DataLoader(
datasets.MNIST('../data', train=False, transform=transform.Compose([
transfroms.ToTensor(),
transforms.Normalize((0.1307,), (0.3081,))
])),
batch_size=args.test_batch_size,
shuffle=True,
**kwargs
)
# 开始训练
model = Net().to(device)
optimizer = optim.SGD(model.parameters(), lr=args.lr)
for epoch in range(1, args.epochs + 1):
train(args, model, device, fed_train_loader, optimizer, epoch)
test(args, model, device, test_loader)
if (args.save_model):
torch.save(model.state_dict(), "mnist_cnn.pt")
总结：
.federate方法，将数据集制作成联邦数据集，然后又使用sy.FedratedDataLoader，制作出联邦dataloader，用于训练。但我有疑问的是，这个federate方法是哪来的呢？那对于我们自己的数据集，如何制作联邦数据集呢？我想后面的学习应该会解答我这一点。最后一件事
与普通的PyTorch相比，进行联邦学习需要多长时间？
实际上，计算时间少于正常PyTorch执行时间的两倍！更准确地说，它需要1.9倍的时间，与我们能够添加的功能相比，这几乎是很少的。
7 使用FedratedDataset进行联邦学习
在这里，介绍了一种使用联合数据集的新工具。 Pysyft创建了一个FederatedDataset类，该类旨在像PyTorch Dataset类一样使用，并提供给联合数据加载器FederatedDataLoader，它将以联合方式对其进行迭代。
我们使用第5节提到的沙箱：
import torch
import syft as sy
sy.create_sandbox(globals(), verbose=False)
找一个数据集：
grid = sy.PrivateGridNetwork(*workers)
boston_data = grid.search("#boston", "#data")
boston_target = grid.search("#boston", "#target")
加载模型
n_features = boston_data['alice'][0].shape[1]
n_targets = 1
model = torch.nn.Linear(n_features, n_targets)
在这里，我们将获取的数据转换为FederatedDataset中的数据【使用了sy.BaseDataset、sy.FederatedDataset】，查看拥有部分数据的工作机，然后给这些工作机加载优化器
datasets = []
for worker in boston_data.keys():
dataset = sy.BaseDataset(boston_data[worker][0], boston_target[worker][0])
datasets.append(dataset)
# 建立FedratedDataset对象
dataset = sy.FederatedDataset(datasets)
print(dataset.workers)
# 加载优化器，注意是个字典
optimizers = {}
for worker in dataset.workers:
optimizer[worker] = torch.optim.Adam(params=model.parameters(), lr=1e-2)
然后，放入一个FederatedDataLoader，并进行设置
train_loader = sy.FederatedDataLoader(dataset, batch_size=32, shuffle=False, drop_last=False)
最后，进行迭代，然后我们会发现这与纯本地Pytorch训练相比有多相似：
for epoch in range(1, epochs + 1):
loss_accum = 0
for batch_idx, (data, target) in enumerate(train_loader):
model.send(data.location)
optimizer = optimizers[data.location.id]
optimizer.zero_grad()
pred = model(data)
loss = ((pred.view(-1) - target)**2).mean()
loss.backward()
optimizer.step()
model.get()
loss = loss.get()
loss_accum += float(loss)
if batch_idx % 8 == 0:
print('Train Epoch: {} [{}/{} ({:.0f}%)]\tBatch loss: {:.6f}'.format(
epoch, batch_idx, len(train_loader),
100. * batch_idx / len(train_loader), loss.item()))
print('Total loss', loss_accum)
8 Plans（计划）的介绍
我们在这里介绍一个对于扩展工业级联邦学习至关重要的对象：Plans。它大大减少了带宽使用，允许使用异步方案，并为远程设备提供了更多的自治权。Plans的原始概念可以在论文大规模联合学习：系统设计中找到，但已在PySyft库中适应了我们的需求。
Plans旨在像函数一样存储一系列的Torch操作，但它可以将该序列的操作发送给远程工作者，并保留对其的引用。这样，要对通过指针引用的某些远程输入上的 n个操作序列进行远程计算，我们现在需要发送包含Plans的引用和指针的单个消息，而不是发送 n 个消息。我们还可以为函数提供张量（我们称为state tensors）以具有扩展的功能。可以将Plans视为可以发送的函数，也可以视为可以远程发送和执行的类。因此，对于高级用户而言，Plans的概念消失了，并被魔术功能所取代，该魔术功能允许向远程工作人员发送包含一系列任意的Torch函数。
需要注意的一件事是，您可以转换为Plans的功能类别目前仅限于Hook的Torch操作序列。即使我们正在努力尽快找到解决方法，这也特别排除了诸如if，for和while语句之类的逻辑结构。 要完全精确，您可以使用这些，但是在Plans的第一次计算中采用的逻辑路径（例如，第一个if到False和for中的5个循环）将是所有后续计算中保留的逻辑路径，在大多数情况下，我们都希望避免这种情况。
8.1 导入包和模型规格
首先导入官方包：
import torch
import torch.nn as nn
import torch.nn.functional as F
这里有一个重要说明：本地工作机不应该是客户工作机。因为非客户工作机可以存储对象，我们需要这种能力来运行Plans，本地工作机应该作为server。
import syft as sy
hook = sy.TorchHook(torch)
# 重要：本地工作机不应该是客户工作机
hook.local_worker.is_client_worker = False
server = hook.local_worker # 本地工作机应该作为server
然后，我们定义远程工作机或devices，以与参考文章中提供的概念一致，并且为他们提供一些数据：
x11 = torch.tensor([-1, 2.]).tag('input_data')
x12 = torch.tensor([1, -2.]).tag('input_data2')
x21 = torch.tensor([-1, 2.]).tag('input_data')
x22 = torch.tensor([1, -2.]).tag('input_data2')
device_1 = sy.VirtualWorker(hook, id='device_1', data=(x11, x12))
device_2 = sy.VirtualWorker(hook, id='device_2', data=(x21, x22))
devices = device_1, device_2 # 相当于 devices = (device_1, device_2)
8.2 基本例子
让我们定义一个我们想要转化为Plan的函数。那么如何转换呢？使用Pysyft提供的装饰器@sy.func2plan()！如下【该装饰器是一个class，后面有po源码】：
@sy.func2plan()
def plan_double_abs(x):
x = x + x
x = torch.abs(x)
return x
此时，plan_double_abs这个函数，已经变成了我们拥有的一个Plan。让我们来输出一下，该plan的信息print(plan_double_abs)：
<Plan plan_double_abs id:54570411307 owner:me>
def plan_double_abs():
return
有了plan之后，我们该如何使用plan呢？我们需要做两件事：1）构建plan；2）将plan发送给工作机。接下来细讲这两个步骤。
构建plan
要构建plan，我们需要对一些数据进行调用。
首先让我们获取一些远程数据的引用：通过网络发送请求，并返回引用指针：
pointer_to_data = device_1.search('input_data')[0]
print(pointer_to_data)
输出：
tensor([-1., 2.])
Tags: input_data
Shape: torch.Size([2])
如果我们告诉plan，它必须在device_1上远程执行，我们将收到错误信息，因为plan尚未建立：
print(plan_double_abs.is_built)
输出：
False
所以要构建一个plan，我们只需要在该plan上调用build，并传递执行该plan所需的参数（也就是一些数据【是函数参数吗】）。注意，当构建一个plan时，所有命令都由本地工作机顺序执行，并被该plan捕获，并存储在其read_plan属性中：
plan_double_abs.build(torch.tensor([1., 2.]))
print(plan_double_abs.is_built)
此时输出：
True
此时我们就构建上了一个plan了
发送plan给工作机
在构建后之后，我们就可以将其直接发送给指定的工作机了，并获得对应的指针，输出一下该指针：
pointer_plan = plan_double_abs.send(device_1)
print(pointer_plan)
输出：
[PointerPlan | me:44754066783 -> device_1:75736204884]
从输出可以看出，和我们之前所学习到的张量指针类似，通过.send()方法即可获得该指针，我们将该指针称为PointerPlan（计划指针）
远程运行plan
现在，我们可以通过使用指向某些数据的指针，来调用指向该plan的指针，来远程运行该plan。这里我们使用了一个命令来远程运行该plan，因此plan输出的预定义位置，现在包含着结果（请记住，我们在计算之前，预先设置了结果位置）。我们得到的结果只是一个指针，就像调用一个普通的函数一样：
pointer_to_result = pointer_plan(pointer_to_data)
print(pointer_to_result)
输出：
(Wrapper)>[PointerTensor | me:1517987496 -> device_1:9214123300]
同样，我们使用.get()方法，取回结果指针的值
print(pointer_to_result.get())
输出：
tensor([2., 4.])
8.3 面向一个具体的例子
但是我们要做的是将plan应用于深度联邦学习，对吗？ 因此，让我们看一个更复杂的示例，使用神经网络。 请注意，我们现在正在将一个类转换为一个plan。 为此，我们的网络这个class要继承sy.Plan，而不是像之前一样继承nn.Module:【之前函数变成Plan时，是使用的函数装饰器】
class Net(sy.Plan):
def __init__(self):
super(Net, self).__init__()
self.fc1 = nn.Linear(2, 3)
self.fc2 = nn.Linear(3, 2)
def forward(self, x):
x = F.relu(self.fc1(x))
x = self.fc2(x)
return F.log_softmax(x, dim=0)
实例化一下我们定义的Net，然后输出一下该实例：
net = Net()
print(net)
输出：
<Net Net id:89645570874 owner:me>
def Net():
return
然后同样，我们使用.build方法，以及一些模拟数据来构建plan：
net.build(torch.tensor([1., 2.]))
现在，我们将plan发送给远程工作机
pointer_to_net = net.send(device_1)
print(pointer_to_net)
输出：
[PointerPlan | me:81332497640 -> device_1:41963989771]
可以看到，该指针仍然是个计划指针。
让我们来检索一些远程数据：
pointer_to_data = device_1.search('input_data')[0]
然后，就像本地执行一样，来进行远程顺序执行。但是，与传统的远程执行相比，该方法每次执行仅进行一次通信：
pointer_to_result = pointer_to_net(pointer_to_data)
同样使用.get()方法，来得到网络输出结果：
print(pointer_to_result.get())
输出：
tensor([-0.8361, -0.5681], requires_grad=True)
注意，从这里我们也可以稍微可以看到，如何显著减少本地工作机（或服务器）与远程工作机之间的通信。
8.4 在工作机之间切换
我们希望拥有的一个主要功能是对多个工作机使用相同的plan，因此我们将根据正在考虑的远程数据批次进行更改。特别是，我们不想要每次更换工作机时，都需要重新构建plan。让我们使用上一个示例，来看看我们如何做到这一点。
和8.3节一样，我们先构建plan：
class Net(sy.Plan):
def __init__(self):
super(Net, self).__init__()
self.fc1 = nn.Linear(2, 3)
self.fc2 = nn.Linear(3, 2)
def forward(self, x):
x = F.relu(self.fc1(x))
x = self.fc2(x)
return F.log_softmax(x, dim=0)
net = Net()
# 建立计划
net.build(torch.tensor([1., 2.]))
然后，利用该plan，得到在device_1工作机上的网络输出：
pointer_to_net_1 = net.send(device_1)
pointer_to_data = device_1.search('input_data')[0]
pointer_to_result = pointer_to_net_1(pointer_to_data)
pointer_to_result.get()
实际上，我们可以利用同一plan，来构建其他PointerPlan，因此语法与在另一台设备上远程运行plan的语法相同：
pointer_to_net_2 = net.send(device_2)
pointer_to_data = device_2.search('input_data')[0]
pointer_to_result = pointer_to_net_2(pointer_to_data)
pointer_to_result.get()
【注意：当前，对于Plan类，我们只能使用一种方法，并且必须将其命名为"forward"】
8.5 自动构建plan函数
对于函数（@sy.func2plan），我们可以自动构建plan，而无需显式调用build，实际上是在创建时就已经构建了plan。
为了获得此功能，创建plan时，我们唯一需要更改的就是将装饰器的参数设置为名为args_shape的，该参数应为包含每个参数shape的列表：
@sy.func2plan(args_shape=[(-1, 1)])
def plan_double_abs(x):
x = x + x
x = torch.abs(x)
return x
print(plan_double_abs.is_built)
输出：
True
参数args_shape在内部用于创建具有给定形状的模拟张量，这些张量用于构建plan，比如这里，就规定了x的shape为(1, 2)，y的shape为(-1, 2)
@sy.func2plan(args_shape=[(1, 2), (-1, 2)])
def plan_sum_abs(x, y):
s = x + y
return torch.abs(s)
同时，我们还可以为函数提供状态元素：
@sy.func2plan(args_shape=[(1,)], state=(torch.tensor([1]), ))
def plan_abs(x, state):
bias, = state.read() # 该条代码在后续章节中会解释
x = x.abs()
return x + bias
使用该plan，并得到plan输出：
pointer_plan = plan_abs.send(device_1)
x_ptr = torch.tensor([-1, 0]).send(device_1)
p = pointer_plan(x_ptr)
p.get()
可以看到，我们因为在装饰器中提供了状态state这个参数，所以原本plan_abs这个函数需要x和state两个参数，但是我们在调用该函数时，只需要提供x参数，而state就不用提供了，而是从装饰器中获取，这就是state的用法。
【
看一下func2plan源码：
实际上，func2plan其实是一个class，直接po出它的源码：
class func2plan(object):
"""Decorator which converts a function to a plan.
Converts a function containing sequential pytorch code into
a plan object which can be sent to any arbitrary worker.
This class should be used only as a decorator.
"""
def __init__(self, args_shape=None, state=None):
self.args_shape = args_shape
self.state_tensors = state or tuple()
# include_state is used to distinguish if the initial plan is a function or a class:
# if it's a function, then the state should be provided in the args, so include_state
# will be true. And to know if it was indeed a function, we just need to see if a
# "manual" state was provided.
self.include_state = state is not None
def __call__(self, plan_function):
plan = Plan(
name=plan_function.__name__,
include_state=self.include_state,
forward_func=plan_function,
state_tensors=self.state_tensors,
id=sy.ID_PROVIDER.pop(),
owner=sy.local_worker,
)
# Build the plan automatically
if self.args_shape:
args = Plan._create_placeholders(self.args_shape)
try:
plan.build(*args)
except TypeError as e:
raise ValueError(
"Automatic build using @func2plan failed!\nCheck that:\n"
" - you have provided the correct number of shapes in args_shape\n"
" - you have no simple numbers like int or float as args. If you do "
"so, please consider using a tensor instead."
)
return plan
从注释可以看出，include_state用于区分初始plan是函数还是类：如果是函数，则应在 args 中提供状态，因此 include_state 将为 true。 要知道它是否确实是一个函数，我们只需要查看是否提供了“手动”状态。 所以，之前我们在构建函数plan的时候，提供了状态，所以我们可以知道plan为一个函数
】
要了解更多信息，我们可以在教程Part 8-bias中发现我们如何使用带有协议的plan！
8bias 协议介绍
现在我们已经完成了plan，我们将引入一个称为协议（protocol）的新对象。协议协调一系列plan，并在部署在远程的工作机上一次性运行它们。
它是一个高级对象，其中包含分布在多个工作程序中的复杂计算的逻辑。协议的主要特征是能够在工作机之间发送/搜索/取回，并最终部署到确定的工作机中。因此，用户可以设计协议，然后将其上载到Cloud Worker，其他任何Worker都可以搜索，下载并在其所连接的Worker上应用其包含的计算程序。
让我们看看如何使用它！
8.1 创建和部署
通过提供（worker,plan）对的列表【包含元组的列表】来创建协议。“worker”可以是工作机，也可以是工作机ID，也可以是表示虚拟工作机的字符串。在创建时可以使用后一种情况来指定部署时同一工作机应拥有（或不拥有）两个plan。“plan”可以是plan，也可以是PointerPlan（计划指针）。
import torch
import syft as sy
hook = sy.TorchHook(torch)
# Local worker should not be a client worker
hook.local_worker.is_client_worker = False
注意，本地工作机不应该作为一个远程工作机（客户机）。现在，让我们定义3个plan，并将其提供给协议，它们都执行增量操作。如何提供给协议呢？使用sy.Protocol
@sy.func2plan(args_shape=[(1,)])
def inc1(x):
return x + 1
@sy.func2plan(args_shape=[(1,)])
def inc2(x):
return x + 1
@sy.func2plan(args_shape=[(1,)])
def inc3(x):
return x + 1
protocol = sy.Protocol([("worker1", inc1), ("worker2", inc2), ("worker3", inc3)]) # 提供给协议
协议收到plan后，就要将协议给绑定到工作机上。绑定过程通过调用.deploy(*workers)完成，这里的workers是工作集的集合，是一个元组。首先让我们创建一些工作机，然后绑定：
bob = sy.VirtualWorker(hook, id='bob')
alice = sy.VirtualWorker(hook, id='alice')
charlie = sy.VirtualWorker(hook, id='charlie')
workers = (alice, bob, charlie)
protocol.deploy(*workers)
这样，我们的3个plans就跟随着协议一起部署到对应的工作机上了，让我们输出一下此时的协议print(protocol)：
<Protocol id:51726074720 owner:me resolved>
- alice: [PointerPlan | me:61159055049 -> alice:51371542861]
- bob: [PointerPlan | me:56614028751 -> bob:42106550172]
- charlie: [PointerPlan | me:83897890237 -> charlie:17599063267]
可以看到，三个工作机分别拿到了对应的plan
总结：这分两个阶段完成：1）我们将创建时提供的虚拟工作机（以字符串命名）映射到提供的工作程序；2）我们将相应的plan发给每一个工作机。
8.2 运行协议
运行协议意味着依次执行所有计划。我们提供一些输入数据，该数据将发送到第一个plan位置，然后运行第一个plan，并将其输出移到第二个plan位置，以此类推。运行完所有plan后，将返回最终结果，该结果由指向最后一个plan位置的指针组成。
使用协议的.run方法来执行协议：
x = torch.tensor([1.0])
ptr = protocol.run(x)
print(ptr)
print(ptr.get())
输出：
send alice
move alice -> bob
move bob -> charlie
(Wrapper)>[PointerTensor | me:5452301871 -> charlie:35148872768]
tensor([4.])
前三行输出为.run()方法的输出。从上面的输出可以知道，我们输入了1.0，该输入实际上顺序经历了3个plans，因此从1增加到了4（1+1+1+1）。并且看到指针的输出，可以看到，该指针指向了最后一个charlie，也就是指向了最后一个plan的位置。
实际上，我们还可以在某些指向数据的指针上，远程运行协议：
james = sy.VirtualWorker(hook, id='james')
protocol.send(james) # 现在protocol在jame手中了
x = torch.tensor([1.0]).send(james) # 现在我们创建了一个tensor，并发给james
# 此时james既有用数据，又拥有协议，我们便可以远程操控james来远程运行协议
ptr = protocol.run(x)
print(ptr)
ptr = ptr.get()
print(ptr)
ptr = ptr.get()
print(ptr)
输出：
send alice
move alice -> bob
move bob -> charlie
send remote run request to james
send alice
move alice -> bob
move bob -> charlie
(Wrapper)>[PointerTensor | me:93015393897 -> james:434626763] # 第一个print
(Wrapper)>[PointerTensor | me:434626763 -> charlie:61110018911] # 第二个print
tensor([4.]) # 第三个print
从输出可知，结果是一个指向james的指针
8.3 搜索协议
在实际设置中，我们可能希望下载一个远程协议，将其部署在您的工作程序上并与数据一起运行：
让我们初始化一个尚未部署的协议，然后将其放在远程工作机上，并获得变量me，来模拟整个搜索过程：
protocol = sy.Protocol([("worker1", inc1), ("worker2", inc2), ("worker3", inc3)])
protocol.tag("my_protocol")
protocol.send(james)
me = sy.hook.local_worker
现在我们开始根据tag，在指定工作机上来搜索协议，使用的是sy.hook.local_worker.request_search方法，该方法就两个参数：
responses = me.request_search(['my_protocol'], location=james)
print(responses)
输出：
[[PointerProtocol | me:22287453707 -> james:3884302893]]
从输出可以看到，该方法返回的responses是个列表，其中包含了若干个协议指针，包含了我们的搜索协议目标.
因此搜索到之后，我们有权访问协议的指针，并且可以将其取回：
ptr_protocol = responses[0]
protocol_back = ptr_protocol.get()
输出：
[PointerProtocol | me:53245575067 -> james:75439236666]
<Protocol id:75439236666 owner:me>
- me: <Plan inc1 id:1158278370 owner:me built>
def inc1(arg_1):
_2 = _1.__add__(1)
return _2
- me: <Plan inc2 id:24680755304 owner:me built>
def inc2(arg_1):
_2 = _1.__add__(1)
return _2
- me: <Plan inc3 id:64530481819 owner:me built>
def inc3(arg_1):
_2 = _1.__add__(1)
return _2
最后，我们也可以像在本章8.1和8.2部分中所做的那样进行操作：
protocol_back.deploy(*workers)
x = torch.tensor([1.0])
ptr = protocol_back.run(x)
print(ptr.get())
输出：
send alice
move alice -> bob
move bob -> charlie
tensor([4.])
协议将附带更多实际示例，但我们已经可以看到此新对象带来的所有可能性！
9 加密程序简介
我们可以在程序中对所有变量进行加密的情况下运行该程序！
在本教程中，我们将逐步介绍加密计算的基本工具。特别地，我们将集中于一种流行的方法，称为安全多方计算。在本课程中，我们将学习如何构建一个可以对加密数字执行计算的加密计算器。
9.1 使用安全的多方计算来进行加密
安全多方计算，叫做Secure Multi-Party Computation，简称SMPC，是一种非常奇怪的“加密”形式。 每个值都被分成多个“共享”，而不是使用公共/私有密钥对变量进行加密，每个共享都像私有密钥一样工作。 通常，这些“份额”将分配给2个或更多owners。 因此，为了解密变量，所有owners必须同意允许解密。 本质上，每个人都有一个私钥。更多的介绍[见这里]([(36条消息) SMPC加密-计算平均值的-小例子-有点神奇]_南瓜派三蔬-CSDN博客)
Encrypt()【加密】
因此，假设我们要“加密”变量“x”，可以通过以下方式进行。
加密不使用浮点数或实数，而是在称为整数商环的数学空间中进行，该空间基本上是介于
0和Q-1之间的整数 ，其中Q是质数，并且“足够大”，以便该空间可以包含我们在实验中使用的所有数字。 实际上，给定值x整数，我们将x％Q放入环中。 （这就是为什么我们避免使用数字“ x”> Q”的原因）。
如下：
import random
Q = 1234567891011
x = 25
def encrypt(x):
share_a = random.randint(-Q, Q)
share_b = random.randint(-Q, Q)
share_c = (x - share_a - share_b) % Q
return (share_a, share_b, share_c)
print(encrypt(x))
输出：
(890804432397, -2305631655, 346069090294)
可以看到，我们将x分为3个不同的份额，可以将其发送给3个不同的owners。
Decrypt()【解密】
def decrypt(*shares):
return sum(shares) % Q
a, b, c = encrypt(x)
d = decrypt(a, b, c)
print(d)
输出：
25
从输出可以看到，我们必须要使用三个共享，才能进行解密，如果单单使用了两个共享进行解密，解密将不会起作用：
decrypt(a, b)
输出：
778460474681
因此，我们需要所有所有者参与才能解密该值。 通过这种方式，shares就像私钥一样，所有私钥都必须存在才能解密值。
9.2 使用SMPC的基本算法
然而，安全多方计算的真正非凡特性是能够在变量被加密的同时进行计算。让我们在下面演示简单的加法。
x = encrypt(25)
y = encrypt(5)
def add(x, y):
z = list()
# 第一个工作机将其共享分片相加
z.append((x[0] + y[0]) % Q)
# 第二个工作机将其共享分片相加
z.append((x[1] + y[1]) % Q)
# 第三个工作机将其共享分片相加
z.append((x[2] + y[2]) % Q)
return z
decrypt(*add(x,y))
输出：
30
即，我们将x和y分别加密成了3个shares，然后在加密状态下进行了加法计算【实质上是分片相加】，最后解密成功，得到了加法结果。
如果每个工作机（分别）将其shares加在一起，则所得shares将解密为正确的值（25 + 5 == 30）。
事实证明，存在SMPC协议，该协议可以允许针对以下操作进行此加密计算：
并使用这些基本的基础原语，我们可以执行任意计算！！！
在下一节中，我们将学习如何使用PySyft库执行这些操作！
9.3 使用Pysyft的SMPC
在前面的部分中，我们概述了关于SMPC应该起作用的一些基本直觉。但是，实际上，在编写加密程序时，我们不需要自己亲自编写所有原始操作。因此，在本节中，我们将逐步介绍如何使用PySyft进行加密计算的基础知识。特别是，我们将集中精力于如何执行前面提到的3个原语：加法，乘法和比较。
基础的加/解密
加密就像获取任何PySyft张量并调用.share()一样简单。解密就像在共享变量上调用.get()一样简单
x = torch.tensor([25])
encrypted_x = x.share(bob, alice, bill)
print(encrypted_x)
print(encrypted_x.get())
输出：
(Wrapper)>[AdditiveSharingTensor]
-> [PointerTensor | me:85135990727 -> bob:19903461041]
-> [PointerTensor | me:33564204399 -> alice:86858171839]
-> [PointerTensor | me:37366246291 -> bill:95595033647]
*crypto provider: me*
tensor([25])
可以看到，.share()方法所需要的参数为所要共享x的工作机，而对加密值encrypted_X进行解密的话，直接调用其的.get()方法即可。
内省加密值
如果我们仔细观察Bob，Alice和Bill的工作机，我们可以看到创建的份额！
bob = sy.VirtualWorker(hook, id='bob')
alice = sy.VirtualWorker(hook, id='alice')
bill = sy.VirtualWorker(hook, id='bill')
print(bob._objects)
x = torch.tensor([25]).share(bob, alice, bill)
print(bob._objects)
输出：
{}
{93159236814: tensor([2310301303372059612])}
可以看到，其实VirtualWorker对象的_objects属性其实是一个字典，调用.share()之后，可以看到bob已经得到了它的份额。这样得到份额值：
bob_share = list(bob._objects.values())[0]
print(bob_share)
输出：
tensor([2310301303372059612])
如果愿意，我们可以使用我们之前讨论的同样的方法解密这些值：
Q = x.child.field
print((bobs_share + alices_share + bills_share) % Q)
输出：
tensor([25])
如我们所见，当我们调用.share()时，它只是将值分割成3股，并向每一方发送了一份！
加密加法
现在您看到我们可以对基础值执行加法了！API的构造使我们可以像执行普通的PyTorch张量那样简单地执行算法：
x = torch.tensor([25]).share(bob, alice)
y = torch.tensor([5]).share(bob, alice)
z = x + y
print(z.get())
z = x - y
print(z.get())
输出：
tensor([30])
tensor([20])
加密乘法
为了进行乘法运算，我们需要一个额外的一方负责连续生成随机数（而不与其他任何一方串通）。 我们称此人为**“加密提供者”。 对于所有密集用途，加密提供者只是一个额外的VirtualWorker，但必须承认加密提供者不是“所有者”**，因为他/她不拥有份额，而是需要信任才能避免与任何现有股东串通。这样实现加密乘法：
crypto_provider = sy.VirtualWorker(hook, id='crypto_provider')
x = torch.tensor([25]).share(bob, alice, crypto_provider=crypto_provider)
y = torch.tensor([5]).share(bob, alice, crypto_provider=crypto_provider)
z = x * y
print(z.get())
输出：
tensor([125])
而且，可以进行加密的矩阵乘法：
x = torch.tensor([[1, 2],[3,4]]).share(bob,alice, crypto_provider=crypto_provider)
y = torch.tensor([[2, 0],[0,2]]).share(bob,alice, crypto_provider=crypto_provider)
z = x.mm(y)
print(z.get())
输出：
tensor([[2, 4],
[6, 8]])
加密比较
私有值之间的私有比较也是可能的。 我们在这里依赖SecureNN协议，其详细信息可以在这里找到。比较的结果也是私有共享张量。同样加密比较需要加密提供者：
x = torch.tensor([25]).share(bob,alice, crypto_provider=crypto_provider)
y = torch.tensor([5]).share(bob,alice, crypto_provider=crypto_provider)
z = x > y
print(z.get())
z = x < y
print(z.get())
z = x == y
print(z.get())
tensor([1])
tensor([0])
tensor([0])
可知道，tensor([1])表示True，tensor([0])表示False
同时，我们还可以对值，或矩阵执行求最值操作：
x = torch.tensor([2, 3, 4, 1]).share(bob,alice, crypto_provider=crypto_provider)
x.max().get()
x = torch.tensor([[2, 3], [4, 1]]).share(bob,alice, crypto_provider=crypto_provider)
max_values, max_ids = x.max(dim=0)
max_values.get()
输出：
tensor([4])
tensor([4, 3])
10 安全聚合的联邦学习
10.1 普通的联邦学习
首先，这是一些代码，用于在Boston Housing Dataset上执行经典的联邦学习。这部分代码分为几个部分。
配置
import pickle
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.utils.data import TensorDataset, DataLoader
class Parser:
"""Parameters for training"""
def __init__(self):
self.epochs = 10
self.lr = 0.001
self.test_batch_size = 8
self.batch_size = 8
self.log_interval = 10
self.seed = 1
args = Parser()
torch.manual_seed(args.seed)
kwargs = {}
加载数据集
with open('../data/BostonHousing/boston_housing.pickle','rb') as f:
((X, y), (X_test, y_test)) = pickle.load(f)
X = torch.from_numpy(X).float() # 将numpy数组转化为tensor向量
y = torch.from_numpy(y).float()
X_test = torch.from_numpy(X_test).float()
y_test = torch.from_numpy(y_test).float()
# preprocessing
mean = X.mean(0, keepdim=True)
dev = X.std(0, keepdim=True)
mean[:, 3] = 0. # the feature at column 3 is binary,
dev[:, 3] = 1. # so we don't standardize it
X = (X - mean) / dev
X_test = (X_test - mean) / dev
train = TensorDataset(X, y)
test = TensorDataset(X_test, y_test)
train_loader = DataLoader(train, batch_size=args.batch_size, shuffle=True, **kwargs)
test_loader = DataLoader(test, batch_size=args.test_batch_size, shuffle=True, **kwargs)
神经网络结构
class Net(nn.Module): # 这里继承的是nn.Module
def __init__(self):
super(Net, self).__init__()
self.fc1 = nn.Linear(13, 32)
self.fc2 = nn.Linear(32, 24)
self.fc3 = nn.Linear(24, 1)
def forward(self, x):
x = x.view(-1, 13)
x = F.relu(self.fc1(x))
x = F.relu(self.fc2(x))
x = self.fc3(x)
return x
model = Net()
optimizer = optim.SGD(model.parameters(), lr=args.lr)
挂钩Pytorch
import syft as sy
hook = sy.TorchHook(torch)
bob = sy.VirtualWorker(hook, id="bob")
alice = sy.VirtualWorker(hook, id="alice")
james = sy.VirtualWorker(hook, id="james")
compute_nodes = [bob, alice]
将数据发送给工作机 （通常工作机已经拥有了数据，这只是出于演示目的，我们选择从本地手动发送数据给工作机，让其拥有自己的数据集）：
train_distributed_dataset = []
for batch_idx, (data,target) in enumerate(train_loader):
# 轮换给每一个compute node分一个batch
data = data.send(compute_nodes[batch_idx % len(compute_nodes)])
target = target.send(compute_nodes[batch_idx % len(compute_nodes)])
train_distributed_dataset.append((data, target))
训练函数【注意这里没有进行聚合算法】
def train(epoch):
model.train()
for batch_idx, (data,target) in enumerate(train_distributed_dataset):
worker = data.location
model.send(worker)
optimizer.zero_grad()
# update the model
pred = model(data)
loss = F.mse_loss(pred.view(-1), target)
loss.backward()
optimizer.step()
model.get()
if batch_idx % args.log_interval == 0:
loss = loss.get()
print('Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}'.format(
epoch, batch_idx * data.shape[0], len(train_loader),
100. * batch_idx / len(train_loader), loss.item()))
测试函数
def test():
model.eval()
test_loss = 0
for data, target in test_loader:
output = model(data)
test_loss += F.mse_loss(output.view(-1), target, reduction='sum').item() # sum up batch loss
pred = output.data.max(1, keepdim=True)[1] # get the index of the max log-probability
test_loss /= len(test_loader.dataset)
print('\nTest set: Average loss: {:.4f}\n'.format(test_loss))
训练模型
import time
t = time.time()
for epoch in range(1, args.epochs + 1):
train(epoch)
total_time = time.time() - t
print('Total', round(total_time, 2), 's')
测试模型
test()
10.2 添加加密聚合
现在，我们将略微修改此示例，以使用加密来聚合梯度。 不同的主要部分实际上是train()函数中的1或2行代码，我们将指出。目前，让我们重新处理数据并初始化bob和alice的模型。
remote_dataset = (list(),list())
train_distributed_dataset = []
for batch_idx, (data,target) in enumerate(train_loader):
data = data.send(compute_nodes[batch_idx % len(compute_nodes)])
target = target.send(compute_nodes[batch_idx % len(compute_nodes)])
remote_dataset[batch_idx % len(compute_nodes)].append((data, target))
def update(data, target, model, optimizer):
model.send(data.location)
optimizer.zero_grad()
pred = model(data)
loss = F.mse_loss(pred.view(-1), target)
loss.backward()
optimizer.step()
return model
bobs_model = Net()
alices_model = Net()
bobs_optimizer = optim.SGD(bobs_model.parameters(), lr=args.lr)
alices_optimizer = optim.SGD(alices_model.parameters(), lr=args.lr)
models = [bobs_model, alices_model]
params = [list(bobs_model.parameters()), list(alices_model.parameters())]
optimizers = [bobs_optimizer, alices_optimizer]
建立我们的训练逻辑
唯一的真正的差异是此训练方法的内部。让我们逐步讲解
A：训练
# 选择训练哪个batch
data_index = 0
# 更新远程模型
# 我们可以在进行此操作之前对其进行多次迭代，但是在这里每个工作机只进行一次迭代
for remote_index in range(len(compute_nodes)):
data, target = remote_dataset[remote_index][data_index]
models[remote_index] = update(data, target, models[remote_index], optimizers[remote_index])
B：加密聚合
SMPC使用要求在整数上工作的加密协议。我们在这里利用PySyft张量抽象来使用
.fix_precision()将PyTorch浮点张量转换为固定精度张量。例如，精度为2的0.123在第二个十进制数字处进行舍入，因此存储的数字为整数12。
# 创建一个列表，我们将在其中存储我们的加密模型平均值
new_params = list()
# 遍历每个参数
for param_i in range(len(params[0])):
# 对每个工作机
spdz_params = list()
for remote_index in range(len(compute_nodes)):
# 从每个工作机中选择相同的参数并复制
copy_of_parameter = params[remote_index][param_i].copy()
# 由于SMPC只能使用整数（不能使用浮点数）
# 因此我们需要使用Integers存储十进制信息。
# 换句话说，我们需要使用“固定精度”编码。
fixed_precision_param = copy_of_parameter.fix_precision()
# 现在我们在远程计算机上对其进行加密。
# 注意，fixed_precision_param“已经”是一个指针。
# 因此，当我们调用share时，它实际上是对指向的数据进行加密。
# 而它会返回一个指向MPC秘密共享对象的指针，也就是我们需要的共享分片。
encrypted_param = fixed_precision_param.share(bob, alice, crypto_provider=james)
# 现在我们获取指向MPC共享值的指针
param = encrypted_param.get()
# 保存参数，以便我们可以使用工作机的相同参数取平均值
spdz_params.append(param)
# 来自多个工作人员的平均参数，将它们提取到本地计算机上
# 以固定精度解密和解码，再返回一个浮点数
new_param = (spdz_params[0] + spdz_params[1]).get().float_precision()/2
# 保存新的平均参数
new_params.append(new_param)
C：清理
with torch.no_grad():
for model in params:
for param in model:
param *= 0
for model in models:
model.get()
for remote_index in range(len(compute_nodes)):
for param_index in range(len(params[remote_index])):
params[remote_index][param_index].set_(new_params[param_index])
把以上代码放在一起
现在我们知道了每个步骤，我们可以将所有步骤放到一个训练循环中！
训练函数
def train(epoch):
for data_index in range(len(remote_dataset[0])-1):
# update remote models
for remote_index in range(len(compute_nodes)):
data, target = remote_dataset[remote_index][data_index]
models[remote_index] = update(data, target, models[remote_index], optimizers[remote_index])
# encrypted aggregation
new_params = list()
for param_i in range(len(params[0])):
spdz_params = list()
for remote_index in range(len(compute_nodes)):
spdz_params.append(params[remote_index][param_i].copy().fix_precision().share(bob, alice, crypto_provider=james).get())
new_param = (spdz_params[0] + spdz_params[1]).get().float_precision()/2
new_params.append(new_param)
# cleanup
with torch.no_grad():
for model in params:
for param in model:
param *= 0
for model in models:
model.get()
for remote_index in range(len(compute_nodes)):
for param_index in range(len(params[remote_index])):
params[remote_index][param_index].set_(new_params[param_index])
测试函数
def test():
models[0].eval()
test_loss = 0
for data, target in test_loader:
output = models[0](data)
test_loss += F.mse_loss(output.view(-1), target, reduction='sum').item() # sum up batch loss
pred = output.data.max(1, keepdim=True)[1] # get the index of the max log-probability
test_loss /= len(test_loader.dataset)
print('Test set: Average loss: {:.4f}\n'.format(test_loss))
训练、 测试模型
t = time.time()
for epoch in range(args.epochs):
print(f"Epoch {epoch + 1}")
train(epoch)
test()
total_time = time.time() - t
print('Total', round(total_time, 2), 's')
11 安全深度学习分类
11.1 前言
我们的模型和数据一样重要
数据是机器学习的推动力，创建和收集数据的组织能够构建和训练自己的机器学习模型。这使他们能够向外部组织提供此类模型即服务（MLaaS）的使用。这对于某些组织很有用——他们无法自行创建这些模型，但仍希望使用此模型对自己的数据进行预测。
但是，托管在云中的模型仍然存在隐私/ IP问题。 为了让外部组织使用它——他们必须上传输入数据（例如要分类的图像）或下载模型。从隐私的角度来看，上传输入数据可能会出现问题，但是如果创建/拥有模型的组织不愿意，则下载模型可能不是一个选择。这就形成了两难的境界。
计算加密数据
在这种情况下，一种潜在的解决方案是以一种方式对模型和数据进行加密，以允许一个组织使用另一组织拥有的模型，而无需将其IP彼此公开。存在几种允许对加密数据进行计算的加密方案，其中最为人熟知的类型是安全多方计算（SMPC），同态加密（FHE / SHE）和功能加密（FE）。我们将在这里集中讨论安全的多方计算（在教程5的此处详细介绍），其中包含私有添加共享。它依赖于SecureNN和SPDZ等加密协议，[在此出色的博客文章中给出了详细信息](https://mortendahl.github.io/2017/09/19/private-image-analysis-with-mpc /)。
这些协议在加密数据上实现了卓越的性能，并且在过去的几个月中，我们一直在努力使这些协议易于使用。具体来说，我们正在构建工具，使您可以使用这些协议，而不必自己重新实现协议（甚至不必知道其工作原理背后的加密方法）。让我们进去看看。
设定
本章教程中的确切场景如下：考虑我们是服务器，并且有一些数据。首先，我们使用此私人训练数据定义和训练模型。然后，我们与一些拥有自己数据的客户联系，该客户希望访问我们的模型以做出一些预测。
我们对模型（神经网络）进行加密。客户端加密其数据。然后，我们都使用这两个加密资产来利用模型对数据进行分类。 最后，预测结果以加密方式发送回客户端，以便服务器（即我们）对客户端数据一无所知（您既不了解输入也不了解预测）。
理想情况下，我们将在“服务器”与“客户端”之间共享输入数据，对模型亦然。为了简单起见，共享将由另外两个工作机“alice”和“ bob”持有。 如果您认为alice由客户端拥有，而bob由服务器拥有，也是一样的。
该计算在许多MPC框架中是标准的半诚实（译者注：honest-but-curious，指的是遵循协议但会试图窃取隐私信息的敌手）敌手模型中是安全的。
万事俱备, 我们开始吧！
11.2 实验
导包以及模型规格
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torchvision import datasets, transforms
我们还需要执行特定于导入/启动PySyft的命令。我们创建了一些工作程序（分别命名为“client”，“bob”和“ alice”）。 最后，我们定义crypto_provider，它提供我们可能需要的所有加密原语（[请参阅我们在SMPC上的教程以了解更多详细信息](https://github.com/OpenMined/PySyft/blob/master/examples/tutorials/Part ％2009％20-％20Intro％20to％20Encrypted％20Programs.ipynb)。
import syft as sy
hook = syft.TorchHook(torch)
client = sy.VirtualWorker(hook, id='client')
bob = sy.VirtualWorker(hook, id='bob')
alice = sy.VirtualWorker(hook, id='alice')
crypto_provider = sy.VirtualWorker(hook, id='crypto_provider')
我们定义学习任务的参数设置
class Arguments():
def __init__(self):
self.batch_size = 64
self.test_batch_size = 50
self.epochs = epochs
self.lr = 0.001
self.log_interval = 100
args = Arguments()
数据加载，并发送给工作机
在我们的设置中，我们假设服务器有权访问某些数据【也就是有训练数据，而测试数据是加密的，服务器无权访问】以首先训练其模型。这是MNIST训练集。
train_loader = torch.utils.data.DataLoader(
datasets.MNIST('../data', train=True, download=True,
transform=transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.1307,), (0.3081,))
])),
batch_size=args.batch_size, shuffle=True)
其次，客户端具有一些数据，并且希望使用服务器的模型对其进行预测。该客户端通过在两个工作人员“alice”和“ bob”之间共享共享数据来加密其数据。
SMPC使用要求在整数上工作的加密协议。我们在这里利用PySyft张量抽象来使用
.fix_precision()将PyTorch浮点张量转换为固定精度张量。例如，精度为2的0.123在第二个十进制数字处进行舍入，因此存储的数字为整数12。
test_loader = torch.utils.data.DataLoader(
datasets.MNIST('../data', train=False,
transform=transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.1307,), (0.3081,))
])),
batch_size=args.test_batch_size, shuffle=True)
private_test_loader = []
# 对输入数据进行加密
for data, target in test_loader:
private_test_loader.append((
data.fix_precision().share(alice, bob, crypto_provider=crypto_provider),
target.fix_precision().share(alice, bob, crypto_provider=crypto_provider)
))
前馈神经网络
这是服务器使用的网络模型
class Net(nn.Module):
def __init__(self):
super(Net, self).__init__()
self.fc1 = nn.Linear(784, 500)
self.fc2 = nn.Linear(500, 10)
def forward(self, x):
x = x.view(-1, 784)
x = self.fc1(x)
x = F.relu(x)
x = self.fc2(x)
return F.log_softmax(x, dim=1)
启动训练
训练是在本地进行的，所以这是纯粹的本地Pytorch训练，没有什么特别的：
def train(args, model, train_loader, optimizer, epoch):
model.train()
for batch_idx, (data, target) in enumerate(train_loader):
optimizer.zero_grad()
output = model(data)
loss = F.nll_loss(output, target)
loss.backward()
optimizer.step()
if batch_idx % args.log_interval == 0:
print('Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}'.format(
epoch, batch_idx * args.batch_size, len(train_loader) * args.batch_size,
100. * batch_idx / len(train_loader), loss.item()))
model = Net()
optimizer = optim.Adam(model.parameters(), lr=args.lr)
for epoch in range(1, args.epochs + 1):
train(args, model, train_loader, optimizer, epoch)
启动测试
def test(args, model, test_loader):
model.eval()
test_loss = 0
correct = 0
with torch.no_grad():
for data, target in test_loader:
output = model(data)
output = F.log_softmax(output, dim=1)
test_loss += F.nll_loss(output, target, reduction='sum').item() # sum up batch loss
pred = output.argmax(1, keepdim=True) # get the index of the max log-probability
correct += pred.eq(target.view_as(pred)).sum().item()
test_loss /= len(test_loader.dataset)
print('\nTest set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\n'.format(
test_loss, correct, len(test_loader.dataset),
100. * correct / len(test_loader.dataset)))
test(args, model, test_loader)
现在我们的模型已经过训练，可以随时提供服务！
安全评估（进行加密的测试）
现在，作为服务器，我们将模型发送给持有数据的工作机。由于模型是敏感信息（我们已经花了很多时间优化它！），因此我们不想透露其权重，因此像我们对数据集所做的那样，我们需要对模型进行加密：
model.fix_precision().share(alice, bob, crypto_provider=crypto_provider)
然后，进行加密的评估测试，即模型权重、数据输入、预测输出以及标签target均以加密。但是，这里的语法和纯Pytorch测试非常相似，这非常好。
我们需要从服务器端解密的唯一一个东西就是最后的评估分数，以验证预测的平均水平：
def test(args, model, test_loader):
model.eval()
n_correct_priv = 0
n_total = 0
with torch.no_grad():
for data, target in test_loader[:n_test_batches]:
output = model(data)
pred = output.argmax(dim=1)
n_correct_priv += pred.eq(target.view_as(pred)).sum()
# 这里我们的model, data, target都是加密状态
# 所以输出的output、pred、n_correct_priv也就都是加密状态，因此需要进行解密
n_correct = n_correct_priv.copy().get().float_precision().long().item()
print('Test set: Accuracy: {}/{} ({:.0f}%)'.format(
n_correct, n_total,
100. * n_correct / n_total))
test(args, model, private_test_loader)
这里就要注意一下了，因为我们的数据、模型在加密之前经过了.fix_precision()，由float精度转换为了固定精度，所以在得到结果之后，除了要对结果进行解密以外，还需要对结果进行.float_precision().long()，将其重新转换为float精度。
在这里，我们已经学会了如何进行端到端的安全预测：服务器模型的权重尚未泄漏到客户端，并且服务器没有有关数据输入或分类输出的信息！
12 在加密数据上训练加密神经网络
在本章中，我们将使用到目前为止所学的所有技术来执行神经网络训练（和预测），同时对模型和数据进行加密。【上一章的安全评估那一节，仅仅是在加密模型上进行测试，而没有涉及到加密模型的训练】
特别是，我们将介绍可用于加密计算的自定义Autograd引擎。
第一步：创建工作机、数据集和模型
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
import syft as sy
# 设置好所需
hook = sy.TorchHook(torch)
alice = sy.VirtualWorker(id="alice", hook=hook)
bob = sy.VirtualWorker(id="bob", hook=hook)
james = sy.VirtualWorker(id="james", hook=hook)
# 一个数据集
data = torch.tensor([[0,0],[0,1],[1,0],[1,1.]])
target = torch.tensor([[0],[0],[1],[1.]])
# 一个模型
class Net(nn.Module):
def __init__(self):
super(Net, self).__init__()
self.fc1 = nn.Linear(2, 2)
self.fc2 = nn.Linear(2, 1)
def forward(self, x):
x = self.fc1(x)
x = F.relu(x)
x = self.fc2(x)
return x
model = Net()
第二步：对模型和数据进行加密
加密分为两个步骤。 由于安全多方计算仅适用于整数，因此为了对带小数点的数字（例如权重和激活数）进行运算，我们需要使用固定精度对所有数字进行编码，这将为我们提供几位十进制精度。 我们通过调用.fix_precision()来实现【但记得，在最后得到预测结果的时候，要将其的精度转换回来】。
然后，我们可以像调用其他演示一样调用.share()，它将通过在Alice和Bob之间共享它们来加密所有值。 请注意，我们还将require_grad设置为True【因为要对加密模型进行训练】，这也为加密数据添加了一种特殊的autograd方法。 确实，由于安全多方计算不适用于浮点值，因此我们无法使用通常的PyTorch自动分级。因此，我们需要添加一个特殊的AutogradTensor节点来计算梯度图以进行反向传播。您可以打印任何此元素以查看其包含AutogradTensor。
# We encode everything
data = data.fix_precision().share(bob, alice, crypto_provider=james, requires_grad=True)
target = target.fix_precision().share(bob, alice, crypto_provider=james, requires_grad=True)
model = model.fix_precision().share(bob, alice, crypto_provider=james, requires_grad=True)
第三步：训练
现在我们可以使用简单的张量逻辑进行训练：
opt = optim.SGD(params=model.parameters(), lr=0,1).fix_precision()
for iter in range(20):
opt.zero_grad()
pred = model(data)
loss = ((pred - target) ** 2).sum()
loss.backward()
opt.step()
print(loss.get().float_precision()) # 解密并转换精度
从上面代码我们可以看到，除了要对模型和数据进行一个float精度到固定精度的转换，我们还需要对优化器进行一个固定精度的转换。最后，在得到loss后，对loss要进行解密以及转换回float精度，获得真正的loss。
固定精度的影响
我们可能想知道加密所有内容如何影响不断减少的损失。实际上，由于理论计算是相同的，因此数字非常接近非加密训练。您可以通过运行相同的示例来验证这一点，而无需加密，并且可以对模型进行确定性的初始化，例如在模型init中：
with torch.no_grad():
self.fc1.weight.set_(torch.tensor([[ 0.0738, -0.2109],[-0.1579, 0.3174]], requires_grad=True))
self.fc1.bias.set_(torch.tensor([0.,0.1], requires_grad=True))
self.fc2.weight.set_(torch.tensor([[-0.5368, 0.7050]], requires_grad=True))
self.fc2.bias.set_(torch.tensor([-0.0343], requires_grad=True)
我们可能会观察到的细微差异是由于在转换为固定精度时对值进行了四舍五入。 默认的precision_fractional为3，如果将其降低到2，则明文训练的差异会增加，而如果选择precision_fractional = 4，则差异会减小。
SMPC使用要求在整数上工作的加密协议。我们在这里利用PySyft张量抽象来使用
.fix_precision()将PyTorch浮点张量转换为固定精度张量。例如，精度为2的0.123在第二个十进制数字处进行舍入，因此存储的数字为整数12。
13 对MNIST进行安全的训练和评估
在构建机器学习即服务解决方案（MLaaS）时，公司可能需要请求其他合作伙伴访问数据以训练其模型。在卫生或金融领域，模型和数据都非常关键：模型参数是业务资产，而数据是严格监管的个人数据。【公司提供模型，卫生/金融机构提供数据】
在这种情况下，一种可能的解决方案是对模型和数据都进行加密，并在加密后的值上训练机器学习模型。例如，这保证了公司不会访问患者的病历，并且医疗机构将无法观察他们贡献的模型。存在几种允许对加密数据进行计算的加密方案，其中包括安全多方计算（SMPC），同态加密（FHE / SHE）和功能加密（FE）。我们将在这里集中讨论多方计算（已在教程5中进行了介绍），它由私有加性共享组成，并依赖于加密协议SecureNN和SPDZ。
本教程的确切设置如下：考虑我们是服务器，并且我们想对n个工作机持有的某些数据进行模型训练。服务器加密共享他的模型，并将每个共享发送给工作机。工作机加密共享他们的数据并在他们之间交换。在我们将要研究的环境配置中，有2个工作机：alice和bob。交换数据共享后，他们每个工作现在拥有（1）自己的共享，（2）另一工作机的数据共享和（3）模型共享。现在，计算可以开始使用适当的加密协议来私下训练模型。训练模型后，所有共享都可以发送回服务器以对其进行解密。下图对此进行了说明：
为了举例说明这个过程，让我们假设alice和bob都拥有MNIST数据集的一部分，然后训练一个模型来执行数字分类！
13.1 在MNIST上进行加密训练的demo
导包以及训练配置
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torchvision import datasets, transforms
import time
此类描述了训练的所有超参数。 请注意，它们在这里都是公开的：
class Arguments():
def __init__(self):
self.batch_size = 64
self.test_batch_size = 64
self.epochs = epochs
self.lr = 0.02
self.seed = 1
self.log_interval = 1 # Log info at each batch
self.precision_fractional = 3
args = Arguments()
_ = torch.manual_seed(args.seed)
我们连接到两个名为alice和bob的远程工作机，并请求另一个名为crypto_provider的工作机，它提供了我们可能需要的所有加密原语：
import syft as sy # import the Pysyft library
hook = sy.TorchHook(torch) # hook PyTorch to add extra functionalities like Federated and Encrypted Learning
# simulation functions
def connect_to_workers(n_workers):
return [
sy.VirtualWorker(hook, id=f"worker{i+1}")
for i in range(n_workers)
]
def connect_to_crypto_provider():
return sy.VirtualWorker(hook, id="crypto_provider")
workers = connect_to_workers(n_workers=2)
crypto_provider = connect_to_crypto_provider()
获取访问权限和加密共享数据
在这里，我们使用一个效用函数来模拟以下行为：我们假设MNIST数据集分布在各个部分中，每个部分都由我们的一个工作机持有。 然后，工作机将其数据分batch拆分，并在彼此之间加密共享其数据。 返回的最终对象是这些加密共享batch上的可迭代对象，我们称之为“私有数据加载器”。 请注意，在此过程中，本地工作人员（即我们）从未访问过数据。
我们像往常一样获得了训练和测试私有数据集，并且输入和标签都是加密共享的：
def get_private_data_loaders(precision_fractional, workers, crypto_provider):
# 该函数用于将target转换为one-hot编码格式
# 为什么要这样做，后面有解释
def one_hot_of(index_tensor):
"""
Transform to one hot tensor
Example:
[0, 3, 9]
=>
[[1., 0., 0., 0., 0., 0., 0., 0., 0., 0.],
[0., 0., 0., 1., 0., 0., 0., 0., 0., 0.],
[0., 0., 0., 0., 0., 0., 0., 0., 0., 1.]]
"""
onehot_tensor = torch.zeros(*index_tensor.shape, 10) # 10 classes for MNIST
onehot_tensor = onehot_tensor.scatter(1, index_tensor.view(-1, 1), 1)
return onehot_tensor
def secret_share(tensor):
"""
Transform to fixed precision and secret share a tensor
"""
return (
tensor
.fix_precision(precision_fractional=precision_fractional
.share(*workers, crypto_provider=crypto_provider,
requires_grad=True)
)
transformation = transforms.Compose([
transforms.ToTensor(),
transforms.Normalize((0.1307,), (0.3081,))
])
train_loader = torch.utils.data.DataLoader(
datasets.MNIST('../data', train=True, download=True, transform=transformation),
batch_size=args.batch_size
)
private_train_loader = [
(secret_share(data), secret_share(one_hot_of(target)))
for i, (data, target) in enumerate(train_loader)
if i < n_train_items / args.batch_size
]
test_loader = torch.utils.data.DataLoader(
datasets.MNIST('../data', train=False, download=True, transform=transformation),
batch_size=args.test_batch_size
)
private_test_loader = [
(secret_share(data), secret_share(target.float()))
for i, (data, target) in enumerate(test_loader)
if i < n_test_items / args.test_batch_size
]
return private_train_loader, private_test_loader
private_train_loader, private_test_loader = get_private_data_loaders(
precision_fractional=args.precision_fractional,
workers=workers,
crypto_provider=crypto_provider
)
模型构建
class Net(nn.Module):
def __init__(self):
super(Net, self).__init__()
self.fc1 = nn.Linear(28 * 28, 128)
self.fc2 = nn.Linear(128, 64)
self.fc3 = nn.Linear(64, 10)
def forward(self, x):
x = x.view(-1, 28 * 28)
x = F.relu(self.fc1(x))
x = F.relu(self.fc2(x))
x = self.fc3(x)
return x
训练和测试
训练几乎像往常一样进行，真正的区别是我们不能使用像负对数可能性（PyTorch中的F.nll_loss）之类的损失，因为使用SMPC再现这些功能相当复杂。相反，我们使用更简单的均方误差损失。所以这里我们在训练之前，需要将target给转换为one-hot编码格式，这样便于计算均方差损失
def train(args, model, private_train_loader, optimizer, epoch):
model.train()
for batch_idx, (data, target) in enumerate(private_train_loader): # <-- now it is a private dataset
start_time = time.time()
optimizer.zero_grad()
output = model(data)
# loss = F.nll_loss(output, target) <-- not possible here
batch_size = output.shape[0]
loss = ((output - target)**2).sum().refresh()/batch_size
loss.backward()
optimizer.step()
if batch_idx % args.log_interval == 0:
loss = loss.get().float_precision()
print('Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}\tTime: {:.3f}s'.format(
epoch, batch_idx * args.batch_size, len(private_train_loader) * args.batch_size,
100. * batch_idx / len(private_train_loader), loss.item(), time.time() - start_time))
测试函数不变：
def test(args, model, private_test_loader):
model.eval()
test_loss = 0
correct = 0
with torch.no_grad():
for data, target in private_test_loader:
start_time = time.time()
output = model(data)
pred = output.argmax(dim=1)
correct += pred.eq(target.view_as(pred)).sum()
correct = correct.get().float_precision()
print('\nTest set: Accuracy: {}/{} ({:.0f}%)\n'.format(
correct.item(), len(private_test_loader)* args.test_batch_size,
100. * correct.item() / (len(private_test_loader) * args.test_batch_size)))
开始训练
关于这里发生的事情的一些注意事项。首先，我们加密共享所有工作机的所有模型参数。其次，我们将优化器的超参数转换为固定精度。注意，我们不需要加密共享它们，因为它们在我们的上下文中是公共的，但是当加密共享值存在于有限域中时，我们仍然需要使用.fix_precision将它们移入有限域中，以便执行一致的操作像权重更新: W←W−α∗ΔW.【比如像a这种超参数，我们也需要将其固定精度，因为W等模型参数已经固定精度了，所以精度和原来的a是不一样的，所以只有把a也固定精度，才能进行正常的梯度下降】
model = Net()
model = model.fix_precision().share(*workers, crypto_provider=crypto_provider, requires_grad=True)
在这！使用MNIST数据集的一小部分，使用100％加密的训练，我们只能获得75％的准确性！
13.2 讨论部分
通过分析我们刚刚做的事情，让我们更仔细地了解加密训练的功能。
计算时间
第一件事显然是运行时间！您肯定已经注意到，它比明文训练要慢得多。特别是，在batch_size=64下进行一次迭代需要3.2s，而在纯PyTorch中只有13 ms。尽管这似乎是一个阻塞程序，但请回想一下，这里的所有事情都是远程发生的，并且是在加密的世界中发生的：没有单个数据项被公开。更具体地说，处理一项的时间是50ms，这还不错。真正的问题是分析何时需要加密训练以及何时仅加密预测就足够了。例如，在生产就绪的情况下，完全可以接受50毫秒执行预测！
一个主要的瓶颈是昂贵的激活功能的使用：SMPC的relu激活非常昂贵，因为它使用私有比较和SecureNN协议。例如，如果我们用二次激活代替relu，就像在CryptoNets等加密计算的几篇论文中所做的那样，我们将从3.2s降到1.2s。
通常，关键思想是仅加密必需的内容，本教程向您展示它可以多么简单。
SMPC的反向传播
您可能会想知道，尽管我们在有限域中使用整数，但我们如何执行反向传播和梯度更新。为此，我们开发了一个新的syft张量，称为AutogradTensor。 尽管您可能没有看过本教程，但还是大量使用它！ 让我们通过打印模型的weight进行检查：
model.fc3.weight
和一个数据项：
frist_batch, input_data = 0, 0
private_train_loader[first_batch][input_data]
正如我们所看到的，AutogradTensor在那里！ 它位于Torch包装器和FixedPrecisionTensor之间，这表明值现在位于有限域中。此AutogradTensor的目标是在对加密值进行操作时存储计算图。这很有用，因为在向后调用反向传播时，此AutogradTensor会覆盖所有与加密计算不兼容的向后函数，并指示如何计算这些梯度。 例如，对于使用Beaver三元组技巧完成的乘法，我们不想再对技巧进行区分，因为区分乘法应该非常容易：∂b(a⋅b)=⋅∂b。 例如，这是我们描述如何计算这些梯度的方法：
class MulBackward(GradFunc):
def __init__(self, self_, other):
super().__init__(self, self_, other)
self.self_ = self_
self.other = other
def gradient(self, grad):
grad_self_ = grad * self.other
grad_other = grad * self.self_ if type(self.self_) == type(self.other) else None
return (grad_self_, grad_other)
如果您想知道更多我们如何实现梯度的，可以查看tensors / interpreters / gradients.py。
就计算图而言，这意味着该图的副本保留在本地，并且协调正向传递的服务器还提供有关如何进行反向传递的指令。 在我们的环境中，这是一个完全正确的假设。
安全保障
最后，让我们给出一些有关我们在此处实现的安全性的提示：我们正在考虑的对手诚实但好奇：这意味着对手无法通过运行此协议来了解有关数据的任何信息，但是恶意的对手仍可能偏离协议，例如尝试破坏共享以破坏计算。在这样的SMPC计算（包括私有比较）中针对恶意对手的安全性仍然是一个未解决的问题。
此外，即使安全多方计算确保不访问训练数据，此处仍然存在来自纯文本世界的许多威胁。例如，当您可以向模型提出请求时（在MLaaS的上下文中），您可以获得可能泄露有关训练数据集信息的预测。特别是，您没有针对成员资格攻击的任何保护措施，这是对机器学习服务的常见攻击，在这种攻击中，对手想确定数据集中是否使用了特定项目。除此之外，其他攻击，例如意外的记忆过程（学习有关数据项特定特征的模型），模型反演或提取仍然可能。
对上述许多威胁有效的一种通用解决方案是添加差异隐私。它可以与安全的多方计算完美地结合在一起，并且可以提供非常有趣的安全性保证。我们目前正在研究几种实现方式，并希望提出一个将两者结合起来的示例！
至此，Pysyft的基本教程已经完结，现在跟着[这篇文章](联邦学习小系统搭建和测试（PySyft + Raspberry Pi 4） – 知乎 (zhihu.com))以及这里学习Pysyft如何实现分布式机器的联邦学习【以上教程均为单机模拟】
14 联邦学习分布式系统搭建
Pysyft0.2.4官方提供了一个分布式的demo，有关MNIST的一个分类任务。现开始介绍跑通demo的步骤。
14.1 配置环境
由于是分布式训练，所以我们要在我们所拥有的机器中均配置好Pysyft0.2.4环境，配置过程如下：
安装Anaconda
创建pysyft环境，并安装Python3.6版本：
conda create -n pysyft python=3.7
安装pysyft（如下指令会一同安装torch1.4.0以及torchvision0.5.0以及其他依赖库）:
pip install syft==0.2.4
安装后，打开python，若import syfyt和import torch能够成功，说明Pysyft和Pytorch均已安装成功。
14.2 启动server
我的分布式环境为一台四卡服务器，一台三卡服务器。三卡服务器充当中央服务器（用于聚合模型）以及模型测试端；四卡服务器充当客户机端，并在该服务器起两个进程，开启两个客户机。所有终端均需要启动server，绑定IP地址，以进行通信：
三卡服务器：
python3 run_websocket_server.py --host '127.0.0.1' --port 8780 --id testing --testing
四卡服务器：
python3 run_websocket_server.py --host '113.54.128.247' --port 8777 --id alice
python3 run_websocket_server.py --host 113.54.128.247 --port 8778 --id bob
正常启动时，可以看到如下输出，以三卡为例：
2022-01-21 11:18:18,979 | MNIST dataset (test set), available numbers on testing:
2022-01-21 11:18:19,001 | 0: 980
2022-01-21 11:18:19,001 | 1: 1135
2022-01-21 11:18:19,001 | 2: 1032
2022-01-21 11:18:19,002 | 3: 1010
2022-01-21 11:18:19,002 | 4: 982
2022-01-21 11:18:19,002 | 5: 892
2022-01-21 11:18:19,002 | 6: 958
2022-01-21 11:18:19,002 | 7: 1028
2022-01-21 11:18:19,002 | 8: 974
2022-01-21 11:18:19,002 | 9: 1009
2022-01-21 11:18:19,002 | datasets: {'mnist_testing': BaseDataset
Data: tensor([[[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
...,
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0]],
[[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
...,
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0]],
[[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
...,
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0]],
...,
[[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
...,
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0]],
[[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
...,
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0]],
[[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
...,
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0],
[0, 0, 0, ..., 0, 0, 0]]], dtype=torch.uint8)
targets: tensor([7, 2, 1, ..., 4, 5, 6])}
Serving. Press CTRL-C to stop.
14.3 修改代码
由于Pysyft官方的分布式项目中，主程序代码均由Jupyter编写，所以我们需要将其转化为.py文件，以在服务器上能够正常运行。但是在该过程中，由于程序中使用到了异步编程，即async、await的联合使用（具体使用在[这里](Python Async/Await入门指南 – dhcn – 博客园 (cnblogs.com))），但是在运行官方程序中的第七个code block时，我们会发现有错误，这是因为await关键字，必须要在async函数中使用，因此我们在转化为.py文件时，可以写一个由async关键字引导的main()函数，然后将我们的await异步代码给移送到main函数中执行，此时便不会报错。最后，我们要这样调用main()函数：
asyncio.get_event_loop().run_until_complete(main())
最后，具体的主程序代码见./A-f-l-o-M-new.py：
import inspect
# Dependencies
import sys
import asyncio
import syft as sy
from syft.workers.websocket_client import WebsocketClientWorker
from syft.frameworks.torch.fl import utils
import torch
from torchvision import datasets, transforms
import numpy as np
import run_websocket_client as rwc
async def main():
hook = sy.TorchHook(torch)
args = rwc.define_and_get_arguments(args=[])
use_cuda = args.cuda and torch.cuda.is_available()
torch.manual_seed(args.seed)
device = torch.device("cuda" if use_cuda else "cpu")
print(args)
import logging
logger = logging.getLogger("run_websocket_client")
if not len(logger.handlers):
FORMAT = "%(asctime)s - %(message)s"
DATE_FMT = "%H:%M:%S"
formatter = logging.Formatter(FORMAT, DATE_FMT)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
logger.addHandler(handler)
logger.propagate = False
LOG_LEVEL = logging.DEBUG
logger.setLevel(LOG_LEVEL)
# 以下6行代码需要根据实际的分布式环境所配置
kwargs_websocket = {"host": "113.54.128.247", "hook": hook, "verbose": args.verbose}
alice = WebsocketClientWorker(id="alice", port=8777, **kwargs_websocket)
bob = WebsocketClientWorker(id="bob", port=8778, **kwargs_websocket)
#charlie = WebsocketClientWorker(id="charlie", port=8779, **kwargs_websocket)
kwargs_websocket = {"host": "127.0.0.1", "hook": hook, "verbose": args.verbose}
testing = WebsocketClientWorker(id="testing", port=8780, **kwargs_websocket)
worker_instances = [alice, bob]
#worker_instances = [alice, bob, charlie]
print(inspect.getsource(rwc.Net))
model = rwc.Net().to(device)
print(model)
traced_model = torch.jit.trace(model, torch.zeros([1, 1, 28, 28], dtype=torch.float))
print("Federate_after_n_batches: " + str(args.federate_after_n_batches))
print("Batch size: " + str(args.batch_size))
print("Initial learning rate: " + str(args.lr))
learning_rate = args.lr
device = "cpu" # torch.device("cpu")
traced_model = torch.jit.trace(model, torch.zeros([1, 1, 28, 28], dtype=torch.float))
for curr_round in range(1, args.training_rounds + 1):
logger.info("Training round %s/%s", curr_round, args.training_rounds)
results = await asyncio.gather(
*[
rwc.fit_model_on_worker(
worker=worker,
traced_model=traced_model,
batch_size=args.batch_size,
curr_round=curr_round,
max_nr_batches=args.federate_after_n_batches,
lr=learning_rate,
)
for worker in worker_instances
]
)
models = {}
loss_values = {}
test_models = curr_round % 10 == 1 or curr_round == args.training_rounds
if test_models:
logger.info("Evaluating models")
np.set_printoptions(formatter={"float": "{: .0f}".format})
for worker_id, worker_model, _ in results:
rwc.evaluate_model_on_worker(
model_identifier="Model update " + worker_id,
worker=testing,
dataset_key="mnist_testing",
model=worker_model,
nr_bins=10,
batch_size=128,
print_target_hist=False,
device=device
)
# Federate models (note that this will also change the model in models[0]
for worker_id, worker_model, worker_loss in results:
if worker_model is not None:
models[worker_id] = worker_model
loss_values[worker_id] = worker_loss
traced_model = utils.federated_avg(models)
if test_models:
rwc.evaluate_model_on_worker(
model_identifier="Federated model",
worker=testing,
dataset_key="mnist_testing",
model=traced_model,
nr_bins=10,
batch_size=128,
print_target_hist=False,
device=device
)
# decay learning rate
learning_rate = max(0.98 * learning_rate, args.lr * 0.01)
if args.save_model:
torch.save(model.state_dict(), "mnist_cnn.pt")
if __name__ == "__main__":
asyncio.get_event_loop().run_until_complete(main())
14.3 运行程序，进行训练和测试
最后，在三卡服务器（中央服务器）上的对应分布式项目目录下运行以上所修改的代码：
python3 ./A-f-l-o-M-new.py
输出：
Namespace(batch_size=32, cuda=False, federate_after_n_batches=10, lr=0.1, save_model=False, seed=1, test_batch_size=128, training_rounds=40, verbose=False)
class Net(nn.Module):
def __init__(self):
super(Net, self).__init__()
self.conv1 = nn.Conv2d(1, 20, 5, 1)
self.conv2 = nn.Conv2d(20, 50, 5, 1)
self.fc1 = nn.Linear(4 * 4 * 50, 500)
self.fc2 = nn.Linear(500, 10)
def forward(self, x):
x = F.relu(self.conv1(x))
x = F.max_pool2d(x, 2, 2)
x = F.relu(self.conv2(x))
x = F.max_pool2d(x, 2, 2)
x = x.view(-1, 4 * 4 * 50)
x = F.relu(self.fc1(x))
x = self.fc2(x)
return F.log_softmax(x, dim=1)
Net(
(conv1): Conv2d(1, 20, kernel_size=(5, 5), stride=(1, 1))
(conv2): Conv2d(20, 50, kernel_size=(5, 5), stride=(1, 1))
(fc1): Linear(in_features=800, out_features=500, bias=True)
(fc2): Linear(in_features=500, out_features=10, bias=True)
)
Federate_after_n_batches: 10
Batch size: 32
Initial learning rate: 0.1
11:24:39 - Training round 1/40
11:24:45 - Evaluating models
11:24:48 - Model update alice: Percentage numbers 0-3: 100%, 4-6: 0%, 7-9: 0%
11:24:48 - Model update alice: Average loss: 0.0251, Accuracy: 2132/10000 (21.32%)
11:24:53 - Model update bob: Percentage numbers 0-3: 0%, 4-6: 100%, 7-9: 0%
11:24:53 - Model update bob: Average loss: 0.0339, Accuracy: 1275/10000 (12.75%)
11:24:56 - Federated model: Percentage numbers 0-3: 10%, 4-6: 89%, 7-9: 0%
11:24:56 - Federated model: Average loss: 0.0186, Accuracy: 2819/10000 (28.19%)
11:24:56 - Training round 2/40
11:25:01 - Training round 3/40
11:25:06 - Training round 4/40
11:25:11 - Training round 5/40
11:25:16 - Training round 6/40
11:25:21 - Training round 7/40
11:25:26 - Training round 8/40
11:25:31 - Training round 9/40
11:25:36 - Training round 10/40
11:25:41 - Training round 11/40
11:25:46 - Evaluating models
11:25:49 - Model update alice: Percentage numbers 0-3: 86%, 4-6: 13%, 7-9: 0%
11:25:49 - Model update alice: Average loss: 0.0281, Accuracy: 5167/10000 (51.67%)
11:25:52 - Model update bob: Percentage numbers 0-3: 26%, 4-6: 73%, 7-9: 0%
11:25:52 - Model update bob: Average loss: 0.0247, Accuracy: 5267/10000 (52.67%)
11:25:55 - Federated model: Percentage numbers 0-3: 56%, 4-6: 43%, 7-9: 0%
11:25:55 - Federated model: Average loss: 0.0209, Accuracy: 6702/10000 (67.02%)
11:25:55 - Training round 12/40
11:26:00 - Training round 13/40
11:26:05 - Training round 14/40
11:26:10 - Training round 15/40
11:26:15 - Training round 16/40
11:26:20 - Training round 17/40
11:26:25 - Training round 18/40
11:26:31 - Training round 19/40
11:26:36 - Training round 20/40
11:26:41 - Training round 21/40
11:26:46 - Evaluating models
11:26:49 - Model update alice: Percentage numbers 0-3: 68%, 4-6: 31%, 7-9: 0%
11:26:49 - Model update alice: Average loss: 0.0241, Accuracy: 6142/10000 (61.42%)
11:26:52 - Model update bob: Percentage numbers 0-3: 35%, 4-6: 64%, 7-9: 0%
11:26:52 - Model update bob: Average loss: 0.0294, Accuracy: 5864/10000 (58.64%)
11:26:55 - Federated model: Percentage numbers 0-3: 51%, 4-6: 48%, 7-9: 0%
11:26:55 - Federated model: Average loss: 0.0241, Accuracy: 6723/10000 (67.23%)
11:26:55 - Training round 22/40
11:27:01 - Training round 23/40
11:27:06 - Training round 24/40
11:27:11 - Training round 25/40
11:27:16 - Training round 26/40
11:27:21 - Training round 27/40
11:27:26 - Training round 28/40
11:27:31 - Training round 29/40
11:27:36 - Training round 30/40
11:27:41 - Training round 31/40
11:27:46 - Evaluating models
11:27:49 - Model update alice: Percentage numbers 0-3: 70%, 4-6: 29%, 7-9: 0%
11:27:49 - Model update alice: Average loss: 0.0260, Accuracy: 6472/10000 (64.72%)
11:27:52 - Model update bob: Percentage numbers 0-3: 37%, 4-6: 62%, 7-9: 0%
11:27:52 - Model update bob: Average loss: 0.0286, Accuracy: 5994/10000 (59.94%)
11:27:55 - Federated model: Percentage numbers 0-3: 55%, 4-6: 44%, 7-9: 0%
11:27:55 - Federated model: Average loss: 0.0248, Accuracy: 6847/10000 (68.47%)
11:27:55 - Training round 32/40
11:28:00 - Training round 33/40
11:28:05 - Training round 34/40
11:28:10 - Training round 35/40
11:28:15 - Training round 36/40
11:28:21 - Training round 37/40
11:28:26 - Training round 38/40
11:28:30 - Training round 39/40
11:28:35 - Training round 40/40
11:28:40 - Evaluating models
11:28:44 - Model update alice: Percentage numbers 0-3: 72%, 4-6: 27%, 7-9: 0%
11:28:44 - Model update alice: Average loss: 0.0266, Accuracy: 6365/10000 (63.65%)
11:28:47 - Model update bob: Percentage numbers 0-3: 40%, 4-6: 59%, 7-9: 0%
11:28:47 - Model update bob: Average loss: 0.0281, Accuracy: 6308/10000 (63.08%)
11:28:50 - Federated model: Percentage numbers 0-3: 57%, 4-6: 42%, 7-9: 0%
11:28:50 - Federated model: Average loss: 0.0247, Accuracy: 6865/10000 (68.65%)
从结果可以看出，由于我们只在四卡上起了两个客户机（alice和bob），其分别拥有0-3和4-6的训练集，所以所训练出的模型在7-9的测试集上没有分类能力，这是正常的。说明我们的分布式训练成功。另外，我们是每10个epochs（这是怎么看出来的呢？源码中args.training_rounds=40，然后run_websocket_client.py中提到每一个round训练一个epoch，所以可将epoch数目看作为training_rounds数目），就更新一次模型，分发给客户机。最后，我们可以看一下这个整个分布式训练所用的时间。对于一个很简单的MNIST任务，我们在使用gpu的情况下，训练40个epochs一共需要4分钟时间，这是很慢的，而且我们现在只起了两个客户机，那么如果集群数量庞大的话，整个时间成本不可估量。我们知道，这大量的时间都浪费在了通信上面，因此我们需要在后期阶段，思考如何降低整个通信成本，将时间成本给移送到客户机端。
11:25:49 – Model update alice: Percentage numbers 0-3: 86%, 4-6: 13%, 7-9: 0%
11:25:49 – Model update alice: Average loss: 0.0281, Accuracy: 5167/10000 (51.67%)
11:25:52 – Model update bob: Percentage numbers 0-3: 26%, 4-6: 73%, 7-9: 0%
11:25:52 – Model update bob: Average loss: 0.0247, Accuracy: 5267/10000 (52.67%)
11:25:55 – Federated model: Percentage numbers 0-3: 56%, 4-6: 43%, 7-9: 0%
11:25:55 – Federated model: Average loss: 0.0209, Accuracy: 6702/10000 (67.02%)
11:25:55 – Training round 12/40
11:26:00 – Training round 13/40
11:26:05 – Training round 14/40
11:26:10 – Training round 15/40
11:26:15 – Training round 16/40
11:26:20 – Training round 17/40
11:26:25 – Training round 18/40
11:26:31 – Training round 19/40
11:26:36 – Training round 20/40
11:26:41 – Training round 21/40
11:26:46 – Evaluating models
11:26:49 – Model update alice: Percentage numbers 0-3: 68%, 4-6: 31%, 7-9: 0%
11:26:49 – Model update alice: Average loss: 0.0241, Accuracy: 6142/10000 (61.42%)
11:26:52 – Model update bob: Percentage numbers 0-3: 35%, 4-6: 64%, 7-9: 0%
11:26:52 – Model update bob: Average loss: 0.0294, Accuracy: 5864/10000 (58.64%)
11:26:55 – Federated model: Percentage numbers 0-3: 51%, 4-6: 48%, 7-9: 0%
11:26:55 – Federated model: Average loss: 0.0241, Accuracy: 6723/10000 (67.23%)
11:26:55 – Training round 22/40
11:27:01 – Training round 23/40
11:27:06 – Training round 24/40
11:27:11 – Training round 25/40
11:27:16 – Training round 26/40
11:27:21 – Training round 27/40
11:27:26 – Training round 28/40
11:27:31 – Training round 29/40
11:27:36 – Training round 30/40
11:27:41 – Training round 31/40
11:27:46 – Evaluating models
11:27:49 – Model update alice: Percentage numbers 0-3: 70%, 4-6: 29%, 7-9: 0%
11:27:49 – Model update alice: Average loss: 0.0260, Accuracy: 6472/10000 (64.72%)
11:27:52 – Model update bob: Percentage numbers 0-3: 37%, 4-6: 62%, 7-9: 0%
11:27:52 – Model update bob: Average loss: 0.0286, Accuracy: 5994/10000 (59.94%)
11:27:55 – Federated model: Percentage numbers 0-3: 55%, 4-6: 44%, 7-9: 0%
11:27:55 – Federated model: Average loss: 0.0248, Accuracy: 6847/10000 (68.47%)
11:27:55 – Training round 32/40
11:28:00 – Training round 33/40
11:28:05 – Training round 34/40
11:28:10 – Training round 35/40
11:28:15 – Training round 36/40
11:28:21 – Training round 37/40
11:28:26 – Training round 38/40
11:28:30 – Training round 39/40
11:28:35 – Training round 40/40
11:28:40 – Evaluating models
11:28:44 – Model update alice: Percentage numbers 0-3: 72%, 4-6: 27%, 7-9: 0%
11:28:44 – Model update alice: Average loss: 0.0266, Accuracy: 6365/10000 (63.65%)
11:28:47 – Model update bob: Percentage numbers 0-3: 40%, 4-6: 59%, 7-9: 0%
11:28:47 – Model update bob: Average loss: 0.0281, Accuracy: 6308/10000 (63.08%)
11:28:50 – Federated model: Percentage numbers 0-3: 57%, 4-6: 42%, 7-9: 0%
11:28:50 – Federated model: Average loss: 0.0247, Accuracy: 6865/10000 (68.65%)
从结果可以看出，由于我们只在四卡上起了两个客户机（alice和bob），其分别拥有0-3和4-6的训练集，所以所训练出的模型在7-9的测试集上没有分类能力，这是正常的。说明我们的分布式训练成功。另外，我们是每10个epochs（这是怎么看出来的呢？源码中`args.training_rounds=40`，然后`run_websocket_client.py`中提到每一个round训练一个epoch，所以可将epoch数目看作为`training_rounds`数目），就更新一次模型，分发给客户机。**最后**，我们可以看一下这个整个分布式训练所用的时间。对于一个很简单的MNIST任务，我们在使用gpu的情况下，训练40个epochs一共需要4分钟时间，这是很慢的，而且我们现在只起了两个客户机，那么如果集群数量庞大的话，整个时间成本不可估量。我们知道，这大量的时间都浪费在了通信上面，因此我们需要在后期阶段，思考如何降低整个通信成本，将时间成本给移送到客户机端。
接下来就是修改代码，实现基于Pysyft的分布式目标检测。
![[python]通过whl文件安装torchvision和torchaudio及国内whl文件下载地址汇总](https://i3.wp.com/i-blog.csdnimg.cn/direct/6d7130def8e741a885c03ffd41a805e2.png)
