9 вопросов по теме «ML Engineering: Реализуй с нуля» на собеседовании

Короткий ответ: K-means чередует назначение точек ближайшим центроидам и пересчёт центроидов как средних. Расстояния векторизуются через broadcasting; цикл остаётся по итерациям и кластерам. Вход нужно преобразовать в float, иначе запись среднего в целочисленный массив обрежет дробную часть.

Подробно:

1. Расстояния — выражение ниже создаёт матрицу (n, k) через временный тензор (n, k, d). На больших данных используйте блоки или формулу ‖x‖² − 2xᵀc + ‖c‖² с учётом погрешности округления.
2. Пустой кластер — переинициализируем его случайной точкой. Это простая политика, которая не гарантирует непустые кластеры, особенно при повторяющихся точках.
3. Остановка — сохраняем новые центроиды до проверки выхода. После выхода заново назначаем метки по возвращаемым центроидам; при лимите итераций сходимость не гарантируется.

import numpy as np

def kmeans(X, k, iters=100, tol=1e-6, seed=0):
    X = np.asarray(X, dtype=float)
    if X.ndim != 2 or 0 in X.shape or not np.isfinite(X).all():
        raise ValueError("X must be a nonempty finite 2-D array")
    if not isinstance(k, (int, np.integer)) or not 1 <= k <= len(X):
        raise ValueError("k must be an integer in [1, len(X)]")
    if not isinstance(iters, (int, np.integer)) or iters < 1:
        raise ValueError("iters must be a positive integer")
    if not np.isfinite(tol) or tol < 0:
        raise ValueError("tol must be finite and nonnegative")
    rng = np.random.default_rng(seed)
    C = X[rng.choice(len(X), k, replace=False)].copy()
    for _ in range(iters):
        d2 = ((X[:, None, :] - C[None, :, :]) ** 2).sum(-1)
        labels = d2.argmin(1)
        newC = C.copy()
        for j in range(k):
            pts = X[labels == j]
            newC[j] = pts.mean(0) if len(pts) else X[rng.integers(len(X))]
        shift = np.linalg.norm(newC - C)
        C = newC
        if shift <= tol:
            break
    labels = ((X[:, None, :] - C[None, :, :]) ** 2).sum(-1).argmin(1)
    return C, labels

⚠️ Частая ошибка: сохранить dtype целочисленного входа или вернуть старые метки после обновления центроидов. Это учебная реализация для умеренного масштаба данных; проверяйте масштабирование признаков, инициализацию и несколько перезапусков.

Короткий ответ: Модель p = σ(Xw), лосс — бинарная кросс-энтропия, а её градиент схлопывается до Xᵀ(p − y)/n. Остаётся стабильный сигмоид, клиппинг вероятностей в лоссе и цикл спуска; bias — колонка единиц в X.

Подробно:

1. Стабильный сигмоид — наивный 1/(1+exp(-z)) переполняется при больших |z|; ветвим по знаку z.
2. Лосс без log(0) — при насыщении сигмоида p упирается в 0 или 1; клиппим np.clip(p, eps, 1-eps) либо считаем через logaddexp.
3. Почему градиент чистый — цепное правило: dL/dp × dp/dz даёт σ′, которая сокращается со знаменателем производной лога, остаётся (p − y). Отсюда Xᵀ(p − y)/n — любимый follow-up.

import numpy as np

def sigmoid(z):
    out = np.empty_like(z)          # стабильно для больших |z|
    pos = z >= 0
    out[pos] = 1 / (1 + np.exp(-z[pos]))
    e = np.exp(z[~pos])
    out[~pos] = e / (1 + e)
    return out

def fit(X, y, lr=0.1, epochs=1000, eps=1e-12):
    Xb = np.hstack([X, np.ones((len(X), 1))])  # bias — колонка единиц
    w = np.zeros(Xb.shape[1])
    for _ in range(epochs):
        p = sigmoid(Xb @ w)
        pc = np.clip(p, eps, 1 - eps)          # log(0) не случится
        loss = -np.mean(y*np.log(pc) + (1-y)*np.log(1-pc))
        w -= lr * Xb.T @ (p - y) / len(y)
    return w

⚠️ Частая ошибка: log(p) без клиппинга — насыщенный сигмоид даёт log(0) = −inf и NaN; и забытый bias — граница решения обязана проходить через ноль.

Короткий ответ: scores = QKᵀ/√d_k; верхний треугольник (будущее) заливаем −inf ДО softmax; softmax — по последней оси и со сдвигом на максимум строки; выход — weights @ V.

Подробно:

1. Зачем √d_k — дисперсия скалярного произведения растёт с размерностью; без нормировки softmax насыщается и градиенты исчезают. Стандартный follow-up: «а зачем корень?»
2. Маска строго до softmax — −inf после экспоненты превращается в ровно 0 веса, а строка по-прежнему суммируется в 1. Занулять веса после softmax — сломать нормировку.
3. Стабильный softmax — вычесть максимум строки перед exp.
4. Ось — softmax по ключам (последняя ось scores), не по запросам.

