這是與人工智能實(shí)施相關(guān)的系列博客文章的一部分。如果您對(duì)故事的背景或進(jìn)展感興趣：

前幾周，我們討論了如何利用SerpApi的Google Images Scraper API來自動(dòng)構(gòu)建個(gè)人圖像數(shù)據(jù)集。本周，我們將使用這些圖像，并通過簡單的命令對(duì)象自動(dòng)訓(xùn)練一個(gè)網(wǎng)絡(luò)模型，隨后將其集成到FastAPI中。

自定義 CSV

為了這一目的，我們需要準(zhǔn)備一個(gè)包含所需圖像的自定義CSV文件。為此，我們將借助pandas庫。以下是相關(guān)要求：

## create.py

from pydantic import BaseModel

from typing import List

import pandas as pd

import os

我們需要?jiǎng)?chuàng)建一個(gè)從 SerpApi 的 Google Image Scraper API 收集的項(xiàng)目列表，設(shè)置要?jiǎng)?chuàng)建的 csv 文檔的名稱，并為訓(xùn)練數(shù)據(jù)定義一個(gè)分?jǐn)?shù)。這里的分?jǐn)?shù)概念很簡單：TestData（測(cè)試數(shù)據(jù)）將包含我們收集的所有圖像，而 TrainingData（訓(xùn)練數(shù)據(jù)）則只會(huì)從中選取一小部分圖像。為了這些目的，我們需要構(gòu)建一個(gè)可以傳遞給相關(guān)端點(diǎn)的對(duì)象。

class ClassificationsArray(BaseModel):

  file_name: str

  classifications_array: List[str]

  train_data_fraction: float

這里提到的分?jǐn)?shù)有一個(gè)簡單的解釋。測(cè)試數(shù)據(jù)集將包含所有圖像，而訓(xùn)練數(shù)據(jù)集將只包含其中的一小部分。這是為了在我們使用尚未訓(xùn)練的圖像訓(xùn)練模型后測(cè)試模型，即測(cè)試數(shù)據(jù)集的差異。
現(xiàn)在我們已經(jīng)定義了負(fù)責(zé)命令的對(duì)象，讓我們定義 CSVCreator 類：

class CSVCreator:

  def __init__(self, ClassificationsArray):

    self.classifications_array = ClassificationsArray.classifications_array

    self.file_name = ClassificationsArray.file_name

    self.rows = []

    self.train_data_fraction = ClassificationsArray.train_data_fraction

  

  def gather(self):

    for label in self.classifications_array:

      images = os.listdir("datasets/test/{}".format(label))

      for image in images:

        row = ["datasets/test/{}/{}".format(label, image), label]

        self.rows.append(row)



  def create(self):

    df = pd.DataFrame(self.rows, columns = ['path', 'label'])

    df.to_csv("datasets/csv/{}.csv".format(self.file_name), index=False)

    train_df = df.sample(frac = self.train_data_fraction)

    train_df.to_csv("datasets/csv/{}_train.csv".format(self.file_name), index=False)

它接收我們提供的參數(shù)列表，這些參數(shù)是我們對(duì)SerpApi的Google Images Scraper API進(jìn)行查詢時(shí)所使用的?；谶@些參數(shù)，它會(huì)從相應(yīng)文件夾中的每張圖像生成一個(gè)CSV文件。在所有圖像處理完畢后，它會(huì)隨機(jī)選取一部分樣本，并創(chuàng)建一個(gè)用于訓(xùn)練的CSV文件。

現(xiàn)在，讓我們?cè)?code>main.py中定義一個(gè)函數(shù)來執(zhí)行這一操作。

## main.py

from create import CSVCreator, ClassificationsArray

這些類是執(zhí)行該操作所必需的，而負(fù)責(zé)具體操作的函數(shù)位于 main.py 中。

@app.post("/create/")

