- **🍨 本文为[🔗365天深度学习训练营](https://mp.weixin.qq.com/s/o-DaK6aQQLkJ8uE4YX1p3Q) 中的学习记录博客**
- **🍖 原作者：[K同学啊](https://mtyjkh.blog.csdn.net/)** 

---

文章目录

• • 概要
  • 整体架构流程
  • 代码运行
  • 技术名词解释
  • 小结

概要

同第6周数据 法语 → 英语 机器翻译

整体架构流程

1. 数据准备
   原始文本 → 分词 → 构建词表 → 句子对
   
2. Encoder处理（修改点：保存所有输出）
   输入序列 → GRU逐词处理 → 保存每步的output
   
3. Attention计算（新增模块）
   Decoder当前状态 + 所有Encoder输出 → 计算权重 → 上下文向量
   
4. Decoder处理（修改点：使用Attention）
   上下文向量 + 当前输入 → GRU → 输出预测
   
5. 反向传播
   计算Loss → 更新参数

代码运行

原代码（同第6周）

from __future__ import unicode_literals, print_function, division
from io import open
import unicodedata
import string
import re
import random

import torch
import torch.nn as nn
from torch import optim
import torch.nn.functional as F

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(device)

# 定义了两个常量，SOS_token和EOS_token分别代表序列的开始和结束
SOS_token = 0  
EOS_token = 1

# 语言类，方便对语料库进行操作
class Lang:
    def __init__(self, name):
        self.name = name
        self.word2index = {}
        self.word2count = {}
        self.index2word = {0: "SOS", 1: "EOS"}
        self.n_words    = 2  # Count SOS and EOS

    def addSentence(self, sentence):
        for word in sentence.split(' '):
            self.addWord(word)

    def addWord(self, word):
        if word not in self.word2index:
            self.word2index[word] = self.n_words
            self.word2count[word] = 1
            self.index2word[self.n_words] = word
            self.n_words += 1
        else:
            self.word2count[word] += 1   # 只增加该单词的出现次数


# 文本处理函数
# 去除重音符号，将Unicode字符转换为基本ASCII字符
def unicodeToAscii(s):
    return ''.join(
        c for c in unicodedata.normalize('NFD', s)
        if unicodedata.category(c) != 'Mn'
    )

# 小写化，剔除标点与非字母符号
def normalizeString(s):
    s = unicodeToAscii(s.lower().strip())
    s = re.sub(r"([.!?])", r" \1", s)
    s = re.sub(r"[^a-zA-Z.!?]+", r" ", s)
    return s

# 文件处理函数
def readLangs(lang1, lang2, reverse=False):
    print("Reading lines...")

    # 以行为单位读取文件 这步将传入数据定义为函数的一部分
    lines = open('/home/zyjiang/eng-fra.txt', encoding='utf-8') \
            .read().strip().split('\n')

    # 将每一行放入一个列表中
    # 一个列表中有两个元素，A语言文本与B语言文本
    pairs = [[normalizeString(s) for s in l.split('\t')] for l in lines]

    # 创建Lang实例，并确认是否反转语言顺序
    if reverse:
        pairs       = [list(reversed(p)) for p in pairs]
        input_lang  = Lang(lang2)
        output_lang = Lang(lang1)
    else:
        input_lang  = Lang(lang1)
        output_lang = Lang(lang2)

    return input_lang, output_lang, pairs

# 过滤
MAX_LENGTH = 10      # 定义语料最长长度

eng_prefixes = (
    "i am ", "i m ",
    "he is", "he s ",
    "she is", "she s ",
    "you are", "you re ",
    "we are", "we re ",
    "they are", "they re "
)

def filterPair(p):
    return len(p[0].split(' ')) < MAX_LENGTH and \
           len(p[1].split(' ')) < MAX_LENGTH and p[1].startswith(eng_prefixes)
def filterPairs(pairs):
    # 选取仅仅包含 eng_prefixes 开头的语料
    return [pair for pair in pairs if filterPair(pair)]

# 辅助函数
def tensorFromSentence(lang, sentence):
    indexes = [lang.word2index[word] for word in sentence.split(' ')]
    indexes.append(EOS_token)
    return torch.tensor(indexes, dtype=torch.long, device=device).view(-1, 1)

def tensorsFromPair(pair):
    input_tensor = tensorFromSentence(input_lang, pair[0])
    target_tensor = tensorFromSentence(output_lang, pair[1])
    return (input_tensor, target_tensor)

def prepareData(lang1, lang2, reverse=False):
    # 读取文件中的数据
    input_lang, output_lang, pairs = readLangs(lang1, lang2, reverse)
    print("Read %s sentence pairs" % len(pairs))
    
    # 按条件选取语料
    pairs = filterPairs(pairs[:])
    print("Trimmed to %s sentence pairs" % len(pairs))
    print("Counting words...")
    
    # 将语料保存至相应的语言类
    for pair in pairs:
        input_lang.addSentence(pair[0])
        output_lang.addSentence(pair[1])
        
    # 打印语言类的信息    
    print("Counted words:")
    print(input_lang.name, input_lang.n_words)
    print(output_lang.name, output_lang.n_words)
    return input_lang, output_lang, pairs

input_lang, output_lang, pairs = prepareData('eng', 'fra', True)
print(random.choice(pairs))

import torch
import torch.nn as nn
import torch.nn.functional as F

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

class EncoderRNN(nn.Module):
    def __init__(self, input_size, hidden_size):
        super(EncoderRNN, self).__init__()
        self.hidden_size = hidden_size
        self.embedding   = nn.Embedding(input_size, hidden_size)
        self.gru         = nn.GRU(hidden_size, hidden_size)

    def forward(self, input, hidden):
        embedded       = self.embedding(input).view(1, 1, -1)
        output         = embedded
        output, hidden = self.gru(output, hidden)
        return output, hidden

    def initHidden(self):
        return torch.zeros(1, 1, self.hidden_size, device=device)

新增Attention代码

# ============================================
# 2. Attention机制 (新增)
# ============================================
class Attention(nn.Module):
    """
    注意力机制模块
    计算decoder当前状态与所有encoder输出的相关性
    """
    def __init__(self, hidden_size):
        super(Attention, self).__init__()
        self.hidden_size = hidden_size
        # 用于计算注意力分数的线性层
        self.attn = nn.Linear(hidden_size * 2, hidden_size)
        self.v = nn.Linear(hidden_size, 1, bias=False)
    
    def forward(self, decoder_hidden, encoder_outputs):
        """
        参数:
            decoder_hidden: [1, 1, hidden_size] Decoder当前隐藏状态
            encoder_outputs: [seq_len, 1, hidden_size] Encoder所有输出
        返回:
            context: [1, 1, hidden_size] 上下文向量
            attention_weights: [1, seq_len] 注意力权重
        """
        seq_len = encoder_outputs.size(0)
        
        # 步骤1: 重复decoder_hidden以匹配encoder_outputs的长度
        # [1, 1, hidden_size] → [seq_len, 1, hidden_size]
        decoder_hidden_repeated = decoder_hidden.repeat(seq_len, 1, 1)
        
        # 步骤2: 拼接decoder_hidden和每个encoder_output
        # [seq_len, 1, hidden_size*2]
        combined = torch.cat((decoder_hidden_repeated, encoder_outputs), dim=2)
        
        # 步骤3: 计算注意力分数
        # [seq_len, 1, hidden_size*2] → [seq_len, 1, hidden_size] → [seq_len, 1, 1]
        energy = torch.tanh(self.attn(combined))
        attention_scores = self.v(energy).squeeze(2)  # [seq_len, 1]
        
        # 步骤4: Softmax归一化得到注意力权重
        # [seq_len, 1] → [1, seq_len]
        attention_weights = F.softmax(attention_scores, dim=0).transpose(0, 1)
        
        # 步骤5: 加权求和得到上下文向量
        # [1, seq_len] x [seq_len, 1, hidden_size] → [1, 1, hidden_size]
        context = torch.bmm(attention_weights.unsqueeze(0), 
                           encoder_outputs.transpose(0, 1))
        
        return context, attention_weights

# ============================================
# 3. 带注意力的Decoder (新版本)
# ============================================
class AttnDecoderRNN(nn.Module):
    def __init__(self, hidden_size, output_size, dropout_p=0.1):
        super(AttnDecoderRNN, self).__init__()
        self.hidden_size = hidden_size
        self.output_size = output_size
        
        # 基础层
        self.embedding = nn.Embedding(output_size, hidden_size)
        self.dropout = nn.Dropout(dropout_p)
        
        # 注意力模块
        self.attention = Attention(hidden_size)
        
        # GRU: 输入是embedding + context，所以是hidden_size * 2
        self.gru = nn.GRU(hidden_size * 2, hidden_size)
        
        # 输出层
        self.out = nn.Linear(hidden_size, output_size)
        self.softmax = nn.LogSoftmax(dim=1)
    
    def forward(self, input, hidden, encoder_outputs):
        """
        参数:
            input: [1] 当前输入词索引
            hidden: [1, 1, hidden_size] 上一步的隐藏状态
            encoder_outputs: [seq_len, 1, hidden_size] Encoder所有输出
        返回:
            output: [1, output_size] 词概率分布
            hidden: [1, 1, hidden_size] 新的隐藏状态
            attention_weights: [1, seq_len] 注意力权重
        """
        # 步骤1: Embedding + Dropout
        embedded = self.embedding(input).view(1, 1, -1)
        embedded = self.dropout(embedded)
        
        # 步骤2: 计算注意力，得到上下文向量
        context, attention_weights = self.attention(hidden, encoder_outputs)
        
        # 步骤3: 拼接embedding和context
        # [1, 1, hidden_size] + [1, 1, hidden_size] → [1, 1, hidden_size*2]
        gru_input = torch.cat((embedded, context), dim=2)
        
        # 步骤4: GRU处理
        output, hidden = self.gru(gru_input, hidden)
        
        # 步骤5: 线性层 + Softmax输出
        output = self.softmax(self.out(output[0]))
        
        return output, hidden, attention_weights
    
    def initHidden(self):
        return torch.zeros(1, 1, self.hidden_size, device=device)

# ============================================
# 4. 训练函数 (修改版)
# ============================================
def train_with_attention(input_tensor, target_tensor, encoder, decoder, 
                        encoder_optimizer, decoder_optimizer, criterion, 
                        max_length=50):
    """
    带注意力机制的训练函数
    """
    # 初始化
    encoder_hidden = encoder.initHidden()
    encoder_optimizer.zero_grad()
    decoder_optimizer.zero_grad()
    
    input_length = input_tensor.size(0)
    target_length = target_tensor.size(0)
    
    # ===== 关键变化1: 保存所有encoder输出 =====
    encoder_outputs = torch.zeros(max_length, 1, encoder.hidden_size, device=device)
    
    loss = 0
    
    # Encoder阶段: 保存每个时间步的输出
    for ei in range(input_length):
        encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)
        encoder_outputs[ei] = encoder_output[0, 0]  # 保存输出
    
    # Decoder阶段
    decoder_input = torch.tensor([[0]], device=device)  # SOS_token
    decoder_hidden = encoder_hidden
    
    # ===== 关键变化2: 传入encoder_outputs =====
    for di in range(target_length):
        decoder_output, decoder_hidden, attention_weights = decoder(
            decoder_input, decoder_hidden, encoder_outputs)
        
        # 计算loss
        loss += criterion(decoder_output, target_tensor[di])
        
        # Teacher forcing: 使用真实目标词作为下一步输入
        decoder_input = target_tensor[di]
    
    # 反向传播
    loss.backward()
    encoder_optimizer.step()
    decoder_optimizer.step()
    
    return loss.item() / target_length

# ============================================
# 5. 预测函数 (修改版)
# ============================================
def evaluate_with_attention(encoder, decoder, input_tensor, max_length=50):
    """
    带注意力机制的预测函数
    返回翻译结果和注意力权重（可用于可视化）
    """
    with torch.no_grad():
        # Encoder
        encoder_hidden = encoder.initHidden()
        encoder_outputs = torch.zeros(max_length, 1, encoder.hidden_size, device=device)
        
        input_length = input_tensor.size(0)
        for ei in range(input_length):
            encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)
            encoder_outputs[ei] = encoder_output[0, 0]
        
        # Decoder
        decoder_input = torch.tensor([[0]], device=device)  # SOS
        decoder_hidden = encoder_hidden
        
        decoded_words = []
        decoder_attentions = []  # 保存注意力权重
        
        for di in range(max_length):
            decoder_output, decoder_hidden, attention_weights = decoder(
                decoder_input, decoder_hidden, encoder_outputs)
            
            decoder_attentions.append(attention_weights.cpu())
            
            # 选择概率最高的词
            topv, topi = decoder_output.topk(1)
            if topi.item() == 1:  # EOS_token
                break
            else:
                decoded_words.append(topi.item())
            
            decoder_input = topi.squeeze().detach()
        
        return decoded_words, decoder_attentions

