PyTorch: LSTM / GRU
Рекуррентные нейронные сети (RNNs) сталкиваются с проблемой исчезающего градиента при обработке последовательных данных, что затрудняет изучение долгосрочных зависимостей.
Долгая краткосрочная память (LSTM)иУправляемый рекуррентный блок (GRU)Благодаря введению механизмов гейтирования они решают эту проблему и являются ключевыми моделями для обработки последовательностных задач, таких как временные ряды, естественный язык и речь.
1. Ограничения RNN и механизмов гейтирования
Стандартная RNN объединяет текущий вход с предыдущим скрытым состоянием на каждом временном шаге, чтобы вычислить новое скрытое состояние:
\[h_t = \tanh(W_h \cdot h_{t-1} + W_x \cdot x_t + b)\]
У этой структуры есть две ключевые проблемы:
Исчезающий градиент: При обратном распространении градиент шаг за шагом умножается на матрицу весов. Когда последовательность длинная, градиент экспоненциально затухает, из-за чего параметры на ранних временных шагах почти не обновляются, и модель не может выучить долгосрочные зависимости.
Взрывающийся градиент: Когда наибольшее собственное значение матрицы весов больше 1, градиент экспоненциально возрастает при обратном распространении, что делает обучение нестабильным (обычно это смягчается отсечением градиентов).
Основная идея механизмов вентилей — ввести обучаемые «переключатели», которые позволяют сети автономно решать: на текущем временном шаге какую информацию следует запомнить, какую следует забыть, а какую новую информацию следует записать в память.
LSTM использует три вентиля (вентиль забывания, входной вентиль, выходной вентиль) плюс отдельное состояние ячейки; GRU упрощает структуру до двух вентилей (вентиль сброса, вентиль обновления), с меньшим числом параметров и более быстрым обучением.
2. Принцип LSTM
2.1 Основная структура и три вентиля
LSTM поддерживает два вектора состояния, передаваемых между временными шагами:
- Состояние ячейки\(c_t\): носитель долговременной памяти, в котором информация может передаваться почти без потерь
- Скрытое состояние\(h_t\): кратковременная память, а также выход текущего временного шага
Все три вентиля — это линейные преобразования, активируемые функцией Sigmoid, с выходными значениями от 0 до 1, выполняющие роль «клапанов»:
遗忘门（Forget Gate）：决定从细胞状态中丢弃哪些信息 输入门（Input Gate）：决定将哪些新信息写入细胞状态 输出门（Output Gate）：决定基于细胞状态输出什么
2.2 Формулы прямого вычисления
\[
\begin{aligned}
\text{Вход:} & x_t \text{ (вход текущего временного шага)}, h_{t-1} \text{ (предыдущее скрытое состояние)}, c_{t-1} \text{ (предыдущее состояние ячейки)} \\
\text{Вентиль забывания:} & f_t = \sigma(W_f \cdot [h_{t-1}, x_t] + b_f) \\
\text{Входной вентиль:} & i_t = \sigma(W_i \cdot [h_{t-1}, x_t] + b_i) \\
\text{Значение-кандидат:} & \tilde{g}_t = \tanh(W_g \cdot [h_{t-1}, x_t] + b_g) \\
\text{Выходной вентиль:} & o_t = \sigma(W_o \cdot [h_{t-1}, x_t] + b_o) \\
\text{Обновление состояния ячейки:} & c_t = f_t \odot c_{t-1} + i_t \odot \tilde{g}_t \\
\text{Обновление скрытого состояния:} & h_t = o_t \odot \tanh(c_t)
\end{aligned}
\]
где \(\odot\) обозначает поэлементное умножение (произведение Адамара), а \(\sigma\) обозначает сигмоидную функцию.
f_t ⊙ c_{t-1}Интерпретация логики вычислений:i_t ⊙ g_t: Вентиль забывания решает, сколько исторической памяти сохранить; при значении, близком к 0, он забывает, при значении, близком к 1, сохраняетg_t: Входной вентиль решает, сколько новой информации записать,o_t ⊙ tanh(c_t)является кандидатом нового содержимого
3. Принцип GRU
3.1 Основная структура и два вентиля
3.1 Основная структура и два вентиляGRU объединяет вентиль забывания и входной вентиль LSTM ввентиль обновления
重置门（Reset Gate）：决定忽略多少历史状态来计算候选隐藏状态 更新门（Update Gate）：决定保留多少历史状态，写入多少新状态
3.2 Формулы прямого вычисления
\[
\begin{aligned}
\text{Вход:} & x_t \text{ (вход текущего временного шага)}, h_{t-1} \text{ (предыдущее скрытое состояние)} \\
\text{Вентиль сброса:} & r_t = \sigma(W_r \cdot [h_{t-1}, x_t] + b_r) \\
\text{Вентиль обновления:} & z_t = \sigma(W_z \cdot [h_{t-1}, x_t] + b_z) \\
\text{Значение-кандидат:} & \tilde{h}_t = \tanh(W_h \cdot [r_t \odot h_{t-1}, x_t] + b_h) \\
\text{Обновление скрытого состояния:} & h_t = (1 - z_t) \odot h_{t-1} + z_t \odot \tilde{h}_t
\end{aligned}
\]
Интерпретация логики вычислений:
- Гейт сброса
r_tКогда значение близко к 0, состояние-кандидатh̃_tпочти не зависит от истории, что эквивалентно началу заново - Гейт обновления
z_tКогда значение близко к 1, новое состояние принимает больше значения-кандидата; когда близко к 0, оно сохраняет больше исторического состояния - У GRU нет отдельного состояния ячейки, и количество её параметров составляет около 75% от количества параметров LSTM
4. LSTM в PyTorch
В этом разделе подробно описаны параметры, формы входа/выхода и методы инициализации скрытого состояния nn.LSTM.
4.1 Подробное объяснение параметров nn.LSTM
Пример
import torch
import torch.nn as nn
lstm = nn.LSTM(
input_size=64, # Размерность входного вектора на каждом временном шаге
hidden_size=128, # Размерность скрытого состояния (и состояния ячейки)
num_layers=2, # Количество слоёв в стеке, по умолчанию 1
bias=True, # Использовать ли смещения, по умолчанию True
batch_first=False, # Определяет, находится ли батч в первом измерении формы входа/выхода, по умолчанию False
dropout=0.0, # Вероятность dropout между слоями (действует только при num_layers > 1)
bidirectional=False, # Использовать ли двунаправленную LSTM, по умолчанию False
proj_size=0, # Размерность слоя проекции (LSTM с проекцией), по умолчанию 0 означает, что не используется
)
# Просмотр количества параметров
total_params = sum(p.numel() for p in lstm.parameters())
print(f"Количество параметров LSTM: {total_params:,}")
# Когда input_size=64, hidden_size=128, num_layers=2, это приблизительно 197,632
import torch.nn as nn
lstm = nn.LSTM(
input_size=64, # Размерность входного вектора на каждом временном шаге
hidden_size=128, # Размерность скрытого состояния (и состояния ячейки)
num_layers=2, # Количество слоёв в стеке, по умолчанию 1
bias=True, # Использовать ли смещения, по умолчанию True
batch_first=False, # Определяет, находится ли батч в первом измерении формы входа/выхода, по умолчанию False
dropout=0.0, # Вероятность dropout между слоями (действует только при num_layers > 1)
bidirectional=False, # Использовать ли двунаправленную LSTM, по умолчанию False
proj_size=0, # Размерность слоя проекции (LSTM с проекцией), по умолчанию 0 означает, что не используется
)
# Просмотр количества параметров
total_params = sum(p.numel() for p in lstm.parameters())
print(f"Количество параметров LSTM: {total_params:,}")
# Когда input_size=64, hidden_size=128, num_layers=2, это приблизительно 197,632
Формула оценки количества параметров (однослойная однонаправленная):
\[
\begin{aligned}
\text{Параметров на слой} & = 4 \times (hidden\_size \times input\_size + hidden\_size \times hidden\_size + hidden\_size) \\
& = 4 \times hidden\_size \times (input\_size + hidden\_size + 1)
\end{aligned}
\]
где 4 соответствует четырём группам весовых матриц: забывающий шлюз, входной шлюз, кандидатное значение и выходной шлюз.
4.2 Формы входа и выхода
Это самая подверженная ошибкам часть использования LSTM, поэтому требуется особое внимание.batch_firstВлияние параметров.
Пример
import torch
import torch.nn as nn
# ── batch_first=False (по умолчанию) ────────────────────────
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=False)
# Форма входных данных: (seq_len, batch_size, input_size)
seq_len, batch_size, input_size = 10, 4, 32
x = torch.randn(seq_len, batch_size, input_size)
output, (h_n, c_n) = lstm(x)
print(f"форма выхода: {output.shape}")
# torch.Size([10, 4, 64]) → (seq_len, batch_size, hidden_size)
# Выход скрытого состояния на каждом временном шаге
print(f"форма h_n: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers * num_directions, batch_size, hidden_size)
# Скрытое состояние на последнем временном шаге
print(f"форма c_n: {c_n.shape}")
# torch.Size([1, 4, 64]) → такой же, как h_n, состояние ячейки на последнем временном шаге
# ── batch_first=True (рекомендуется, более интуитивно) ────────
lstm_bf = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Форма входа: (batch_size, seq_len, input_size)
x = torch.randn(batch_size, seq_len, input_size)
output, (h_n, c_n) = lstm_bf(x)
print(f"форма выхода: {output.shape}")
# torch.Size([4, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"форма h_n: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers, batch_size, hidden_size)
# Примечание: форма h_n не зависит от batch_first
# ── Форма выхода многослойной двунаправленной LSTM ──────────────────────
lstm_bd = nn.LSTM(input_size=32, hidden_size=64,
num_layers=3, bidirectional=True, batch_first=True)
x = torch.randn(batch_size, seq_len, input_size)
output, (h_n, c_n) = lstm_bd(x)
print(f"форма выхода: {output.shape}")
# torch.Size([4, 10, 128])
# hidden_size × 2 = 128, потому что двунаправленная конкатенация
print(f"форма h_n: {h_n.shape}")
# torch.Size([6, 4, 64])
# num_layers × num_directions = 3 × 2 = 6
import torch.nn as nn
# ── batch_first=False (по умолчанию) ────────────────────────
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=False)
# Форма входных данных: (seq_len, batch_size, input_size)
seq_len, batch_size, input_size = 10, 4, 32
x = torch.randn(seq_len, batch_size, input_size)
output, (h_n, c_n) = lstm(x)
print(f"форма выхода: {output.shape}")
# torch.Size([10, 4, 64]) → (seq_len, batch_size, hidden_size)
# Выход скрытого состояния на каждом временном шаге
print(f"форма h_n: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers * num_directions, batch_size, hidden_size)
# Скрытое состояние на последнем временном шаге
print(f"форма c_n: {c_n.shape}")
# torch.Size([1, 4, 64]) → такой же, как h_n, состояние ячейки на последнем временном шаге
# ── batch_first=True (рекомендуется, более интуитивно) ────────
lstm_bf = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Форма входа: (batch_size, seq_len, input_size)
x = torch.randn(batch_size, seq_len, input_size)
output, (h_n, c_n) = lstm_bf(x)
print(f"форма выхода: {output.shape}")
# torch.Size([4, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"форма h_n: {h_n.shape}")
# torch.Size([1, 4, 64]) → (num_layers, batch_size, hidden_size)
# Примечание: форма h_n не зависит от batch_first
# ── Форма выхода многослойной двунаправленной LSTM ──────────────────────
lstm_bd = nn.LSTM(input_size=32, hidden_size=64,
num_layers=3, bidirectional=True, batch_first=True)
x = torch.randn(batch_size, seq_len, input_size)
output, (h_n, c_n) = lstm_bd(x)
print(f"форма выхода: {output.shape}")
# torch.Size([4, 10, 128])
# hidden_size × 2 = 128, потому что двунаправленная конкатенация
print(f"форма h_n: {h_n.shape}")
# torch.Size([6, 4, 64])
# num_layers × num_directions = 3 × 2 = 6
Сводка размеров выходных тензоров:
| Переменные | batch_first=False | batch_first=True |
|---|---|---|
output |
(seq_len, N, H * D) |
(N, seq_len, H * D) |
h_n |
(L * D, N, H) |
(L * D, N, H) |
c_n |
(L * D, N, H) |
(L * D, N, H) |
\(N\) = batch_size, \(H\) = hidden_size, \(L\) = num_layers, \(D\) = 2 (двунаправленный) или 1 (однонаправленный)
4.3 Инициализация скрытого состояния
Пример
import torch
import torch.nn as nn
lstm = nn.LSTM(input_size=32, hidden_size=64, num_layers=2, batch_first=True)
batch_size = 8
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
lstm = lstm.to(device)
# Метод 1: не передавать начальное состояние; PyTorch автоматически использует нулевую инициализацию
x = torch.randn(batch_size, 10, 32).to(device)
output, (h_n, c_n) = lstm(x)
# Метод 2: вручную инициализировать нулями (эквивалентно методу 1, но явно показывает намерение)
num_layers, num_directions = 2, 1
h_0 = torch.zeros(num_layers * num_directions, batch_size, 64).to(device)
c_0 = torch.zeros(num_layers * num_directions, batch_size, 64).to(device)
output, (h_n, c_n) = lstm(x, (h_0, c_0))
# Метод 3: режим с сохранением состояния — передача состояния между батчами
# Подходит для нарезки сверхдлинных последовательностей (языковые модели, генерация длинного текста и т. д.)
# Нужно вызвать detach(), чтобы отсечь граф вычислений от предыдущего батча и предотвратить утечку памяти GPU
h, c = h_0, c_0
for batch_x in data_loader:
batch_x = batch_x.to(device)
output, (h, c) = lstm(batch_x, (h, c))
h = h.detach() # Отсечь градиенты, сохранить только значения
c = c.detach()
# Метод 4: использовать инициализацию Xavier или нормальным распределением (в некоторых сценариях сходится быстрее)
def init_hidden(lstm_module, batch_size, device):
num_layers = lstm_module.num_layers
hidden_size = lstm_module.hidden_size
directions = 2 if lstm_module.bidirectional else 1
h = torch.zeros(num_layers * directions, batch_size, hidden_size, device=device)
c = torch.zeros(num_layers * directions, batch_size, hidden_size, device=device)
nn.init.orthogonal_(h) # Ортогональная инициализация помогает стабилизировать обучение
return h, c
import torch.nn as nn
lstm = nn.LSTM(input_size=32, hidden_size=64, num_layers=2, batch_first=True)
batch_size = 8
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
lstm = lstm.to(device)
# Метод 1: не передавать начальное состояние; PyTorch автоматически использует нулевую инициализацию
x = torch.randn(batch_size, 10, 32).to(device)
output, (h_n, c_n) = lstm(x)
# Метод 2: вручную инициализировать нулями (эквивалентно методу 1, но явно показывает намерение)
num_layers, num_directions = 2, 1
h_0 = torch.zeros(num_layers * num_directions, batch_size, 64).to(device)
c_0 = torch.zeros(num_layers * num_directions, batch_size, 64).to(device)
output, (h_n, c_n) = lstm(x, (h_0, c_0))
# Метод 3: режим с сохранением состояния — передача состояния между батчами
# Подходит для нарезки сверхдлинных последовательностей (языковые модели, генерация длинного текста и т. д.)
# Нужно вызвать detach(), чтобы отсечь граф вычислений от предыдущего батча и предотвратить утечку памяти GPU
h, c = h_0, c_0
for batch_x in data_loader:
batch_x = batch_x.to(device)
output, (h, c) = lstm(batch_x, (h, c))
h = h.detach() # Отсечь градиенты, сохранить только значения
c = c.detach()
# Метод 4: использовать инициализацию Xavier или нормальным распределением (в некоторых сценариях сходится быстрее)
def init_hidden(lstm_module, batch_size, device):
num_layers = lstm_module.num_layers
hidden_size = lstm_module.hidden_size
directions = 2 if lstm_module.bidirectional else 1
h = torch.zeros(num_layers * directions, batch_size, hidden_size, device=device)
c = torch.zeros(num_layers * directions, batch_size, hidden_size, device=device)
nn.init.orthogonal_(h) # Ортогональная инициализация помогает стабилизировать обучение
return h, c
5. GRU в PyTorch
Интерфейс GRU почти идентичен LSTM; главное отличие в том, что нет состояния ячейки.c。
5.1 Подробное объяснение параметров nn.GRU
Пример
import torch.nn as nn
gru = nn.GRU(
input_size=64,
hidden_size=128,
num_layers=2,
bias=True,
batch_first=True, # Рекомендуется установить True
dropout=0.3, # Dropout между слоями
bidirectional=False,
)
gru = nn.GRU(
input_size=64,
hidden_size=128,
num_layers=2,
bias=True,
batch_first=True, # Рекомендуется установить True
dropout=0.3, # Dropout между слоями
bidirectional=False,
)
5.2 Пример базового использования
Пример
import torch
import torch.nn as nn
gru = nn.GRU(input_size=32, hidden_size=64, batch_first=True)
batch_size, seq_len = 8, 10
x = torch.randn(batch_size, seq_len, 32)
# GRU возвращает только output и h_n, без c_n
output, h_n = gru(x)
print(f"форма выхода: {output.shape}")
# torch.Size([8, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"форма h_n: {h_n.shape}")
# torch.Size([1, 8, 64]) → (num_layers, batch_size, hidden_size)
# Взять выход на последнем временном шаге (для таких задач, как классификация)
last_hidden = output[:, -1, :] # (batch_size, hidden_size)
# Или эквивалентно:
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
import torch.nn as nn
gru = nn.GRU(input_size=32, hidden_size=64, batch_first=True)
batch_size, seq_len = 8, 10
x = torch.randn(batch_size, seq_len, 32)
# GRU возвращает только output и h_n, без c_n
output, h_n = gru(x)
print(f"форма выхода: {output.shape}")
# torch.Size([8, 10, 64]) → (batch_size, seq_len, hidden_size)
print(f"форма h_n: {h_n.shape}")
# torch.Size([1, 8, 64]) → (num_layers, batch_size, hidden_size)
# Взять выход на последнем временном шаге (для таких задач, как классификация)
last_hidden = output[:, -1, :] # (batch_size, hidden_size)
# Или эквивалентно:
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
6. Варианты LSTM и GRU
В этом разделе представлены двунаправленные, многослойные стекированные и многослойные структуры с Dropout.
6.1 Двунаправленные LSTM / GRU
Двунаправленные модели одновременно обрабатывают информацию из прямого и обратного направлений последовательности. Выход на каждом временном шаге содержит прошлый и будущий контекст, что делает их подходящими для задач, требующих глобального контекста, таких как классификация текста и распознавание именованных сущностей.
Пример
import torch
import torch.nn as nn
# Двунаправленная LSTM
bilstm = nn.LSTM(
input_size=32,
hidden_size=64,
num_layers=2,
batch_first=True,
bidirectional=True, # Включить двунаправленность
)
x = torch.randn(8, 10, 32)
output, (h_n, c_n) = bilstm(x)
print(f"форма выхода: {output.shape}")
# torch.Size([8, 10, 128])
# Прямое 64-мерное + обратное 64-мерное = 128-мерное
print(f"форма h_n: {h_n.shape}")
# torch.Size([4, 8, 64])
# num_layers(2) × num_directions(2) = 4
# Разделить конечные скрытые состояния прямого и обратного направлений
# порядок h_n: [прямой слой 0, обратный слой 0, прямой слой 1, обратный слой 1]
h_forward = h_n[-2, :, :] # (batch_size, hidden_size) последний слой прямого направления
h_backward = h_n[-1, :, :] # (batch_size, hidden_size) последний слой обратного направления
h_combined = torch.cat([h_forward, h_backward], dim=-1) # (batch_size, 128)
# Использование двунаправленной GRU точно такое же
bigru = nn.GRU(input_size=32, hidden_size=64,
num_layers=2, batch_first=True, bidirectional=True)
output, h_n = bigru(x)
import torch.nn as nn
# Двунаправленная LSTM
bilstm = nn.LSTM(
input_size=32,
hidden_size=64,
num_layers=2,
batch_first=True,
bidirectional=True, # Включить двунаправленность
)
x = torch.randn(8, 10, 32)
output, (h_n, c_n) = bilstm(x)
print(f"форма выхода: {output.shape}")
# torch.Size([8, 10, 128])
# Прямое 64-мерное + обратное 64-мерное = 128-мерное
print(f"форма h_n: {h_n.shape}")
# torch.Size([4, 8, 64])
# num_layers(2) × num_directions(2) = 4
# Разделить конечные скрытые состояния прямого и обратного направлений
# порядок h_n: [прямой слой 0, обратный слой 0, прямой слой 1, обратный слой 1]
h_forward = h_n[-2, :, :] # (batch_size, hidden_size) последний слой прямого направления
h_backward = h_n[-1, :, :] # (batch_size, hidden_size) последний слой обратного направления
h_combined = torch.cat([h_forward, h_backward], dim=-1) # (batch_size, 128)
# Использование двунаправленной GRU точно такое же
bigru = nn.GRU(input_size=32, hidden_size=64,
num_layers=2, batch_first=True, bidirectional=True)
output, h_n = bigru(x)
6.2 Многослойное стекирование
Пример
import torch
import torch.nn as nn
# 3-слойный стековый LSTM
deep_lstm = nn.LSTM(
input_size=32,
hidden_size=128,
num_layers=3, # Стекировать 3 слоя
batch_first=True,
)
x = torch.randn(8, 20, 32)
output, (h_n, c_n) = deep_lstm(x)
print(f"форма выхода: {output.shape}")
# torch.Size([8, 20, 128]) содержит только выход верхнего слоя
print(f"форма h_n: {h_n.shape}")
# torch.Size([3, 8, 128]) содержит конечное скрытое состояние каждого слоя
# Получить конечное скрытое состояние каждого слоя
h_layer1 = h_n[0] # (batch_size, hidden_size) первый слой
h_layer2 = h_n[1] # (batch_size, hidden_size) второй слой
h_layer3 = h_n[2] # (batch_size, hidden_size) третий слой (верхний слой)
import torch.nn as nn
# 3-слойный стековый LSTM
deep_lstm = nn.LSTM(
input_size=32,
hidden_size=128,
num_layers=3, # Стекировать 3 слоя
batch_first=True,
)
x = torch.randn(8, 20, 32)
output, (h_n, c_n) = deep_lstm(x)
print(f"форма выхода: {output.shape}")
# torch.Size([8, 20, 128]) содержит только выход верхнего слоя
print(f"форма h_n: {h_n.shape}")
# torch.Size([3, 8, 128]) содержит конечное скрытое состояние каждого слоя
# Получить конечное скрытое состояние каждого слоя
h_layer1 = h_n[0] # (batch_size, hidden_size) первый слой
h_layer2 = h_n[1] # (batch_size, hidden_size) второй слой
h_layer3 = h_n[2] # (batch_size, hidden_size) третий слой (верхний слой)
6.3 Многослойная структура с Dropout
nn.LSTMВстроеннаяdropoutпараметр применяется только ксвязям между слоями, а не к выходу последнего слоя. Если вам нужно добавить Dropout также и после последнего слоя, вы должны добавить его вручную:
Пример
import torch
import torch.nn as nn
class StackedLSTM(nn.Module):
"""
Многослойная LSTM + межслойный Dropout + выходной Dropout
"""
def __init__(self, input_size, hidden_size, num_layers,
num_classes, dropout=0.3):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
# Когда num_layers=1, dropout не действует; установите его в 0, чтобы избежать предупреждений
)
self.dropout = nn.Dropout(dropout) # Dropout после последнего слоя
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x):
output, (h_n, c_n) = self.lstm(x)
# Взять скрытое состояние на последнем временном шаге
last_output = output[:, -1, :] # (batch_size, hidden_size)
last_output = self.dropout(last_output)
return self.fc(last_output)
model = StackedLSTM(input_size=64, hidden_size=128,
num_layers=3, num_classes=5, dropout=0.3)
x = torch.randn(16, 20, 64)
print(model(x).shape) # torch.Size([16, 5])
import torch.nn as nn
class StackedLSTM(nn.Module):
"""
Многослойная LSTM + межслойный Dropout + выходной Dropout
"""
def __init__(self, input_size, hidden_size, num_layers,
num_classes, dropout=0.3):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
# Когда num_layers=1, dropout не действует; установите его в 0, чтобы избежать предупреждений
)
self.dropout = nn.Dropout(dropout) # Dropout после последнего слоя
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x):
output, (h_n, c_n) = self.lstm(x)
# Взять скрытое состояние на последнем временном шаге
last_output = output[:, -1, :] # (batch_size, hidden_size)
last_output = self.dropout(last_output)
return self.fc(last_output)
model = StackedLSTM(input_size=64, hidden_size=128,
num_layers=3, num_classes=5, dropout=0.3)
x = torch.randn(16, 20, 64)
print(model(x).shape) # torch.Size([16, 5])
7. Обработка последовательностей переменной длины
В реальных задачах последовательности внутри одного батча обычно имеют разную длину. PyTorch предоставляетpack_padded_sequenceиpad_packed_sequenceдля решения этой проблемы, не позволяя LSTM выполнять некорректные вычисления на позициях заполнения.
7.1 pack_padded_sequence
Пример
import torch
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence, pad_sequence
# Симулировать последовательности переменной длины в батче (уже отсортированные от самой длинной к самой короткой)
seq1 = torch.randn(5, 32) # Длина последовательности 5
seq2 = torch.randn(3, 32) # Длина последовательности 3
seq3 = torch.randn(2, 32) # Длина последовательности 2
# pad_sequence автоматически дополняет нулями, чтобы выровнять по самой длинной последовательности
# Когда batch_first=True, форма выхода — (batch_size, max_seq_len, input_size)
padded = pad_sequence([seq1, seq2, seq3], batch_first=True, padding_value=0.0)
lengths = torch.tensor([5, 3, 2])
print(f"Форма после дополнения: {padded.shape}") # torch.Size([3, 5, 32])
# pack_padded_sequence: сжимает padding, сообщает LSTM реальные длины
packed = pack_padded_sequence(
padded,
lengths=lengths,
batch_first=True,
enforce_sorted=True, # Последовательности должны быть отсортированы в порядке убывания длины
# enforce_sorted=False # Допускает любой порядок (автоматически сортируется внутри), рекомендуется установить False
)
print(type(packed)) # <class 'torch.nn.utils.rnn.PackedSequence'>
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence, pad_sequence
# Симулировать последовательности переменной длины в батче (уже отсортированные от самой длинной к самой короткой)
seq1 = torch.randn(5, 32) # Длина последовательности 5
seq2 = torch.randn(3, 32) # Длина последовательности 3
seq3 = torch.randn(2, 32) # Длина последовательности 2
# pad_sequence автоматически дополняет нулями, чтобы выровнять по самой длинной последовательности
# Когда batch_first=True, форма выхода — (batch_size, max_seq_len, input_size)
padded = pad_sequence([seq1, seq2, seq3], batch_first=True, padding_value=0.0)
lengths = torch.tensor([5, 3, 2])
print(f"Форма после дополнения: {padded.shape}") # torch.Size([3, 5, 32])
# pack_padded_sequence: сжимает padding, сообщает LSTM реальные длины
packed = pack_padded_sequence(
padded,
lengths=lengths,
batch_first=True,
enforce_sorted=True, # Последовательности должны быть отсортированы в порядке убывания длины
# enforce_sorted=False # Допускает любой порядок (автоматически сортируется внутри), рекомендуется установить False
)
print(type(packed)) # <class 'torch.nn.utils.rnn.PackedSequence'>
7.2 pad_packed_sequence
Пример
lstm = nn.LSTM(input_size=32, hidden_size=64, batch_first=True)
# Передать PackedSequence в LSTM
packed_output, (h_n, c_n) = lstm(packed)
# pad_packed_sequence: восстановить в тензор с паддингом
output, output_lengths = pad_packed_sequence(packed_output, batch_first=True)
print(f"Восстановленная форма выхода: {output.shape}")
# torch.Size([3, 5, 64]) → (batch_size, max_seq_len, hidden_size)
# Выход в позициях паддинга равен 0
print(f"Фактические длины каждой последовательности: {output_lengths}")
# tensor([5, 3, 2])
# Передать PackedSequence в LSTM
packed_output, (h_n, c_n) = lstm(packed)
# pad_packed_sequence: восстановить в тензор с паддингом
output, output_lengths = pad_packed_sequence(packed_output, batch_first=True)
print(f"Восстановленная форма выхода: {output.shape}")
# torch.Size([3, 5, 64]) → (batch_size, max_seq_len, hidden_size)
# Выход в позициях паддинга равен 0
print(f"Фактические длины каждой последовательности: {output_lengths}")
# tensor([5, 3, 2])
7.3 Полный конвейер обработки последовательностей переменной длины
Пример
import torch
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence
class LSTMClassifier(nn.Module):
def __init__(self, vocab_size, embed_dim, hidden_size, num_classes, padding_idx=0):
super().__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=padding_idx)
self.lstm = nn.LSTM(embed_dim, hidden_size, batch_first=True)
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x, lengths):
# x: (batch_size, max_seq_len) — индексы слов
embedded = self.embedding(x) # (batch_size, max_seq_len, embed_dim)
# Упаковка
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
# Прямой проход LSTM (пропуск позиций паддинга)
packed_output, (h_n, c_n) = self.lstm(packed)
# Метод 1: использовать h_n последнего слоя как представление последовательности
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
# Метод 2: После распаковки взять выход на истинной последней позиции каждой последовательности (эквивалентно методу 1)
# output, _ = pad_packed_sequence(packed_output, batch_first=True)
# last_hidden = output[range(len(lengths)), lengths - 1, :]
return self.fc(last_hidden)
model = LSTMClassifier(vocab_size=10000, embed_dim=128,
hidden_size=256, num_classes=5)
# Симуляция батча
batch_tokens = torch.randint(1, 10000, (8, 30)) # (batch_size=8, max_len=30)
batch_lengths = torch.randint(5, 31, (8,)) # Истинная длина каждой последовательности
output = model(batch_tokens, batch_lengths)
print(output.shape) # torch.Size([8, 5])
import torch.nn as nn
from torch.nn.utils.rnn import pack_padded_sequence, pad_packed_sequence
class LSTMClassifier(nn.Module):
def __init__(self, vocab_size, embed_dim, hidden_size, num_classes, padding_idx=0):
super().__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=padding_idx)
self.lstm = nn.LSTM(embed_dim, hidden_size, batch_first=True)
self.fc = nn.Linear(hidden_size, num_classes)
def forward(self, x, lengths):
# x: (batch_size, max_seq_len) — индексы слов
embedded = self.embedding(x) # (batch_size, max_seq_len, embed_dim)
# Упаковка
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
# Прямой проход LSTM (пропуск позиций паддинга)
packed_output, (h_n, c_n) = self.lstm(packed)
# Метод 1: использовать h_n последнего слоя как представление последовательности
last_hidden = h_n.squeeze(0) # (batch_size, hidden_size)
# Метод 2: После распаковки взять выход на истинной последней позиции каждой последовательности (эквивалентно методу 1)
# output, _ = pad_packed_sequence(packed_output, batch_first=True)
# last_hidden = output[range(len(lengths)), lengths - 1, :]
return self.fc(last_hidden)
model = LSTMClassifier(vocab_size=10000, embed_dim=128,
hidden_size=256, num_classes=5)
# Симуляция батча
batch_tokens = torch.randint(1, 10000, (8, 30)) # (batch_size=8, max_len=30)
batch_lengths = torch.randint(5, 31, (8,)) # Истинная длина каждой последовательности
output = model(batch_tokens, batch_lengths)
print(output.shape) # torch.Size([8, 5])
8. Полный практический пример: классификация тональности текста
На примере бинарной классификации тональности отзывов о фильмах IMDB продемонстрируем полный конвейер обработки текста с помощью двунаправленной LSTM:
Пример
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
from torch.nn.utils.rnn import pack_padded_sequence, pad_sequence
from torch.optim.lr_scheduler import ReduceLROnPlateau
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# ── Определение модели ──────────────────────────────────────
class BiLSTMClassifier(nn.Module):
"""
Модель классификации текста на основе двунаправленной LSTM
Архитектура: Embedding -> BiLSTM -> Dropout -> FC
"""
def __init__(self, vocab_size, embed_dim, hidden_size,
num_layers, num_classes, dropout=0.5, padding_idx=0):
super().__init__()
self.embedding = nn.Embedding(
vocab_size, embed_dim, padding_idx=padding_idx
)
# При использовании предобученных векторов слов:
# self.embedding = nn.Embedding.from_pretrained(pretrained_vectors)
self.lstm = nn.LSTM(
input_size=embed_dim,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
bidirectional=True,
)
self.dropout = nn.Dropout(dropout)
# Двунаправленная LSTM: объединить последние скрытые состояния прямого и обратного проходов
self.fc = nn.Linear(hidden_size * 2, num_classes)
def forward(self, x, lengths):
# x: (batch_size, max_seq_len)
embedded = self.dropout(self.embedding(x))
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
packed_output, (h_n, c_n) = self.lstm(packed)
# h_n: (num_layers * 2, batch_size, hidden_size)
# Взять скрытые состояния прямого и обратного проходов последнего слоя и объединить их
h_forward = h_n[-2, :, :] # (batch_size, hidden_size)
h_backward = h_n[-1, :, :] # (batch_size, hidden_size)
h_combined = torch.cat([h_forward, h_backward], dim=-1)
# (batch_size, hidden_size * 2)
out = self.dropout(h_combined)
return self.fc(out)
# ── Пользовательский набор данных ────────────────────────────────
class TextDataset(Dataset):
def __init__(self, texts, labels, vocab, max_len=200):
self.data = texts
self.labels = labels
self.vocab = vocab
self.max_len = max_len
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
tokens = self.data[idx][:self.max_len]
ids = [self.vocab.get(t, 1) for t in tokens] # 1 = <UNK>
return torch.tensor(ids, dtype=torch.long), torch.tensor(self.labels[idx])
def collate_fn(batch):
"""Пользовательский collate: дополнить последовательности переменной длины нулями, записать истинные длины"""
sequences, labels = zip(*batch)
lengths = torch.tensor([len(s) for s in sequences])
padded = pad_sequence(sequences, batch_first=True, padding_value=0)
labels = torch.stack(labels)
return padded, lengths, labels
# ── Функции обучения и оценки ────────────────────────────────
def train_epoch(model, loader, optimizer, criterion):
model.train()
total_loss, correct = 0.0, 0
for texts, lengths, labels in loader:
texts, labels = texts.to(device), labels.to(device)
optimizer.zero_grad()
outputs = model(texts, lengths)
loss = criterion(outputs, labels)
loss.backward()
# Отсечение градиента: предотвратить взрыв градиентов в RNN
nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
total_loss += loss.item() * len(labels)
correct += (outputs.argmax(1) == labels).sum().item()
n = len(loader.dataset)
return total_loss / n, correct / n
def eval_epoch(model, loader, criterion):
model.eval()
total_loss, correct = 0.0, 0
with torch.no_grad():
for texts, lengths, labels in loader:
texts, labels = texts.to(device), labels.to(device)
outputs = model(texts, lengths)
loss = criterion(outputs, labels)
total_loss += loss.item() * len(labels)
correct += (outputs.argmax(1) == labels).sum().item()
n = len(loader.dataset)
return total_loss / n, correct / n
# ── Инициализация и обучение ──────────────────────────────────
VOCAB_SIZE = 50000
EMBED_DIM = 128
HIDDEN_SIZE = 256
NUM_LAYERS = 2
NUM_CLASSES = 2
DROPOUT = 0.5
EPOCHS = 15
model = BiLSTMClassifier(VOCAB_SIZE, EMBED_DIM, HIDDEN_SIZE,
NUM_LAYERS, NUM_CLASSES, DROPOUT).to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
scheduler = ReduceLROnPlateau(optimizer, mode="max",
patience=3, factor=0.5)
best_acc = 0.0
for epoch in range(1, EPOCHS + 1):
train_loss, train_acc = train_epoch(model, train_loader, optimizer, criterion)
val_loss, val_acc = eval_epoch(model, val_loader, criterion)
scheduler.step(val_acc)
print(f"Эпоха {epoch:2d}/{EPOCHS} | "
f"Потери на обучении: {train_loss:.4f}, Точность: {train_acc:.4f} | "
f"Потери на валидации: {val_loss:.4f}, Точность: {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> Сохранение лучшей модели, точность на валидации: {best_acc:.4f}")
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
from torch.nn.utils.rnn import pack_padded_sequence, pad_sequence
from torch.optim.lr_scheduler import ReduceLROnPlateau
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# ── Определение модели ──────────────────────────────────────
class BiLSTMClassifier(nn.Module):
"""
Модель классификации текста на основе двунаправленной LSTM
Архитектура: Embedding -> BiLSTM -> Dropout -> FC
"""
def __init__(self, vocab_size, embed_dim, hidden_size,
num_layers, num_classes, dropout=0.5, padding_idx=0):
super().__init__()
self.embedding = nn.Embedding(
vocab_size, embed_dim, padding_idx=padding_idx
)
# При использовании предобученных векторов слов:
# self.embedding = nn.Embedding.from_pretrained(pretrained_vectors)
self.lstm = nn.LSTM(
input_size=embed_dim,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
bidirectional=True,
)
self.dropout = nn.Dropout(dropout)
# Двунаправленная LSTM: объединить последние скрытые состояния прямого и обратного проходов
self.fc = nn.Linear(hidden_size * 2, num_classes)
def forward(self, x, lengths):
# x: (batch_size, max_seq_len)
embedded = self.dropout(self.embedding(x))
packed = pack_padded_sequence(
embedded, lengths.cpu(), batch_first=True, enforce_sorted=False
)
packed_output, (h_n, c_n) = self.lstm(packed)
# h_n: (num_layers * 2, batch_size, hidden_size)
# Взять скрытые состояния прямого и обратного проходов последнего слоя и объединить их
h_forward = h_n[-2, :, :] # (batch_size, hidden_size)
h_backward = h_n[-1, :, :] # (batch_size, hidden_size)
h_combined = torch.cat([h_forward, h_backward], dim=-1)
# (batch_size, hidden_size * 2)
out = self.dropout(h_combined)
return self.fc(out)
# ── Пользовательский набор данных ────────────────────────────────
class TextDataset(Dataset):
def __init__(self, texts, labels, vocab, max_len=200):
self.data = texts
self.labels = labels
self.vocab = vocab
self.max_len = max_len
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
tokens = self.data[idx][:self.max_len]
ids = [self.vocab.get(t, 1) for t in tokens] # 1 = <UNK>
return torch.tensor(ids, dtype=torch.long), torch.tensor(self.labels[idx])
def collate_fn(batch):
"""Пользовательский collate: дополнить последовательности переменной длины нулями, записать истинные длины"""
sequences, labels = zip(*batch)
lengths = torch.tensor([len(s) for s in sequences])
padded = pad_sequence(sequences, batch_first=True, padding_value=0)
labels = torch.stack(labels)
return padded, lengths, labels
# ── Функции обучения и оценки ────────────────────────────────
def train_epoch(model, loader, optimizer, criterion):
model.train()
total_loss, correct = 0.0, 0
for texts, lengths, labels in loader:
texts, labels = texts.to(device), labels.to(device)
optimizer.zero_grad()
outputs = model(texts, lengths)
loss = criterion(outputs, labels)
loss.backward()
# Отсечение градиента: предотвратить взрыв градиентов в RNN
nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
total_loss += loss.item() * len(labels)
correct += (outputs.argmax(1) == labels).sum().item()
n = len(loader.dataset)
return total_loss / n, correct / n
def eval_epoch(model, loader, criterion):
model.eval()
total_loss, correct = 0.0, 0
with torch.no_grad():
for texts, lengths, labels in loader:
texts, labels = texts.to(device), labels.to(device)
outputs = model(texts, lengths)
loss = criterion(outputs, labels)
total_loss += loss.item() * len(labels)
correct += (outputs.argmax(1) == labels).sum().item()
n = len(loader.dataset)
return total_loss / n, correct / n
# ── Инициализация и обучение ──────────────────────────────────
VOCAB_SIZE = 50000
EMBED_DIM = 128
HIDDEN_SIZE = 256
NUM_LAYERS = 2
NUM_CLASSES = 2
DROPOUT = 0.5
EPOCHS = 15
model = BiLSTMClassifier(VOCAB_SIZE, EMBED_DIM, HIDDEN_SIZE,
NUM_LAYERS, NUM_CLASSES, DROPOUT).to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
scheduler = ReduceLROnPlateau(optimizer, mode="max",
patience=3, factor=0.5)
best_acc = 0.0
for epoch in range(1, EPOCHS + 1):
train_loss, train_acc = train_epoch(model, train_loader, optimizer, criterion)
val_loss, val_acc = eval_epoch(model, val_loader, criterion)
scheduler.step(val_acc)
print(f"Эпоха {epoch:2d}/{EPOCHS} | "
f"Потери на обучении: {train_loss:.4f}, Точность: {train_acc:.4f} | "
f"Потери на валидации: {val_loss:.4f}, Точность: {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> Сохранение лучшей модели, точность на валидации: {best_acc:.4f}")
9. Полный практический пример: прогнозирование временных рядов
На примере многошагового прогнозирования временных рядов используйте LSTM для предсказания значений на следующие N шагов:
Пример
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from torch.utils.data import Dataset, DataLoader
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# ── Набор данных со скользящим окном ──────────────────────────────
class TimeSeriesDataset(Dataset):
"""
Сегментация временного ряда методом скользящего окна
Вход: прошлые input_len шагов
Цель: будущие output_len шагов
"""
def __init__(self, series, input_len, output_len):
self.data = torch.tensor(series, dtype=torch.float32)
self.input_len = input_len
self.output_len = output_len
def __len__(self):
return len(self.data) - self.input_len - self.output_len + 1
def __getitem__(self, idx):
x = self.data[idx : idx + self.input_len]
y = self.data[idx + self.input_len : idx + self.input_len + self.output_len]
return x.unsqueeze(-1), y # x: (input_len, 1), y: (output_len,)
# ── Определение модели ──────────────────────────────────────
class LSTMForecaster(nn.Module):
"""
Модель многошагового прогнозирования временных рядов
Архитектура: LSTM -> Dropout -> FC
"""
def __init__(self, input_size, hidden_size, num_layers,
output_len, dropout=0.2):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
)
self.dropout = nn.Dropout(dropout)
self.fc = nn.Linear(hidden_size, output_len)
def forward(self, x):
# x: (batch_size, input_len, input_size)
output, (h_n, c_n) = self.lstm(x)
# Взять выход последнего временного шага
last = output[:, -1, :] # (batch_size, hidden_size)
last = self.dropout(last)
return self.fc(last) # (batch_size, output_len)
# ── Подготовка данных (на примере синусоиды)──────────────────────
t = np.linspace(0, 200, 10000)
series = np.sin(t) + 0.1 * np.random.randn(len(t))
INPUT_LEN = 60 # Использовать последние 60 шагов
OUTPUT_LEN = 10 # Предсказать следующие 10 шагов
split = int(len(series) * 0.8)
train_data = series[:split]
val_data = series[split:]
train_dataset = TimeSeriesDataset(train_data, INPUT_LEN, OUTPUT_LEN)
val_dataset = TimeSeriesDataset(val_data, INPUT_LEN, OUTPUT_LEN)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=64, shuffle=False)
# ── Обучение ──────────────────────────────────────────
model = LSTMForecaster(input_size=1, hidden_size=128,
num_layers=2, output_len=OUTPUT_LEN).to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
def train_epoch_ts(model, loader, optimizer, criterion):
model.train()
total_loss = 0.0
for x, y in loader:
x, y = x.to(device), y.to(device)
optimizer.zero_grad()
pred = model(x)
loss = criterion(pred, y)
loss.backward()
nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
total_loss += loss.item() * x.size(0)
return total_loss / len(loader.dataset)
def eval_epoch_ts(model, loader, criterion):
model.eval()
total_loss = 0.0
with torch.no_grad():
for x, y in loader:
x, y = x.to(device), y.to(device)
pred = model(x)
total_loss += criterion(pred, y).item() * x.size(0)
return total_loss / len(loader.dataset)
for epoch in range(1, 31):
train_loss = train_epoch_ts(model, train_loader, optimizer, criterion)
val_loss = eval_epoch_ts(model, val_loader, criterion)
print(f"Epoch {epoch:2d}/30 | Train MSE: {train_loss:.6f} | Val MSE: {val_loss:.6f}")
# ── Многошаговое рекурсивное предсказание (другая стратегия)────────────────────
def recursive_forecast(model, init_sequence, steps, device):
"""
Рекурсивное предсказание: предсказываем по одному шагу за раз, добавляем предсказанное значение в последовательность, затем предсказываем следующий шаг
Подходит для моделей одношагового предсказания с output_len=1
"""
model.eval()
sequence = list(init_sequence)
predictions = []
with torch.inference_mode():
for _ in range(steps):
x = torch.tensor(sequence[-INPUT_LEN:], dtype=torch.float32)
x = x.unsqueeze(0).unsqueeze(-1).to(device) # (1, input_len, 1)
pred = model(x).item()
predictions.append(pred)
sequence.append(pred)
return predictions
import torch.nn as nn
import torch.optim as optim
import numpy as np
from torch.utils.data import Dataset, DataLoader
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# ── Набор данных со скользящим окном ──────────────────────────────
class TimeSeriesDataset(Dataset):
"""
Сегментация временного ряда методом скользящего окна
Вход: прошлые input_len шагов
Цель: будущие output_len шагов
"""
def __init__(self, series, input_len, output_len):
self.data = torch.tensor(series, dtype=torch.float32)
self.input_len = input_len
self.output_len = output_len
def __len__(self):
return len(self.data) - self.input_len - self.output_len + 1
def __getitem__(self, idx):
x = self.data[idx : idx + self.input_len]
y = self.data[idx + self.input_len : idx + self.input_len + self.output_len]
return x.unsqueeze(-1), y # x: (input_len, 1), y: (output_len,)
# ── Определение модели ──────────────────────────────────────
class LSTMForecaster(nn.Module):
"""
Модель многошагового прогнозирования временных рядов
Архитектура: LSTM -> Dropout -> FC
"""
def __init__(self, input_size, hidden_size, num_layers,
output_len, dropout=0.2):
super().__init__()
self.lstm = nn.LSTM(
input_size=input_size,
hidden_size=hidden_size,
num_layers=num_layers,
batch_first=True,
dropout=dropout if num_layers > 1 else 0.0,
)
self.dropout = nn.Dropout(dropout)
self.fc = nn.Linear(hidden_size, output_len)
def forward(self, x):
# x: (batch_size, input_len, input_size)
output, (h_n, c_n) = self.lstm(x)
# Взять выход последнего временного шага
last = output[:, -1, :] # (batch_size, hidden_size)
last = self.dropout(last)
return self.fc(last) # (batch_size, output_len)
# ── Подготовка данных (на примере синусоиды)──────────────────────
t = np.linspace(0, 200, 10000)
series = np.sin(t) + 0.1 * np.random.randn(len(t))
INPUT_LEN = 60 # Использовать последние 60 шагов
OUTPUT_LEN = 10 # Предсказать следующие 10 шагов
split = int(len(series) * 0.8)
train_data = series[:split]
val_data = series[split:]
train_dataset = TimeSeriesDataset(train_data, INPUT_LEN, OUTPUT_LEN)
val_dataset = TimeSeriesDataset(val_data, INPUT_LEN, OUTPUT_LEN)
train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=64, shuffle=False)
# ── Обучение ──────────────────────────────────────────
model = LSTMForecaster(input_size=1, hidden_size=128,
num_layers=2, output_len=OUTPUT_LEN).to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=1e-3)
def train_epoch_ts(model, loader, optimizer, criterion):
model.train()
total_loss = 0.0
for x, y in loader:
x, y = x.to(device), y.to(device)
optimizer.zero_grad()
pred = model(x)
loss = criterion(pred, y)
loss.backward()
nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
optimizer.step()
total_loss += loss.item() * x.size(0)
return total_loss / len(loader.dataset)
def eval_epoch_ts(model, loader, criterion):
model.eval()
total_loss = 0.0
with torch.no_grad():
for x, y in loader:
x, y = x.to(device), y.to(device)
pred = model(x)
total_loss += criterion(pred, y).item() * x.size(0)
return total_loss / len(loader.dataset)
for epoch in range(1, 31):
train_loss = train_epoch_ts(model, train_loader, optimizer, criterion)
val_loss = eval_epoch_ts(model, val_loader, criterion)
print(f"Epoch {epoch:2d}/30 | Train MSE: {train_loss:.6f} | Val MSE: {val_loss:.6f}")
# ── Многошаговое рекурсивное предсказание (другая стратегия)────────────────────
def recursive_forecast(model, init_sequence, steps, device):
"""
Рекурсивное предсказание: предсказываем по одному шагу за раз, добавляем предсказанное значение в последовательность, затем предсказываем следующий шаг
Подходит для моделей одношагового предсказания с output_len=1
"""
model.eval()
sequence = list(init_sequence)
predictions = []
with torch.inference_mode():
for _ in range(steps):
x = torch.tensor(sequence[-INPUT_LEN:], dtype=torch.float32)
x = x.unsqueeze(0).unsqueeze(-1).to(device) # (1, input_len, 1)
pred = model(x).item()
predictions.append(pred)
sequence.append(pred)
return predictions
10. Сравнение и выбор LSTM и GRU
В этом разделе сравниваются структурные различия между LSTM и GRU и даются рекомендации по выбору.
Структурное сравнение
| Пункт сравнения | LSTM | GRU |
|---|---|---|
| Количество ворот | 3 (Забывание, Вход, Выход) | 2 (Сброс, Обновление) |
| Векторы состояний | Состояние ячейки + Скрытое состояние | Только скрытое состояние |
| Количество параметров (тот же hidden_size) | Базовый уровень | Около 75% |
| Скорость обучения | Медленнее | Быстрее |
| Производительность на длинных последовательностях | Обычно лучше | Comparable |
| Производительность на коротких последовательностях | Аналогично | Аналогично |
| Сложность реализации | Повышенный | Нижний |
Рекомендации по выбору
数据量小、训练资源有限
-> 优先选 GRU，参数少，不容易过拟合，训练快
序列较长（> 100 步）、长距离依赖重要
-> 优先选 LSTM，细胞状态更擅长保留远期信息
需要快速实验和基线对比
-> 先用 GRU，效果差再换 LSTM
任务对准确率要求极高，有充足数据
-> 两者都试，结合交叉验证选择
2020 年后的新项目
-> 考虑 Transformer 架构（BERT、GPT），在大数据量下通常优于 LSTM/GRU
LSTM/GRU 仍在边缘设备、低延迟推理、数据量较小的场景中有优势
Краткая справка по частым вопросам
| Проблема | Причина | Решение |
|---|---|---|
| Потери при обучении не уменьшаются | Слишком высокая скорость обучения или взрыв градиента | Уменьшите скорость обучения; добавьте отсечение градиентаclip_grad_norm_ |
| Потери на валидации намного выше, чем потери на обучении | Переобучение | Увеличьте Dropout; уменьшите количество слоёв или hidden_size |
| Плохое предсказание в конце последовательности | Исчезающий градиент | Увеличьте количество слоёв; используйте двунаправленную структуру; рассмотрите механизм внимания |
| Чрезмерное использование памяти GPU | Слишком длинная последовательность или слишком большой батч | Уменьшите seq_len или batch_size; используйте контрольные точки градиента |
| Аномальные потери при обучении с сохранением состояния | Состояния, передаваемые между батчами, не отсоединены | Вызывать после каждого батчаh.detach_() |
| Ошибка размерности конкатенации двунаправленной LSTM | Непонимание индекса h_n | использоватьh_n[-2](forward) иh_n[-1](backward) |
| Ошибка PackedSequence | Длины последовательностей не отсортированы по убыванию | Setenforce_sorted=False |
| Путаница с batch_first | Забыли установить согласованно | Рекомендуется использовать повсюдуbatch_first=True |