def create_csv(arr: ClassificationsArray):

  csv = CSVCreator(arr)

  csv.gather()

  csv.create()

  return {"status": "Complete"}

舉個(gè)直觀的例子，如果你前往并使用以下參數(shù)進(jìn)行嘗試：http://localhost:8000/docs/create/

{

  "file_name": "apples_and_oranges",

  "classifications_array": [

    "Apple",

    "Orange"

  ],

  "train_data_fraction": 0.8

}

您將在 called 和 中創(chuàng)建兩個(gè) csv 文件datasets/csvapples_and_oranges.csvapples_and_oranges_train.csv

apples_and_oranges.csv將是測(cè)試 CSV，將被排序，將包含所有圖像，如下所示：

“apples_and_oranges_train.csv” 將作為訓(xùn)練用的 CSV 文件，其內(nèi)容將被隨機(jī)排列，并且會(huì)包含 80% 的圖像，具體細(xì)節(jié)如下：

這兩個(gè)項(xiàng)目將用于創(chuàng)建 Dataset 項(xiàng)。

自定義訓(xùn)練命令

我們需要一個(gè)對(duì)象來指定訓(xùn)練操作的詳細(xì)信息，并在多個(gè)類之間共享使用以避免循環(huán)導(dǎo)入：

## commands.py

from pydantic import BaseModel



class TrainCommands(BaseModel):

	model_name: str = "apples_and_oranges"

	criterion: str = "CrossEntropyLoss"

	annotations_file: str = "apples_and_oranges"

	optimizer: str = "SGD"

	lr: float = 0.001

	momentum: float = 0.9

	batch_size: int = 4

	n_epoch: int = 2

	n_labels: int = None

	image_height: int = 500

	image_width: int = 500

	transform: bool = True

	target_transform: bool = True

	shuffle: bool = True

讓我們分解此對(duì)象中的項(xiàng)：

僅僅固定圖像的高度和寬度是不夠的，因?yàn)檫@樣做可能會(huì)導(dǎo)致圖像失真。本周，我們不會(huì)實(shí)施任何去噪的變換操作，但這類操作在批量加載時(shí)是必要的，因?yàn)榕恐械膹埩繄D像需要保持相同的大小。

自定義數(shù)據(jù)集和自定義數(shù)據(jù)加載器

現(xiàn)在我們已經(jīng)有了所需的命令，讓我們開始了解創(chuàng)建 dataset 和 dataloader 的要求：

## dataset.py

import os

import pandas as pd

import numpy as np

from PIL import Image

from torchvision import transforms

from torch.utils.data import Dataset, DataLoader

from commands import TrainCommands

然后讓我們初始化我們的 dataset 類：

class CustomImageDataset(Dataset):

	def __init__(self, tc: TrainCommands, type: str):

		transform = tc.transform

		target_transform = tc.target_transform

		annotations_file = tc.annotations_file

		self.image_height = tc.image_height

		self.image_width = tc.image_width

		if type == "train":

			annotations_file = "{}_train".format(annotations_file)

		self.img_labels = pd.read_csv("datasets/csv/{}.csv".format(annotations_file))

		unique_labels = list(set(self.img_labels['label'].to_list()))

		tc.n_labels = len(unique_labels)

		dict_labels = {}

		for label in unique_labels:

			dict_labels[label] = unique_labels.index(label)

		self.dict_labels = dict_labels

		

		if transform == True:

			self.transform = transforms.Compose([

				transforms.ToTensor(),

				transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5)),

			])

		else:

			self.transform == False

		if target_transform == True:

			self.target_transform = transforms.Compose([

				transforms.ToTensor(),

			])

		else:

			self.transform == False

我們使用參數(shù)（parameter）來定義我們的操作目標(biāo)：是初始化數(shù)據(jù)庫，還是指定數(shù)據(jù)庫具有的類型（如 train 或 test）。

		if type == "train":

			annotations_file = "{}_train".format(annotations_file)

    self.img_labels = pd.read_csv("datasets/csv/{}.csv".format(annotations_file))

