边缘计算 课件 第五章 边缘智能

边缘智能倡和正学出新明德

1.边缘智能的基本概念？

2.联邦学习的挑战及技术是什么？

3.解决联邦学习挑战的方法？

这节课的问题是什么？问题的答案是什么？边缘智能的基本概念？

(Edgeintelligence)目录1.1边缘智能基本概念011.2边缘智能vs云计算021.3典型应用场景03边缘智能（EdgeIntelligence）的概念随着人工智能与物联网的深度融合而兴起，旨在解决云端集中处理模式的瓶颈。权威研究与行业报告指出，智能计算的时代正从云端向边缘侧延伸，世界上无数的设备，从智能家居传感器到自动驾驶汽车，从工业机器人到智能手机，都将在数据产生的源头就地完成智能分析与决策。边缘计算技术、轻量化AI模型、专用AI芯片及联邦学习技术将得到广泛的应用。1.1边缘智能基本概念1.1边缘智能基本概念边缘智能（EI,EdgeIntelligence）即“智能下沉于边缘”，是云计算基础上的延伸与演进。它将人工智能模型与算法部署在网络边缘侧，就近处理终端设备产生的数据，从而实现任何时间、任何地点的实时智能决策与响应。边缘智能的本质是云智能的分布式下沉，其核心不再仅是强大的云端服务器，而是嵌入在各类终端设备及边缘节点中的轻量化智能模型。。边缘计算本质是计算，只是处理位置不再仅限于云端数据中心，而是网络边缘的终端设备与边缘节点边缘智能1.2边缘智能vs云计算边缘智能云计算对比维度分布式集中化核心理念网络边缘、终端设备附近远程数据中心处理位置本地实时推理、及时响应、数据初步处理海量数据存储、复杂模型训练、非实时分析主要任务极低，本地处理，实时响应高，受网络传输影响延迟低，仅上传关键结果或模型更新高，需上传全部原始数据带宽消耗数据留在本地，隐私保护性更强数据离设备，隐私泄露风险较高隐私安全弱网或断网环境下仍可独立工作依赖稳定的网络连接可靠性1.3典型应用场景自动驾驶汽车依赖大量传感器和摄像头来收集周围环境数据。边缘智能可以实时处理这些数据，做出驾驶决策，如避让障碍物、识别交通标志等。这减少了车辆对云端计算的依赖，提升了行驶的安全性和实时性自动驾驶

智能家居智能城市边缘智能在智能家居中，将机器学习模型部署于本地设备，能实时记录调节室内温湿度等，还能在保护用户隐私（数据不上传云端）的情况下，为不同家庭提供个性化服务，解决传统云端处理的时延与隐私泄露问题。边缘智能在智能城市的交通管理、环境监测、能源管理等方面发挥了重要作用。例如，智能交通系统可以通过边缘设备监控交通流量，并根据实时数据调整信号灯，缓解交通拥堵。联邦学习的原理及技术是什么？

(Federatedlearning)目录2.1联邦学习原理012.2联邦学习面临的挑战022.3联邦平均算法032.4卷积神经网络042.5轻量级框架052.1联邦学习原理联邦学习流程图

客户端：拥有本地私有数据的多个设备或机构（如手机、医院、传感器）。负责在本地数据进行模型训练。服务器：一个中央协调者。负责分发全局模型、挑选客户端、聚合模型更新，并维护全局模型。通信框架：连接客户端与服务器，安全地传输模型参数（而非原始数据）。2.1联邦学习原理

2.2联邦学习面临的挑战资源受限数据异构系统动态安全隐私设备有限算力与庞大模型间的矛盾，导致部署困难与高昂开销。数据非独立同分布，致使全局模型泛化性能显著下降。设备频繁掉线与性能波动，造成训练过程不稳定、效率低下。参数仍可泄露隐私并遭受攻击，安全与性能难以兼顾。