import time
import math
import random
import torch.optim as optim

def asMinutes(s):
    m = math.floor(s / 60)
    s -= m * 60
    return '%dm %ds' % (m, s)

def timeSince(since, percent):
    now = time.time()
    s = now - since
    es = s / (percent)
    rs = es - s
    return '%s (- %s)' % (asMinutes(s), asMinutes(rs))

def trainIters_with_attention(encoder, decoder, n_iters, pairs, 
                              print_every=1000, plot_every=100, learning_rate=0.01):
    """
    带注意力机制的完整训练循环
    """
    start = time.time()
    plot_losses = []
    print_loss_total = 0
    plot_loss_total = 0
    
    encoder_optimizer = optim.SGD(encoder.parameters(), lr=learning_rate)
    decoder_optimizer = optim.SGD(decoder.parameters(), lr=learning_rate)
    
    # 随机选取训练数据
    training_pairs = [tensorsFromPair(random.choice(pairs)) for i in range(n_iters)]
    criterion = nn.NLLLoss()
    
    for iter in range(1, n_iters + 1):
        training_pair = training_pairs[iter - 1]
        input_tensor = training_pair[0]
        target_tensor = training_pair[1]
        
        # 使用带注意力的训练函数
        loss = train_with_attention(input_tensor, target_tensor, encoder,
                                   decoder, encoder_optimizer, decoder_optimizer, criterion)
        
        print_loss_total += loss
        plot_loss_total += loss
        
        if iter % print_every == 0:
            print_loss_avg = print_loss_total / print_every
            print_loss_total = 0
            print('%s (%d %d%%) %.4f' % (timeSince(start, iter / n_iters),
                                         iter, iter / n_iters * 100, print_loss_avg))
        
        if iter % plot_every == 0:
            plot_loss_avg = plot_loss_total / plot_every
            plot_losses.append(plot_loss_avg)
            plot_loss_total = 0
    
    return plot_losses

