import random class Seq2Seq(nn.Module): “””Seq2Seqモデル（Encoder + Decoder）””” def __init__(self, encoder, decoder, device): super(Seq2Seq, self).__init__() self.encoder = encoder self.decoder = decoder self.device = device def forward(self, src, trg, teacher_forcing_ratio=0.5): “”” Args: src: 入力系列 (batch_size, src_length) trg: 正解系列 (batch_size, trg_length) teacher_forcing_ratio: Teacher Forcingを使う確率 Returns: outputs: 全時刻の予測 (batch_size, trg_length, output_dim) “”” batch_size = src.shape[0] trg_length = trg.shape[1] trg_vocab_size = self.decoder.output_dim # 出力を格納するテンソル outputs = torch.zeros(batch_size, trg_length, trg_vocab_size).to(self.device) # 1. Encoderで入力を処理 hidden, cell = self.encoder(src) # hidden, cell: Context Vector（入力の「意味」） # 2. Decoderの最初の入力は<START>トークン # trg[:, 0]: 正解系列の最初の要素（= <START>） input = trg[:, 0].unsqueeze(1) # (batch_size, 1) # 3. 1単語ずつ生成 for t in range(1, trg_length): # Decoderで1単語予測 output, hidden, cell = self.decoder(input, hidden, cell) # output: (batch_size, output_dim) # 予測を保存 outputs[:, t] = output # Teacher Forcingを使うか決定 teacher_force = random.random() < teacher_forcing_ratio # 最も確率の高い単語を取得 top1 = output.argmax(1).unsqueeze(1) # (batch_size, 1) # 次の入力を決定 # Teacher Forcing: 正解を使う # そうでなければ: 自分の予測を使う input = trg[:, t].unsqueeze(1) if teacher_force else top1 return outputs

import torch.optim as optim # 損失関数（パディングを無視） PAD_IDX = 0 # <PAD>のインデックス criterion = nn.CrossEntropyLoss(ignore_index=PAD_IDX) # 最適化器 optimizer = optim.Adam(model.parameters(), lr=0.001) def train_epoch(model, dataloader, optimizer, criterion, clip=1.0): “””1エポックの訓練””” model.train() total_loss = 0 for src, trg in dataloader: src, trg = src.to(device), trg.to(device) # 勾配をゼロに optimizer.zero_grad() # Forward pass（Teacher Forcing率50%） output = model(src, trg, teacher_forcing_ratio=0.5) # output: (batch_size, trg_length, output_dim) # 損失計算 # <START>を除く（最初のトークンは予測しない） output_dim = output.shape[-1] output = output[:, 1:].reshape(-1, output_dim) # (batch_size * (trg_len-1), output_dim) trg = trg[:, 1:].reshape(-1) # (batch_size * (trg_len-1)) loss = criterion(output, trg) # Backward pass loss.backward() # 勾配クリッピング（重要！） torch.nn.utils.clip_grad_norm_(model.parameters(), clip) # パラメータ更新 optimizer.step() total_loss += loss.item() return total_loss / len(dataloader)

def translate(model, src_sentence, src_vocab, trg_vocab, max_length=50): “”” 文を翻訳する（Greedy Search） Args: model: 訓練済みSeq2Seqモデル src_sentence: 入力文（単語のリスト） src_vocab: 入力語彙（単語→ID） trg_vocab: 出力語彙（ID→単語） max_length: 最大生成長 Returns: 翻訳結果（単語のリスト） “”” model.eval() # 1. 入力をIDに変換 src_ids = [src_vocab.get(word, src_vocab[‘<unk>’]) for word in src_sentence] src_tensor = torch.LongTensor(src_ids).unsqueeze(0).to(device) # src_tensor: (1, src_length) with torch.no_grad(): # 2. Encoderで入力を処理 hidden, cell = model.encoder(src_tensor) # 3. Decoderで1単語ずつ生成 trg_ids = [trg_vocab[‘<start>’]] for _ in range(max_length): # 現在の入力 input = torch.LongTensor([trg_ids[-1]]).unsqueeze(0).to(device) # Decoder output, hidden, cell = model.decoder(input, hidden, cell) # 最も確率が高い単語 pred_id = output.argmax(1).item() trg_ids.append(pred_id) # <END>が出たら終了 if pred_id == trg_vocab[‘<end>’]: break # 4. IDを単語に変換 id_to_word = {v: k for k, v in trg_vocab.items()} translation = [id_to_word[id] for id in trg_ids[1:-1]] # <START>と<END>を除く return translation