趣旨
RNNの一番シンプルなモデルであるElam RNNを使って大まかな仕組みを理解する事が目的です
RNNとは
1. RNNとは？
意味：Recurrent Neural Network（再帰型ニューラルネットワーク）
特徴：時系列や系列データの解析に強みを持つニューラルネットワークの一種
主な用途：音声認識、自然言語処理（NLP）、時系列予測、動画解析など
2. なぜRNNが必要か？
通常のニューラルネットワーク（Feedforward NN）は入力データの順序情報を扱えない
連続したデータ（文章の単語列、株価の時系列、音声の波形など）は時間や順序が重要
RNNは「隠れ状態（hidden state）」を持ち、過去の情報を内部に記憶しながら次の処理に活かせる
※今や過去の学習方法で陳腐化した技術ですが、小規模ならわざわざオーバースペックな生成AIモデルを使わなくても活用できる技術だと思います。
3. RNNの基本構造
入力系列
$$
h_t = f(W_{xh} x_t + W_{hh} h_{t-1} + b_h)
$$
- ( W_{xh} ), ( W_{hh} ) は重み行列
- ( b_h ) はバイアス
- ( f ) は活性化関数（例：tanh、ReLU）です
出力 ( y_t ) は通常隠れ状態から計算され、
$$
y_t = g(W_{hy} h_t + b_y)
$$
例えば、分類問題の場合は ( g ) に softmax 関数を用います。
4. RNNの強みと課題
強み
・時系列の依存関係をモデル化可能
・変動長の入力に対応可能
・自然言語のようなシーケンスに強い
課題
・長期依存の情報を保持しにくい（勾配消失問題）
・学習が難しく計算コストが高い
・標準的なRNNは長期記憶が弱い
5. RNNの発展系
LSTM（Long Short-Term Memory）
勾配消失問題を解決し、長期依存性を保持しやすいゲート機構を持つ
GRU（Gated Recurrent Unit）
LSTMの簡略版で計算コストを抑えつつ長期依存の記憶を可能にする
6. 具体的なビジネス応用例
チャットボット・対話システム：文章の文脈を理解し応答生成
売上予測：過去の売上データの時系列解析で将来予測
異常検知：設備センサーデータの時系列から異常を検知
音声認識・翻訳：音声の連続的な波形をテキスト化、翻訳
模擬データ
営業と購買のcall reportをモデルとして学習し、未知のcall reportに対して推論します。
python code
data import
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
print("PyTorch version:", torch.__version__)
print("CUDA available:", torch.cuda.is_available())
print("CUDA version:", torch.version.cuda)
学習用データ
0を営業のcall report, 1を購買のcall reportとします。
# --- データ準備 ---
texts = [
# ラベル0：営業のコールレポート風
"Discussed client requirements and upcoming project deadlines",
"Followed up with customer regarding contract renewal and pricing",
"Presented new product features and addressed client concerns",
"Scheduled next meeting to review sales targets and pipeline",
"Negotiated terms for bulk order and finalized delivery dates",
# ラベル1：購買のコールレポート風
"Inquired about supplier availability and lead times for materials",
"Reviewed purchase order status and confirmed shipment details",
"Discussed pricing adjustments due to recent market fluctuations",
"Coordinated with vendor on quality control and inspection process",
"Approved invoice discrepancies and updated procurement records"
]
labels = [0, 0, 0, 0, 0, 1, 1, 1, 1, 1]
ボキャブラリ作成
この操作で出現単語を数字に変換し、テキストを数字のリストに変換します。
# --- ボキャブラリ作成 ---
# この操作でテキスト文字列を数字に変換する
# テキストを半角スペースでjoinした後splitして、集合としてvocabに渡す
vocab = set(" ".join(texts).split())
print(vocab)
# 集合の番号順番をvalueにして、keyをwordにして辞書に入れる
word2idx = {word: idx+1 for idx, word in enumerate(vocab)} # 0はPAD用
print("="*30)
print(word2idx)
def encode(text: str) -> list[int]:
"""テキスト文字列をword2idxのvalue listに変換する関数"""
return [word2idx[w] for w in text.split()]
encoded_texts = [encode(t) for t in texts]
print("="*30)
print(encoded_texts)
パディングして最大テキスト長に他のテキスト長をそろえるためゼロで埋めます
# パディング
# テキスト文字列の最大長を確認
max_len = max(len(seq) for seq in encoded_texts)
print("max text length: ",max_len)
def pad(seq):
# テキスト最大長にそろうように、value = 0 で埋める関数
return seq + [0] * (max_len - len(seq))
padded_texts = [pad(seq) for seq in encoded_texts]
データセットを作ります
# PyTorchのDatasetを継承して、テキストデータとラベルを扱うクラスを定義
class TextDataset(Dataset):
def __init__(self, texts, labels):
# テキストのリストをPyTorchのLongTensorに変換（単語IDの系列を表現）
self.texts = torch.tensor(texts, dtype=torch.long)
# ラベルのリストをLongTensorに変換
self.labels = torch.tensor(labels, dtype=torch.long)
def __len__(self):
# データセットのサイズ（サンプル数）を返す
return len(self.labels)
def __getitem__(self, idx):
# インデックスidxに対応するテキストとラベルを返す
return self.texts[idx], self.labels[idx]
# パディング済みテキストとラベルからDatasetのインスタンスを作成
dataset = TextDataset(padded_texts, labels)
# Datasetからミニバッチを生成するDataLoaderを作成
# バッチサイズは2、エポックごとにデータをシャッフルしてミニバッチを作成する設定
loader = DataLoader(dataset, batch_size=2, shuffle=True)
# Datasetの2番目（インデックス1）の要素（テキストのTensorとラベル）を表示
print(dataset[1])
# DataLoaderオブジェクトの中身を表示（ミニバッチのイテレータ）
print(loader)
実際にどのように引数に渡されているのかを理解します
batch = next(iter(loader))
# 選ばれた二つのIDベクトルとその時のlabelが入っている
batch
モデル定義
モデルをクラスで作成して学習する準備を行います
# --- モデル定義 ---
class SimpleElmanRNN(nn.Module):
"""
シンプルなElman RNNモデル。
単語IDの系列を受け取り、Embedding層でベクトル化した後、
RNNで時系列処理を行い、最終時刻の隠れ状態を用いて
全結合層で分類出力を行う。
Args:
vocab_size: 語彙数（単語の種類数）
embed_size: 埋め込みベクトルの次元数
hidden_size: RNNの隠れ状態の次元数
output_size: 出力クラス数
Methods:
forward(x: Tensor) -> Tensor:
順伝播処理。入力バッチの単語ID系列を受け取り、
分類スコアを返す。
"""
def __init__(self, vocab_size: int, embed_size: int, hidden_size: int, output_size: int) -> None:
super().__init__()
self.embedding: nn.Embedding = nn.Embedding(vocab_size + 1, embed_size, padding_idx=0)
self.rnn: nn.RNN = nn.RNN(embed_size, hidden_size, batch_first=True)
self.fc: nn.Linear = nn.Linear(hidden_size, output_size)
def forward(self, x):
"""
モデルの順伝播処理。
Args:
x: 入力テンソル。形状は (batch_size, sequence_length) の単語ID系列。
Returns:
出力テンソル。形状は (batch_size, output_size) の分類スコア。
"""
x = self.embedding(x)
out, h_n = self.rnn(x) # out:(batch, seq_len, hidden), h_n: (num_layers,batch, hidden)
# 単純RNNで1層だから h_n.shape = (1, batch, hidden_size)になっている。
out = out[:, -1, :] # シーケンスの最終時刻の隠れ状態を抽出
out = self.fc(out)
return out, h_n.squeeze(0)
ハイパーパラメータ設定
学習量は少ないので、ハイパーパラメータは低めでも問題なく損失関数は小さくなります
# ハイパーパラメータ設定
vocab_size = len(vocab) # 語彙数 = テキスト文字列の長さ
embed_size = 4
hidden_size = 16
output_size = 2 # 出力クラス数 labelsの変量
model = SimpleElmanRNN(vocab_size, embed_size, hidden_size, output_size)
学習のため、隠れ層hを保存して、隠れ層がどのように推移して値が変化しているのかを理解します
# --- 訓練設定 ---
criterion = nn.CrossEntropyLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)
# --- 訓練ループ ---
hidden_states_over_epoches = []
epoch_list = []
avg_loss_list = []
for epoch in range(30):
total_loss = 0
all_hidden_states = []
for batch_x, batch_y in loader:
optimizer.zero_grad()
outputs, h_n = model(batch_x)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
total_loss += loss.item() * batch_x.size(0)
all_hidden_states.append(h_n[0].detach().cpu().numpy()) # 1層目のサンプルの隠れ層hを記録
avg_loss = total_loss / len(dataset)
print(f"Epoch {epoch+1}, Avg Loss: {avg_loss:.4f}")
epoch_list.append(epoch)
avg_loss_list.append(avg_loss)
hidden_states_over_epoches.append(np.mean(all_hidden_states, axis=0)) # バッチ内平均
隠れ層hの推移を可視化
hidden_states_over_epoches = np.array(hidden_states_over_epoches)
plt.figure(figsize=(18,6))
for i in range(hidden_states_over_epoches.shape[1]):
plt.plot(hidden_states_over_epoches[:,i], label=f"hidden_{i}") # 次元の数だけ可視化
plt.xlabel("Epoch")
plt.ylabel("Average Hidden State Value")
plt.title("RNN Hidden State Dynamics over Epochs")
plt.legend(loc="upper right")
plt.grid(True)
plt.show()
ここからわかることは、
1. 学習の安定化・収束の兆候
- 学習開始直後（前半）は、モデルのパラメータがランダムに初期化されており、隠れ状態もランダムに振る舞っています。
- Epoch10以降、モデルが訓練データに対する特徴を捉え、パラメータが更新されることで、隠れ状態の各ユニットが特定のパターンや機能を持つように学習されていることを示します。
- 「一貫した数値の変化」とは、各ユニットがある意味のある特徴を捉えた信号を出している状態であり、モデルが入力系列から意味のある情報を抽出できている証拠です。
2. 隠れ状態の「機能分化」
- 16次元それぞれの隠れユニットが異なる特徴やパターンを表現し始めている可能性があります。
- 例えば、あるユニットはある種のパターンを強調し、別のユニットは異なる情報を表しているかもしれません。
- グラフの色ごとに異なる挙動が見えるのは、この機能分化の表れです。
- 初期の乱雑さの消失＝学習によるノイズ除去
- 初期のランダム状態はノイズの多い未学習状態。
- 学習進行で、ノイズが減り安定した信号に変わっていることを示しています。
損失関数の推移
epoch_list = list(range(1, len(avg_loss_list)+1))
plt.figure(figsize=(8,6))
plt.plot(epoch_list,avg_loss_list)
plt.xlabel("epoch no.")
plt.ylabel("avg_loss (log)")
plt.yscale("log")
plt.title("RNN avg_loss step")
plt.grid(True)
plt.show()
※学習するtrainデータも少なかったので、おおよそ低い値まで下がり、やや過学習気味のようにも見えます。
未知データを使って推論してみる
# 未知のテキストの分類を試す
# 学習済み語彙辞書
print(word2idx)
# 未知単語ように<UNK>を追加
word2idx["<UNK>"] = 0
# padding用に<PAD>を追加
word2idx["<PAD>"] = 99
print(word2idx)
# 未知のテキスト
# 1テキスト目が営業のcall report, 2テキスト目が購買のcall report
unknown_texts = [
"discuss the contract of the new project with the client",
"confirm supplier shipment schedule and status"
]
# 1) トークン化・エンコード関数（学習時と同じ）
def encode(text: str, word2idx: dict) -> list[int]:
return [word2idx.get(w, 0) for w in text.split()]
encoded_unknown = [encode(t, word2idx) for t in unknown_texts]
# unknownの最大長を確認
max_len = max(len(seq) for seq in encoded_unknown)
print("unknown max length: ", max_len)
padded_unknown = [pad(seq) for seq in encoded_unknown]
print("\nafter padding: ")
print(padded_unknown)
# 3) Tensorに変換
input_tensor = torch.tensor(padded_unknown, dtype=torch.long)
print("\ninput tensor values: ")
print(input_tensor)
# model mode
model.eval()
# 推論モード
with torch.no_grad():
outputs, h_n = model(input_tensor) # 出力はlogits, 隠れ層も出力するモデルなのでh_nを作成
print("\nevaluated output: ")
print(outputs) # 行がテキスト番号、列がラベル確率(log)
# 確率に変換
probabilities = torch.softmax(outputs, dim=1) # dim:次元数の指定 1->2次元
print("\nresult probabilities: ")
print(probabilities)
# 予測クラスの取得
predicted_classes = torch.argmax(probabilities,dim=1)
print("\npredicted class of sentences: ")
print(unknown_texts)
print("0: sales call report / 1: purchase call report ",predicted_classes)
未知の単語が多く、ゼロ値が半分ほどありましたが分類確率は0.99を超えており正しく推論できていることが分かりました。