NLP中的对话机器人——问答机器人的应用场景

Posted 愤怒的可乐

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了NLP中的对话机器人——问答机器人的应用场景相关的知识,希望对你有一定的参考价值。

引言

本文是七月在线《NLP中的对话机器人》的视频笔记,主要介绍FAQ问答型聊天机器人的实现。

讲得还不错,关键是只要1分钱

FAQ问答机器人

FAQ就是一些常见问题与回答,比如https://letsencrypt.org/docs/faq/。

但是我们要做的不是一问一答形式的,而是类似stackoverflow那种一问多答,即包括用户提问、网友回答和最佳答案。有人提问,然后会有人在上面回复,每个问题可能有多个回答。

数据集仓库地址 : https://github.com/SophonPlus/ChineseNlpCorpus

数据集

我们先来了解下数据集。

可以看到,有4个字段,其中标题和问题类似发帖时的标题和正文,问题可以为空。

剩下的是reply和is_best分别代表回复和是否为最佳答案。

从这种数据集我们可以想一下它的应用场景。

场景一

假设在提问者手动选择最佳答案之前,我们可以

  • 对多个回答进行排序,最相关的、最好的排在前面,不好的排在后面
  • 从这些回答中找出最佳回答。二分类任务

数据

我们这里用的数据是农行问答数据,下载地址: https://pan.baidu.com/s/1n-jT9SKkt6cwI_PjCd7i_g

模型

关于这种类似的任务,我们应该得到句子的向量表示,即句向量。可能说到句向量,大家第一时间想到的都是BERT来实现,但这里我们先用简单的模型来实现,简单的模型速度快,可以快速验证我们的思路。

关于问题和回答我们都需要一个句向量编码,我们可以采用Dual Encoder的架构,训练两个编码器,一个用于问题的编码,另一个用于回答的编码。这两个编码器是独立的。我们知道,编码器的选择一般有RNN、CNN和Transformer等。得到了问题和回答的句向量编码后,我们可以使用余弦相似度来计算问题和回答的匹配程度,也可以使用一个复杂一点的神经网络来计算匹配度。

每当拿到一个新的问题和数据,建议从最简单的模型开始,搭建出一个baseline,然后以这个baseline为基础开始调试自己的模型。因为简单的模型往往表现不会太差,容易调试,且模型的可解释性好。当我们确定自己的baseline没有问题之后可以开始尝试更复杂的模型,通过循序渐进地尝试。一般从一些简单的模型开始一步步叠加新的组件,直到效果令人满意位置。如果一开始就采用太复杂的模型,很多情况下我们就无法理解究竟模型中的哪些组件时重要的,哪些是不重要的。

实现

本节包含完整的代码,首先是需要引入的依赖:

from collections import Counter
import pickle
import os

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

import numpy as np
import pandas as pd

from tqdm import tqdm

DualEncoder

Dual Encoder即两个独立的Encoder,这里分别计算问题和回答的句向量,最后通过余弦相似度计算它们之间的关联程度。

# DualEncoder
class DualEncoder(nn.Module):
  def __init__(self, encoder1, encoder2, type="cosine"):
    super(DualEncoder, self).__init__()
    self.encoder1 = encoder1
    self.encoder2 = encoder2
    if type != 'cosine':
      # 训练一个简单的神经网络来计算相似度
      self.linear = nn.Sequential(
          # 拼接encoder1和encoder2的输出(向量),转换成100维的表示
          nn.Linear(self.encoder1.hidden_size + self.encoder2.hidden_size, 100),
          # 经过ReLU激活
          nn.ReLU(),
          # 再转换成一个数值,表示相似程度
          nn.Linear(100, 1)
      )
  
  def forward(self, x, x_mask, y, y_mask):
    x_rep = self.encoder1(x, x_mask)
    y_rep = self.encoder2(y, y_mask)
    return x_rep, y_rep
  

主要实现了前向传播方法。

Encoder

