Привет, чемпионы! Давайте сегодня рассмотрим 5 инструментов, которые стоит применять в своих проектах прямо сейчас и становиться круче. Посмотрим, как улучшить ваш код, чтобы он был без запаха, как сделать ваш pipeline более стабильным и фиксировать выбросы, как не писать один код по 10 раз

Сегодня окунемся в работу с:

• Makefile - проект сам соберется на работу

• Linters - делает джентельменский код

• Lightning - прокачаем PyTorch, забываем про train loop

• DVC - правильно храним данные

• ClearML - настраивает слежку за обучением

? Makefile - твой код заработает на другой машине, прикинь!

Допустим мы создали свой первый пет-проект. И сейчас наша цель, чтобы этот код запустился и у других, а не только у нас. А чтобы его запустить - надо исполнить целую портянку команд: от установки библиотек, внесения в реестр пути для используемых библиотек, до запуска кода... Согласитесь звучит невкусно

Намного приятнее, когда написал одну команду, а все установки и проверки делаются автоматически. Пользователи Linux могут сказать, что пишем start.sh на bash и все работает. Действительно заработает, но только на Linux и MacOS, также достаточно трудно новичку написать bash-скрипт. Самый простой выход из этой ситуации - Makefile

Достаточно создать внутри проекта файлик Makefile, со всеми этапами сборки, а пользователь может выбрать действие и запустить необходимый этап. Не приходится задумываться, что происходит под капотом - откинулись на спинку кресла и ждем пока завершится процесс. То что мы хотели!

# Переменные нужные в процессе сборки
VENV=venv
PYTHON3=$(VENV)/bin/python3

# Название этапа
run_script: build_venv # то что нужно запустить перед исполнением
	# Сами комманды
	@echo "===Устанавливаем пакеты==="
	$(PYTHON3) -m pip install -r requirements.txt
	# ...

То есть вы пишете те же самые команды, но пользователю будет достаточно написать команду make run_scriptв консоле. После чего окружение соберется, и пользователь сможет работать дальше с вашим проектом. Причем все опытные программисты уже знают про эту договоренность и сразу ставят ее без лишних вопросов.

Также правильно написанный Makefile вы можете тянуть в любой свой код, изменяя переменные. Так вы сможете сохранить время, которое можно будет потратить на разработку, а не настройку каких-то пакетов.

?‍? Linters - твой код теперь пахнет ромашками

Линтеры - инструменты, которые помогают повышать читабельность вашего кода, а также указывают на ваши ошибки в стиле оформления. Для Python: Black, Flake8, Pylint, Bandit, Ruff. Сегодня заострим внимание на Pylint, тк его используют много проектов и он выдает оценку вашего кода. Так что сможете ходить и перед всеми красоваться, как вы написали код на -40 из 10 возможных

Перед нами стоит кейс: собрать все данные о пользователе в удобном для дальнейшей работы формате. За 3 минуты крутейший программист приемный сын маминой подруги выдал вот такой страшный код

name_per = 'Bob'
lst = ['315-194-6020', '555-2368', '8-800-555-35-35']
lst2 = ['(1)', '(2)', '(3)']

import copy
dct = {}
lst3 = []
lst4 = []
for i in range(len(lst2)):
    dct.update({'описание': lst2[i], 'номер': lst[i]})
    lst3.append(copy.deepcopy(dct))
lst4.append(
    {'имя': name_per, 'телефоны': lst3},
)

Сразу замечаем странное название переменных dct, lst. Если вы до сих пор так называете свои переменные - перестаньте так их называть! Сегодня мы автоматизируем свою жизнь, так что переходим от ручного рефакторинга к автоматическому с помощью Pylint. Пишем команду pylint file_name.py

Получаем:

main.py:1:0: C0114: Missing module docstring (missing-module-docstring)
main.py:1:0: C0103: Constant name "name_per" doesn't conform to UPPER_CASE naming style (invalid-name)
main.py:5:0: C0413: Import "import copy" should be placed at the top of the module (wrong-import-position)
main.py:9:0: C0200: Consider using enumerate instead of iterating with range and len (consider-using-enumerate)

-----------------------------------
Your code has been rated at 6.36/10

