双向RNN

如果不加特别说明RNN的计算一般是从左到右的即第t步的特征计算只能“看到”的信息但从语义理解的角度来看有时只看前边时间步的信息是不够的。我们来看下边的例子。南京市 长江 大桥 是 一个 工程 奇迹。 南京 市长 江大桥 是 一个 好同志。“南京市长江大桥”这个短语的理解存在歧义在没有看到后边的上下文时无法对其进行正确的分词。双向RNN使得模型可以同时看到前边和后边的信息。上图显示的是利用双向RNN的示意图。在计算时先从前到后计算再将作为后向RNN序列的初始化 (相当于)再从后到前依次计算。前向RNN和后向RNN是两套独立的RNN它们的参数是不同的。在前向RNN的计算中包含了全部的序列信息在后向RNN的计算中包含了全部的序列信息。在进行序列分类任务时我们将与合并在一起作为整个序列的表征。在经过输出层之后 (Linear Softmax)最后输出其中Y为表示预测输出的随机变量概率的预测公式为在进行序列标注任务时将与合并在一起其中集成了的信息集成了的信息因此能够看到序列里全部的信息。之后基于计算每个时间步的输出其中test_rnn_sequence_labeling.py# -*- coding: utf-8 -*- 基于 RNN 的序列标注Sequence Labeling示例程序 功能使用单层 RNN 对句子中的每个字符进行标注S/B/M/E 四类并演示一次完整的训练迭代前向传播 → 计算损失 → 反向传播 → 参数更新。 标注体系 S (Single) : 单字词 B (Begin) : 词首 M (Middle) : 词中 E (End) : 词尾 运行前提 1. 当前目录下存在 vocab.txt词表文件 2. 同目录下存在 tokenizer.py提供 MyTokenizer 类。 import torch.nn as nn import torch from tokenizer import MyTokenizer import torch.optim as optim class MySequenceLabelingModel(nn.Module): 基于 RNN 的序列标注模型。 结构Embedding词嵌入 → RNN循环层 → Linear输出层 输入形如 (batch_size, max_length) 的词索引矩阵 输出形如 (batch_size, max_length, output_dim) 的 logits 矩阵 def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim, pad_id0, embedding_matrixNone): 初始化模型各层。 Args: vocab_size: 词表大小即 Embedding 的词典数量。 embedding_dim: 词向量维度。 hidden_dim: RNN 隐藏层维度。 output_dim: 输出类别数此处为标签类别数 4。 pad_id: padding 位置对应的词表索引其向量恒为 0 且不参与更新。 embedding_matrix: 预训练词向量矩阵torch.Tensor为 None 时随机初始化 Embedding。 super().__init__() # 词嵌入层支持随机初始化与预训练初始化两种方式 if embedding_matrix is None: # 随机初始化词向量 self.embedding nn.Embedding(vocab_size, embedding_dim, padding_idxpad_id) else: # 使用预训练矩阵初始化freezeFalse 表示随训练微调 self.embedding nn.Embedding.from_pretrained(embeddingsembedding_matrix, freezeFalse, padding_idxpad_id) # 循环层单层标准 RNN默认 tanh 激活batch_firstTrue 使输入为 (batch, seq, feat) self.rnn nn.RNN(input_sizeembedding_dim, hidden_sizehidden_dim, batch_firstTrue) # 输出层将每个时间步的隐藏状态映射为各类别的 logits self.linear nn.Linear(hidden_dim, output_dim) def forward(self, inputs): 前向传播。 Args: inputs: 词索引张量tensor形状 (batch_size, max_length)。 Returns: 每个位置各类别的 logits形状 (batch_size, max_length, output_dim)。 # 词嵌入索引 → 词向量 (batch_size, max_length, embedding_dim) embedding_outputs self.embedding(inputs) # RNN 编码返回所有时间步输出与最后一步隐藏状态此处只需前者 (batch_size, seq_len, hidden_dim) rnn_outputs, _ self.rnn(embedding_outputs) # 线性输出层每个时间步独立映射 (batch_size, seq_len, output_dim) return self.linear(rnn_outputs) if __name__ __main__: # ---------- 1. 数据准备 ---------- # 加载词表相对路径需在项目根目录运行 tokenizer MyTokenizer(vocab.txt) # 标签编码S单字词、B词首、M词中、E词尾 label_dict { S: 0, B: 1, M: 2, E: 3 } # 训练样本两个短句及其逐字标签标签长度与句长一一对应 texts [今天天气很好。, 但是明天要下雨。] labels [BEBEBES, BEBESBES] # 统一序列长度最长句子字符数 4s、/s 各占 1 位再预留 2 位 max_length max([len(text) for text in texts]) 4 samples [] # 存放 (x, y) 样本对 for text, label in zip(texts, labels): # 句子首尾添加开始/结束标记 char_list [s] [char for char in text] [/s] # 标签序列首尾对应补 S label_list [S] [label for label in label] [S] # 字符 → 词表索引 x [tokenizer.word2index[c] for c in char_list] # 标签字符 → 数字标签 y [label_dict[l] for l in label_list] # 不足 max_length 的部分补齐 if len(x) max_length: x [tokenizer.pad_id] * (max_length - len(x)) # 输入用 pad_id 填充 y [-100] * (max_length - len(y)) # padding 位置的 label 设置为 -100 (标签用 -100 填充)会被 CrossEntropyLoss 忽略 samples.append((x, y)) # ---------- 2. 构建模型、损失函数与优化器 ---------- model MySequenceLabelingModel(len(tokenizer.vocab), 300, 300, 4, None) loss_fn nn.CrossEntropyLoss() optimizer optim.SGD(model.parameters(), lr0.01) # ---------- 3. 前向传播与损失计算 ---------- batch_x torch.LongTensor([sample[0] for sample in samples]) batch_y torch.LongTensor([sample[1] for sample in samples]) print(batch_x.shape, batch_y.shape) outputs model(batch_x) preds outputs.reshape(-1, 4) labels batch_y.reshape(-1) loss loss_fn(preds, labels) print(loss) # ---------- 4. 反向传播与参数更新 ---------- loss.backward() optimizer.step() optimizer.zero_grad() # ---------- 5. 验证更新后再前向一次loss 应有所下降 ---------- outputs model(batch_x) preds outputs.reshape(-1, 4) loss loss_fn(preds, labels) print(loss)tokenizer.pyimport fasttext import torch import numpy as np class MyTokenizer: def __init__(self, vocab, padpad): self.vocab self.load_vocab(vocab) self.vocab_size len(self.vocab) self.word2index {word: i for i, word in enumerate(self.vocab)} self.index2word {i: word for i, word in enumerate(self.vocab)} self.pad pad self.pad_id self.word2index[pad] def tokenize(self, text): words text.split() return [self.word2index[word] for word in words if word in self.vocab] staticmethod def load_vocab(file_path): vocab [] with open(file_path, r) as f: for line in f.readlines(): vocab.append(line.strip()) return vocab #def tokenize_file(toknizer, input_file, output_file): # with open(input_file, r) as f: # with open(output_file, w) as out: # for line in f.readlines(): # try: # label, text line.strip().split(\t) # tokenized_text tokenizer.tokenize(text) # out.write(label \t .join(map(str, tokenized_text)) \n) # except: # continue ## main #if __name__ __main__: # # tokenizer MyTokenizer(vocab.txt, cc.zh.300.bin) # #vectors tokenizer.get_vector_tensor() # # tokenize_file(tokenizer, data/toutiao_cat_data_train_2.txt, data/toutiao_cat_data_train_2_tokenized.txt) # #tokenize_file(tokenizer, toutiao_cat_data_test.txt, toutiao_cat_data_test_tokenized.txt)