2.3联邦平均算法联邦学习是一个迭代求解全局模型的过程，客户端通过聚合服务器协同训练出全局模型，直到全局模型收敛或达到其终止条件为止。联邦平均算法的具体实现过程如算法5-1所示。2.3联邦平均算法参与方1参与方2参与方N私有数据1私有数据2私有数据N服务器聚合在学习率η下，每个客户端使用梯度下降算法更新权重，则有：每个客户端来讲，损失函数定义为：优化函数可以表示为：2.4卷积神经网络卷积网络的基本结构示意图卷积神经网络是深度学习的核心模型，广泛应用于计算机视觉和自然语言处理领域。其通过卷积层局部连接和池化层降维的特性，显著减少模型参数量，提升特征提取能力和训练效率，在图像分类等任务中表现出色。2.4卷积神经网络卷积层：功能：核心特征提取器。使用卷积核在输入数据上滑动，通过局部连接和权重共享机制，提取局部特征（如边缘、角点、纹理）。优势：与全连接相比，大幅减少了参数量，降低了模型复杂度，并赋予了模型对平移的不变性。池化层：功能：特征降维与压缩。在保留显著特征的同时，减少数据量和计算负荷。常见操作有最大池化（提取最显著特征）和平均池化。优势：增强了模型的尺度不变性和旋转不变性，同时有效防止过拟合。全连接层：功能：负责将前面提取到的分布式特征进行整合，并映射到最终的样本标签空间，完成分类或回归任务。注意：在全连接层中，每个神经元都与前一层的所有神经元连接，因此参数量巨大，是模型剪枝和压缩的重点目标。2.4卷积神经网络

输出概率分布Y：

2.5轻量级框架TensorFlowLite是Google推出的轻量级AI推理框架，专为移动和嵌入式设备优化。通过模型转换和8位量化技术，大幅提升模型运行效率，支持设备端本地部署。其优势包括数据本地处理保障隐私、降低交互延迟、减少设备耗电，为边缘计算提供高效的解决方案。2.5轻量级框架TensorFlowLite的核心目标：为移动设备、嵌入式设备和边缘设备提供高效的机器学习推理框架，通过模型优化与硬件加速，实算资源受限环境下的低延迟、高能效和隐私安全的端侧AI部署。

2.5轻量级框架TensorFlowLite框架TensorFlowModel保存预训练模型TensorFlowLiteConverter将模型转换为

.tflite

格式TensorFlowLiteModelFile优化模型速度与大小的最终文件

2.5轻量级框架关键技术：模型转换与量化将TensorFlow模型转换为压缩的FlatBuffer格式文件（.tflite）。通过将32位浮点数转换为高效的8位整数进行量化，以实现模型快速运算。三大优势：轻量级、跨平台、快速减少传输时延：数据在本地处理，无需与服务器频繁交互。有效保护隐私：用户数据停留在本地，不上传至云中心。解决联邦学习挑战的方法？

(Federatedlearning)目录3.1高精度轻量化模型013.2个性化轻量化模型023.3边缘智能实验平台033.1高精度轻量化模型根据服务器是否参与模型训练，分为客户端-服务器模式(client-server,C-S)和点对点模式(peer-to-peer,P2P)。P2P模式无中心服务器，设备间直接交换模型参数并随机选择聚合节点，以实现模型收敛。C-S模式通过服务器协调，虽增加部分能耗，但显著提升收敛速度。客户端-服务器模式（C-S）和点对点模式（P2P）训练模型的框架图3.1高精度轻量化模型C-S模式：服务器居中协调，收敛快，易于管理。P2P模式：完全分布式，灵活但收敛慢，协调开销大

3.1高精度轻量化模型

3.1高精度轻量化模型

