自然言語処理（NLP）における機械翻訳は、近年ディープラーニング技術の進歩により大きな進展を遂げています。本記事では、GRUベースのSeq2Seqモデルを使用した日本語-英語翻訳システムの実装と評価方法について説明します。
機械翻訳とは？
機械翻訳（Machine Translation, MT）は、ある言語で書かれたテキストを別の言語に自動的に変換する技術です。MTの歴史は1950年代までさかのぼり、規則ベースや統計的アプローチから始まり、現在ではニューラルネットワークを利用したNeural Machine Translation（NMT）が主流となっています。特にLSTMやTransformerなどのモデルは、複雑な言語間のマッピングを学習し、高い翻訳精度を達成しています。
データセットの分割
機械学習プロジェクトでは、データセットを以下の3つに分けることが一般的です：
- トレーニングセット
- 役割：モデルの学習に使用されます。モデルはこのデータに基づいてパラメータを調整し、予測誤差を最小化します。
- バリデーションセット
- 役割：超パラメータの調整やモデル選択に使用されます。トレーニングセットとは独立したデータであり、モデルの汎化性能を評価します。
- テストセット
- 役割：最終的なモデルの性能評価に使用されます。トレーニングやチューニング後に適用され、未知のデータに対する性能を測定します。
コンテスト概要
本コンテストでは、英語から日本語への翻訳タスクが設定されています。提供されるリソースには、双方向の並び文データおよび英日対応の用語辞書が含まれます。参加者はこれらのデータを基に多言語翻訳モデルを構築・訓練し、テストセットに対して翻訳結果を提出します。
使用するモデル
GRUモデル
GRU（Gated Recurrent Unit）はRNNの一種で、長期依存関係を効率的に捕捉できるように設計されています。主な特徴として更新ゲートとリセットゲートがあり、これらは次のように機能します：
- 更新ゲート：過去の情報をどれだけ保持するかを制御します。
- リセットゲート：過去の情報のどれだけを忘却するかを決定します。
Seq2Seqモデル
Seq2Seqモデルはエンコーダーとデコーダーの2つの部分から構成されています。エンコーダーは入力シーケンスを固定長のベクトルに変換し、デコーダーはそのベクトルを出力シーケンスに復元します。
import torch
import torch.nn as nn
import random
class Encoder(nn.Module):
def __init__(self, vocab_size, embed_dim, hidden_dim, num_layers, dropout_rate):
super(Encoder, self).__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim)
self.rnn = nn.GRU(embed_dim, hidden_dim, num_layers, dropout=dropout_rate, batch_first=True)
self.dropout = nn.Dropout(dropout_rate)
def forward(self, x):
embedded = self.dropout(self.embedding(x))
output, hidden = self.rnn(embedded)
return hidden
class Decoder(nn.Module):
def __init__(self, vocab_size, embed_dim, hidden_dim, num_layers, dropout_rate):
super(Decoder, self).__init__()
self.embedding = nn.Embedding(vocab_size, embed_dim)
self.rnn = nn.GRU(embed_dim, hidden_dim, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_dim, vocab_size)
self.dropout = nn.Dropout(dropout_rate)
def forward(self, x, hidden):
embedded = self.dropout(self.embedding(x))
output, hidden = self.rnn(embedded, hidden)
prediction = self.fc(output.squeeze(1))
return prediction, hidden
class Seq2SeqModel(nn.Module):
def __init__(self, encoder, decoder, device):
super(Seq2SeqModel, self).__init__()
self.encoder = encoder
self.decoder = decoder
self.device = device
def forward(self, src, trg, teacher_forcing_ratio=0.5):
batch_size, max_len = src.size(0), trg.size(1)
outputs = torch.zeros(batch_size, max_len, self.decoder.vocab_size).to(self.device)
hidden = self.encoder(src)
input_token = trg[:, 0].unsqueeze(1)
for t in range(1, max_len):
output, hidden = self.decoder(input_token, hidden)
outputs[:, t, :] = output
teacher_force = random.random() < teacher_forcing_ratio
top1 = output.argmax(1)
input_token = trg[:, t].unsqueeze(1) if teacher_force else top1.unsqueeze(1)
return outputs
データセットの定義
from torch.utils.data import Dataset, DataLoader
from collections import Counter
class TranslationDataset(Dataset):
def __init__(self, file_path, term_dict):
self.data = []
with open(file_path, 'r', encoding='utf-8') as f:
for line in f:
en, ja = line.strip().split('\t')
self.data.append((en, ja))
self.term_dict = term_dict
self.en_vocab = ['<pad>', '<sos>', '<eos>'] + list(term_dict.keys())
self.ja_vocab = ['<pad>', '<sos>', '<eos>']
en_counter, ja_counter = Counter(), Counter()
for en, ja in self.data:
en_tokens = en.split()
ja_tokens = list(ja)
en_counter.update(en_tokens)
ja_counter.update(ja_tokens)
self.en_vocab += [word for word, _ in en_counter.most_common(10000)]
self.ja_vocab += [word for word, _ in ja_counter.most_common(10000)]
self.en_word2idx = {word: idx for idx, word in enumerate(self.en_vocab)}
self.ja_word2idx = {word: idx for idx, word in enumerate(self.ja_vocab)}
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
en, ja = self.data[idx]
en_tensor = torch.tensor([self.en_word2idx.get(word, 1) for word in en.split()] + [2])
ja_tensor = torch.tensor([self.ja_word2idx.get(word, 1) for word in list(ja)] + [2])
return en_tensor, ja_tensor
モデルの訓練
def train_model(model, data_loader, optimizer, criterion):
model.train()
total_loss = 0
for src, trg in data_loader:
src, trg = src.to(device), trg.to(device)
optimizer.zero_grad()
output = model(src, trg)
output_dim = output.shape[-1]
output = output[:, 1:].contiguous().view(-1, output_dim)
trg = trg[:, 1:].contiguous().view(-1)
loss = criterion(output, trg)
loss.backward()
optimizer.step()
total_loss += loss.item()
return total_loss / len(data_loader)