11月份ai智慧交通训练营

1

我们直接看一下部署的效果:

回顾开发过程知识：

1 常用AI对抗攻击算法讲解

对抗攻击（adversarial attack）是AI安全方向的重要分支，其核心逻辑是在数据中增加一些微小扰动，在人类视觉系统无法察觉的情况下，使得算法模型对这些数据产生误判。其中被增加扰动的数据也称为对抗样本。

下面是一些直观的通过增加对抗噪声来让算法模型误判的例子：

在上图的熊猫图片中加入微小的对抗噪声，就能让AI模型将其误判为长臂猿。

下图是之前第一节课和第二节课带大家认识的不同对抗攻击逻辑，大家可以看图温习一下，也可以对照着之前的课件回顾一下：

我们该怎么去理解对抗样本呢？总的来说，我把它当作生成噪声的艺术，其攻击价值的底层依托是算法模型的过拟合问题。

更多攻击算法的介绍可以查阅以下的相关材料：

https://zhuanlan.zhihu.com/p/493333024

2 常用AI对抗防御算法讲解

在对抗攻击技术出现之后，对抗防御技术也孕育而生。对抗防御一般针对对抗攻击技术而展开，有针对特定攻击的特定防御，也有一般性防御策略，下面我们就一起学习一下。

2.1 对抗防御算法的场景应用

在智慧交通场景中，常用的对抗防御方法有与对抗攻击结合紧密的AI业务的鲁棒性与安全性检查工具；研究对抗攻击，来增强对实际场景中受到的攻击情形的判断力；使用对抗攻击对敏感信息进行隐藏；使用对抗样本在模型训练过程中进行对抗训练。

当然的，除了上面讲到的与对抗攻击结合的形式，还包括在算法解决方案中加入防御预处理，设计鲁棒模型结构，加入对抗攻击检测模型等方法。

由于智慧交通场景的算法解决方案对耗时和内存占用有较高的要求，所以防御预处理与防御后处理作为常规防御的首要选择。

同时，针对可能存在的安全风险，在开发阶段，设计鲁棒性的模型结构，提升算法解决方案的整体安全性。

或者训练轻量级的对抗攻击检测模型，作为算法解决方案的安全模块，在受到攻击风险时，启动防御算法功能。

具体来说，我们需要生成一些对抗样本，并将其和干净样本一同作为训练集，训练一个能够捕捉对抗扰动的二分类模型，并将其作为实际AI项目中的一个前置防御模块。

2.2 常用AI对抗防御算法划分

上面我们了解了对抗防御的场景应用，接下来我们再了解一下目前主流对抗防御的总体分支与逻辑：

其中对抗训练是指在训练过程中加入对抗样本，通过不断的学习对抗样本的特征，从而提升模型的鲁棒性。

监测识别对抗样本顾名思义，在项目关键节点处，设置一些能够识别对抗样本的特种模型，从而提前预警对抗攻击风险。

模型鲁棒结构设计是指在模型中设计特定的滤波结构能够一定程度上增强模型鲁棒性，抵御对抗噪声。

对抗扰动结构破坏主要在数据流处理的时候使用，通过一些滤波算法，噪声结构破坏算法，噪声覆盖算法等策略，减弱对抗噪声的影响，使其不能对模型造成攻击。

梯度掩膜则是在白盒对抗防御中非常高效的一种算法，因为其能掩盖真实梯度，从而能够使得白盒攻击算法失效。

本章中在后面小节使用到的对抗防御方法主要是基于梯度掩膜的GCM模块。

GCM模块通过改造输入，使模型在前向测试时并不影响原来的结果，但是在进行梯度反向传播时，能隐藏原来的梯度，使得白盒攻击无所适从。

下图是加入GCM模块后的模型梯度反向传播图示：

训练营大作业：

利用yolov5检测出来的车辆检测结果进行对抗样本的生成，将对抗样本传进模型，通过ai检测模型可以判断这张图片是否带有恶意，以提高鲁棒性，进行一个告警功能的实现。

  1. 第一步：车辆检测，生成检测框里的车图片

首先找一张有关车的图片，我们通过yolov5找到图片中车的具体位置，再通过相关函数进行裁剪，提取车辆目标区域

我们试着利用模型判断这个车的类型

  1. 生成对抗样本

可以看到加入了很多的噪声

可以观察到模型输出的类型与之前不同，可以知道这对抗样本干扰到模型的准确性。

  1. 对抗攻击检测模型

