Python客服机器人-EW帮帮网

1.功能概述

实现一个简单的客服机器人应用，使用Python的Tkinter库构建了图形用户界面(GUI)，并通过与MySQL数据库交互来查询和回复用户的提问。此外，它还支持从CSV或Excel文件中导入话术模板，并提供下载模板的功能。

2.实现逻辑

初始化与GUI设置：通过tkinter库创建了一个窗口应用程序，设置了文本显示区、用户输入区、发送按钮、导入话术按钮和下载模板按钮。
连接到数据库：尝试连接到本地MySQL数据库，并准备执行查询操作。
消息处理：
- 用户输入消息后，首先清理消息内容（移除标点符号），然后根据清理后的消息查询数据库中的匹配项。
- 如果找到一个确切的匹配，则直接返回相应的回答；如果找到多个可能的匹配，则向用户询问确认问题。
- 若无匹配项，则返回默认响应。
话术导入与模板下载：允许用户从外部文件（CSV或Excel）导入新的对话模式到数据库中，并提供下载预设模板的功能以便于用户自定义话术。

3.使用的技术

前端：使用tkinter库构建GUI。
后端：利用mysql.connector与MySQL数据库进行交互，pandas用于处理CSV和Excel文件。
其他：re用于正则表达式处理，random用于随机选择默认响应，chardet用于检测文件编码。

4.主要功能点

简化状态管理：

移除不必要的状态变量 current_pattern 和 current_response。

class ChatbotApp:
    def __init__(self, root):
        ...
        self.patterns_queue = []

处理单条匹配结果：

如果查询结果只有一条，则直接响应。

def query_response(self, msg):
    try:
        query = "SELECT pattern, response FROM patterns WHERE pattern LIKE %s"
        search_pattern = f"%{msg}%"
        print(f"执行查询: {query} WITH {search_pattern}")  # 记录查询语句
        self.cursor.execute(query, (search_pattern,))
        results = self.cursor.fetchall()
        print(f"查询结果: {results}")  # 记录查询结果
        if len(results) == 1:
            pattern, response = results[0]
            return response
        elif len(results) > 1:
            self.patterns_queue = list(results)
            pattern, response = self.patterns_queue.pop(0)
            self.current_pattern = pattern
            self.current_response = response
            return f"您好 {pattern} 吗?"
        else:
            return self.query_default_response(msg)
    except mysql.connector.Error as err:
        messagebox.showerror("错误", f"查询数据库失败: {err}")
        print(f"查询数据库失败: {err}")
        return "发生内部错误，请稍后再试。"

处理多条匹配结果：

如果查询结果有多条，则弹出第一个结果让用户确认。

def query_response(self, msg):
    try:
        query = "SELECT pattern, response FROM patterns WHERE pattern LIKE %s"
        search_pattern = f"%{msg}%"
        print(f"执行查询: {query} WITH {search_pattern}")  # 记录查询语句
        self.cursor.execute(query, (search_pattern,))
        results = self.cursor.fetchall()
        print(f"查询结果: {results}")  # 记录查询结果
        if len(results) == 1:
            pattern, response = results[0]
            return response
        elif len(results) > 1:
            self.patterns_queue = list(results)
            pattern, response = self.patterns_queue.pop(0)
            self.current_pattern = pattern
            self.current_response = response
            return f"您好 {pattern} 吗?"
        else:
            return self.query_default_response(msg)
    except mysql.connector.Error as err:
        messagebox.showerror("错误", f"查询数据库失败: {err}")
        print(f"查询数据库失败: {err}")
        return "发生内部错误，请稍后再试。"

处理用户响应：

根据用户的回复决定是否继续匹配下一条消息。