import numpy as np

def softmax(x, axis=-1):
    x = x - x.max(axis=axis, keepdims=True)  # стабильность
    e = np.exp(x)
    return e / e.sum(axis=axis, keepdims=True)

def attention(Q, K, V, causal=True):
    d_k = Q.shape[-1]
    scores = Q @ K.swapaxes(-1, -2) / np.sqrt(d_k)
    if causal:
        T = scores.shape[-1]
        mask = np.triu(np.ones((T, T), dtype=bool), k=1)
        scores = np.where(mask, -np.inf, scores)  # ДО softmax!
    w = softmax(scores, axis=-1)  # по оси ключей
    return w @ V

⚠️ Частая ошибка: softmax не по той оси — и маскирование после softmax: веса перестают суммироваться в единицу, а модель тихо подглядывает в будущее.

Короткий ответ: В train нормируем по статистикам текущего батча и обновляем бегущие mean/var с momentum; в eval используем ТОЛЬКО бегущие статистики. Затем scale-shift: y = γ·x̂ + β.

Подробно:

1. Train — μ, σ² по батчу, x̂ = (x − μ)/√(σ² + ε), running-статистики обновляются экспоненциальным средним.
2. Eval — batch-статистики брать нельзя: предсказание начинает зависеть от состава батча, а при batch=1 дисперсия равна нулю.
3. γ и β — обучаемые; дают слою право при необходимости отменить нормализацию.
4. Почему LayerNorm иначе — она нормирует по признакам одного примера, не по батчу, поэтому не имеет режимов и живёт в трансформерах и RNN.

import numpy as np

class BatchNorm:
    def __init__(self, d, momentum=0.1, eps=1e-5):
        self.gamma, self.beta = np.ones(d), np.zeros(d)
        self.run_mean, self.run_var = np.zeros(d), np.ones(d)
        self.momentum, self.eps = momentum, eps

    def __call__(self, x, training):
        if training:
            mu, var = x.mean(0), x.var(0)
            m = self.momentum
            self.run_mean = (1 - m) * self.run_mean + m * mu
            self.run_var = (1 - m) * self.run_var + m * var
        else:
            mu, var = self.run_mean, self.run_var  # НЕ batch-статистики!
        x_hat = (x - mu) / np.sqrt(var + self.eps)
        return self.gamma * x_hat + self.beta

⚠️ Частая ошибка: один режим на все случаи — batch-статистики на инференсе. Метрики скачут от размера и состава батча, при batch=1 всё ломается. Именно это переключает model.eval() в PyTorch.

Короткий ответ: Перед exp вычитаем максимум строки — уже exp(89) переполняет float32. Для лосса softmax и log не считаем по отдельности, а объединяем в log-sum-exp: CE = logsumexp(z) − z[y].

Подробно:

1. Сдвиг на максимум — softmax инвариантен к сдвигу: exp(z − m)/Σexp(z − m) — та же величина, но без overflow.
2. log-softmax одним куском — log(softmax(z)) = z − m − log Σ exp(z − m); нигде не берём log от почти-нуля.
3. CE как индексирование — минус log-softmax в позиции правильного класса, среднее по батчу.

import numpy as np

def log_softmax(z):
    z = z - z.max(axis=1, keepdims=True)  # exp(89) переполняет fp32
    return z - np.log(np.exp(z).sum(axis=1, keepdims=True))

def cross_entropy(z, y):
    # CE = logsumexp(z) - z[y]; никакого softmax -> log по отдельности
    n = len(y)
    return -log_softmax(z)[np.arange(n), y].mean()

⚠️ Частая ошибка: посчитать softmax, потом взять log — при насыщении вероятность округляется до 0 и log даёт −inf. Именно поэтому F.cross_entropy в PyTorch принимает логиты, а не вероятности.

Короткий ответ: Train: model.train(), затем цикл zero_grad → forward → loss → backward → step. Eval: model.eval() плюс torch.no_grad(), метрики копим взвешенно по размеру батча.

Подробно:

1. zero_grad обязателен — PyTorch суммирует градиенты в .grad; забудешь — получишь градиенты сразу за несколько батчей.
2. eval() и no_grad() — разные вещи — первый переключает dropout и BatchNorm на инференс-режим, второй отключает построение графа. Нужны оба.
3. Взвешивание метрик — последний батч обычно короче; среднее по батчам ≠ среднее по датасету, умножаем на len(x).
4. .item() — иначе тензор тянет за собой граф вычислений, и память течёт.

def train_epoch(model, loader, opt, crit, device):
    model.train()
    for x, y in loader:
        x, y = x.to(device), y.to(device)
        opt.zero_grad()            # иначе градиенты копятся
        loss = crit(model(x), y)
        loss.backward()
        opt.step()

@torch.no_grad()
def evaluate(model, loader, crit, device):
    model.eval()                   # dropout выкл, BN -> running stats
    total, n = 0.0, 0
    for x, y in loader:
        x, y = x.to(device), y.to(device)
        loss = crit(model(x), y)
        total += loss.item() * len(x)  # последний батч короче
        n += len(x)
    return total / n