要定義要在模型整形中使用的標(biāo)簽列表，我們使用以下行：

		unique_labels = list(set(self.img_labels['label'].to_list()))

		tc.n_labels = len(unique_labels)

		dict_labels = {}

		for label in unique_labels:

			dict_labels[label] = unique_labels.index(label)

		self.dict_labels = dict_labels

這為我們提供了一個(gè)要分類的項(xiàng)目的字典，每個(gè)項(xiàng)目都有自己唯一的整數(shù)：

## self.dict_labels

{

  "Apple": 0,

  "Orange": 1

}

我們必須為 input 和 label 定義某些轉(zhuǎn)換。這些轉(zhuǎn)換定義了如何將它們轉(zhuǎn)換為用于訓(xùn)練的張量，以及在轉(zhuǎn)換后應(yīng)應(yīng)用哪些操作：

		if transform == True:

			self.transform = transforms.Compose([

				transforms.ToTensor(),

				transforms.Normalize((0.5, 0.5, 0.5), (0.5, 0.5, 0.5)),

			])

		else:

			self.transform == False

		if target_transform == True:

			self.target_transform = transforms.Compose([

				transforms.ToTensor(),

			])

		else:

			self.transform == False

我們還定義一個(gè)函數(shù)，為我們提供給定數(shù)據(jù)集中的圖像數(shù)量：

	def __len__(self):

			return len(self.img_labels)

最后，我們需要定義要給出的內(nèi)容，一個(gè) image 的張量和一個(gè) label 的張量：

	def __getitem__(self, idx):

		img_path = os.path.join(self.img_labels.iloc[idx, 0])

		label = self.img_labels.iloc[idx, 1]

		label = self.dict_labels[label]

		label_arr = np.full((len(self.dict_labels), 1), 0, dtype=float) #[0.,0.]

		label_arr[label] = 1.0 #[0.,1.]

		image = Image.open(img_path).convert('RGB')

		image = image.resize((self.image_height,self.image_width), Image.ANTIALIAS)

		if not self.transform	== False:

			image = self.transform(image)

		if not self.target_transform == False:

			label = self.target_transform(label_arr)

		return image, label

讓我們逐個(gè)部分地分解它。
以下行將獲取具有給定索引的圖像路徑：

img_path = os.path.join(self.img_labels.iloc[idx, 0])

假設(shè)數(shù)據(jù)集是訓(xùn)練數(shù)據(jù)集，索引為 0：

路徑包含的原因是我們目前將所有文件都保存在同一目錄中。該圖像數(shù)據(jù)來源于一個(gè)dataframe，因此不會(huì)引發(fā)問題。接下來的幾行代碼將從相關(guān)標(biāo)簽的索引中創(chuàng)建一個(gè)one-hot向量，例如使用self.img_labels.iloc[0, 0]

		label = self.img_labels.iloc[idx, 1]

		label = self.dict_labels[label]

		label_arr = np.full((len(self.dict_labels), 1), 0, dtype=float) #[0.,0.]

		label_arr[label] = 1.0 #[0.,1.]

我進(jìn)行評(píng)論的原因是，在我們的例子中存在兩個(gè)標(biāo)簽，即“Apple”和“Orange”，因此one-hot向量的大小將根據(jù)這兩個(gè)標(biāo)簽來定義。接下來，我們會(huì)將這個(gè)向量轉(zhuǎn)換為numpy數(shù)組，以便進(jìn)一步轉(zhuǎn)換為張量。

		image = Image.open(img_path).convert('RGB')

		image = image.resize((self.image_height,self.image_width), Image.ANTIALIAS)