# ============================================
# 7. 训练模型（假设你已经有 input_lang, output_lang, pairs）
# ============================================

# 创建带注意力的模型
hidden_size = 256
encoder_attn = EncoderRNN(input_lang.n_words, hidden_size).to(device)
decoder_attn = AttnDecoderRNN(hidden_size, output_lang.n_words).to(device)

print("开始训练带注意力机制的模型...")
plot_losses_attn = trainIters_with_attention(
    encoder_attn, decoder_attn, 
    n_iters=20000, 
    pairs=pairs,
    print_every=5000
)

print("训练完成！")


# ============================================
# 8. 画图对比（如果你之前训练了无注意力的模型）
# ============================================
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings("ignore")
plt.rcParams['axes.unicode_minus'] = False
plt.rcParams['figure.dpi'] = 100

plt.figure(figsize=(12, 5))

plt.plot(plot_losses_attn, label='With Attention', color='orange')
plt.xlabel('Iterations (x100)')
plt.ylabel('Loss')
plt.title('Training Loss - With Attention')
plt.legend()

plt.tight_layout()
plt.show()

# ============================================
# 9. 注意力可视化
# ============================================
import matplotlib.pyplot as plt
import matplotlib.ticker as ticker
import warnings
warnings.filterwarnings("ignore")