# GRUEncoder
class GRUEncoder(nn.Module):
  def __init__(self, vocab_size, embed_size, hidden_size, dropout_p=0.1, avg_hidden=True, n_layers=1, bidirectional=True):
    super(GRUEncoder, self).__init__()
    self.hidden_size = hidden_size
    self.embed = nn.Embedding(vocab_size, embed_size)
    if bidirectional:
      # 大小除以2,使得拼接两个方向后大小不变
      hidden_size //= 2
    # 这种生成句子表征的建议使用bidirectional=True
    self.rnn = nn.GRU(embed_size, hidden_size, num_layers=n_layers, bidirectional=bidirectional,dropout=dropout_p)
    self.dropout = nn.Dropout(dropout_p)
    self.bidirectional = bidirectional
    self.avg_hidden = avg_hidden
  
  def forward(self, x, mask):
    x_embed = self.embed(x) # 先得到嵌入表示
    x_embed = self.dropout(x_embed) # 再经过dropout
    seq_len = mask.sum(1) # 计算有效长度
    # 压缩批次内填充数据
    # 通过压缩填充加快训练效率,具体可参考文章: https://blog.csdn.net/yjw123456/article/details/118855324
    x_embed = torch.nn.utils.rnn.pack_padded_sequence(
      input=x_embed,
      lengths=seq_len.cpu(),
      batch_first=True,
      enforce_sorted=False
    )
    output, hidden = self.rnn(x_embed)


    # output (batch_size, seq_len, hidden_size)
    # hidden (num_directions * num_layers, batch_size, hidden_size)
    output, seq_len = torch.nn.utils.rnn.pad_packed_sequence(
      sequence=output,
      batch_first=True,
      padding_value=0,
      total_length=mask.shape[1]
    )
 
  

    if self.avg_hidden:
      # 对RNN输出每个时刻的输出求均值
      # mask.unsqueeze(2) 使维度个数和output一致
      # hiden (batch_size, hidden_size)
      hidden = torch.sum(output * mask.unsqueeze(2), 1) / torch.sum(mask, 1, keepdim=True)
    else:
      if self.bidirectional:
        # 拼接两个方向上的输出
        # hidden[-2,:,:] (batch_size, hidden_size / 2)
        # hidden[-1,:,:] (batch_size, hidden_size / 2)
        # hidden (batch_size, hidden_size)
        hidden = torch.cat((hidden[-2,:,:], hidden[-1,:,:]),dim=1)
      else:
        # 取出最顶层(若num_layers > 1)的hidden
        hidden = hidden[-1,:,:]
    
    # 需要保证各种情况下的hidden大小都是一致的

    # 经过一层dropout
    hidden = self.dropout(hidden)
    return hidden
   

这里采用GRU作为Encoder的实现,支持多种表征的获取,默认是平均每个时刻的输出。

词典

处理NLP任务基本上都需要一个词典:

# 构建分词器
class Tokenizer:
  def __init__(self, vocab):
    self.id2word = ["UNK"] + vocab # 保证未知词UNK的id为0
    self.word2id = w:i for i,w in enumerate(vocab)
  
  def text2id(self, text):
    # 对中文简单的按字拆分
    return [self.word2id.get(w, 0) for w in str(text)]
  
  def id2text(self, ids):
    return "".join([self.id2word[id] for id in ids])
  
  def __len__(self):
    return len(self.id2word)

def create_tokenizer(texts, vocab_size):
  """
  创建分词器,输入文本列表和词典大小
  """
  all_vocab = ""
  for text in texts:
    all_vocab += str(text)
  
  vocab_count = Counter(all_vocab) # 按字拆分
  # 最频繁的vocab_size个单词
  vocab = vocab_count.most_common(vocab_size)
  # (char, count) 从中取出char
  vocab = [w[0] for w in vocab]
  return Tokenizer(vocab)

def list2tensor(sents, tokenizer):
  """
  将文本列表结合分词器转换为tensor
  """
  res = []
  mask = []
  for sent in sents:
    res.append(tokenizer.text2id(sent))
  
  max_len = max([len(sent) for sent in res])
  # 按最大长度进行填充
  for i in range(len(res)):
    _mask = np.zeros((1, max_len)) # 中的元素0表示填充词,1表示非填充
    _mask[:,:len(res[i])] = 1 # 有效位元素置1
    
    res[i] = np.expand_dims(np.array(res[i] + [0] * (max_len - len(res[i]))), 0) # 增加一个维度
    mask.append(_mask)

  res = np.concatenate(res, axis=0) # 按维度0进行拼接
  mask = np.concatenate(mask, axis=0)
  # 分别转换为long类型和float类型的tensor
  res = torch.tensor(res).long()
  mask = torch.tensor(mask).float()
  return res, mask
    
    