对抗攻击检测模型是一个能够捕捉对抗扰动的二分类模型，这里使用一个已经训练好的基于ResNet50的模型，作为检测模型。再通过喵提醒将检测模型的告警结果发送到手机公众号中。

  1. 完整流程串联：

车辆检测+检测框提取+对抗样本+ai安全检测警告功能。

判断是否进行攻击，如果是，则通过公众号进行消息提醒

看一下整个实现算法：

import os
import torch
import requests
import time
import torch.nn as nn
from torchvision.models import mobilenet_v2,resnet18
from advertorch.utils import predict_from_logits
from advertorch.utils import NormalizeByChannelMeanStd
from robust_layer import GradientConcealment, ResizedPaddingLayer
from timm.models import create_model
from advertorch.attacks import LinfPGDAttack
from advertorch_examples.utils import ImageNetClassNameLookup
from advertorch_examples.utils import bhwc2bchw
from advertorch_examples.utils import bchw2bhwc
import torchvision.utils
from cvs import *
import cvs
import aidlite_gpu
from aidlux.utils import detect_postprocess, preprocess_img, draw_detect_res, extract_detect_res
import cv2

device = "cuda" if torch.cuda.is_available() else "cpu"


#读取图片

def get_image():

img_path = os.path.join("/home/carJiance/Lesson5_code/adv_code/duikangpicture", "vid_5_27620.jpg_0.jpg")

img_url = "https://farm1.static.flickr.com/230/524562325_fb0a11d1e1.jpg"


def _load_image():
    from skimage.io import imread
    return imread(img_path) / 255.

if os.path.exists(img_path):
    return _load_image()
else:
    import urllib
    urllib.request.urlretrieve(img_url, img_path)
    return _load_image()



##模型加载

normalize = NormalizeByChannelMeanStd(

mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])


常规模型加载


model = mobilenet_v2(pretrained=True)

model.eval()

model = nn.Sequential(normalize, model)

model = model.to(device)


替身模型加载


model_su = resnet18(pretrained=True)

model_su.eval()

model_su = nn.Sequential(normalize, model_su)

model_su = model_su.to(device)


def tensor2npimg(tensor):

return bchw2bhwc(tensor[0].cpu().numpy())


常规模型加载


class Model(nn.Module):

def init(self, l=290):

super(Model, self).init()


    self.l = l
    self.gcm = GradientConcealment()
    #model = resnet18(pretrained=True)
    model = mobilenet_v2(pretrained=True)

    self.model = nn.Sequential(normalize, model)

def load_params(self):
    pass

def forward(self, x):
    #x = self.gcm(x)
    #x = ResizedPaddingLayer(self.l)(x)
    out = self.model(x)
    return out



对抗攻击监测模型


class Detect_Model(nn.Module):

def init(self, num_classes=2):

super(Detect_Model, self).init()

self.num_classes = num_classes

#model = create_model(‘mobilenetv3_large_075’, pretrained=False, num_classes=num_classes)

model = create_model(‘resnet50’, pretrained=False, num_classes=num_classes)


    # self.multi_PreProcess = multi_PreProcess()
    pth_path = os.path.join("/home/carJiance/Lesson5_code/model", 'track2_resnet50_ANT_best_albation1_64_checkpoint.pth')
    #pth_path = os.path.join("/Users/rocky/Desktop/训练营/Lesson5_code/model/", "track2_tf_mobilenetv3_large_075_64_checkpoint.pth")
    state_dict = torch.load(pth_path, map_location='cpu')
    is_strict = False
    if 'model' in state_dict.keys():
        model.load_state_dict(state_dict['model'], strict=is_strict)
    else:
        model.load_state_dict(state_dict, strict=is_strict)
    
    # self.model = nn.Sequential(normalize, self.multi_PreProcess, model)
    self.model = nn.Sequential(normalize, model)

def load_params(self):
    pass

def forward(self, x):
    # x = x[:,:,32:193,32:193]
    # x = F.interpolate(x, size=(224,224), mode="bilinear", align_corners=True)
    # x = self.multi_PreProcess.forward(x)
    out = self.model(x)
    if self.num_classes == 2:
        out = out.softmax(1)
        #return out[:,1:]
        return out[:,1:]



model = Model().eval().to(device)


detect_model = Detect_Model().eval().to(device)


def attack():

