小型中文版聊天机器人

import re
from tqdm import tqdm
import zhconv
import config

#处理重复符号的表达，如替换多个重复符号
def delete_repeat(s):
    #注释掉的是英文的表达
    #s = re.sub('[!]+','!', s)
    #s = re.sub('[?]+','?', s)
    #s = re.sub('[,]+',',', s)
    #s = re.sub('[:]+',':', s)
    #s = re.sub('[;]+',';', s)
    s = re.sub('[，]+','，', s)
    s = re.sub('[！]+','！', s)
    s = re.sub('[？]+','？', s)
    s = re.sub('[：]+','：', s)
    s = re.sub('[；]+','；', s)
    s = re.sub('[。]+','。', s)
    s = re.sub('[、]+','、', s)
    return s

with open('data/origin_train.txt','r',encoding='utf-8') as f: #打开原始数据集
    lines = f.readlines()

train_datas = []
temp_data = ''
#每个多轮对话中使用'<EOS>'将其划分
for line in tqdm(lines):

    if line!='\n':
        line = line.strip() #去除前导后方空格
        #英文标点符号置换为中文标点符号
        line = line.replace('!','！')
        line = line.replace('?','？')
        line = line.replace(',','，')
        line = line.replace('.','。')
        line = line.replace(':','：')
        line = line.replace(';','；')
        line = zhconv.convert(line, 'zh-cn') #转为简体字
        line = " ".join(line)
        temp_data+=(line+' <EOS> ')
    else:
        if len(temp_data.split()) <= config.max_len: #限制长度
            train_datas.append(temp_data)
        temp_data=''

with open(config.data_path_txt,'w',encoding='utf-8') as f: #将处理后的数据保存在另一个文件中
    for train_data in train_datas:
        f.write(train_data+'\n')

7、处理完的数据展示

四、词表制作以及转化（word2seq.py文件）

先定义填充符<PAD>，未知符<UNK>和结束符<EOS>，然后再对数据集中的词进行标号，生成词表与转义词表，最后统计数据集中每个词出现的词频，生成一个词频表可以直观看看咱们的词表情况。

并且定义词到标号，标号到词的转化的方法，方便后期训练以及测试时使用。

其中，config.word_sequence_dict是保存词典的位置

#生成词表
#构造文本序列化和反序列化方法（文本转数字）
import pickle
import config
from tqdm import tqdm

class Word2Sequence():
    PAD_TAG = "<PAD>" #填充编码
    UNK_TAG = "<UNK>" #未知编码
    EOS_TAG = "<EOS>" #句子结尾

    #上面四种情况的对应编号
    PAD = 0
    UNK = 1
    EOS = 2

    def __init__(self):

        #文字——标号字典
        self.dict = {
            self.PAD_TAG :self.PAD,
            self.UNK_TAG :self.UNK,
            self.EOS_TAG :self.EOS
        }
        #词频统计
        self.count = {}
        self.fited = False #是否统计过词典了

    #以下两个转换都不包括'\t'
    #文字转标号（针对单个词）
    def to_index(self,word):
        """word -> index"""
        assert self.fited == True,"必须先进行fit操作"
        return self.dict.get(word,self.UNK) #无这个词则用未知代替

    #标号转文字（针对单个词）
    def to_word(self,index):
        """index -> word"""
        assert self.fited == True, "必须先进行fit操作"
        if index in self.inversed_dict:
            return self.inversed_dict[index]
        return self.UNK_TAG

    # 获取词典长度
    def __len__(self):
        return len(self.dict)

    #统计词频生成词典
    def fit(self, sentence):
        """
        :param sentence:[word1,word2,word3]
        """
        for a in sentence:
            if a not in self.count:
                self.count[a] = 0
            self.count[a] += 1

        self.fited = True

    def build_vocab(self, min_count=config.min_count, max_count=None, max_feature=None):

        """
        :param min_count: 最小出现的次数
        :param max_count: 最大出现的次数
        :param max_feature: 总词语的最大数量
        """

        # 限定统计词频范围
        if min_count is not None:
            self.count = {k: v for k, v in self.count.items() if v >= min_count}
        if max_count is not None:
            self.count = {k: v for k, v in self.count.items() if v <= max_count}

        # 给对应词进行编号
        if isinstance(max_feature, int): #是否限制词典的词数
            #词频从大到小排序
            count = sorted(list(self.count.items()), key=lambda x: x[1])
            if max_feature is not None and len(count) > max_feature:
                count = count[-int(max_feature):]
            for w, _ in count:
                self.dict[w] = len(self.dict)
        else: #按字典序(方便debug查看)
            for w in sorted(self.count.keys()):
                self.dict[w] = len(self.dict)

        # 准备一个index->word的字典
        self.inversed_dict = dict(zip(self.dict.values(), self.dict.keys()))

        #debug专用
        f_debug_word = open("data/debug_word.txt","w",encoding='utf-8')
        t = 0
        for key,_ in self.dict.items():
            t = t + 1
            if t >= 4: #排除那3种情况（填充，未知，结尾）
                f_debug_word.write(key+"★ "+str(self.count[key]) + "\n") #使用★ 区分是为了防止其中的词语包含分隔符，对我们后续的操作不利

        f_debug_word.close()

    def transform(self, sentence,max_len=None,add_eos=True):
        """
        实现把句子转化为向量
        :param max_len: 限定长度
        :param add_eos: 是否在最后再补上<EOS>结束符
        :return:
        """
        assert self.fited == True, "必须先进行fit操作"

        r = [self.to_index(i) for i in sentence]
        if max_len is not None: #限定长度
            if max_len>len(sentence):
                if add_eos:
                    #添加结束符与填充符达到一定长度
                    r+=[self.EOS]+[self.PAD for _ in range(max_len-len(sentence)-2)]
                else: #添加填充符达到一定长度
                    r += [self.PAD for _ in range(max_len - len(sentence)-1)]
            else:
                if add_eos:
                    r = r[:max_len-2]
                    r += [self.EOS]
                else:
                    r = r[:max_len-1]
        else:
            if add_eos:
                r += [self.EOS]

        return r

    def inverse_transform(self,indices):
        """
        实现从句子向量 转化为 词（文字）
        :param indices: [1,2,3....]
        :return:[word1,word2.....]
        """
        sentence = []
        for i in indices:
            word = self.to_word(i)
            sentence.append(word)
        return sentence