3.1高精度轻量化模型结合上述所得公式和逆幂律可以得到总体优化目标：在该方法中，我们要求C-S模式下的时延和能耗优于P2P模式，以此为约束条件，最大化模型精度，在联邦协同训练中得到轻量化模型。在设备中部署的模型具有快响应和低能耗的特点，考虑了联邦训练中系统的计算和通信等的动态变化性，提高了资源的利用率。3.1高精度轻量化模型在该方法中，我们要求C-S模式下的时延和能耗优于P2P模式，以此为约束条件，最大化模型精度，在联邦协同训练中得到轻量化模型。在设备中部署的模型具有快响应和低能耗的特点，考虑了联邦训练中系统的计算和通信等的动态变化性，提高了资源的利用率。3.2个性化轻量化模型轻量个性化模型的训练流程图采用ADMM算法实现稀疏约束下的权重剪枝引入掩码机制(Mask)实现个性化建模不同客户端传输全局模型的子网（而非完整模型）用户可根据设备性能选择剪枝率，实现灵活部署优势：轻量高效+个性化+隐私增强3.2个性化轻量化模型三变量交替迭代优化（更新w,z,λ）剪枝后参数低于阈值→置零并冻结用户可控制剪枝率，平衡精度与计算量。稀疏约束下的损失函数：指示函数引入稀疏正则项：构造增广拉格朗日函数：ADMM剪枝算法原理：3.2个性化轻量化模型剪枝流程：前提条件：完成ADMM稀疏训练，使不重要的参数趋近于零。精度保护：仅在当前精度高于阈值且未达目标剪枝率时执行。执行操作：将低于阈值的参数置零并永久冻结。如图所示，剪枝过程通过移除冗余的突触连接乃至整个神经元，实现从密集网络到稀疏化网络的高效转化。这一过程既保障了模型精度不受永久性损伤，又通过减少可训练参数量显著降低了计算负载，最终生成高度精简的模型完美适配资源受限的边缘设备。3.3边缘智能实验平台平台架构与部署流程模型转换与Android端部署三种模型性能对比验证离线推理与实时响应测试实验结果与分析(Federatedlearning)目录4.1实验参数设置014.2实验源代码说明024.3实验结果034.1实验参数设置参数取值客户端总数N2000.05、0.075、0.1230、40、5010%10%-90%4.2实验源代码说明

以CIFAR-10数据集为例，共包括12个内容：为客户端划分非独立同分布数据、初始化掩码、获取全局模型的子网、训练模型、更新ADMM算法中的X、更新ADMM算法中的Z、更新ADMM算法中的U、ADMM损失函数、判断是否需要剪枝、剪枝、服务器聚合非0值、计算模型准确率和损失值。4.2实验源代码说明为客户端划分非独立同分布数据defcifar_extr_noniid(train_dataset,test_dataset,num_users,n_class,num_samples,rate_unbalance):（sampling.py189）#把CIFAR-10训练集中的50000个数据划分为num_shards_train个切片

num_shards_train,num_imgs_train=int(50000/num_samples),num_samples

#CIFAR-10包括10种数据

num_classes=10#CIFAR-10测试集中每个种类包括1000个数据，10个种类共有10000个数据

num_imgs_perc_test,num_imgs_test_total=1000,10000#防止为客户端分配的数据超过总的数据量

assert(n_class*num_users<=num_shards_train)

assert(n_class<=num_classes)

4.2实验源代码说明#初始化

idx_class

=[i

fori

inrange(num_classes)]

idx_shard=[i

fori

inrange(num_shards_train)]

dict_users_train={i:np.array([])fori

inrange(num_users)}

dict_users_test={i:np.array([])fori

inrange(num_users)}

idxs=np.arange(num_shards_train*num_imgs_train)labels=np.array(train_dataset.targets)

idxs_test=np.arange(num_imgs_test_total)

labels_test=np.array(test_dataset.targets)#按照标签为训练集排序idxs_labels=np.vstack((idxs,labels))

idxs_labels=idxs_labels[:,idxs_labels[1,:].argsort()]

idxs=idxs_labels[0,:]

labels=idxs_labels[1,:]#按照标签为测试集排序

idxs_labels_test=np.vstack((idxs_test,labels_test))

idxs_labels_test=idxs_labels_test[:,idxs_labels_test[1,:].argsort()]

idxs_test=idxs_labels_test[0,:]

#为参与训练的num_users个客户端分配数据fori

inrange(num_users):

user_labels=np.array([])

#为客户端i随机分配n_class个切片rand_set=set(np.random.choice(idx_shard,n_class,replace=False))