def send_message(self, event=None):
    user_message = self.user_input.get().strip()
    if user_message.lower() == "退出":
        self.root.quit()
    else:
        cleaned_message = self.remove_punctuation(user_message).lower()
        print(f"清理后的消息: {cleaned_message}")  # 调试信息
        self.display_message(f"你: {user_message}", "blue")
        
        if self.patterns_queue:
            if user_message.lower() in ["是", "yes"]:
                response = self.current_response
                self.display_message(f"机器人: {response}", "green")
                self.patterns_queue.clear()
            elif user_message.lower() in ["否", "no"]:
                if self.patterns_queue:
                    pattern, response = self.patterns_queue.pop(0)
                    self.current_pattern = pattern
                    self.current_response = response
                    self.display_message(f"机器人: 您好 {pattern} 吗?", "green")
                else:
                    default_response = self.query_default_response(cleaned_message)
                    self.display_message(f"机器人: {default_response}", "green")
            else:
                self.display_message(f"机器人: 请输入 是 或 否", "red")
        else:
            response = self.query_response(cleaned_message)
            self.display_message(f"机器人: {response}", "green")

        self.user_input.delete(0, tk.END)

5.运行步骤

确保安装必要的库：
- 安装 mysql-connector-python, pandas, openpyxl, 和 chardet 库。
```
pip install mysql-connector-python pandas openpyxl chardet
```

创建数据库和表：

使用以下SQL语句创建数据库和表：

CREATE DATABASE chatbot_db;
USE chatbot_db;

CREATE TABLE patterns (
    id INT AUTO_INCREMENT PRIMARY KEY,
    pattern VARCHAR(255) NOT NULL,
    response TEXT NOT NULL
);

保存代码：
- 将上述代码保存为一个Python文件（例如chatbot_gui_import_download.py）。
运行代码：
- 在命令行中运行该文件：
```
python chatbot_gui_import_download.py
```
测试默认话术：
- 输入一些默认的话术（例如“你好”，“你是谁？”），确保机器人能够正确回应。
测试模糊检索：
- 输入“今天天气”，确保机器人能够正确回应类似于“今天天气不错！”的回答。
- 输入“今天的天气”，确保机器人能够正确回应类似的回答。
- 输入“今天天气？”或“今天天气.”，确保机器人仍然能够正确回应。

导入话术：

点击“导入话术”按钮，选择包含话术数据的CSV或Excel文件。

导入完成后，在控制台中查看调试信息，确认数据是否正确插入到数据库中。

数据库连接成功
检测到的编码: utf-8 (置信度: 0.99)
话术导入成功
执行查询: SELECT pattern, response FROM patterns WHERE pattern LIKE %s WITH %今天天气%
查询结果: [('今天天气怎么样|今天天气如何|今天的天气|今天是一个好天气吗', '今天天气不错！ 今天的天气还行。 天气非常好，适合外出。')]
匹配到模式: 今天天气怎么样|今天天气如何|今天的天气|今天是一个好天气吗, 响应: 今天天气不错！ 今天的天气还行。 天气非常好，适合外出。

再次测试话术是否能够正确匹配和响应。

下载模板：
- 点击“下载模板”按钮，选择保存路径和文件类型（CSV或Excel）。
- 下载完成后，打开文件检查模板内容是否正确。

6.测试示例

下载模板

点击“下载模板”按钮。
选择保存路径和文件类型（例如CSV）。
打开下载的 patterns_template.csv 文件，确认内容如下：

pattern,response
我叫(.*),"你好 %1，有什么我可以帮忙的吗？"
你好|嗨|您好,"你好！ 嘿，很高兴见到你！ 您好！有什么问题吗？"
你是谁,"我是一个简单的聊天机器人，可以帮助你。 我是你的客服助手。"
你怎么样,"我只是个程序，不过谢谢关心！你呢？ 我很好，谢谢！"
对不起(.*),"没关系！ 别担心。 没事的，继续吧。"
退出,"再见！祝你有美好的一天！ 回头见！ 感谢使用，再见！"
今天天气怎么样|今天天气如何|今天的天气|今天是一个好天气吗,"今天天气不错！ 今天的天气还行。 天气非常好，适合外出。"
订单在哪里,"您的订单正在处理中，稍后会送到。 请您耐心等待，订单很快就会到达。"
退款怎么操作,"您可以通过我们的网站或联系客服进行退款操作。"
产品有问题,"请提供订单号和详细信息，我们会尽快处理。"
发货时间,"预计明天发货，具体时间请关注物流信息。"
售后服务,"您可以联系我们的售后部门获取帮助。"
促销活动,"我们正在进行打折促销，详情请查看我们的官方网站。"
(.),"我不太明白你在说什么，能再说一遍吗？ 抱歉，我没有理解你的意思。"

