基于Seq2Seq的聊天机器人

class Encoder(nn.Layer):
    def __init__(self,vocab_size,emb_dim,hid_dim,drop_out,n_layers):
        #vocab_size:输入张量的维度，即字典的大小
        #emb_dim:嵌入层的维度
        #hid_dim:隐藏状态与单元状态的维度
        #n_layers:RNN的层数
        #dropout:丢弃的概率，防止过拟合
        super(Encoder, self).__init__()
        self.hid_dim=hid_dim
        self.n_layers=n_layers
        
        self.emb=nn.Embedding(vocab_size,emb_dim)        #[batch_size,vocab_size,hid_dim]
        self.lstm=nn.LSTM(emb_dim,hid_dim,n_layers)
        self.drop=nn.Dropout(drop_out)    def forward(self,x):
        #src:[batch_size,sen_len]
        x=self.drop(self.emb(x))        #x:[batch_size,sen_len,emb_dim]
        y,(h,c)=self.lstm(x)        #y:[batch size,sen_len,hid dim*n_directions]
        #h:[n layers*n_directions,batch_size,hid_dim]
        #c:[n layers*n_directions,batch size,hid_dim]
        return h,c

!pip install jieba
!pip install --upgrade pip

import jiebaimport numpy as npimport re#将无效字符去掉with open("datasets/one.txt","r",encoding="utf-8") as f:# with open("data/data86810/human_chat.txt","r",encoding="utf-8") as f:
    data=f.read().replace("Human 1"," ").replace("Human 2"," ").replace("."," ").replace("*"," ").replace("@"," ").replace("^"," ").replace("&"," ").replace("!"," ").replace("#"," ").replace("$"," ").replace("?"," ").replace(";"," ").replace(":"," ").replace(","," ").replace('"',' ').replace("%"," ").replace("/"," ").replace("@"," ").replace("("," ").replace(")"," ").replace("'"," ").lower()
data=list(data.split("\n"))#print(len(data))lst=[]#分割出单词，连成序列for obj in data:    # sen=list(obj.split(" "))
    sen=list(jieba.cut(obj, cut_all=False))

    lst.append(sen)

# 分词结果lst

#将字符连接起来，制作字典zh_data = [' '.join(i) for i in lst]
string=" ".join(zh_data)#将特殊字符添入string1=" pad sos eos"#合并字符串string=string+string1#string=string.replace(''," ")#使用正则分割，防止有多个空格words=re.split(" +",string)#使用集合，防止单词重复words=list(set(words))
DIM = len(words)print(len(words))#获取字典dic=dict([(word,i) for i,word in enumerate(words)])

dic

#存储对话序列index_data=[]#每句话的长度，短句添加"pad",长句切至10sen_len=10for i,sen in enumerate(lst):    #token映射至index，并防止出现空字符
    sen=[dic[word] for word in sen if word!='' and word!=' ']    #在开头添加"sos"
    sen.insert(0,dic["sos"])    while len(sen)<sen_len-1:        #填充"pad",防止长度不够
        sen.append(dic["pad"])    #切取sen_len-1个词
    sen=sen[:sen_len-1]    #末尾添加"eos"
    sen.append(dic["eos"])    #将ask与answer分割
    if i%2==0:
        one=[]
        one.append(sen)    else:
        one.append(sen)
        index_data.append(one)#print(len(index_data))index_data=np.array(index_data)print(index_data.shape)print(index_data[0])

#挑一个看看效果ask,ans=index_data[3]#将index序列转化为字符串ask_str=[words[i] for i in ask]
ans_str=[words[i] for i in ans]print(ask_str)print(ans_str)#print(dic)

import paddlefrom paddle.io import Dataset,DataLoaderimport paddle.nn as nnimport random#batch大小batch_size=128class Mydataset(Dataset):
    def __init__(self,index_data,dic):
        super(Mydataset, self).__init__()
        self.index_data=index_data
        self.dic=dic    def __getitem__(self,index):
        ask_data,ans_data=self.index_data[index]        #ask部分倒序，引入更多短期依赖关系
        ask_data,ans_data=ask_data[:][::-1],ans_data        return ask_data,ans_data    def __len__(self):
        return self.index_data.shape[0]#实例化读取器dataset=Mydataset(index_data,dic)#封装为迭代器dataloader=DataLoader(dataset,batch_size=batch_size,shuffle=True,drop_last=True)#看看效果for _,__ in dataloader():    print(_,__)    # break