#从总切片中减去已经分配过的切片，防止为客户端重复分配数据idx_shard=list(set(idx_shard)-rand_set)

unbalance_flag=0

4.2实验源代码说明#在客户端i中添加切片所对应的训练集数据forrandinrand_set:

ifunbalance_flag==0:

dict_users_train[i]=np.concatenate((dict_users_train[i],idxs[rand*num_imgs_train:(rand+1)*num_imgs_train]),axis=0)

user_labels=np.concatenate((user_labels,labels[rand*num_imgs_train:(rand+1)*num_imgs_train]),axis=0)

else:

dict_users_train[i]=np.concatenate((dict_users_train[i],idxs[rand*num_imgs_train:int((rand+rate_unbalance)*num_imgs_train)]),axis=0)user_labels=np.concatenate((user_labels,labels[rand*num_imgs_train:int((rand+rate_unbalance)*num_imgs_train)]),axis=0)unbalance_flag=1

user_labels_set=set(user_labels)

#在客户端i中添加切片所对应的测试集数据forlabelinuser_labels_set:

dict_users_test[i]=np.concatenate((dict_users_test[i],idxs_test[int(label)*num_imgs_perc_test:int(label+1)*num_imgs_perc_test]),axis=0)returndict_users_train,dict_users_test

4.2实验源代码说明初始化掩码defmake_mask(model):(utils.py209)

step=0

forname,paramind_parameters():

if'weight'inname:

step=step+1

#掩码的个数与模型参数的个数相同

mask=[None]*step

step=0

#把掩码初始化为1forname,paramind_parameters():

if'weight'inname:

tensor=param.data.cpu().numpy()

mask[step]=np.ones_like(tensor)

step=step+1

returnmask

4.2实验源代码说明获取全局模型的子网defmask_model(model,mask,initial_state_dict):(274)

step=0

forname,paramind_parameters():

if"weight"inname:

weight_dev=param.device

#将掩码与模型参数对应位置相乘，得到全局模型的子网param.data=torch.from_numpy(mask[step]*initial_state_dict[name].cpu().numpy()).to(weight_dev)step=step+1if"bias"inname:

param.data=initial_state_dict[name]

4.2实验源代码说明训练模型defupdate_weights(self,model,global_round,device):(update.py78)

EPS=1e-6#把模型切换为训练模式

model.train()

epoch_loss=[]#设置优化器

ifself.args.optimizer=='sgd':

optimizer=SGD(d_parameters(),lr=1e-2,momentum=0.5,

weight_decay=5e-4)

elif

self.args.optimizer=='adam':

optimizer=torch.optim.Adam(model.parameters(),lr=self.args.lr,

weight_decay=1e-4)

#初始化ADMM算法中的Z和UZ,U=initialize_Z_and_U(model)

4.2实验源代码说明#在客户端本地训练模型foriter

inrange(self.args.local_ep):

batch_loss=[]

forbatch_idx,(images,labels)inenumerate(self.trainloader):

images,labels=images.to(self.device),labels.to(self.device)#梯度清零

model.zero_grad()

#得到模型对数据images的预测结果log_probslog_probs=model(images)#调用admm_loss函数来计算损失

loss=admm_loss(device,model,Z,U,log_probs,labels)

#反向传播loss.backward()#冻结已修剪过的权重forname,pind_parameters():

if'weight'inname:

tensor=p.data.cpu().numpy()

grad_tensor=p.grad.data.cpu().numpy()

grad_tensor=np.where(abs(tensor)<EPS,0,grad_tensor)

p.grad.data=torch.from_numpy(grad_tensor).to(device)

4.2实验源代码说明

#更新优化器optimizer.step()

batch_loss.append(loss.item())

epoch_loss.append(sum(batch_loss)/len(batch_loss))

#调用update_X,来更新ADMM算法中的XX=update_X(model)#调用update_Z,来更新ADMM算法中的Z

Z=update_Z(X,U)#调用update_U,更新ADMM算法中的U

U=update_U(U,X,Z)