Даже так мы набрали больше 6 баллов (Чтобы набрать -100 надо конкретно попотеть). Данный инструмент вы можете встроить в git precommit, чтобы перед коммитом код прогонялся через линтер и если балл ниже определенного, то разработчику надо будет дописать код. Так вы сэкономите кучу времени на дальнейшем рефакторинге кода

⚡ Ligthning - подзаряди свой pytorch

PyTorch Lightning - целый фреймворк, который следит за обучением, дообучением и выпуском моделей в прод. Lightning - это фактически огромная надстройка над PyTorch, которая бустит процесс проектирования моделей в разы

Если кратко, как интегрирована Lightning:

Давайте представим, что мы реализовали следующую простую модель:

class FCN(nn.Module):
    def __init__(self, input_size, num_classes):
        super().__init__()
        self.fc1 = nn.Linear(input_size, 50)
        self.fc2 = nn.Linear(50, num_classes)

    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = self.fc2(x)
        return x

Для обучения нам придется написать следующее:

for epoch in range(config.num_epochs):
    for batch_idx, (data, targets) in enumerate(tqdm(train_loader)):
        data = data.to(device=config.device)
        targets = targets.to(device=config.device)

        data = data.reshape(data.shape[0], -1)

        scores = model(data)
        loss = criterion(scores, targets)

        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

Так мы действительно обучим модель и получим какие-то показатели. Но если я захочу отслеживать метрики по мере обучения? А как включить мой любимый Early stopping? При увеличении модели может понадобится gradient accumulation. И все это придется писать вручную

Хорошо, что есть Lightning. Мы можем дописать некоторые функции и добавить функционала без лишней головной боли.

class Lightning_FCN(pl.LightningModule):
    def __init__(self, input_size, num_classes):
        super().__init__()
        self.fc1 = nn.Linear(input_size, 50)
        self.fc2 = nn.Linear(50, num_classes)
        self.loss_fn = nn.CrossEntropyLoss()

    def forward(self, x):
        x = F.relu(self.fc1(x))
        x = self.fc2(x)
        return x

    def training_step(self, batch, batch_idx):
        loss, scores, y = self._common_step(batch, batch_idx)
        self.log("train_loss", loss)
        return loss

    def validation_step(self, batch, batch_idx):
        loss, scores, y = self._common_step(batch, batch_idx)
        self.log("val_loss", loss)
        return loss

    def test_step(self, batch, batch_idx):
        loss, scores, y = self._common_step(batch, batch_idx)
        self.log("test_loss", loss)
        return loss

    def _common_step(self, batch, batch_idx):
        x, y = batch
        x = x.reshape(x.size(0), -1)
        scores = self.forward(x)
        loss = self.loss_fn(scores, y)
        return loss, scores, y

    def predict_step(self, batch, batch_idx):
        x, y = batch
        x = x.reshape(x.size(0), -1)
        scores = self.forward(x)
        preds = torch.argmax(scores, dim=1)
        return preds

    def configure_optimizers(self):
        return optim.Adam(self.parameters(), lr=config.learning_rate)

Чтобы моделька завелась в Lightning необходимо дописать training_step, test_step, validation_step . Чтобы отслеживать метрики мы дописали self.log так мы сможем логировать все что заходим

И опять обучать модельку через циклы? Нет и еще раз нет! Создадим "тренера", который следить за обучением вашей модели и показывать все показатели в удобном формате:

trainer = pl.Trainer(
    accelerator="gpu", 
    devices=1, 
    min_epochs=1, 
    max_epochs=3, 
    precision=16,
)
trainer.fit(model, train_loader, val_loader)
trainer.validate(model, val_loader)
trainer.test(model, test_loader)

Если же мы хотим отслеживать какие-то специальные метрики, то:

self.log_dict({'train_loss': loss, 'train_accuracy': accuracy, 'train_f1_score': f1_score},
                      on_step=False, on_epoch=True, prog_bar=True)

Позволит логировать не только лосс, но и accuracy и f1 примерно в таком виде

Также можно легко настроить распределенные вычисления, логирование в MLFlow и другие сервисы. В общем Lightning это огромный фреймворк, который призван убирать рутину в работе и наводить порядок в экспериментах

? DVC - свой такой git для датасетов

