wiki语料库训练检索式聊天机器人

环境：

python3.7~9，chatterbot库

语料库：

中文语料库github.com/brightmart/nlp_chinese_corpus

可以从我的网盘下:

trained_txt: (我训练时踩坑然后去掉每一步遇到的非法json处理后的txt）

https://www.aliyundrive.com/s/GwtwpfH56fj

原始语料库:

https://pan.baidu.com/s/1cOF_uYC1eymOafeG690H4Q?pwd=y79xpan.baidu.com/s/1cOF_uYC1eymOafeG690H4Q?pwd=y79x

用的是wiki

chatterbot

pip install ChatterBot==1.0.8

chatterbot可以从这里下（如果pip安装失败的话）

ChatterBot-1.0.8 https://www.aliyundrive.com/s/9TnnaGEAiFB

https://pan.baidu.com/s/1aZatb1Lj8bL42Lex62YEzw?pwd=l00dpan.baidu.com/s/1aZatb1Lj8bL42Lex62YEzw?pwd=l00d

下载完解压，然后在chatterbot-1.0.8根目录执行

# 1 构建
python setup.py build
# 2 安装
python setup.py install

数据预处理

该语料库的单行是这种结构

{"id": "53", "url": "https://zh.wikipedia.org/wiki?curid=53", "title": "经济学", "text": "经济学\n\n经济学是一门对产品和服务的生产、分配以及消费进行研究的社会科学。西方语言中的“经济学”一词源于古希腊的。\n\n经济学注重的是研究经济行为者在一个经济体系下的行为，以及他们彼此之间的互动。在现代，经济学的教材通常将这门领域的研究分为总体经济学和个体经济学。微观经济学检视一个社会里基本层次的行为，包括个体的行为者（例如个人、公司、买家或卖家）以及与市场的互动。而宏观经济学则分析整个经济体和其议题，包括失业、通货膨胀、经济成长、财政和货币政策等。..."}

而chatterbot需要的数据是一问一答的，为了方便，一问一答放在不同行，所以我们需要提取这个json里的title（问）和text（答）

我一开始是想遍历文件夹（AA~AM）然后发现运行到一半报错(而且好像文件太大没办法一次性搞完），因为它的json里面有一些字符没办法解析（也不知道为啥）然后就一个一个文件夹去遍历，遇到错的就去掉那一行继续遍历……

处理逻辑（可以跳过，直接用我上面分享的trained_txt里的文件训练）

# 文件
import os
# 处理json
import json# wiki_zh下所有子目录名称,用于遍历
# AM目录只有74个文件(wiki_73)
# todo 每次训练完一个文件夹，就把该文件夹名从这里去掉
wiki_c_file_name = ['AA', 'AB', 'AC', 'AD', 'AE', 'AF', 'AG', 'AH', 'AI', 'AJ', 'AK', 'AL', 'AM']
path = './wiki_zh/'  # 目录路径
for c_dir_name in wiki_c_file_name:# 拼接 -> ./wiki_zh/AAdir_name = os.path.join(path, c_dir_name)# ./wiki_zh/AA/*files = os.listdir(dir_name)# print(files)for filename in files:if os.path.isfile(os.path.join(dir_name, filename)):# 控制当前训练的文件到精确位置# 1,训练# 2,到了异常的地方就复制 id# 3,ctrl shift f -> id# 4,ctrl x 该id对应的那行,黏贴到桌面的txt# 5,修改文件名 train_before_${id}# 6,修改判断条件 int(data['id']) < ${id}# 7,修改判断条件 int(filename[(wiki + 1):]) < id所在文件的编号wiki = filename.find("_")if (c_dir_name != 'AA'):print(filename)break# 跳过 上一次报错的位置 之前的数据 todo 每次报错重新训练要更改if (int(filename[(wiki + 1):]) < 0):print(filename[(wiki + 1):])continuewith open(os.path.join(dir_name, filename), 'r', encoding='utf-8') as f:# 读取所有内容content = f.read()content = content.replace("\n", '')content = content.replace("\\n", '')while content.find('{"id":') != -1:# 从 {"id": 开始，到 "} 结束 start = content.find('{"id":')end = content.find('"}')print(content[start:(end + 2)])# start到end这段(json) -> 转为json对象 -> 字典json_str = content[start:(end + 2)]json_str = json_str.replace('\\n', '')json_str = json_str.replace('\n', '')# 加载、解析json数据data = json.loads(json_str)# 从原来的文本里去掉start到end的这段content = content[(end + 2):]print(data['text'])# 跳过 上次报错的json的id 之前的数据 todo 每次报错重新训练要更改if (int(data['id']) <= 0):continue# 打开文件，如果不存在则创建, 把训练完的数据保存到txt文件# todo 每次报错把创建的文件改名 train_before_AA_错误的json的id# todo 每次训练完一个文件夹下的所有文件，需要把AA改成下一个要训练的文件夹（AB、AC...）with open('train_before_AA_.txt', 'a', encoding='utf-8') as file:file.write(data['title'] + '\n')file.write(data['text'] + '\n')file.close()f.close()

训练(批量）

from chatterbot import ChatBot
from chatterbot.trainers import ListTrainer
import ospath = './train_txt/'  # 目录路径
files = os.listdir(path) # 该目录下的子目录s
print(files) # xxx.txt
corpus = []  # 加载的语料,一行问,一行答
for filename in files:if os.path.isfile(os.path.join(path, filename)):print(filename)open_file = os.path.join(path, filename)print(open_file)# 导入语料库file = open(open_file, 'r', encoding='utf-8')print('开始加载语料！')while 1:try:line = file.readline()print(line)if not line:breakcorpus.append(line.strip('\n'))except:passfile.close()
print('语料加载完毕')
my_bot = ChatBot("wiki",database_uri='sqlite:///db.sqlite3')
print('Type something to begin...')
trainer = ListTrainer(my_bot)
print('开始训练！')
# trainer.train(corpus[:10000])
trainer.train(corpus)
print('训练完毕！')

如果批量不行，就单个

trained_txt

from chatterbot import ChatBot
from chatterbot.trainers import ListTrainercorpus = []  # 加载的语料,一行问,一行答
# 导入语料库
# todo 训练完改为下一个文件
file = open('train_txt/train_before_53890.txt', 'r', encoding='utf-8')
print('开始加载语料！')
while 1:try:line = file.readline()print(line.strip('\n'))if not line:breakcorpus.append(line.strip('\n'))except:pass
file.close()
print(corpus)
print('语料加载完毕')
# 使用不同语料+同个数据库训练不会覆盖数据库原有数据
my_bot = ChatBot("wiki", database_uri='sqlite:///db.sqlite3')
print('Type something to begin...')
trainer = ListTrainer(my_bot)
print('开始训练！')
# trainer.train(corpus[:10000])
trainer.train(corpus)
print('训练完毕！')# 测试
# text = input("You: ")
# res = my_bot.get_response(text)
# print(res)

使用flask作为服务器提供服务：

from chatterbot import ChatBot
# flask创建http接口
from flask import Flask, request, jsonify
from flask_cors import CORS# 使用训练好保存了的数据
# 初始模型
my_bot = ChatBot("wiki", database_uri='sqlite:///db.sqlite3')
# flask
app = Flask(__name__)
CORS(app, resources=r'/*')  # 注册CORS, "/*" 允许访问所有api# /api/wiki/q
@app.route("/api/wiki/<q>", methods=['get'])
def wiki_answer_path(q):res = str(my_bot.get_response(q))return jsonify({"res": res}), 200# /api/wiki?q=xxx
@app.route("/api/wiki", methods=['get'])
def wiki_answer_query():# 获取路径参数 ?q=xxxq = request.args.get("q")res = str(my_bot.get_response(q))return jsonify({"res": res}), 200if __name__ == "__main__":app.run(host='0.0.0.0', port=8181)

如果报错需要en，可以从这里下载

链接：https://pan.baidu.com/s/1urcaSqBz3nv2K73naGZV9A?pwd=5ino

补：打包...