returnmodel.state_dict(),sum(epoch_loss)/len(epoch_loss)

4.2实验源代码说明更新ADMM算法中的Xdefupdate_X(model):(admm.py31)

X=()

forname,paramind_parameters():

ifname.split('.')[-1]=="weight":

X+=(param.detach().cpu().clone(),)

returnX

更新ADMM算法中的Zdefupdate_Z(X,U):

new_Z=()

idx=0

forx,uinzip(X,U):

z=x+u#计算Z的阈值

pcen=np.percentile(abs(z),10)

under_threshold=abs(z)<pcen

#若Z的值小于阈值，则把Z设为0z.data[under_threshold]=0

new_Z+=(z,)

idx+=1

returnnew_Z

4.2实验源代码说明更新ADMM算法中的Udefupdate_U(U,X,Z):

new_U=()

foru,x,zinzip(U,X,Z):

new_u=u+1e-4*(x-z)

new_U+=(new_u,)

returnnew_U

ADMM损失函数defadmm_loss(device,model,Z,U,output,target):

idx=0

loss=F.nll_loss(output,target)

forname,paramind_parameters():

ifname.split('.')[-1]=="weight":

u=U[idx].to(device)

z=Z[idx].to(device)#计算ADMM损失

loss+=1e-4/2*(param-z+u).norm()idx+=1

returnloss

4.2实验源代码说明判断是否需要剪枝#若当前精度高于精度阈值且当前剪枝率未达到目标剪枝率，则剪枝if(acc_beforePrune>args.prune_start_acc

andpruning_rate[idx]>args.prune_end_rate):(pruneFL.py175)#调用prune_by_percentile函数，剪枝

prune_by_percentile(train_model,masks[idx],args.prune_percent)#计算当前剪枝率

pruning_rate[idx]=pruning_rate[idx]*(1-args.prune_percent/100)

剪枝defprune_by_percentile(model,mask,percent,resample=False,reinit=False,**kwargs):(utils.py243)

step=0

forname,paramind_parameters():

if'weight'inname:tensor=param.data.cpu().numpy()

#展开非0值alive=tensor[np.nonzero(tensor)]#计算模型参数的剪枝阈值percentile_value=np.percentile(abs(alive),percent)

weight_dev=param.device

#把小于剪枝阈值的模型参数对应的掩码置为0new_mask=np.where(abs(tensor)<percentile_value,0,mask[step])#把掩码为0对应的参数置为0

param.data=torch.from_numpy(tensor*new_mask).to(weight_dev)

mask[step]=new_mask

step+=1

4.2实验源代码说明服务器聚合非0值defaverage_weights_with_masks(w,masks,device):(utils.py158)

step=0

w_avg=copy.deepcopy(w[0])

forkeyinw_avg.keys():

if'weight'inkey:

mask=masks[0][step]

fori

inrange(1,len(w)):

w_avg[key]+=w[i][key]#把该参数所对应的非0掩码相加，判断没有对该参数进行剪枝的客户端数量

mask+=masks[i][step]#只聚合未剪枝的参数w_avg[key]=torch.from_numpy(np.where(mask<1,0,w_avg[key].cpu().numpy()/mask)).to(device)

step+=1else:

fori

inrange(1,len(w)):

w_avg[key]+=w[i][key]

w_avg[key]=torch.div(w_avg[key],len(w))

returnw_avg

4.2实验源代码说明计算模型准确率和损失值definference(self,model):(update.py140)

#把模型切换为评估模式model.eval()

loss,total,correct=0.0,0.0,0.0

forbatch_idx,(images,labels)inenumerate(self.testloader):

images,labels=images.to(self.device),labels.to(self.device)#输出模型对图片images的预测分数

outputs=model(images)#计算模型的输出分数outputs和图片的真实标签labels之间的损失

batch_loss=F.nll_loss(outputs,labels)

loss+=batch_loss.item()#得到模型的预测标签

_,pred_labels=torch.max(outputs,1)

pred_labels=pred_labels.view(-1)

#计算预测正确的次数correct+=torch.sum(torch.e