#以下可供第一次运行，下一次就可以注释掉了

#初始
word_sequence = Word2Sequence()
#词语导入
for line in tqdm(open(config.data_path.txt,encoding='utf-8').readlines()):
    word_sequence.fit(line.strip().split())

print("生成词典...")
word_sequence.build_vocab(min_count=None,max_count=None,max_feature=None)
print("词典大小：",len(word_sequence.dict))
pickle.dump(word_sequence,open(config.word_sequence_dict,"wb")) #保存词典

五、数据集加载（dataset.py）

定义一个ChatDataset类，可以逐一取出数据，并且获取数据集大小。

并且定义一个处理数据的方法——将句子中的词转为标号，并且进行填充。这里并不是整份数据集都是一样的样本长度，只要保证一个batch里的样本长度一致就好了（不一致就填充），这样设计的原因见后面的模型原理分析。

#构建数据集
import torch
import pickle
import config
from torch.utils.data import Dataset,DataLoader
from tqdm import tqdm
from word2seq import Word2Sequence

word_sequence = pickle.load(open(config.word_sequence_dict,"rb")) #词典加载

class ChatDataset(Dataset):
    def __init__(self):
        super(ChatDataset,self).__init__()

        #读取内容
        data_path = config.data_path_txt
        self.data_lines = open(data_path,encoding='utf-8').readlines()

    #获取对应索引的问答
    def __getitem__(self, index):
        input = self.data_lines[index].strip().split()[:-1]
        target = self.data_lines[index].strip().split()[1:]
        #为空则默认读取下一条
        if len(input) == 0 or len(target)==0:
            input = self.data_lines[index+1].split()[:-1]
            target = self.data_lines[index+1].split()[1:]
        #此处句子的长度如果大于max_len，那么应该返回max_len
        return input,target,len(input),len(target)

    #获取数据长度
    def __len__(self):
        return len(self.data_lines)
    
# 整理数据————数据集处理方法
def collate_fn(batch):

    # 排序
    batch = sorted(batch,key=lambda x:x[2],reverse=True) #输入长度排序
    input, target, input_length, target_length = zip(*batch)

    max_len = max(input_length[0],target_length[0]) #这里只需要固定每个batch里面的样本长度一致就好，并不需要整个数据集的所有样本长度一致

    # 词变成词向量，并进行padding的操作
    input = torch.LongTensor([word_sequence.transform(i, max_len=max_len, add_eos=False) for i in input])
    target = torch.LongTensor([word_sequence.transform(i, max_len=max_len, add_eos=False) for i in target])

    input_length = torch.LongTensor(input_length)
    target_length = torch.LongTensor(target_length)

    return input, target

print("数据集装载...")
data_loader = DataLoader(dataset=ChatDataset(),batch_size=config.batch_size,shuffle=True,collate_fn=collate_fn,drop_last=True)

'''''
#测试专用（debug）
if __name__ == '__main__':
    for idx, (input, target) in enumerate(data_loader):
        print(idx)
        print(input)
        print(target)
'''''

六、GPT模型搭建（gpt_model.py）

1、原理解析

（1）Transformer与GPT

说到GPT就要提到Transformer啦。GPT是Transformer的Decoder部分。

Transformer的网络结构如下：（图是网上找的，侵权立删）

而GPT则如下：

因为其没有encoder的输出作为另一个输入分支，所以去掉了encoder-decoder的attention机制。

（2）多头注意力机制

A、提出原因

self attention是通过某种运算来直接计算得到句子在编码过程中每个位置上的注意力权重，然后再以权重和的形式来计算得到整个句子的隐含向量表示（self attention提出原因：在深度学习领域，模型往往需要接收和处理大量的数据，然而在特定的某个时刻，往往只有少部分的数据是重要的。这种情况下应该让模型更加关注这些重要数据，这样他就可以在计算能力有限的情况下，将计算资源分配给更重要的任务，同时解决信息超载问题）。

但self attention的缺陷是：模型在对当前位置的信息进行编码时，会过度的将注意力集中于自身的位置，因此提出了通过多头注意力机制来解决这一问题。

注：为了更好发挥并行输入的特点，首先要解决的问题就是要让输入的内容具有一定的位置信息，因此引入位置编码。

B、注意力机制

键值对注意力机制公式如下：

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)