导入话术

修改或扩展 patterns_template.csv 文件中的内容。

添加一个新的模式和响应，例如：

pattern,response
产品怎么购买,"您可以通过我们的官网或者联系客服了解购买详情。"

确保文件编码为UTF-8（推荐使用UTF-8编码以避免编码问题）。
点击“导入话术”按钮，选择修改后的 patterns_template.csv 文件。

查看控制台输出，确认数据是否正确插入到数据库中。

数据库连接成功
检测到的编码: utf-8 (置信度: 0.99)
话术导入成功

再次测试话术是否能够正确匹配和响应。

输入“产品怎么购买”，预期响应：“您可以通过我们的官网或者联系客服了解购买详情。”。

查看控制台输出：

清理后的消息: 产品怎么购买
执行查询: SELECT pattern, response FROM patterns WHERE pattern LIKE %s WITH %产品怎么购买%
查询结果: [('产品怎么购买', '您可以通过我们的官网或者联系客服了解购买详情。')]
匹配到模式: 产品怎么购买, 响应: 您可以通过我们的官网或者联系客服了解购买详情。
收到的响应: 您可以通过我们的官网或者联系客服了解购买详情。

测试单条匹配结果

输入“今天天气”。

查看控制台输出：

清理后的消息: 今天天气
执行查询: SELECT pattern, response FROM patterns WHERE pattern LIKE %s WITH %今天天气%
查询结果: [('今天天气怎么样|今天天气如何|今天的天气|今天是一个好天气吗', '今天天气不错！ 今天的天气还行。 天气非常好，适合外出。')]
匹配到模式: 今天天气怎么样|今天天气如何|今天的天气|今天是一个好天气吗, 响应: 今天天气不错！ 今天的天气还行。 天气非常好，适合外出。

机器人响应：“今天天气不错！今天的天气还行。天气非常好，适合外出。”

测试多条匹配结果

输入“换货”。

查看控制台输出：

清理后的消息: 换货
执行查询: SELECT pattern, response FROM patterns WHERE pattern LIKE %s WITH %换货%
查询结果: [('能不能退换货', '如果符合我们的退换货政策是可以的。具体政策您可以在网站上查看128221'), ('退换货需要承担运费吗', '根据不同情况而定哦如果是产品质量问题我们承担运费若是其他原因可能需要您承担部分运费128666128176')]
匹配到模式: 能不能退换货, 响应: 如果符合我们的退换货政策是可以的。具体政策您可以在网站上查看128221

机器人询问：“您好能不能退换货吗?”。
用户回复“是”。
机器人响应：“如果符合我们的退换货政策是可以的。具体政策您可以在网站上查看128221”。
如果用户回复“否”：
- 机器人询问：“您好退换货需要承担运费吗吗?”。
- 用户回复“是”。
- 机器人响应：“根据不同情况而定哦如果是产品质量问题我们承担运费若是其他原因可能需要您承担部分运费128666128176”。
如果用户再次回复“否”且没有更多匹配结果：
- 机器人返回默认响应：“抱歉，我不太明白你在说什么，能再说一遍吗？抱歉，我没有理解你的意思。”

7.完整代码

import tkinter as tk
from tkinter import scrolledtext, messagebox, filedialog
import re
import random
import mysql.connector
import csv
import pandas as pd
import chardet