我們將圖像轉(zhuǎn)換為 RGB 格式，以獲得一個(gè)包含三個(gè)維度的向量，其中第三個(gè)維度代表顏色。接著，我們使用 ANTIALIAS 方法對(duì)圖像進(jìn)行大小調(diào)整，以確保調(diào)整后的圖像仍然能夠被人類視覺所識(shí)別。盡管我之前提到過，這樣的處理通常并不足夠，但目前我們就按照這種方式來進(jìn)行。

現(xiàn)在是自定義數(shù)據(jù)加載器：

class CustomImageDataLoader:

	def __init__(self, tc: TrainCommands, cid: CustomImageDataset):

		batch_size = tc.batch_size

		train_data = cid(tc, type = "train")

		test_data =  cid(tc, type = "test")

		self.train_dataloader = DataLoader(train_data, batch_size = tc.batch_size, shuffle = tc.shuffle)

		self.test_dataloader = DataLoader(test_data, batch_size = batch_size, shuffle = tc.shuffle)

	

	def iterate_training(self):

		train_features, train_labels = next(iter(self.train_dataloader))

		print(f"Feature batch shape: {train_features.size()}")

		print(f"Ladabels batch shape: {train_labels.size()}")

		return train_features, train_labels

如前所述，我們首先使用參數(shù)進(jìn)行初始化，并在其中定義了數(shù)據(jù)集。接著，我們利用Pytorch的DataLoader函數(shù)來聲明一個(gè)數(shù)據(jù)加載器。在調(diào)用數(shù)據(jù)加載器時(shí)，我們通過batch size參數(shù)來指定每次迭代中要獲取的圖像數(shù)量。

在迭代過程中，我們會(huì)使用在初始化階段定義的自定義圖像數(shù)據(jù)集（Custom Image Dataset），該數(shù)據(jù)集為我們提供圖像及其對(duì)應(yīng)的標(biāo)簽。self.train_dataloader和self.test_dataloader分別代表訓(xùn)練集和測(cè)試集的數(shù)據(jù)加載器。當(dāng)我們從數(shù)據(jù)集中獲取一批圖像時(shí)，train_features將表示這批圖像的張量，而train_labels則代表這些圖像對(duì)應(yīng)的標(biāo)簽的張量。

自動(dòng)訓(xùn)練

現(xiàn)在我們已經(jīng)擁有了一個(gè)自定義的圖像數(shù)據(jù)集以及一個(gè)用于從數(shù)據(jù)集中加載圖像的自定義數(shù)據(jù)加載器。接下來，我們將使用object來進(jìn)行自動(dòng)訓(xùn)練。訓(xùn)練類和模型的要求是：TrainCommands

# train.py

import torch

import torch.nn as nn

import torch.nn.functional as F

from dataset import CustomImageDataLoader, CustomImageDataset

from commands import TrainCommands

我們還聲明我們要使用的 CNN 模型：

class CNN(nn.Module):

	def __init__(self, tc: TrainCommands):

		super().__init__()

		n_labels = tc.n_labels

		self.conv1 = nn.Conv2d(3, 6, 5)

		self.pool = nn.MaxPool2d(2, 2)

		self.conv2 = nn.Conv2d(6, 16, 5)

		self.flatten = nn.Flatten(start_dim=1)

		self.fc1 = nn.Linear(16*122*122, 120) # Manually calculated I will explain next week

		self.fc2 = nn.Linear(120, 84)

		self.fc3 = nn.Linear(84, n_labels) #unique label size



	def forward(self, x):

		x = self.pool(F.relu(self.conv1(x)))

		x = self.pool(F.relu(self.conv2(x)))

		x = self.flatten(x)

		x = F.relu(self.fc1(x))

		x = F.relu(self.fc2(x))

		x = self.fc3(x)

		return x