class Encoder(nn.Layer):
    def __init__(self,vocab_size,emb_dim,hid_dim,drop_out,n_layers):
        #vocab_size:输入张量的维度，即字典的大小
        #emb_dim:嵌入层的维度
        #hid_dim:隐藏状态与单元状态的维度
        #n_layers:RNN的层数
        #dropout:丢弃的概率，防止过拟合
        super(Encoder, self).__init__()
        self.hid_dim=hid_dim
        self.n_layers=n_layers
        
        self.emb=nn.Embedding(vocab_size,emb_dim)        #[batch_size,vocab_size,hid_dim]
        self.lstm=nn.LSTM(emb_dim,hid_dim,n_layers)
        self.drop=nn.Dropout(drop_out)    def forward(self,x):
        #src:[batch_size,sen_len]
        x=self.drop(self.emb(x))        #x:[batch_size,sen_len,emb_dim]
        y,(h,c)=self.lstm(x)        #y:[batch size,sen_len,hid dim*n_directions]
        #h:[n layers*n_directions,batch_size,hid_dim]
        #c:[n layers*n_directions,batch size,hid_dim]
        return h,c

vocab_size=len(dic)
emb_dim=128hid_dim=256drop_out=0.7n_layers=2#实例化encoderencoder=Encoder(vocab_size,emb_dim,hid_dim,drop_out,n_layers)
x=paddle.randint(0,130,[batch_size,sen_len])
h,c=encoder(x)#看看形状print(h.shape,c.shape)

class Decoder(nn.Layer):
    def __init__(self,vocab_size,emb_dim,hid_dim,drop_out,n_layers):
        super(Decoder, self).__init__()
        self.vocab_size=vocab_size
        self.emb_dim=emb_dim
        self.hid_dim=hid_dim

        self.emb=nn.Embedding(vocab_size,emb_dim)
        self.lstm=nn.LSTM(emb_dim,hid_dim,n_layers)
        self.drop=nn.Dropout(drop_out)
        self.fc=nn.Linear(hid_dim,vocab_size)    def forward(self,x,hidden,cell):
        #x = [batch_size]
        #hidden = [n_layers*n_directions, batch_size, hid_dim]
        #cell = [n_layers*n_directions, batch_size, hid_dim]
        #扩维
        x=paddle.unsqueeze(x,axis=1)        #x=[batch_size,1]
        x=self.drop(self.emb(x))        #x=[batch_size,emb_dim]
        output,(h,c)=self.lstm(x,(hidden,cell))        #output = [batch_size,1, hid_dim * n_directions]
        #hidden = [n_layers * n_directions, batch_size, hid_dim]
        #cell = [n_layers * n_directions, batch_size, hid_dim]
        prediction=self.fc(output.squeeze(1))        #prediction=[batch_size,vocab_size]
        return prediction,h,c

decoder=Decoder(vocab_size,emb_dim,hid_dim,drop_out,n_layers)
x=paddle.randint(0,136,[batch_size])
y,h,c=decoder(x,h,c)print(y.shape)

import randomclass seq2seq(nn.Layer):
    def __init__(self,encoder,decoder):
        super(seq2seq, self).__init__()
        nn.initializer.set_global_initializer(nn.initializer.XavierNormal(),nn.initializer.Constant(0.))
        self.encoder=encoder
        self.decoder=decoder    def forward(self,source,target,teacher_forcing_ratio=0.5):
        #src = [batch_size,src_len]
        #trg = [batch_size,trg_len]
        #teacher_forcing_ratio is probability to use teacher forcing
        target_len=target.shape[1]
        batch_size=target.shape[0]
        outputs=paddle.zeros([target_len,batch_size,decoder.vocab_size])        #outputs=[tar_len,batch_size,vocab_size]

        hidden,cell=self.encoder(source)        #x为第一个词"sos"
        x=target[:,0]        #loop (tar_len-1)次
        for t in range(1,target_len):
            output,hidden,cell=self.decoder(x,hidden,cell)            #保存token的张量
            outputs[t]=output            #判断是否动用teacher_forcing
            flag=random.random()<teacher_forcing_ratio            #目标token
            top1=paddle.argmax(output,axis=1)            #x为下一个输入token
            x=target[:,t] if flag else top1        return outputs