Самое вкусное оставили на конец. Часто ли у вас бывало, когда хакатоните и у вас собирается 100+ версий исходного датасета, а вы не понимаете на чем обучать модель... Какие фичи были внедрены, какие нужны/не нужны, была ли фильтрация
DVC призван помочь вам в этом деле. Давайте посмотрим как его стоит применять

DVC (Data Version Control) умеет ооочень много, но сегодня мы пощупаем эту библиотеку: попробуем создать локальное хранилище, создать некоторые изменения и запушить изменения на сервер.

Создадим новый проект:

poetry new dvc_learning
cd 
poetry add dvc
dvc init

Фактически DVC умеет взаимодействовать со всеми популярными хранилищами данных начиная от Google Drive и до AWS. Но мы будем использовать репу на Github

Любезно воспользуемся датасетом с kaggle для примера. Фиксируем первоначальные изменения dvc add data/red_wine.csv

Появляется еще один файлик - red_wine.csv.dvc. В котором записывается служебная информация о нем: хеш файла, по какому пути он располагается и тд. Размер такого файла около 1KB, поэтому не стоит переживать, что у вас будет засорятся репка

Теперь давайте переходить от теории к практике - напишем функцию генерации фичей, а потом прогоним наш условный pipeline

import pandas as pd
from sklearn.decomposition import PCA
from dataclasses import dataclass

  

@dataclass(frozen=True)
class Config:
	n_components: int = 8
	data_path: str = "./data/winequality-red.csv"
config = Config()


def get_pca(df: pd.DataFrame, target_col: str) -> tuple[pd.DataFrame, pd.Series]:
	pca = PCA(config.n_components)
	X = df.drop(columns=target_col)
	y = df[target_col]

	X_tr = pca.fit_transform(X)
	return (X_tr, y)


def get_features(df: pd.DataFrame, target_col: str) -> pd.DataFrame:
	df = df.copy()
	alc_by_acid = df["alcohol"] / df["volatile acidity"]
	suplhates_by_total_sulfur = df["sulphutes"] / df["total sulfur dioxide"]
	df["alc_by_acid"] = alc_by_acid
	df["suplhates_by_total_sulfur"] = suplhates_by_total_sulfur

	return df

  
if __name__ == "__main__":
	df = pd.read_csv(config.data_path)
	df_featured = get_features(df)
	X, y = get_pca(df, "quality")
	new_df = pd.DataFrame([X, y], columns=X.columns + "quality")

	new_df.to_csv("./data/winequality-featured.csv")

Нагенерировали две фичи и также применили PCA, после этого сохранили все это в файл. Давайте пробовать прогнать наш пайплайн через DVC:

dvc run -f feature_generation.dvc \
	-d feature_engineering.py \
	-d data/winequality-red.csv \
	-o data/winequality-featured.csv \
	python feature_engineering.py

Через аргумент -f мы даем понять DVC где он будет сохранять метаданные датасета. -d необходимые для сборки проекта. -o куда сохраняем новый датасет, и наконец сама команда запуска pipeline. Также вы можете сохранять метрики на вашем новом датасете -
это решает нашу основную проблему!

Если мы хотим все это дело сохранить на Github, то нам очень повезло тк DVC по умолчанию использует Git для сохранения файлов, поэтому: команда git init

После настраиваем Git и в путь: dvc push

И вы прекрасны! Изменения появились на сервере. При новых датасетов напишите краткое, но емкое описание новых данных, чтобы коллеги смогли разобрать стоит использовать их при обучении или нет

Также при добавлении датасета, можно прогнать легкую модель и посмотреть на ее производительность. И при dvc run можно добавить метрики и таким образом выбрать лучший датасет из генераций

? ClearML - убийца Weights & Bias?!

ClearML - относительно не новый фреймворк, который нацелен на трекинг ваших экспериментов с моделями. Но сейчас он умеет намного больше:

ClearML умеет буквально все, от сохранения артефактов вашего процесса обучения до оркестраций решений. Также он умеет версионировать датасеты, как DVC
Чтобы изучить все, что этот проект умеет потребуется не одна статья, поэтому давайте посмотрим только на создание pipeline. Наш пробный pipeline: загрузка данных -> разбиение кросс валидацией -> тест -> получение метрик -> обучение модели и выгрузка. Давайте начинать!

def load_dataset() -> pd.DataFrame:
	iris = datasets.load_iris(as_frame=True)
	print('===Fetched successfully!===')
	X, y = iris['data'], iris['target']
	df = X.copy()
	df['target'] = y
	return df