這里需要強(qiáng)調(diào)的一點(diǎn)是，在我們的案例中，輸出大小?n_labels?被設(shè)定為 2，因?yàn)槲覀儍H在兩個(gè)類別間進(jìn)行分類。此外，還有一個(gè)計(jì)算步驟是基于我手動(dòng)確定的圖像嵌入大小以及圖像的高度和寬度?？傮w而言，這是一個(gè)高度通用的圖像分類函數(shù)。在接下來的幾周里，我們將探討如何自動(dòng)完成我原先手動(dòng)進(jìn)行的大小計(jì)算，并研究如何向該函數(shù)中添加更多層，以實(shí)現(xiàn)更進(jìn)一步的自動(dòng)化。
現(xiàn)在，我們來定義一個(gè)訓(xùn)練函數(shù)?AppleOrangeTrainCommands，該函數(shù)將使用自定義的數(shù)據(jù)集和數(shù)據(jù)加載器。

class Train:

	def __init__(self, tc: TrainCommands, cnn: CNN, cidl: CustomImageDataLoader, cid: CustomImageDataset):

		self.loader = cidl(tc, cid)

		self.cnn = cnn(tc)

		self.criterion = getattr(nn, tc.criterion)()

		self.optimizer = getattr(torch.optim, tc.optimizer)(self.cnn.parameters(), lr=tc.lr, momentum=tc.momentum)

		self.n_epoch = tc.n_epoch

		self.model_name = tc.model_name

				



	def train(self):

		for epoch in range(self.n_epoch):  # how many times it'll loop over

			running_loss = 0.0

			for i, data in enumerate(self.loader.train_dataloader):

				inputs, labels = data

				self.optimizer.zero_grad()

				outputs = self.cnn(inputs)

				loss = self.criterion(outputs, labels.squeeze())



				loss.backward()

				self.optimizer.step()

				running_loss = running_loss + loss.item()

				if i % 5 == 4:

					print(

							f'[{epoch + 1}, {i + 1:5d}] loss: {running_loss / 2000:.3f}')

					running_loss = 0.0

		torch.save(self.cnn.state_dict(), "models/{}.pt".format(self.model_name))

讓我們逐個(gè)部分分解它。以下行使用訓(xùn)練命令使用自定義數(shù)據(jù)集初始化自定義訓(xùn)練數(shù)據(jù)集。

self.loader = cidl(tc, cid)

下一行使用訓(xùn)練命令聲明模型 （卷積神經(jīng)網(wǎng)絡(luò)）：

self.cnn = cnn(tc)

以下行負(fù)責(zé)創(chuàng)建標(biāo)準(zhǔn)：

self.criterion = getattr(nn, tc.criterion)()

在我們的例子中，使用torch.nn.CrossEntropyLoss()是等價(jià)的。

下一行用于創(chuàng)建具有所需參數(shù)的優(yōu)化器：

self.optimizer = getattr(torch.optim, tc.optimizer)(self.cnn.parameters(), lr=tc.lr, momentum=tc.momentum)

這相當(dāng)于說，在未來的幾周內(nèi)，我們將開發(fā)一種機(jī)制，允許為可選參數(shù)指定可選的名稱，從而能夠靈活地配置優(yōu)化器和標(biāo)準(zhǔn)。但就目前而言，現(xiàn)有的方法已經(jīng)足夠使用。例如，我們可以使用 torch.optim.SGD(CNN.parameters(), lr=0.001, momentum=0.9) 來初始化一個(gè)帶有特定學(xué)習(xí)率和動(dòng)量的隨機(jī)梯度下降優(yōu)化器。

最后，我們需要設(shè)定要運(yùn)行的訓(xùn)練周期（epoch）數(shù)，并確定保存模型的名稱:

self.n_epoch = tc.n_epoch

self.model_name = tc.model_name

在以下部分中，我們將迭代 epochs，聲明損失，并使用 function 從數(shù)據(jù)集中調(diào)用圖像和標(biāo)簽：

	def train(self):

		for epoch in range(self.n_epoch):  # loop over the dataset multiple times

			running_loss = 0.0

			for i, data in enumerate(self.loader.train_dataloader):