⚠️ Частая ошибка: забытый zero_grad — градиенты аккумулируются и лосс «странно скачет»; валидация без model.eval() — dropout продолжает выключать нейроны; лосс без .item() — граф копится в памяти.

Короткий ответ: Сначала делим логиты на температуру, потом отсекаем: top-k оставляет k наибольших логитов, top-p — минимальный префикс отсортированных вероятностей с суммой ≥ p (минимум один токен). После отсечения — перенормировка и сэмплирование.

Подробно:

1. Порядок операций — температура меняет форму всего распределения, поэтому идёт до отсечения; T после top-k перевзвешивает уже усечённый набор.
2. top-k — порог по k-му логиту (np.sort(z)[-k] или argpartition), остальное в −inf.
3. top-p — сортировка по убыванию, cumsum, срез там, где сумма достигла p; searchsorted + 1 включает пересекающий токен.
4. Перенормировка — после зануления хвоста делим вероятности на новую сумму.

import numpy as np

def sample(logits, T=1.0, k=None, p=None, rng=np.random.default_rng()):
    z = logits / T                        # температура ДО отсечения
    if k is not None:
        kth = np.sort(z)[-k]              # k-й по величине логит
        z = np.where(z < kth, -np.inf, z)
    probs = np.exp(z - z.max())
    probs /= probs.sum()
    if p is not None:
        order = np.argsort(-probs)
        cum = np.cumsum(probs[order])
        cut = np.searchsorted(cum, p) + 1  # хотя бы один токен
        probs[order[cut:]] = 0.0           # хвост в ноль
        probs /= probs.sum()               # перенормировка
    return rng.choice(len(probs), p=probs)

⚠️ Частая ошибка: температура после отсечения — и срез top-p строго до p без пересекающего токена: на пиковом распределении так можно остаться вовсе без кандидатов.

Короткий ответ: Нормируем строки базы один раз — косинус превращается в matmul единичных векторов; top-k берём через np.argpartition за O(n) вместо полной сортировки. Для 1M×d это одна BLAS-операция.

Подробно:

1. Нормировка заранее — cos(a, b) = â·b̂; базу нормируем один раз офлайн, на каждый запрос остаётся только матричное умножение.
2. argpartition — выбор k наибольших за O(n) без сортировки всего массива; досортировываем только k элементов.
3. Нулевые векторы — деление на нулевую норму даёт NaN; добавляем ε к норме.
4. Мост к продовому ответу — брутфорс на 1M векторов ещё жив (миллисекунды на BLAS), дальше — ANN: HNSW или IVF (faiss), размен точности на скорость.

import numpy as np

def build_index(X, eps=1e-12):
    # один раз офлайн: нормируем строки
    return X / (np.linalg.norm(X, axis=1, keepdims=True) + eps)

def top_k(Xn, q, k=10, eps=1e-12):
    qn = q / (np.linalg.norm(q) + eps)
    sims = Xn @ qn                        # косинус = matmul единичных
    idx = np.argpartition(-sims, k)[:k]   # O(n), не полная сортировка
    return idx[np.argsort(-sims[idx])]    # сортируем только k штук

⚠️ Частая ошибка: np.sort всего массива сходств ради top-10 — O(n log n) вместо O(n); и косинус «в лоб» с перенормировкой всей базы на каждый запрос.

Короткий ответ: Накапливайте градиенты группы микробатчей, нормируйте по фактическому числу примеров в группе и обновляйте веса один раз. Для одинаковых микробатчей это деление каждого среднего лосса на размер группы, включая короткую последнюю группу.

from itertools import islice

accum = 4
batches = iter(loader)
while group := list(islice(batches, accum)):
    total = sum(y.shape[0] for _, y in group)
    opt.zero_grad()
    for x, y in group:
        loss = crit(model(x), y)  # средний лосс по примерам
        (loss * y.shape[0] / total).backward()
    opt.step()

Подробно:

• Пример обрабатывает пустой loader, неполную последнюю группу и микробатчи разного размера. Для токенных лоссов нормируйте по реальному числу учитываемых токенов.
• Одновременно хранится граф вычислений только одного микробатча. Этот вариант буферизует входные тензоры группы; держите батчи loader на CPU и переносите каждый микробатч внутри цикла, если важна память ускорителя.
• Без нормировки градиент одинаковых микробатчей умножается на размер группы. Для обычного SGD это масштабирует градиентный шаг; с адаптивным оптимизатором, clipping или регуляризацией это не всегда эквивалентно умножению learning rate.
• BatchNorm всё равно видит каждый микробатч отдельно: аккумуляция не воспроизводит все свойства обучения на большом батче.

Важно: Если хвост из двух микробатчей делить на четыре, его градиент окажется вдвое меньше нужного. Нормируйте хвост по его реальному размеру.

Источник: Аккумуляция градиентов PyTorch.