### 数据预处理

np_img = get_image()

img = torch.tensor(bhwc2bchw(np_img))[None, :, :, :].float().to(device)

imagenet_label2classname = ImageNetClassNameLookup()


### 测试模型输出结果
pred = imagenet_label2classname(predict_from_logits(model(img)))
print("车的类型:", pred)

### 输出原label
pred_label = predict_from_logits(model_su(img))

### 对抗攻击：PGD攻击算法
adversary = LinfPGDAttack(
model_su, eps=8/255, eps_iter=2/255, nb_iter=80,
rand_init=True, targeted=False)

### 完成攻击，输出对抗样本
advimg = adversary.perturb(img, pred_label)
return advimg



AidLite初始化：调用AidLite进行AI模型的加载与推理，需导入aidlite


aidlite = aidlite_gpu.aidlite()


Aidlite模型路径


model_path = ‘/home/carJiance/Lesson5_code/yolov5_code/models/yolov5_car_best-fp16.tflite’


定义输入输出shape


in_shape = [1 * 640 * 640 * 3 * 4]

out_shape = [1 * 25200 * 6 * 4]


加载Aidlite检测模型：支持tflite, tnn, mnn, ms, nb格式的模型加载


aidlite.ANNModel(model_path, in_shape, out_shape, 4, 0)


def main():

# 读取图片进行推理

# 设置测试集路径

source = "/home/carJiance/Lesson5_code/adv_code/test_images"

images_list = os.listdir(source)

print(images_list)

frame_id = 0

# 读取数据集

for image_name in images_list:

frame_id += 1

print("frame_id:", frame_id)

image_path = os.path.join(source, image_name)

frame = cv2.imread(image_path)

print("数据预处理开始")

# 预处理

img = preprocess_img(frame, target_shape=(640, 640), div_num=255, means=None, stds=None)

# 数据转换：因为setTensor_Fp32()需要的是float32类型的数据，所以送入的input的数据需为float32,大多数的开发者都会忘记将图像的数据类型转换为float32

aidlite.setInput_Float32(img, 640, 640)

# 模型推理API

aidlite.invoke()

# 读取返回的结果

pred = aidlite.getOutput_Float32(0)


    # 数据维度转换
    pred = pred.reshape(1, 25200, 6)[0]
    # 模型推理后处理
    pred = detect_postprocess(pred, frame.shape, [640, 640, 3], conf_thres=0.25, iou_thres=0.45)
    # 绘制推理结果
    res_img = draw_detect_res(frame, pred)

    # 图片裁剪，提取车辆目标区域
    extract_detect_res(frame, pred, image_name)
    print("图片裁剪完成")
    ### 数据预处理
    np_img = get_image()
    print("2")
    img = torch.tensor(bhwc2bchw(np_img))[None, :, :, :].float().to(device)
    print("3")
    imagenet_label2classname = ImageNetClassNameLookup()
    print("测试模型输出结果")
    ### 测试模型输出结果
    pred = imagenet_label2classname(predict_from_logits(model(img)))
    print("输出原label")
    ### 输出原label
    pred_label = predict_from_logits(model_su(img))
    print("进行输出对抗样本")
    ### 对抗攻击：PGD攻击算法
    adversary = LinfPGDAttack(
    model_su, eps=8/255, eps_iter=2/255, nb_iter=80,
    rand_init=True, targeted=False)

    ### 完成攻击，输出对抗样本
    advimg = adversary.perturb(img, pred_label)
    print("完成输出对抗样本")
    print("对抗攻击检测")
    ### 对抗攻击监测
    detect_pred = detect_model(advimg)
    if detect_pred > 0.5:
        id = 'tLCW5i9'
        # 填写喵提醒中，发送的消息，这里放上前面提到的图片外链
        text = "出现对抗攻击风险！！"
        ts = str(time.time())  # 时间戳
        type = 'json'  # 返回内容格式
        request_url = "http://miaotixing.com/trigger?"

        headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/87.0.4280.67 Safari/537.36 Edg/87.0.664.47'}

        result = requests.post(request_url + "id=" + id + "&text=" + text + "&ts=" + ts + "&type=" + type,
                            headers=headers)
    else:
        pred = imagenet_label2classname(predict_from_logits(model(img)))
        print("车的类型为："+pred)
        
print("结束")



if name=="main":

print("流程开始")

main()

print("流程结束")

2

想去看看呢