數(shù)據(jù)將以元組形式出現(xiàn)：

inputs, labels = data

然后我們?cè)?optimizer 中將梯度歸零：

self.optimizer.zero_grad()

運(yùn)行預(yù)測(cè)：

outputs = self.cnn(inputs)

然后，使用我們的標(biāo)準(zhǔn)將預(yù)測(cè)與實(shí)際標(biāo)簽進(jìn)行比較，以計(jì)算損失：

loss = self.criterion(outputs, labels.squeeze())

標(biāo)簽在此處被擠壓以匹配要在 criterion function 中計(jì)算的輸入的形狀。

然后我們運(yùn)行反向傳播以自動(dòng)重新累積梯度：

loss.backward()

我們逐步執(zhí)行優(yōu)化器：

self.optimizer.step()

然后更新 ：running_loss

running_loss = running_loss + loss.item()

以下行是每 5 個(gè)步驟的流程輸出：

				if i % 5 == 4:

					print(

							f'[{epoch + 1}, {i + 1:5d}] loss: {running_loss / 2000:.3f}')

					running_loss = 0.0

最后，我們將模型保存到所需的位置：

torch.save(self.cnn.state_dict(), "models/{}.pt".format(self.model_name))

現(xiàn)在我們已經(jīng)萬事俱備，接下來將在main.py中聲明一個(gè)端點(diǎn)，通過這個(gè)端點(diǎn)我們可以接收訓(xùn)練命令。

# main.py

from fastapi import FastAPI

from add import Download, Query

from create import CSVCreator, ClassificationsArray

from dataset import CustomImageDataLoader, CustomImageDataset

from train import CNN, Train

from commands import TrainCommands



app = FastAPI()



@app.get("/")

def read_root():

  return {"Hello": "World"}



@app.post("/add/")

def create_query(query: Query):

  ## Create unique links

  serpapi = Download(query)

  serpapi.download_all_images()

  return  {"status": "Complete"}



@app.post("/create/")

def create_csv(arr: ClassificationsArray):

  csv = CSVCreator(arr)

  csv.gather()

  csv.create()

  return {"status": "Complete"}



@app.post("/train/")

def train(tc: TrainCommands):

  trainer = Train(tc, CNN, CustomImageDataLoader, CustomImageDataset)

  trainer.train()

  return {"status": "Success"}

/train/endpoint 將接收您的命令，并自動(dòng)為您訓(xùn)練一個(gè)模型:

現(xiàn)在，如果您訪問并使用以下路徑嘗試我們的服務(wù)：localhost:8000/docs/train/，并附帶相應(yīng)的參數(shù)。

{

  "model_name": "apples_and_oranges",

  "criterion": "CrossEntropyLoss",

  "annotations_file": "apple_orange",

  "optimizer": "SGD",

  "lr": 0.001,

  "momentum": 0.9,

  "batch_size": 4,

  "n_epoch": 2,

  "n_labels": 0,

  "image_height": 500,

  "image_width": 500,

  "transform": true,

  "target_transform": true,

  "shuffle": true

}

您可以從終端觀察訓(xùn)練過程，因?yàn)槲覀優(yōu)?epochs 聲明了 print 函數(shù)：

訓(xùn)練結(jié)束后，您將在文件夾中保存一個(gè)模型，其中包含您指定的所需名稱：

結(jié)論

我衷心感謝SerpApi的杰出團(tuán)隊(duì)，正是他們的努力使得這篇博文得以問世。同時(shí)，我也非常感謝讀者的關(guān)注與支持。在接下來的幾周里，我們將深入探討如何提升本文提及的某些部分的效率與可定制性。此外，我們還將更全面地討論FastAPI的異步處理機(jī)制，以及如何實(shí)現(xiàn)對(duì)SerpApi的Google Images Scraper API的異步調(diào)用。

原文鏈接：https://serpapi.com/blog/automatic-training-fastapi-pytorch-serpapi/