def evaluate(encoder, decoder, sentence, max_length=50):
    """
    评估单个句子，返回输出词列表和注意力权重矩阵
    """
    with torch.no_grad():
        # 将句子转换为张量
        input_tensor = tensorFromSentence(input_lang, sentence)
        input_length = input_tensor.size(0)
        
        # Encoder
        encoder_hidden = encoder.initHidden()
        encoder_outputs = torch.zeros(max_length, 1, encoder.hidden_size, device=device)
        
        for ei in range(input_length):
            encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden)
            encoder_outputs[ei] = encoder_output[0, 0]
        
        # Decoder
        decoder_input = torch.tensor([[SOS_token]], device=device)
        decoder_hidden = encoder_hidden
        
        decoded_words = []
        # 注意力矩阵: [max_output_length, input_length]
        attentions = torch.zeros(max_length, max_length)
        
        for di in range(max_length):
            decoder_output, decoder_hidden, attention_weights = decoder(
                decoder_input, decoder_hidden, encoder_outputs
            )
            
            # 保存注意力权重 (只取input_length部分)
            attentions[di, :attention_weights.size(1)] = attention_weights.squeeze(0).cpu()
            
            # 获取预测词
            topv, topi = decoder_output.topk(1)
            if topi.item() == EOS_token:
                decoded_words.append('<EOS>')
                break
            else:
                decoded_words.append(output_lang.index2word[topi.item()])
            
            decoder_input = topi.squeeze().detach().view(1, 1)
        
        # 截取实际长度的注意力矩阵
        return decoded_words, attentions[:len(decoded_words), :input_length]


