PyTorch LSTM / GRU
リカレントニューラルネットワーク（RNN）はシーケンスデータを処理する際に勾配消失問題に直面し、長距離依存関係を学習することが困難になります。
長短期記憶ネットワーク（Long Short-Term Memory、LSTM）とゲート付き回帰ユニット（Gated Recurrent Unit、GRU）ゲート機構を導入することでこの問題を解決し、時系列、自然言語、音声などのシーケンスタスクを処理する中核的なモデルです。
1. RNNの限界とゲート機構
標準RNNは各タイムステップで現在の入力と前ステップの隠れ状態を結合して新しい隠れ状態を計算します：
\[h_t = \tanh(W_h \cdot h_{t-1} + W_x \cdot x_t + b)\]
この構造には2つの中核的な問題があります：
勾配消失：逆伝播時に勾配は重み行列を段階的に乗算する必要があり、シーケンスが長いと勾配が指数関数的に減衰し、初期のタイムステップのパラメータがほとんど更新されず、モデルは長距離依存を学習できません。
勾配爆発：重み行列の最大固有値が1より大きい場合、勾配は逆伝播中に指数関数的に増大し、訓練が不安定になります（通常は勾配クリッピングで緩和します）。
ゲート機構の中核的な考え方は、学習可能な「スイッチ」を導入し、ネットワークが現在のタイムステップでどの情報を記憶すべきか、どの情報を忘れるべきか、どの新しい情報をメモリに書き込むべきかを自律的に決定できるようにすることです。
LSTMは3つのゲート（忘れゲート、入力ゲート、出力ゲート）と独立したセル状態を使用します。GRUは構造を2つのゲート（リセットゲート、更新ゲート）に簡素化し、パラメータが少なく、訓練が高速です。
2. LSTMの原理
2.1 中核構造と3つのゲート
LSTMは2つの状態ベクトルをタイムステップ間で受け渡します：
- セル状態（Cell State）\(c_t\)：長期記憶の保持役であり、情報はほぼ損失なく流れます
- 隠れ状態（Hidden State）\(h_t\)：短期記憶であり、現在のタイムステップの出力でもあります
3つのゲートはすべて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\) は要素ごとの乗算（Hadamard積）、\(\sigma\) はSigmoid関数を表します。
計算ロジックの解釈：
f_t ⊙ c_{t-1}：忘れゲートは履歴メモリをどれだけ保持するかを決定し、0に近いと忘れ、1に近いと保持しますi_t ⊙ g_t：入力ゲートはどれだけの新しい情報を書き込むかを決定し、g_tは候補となる新しい内容ですo_t ⊙ tanh(c_t)：出力ゲートはセル状態から何を抽出して隠れ状態として出力するかを決定します
3. GRUの原理
3.1 中核構造と2つのゲート
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には独立したセル状態がなく、パラメータ数はLSTMの約75%です。
4. PyTorchにおけるLSTM
この節では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, # 入力/出力shapeにおいてbatchが最初の次元にあるかどうか、デフォルトはFalse
dropout=0.0, # 層間dropout確率（num_layers > 1の場合のみ有効）
bidirectional=False, # 双方向LSTMを使用するかどうか、デフォルトはFalse
proj_size=0, # 投影層の次元（LSTM with projection）、デフォルトは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, # 入力/出力shapeにおいてbatchが最初の次元にあるかどうか、デフォルトはFalse
dropout=0.0, # 層間dropout確率（num_layers > 1の場合のみ有効）
bidirectional=False, # 双方向LSTMを使用するかどうか、デフォルトはFalse
proj_size=0, # 投影層の次元（LSTM with projection）、デフォルトは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組の重み行列、すなわち忘却ゲート・入力ゲート・候補値・出力ゲートに対応する。
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 の形状: {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 の形状: {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 の形状: {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 の形状: {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 の形状: {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 の形状: {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：ステートフル（stateful）モード —— batch をまたいで状態を渡す
# 超長系列のスライス（言語モデル、長文生成など）に適する
# detach() を呼び出して前の batch との計算グラフを切断し、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：ステートフル（stateful）モード —— batch をまたいで状態を渡す
# 超長系列のスライス（言語モデル、長文生成など）に適する
# detach() を呼び出して前の batch との計算グラフを切断し、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. PyTorchにおけるGRU
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 の形状: {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 の形状: {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 の形状: {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 の並び順：[順方向layer0, 逆方向layer0, 順方向layer1, 逆方向layer1]
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 の形状: {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 の並び順：[順方向layer0, 逆方向layer0, 順方向layer1, 逆方向layer1]
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 の形状: {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) 第1層
h_layer2 = h_n[1] # (batch_size, hidden_size) 第2層
h_layer3 = h_n[2] # (batch_size, hidden_size) 第3層（最上位層）
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 の形状: {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) 第1層
h_layer2 = h_n[1] # (batch_size, hidden_size) 第2層
h_layer3 = h_n[2] # (batch_size, hidden_size) 第3層（最上位層）
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. 可変長シーケンスの処理
実際のタスクでは、同じ batch 内の系列の長さが異なることがよくある。PyTorch にはpack_padded_sequenceとpad_packed_sequenceこの問題を処理し、LSTM がパディング位置（padding）で無駄な計算を行わないようにするための機能がある。
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
# batch 内の可変長系列をシミュレート（長い順にソート済み）
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：パディングを圧縮し、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
# batch 内の可変長系列をシミュレート（長い順にソート済み）
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：パディングを圧縮し、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 の形状: {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 の形状: {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)
# ── カスタム Dataset ────────────────────────────────
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 {epoch:2d}/{EPOCHS} | "
f"Train Loss: {train_loss:.4f}, Acc: {train_acc:.4f} | "
f"Val Loss: {val_loss:.4f}, Acc: {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> 最適モデルを保存、Val Acc: {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)
# ── カスタム Dataset ────────────────────────────────
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 {epoch:2d}/{EPOCHS} | "
f"Train Loss: {train_loss:.4f}, Acc: {train_acc:.4f} | "
f"Val Loss: {val_loss:.4f}, Acc: {val_acc:.4f}")
if val_acc > best_acc:
best_acc = val_acc
torch.save(model.state_dict(), "best_bilstm.pth")
print(f" -> 最適モデルを保存、Val Acc: {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")
# ── スライディングウィンドウ Dataset ──────────────────────────────
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):
"""
再帰予測：毎回1ステップ予測し、予測値をシーケンスに追加して次のステップを予測する
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")
# ── スライディングウィンドウ Dataset ──────────────────────────────
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):
"""
再帰予測：毎回1ステップ予測し、予測値をシーケンスに追加して次のステップを予測する
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% |
| トレーニング速度 | 比較的遅い | 比較的速い |
| 長いシーケンスの性能 | 通常より良い | ほぼ同等 |
| 短いシーケンスの性能 | ほぼ同等 | ほぼ同等 |
| 実装の複雑さ | 比較的高い | 比較的低い |
選定のアドバイス
数据量小、训练资源有限
-> 优先选 GRU，参数少，不容易过拟合，训练快
序列较长（> 100 步）、长距离依赖重要
-> 优先选 LSTM，细胞状态更擅长保留远期信息
需要快速实验和基线对比
-> 先用 GRU，效果差再换 LSTM
任务对准确率要求极高，有充足数据
-> 两者都试，结合交叉验证选择
2020 年后的新项目
-> 考虑 Transformer 架构（BERT、GPT），在大数据量下通常优于 LSTM/GRU
LSTM/GRU 仍在边缘设备、低延迟推理、数据量较小的场景中有优势
よくある質問の早見表
| 問題 | 原因 | 解決方法 |
|---|---|---|
| トレーニング Loss が下がらない | 学習率が大きすぎる、または勾配爆発 | 学習率を下げる；勾配クリッピングを追加するclip_grad_norm_ |
| 検証セットの Loss がトレーニングセットよりはるかに高い | 過学習 | Dropout を増やす；層数または hidden_size を減らす |
| シーケンス末尾の予測が悪い | 勾配消失 | 層数を増やす；双方向構造を使用する；アテンション機構を検討する |
| GPUメモリ使用量が大きすぎる | シーケンスが長すぎる、またはバッチが大きすぎる | seq_len または batch_size を減らす；勾配チェックポイントを使用する |
| ステートフルトレーニング中に Loss が異常 | バッチ間で状態を渡す際に detach していない | 各バッチの後に呼び出すh.detach_() |
| 双方向 LSTM の連結次元の誤り | h_n のインデックス理解の誤り | 用 h_n[-2]（順方向）とh_n[-1]（逆方向） |
| PackedSequence エラー | シーケンスの長さが降順に並んでいない | 設定enforce_sorted=False |
| batch_first の混同 | 統一設定を忘れる | 全体を通して使用することを推奨batch_first=True |