x=paddle.randint(0,136,[20,sen_len])
y=paddle.randint(0,136,[20,sen_len])
model=seq2seq(encoder,decoder)
predict=model(x,y)print(predict.shape)

#截断梯度@paddle.no_grad()def init_weights(m):
    for name, param in m.named_parameters():        #均匀分布初始化
        param.data=paddle.uniform(min=-0.2,max=0.2,shape=param.shape)#模型初始化model.apply(init_weights)

def check(str_lst):
    index_set=set(str_lst)    #筛掉重复的单词
    lst=list(index_set)    #重复次数
    zeros=[0 for index in lst]    #组合为字典
    index_dic=dict(zip(lst,zeros))
    index_list=[]    #找出重复的index地方
    for i in range(len(str_lst)):
        index=str_lst[i]        if index in index_set:
            index_dic[index]+=1
            if index_dic[index]>1:
                index_list.append(i)    #删除重复处
    str_lst=np.delete(str_lst,index_list)
    str_lst=paddle.to_tensor(str_lst,dtype="int64")    return str_lst
arr=np.array([1,2,3,4,1,1,2,2])print(check(arr))

#测试函数def evaluate(model,ask_sen=ask):
    ask_sen=paddle.to_tensor(ask_sen).unsqueeze(axis=0)    #tar设为全零张量
    tar=paddle.zeros([1,sen_len])    #第一个token设为"sos"
    tar[0,0]=dic["sos"]
    tar=tar.astype("int64")    #获取answer
    ans=model(ask_sen,tar,0)    #压扁batch_size层
    ans=ans.squeeze(axis=1)    #获取概率最大的token
    ans=ans.argmax(axis=1)
    ans=check(ans.numpy())    #获取字符串
    ans_str=[words[i] for i in ans]    #连接字符串
    string=" ".join(ans_str)    return stringprint(evaluate(model,ask))

learning_rate=2e-4epoch_num=1000#梯度裁剪，防止LSTM梯度爆炸clip_grad=nn.ClipGradByNorm(1)#设定loss，并忽略"pad"这个tokenloss=nn.CrossEntropyLoss(ignore_index=dic["pad"])#实例化优化器optimize=paddle.optimizer.Momentum(learning_rate,parameters=model.parameters(),grad_clip=clip_grad)
model.train()for epoch in range(epoch_num):    for i,(user_data,assist_data) in enumerate(dataloader()):        #清除梯度
        optimize.clear_grad()        #获取预测张量
        predict=model(user_data,assist_data,0)        #将predict展开，并去掉第一个词
        predict=paddle.reshape(predict[1:],[-1,vocab_size])        #将assist_data展开，并去掉第一个词
        assist_data=paddle.reshape(assist_data[:,1:],[-1])        #原predict=[0,y_hat1,y_hat2...]
        #原assist_data=["sos",y1,y2...]
        #因此要将第一个词去掉
        predict=paddle.to_tensor(predict,dtype="float32")
        str_predict=predict.argmax(axis=1)
        str_del=check(str_predict.numpy())        #print("predict:",str_predict)
        #print("del:",str_del)
        num=str_predict.shape[0]-str_del.shape[0]
        assist_data=paddle.to_tensor(assist_data,dtype="int64")        #获取损失值
        avg_loss=loss(predict,assist_data)        #print(avg_loss.numpy(),num)
        avg_loss+=num        #反向传播求梯度
        avg_loss.backward()        #优化参数
        optimize.minimize(avg_loss)        #清除梯度
        optimize.clear_grad()        if i%10==0:            print("epoch:%d,i:%d,loss:%f"%(epoch,i,avg_loss.numpy()))            print(evaluate(model,index_data[random.randint(0,500)][0]))    if epoch%10==0:        #保存模型参数
        paddle.save(model.state_dict(),"work/zh/seq2seq_1.pdparams")

encoder=Encoder(vocab_size,emb_dim,hid_dim,drop_out,n_layers)
decoder=Decoder(vocab_size,emb_dim,hid_dim,drop_out,n_layers)
model=seq2seq(encoder,decoder)
state_dict=paddle.load("work/zh/seq2seq.pdparams")
model.load_dict(state_dict)

def transform(index_tensor):
    string=[words[i] for i in index_tensor]    return " ".join(string)

print("human 1:",transform(index_data[10][0]))print("human 2",evaluate(model,index_data[10][1]))

transform(index_data[10][0])