def showAttention(input_sentence, output_words, attentions):
    """
    用热力图可视化注意力权重
    """
    fig = plt.figure(figsize=(10, 10))
    ax = fig.add_subplot(111)
    
    # 绘制热力图，使用 'bone' 配色（黑白灰）
    cax = ax.matshow(attentions.numpy(), cmap='bone')
    fig.colorbar(cax)
    
    # 设置坐标轴标签
    # X轴: 输入词（法语）
    ax.set_xticklabels([''] + input_sentence.split(' ') + ['<EOS>'], rotation=90)
    # Y轴: 输出词（英语）
    ax.set_yticklabels([''] + output_words)
    
    # 确保每个刻度都显示
    ax.xaxis.set_major_locator(ticker.MultipleLocator(1))
    ax.yaxis.set_major_locator(ticker.MultipleLocator(1))
    
    plt.tight_layout()
    plt.show()


def evaluateAndShowAttention(input_sentence):
    """
    评估并可视化注意力
    """
    output_words, attentions = evaluate(encoder_attn, decoder_attn, input_sentence)
    print('input =', input_sentence)
    print('output =', ' '.join(output_words))
    showAttention(input_sentence, output_words, attentions)


# ============================================
# 10. 测试可视化
# ============================================
evaluateAndShowAttention("elle a cinq ans de moins que moi .")
evaluateAndShowAttention("elle est trop petit .")
evaluateAndShowAttention("je ne crains pas de mourir .")
evaluateAndShowAttention("c est un jeune directeur plein de talent .")

input = elle a cinq ans de moins que moi .

output = she is as his own as me . <EOS>

---

优化

第七步 训练模型

# ============================================
# 7. 训练模型
# ============================================

# 创建带注意力的模型
hidden_size = 256
encoder_attn = EncoderRNN(input_lang.n_words, hidden_size).to(device)
decoder_attn = AttnDecoderRNN(hidden_size, output_lang.n_words).to(device)

def trainIters_improved(encoder, decoder, n_iters, pairs, 
                        print_every=5000, plot_every=100, learning_rate=0.001):
    start = time.time()
    plot_losses = []
    print_loss_total = 0
    plot_loss_total = 0
    
    # 改用 Adam 优化器（比SGD收敛快很多）
    encoder_optimizer = optim.Adam(encoder.parameters(), lr=learning_rate)
    decoder_optimizer = optim.Adam(decoder.parameters(), lr=learning_rate)
    
    training_pairs = [tensorsFromPair(random.choice(pairs)) for i in range(n_iters)]
    criterion = nn.NLLLoss()
    
    for iter in range(1, n_iters + 1):
        training_pair = training_pairs[iter - 1]
        input_tensor = training_pair[0]
        target_tensor = training_pair[1]
        
        loss = train_with_attention(input_tensor, target_tensor, encoder,
                                   decoder, encoder_optimizer, decoder_optimizer, criterion)
        
        print_loss_total += loss
        plot_loss_total += loss
        
        if iter % print_every == 0:
            print_loss_avg = print_loss_total / print_every
            print_loss_total = 0
            print('%s (%d %d%%) %.4f' % (timeSince(start, iter / n_iters),
                                         iter, iter / n_iters * 100, print_loss_avg))
        
        if iter % plot_every == 0:
            plot_loss_avg = plot_loss_total / plot_every
            plot_losses.append(plot_loss_avg)
            plot_loss_total = 0
    
    return plot_losses

# 重新训练：迭代次数增加到75000
print("开始改进版训练...")
plot_losses_attn = trainIters_improved(
    encoder_attn, decoder_attn, 
    n_iters=75000,          # 增加到75000
    pairs=pairs,
    print_every=5000,
    learning_rate=0.001     # Adam用较小学习率
)
print("训练完成！")


# ============================================
# 8. 画图对比
# ============================================
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings("ignore")
plt.rcParams['axes.unicode_minus'] = False
plt.rcParams['figure.dpi'] = 100

plt.figure(figsize=(12, 5))

plt.plot(plot_losses_attn, label='With Attention', color='orange')
plt.xlabel('Iterations (x100)')
plt.ylabel('Loss')
plt.title('Training Loss - With Attention')
plt.legend()

plt.tight_layout()
plt.show()

注意力可视化结果

正确版本

小结