Черновик для преподавателя. Если ссылка на Colab не открывается, скачайте ноутбук и загрузите его в Colab. Выберите среду Python с CPU.
Выполните настройку и перейдите к заданию, которое объявит преподаватель. Сначала предположите результат. Затем заполните отмеченные строки по соседним примерам и изучите изображения и таблицу. Ответьте в Moodle по результатам своего запуска. Подготовьтесь объяснить один пример, на котором метод ошибся.
На настройку отведите 5–10 минут, на каждое задание — 15–20 минут. Три основных задания посвящены восстановлению изображений, поиску похожих цифр и обнаружению аномалий. Два дополнительных задания посвящены картам и генерации. Сравните качество и время работы. Подумайте, чего не учитывает каждая метрика.
TODO(instructor): подтвердить темп занятия, набор заданий и набор данных перед публикацией.
Запустите свёрнутую ячейку с функциями визуализации в начале, затем обе ячейки настройки. Введите месяц и день рождения без года или придумайте допустимую дату. Сохраните дату и назначенный вариант у себя. В Moodle нужно указать только лучший метод по каждому основному заданию.
Используем 1797 рукописных цифр размером 8×8. Данные доступны без загрузки. В учебнике используется MNIST с изображениями 28×28. Строка матрицы X содержит D = 64 пикселя, а строка H — d координат скрытого кода. x_hat обозначает восстановленное изображение, y — метку цифры. Метки нужны для разбиения данных, выбора аномалий и проверки результатов. Модели обучаются по значениям пикселей.
Фиксированное разбиение: около 60% для обучения, 20% для валидации и 20% для тестирования. Для ответа в Moodle сохраните назначенные настройки. Для дополнительных экспериментов используйте валидационную выборку.
from importlib.metadata import PackageNotFoundError, versionimport platformimport shleximport subprocessimport syspackages = ("numpy", "scipy", "scikit-learn", "matplotlib", "torch", "threadpoolctl")def package_versions(): versions = {}for package in packages:try: versions[package] = version(package)except PackageNotFoundError: versions[package] =Nonereturn versionsbefore = package_versions()missing = [name for name, installed in before.items() if installed isNone]if missing: command = [sys.executable, "-m", "pip", "install", *missing]if"google.colab"notin sys.modules:raiseRuntimeError(f"Install missing packages and rerun Setup: {shlex.join(command)}")try: subprocess.run(command, check=True)except (OSError, subprocess.CalledProcessError) as exc:raiseRuntimeError("Package setup failed. Try a fresh Colab session and rerun Setup.") from exc after = package_versions() loaded_names = {"scikit-learn": "sklearn"}ifany(before[name] != after[name] and loaded_names.get(name, name) in sys.modulesfor name in packages):raiseRuntimeError("Restart the session, then rerun Setup to load the installed versions.")print("Python", platform.python_version(), "| CPU")print(package_versions())
Для каждого задания создаются новые модели и используется отдельное начальное значение генератора. Время обучения измеряется один раз. Для поиска и вычисления оценок берётся медиана 20 замеров после прогрева, с чередованием порядка методов. Сравнивайте время на одном CPU. Пороги ничьей приняты для этого занятия.
incomplete ="I could not obtain a complete result"def run_activity(activity, offset): seed = personal_seed + offset torch.manual_seed(seed) np.random.seed(seed)try: activity(seed)except (AssertionError, TypeError, ValueError, IndexError, RuntimeError) as exc:print(f"{incomplete}. Проверьте отмеченные строки. {exc}")def timed(operation): start = perf_counter() result = operation()return result, perf_counter() - startdef batch_times(operations): samples = {name: [] for name in operations}for operation in operations.values(): operation() names =list(operations)for repeat inrange(20):for name in (names if repeat %2==0else names[::-1]): _, seconds = timed(operations[name]) samples[name].append(seconds)return {name: float(np.median(times)) for name, times in samples.items()}def report(scores, metric, question=None, tolerance=0, maximize=False, timing=False):assert np.isfinite(list(scores.values())).all(), "A comparison contains a nonfinite result."print(f"\n{metric}")for name, value in scores.items():print(f"{name:34}{value:.6g}") best = (maxif maximize elsemin)(scores.values()) margin = best * tolerance if timing ornot maximize else tolerance winners = [name for name, value in scores.items() ifabs(value - best) <= margin] tie ="Tie / too close to distinguish"if timing else"Tie / practically the same" prefix =f"Moodle Q{question}"if question isnotNoneelse"Result"print(f"{prefix}: {winners[0] iflen(winners) ==1else tie}")class Autoencoder(nn.Module):def__init__(self, width):super().__init__()self.encoder = nn.Sequential(nn.Linear(64, 64), nn.ReLU(), nn.Linear(64, width))self.decoder = nn.Sequential(nn.Linear(width, 64), nn.ReLU(), nn.Linear(64, 64), nn.Sigmoid())def forward(self, x):returnself.decoder(self.encoder(x))def fit_network(model, dataset, seed, objective, epochs=40): batches = DataLoader(dataset, batch_size=128, shuffle=True, generator=torch.Generator().manual_seed(seed)) optimizer = torch.optim.Adam(model.parameters(), lr=0.001)for epoch inrange(epochs):for batch_number, (x, target) inenumerate(batches):assert x.ndim ==2and x.shape == target.shape and x.shape[1] ==64 optimizer.zero_grad() loss = objective(model, x, target)assert torch.isfinite(loss), "Training loss is nonfinite." loss.backward()if epoch ==0and batch_number ==0: gradients = [p.grad for p in model.parameters() if p.grad isnotNone]assertall(torch.isfinite(g).all() for g in gradients)assertany(torch.count_nonzero(g) for g in gradients), "No learning signal." optimizer.step()return model.eval()def mse(model, x, target):return (model(x) - target).square().mean()def clean_ae(images, seed): x = torch.from_numpy(images)return fit_network(Autoencoder(d), TensorDataset(x, x), seed, mse)@torch.inference_mode()def predict(operation, images):return operation(torch.from_numpy(images)).numpy()def fit_pca(images, width):return PCA(n_components=width, svd_solver="full").fit(images)def reconstruct_pca(model, images):return model.inverse_transform(model.transform(images))
Задание 1 — Восстановить зашумлённую цифру
Предскажите: какой метод лучше восстановит штрихи? Какой быстрее обучится?
Дополните первую свёртку и линейную проекцию в скрытый код. Формы тензоров: [B, 1, 8, 8] → [B, 8, 8, 8] → [B, 512] → [B, d]. Возьмите ядро 3×3 с шагом 1. В каждой паре зашумлённое изображение служит входом, а чистое — целью. Декодер восстанавливает вектор из 64 пикселей.
PCA обучается на чистых тренировочных изображениях. Оба метода получают одни и те же зашумлённые тестовые изображения со значениями в диапазоне [0, 1]. MSE для PCA вычисляется по его исходному выходу. Ограничения шкалы серого действуют только при отрисовке. Строка с шумом служит ориентиром.
Moodle Q1: выберите метод с меньшей MSE относительно чистых изображений. Ошибки в пределах 1% от минимума считаются ничьей. Время обучения обсудите устно. Разница в пределах 10% от минимального времени считается ничьей. Найдите пример, где численный результат и визуальная оценка расходятся.
def denoising(seed): rng = np.random.default_rng(seed) noisy = np.clip(X + rng.normal(0, 0.35, X.shape), 0, 1).astype(np.float32) noisy_train, clean_train =map(torch.from_numpy, (noisy[train], X[train]))# Пример: свёртка 3×3 из 2 в 4 канала сохраняет высоту и ширину.# nn.Conv2d(2, 4, kernel_size=3, stride=1, padding=1) convolution =None# TODO(student): 1 input channel, 8 output channels# Пример: nn.Linear(4 * 8 * 8, 6) переводит развёрнутое 4-канальное изображение в 6 координат. projection =None# TODO(student): project the flattened features to dif convolution isNoneor projection isNone:print(incomplete, "Заполните свёртку и проекцию.")returnwith torch.no_grad(): features = convolution(noisy_train[:2].reshape(2, 1, 8, 8))assert features.shape == (2, 8, 8, 8), "Проверьте число каналов, шаг и отступ."assert projection(features.flatten(1)).shape == (2, d), "Проверьте размеры линейного слоя."def fit_denoiser(): model = Autoencoder(d) model.encoder = nn.Sequential( nn.Unflatten(1, (1, 8, 8)), convolution, nn.ReLU(), nn.Flatten(1), projection )return fit_network(model, TensorDataset(noisy_train, clean_train), seed, mse) pca, pca_seconds = timed(lambda: fit_pca(X[train], d)) ae, ae_seconds = timed(fit_denoiser)assert predict(ae.encoder, X[test[:8]]).shape == (8, d) results = {"PCA": reconstruct_pca(pca, noisy[test]),"Denoising autoencoder": predict(ae, noisy[test]), } clean = X[test]print(f"MSE зашумлённых изображений (ориентир): {np.mean((noisy[test] - clean)**2):.6f}") report({name: np.mean((out - clean)**2) for name, out in results.items()},"Тестовая MSE относительно чистых изображений", 1, tolerance=0.01) report(dict(zip(results, (pca_seconds, ae_seconds))), "Время обучения, с", tolerance=0.10, timing=True) rows = [("Чистые", clean[:8], None), ("С шумом", noisy[test[:8]], None)] rows += [(name, out[:8], None) for name, out in results.items()] image_rows(rows, "Восемь тестовых цифр: чистая цель, вход с шумом и два восстановления.")run_activity(denoising, 100)
Задание 2 — Найти такую же цифру
Предскажите: помогает ли код, удобный для восстановления, при поиске похожих цифр?
Для каждого тестового запроса найдите одно ближайшее тренировочное изображение. Сравните исходные пиксели, коды PCA и коды автоэнкодера. В каждом пространстве используется евклидово расстояние. Представления обучаются на чистых тренировочных изображениях. Совпадение означает одинаковую метку цифры. Идентификаторы запросов и изображений галереи не пересекаются.
Выберите один индекс галереи на запрос, посчитайте одинаковые метки и вычислите долю правильных первых ответов (top-1 accuracy). Строка матрицы расстояний соответствует запросу, столбец — изображению галереи.
Moodle Q2: выберите представление с наибольшим числом совпадений. Отставание на одно совпадение считается ничьей. Устно сравните время кодирования запросов и поиска в готовой галерее. Время в пределах 10% от минимума считается ничьей.
def find_matches(encode, query, gallery): query_codes = encode(query) distances = cdist(query_codes, gallery, metric="sqeuclidean")assert distances.shape == (len(query), len(gallery))# Пример: np.argmin(table, axis=0) выбирает строку для каждого столбца. nearest =None# TODO(student): one minimum index per queryif nearest isnotNone:assert nearest.shape == (len(query),) and ((0<= nearest) & (nearest <len(gallery))).all()return nearestdef match_accuracy(found_labels, query_labels):# Пример: int(np.sum(np.array([1, 4]) == np.array([1, 5]))) даёт 1. count =None# TODO(student): count equal labelsif count isNone:returnNone, None# Пример: доля для 3 правильных ответов из 5 равна 3 / 5. accuracy =None# TODO(student): fraction of queries matchedreturn count, accuracydef retrieval(seed): probe = find_matches(np.asarray, X[test[:2]], X[train[:3]]) count, accuracy = match_accuracy(np.array([1, 4]), np.array([1, 5]))if probe isNoneor count isNoneor accuracy isNone:print(incomplete, "Заполните строки выбора соседа и расчёта точности.")returnassert (count, accuracy) == (1, 0.5), "Проверьте сравнение меток и знаменатель."assertnot np.intersect1d(train, test).size pca = fit_pca(X[train], d) ae = clean_ae(X[train], seed) encoders = {"Original pixels": np.asarray, "PCA codes": pca.transform,"Autoencoder codes": lambda x: predict(ae.encoder, x)} operations = {}for name, encode in encoders.items(): gallery = encode(X[train]) operations[name] =lambda encode=encode, gallery=gallery: find_matches(encode, X[test], gallery) matches = {name: train[operation()] for name, operation in operations.items()} scores = {}for name, ids in matches.items(): count, accuracy = match_accuracy(y[ids], y[test]) scores[name] = countprint(f"{name}: {count}/{len(test)} ({accuracy:.1%})") report(scores, "Правильные совпадения", 2, tolerance=1, maximize=True) report(batch_times(operations), "Время поиска по пакету, с", tolerance=0.10, timing=True) rows = [("Запрос", X[test[:8]], y[test[:8]])] rows += [(name, X[ids[:8]], y[ids[:8]]) for name, ids in matches.items()] image_rows(rows, "Восемь тестовых запросов и найденные соседи с истинными метками.")run_activity(retrieval, 200)
Задание 3 — Найти чужака
Предскажите: может ли модель хорошо восстановить незнакомую цифру?
Цифры 0–8 считаются обычными. Обучите k-means, PCA и автоэнкодер только на таких тренировочных изображениях. В тестовом пакете 80 обычных изображений и 20 девяток. Каждый метод помечает десять изображений. Метки задают состав выборки и используются для проверки после вычисления оценок.
Сведите квадраты ошибок по пикселям к одной оценке на изображение. Отсортируйте оценки, поставив самые подозрительные изображения первыми, и посчитайте девятки среди отметок. При одинаковых оценках порядок задаётся исходными идентификаторами.
Moodle Q3: выберите метод, который нашёл больше девяток среди десяти отметок. При равном максимуме выберите ничью. Время вычисления оценок обсудите устно. Время в пределах 10% от минимума считается ничьей. Сами оценки у методов имеют разные шкалы.
def reconstruction_scores(x, x_hat): squared = (x - x_hat)**2# Пример: squared.mean(axis=0) даёт по одному значению на пиксель.returnNone# TODO(student): reduce the pixel axisdef rank_flags(scores):# Пример: np.argsort(scores, kind="stable") ставит меньшие оценки первыми.returnNone# TODO(student): descending stable argsortdef count_intruders(flag_labels):# Пример: int(np.sum(flag_labels == 2)) считает двойки.returnNone# TODO(student): count actual anomaliesdef anomalies(seed):if reconstruction_scores(X[test[:2]], X[test[:2]]) isNoneor rank_flags(np.array([0., 1.])) isNoneor count_intruders(np.array([9, 2, 9])) isNone:print(incomplete, "Заполните усреднение, сортировку и подсчёт.")returnassert count_intruders(np.array([9, 2, 9])) ==2, "Посчитайте только девятки." rng = np.random.default_rng(seed) normal_train = train[y[train] !=9] normal_test, unfamiliar = test[y[test] !=9], test[y[test] ==9] ids = np.r_[rng.choice(normal_test, 80, replace=False), rng.choice(unfamiliar, 20, replace=False)] rng.shuffle(ids)assertlen(np.unique(ids)) ==100and (y[ids] ==9).sum() ==20assert (y[normal_train] !=9).all()# Scores use a shuffled batch. Original IDs break ties at the ranking step. x = X[ids] kmeans = KMeans(10, n_init=10, random_state=seed).fit(X[normal_train]) pca = fit_pca(X[normal_train], d) ae = clean_ae(X[normal_train], seed) operations = {"Distance to k-means centers": lambda: cdist(x, kmeans.cluster_centers_, "sqeuclidean").min(axis=1),"PCA reconstruction error": lambda: reconstruction_scores(x, reconstruct_pca(pca, x)),"Autoencoder reconstruction error": lambda: reconstruction_scores(x, predict(ae, x)), } flags = {} row_order = np.argsort(ids)for name, operation in operations.items(): scores = np.asarray(operation())assert scores.shape == (100,), "Усредните по пикселям, чтобы получить одну оценку на изображение." scores = scores[row_order] rank = rank_flags(scores)assert np.array_equal(np.sort(rank), np.arange(100)), "Верните 100 индексов строк."assert np.all(np.diff(scores[rank]) <=0), "Поставьте наибольшие оценки первыми." flags[name] = ids[row_order[rank[:10]]] report({name: count_intruders(y[flag]) for name, flag in flags.items()},"Девятки среди десяти отметок (80 обычных / 20 незнакомых)", 3, maximize=True) report(batch_times(operations), "Время расчёта оценок без сортировки, с", tolerance=0.10, timing=True) image_rows([(name, X[flag], y[flag]) for name, flag in flags.items()],"Десять изображений с наибольшими оценками каждого метода и их истинные метки.")run_activity(anomalies, 300)
Дополнение A — Можно ли доверять карте?
Предскажите: какая карта сохранит больше из десяти ближайших соседей?
Постройте двумерные карты PCA и t-SNE по одним и тем же 500 изображениям. Исключите каждое изображение из списка его соседей, изменив диагональ матрицы расстояний. Затем вычислите долю сохранённых соседей. Метки используются для окраски точек после построения карт.
Обсудите: сравните PCA с t-SNE при перплексии 30. Посмотрите на среднюю долю сохранённых соседей и время построения. Разница долей до 0,01 и время в пределах 10% от минимального считаются ничьей. Сравните с картой при перплексии 10. Какие выводы можно сделать о соседях? Что по расстояниям между группами заключить нельзя?
def map_neighbors(coordinates): distances = cdist(coordinates, coordinates, "sqeuclidean")# Пример: расстояние np.inf исключает кандидата из десяти ближайших. self_distance =None# TODO(student): value to put on the diagonalif self_distance isNone:returnNone np.fill_diagonal(distances, self_distance) neighbors = np.argsort(distances, axis=1, kind="stable")[:, :10]assertnot np.any(neighbors == np.arange(len(coordinates))[:, None]), "Исключите каждую строку из списка её соседей."return neighborsdef neighbor_overlap(original, mapped):# Пример: np.intersect1d([0, 1], [1, 2]) возвращает array([1]).returnNone# TODO(student): fraction of original neighbors retaineddef maps(seed): ids, _ = train_test_split(train, train_size=500, stratify=y[train], random_state=1010) pixels = X[ids] original = map_neighbors(pixels) probe = neighbor_overlap(np.array([0, 1]), np.array([1, 2]))if original isNoneor probe isNone:print(incomplete, "Исключите сам объект и вычислите долю общих соседей.")returnassert probe ==0.5, "Должен сохраниться один из двух исходных соседей." pca_map, pca_seconds = timed(lambda: PCA(2, svd_solver="full").fit_transform(pixels)) tsne_map, tsne_seconds = timed(lambda: tsne(30, seed).fit_transform(pixels)) views = {"PCA map": pca_map, "t-SNE map (perplexity 30)": tsne_map} neighbors = {name: map_neighbors(coords) for name, coords in views.items()} retention = { name: np.mean([neighbor_overlap(a, b) for a, b inzip(original, mapped)])for name, mapped in neighbors.items() } report(retention, "Доля сохранённых соседей из десяти исходных", tolerance=0.01, maximize=True) report(dict(zip(views, (pca_seconds, tsne_seconds))), "Время построения карты, с", tolerance=0.10, timing=True) views["t-SNE map (perplexity 10)"] = tsne(10, seed).fit_transform(pixels) plot_maps(views, y[ids])for i in (0, 1, 2): rows = [("Запрос + пиксели", pixels[np.r_[i, original[i, :4]]], None)] rows += [(name, pixels[np.r_[i, near[i, :4]]], None) for name, near in neighbors.items()] image_rows(rows, f"Слева изображение {ids[i]}, затем четыре соседа в каждом пространстве.")run_activity(maps, 400)
Дополнение B — Восстановить или сгенерировать?
Предскажите: даст ли лучший по восстановлению метод больше узнаваемых цифр при декодировании случайных кодов?
Дополните выборку кода и сумму вкладов KL для каждого изображения. Энкодер предсказывает среднее и логарифм дисперсии. Получите стандартное отклонение, умножьте на него стандартный нормальный шум и прибавьте среднее.
VAE минимизирует отрицательное логарифмическое правдоподобие гауссовского восстановления со стандартным отклонением 0,25 и KL-дивергенцию до стандартного нормального распределения. Сложите вклады по координатам каждого слагаемого и усредните по пакету. Автоэнкодер обучается по MSE пикселей. Сравните модели на одной тестовой MSE. Для воспроизводимого восстановления VAE декодирует среднее энкодера.
Обсудите: выберите меньшую тестовую MSE. Разница в пределах 1% считается ничьей. Затем решите, в какой сетке больше узнаваемых цифр. Обе модели декодируют одни и те же 16 случайных кодов. На изображениях показаны средние декодеров. Автоэнкодер не обучался согласовывать коды со стандартным нормальным распределением.
def sample_code(mu, logvar, epsilon):# Пример: mean + std * noise масштабирует и сдвигает стандартный нормальный шум. std =None# TODO(student): convert log variance to standard deviationreturnNoneif std isNoneelse mu + std * epsilonclass VariationalAutoencoder(nn.Module):def__init__(self, width):super().__init__()self.encoder = nn.Sequential(nn.Linear(64, 64), nn.ReLU())self.mu = nn.Linear(64, width)self.logvar = nn.Linear(64, width)self.decoder = nn.Sequential(nn.Linear(width, 64), nn.ReLU(), nn.Linear(64, 64), nn.Sigmoid())def encode(self, x): h =self.encoder(x)returnself.mu(h), self.logvar(h)def kl_to_normal(mu, logvar): per_coordinate =0.5* (mu.square() + logvar.exp() -1- logvar)# Пример: per_coordinate.sum(dim=0) даёт сумму для каждой скрытой координаты.returnNone# TODO(student): sum the KL coordinates for each imagedef vae_objective(model, x, target): mu, logvar = model.encode(x) z_sample = sample_code(mu, logvar, torch.randn_like(mu))assert z_sample.shape == mu.shape x_hat = model.decoder(z_sample) sd =0.25 nll = (0.5* ((target - x_hat) / sd).square() + math.log(sd * math.sqrt(2* math.pi))).sum(dim=1)return (nll + kl_to_normal(mu, logvar)).mean()def generation(seed): zero, one = torch.zeros(2, d), torch.ones(2, d) sample = sample_code(zero, zero, one)if sample isNoneor kl_to_normal(zero, zero) isNone:print(incomplete, "Заполните выборку и суммирование KL.")returnassert torch.allclose(sample, one), "При нулевом среднем и единичной дисперсии выборка равна шуму."assert torch.allclose(sample_code(one, one * math.log(4), one), 3* one), "Проверьте стандартное отклонение."assert torch.equal(kl_to_normal(zero, zero), torch.zeros(2))assert torch.allclose(kl_to_normal(one, zero), torch.full((2,), d /2)), "Проверьте ось суммирования." x = torch.from_numpy(X[train]) ae = clean_ae(X[train], seed) vae = fit_network(VariationalAutoencoder(d), TensorDataset(x, x), seed, vae_objective) clean = X[test] outputs = {"Autoencoder": predict(ae, clean),"Variational autoencoder": predict(lambda x: vae.decoder(vae.encode(x)[0]), clean), } report({name: np.mean((out - clean)**2) for name, out in outputs.items()}, "Тестовая MSE", tolerance=0.01) rows = [("Чистые", clean[:8], None)] + [(name, out[:8], None) for name, out in outputs.items()] image_rows(rows, "Восемь тестовых цифр и их детерминированные восстановления.") codes = np.random.default_rng(seed).standard_normal((16, d)).astype(np.float32) grids = [("Autoencoder grid", predict(ae.decoder, codes)), ("Variational autoencoder grid", predict(vae.decoder, codes))] image_rows([(name if start ==0else"", out[start:start +8], None)for name, out in grids for start in (0, 8)],"Сравните две сетки по 16 изображений из одних и тех же нормальных кодов.")run_activity(generation, 500)
Обсудите результаты
Выберите метод для одной задачи. Приведите один результат в его пользу, один пример ошибки и одну затрату, которую стоит учесть. Сравните ответы группы как наблюдения по конкретным запускам. Учтите ничьи и неполные результаты.
TODO(instructor): проверить доступ и отчёты Moodle, учёт оценок, темп занятия и актуальную среду Colab перед публикацией.