How to Implement a Chatbot with Python and Tensorflow

 * 训练模型
 * 测试模型

   * 基于检索的对话系统应用 
 * 构建消息查询函数
 * 启动聊天窗口

              + [获取用户信息](#subsubsubsubsection_1)

              + [获取候选回复](#subsubsubsubsection_2)
              + [生成回复语句](#subsubsubsubsection_3)
                
                           
                           
            代码解读

    import tensorflow as tf
    from sklearn.model_selection import train_test_split
    from sklearn.preprocessing import LabelEncoder
    from collections import defaultdict
    from nltk.tokenize import word_tokenize
    from gensim.models.keyedvectors import KeyedVectors
    import jieba
    import re
    
      
      
      
      
      
      
      
    
    代码解读

    # load little bird corpus dataset
    with open('little_bird_corpus.txt', 'r', encoding='utf8') as f:
    data = f.readlines()
    data = list(map(lambda x: x[:-1], data)) # remove '\n' character
    labels = ['小黄鸡故事'+str(i+1) for i in range(len(data))]
    
      
      
      
      
    
    代码解读

    # load qq chat corpus dataset
    with open('qq_chat_corpus.txt', 'r', encoding='utf8') as f:
    data += f.readlines()
    data = list(set(data)) # remove duplicated text
    data = list(filter(None, data)) # remove empty strings
    labels += ['腾讯'+str(i+1) for i in range(len(data)-len(labels))]
    
      
      
      
      
      
    
    代码解读

    train_data, test_data, train_label, test_label = train_test_split(data, labels, test_size=0.2, random_state=42)
    train_data, val_data, train_label, val_label = train_test_split(train_data, train_label, test_size=0.25, random_state=42)
    
      
    
    代码解读

    params = {
    'embed_dim': 300, # embedding dimensionality
       'max_seq_length': 10, # maximum sequence length
    'lstm_units': 64, # number of LSTM units
    'learning_rate': 0.001, # learning rate
    'num_epochs': 10, # number of epochs
    'batch_size': 64, # batch size
       'margin': 0.2, # margin parameter used in loss function
    'loss_type': 'cross-entropy', # loss type (can be cross-entropy or contrastive)
    }
    
      
      
      
      
      
      
      
      
      
    
    代码解读

    def preprocess_text(text):
    """ Preprocess input text by tokenizing, converting to lowercase and removing punctuations."""
    tokens = word_tokenize(text)
    table = str.maketrans('', '', string.punctuation)
    stripped = [w.translate(table).lower().strip() for w in tokens]
    return stripped 
    
    def encode_sequence(tokenizer, max_seq_length, sentences):
    """ Encode the given sentence into padded sequences."""
    seqs = tokenizer.texts_to_sequences(sentences)
    seqs = tf.keras.preprocessing.sequence.pad_sequences(seqs, padding='post', maxlen=max_seq_length)
    return seqs
    
    def create_embedding_matrix(filepath, words):
    """ Load pre-trained word embeddings and add them to an embedding matrix."""
    embedding_dict = {}
    with open(filepath, 'r', encoding='utf8') as f:
        next(f)
        for line in f:
            values = line.split()
            word = values[0]
            vector = np.asarray(values[1:], dtype='float32')
            if word in words:
                embedding_dict[word] = vector
    
    num_words = len(words) + 1
    embedding_matrix = np.zeros((num_words, params['embed_dim']))
    for word, i in tokenizer.word_index.items():
        embedding_vector = embedding_dict.get(word)
        if embedding_vector is not None:
            # words not found in embedding index will be all-zeros.
            embedding_matrix[i] = embedding_vector
    
    return embedding_matrix
    
    def define_model(embedding_matrix):
    """ Define model architecture using TensorFlow layers."""
    inputs = tf.keras.layers.Input(shape=(params['max_seq_length'], ))
    embedding_layer = tf.keras.layers.Embedding(input_dim=embedding_matrix.shape[0], output_dim=embedding_matrix.shape[1], weights=[embedding_matrix])(inputs)
    lstm_out = tf.keras.layers.Bidirectional(tf.keras.layers.LSTM(params['lstm_units']//2, dropout=0.2, recurrent_dropout=0.2))(embedding_layer)
    dense_out = tf.keras.layers.Dense(params['lstm_units'], activation='relu')(lstm_out)
    outputs = tf.keras.layers.Dense(1)(dense_out)
    model = tf.keras.Model(inputs=inputs, outputs=outputs)
    optimizer = tf.keras.optimizers.Adam(lr=params['learning_rate'])
    model.compile(optimizer=optimizer, loss=params['loss_type'], metrics=['accuracy'])
    return model
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    def map_text_to_int(vocab):
    """ Create mapping between words and their corresponding integers."""
    tokenizer = tf.keras.preprocessing.text.Tokenizer(filters='', split=' ')
    tokenizer.fit_on_texts([vocab])
    vocab_size = len(tokenizer.word_index) + 1
    return tokenizer, vocab_size
    
      
      
      
      
      
    
    代码解读

    def train_retriever(train_data, val_data, test_data, train_label, val_label, test_label):
    """ Train a retrieval model on the given training set."""
    tokenizer, vocab_size = map_text_to_int('\n'.join(train_data)+'\n'+'\n'.join(val_data))
    encoded_train_data = encode_sequence(tokenizer, params['max_seq_length'], train_data)
    encoded_val_data = encode_sequence(tokenizer, params['max_seq_length'], val_data)
    encoded_test_data = encode_sequence(tokenizer, params['max_seq_length'], test_data)
    label_encoder = LabelEncoder()
    y_train = label_encoder.fit_transform(train_label)
    y_val = label_encoder.transform(val_label)
    y_test = label_encoder.transform(test_label)
    
    embedding_matrix = create_embedding_matrix('sgns.sogou.char', tokenizer.word_index)
    print("Building model...")
    model = define_model(embedding_matrix)
    print("Model built successfully!")
    
    print("Training model...")
    history = model.fit(encoded_train_data, y_train, validation_data=(encoded_val_data, y_val), 
                        epochs=params['num_epochs'], batch_size=params['batch_size'])
    
    print("Evaluating model...")
    scores = model.evaluate(encoded_test_data, y_test, verbose=0)
    print("Test accuracy:", scores[1])
    
    return model, label_encoder, tokenizer
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    model, encoder, tokenizer = train_retriever(train_data, val_data, test_data, train_label, val_label, test_label)
    
    
    代码解读

    print("\nExample queries:")
    for example in ['我有一只小黄鸡', '小黄鸡和狗的爱恨纠葛', '怎么卖掉小黄鸡的屎']:
    query_vec = np.array(encode_sequence(tokenizer, params['max_seq_length'], [example]))
    pred = model.predict(query_vec)[0][0]
    top_indices = (-pred).argsort()[0][:10]
    top_matches = [(encoder.inverse_transform([idx])[0].replace(' ', ''), float(-pred[idx])) for idx in top_indices]
    print('- Query:', example)
    print('- Top matches:')
    for match in top_matches:
        print('--', match[0]+' ({:.2f})'.format(match[1]*100))
    print('')
    
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    def retrieve_docs(msg, k=10):
    """ Retrieve documents that are most relevant to the given message."""
    query_vec = np.array(encode_sequence(tokenizer, params['max_seq_length'], [msg])).astype('float32')
    pred = model.predict(query_vec)[0][0]
    top_indices = (-pred).argsort()[0][:k]
    top_matches = [(encoder.inverse_transform([idx])[0].replace(' ', ''), float(-pred[idx])) for idx in top_indices]
    return sorted(top_matches, key=lambda x: x[1], reverse=True)
    
      
      
      
      
      
      
    
    代码解读

    import tkinter as tk
    from PIL import ImageTk, Image
    from tkinter import scrolledtext
    import os
    
    class ConvoBotGUI(tk.Frame):
    def __init__(self, master=None):
        super().__init__(master)
        self.master = master
        self.pack()
    
        self.create_widgets()
    
    def create_widgets(self):
    
        self.greetings = tk.Label(self, text="Welcome to our Chinese chatbot!", font=('Arial', 20))
        self.greetings.grid(row=0, columnspan=2, pady=10)
    
        self.photo = tk.Label(self, image=self.img)
        self.photo.grid(row=1, rowspan=2, column=0, sticky=tk.W)
    
        self.usrMsg = scrolledtext.ScrolledText(self, width=50, height=5, wrap=tk.WORD)
        self.usrMsg.grid(row=1, column=1, padx=10, pady=10, sticky=tk.E+tk.N+tk.S)
    
        self.showMsg = tk.Label(self, fg="blue", anchor="e")
        self.showMsg.grid(row=3, column=1, padx=10, pady=10, sticky=tk.E+tk.N+tk.S)
    
        self.replyMsg = scrolledtext.ScrolledText(self, width=50, height=10, wrap=tk.WORD)
        self.replyMsg.grid(row=4, column=1, padx=10, pady=10, sticky=tk.E+tk.N+tk.S)
    
        self.sendBtn = tk.Button(self, text="Send", command=self.send_message)
        self.sendBtn.grid(row=5, column=1, pady=10, ipadx=10, ipady=10)
    
    def send_message(self):
        msg = self.usrMsg.get(1.0, "end-1c")
        reply = ''
        results = []
        try:
            results = retrieve_docs(msg, k=10)
            result_text = ''.join(['\n{:d}. {:.2f}% {}\t({})'.format(rank+1, score*100, docname, author)
                                    for rank,(docname,score) in enumerate(results)])
            if not result_text:
                raise ValueError('No matching document found.')
            else:
                reply = 'Here are some related articles:\n{}'.format(result_text)
    
        except Exception as e:
            print(repr(e))
            reply = "I'm sorry I couldn't find any helpful information about your question."
    
        self.usrMsg.delete(1.0, "end")
        self.showMsg.config(text='\nUser says:{}\n'.format(msg)+
                               'Reply from Bot:\n{}\n'.format(reply))
        self.replyMsg.insert(tk.END,'\nMe says:{}\n'.format(reply))
    
    root = tk.Tk()
    app = ConvoBotGUI(master=root)
    app.mainloop()
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    msg = app.usrMsg.get(1.0, "end-1c")
    
    
    代码解读

    reply = ''
    results = []
    try:
    results = retrieve_docs(msg, k=10)
       ...
    except Exception as e:
    print(repr(e))
    reply = "I'm sorry I couldn't find any helpful information about your question."
    
      
      
      
      
      
      
      
    
    代码解读

    if not result_text:
    raise ValueError('No matching document found.')
    else:
    reply = 'Here are some related articles:\n{}'.format(result_text)
    
      
      
      
    
    代码解读