class ChatbotApp:
    def __init__(self, root):
        self.root = root
        self.root.title("客服机器人")
        self.root.geometry("1000x900")

        # 设置主框架样式
        main_frame = tk.Frame(root, bg="#f0f0f0")
        main_frame.pack(fill=tk.BOTH, expand=True)

        # 创建滚动文本框用于显示对话
        self.chat_display = scrolledtext.ScrolledText(main_frame, width=70, height=15, wrap=tk.WORD,
                                                    bg="#ffffff", fg="#000000", font=("Arial", 12),
                                                    borderwidth=2, relief=tk.SUNKEN)
        self.chat_display.grid(row=0, column=0, columnspan=4, padx=10, pady=10, sticky="nsew")
        self.chat_display.config(state=tk.DISABLED)

        # 创建输入框用于用户输入
        self.user_input = tk.Entry(main_frame, width=50, font=("Arial", 12), bg="#ffffff", fg="#000000",
                                  borderwidth=2, relief=tk.SUNKEN)
        self.user_input.grid(row=1, column=0, padx=10, pady=10, sticky="ew")
        self.user_input.bind("<Return>", self.send_message)

        # 创建发送按钮
        self.send_button = tk.Button(main_frame, text="发送", command=self.send_message, font=("Arial", 12),
                                     bg="#4CAF50", fg="#FFFFFF", borderwidth=2, relief=tk.RAISED)
        self.send_button.grid(row=1, column=1, padx=10, pady=10, sticky="ew")

        # 创建导入按钮
        self.import_button = tk.Button(main_frame, text="导入话术", command=self.import_patterns, font=("Arial", 12),
                                       bg="#FF9800", fg="#FFFFFF", borderwidth=2, relief=tk.RAISED)
        self.import_button.grid(row=1, column=2, padx=10, pady=10, sticky="ew")

        # 创建下载模板按钮
        self.download_template_button = tk.Button(main_frame, text="下载模板", command=self.download_template, font=("Arial", 12),
                                                 bg="#008CBA", fg="#FFFFFF", borderwidth=2, relief=tk.RAISED)
        self.download_template_button.grid(row=1, column=3, padx=10, pady=10, sticky="ew")

        # 配置网格布局权重
        main_frame.grid_rowconfigure(0, weight=1)
        main_frame.grid_columnconfigure(0, weight=1)
        main_frame.grid_columnconfigure(1, weight=1)
        main_frame.grid_columnconfigure(2, weight=1)
        main_frame.grid_columnconfigure(3, weight=1)

        # 初始化聊天机器人
        self.connect_to_database()
        self.current_pattern = None
        self.current_response = None
        self.patterns_queue = []

    def connect_to_database(self):
        try:
            self.db_connection = mysql.connector.connect(
                host="xxxxx",
                user="xxxxx",
                password="xxxxx",
                database="xxxxx"
            )
            self.cursor = self.db_connection.cursor()
            print("数据库连接成功")
        except mysql.connector.Error as err:
            messagebox.showerror("错误", f"无法连接到数据库: {err}")
            print(f"无法连接到数据库: {err}")

    def send_message(self, event=None):
        user_message = self.user_input.get().strip()
        if user_message.lower() == "退出":
            self.root.quit()
        else:
            cleaned_message = self.remove_punctuation(user_message).lower()
            print(f"清理后的消息: {cleaned_message}")  # 调试信息
            self.display_message(f"你: {user_message}", "blue")
            
            if self.current_pattern is not None:
                if user_message.lower() in ["是", "yes"]:
                    response = self.current_response
                    self.display_message(f"机器人: {response}", "green")
                    self.current_pattern = None
                    self.current_response = None
                    self.patterns_queue.clear()
                elif user_message.lower() in ["否", "no"]:
                    if self.patterns_queue:
                        pattern, response = self.patterns_queue.pop(0)
                        self.current_pattern = pattern
                        self.current_response = response
                        self.display_message(f"机器人: 您问的是 {pattern} 吗?", "green")
                    else:
                        default_response = self.query_default_response(cleaned_message)
                        self.display_message(f"机器人: {default_response}", "green")
                        self.current_pattern = None
                        self.current_response = None
                else:
                    self.display_message(f"机器人: 请输入 是 或 否", "red")
            else:
                response = self.query_response(cleaned_message)
                self.display_message(f"机器人: {response}", "green")

            self.user_input.delete(0, tk.END)

    def remove_punctuation(self, text):
        return re.sub(r'[^\w\s]', '', text)

    def display_message(self, message, color):
        self.chat_display.config(state=tk.NORMAL)
        self.chat_display.insert(tk.END, message + "\n", color)
        self.chat_display.tag_config(color, foreground=color)
        self.chat_display.yview(tk.END)
        self.chat_display.config(state=tk.DISABLED)

    def query_response(self, msg):
        try:
            query = "SELECT pattern, response FROM patterns WHERE pattern LIKE %s"
            search_pattern = f"%{msg}%"
            print(f"执行查询: {query} WITH {search_pattern}")  # 记录查询语句
            self.cursor.execute(query, (search_pattern,))
            results = self.cursor.fetchall()
            print(f"查询结果: {results}")  # 记录查询结果
            if len(results) == 1:
                pattern, response = results[0]
                return response
            elif len(results) > 1:
                self.patterns_queue = list(results)
                pattern, response = self.patterns_queue.pop(0)
                self.current_pattern = pattern
                self.current_response = response
                return f"您问的是 {pattern} 吗?"
            else:
                return self.query_default_response(msg)
        except mysql.connector.Error as err:
            messagebox.showerror("错误", f"查询数据库失败: {err}")
            print(f"查询数据库失败: {err}")
            return "发生内部错误，请稍后再试。"

    def query_default_response(self, msg):
        try:
            default_query = "SELECT response FROM patterns WHERE pattern LIKE '%(.*)%'"
            print(f"执行默认查询: {default_query}")  # 记录默认查询语句
            self.cursor.execute(default_query)
            default_results = self.cursor.fetchall()
            print(f"默认查询结果: {default_results}")  # 记录默认查询结果
            if default_results:
                default_responses = [row[0] for row in default_results]
                return random.choice(default_responses)
            return "抱歉，我不太明白你在说什么，能再说一遍吗？ 抱歉，我没有理解你的意思。"
        except mysql.connector.Error as err:
            messagebox.showerror("错误", f"查询数据库失败: {err}")
            print(f"查询数据库失败: {err}")
            return "发生内部错误，请稍后再试。"

    def _substitute(self, str, args):
        if "%1" in str and len(args) > 0: str = str.replace("%1", args[0])
        if "%2" in str and len(args) > 1: str = str.replace("%2", args[1])
        if "%3" in str and len(args) > 2: str = str.replace("%3", args[2])
        if "%4" in str and len(args) > 3: str = str.replace("%4", args[3])
        if "%5" in str and len(args) > 4: str = str.replace("%5", args[4])
        return str

    def import_patterns(self):
        file_path = filedialog.askopenfilename(filetypes=[("CSV files", "*.csv"), ("Excel files", "*.xlsx *.xls")])
        if not file_path:
            return

        try:
            encoding = self.detect_file_encoding(file_path)
            if file_path.endswith('.csv'):
                df = pd.read_csv(file_path, encoding=encoding)
            elif file_path.endswith(('.xlsx', '.xls')):
                df = pd.read_excel(file_path, engine='openpyxl')
            else:
                messagebox.showwarning("警告", "不支持的文件类型")
                return

            if 'pattern' not in df.columns or 'response' not in df.columns:
                messagebox.showwarning("警告", "文件缺少必要的列 (pattern 或 response)")
                return

            for index, row in df.iterrows():
                pattern = row['pattern']
                response = row['response']
                query = "INSERT INTO patterns (pattern, response) VALUES (%s, %s)"
                self.cursor.execute(query, (pattern, response))
            self.db_connection.commit()
            messagebox.showinfo("成功", "话术导入成功")
            print("话术导入成功")
        except Exception as e:
            messagebox.showerror("错误", f"导入失败: {e}")
            print(f"导入失败: {e}")

    def detect_file_encoding(self, file_path):
        with open(file_path, 'rb') as f:
            result = chardet.detect(f.read())
            encoding = result['encoding']
            confidence = result['confidence']
            print(f"检测到的编码: {encoding} (置信度: {confidence})")
            return encoding

    def download_template(self):
        file_path = filedialog.asksaveasfilename(defaultextension=".csv",
                                                filetypes=[("CSV files", "*.csv"),
                                                           ("Excel files", "*.xlsx *.xls")],
                                                initialfile="patterns_template.csv")
        if not file_path:
            return

        try:
            data = {
                'pattern': [
                    '我叫(.*)',
                    '你好|嗨|您好',
                    '你是谁',
                    '你怎么样',
                    '对不起(.*)',
                    '退出',
                    '今天天气怎么样|今天天气如何|今天的天气|今天是一个好天气吗',
                    '订单在哪里',
                    '退款怎么操作',
                    '产品有问题',
                    '发货时间',
                    '售后服务',
                    '促销活动',
                    '(.)'
                ],
                'response': [
                    '你好 %1，有什么我可以帮忙的吗？',
                    '你好！ 嘿，很高兴见到你！ 您好！有什么问题吗？',
                    '我是一个简单的聊天机器人，可以帮助你。 我是你的客服助手。',
                    '我只是个程序，不过谢谢关心！你呢？ 我很好，谢谢！',
                    '没关系！ 别担心。 没事的，继续吧。',
                    '再见！祝你有美好的一天！ 回头见！ 感谢使用，再见！',
                    '今天天气不错！ 今天的天气还行。 天气非常好，适合外出。',
                    '您的订单正在处理中，稍后会送到。 请您耐心等待，订单很快就会到达。',
                    '您可以通过我们的网站或联系客服进行退款操作。',
                    '请提供订单号和详细信息，我们会尽快处理。',
                    '预计明天发货，具体时间请关注物流信息。',
                    '您可以联系我们的售后部门获取帮助。',
                    '我们正在进行打折促销，详情请查看我们的官方网站。',
                    '我不太明白你在说什么，能再说一遍吗？ 抱歉，我没有理解你的意思。'
                ]
            }

            if file_path.endswith('.csv'):
                df = pd.DataFrame(data)
                df.to_csv(file_path, index=False, encoding='utf-8-sig')
            elif file_path.endswith(('.xlsx', '.xls')):
                df = pd.DataFrame(data)
                df.to_excel(file_path, index=False, engine='openpyxl')
            else:
                messagebox.showwarning("警告", "不支持的文件类型")
                return

            messagebox.showinfo("成功", "模板下载成功")
            print("模板下载成功")
        except Exception as e:
            messagebox.showerror("错误", f"下载模板失败: {e}")
            print(f"下载模板失败: {e}")

if __name__ == "__main__":
    root = tk.Tk()
    app = ChatbotApp(root)
    root.mainloop()

8.注意修改

这里要修改成自己数据库的信息

try:
            self.db_connection = mysql.connector.connect(
                host="xxxxx",
                user="xxxxx",
                password="xxxxx",
                database="xxxxx"

9.展示效果

在这里插入图片描述

10.运行日志

在这里插入图片描述

Python客服机器人

1.功能概述

2.实现逻辑

3.使用的技术

4.主要功能点

5.运行步骤

6.测试示例

下载模板

导入话术

测试单条匹配结果

测试多条匹配结果

7.完整代码

8.注意修改

9.展示效果

10.运行日志

网站公告

今日签到

热门文章

最新发布