// Deep Dive · NLP · 2024-2025
От математики позиционного кодирования до pipeline("sentiment-analysis") — пошаговый разбор с кодом, формулами и примерами.
// 01 · recap
Трансформер, предложенный в статье «Attention Is All You Need» (Vaswani et al., 2017), полностью отказался от рекуррентности. Вместо RNN используется механизм self-attention, который позволяет каждому токену «смотреть» на все остальные токены последовательности за одну операцию — без шага по времени.
Оригинальная модель состоит из энкодера (стек из N одинаковых слоёв) и декодера (тоже стек из N слоёв, но с дополнительным cross-attention). Современные LLM — GPT-4, LLaMA, Mistral — используют decoder-only вариант. BERT — encoder-only.
// АРХИТЕКТУРА ТРАНСФОРМЕРА · ENCODER + DECODER BLOCK
Кликни на любой компонент выше, чтобы увидеть подробное объяснение, математику и советы из практики.
В зависимости от того, какие блоки используются, трансформеры делятся на три архитектурных класса:
// ТРИ АРХИТЕКТУРНЫХ СЕМЕЙСТВА · encoder-only / decoder-only / encoder-decoder
// 01.5 · tokenization
До того как текст попадёт в трансформер, он должен быть разбит на токены — минимальные единицы словаря. Современные модели используют субслóвные алгоритмы: слово может быть разбито на несколько токенов, зато словарь остаётся компактным (~30–100k элементов), и модель справляется с редкими словами.
// BPE TOKENIZATION · «transformer» → 2 токена: [transform][##er]
from transformers import AutoTokenizer # BERT использует WordPiece tok_bert = AutoTokenizer.from_pretrained("bert-base-uncased") # GPT-2 использует BPE tok_gpt = AutoTokenizer.from_pretrained("gpt2") word = "transformers" print(tok_bert.tokenize(word)) # ['transform', '##ers'] ← WordPiece: ## = продолжение print(tok_gpt.tokenize(word)) # ['transform', 'ers'] ← BPE: без ## префикса # Специальные токены enc = tok_bert("Hello world", return_tensors="pt") ids = enc.input_ids[0].tolist() # [101, 7592, 2088, 102] # 101=[CLS], 102=[SEP] — всегда добавляются print(tok_bert.convert_ids_to_tokens(ids)) # ['[CLS]', 'hello', 'world', '[SEP]'] # Как выглядит attention_mask: # 1 = реальный токен, 0 = PAD (при батчинге) print(enc.attention_mask) # tensor([[1, 1, 1, 1]])
[CLS]=101, [SEP]=102, [PAD]=0, [MASK]=103 — это специальные токены BERT. GPT использует только eos_token (конец текста). Никогда не хардкодируй id — используй tokenizer.cls_token_id.
// 02 · positional encoding
Self-attention по природе перестановочно инвариантна: порядок токенов не важен. Чтобы модель «знала», где стоит токен, к эмбеддингу добавляется позиционный вектор — либо синусоидальный (оригинальная статья), либо обучаемый.
Для позиции pos и измерения i в векторе размерности d_model:
Смысл: каждое измерение — своя «частота». Низкочастотные измерения кодируют грубую позицию, высокочастотные — тонкие различия. Модель может интерполировать относительные расстояния через свойства синуса/косинуса.
import torch import torch.nn as nn import math class PositionalEncoding(nn.Module): def __init__(self, d_model: int, max_len: int = 5000, dropout: float = 0.1): super().__init__() self.dropout = nn.Dropout(p=dropout) # Матрица позиций: shape (max_len, d_model) pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp( torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model) ) pe[:, 0::2] = torch.sin(position * div_term) # чётные индексы pe[:, 1::2] = torch.cos(position * div_term) # нечётные # Добавляем batch-измерение: (1, max_len, d_model) pe = pe.unsqueeze(0) self.register_buffer('pe', pe) # не обучаем, но в state_dict def forward(self, x: torch.Tensor) -> torch.Tensor: # x: (batch, seq_len, d_model) x = x + self.pe[:, :x.size(1)] return self.dropout(x)
Каждая строка — вектор для одной позиции, каждый столбец — одно измерение. Левые (низкочастотные) измерения меняются медленно; правые — быстро.
// СИНУСОИДАЛЬНАЯ PE · строки = позиции (0..7) · столбцы = измерения d_model
В отличие от абсолютного PE, RoPE кодирует относительную позицию непосредственно в матрицу attention, поворачивая векторы Q и K на угол, пропорциональный их позициям. Ключевое преимущество: dot product Q·K зависит только от разницы позиций, а не от абсолютных координат.
def rotate_half(x): """Вращение: [x1, x2, x3, x4] → [-x2, x1, -x4, x3]""" x1, x2 = x[..., :x.shape[-1]//2], x[..., x.shape[-1]//2:] return torch.cat([-x2, x1], dim=-1) def apply_rope(q, k, cos, sin): """cos/sin: precomputed для каждой позиции, shape (seq, d_k)""" q_rot = q * cos + rotate_half(q) * sin k_rot = k * cos + rotate_half(k) * sin return q_rot, k_rot # В LLaMA: θ_i = 1 / 10000^(2i/d), i = 0..d/2-1 # Затем cos/sin кешируются для всей max_seq_len
// 03 · multi-head self-attention
Сердце трансформера. Каждый токен «спрашивает» (Query), «предлагает информацию» (Key+Value) — и получает взвешенную сумму значений от всех токенов последовательности.
Зачем √d_k? При больших d_k скалярные произведения Q·Kᵀ становятся очень большими → softmax уходит в насыщение → градиенты исчезают. Деление на √d_k нормирует дисперсию.
Запускаем h независимых «голов», каждая проецирует Q/K/V в подпространство размера d_k = d_model / h. Головы учатся смотреть на разные типы зависимостей одновременно (синтаксис, семантика, анафора и т.д.).
Строка = токен-Query, столбец = токен-Key. Яркость = вес внимания. Пример: «Кот сидел на коврике».
// Q, K, V — три разные линейные проекции ОДНОГО входного вектора x
В стандартном MHA каждая из h голов имеет собственные K и V. При длинных контекстах это даёт огромный KV-cache. GQA делит головы на группы: внутри группы K и V разделяются между несколькими Q-головами — это сокращает память в несколько раз без заметной потери качества.
// MHA vs GQA: GQA разделяет KV между группами Q-голов — экономия KV-cache
import torch import torch.nn.functional as F def scaled_dot_product_attention( Q: torch.Tensor, K: torch.Tensor, V: torch.Tensor, mask: torch.Tensor = None ) -> tuple[torch.Tensor, torch.Tensor]: """ Q, K, V: (batch, heads, seq_len, d_k) mask: (batch, 1, 1, seq_len) — True там, где нужно заблокировать """ d_k = Q.size(-1) scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(d_k) if mask is not None: scores = scores.masked_fill(mask, float('-inf')) weights = F.softmax(scores, dim=-1) output = torch.matmul(weights, V) return output, weights # weights нужны для интерпретируемости class MultiHeadAttention(nn.Module): def __init__(self, d_model: int, num_heads: int): super().__init__() assert d_model % num_heads == 0 self.d_k = d_model // num_heads self.num_heads = num_heads self.W_Q = nn.Linear(d_model, d_model) self.W_K = nn.Linear(d_model, d_model) self.W_V = nn.Linear(d_model, d_model) self.W_O = nn.Linear(d_model, d_model) def split_heads(self, x): B, T, D = x.shape return x.view(B, T, self.num_heads, self.d_k).transpose(1, 2) def forward(self, x, mask=None): Q = self.split_heads(self.W_Q(x)) K = self.split_heads(self.W_K(x)) V = self.split_heads(self.W_V(x)) attn_out, _ = scaled_dot_product_attention(Q, K, V, mask) B, H, T, D = attn_out.shape out = attn_out.transpose(1, 2).reshape(B, T, H * D) return self.W_O(out)
// 04 · FFN · нормализация · residual
«Skip connection» из ResNet. Выход блока прибавляется к входу: x + Sublayer(x). Это решает проблему затухающих градиентов и позволяет обучать очень глубокие сети.
Нормирует активации по измерениям (а не по батчу, как Batch Norm). Стабилизирует обучение при переменных длинах последовательностей.
Два линейных слоя с активацией ReLU/GELU между ними. Применяется позиционно: одна и та же функция к каждому токену независимо.
Оригинал (Post-Norm): LayerNorm(x + sub(x)).
Современные модели (Pre-Norm): x + sub(LayerNorm(x)) — устойчивее при большом числе слоёв.
class TransformerEncoderLayer(nn.Module): def __init__(self, d_model=512, num_heads=8, d_ff=2048, dropout=0.1): super().__init__() self.self_attn = MultiHeadAttention(d_model, num_heads) # FFN: d_model → d_ff (×4) → d_model self.ffn = nn.Sequential( nn.Linear(d_model, d_ff), nn.GELU(), # GELU: плавнее ReLU, стандарт в BERT/GPT nn.Dropout(dropout), nn.Linear(d_ff, d_model), ) self.norm1 = nn.LayerNorm(d_model) self.norm2 = nn.LayerNorm(d_model) self.dropout = nn.Dropout(dropout) def forward(self, x, mask=None): # Pre-Norm вариант (как в GPT-2) x = x + self.dropout(self.self_attn(self.norm1(x), mask)) # residual x = x + self.dropout(self.ffn(self.norm2(x))) # residual return x
// АКТИВАЦИИ В FFN · ReLU (GPT-1) → GELU (BERT, GPT-2) → SwiGLU (LLaMA, PaLM)
class SwiGLUFFN(nn.Module): """ SwiGLU из статьи «GLU Variants Improve Transformer» (Noam Shazeer, 2020). Используется в LLaMA, PaLM, Mistral. В отличие от стандартного FFN — три матрицы вместо двух, но d_ff обычно уменьшают до ~2.67×d_model для того же числа FLOP. """ def __init__(self, d_model: int, d_ff: int): super().__init__() self.w_gate = nn.Linear(d_model, d_ff, bias=False) # gate projection self.w_up = nn.Linear(d_model, d_ff, bias=False) # up projection self.w_down = nn.Linear(d_ff, d_model, bias=False) # down projection def forward(self, x: torch.Tensor) -> torch.Tensor: gate = F.silu(self.w_gate(x)) # Swish(x·W_gate) up = self.w_up(x) # x·W_up fused = gate * up # поэлементное умножение (gate) return self.w_down(fused) # проекция обратно # F.silu(x) = x * sigmoid(x) — функция Swish # Идея: gate контролирует, какая часть информации пройдёт дальше
// POST-NORM vs PRE-NORM · современные LLM используют Pre-Norm для стабильности обучения
// 05 · masked self-attention
В декодере при авторегрессивной генерации модель должна предсказывать следующий токен, не видя будущих. Для этого в матрицу scores подставляется causal mask (нижнетреугольная маска): верхний треугольник заполняется -∞ → после softmax → 0.
def make_causal_mask(seq_len: int, device) -> torch.Tensor: """ Возвращает булеву маску (seq_len, seq_len): True → заблокировать (будущие позиции) False → разрешить """ # torch.triu: верхний треугольник, diagonal=1 → исключаем диагональ mask = torch.triu(torch.ones(seq_len, seq_len, device=device), diagonal=1).bool() return mask # (T, T) # Пример: последовательность длиной 4 mask = make_causal_mask(4, 'cpu') # tensor([[False, True, True, True], # [False, False, True, True], # [False, False, False, True], # [False, False, False, False]])
// 06 · linear + softmax
После стека декодеров вектор последнего токена (размер d_model) проецируется линейным слоём на пространство словаря (vocab_size) — это logits. Затем softmax даёт вероятностное распределение по токенам.
Weight tying: во многих моделях (GPT-2, LLaMA) матрица W_vocab = Embedding_matrixT. Это сокращает параметры и улучшает качество.
Temperature τ: τ < 1 → распределение «острее», детерминированнее. τ > 1 → «мягче», креативнее. τ = 0 → greedy (argmax). Используется только при инференсе, не при обучении.
def sample_next_token(logits: torch.Tensor, temperature: float = 1.0, top_k: int = 50) -> int: """logits: (vocab_size,)""" logits = logits / max(temperature, 1e-8) # масштабируем # Top-k фильтрация: оставляем только top_k токенов values, _ = torch.topk(logits, top_k) threshold = values[-1] logits = logits.masked_fill(logits < threshold, float('-inf')) probs = F.softmax(logits, dim=-1) next_token = torch.multinomial(probs, num_samples=1).item() return next_token
top_p=0.9.
// 07 · hugging face transformers
Hugging Face Transformers — библиотека с тысячами готовых моделей. pipeline() — высокоуровневый API, скрывающий токенизацию, батчинг и постобработку.
from transformers import pipeline # Загружает: distilbert-base-uncased-finetuned-sst-2-english # (или любой другой text-classification checkpoint) classifier = pipeline( "sentiment-analysis", model="distilbert/distilbert-base-uncased-finetuned-sst-2-english", device=0, # GPU; -1 для CPU truncation=True, ) texts = [ "Transformers make NLP incredibly accessible!", "The model was slow and the output was poor.", ] results = classifier(texts, batch_size=8) for text, r in zip(texts, results): print(f"{text[:40]}... → {r['label']} ({r['score']:.3f})") # → Transformers make NLP... → POSITIVE (0.9998) # → The model was slow... → NEGATIVE (0.9996) ## Что происходит под капотом TextClassificationPipeline: # 1. tokenizer(texts, ...) → input_ids, attention_mask # 2. model(**inputs) → BaseModelOutput(logits=...) # 3. softmax(logits) → вероятности по классам # 4. argmax → id2label → "POSITIVE" / "NEGATIVE"
from transformers import AutoTokenizer, AutoModelForCausalLM import torch model_id = "openai-community/gpt2" tokenizer = AutoTokenizer.from_pretrained(model_id) model = AutoModelForCausalLM.from_pretrained(model_id) model.eval() prompt = "The transformer architecture revolutionized NLP because" inputs = tokenizer(prompt, return_tensors="pt") with torch.no_grad(): output_ids = model.generate( **inputs, max_new_tokens=80, do_sample=True, temperature=0.8, top_p=0.9, # nucleus sampling repetition_penalty=1.1, # штраф за повторы pad_token_id=tokenizer.eos_token_id, ) generated = tokenizer.decode(output_ids[0], skip_special_tokens=True) print(generated) ## Автоматически выполняется при каждом шаге generate(): # 1. Encoder-кэш KV не пересчитывается (KV-cache) # 2. Causal mask применяется автоматически # 3. Top-p / top-k / temperature применяются к logits # 4. EOS-токен останавливает генерацию
from transformers import AutoTokenizer, AutoModel import torch tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased") model = AutoModel.from_pretrained("bert-base-uncased") text = "Hello, transformer world!" enc = tokenizer(text, return_tensors="pt") # enc содержит: input_ids, attention_mask, token_type_ids print(enc.input_ids) # tensor([[101, 7592, 1010, 19081, 2088, 999, 102]]) print(tokenizer.convert_ids_to_tokens(enc.input_ids[0].tolist())) # ['[CLS]', 'hello', ',', 'transform', '##er', 'world', '!', '[SEP]'] with torch.no_grad(): outputs = model(**enc) # last_hidden_state: (batch, seq_len, hidden_size) = (1, 8, 768) # pooler_output: (batch, hidden_size) — [CLS]-токен через linear+tanh cls_emb = outputs.last_hidden_state[:, 0, :] # [CLS] → sentence embedding print(cls_emb.shape) # torch.Size([1, 768])
tokenizer(texts, padding=True, truncation=True, max_length=512). Паддинг маскируется через attention_mask — модель не «видит» PAD-токены благодаря маске в attention.
// 08 · кастомный train-loop · pytorch lstm
LSTM (Long Short-Term Memory) — классическая рекуррентная архитектура. В отличие от трансформера, обрабатывает последовательность шаг за шагом. Пример ниже показывает, как устроен полный train-loop без высокоуровневых абстракций — это даёт понимание того, что HuggingFace делает автоматически.
import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset # ── 1. Модель ────────────────────────────────────────────── class LSTMClassifier(nn.Module): def __init__(self, vocab_size, embed_dim=128, hidden=256, num_layers=2, num_classes=2): super().__init__() self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0) self.lstm = nn.LSTM( embed_dim, hidden, num_layers=num_layers, batch_first=True, # (batch, seq, features) dropout=0.3, bidirectional=True, # BiLSTM: читает в обе стороны ) self.fc = nn.Linear(hidden * 2, num_classes) # *2 за bidirectional self.dropout = nn.Dropout(0.3) def forward(self, x): emb = self.dropout(self.embedding(x)) # (B, T, E) out, (h_n, _) = self.lstm(emb) # out: (B, T, 2H) # Берём последний скрытый слой обоих направлений h_fwd = h_n[-2] # последний forward-слой h_bwd = h_n[-1] # последний backward-слой h = torch.cat([h_fwd, h_bwd], dim=-1) # (B, 2H) return self.fc(self.dropout(h)) # ── 2. Данные (заглушка) ─────────────────────────────────── VOCAB, SEQ, N = 10000, 128, 1000 X = torch.randint(1, VOCAB, (N, SEQ)) y = torch.randint(0, 2, (N,)) loader = DataLoader(TensorDataset(X, y), batch_size=64, shuffle=True) # ── 3. Инициализация ─────────────────────────────────────── device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = LSTMClassifier(VOCAB).to(device) optim = torch.optim.AdamW(model.parameters(), lr=3e-4, weight_decay=1e-2) sched = torch.optim.lr_scheduler.CosineAnnealingLR(optim, T_max=5) crit = nn.CrossEntropyLoss() # ── 4. Train-loop ────────────────────────────────────────── for epoch in range(5): model.train() total_loss = 0.0 for xb, yb in loader: xb, yb = xb.to(device), yb.to(device) optim.zero_grad() # обнуляем накопленные градиенты logits = model(xb) # forward pass loss = crit(logits, yb) # CrossEntropy = LogSoftmax + NLLLoss loss.backward() # вычисляем градиенты (autograd) nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) # gradient clipping optim.step() # обновляем веса total_loss += loss.item() sched.step() avg = total_loss / len(loader) print(f"Epoch {epoch+1:02d} | loss={avg:.4f} | lr={sched.get_last_lr()[0]:.2e}") # ── 5. Инференс ──────────────────────────────────────────── model.eval() with torch.no_grad(): test_x = torch.randint(1, VOCAB, (4, SEQ)).to(device) preds = model(test_x).argmax(dim=-1) print("Predicted classes:", preds.tolist())
Trainer делает внутри. Понимание этого цикла критично для дебага, кастомных loss-функций и mixed-precision обучения.
// 09 · сравнение
| Характеристика | LSTM (RNN) | Transformer (с нуля) | HuggingFace pipeline |
|---|---|---|---|
| Параллелизм при обучении | ❌ Последовательно по времени | ✅ Полный параллелизм по sequence | ✅ (внутри трансформер) |
| Длинные зависимости | ⚠️ Затухает после ~200 токенов | ✅ O(1) между любыми токенами | ✅ до context window модели |
| Сложность (время) | O(n) по длине | O(n²) по длине (attention) | O(n²) или O(n) с Flash Attention |
| Память при инференсе | Константная (hidden state) | Линейная (KV-cache растёт) | Управляется автоматически |
| Строк кода для задачи | 50–200 строк (полный loop) | 200–500 строк архитектуры | 3–10 строк |
| Контроль над обучением | ✅ Полный | ✅ Полный | ⚠️ Ограничен (нужен Trainer API) |
| Interpretability | Сложно (hidden state непрозрачен) | Attention weights (с оговорками) | Зависит от модели; есть SHAP/LIME интеграции |
| Когда использовать | Потоки данных, онлайн-обработка, малый датасет | Исследования, кастомные архитектуры | Прод, быстрый прототип, fine-tuning |
Contrastive learning, ranking loss, multi-task — нужен кастомный loop или переопределение compute_loss в Trainer.
Мультимодальные данные (текст + таблицы + изображения) требуют кастомного Collator и архитектуры.
Для экспериментов с новыми механизмами attention, нормализации или positional encoding нужен чистый PyTorch.
pipeline() → затем Trainer для fine-tuning → потом кастомный loop. Каждый уровень абстракции уместен для своей задачи.