def split_data(data: pd.DataFrame, target: str='target', n_splits: int=5, cv=StratifiedKFold) -> list:	
	cv = cv(n_splits=n_splits, random_state=2024)
	X = data.drop(target, axis=1)
	y = data[target]
	
	splitted_data = []
	for idx, (train_idx, test_idx) in enumerate(cv.split(X, y), start=1):
		print(f'Fold: {idx}')
		X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
		y_train, y_test = y[train_idx], y[test_idx]
		splitted_data.append((X_train, y_train), (X_test, y_test))
	return splitted_data

def fit_model(model, data: pd.DataFrame, target: str='target') -> None:
	X, y = data.drop(target, axis=1), data[target]
	print('===Start fit model===')	
	model.fit(X, y)
	print('===Successfully!')	
	joblib.dump(model, 'model.joblib')

def cv_model(data: list, model, metric=accuracy_score) -> float:
	scores = []
	for idx, (train, test) in enumerate(data, start=1):
		print('===Fold {idx}===')
		model.fit(*train)
		score = accuracy_score(test[1], test[0])
		print('===Score: {score:.3}')
		scores.append(score)
	mean_scores = np.mean(scores) - np.std(scores)
	print('\n' + f'Mean Score: {mean_scores}')
	return mean_scores

Все необходимые функции для выполнения кода ☝️ Приступаем к созданию pipeline:

pipe = PipelineController(
	project='Data Feeling ?',
	name='ClearML Demo',
	version='1.0',
	add_pipeline_tags=False
)

pipe.set_default_execution_queue('default')

Обращаемся к классу из ClearML - даем название проекта, название таски, версия и будем ли давать теги нашим функция. А дальше добавляем этапы:

pipe.add_function_step(
	name='load_dataset',
	function=load_dataset,
	function_return=['df'],
	cache_executed_step=True,
)

pipe.add_function_step(
	name='split_data',
	function=split_data,
	function_kwargs=dict(data_frame='${load_dataset.df}'),
	function_return=['splitted_data'],
	cache_executed_step=True,
)

pipe.add_function_step(
	name='Test model',
	function=cv_model,
	function_kwargs=dict(data='${split_data.splitted_data}', model=model),
	function_return=['mean_scores'],
	cache_executed_step=True,
)

pipe.add_function_step(
	name='Fit model',
	function=fit_model,
	function_kwargs=dict(data='${split_data.splitted_data}', model=model),
	function_return=['model_pkl'],
	cache_executed_step=True
)

Необходимо давать название этапам, также функция которая будет ответственна за степ. Потом передаем все необходимые зависимости для функции - мы их можем брать, как с предыдущего этапа или же сами где-то сохранили. А также если вы хотите поменять выход функции вы это можете сделать в function_return и все. Запуск!

Для запуска: pipe.start() . После чего на странице ClearML можем увидеть схему нашего обучения. Можем потыкать и посмотреть на артефакты обучения

Отличительнная черта ClearML от других фреймворков в MLOps заключается в том, что он старается сохранять все, что можно сохранить. Вы сделали EDA в своем проекте - супер, мы их сохраним, чтобы в ничего не потеряли. Обучаете модель через Lightning? Мы ее тоже сохраним

Таким образом ClearML - отдельная платформа, когда вы хотите настроить логирование обучения без лишней мороки и когда не хотите думать, как сохранять свою модель. А после обучения хотите скинуть своему коллеге графики обучения и метрики, чтобы улучшить модель. Не жизнь, а сказка!

После того, как проект инициализировался мы видим следующую красоту

Выводы

Мы рассмотрели сегодня 5 сочных инструментов, которые увеличат вашу эффективность. Выучите их за ближайшие выходные. Эти инструменты выгодно выделят вас среди остальных коллег.

Делитесь в комментариях, какие инструменты вы используете в работе и почерпнули что-то новое в свой рабочий процесс?

Еще больше про полезные технологии Data Science и ML инструменты вы можете найти в моем тг-канале. В канале я рассказываю про главные новости в AI и ML, а также пишу про свой практический опыт решения ML кейсов. Например недавно писал про собранную демку голосовой заказчика в Додо с помощью RAG + few-shots техники.