这里的分词器结合了词典的功能,代码如上,我们可以通过text2id方法获取文本中每个字的ID。

加载数据集

数据集下载地址: https://pan.baidu.com/s/1n-jT9SKkt6cwI_PjCd7i_g

# 数据集位置
file_path = '../dataset/nonghangzhidao_filter.csv'
df = pd.read_csv(file_path)[["title", "reply", "is_best"]] # 只需要这三个字段
df.head()

看一下开头那么几条数据,对数据长什么样的有一个基本的了解。

拆分数据集

由于数据集本身没有进行拆分,因此我们这里实现拆分数据集的代码:

np.random.seed(42) # 设定随机种子可以防止每次的训练/测试集数据不一样
# 拆分训练/测试集
def shuffle_and_split_data(data, test_ratio):
  shuffled_indices = np.random.permutation(len(data))
  test_set_size = int(len(data) * test_ratio)
  test_indices = shuffled_indices[:test_set_size] # 前test_set_size作为测试集
  train_indices = shuffled_indices[test_set_size:] # 剩下的作为训练集
  return data.iloc[train_indices], data.iloc[test_indices]

# 20%的数据作为测试集
train_set, test_set = shuffle_and_split_data(df, 0.2)

print(len(train_set), len(test_set))
(31876, 7968)

在拆分数据集的同时进行了洗牌操作,打散数据。

这里设置了随机种子,方式每次运行的训练集、测试集中的数据都不一样。

将文本转换为张量

首先我们取出文本内容:

texts = list(train_set["title"]) + list(train_set["reply"]) # 取出文本内容

查看前10条:

print(texts[:10])
['有没有什么借款的口子?',
 '农行信用卡办哪一张好',
 '窝的银行卡必须要办理转账才能收到钱吗?',
 '请问成都兴百惠公司贷款不成功不收费是真的吗?',
 '得到死者老婆的身份证,还得到死者的户口本和火化证能取走死者银行卡里的钱吗',
 '借个爱奇艺vip会员,只用1天',
 '微信怎么能有微',
 '农村60岁拿钱的,在外地农行能交钱吗',
 '信而富为什么不下款了',
 '大众金融为什么还完贷款,不给办解压绿本']

下面我们创建词典,并基于词典将文本转换为张量:

tokenizer = create_tokenizer(texts, 5000)
print(len(tokenizer)) 
3383

词典中的单词(字)个数为3383。调用上面实现的list2tensor函数:

sents = list(train_set["title"][:3])
print(list2tensor(sents, tokenizer))
(tensor([[  15,  211,   15,  238,   97,   52,    4,    2,  488,  233,  125,    0,
             0,    0,    0,    0,    0,    0,    0],
         [   9,    1,   12,   13,   18,   62,  248,   39,  379,  178,    0,    0,
             0,    0,    0,    0,    0,    0,    0],
         [2231,    2,    7,    1,   18,  373,  442,   71,   62,   34,   64,   21,
           493,   63,   81,   41,  206,  100,  125]]),
 tensor([[1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 0., 0., 0., 0., 0., 0., 0.,
          0.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 0., 0., 0., 0., 0., 0., 0., 0.,
          0.],
         [1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 1., 1.,
          1.]]))

可以看到,同时返回了两个tensor,一个是文本对应的ID列表,另一个是mask,表示对应位置的单词是否为填充单词。填充单词用0表示,非填充单词用1表示。

这样做的好处是,可以在计算损失时乘以这个mask,就可以忽略0处(填充词)的损失;通过mask.sum()就可以知道句子的有效长度。

编写训练代码

数据和模型都准备好了,下一步是编写训练代码:

# 编写训练代码
def train(df, model, loss_fn, optimizer, device, tokenizer, loss_function, batch_size):
  # 设成训练模型,让dropout生效
  model.train() 
  df = df.sample(frac=1) # 每次训练时shuffle数据
  # 分批处理
  for i in range(0, df.shape[0], batch_size): 
    # 得到批次数据
    batch_df = df.iloc[i:i+batch_size]
    title = list(batch_df["title"])
    reply = list(batch_df["reply"])
    # 构建目标tensor(1或0)
    target = torch.tensor(batch_df["is_best"].to_numpy()).float()
    if loss_function == "cosine":
       # 为了符合CosineEmbeddingLoss的要求,将0替换成-1
      target[target == 0] = -1 
    
    x, x_mask = list2tensor(title, tokenizer)
    y, y_mask = list2tensor(reply, tokenizer)

    # 都切换到同一设备(cpu/gpu)
    x, x_mask, y, y_mask, target = x.to(device), x_mask.to(device), y.to(device), y_mask.to(device), target.to(device)
    # 计算x和y的表征
    x_rep, y_rep = model(x, x_mask, y, y_mask)

    # 根据需要使用不同的损失
    if loss_function == "cosine":
      loss = loss_fn(x_rep, y_rep, target)
    else:
      # 拼接x_pre和y_rep,并传入linear
      logits = model.linear(torch.cat([x_rep, y_rep], 1)) 
      loss = loss_fn(logits, target)

    optimizer.zero_grad()
    loss.backward()
    # 梯度裁剪
    nn.utils.clip_grad_norm_(model.parameters(), 1.0)
    optimizer.step()


    if loss_function == "cosine":
      sim = F.cosine_similarity(x_rep, y_rep)
      # 余弦相似度范围在[-1,1]之间,因此这里使用0作为分界
      sim[sim < 0] = -1
      sim[sim >= 0] = 1
    else:
      sim = model.linear(torch.cat([x_rep, y_rep], 1))
      # sim = torch.sigmoid(logits) 可以不用sigmoid
      sim[sim < 0] = 0
      sim[sim >= 0] = 1
    
    sim = sim.view(-1)
    target = target.view(-1)
    # 计算准确率
    num_corrects = torch.sum(sim == target).item()
    total_counts = target.shape[0]

  print(f"accuracy:num_corrects / total_counts")
  return num_corrects / total_counts

这里根据不同的设置使用了不同的损失函数,我们后面介绍。

定义参数

# 定义参数

loss_function = "cosine"
batch_size  = 64
output_dir  = "./models"
num_epochs  = 10
vocab_size  = 5000
hidden_size = 300
embed_size  = 600

我们可以根据需要调整这里的参数。

开始训练

# 构建两个Encoder
title_encoder = GRUEncoder(len(tokenizer), embed_size, hidden_size)
reply_encoder = GRUEncoder(len(tokenizer), embed_size, hidden_size)
# 传入DualEncoder模型
model = DualEncoder(title_encoder, reply_encoder, type=loss_function)
# 设置特定的损失函数
if loss_function == "cosine":
  loss_fn = nn.CosineEmbeddingLoss()
else:
  loss_fn = nn.BCEWithLogitsLoss()
# Adam优化器
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
# 有GPU就用GPU
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu") 
model = model.to(device)

if not os.path.exists(output_dir):
  os.makedirs(output_dir)

# 分词器(词典)可以重复使用,为了不出问题,在推理时要使用和训练时同样的分词器(词典)
pickle.dump(tokenizer, open(os.path.join(output_dir,"tokenizer.pickle"), "wb"))
for epoch in tqdm(range(num_epochs)):
  train(train_set, model, loss_fn, optimizer, device, tokenizer, loss_function, batch_size)

可以看到,这里有两种损失函数,分别是CosineEmbeddingLossBCEWithLogitsLoss

我们分别来看看。

CosineEmbeddingLoss

官方文档: https://pytorch.org/docs/stable/generated/torch.nn.CosineEmbeddingLoss.html

用于衡量两个向量是相似还是不相似的。输入主要为input1( x 1 x_1 x1)、input2( x 2 x_2 x2)、target( y y y)。

这里要求target也就是 y y y取值-11,取值1代表正例。

对于正例来说,我们希望它们的余弦距离(1-cosine相似度)尽可能小,余弦相似度取值范围是[-1,1],越接近1表示越相似,在向量中余弦相似度-1表示两个向量方向相反。所以 1 − cos ⁡ ( x 1 , x 2 ) 1-\\cos(x_1,x_2) 1cos(x1,x2)当相似度为以上是关于NLP中的对话机器人——问答机器人的应用场景的主要内容,如果未能解决你的问题,请参考以下文章

NLP人机对话系统应用场景及技术

NLP中的对话机器人——模型的评估

万字长文概述NLP中的深度学习技术

社区说|对话机器人技术在元宇宙之虚拟社交/虚拟替身场景下的应用

社区说|对话机器人技术在元宇宙之虚拟社交/虚拟替身场景下的应用

基于百度理解与交互技术实现机器问答