上一篇 我们安装好了Chatterbot并做了中文适配，然后发现它不太能满足问答型的需求，训练数据中的问句也可能作为回复输出（机器人：你问我，我还想问问你呢）。在Chatterbot的设计中，训练基于对话（Conversation），一个对话通常由一系列相关联的语句组成，比如：
conversation = [
'早上好，二队长。',
'早上好呀。',
'这么早上哪儿去啊？',
'别提了，宿舍马桶又堵了，通马桶去呢。',
'愿马桶精灵保佑你。',
]
训练后，对话中的语句将由上而下两两关联存储在数据库中。只要用户输入的句子能与训练数据匹配，就会给出相关联的回复（绿字是用户输入）：
最后一句话“愿马桶精灵保佑你。”没有关联的语句，会给出相似或随机的回复，学习功能开启时，也可能回复用户说过的话。
在这种场景下它的表现还是比较不错的，但在某些场景，比如我们的机器人就是二队长，她通常是回答的那一方，不可能说出“早上好，二队长”、“愿马桶精灵保佑你”之类的语句；另外我们还希望某些回答只与特定的问题关联，不进入随机回复的选择范围，比如—“今晚的星空格外地璀璨，你觉得呢？”—“就像你一样美丽”。针对这些问题我们需要修改Chatterbot的训练与回复逻辑。
如果对上下文支持要求较高，比如客服机器人，主打任务型对话的RASA可能是更好的选择，或者直接使用一些现成的机器人API。
接下来的修改可能偏离了Chatterbot的设计理念，现在看来也不是特别好，总之能解决上面提到的问题，并且能兼容默认的基于对话的训练。
训练部分修改
对于训练部分，要修改的有两点：
- 标记问句，并且在一组训练数据中，只有第一句为问句，其他句子是这句话的候选回答
- 标记特殊回答，特殊回答不会进入随机回复的选择范围
比如下面的训练数据：
[
'这么早去哪儿啊？',
'别提了，宿舍马桶又堵了，通马桶去呢。',
'去执行任务。-s',
'没事，我溜达。'
]
我们定义第一句为问句，其他语句都是对“这么早去哪儿啊？”的可能回复，同时用“-s”结尾标记特殊回答。
为了标记出问句与特殊回答，直接在数据库中加两个字段是最方便的，可以先看看目前训练后的数据在数据库中的存储格式：
text: 语句
search_text: 语句通过tagger.get_bigram_pair_string处理后的检索文本
conversation: 所属对话
persona: 所属机器人
in_response_to: 关联的语句
search_in_response_to: 关联语句的检索文本
所以可以增加“is_question”字段来区别是否为问句，“is_special”字段来区别是否为特殊回答。
顺带一提，分词逻辑非常完美地理解错了“这么早上哪儿去啊”的意思，但用户输入“去哪儿啊”时依然可以根据文本相似度匹配上。
数据对应的实体类为Statement，位于源码的conversation.py文件中，如果要加字段，可能就不得不修改源码了，这非常不优雅。冷静下来之后想到，增加的字段目前只会在数据存取时用到，所以和实体类Statement没有多大关系，不修改它也是可以的。
Trainer
可以冻手了，先来写自定义的问答型Trainer类，训练逻辑与默认的ListTrainer没有太大区别。在mybot模块下新建trainers.py：
编写类QATrainer继承ListTrainer，照抄并修改train方法：
trainers.py
import re
from chatterbot.conversation import Statement
from chatterbot.trainers import ListTrainer
from chatterbot.utils import print_progress_bar
class QATrainer(ListTrainer):
def train(self, conversation):
# 修改：问句的文本与检索文本
question_statement_text = None
question_statement_search_text = ''
# 修改end
statements_to_create = []
conversation_len = len(conversation)
for conversation_count, text in enumerate(conversation):
if not text.strip():
continue
if self.show_training_progress:
print_progress_bar(
'QA Trainer',
conversation_count + 1, conversation_len
)
# 修改：-s结尾的语句标记为特殊语句，特殊语句不出现在随机回复中
is_special = False
if re.match(r'.*-s$', text):
text = re.sub(r'-s', '', text)
is_special = True
print('Special statement: ' + text)
statement_search_text = self.chatbot.storage.tagger.get_bigram_pair_string(text)
# 如果没有分词结果，是问句则结束并提示错误，是回答则跳过
if not statement_search_text.strip():
if conversation_count == 0:
self.chatbot.logger.error(f'Question has no search text, abort: {text}')
return
else:
self.chatbot.logger.info(f'Answer has no search text, skip: {text}')
continue
# 修改end
# 修改：增加字段，回答与问句相关联
statement = Statement(
text=text,
search_text=statement_search_text,
in_response_to=question_statement_text,
search_in_response_to=question_statement_search_text,
conversation='training')
# statement序列化需要向extra_statement_field_names增加额外字段
statement.extra_statement_field_names.append('is_question')
statement.extra_statement_field_names.append('is_special')
# 额外字段赋值
statement.is_question = conversation_len > 1 and conversation_count == 0
statement.is_special = is_special
# 修改end
statement = self.get_preprocessed_statement(statement)
# 修改：记录问句的文本与检索文本
if conversation_count == 0:
question_statement_text = statement.text
question_statement_search_text = statement_search_text
# 修改end
statements_to_create.append(statement)
self.chatbot.storage.create_many(statements_to_create)
在QATrainer的train方法中，我们完成了问句与特殊回答的标记，将标记作为额外字段添加到了statement中。此时如果调用train方法可以在数据库中看到记录中新增的字段：
StorageAdapter
顺便解决上一篇提到的问题三：重复运行训练代码，数据库中的训练数据会重复添加。在上面的train方法最后，调用了storage的create_many方法来将处理好的训练数据存储到数据库，修改这个create_many，在中间加入对重复数据的判断：
storage_adapter.py
...
class MyMongoDatabaseAdapter(MongoDatabaseAdapter):
...
def create_many(self, statements):
"""
存储多个实体
"""
create_statements = []
for statement in statements:
statement_data = statement.serialize()
tag_data = list(set(statement_data.pop('tags', [])))
statement_data['tags'] = tag_data
if not statement.search_text:
statement_data['search_text'] = self.tagger.get_bigram_pair_string(statement.text)
if not statement.search_in_response_to and statement.in_response_to:
statement_data['search_in_response_to'] = self.tagger.get_bigram_pair_string(statement.in_response_to)
# 修改: 为避免重复，判断是否已经存在，若存在则跳过
exists = self.statements.find_one({
'text': statement_data['text'],
'search_text': statement_data['search_text'],
'conversation': statement_data['conversation'],
'persona': statement_data['persona'],
'in_response_to': statement_data['in_response_to'],
'search_in_response_to': statement_data['search_in_response_to'],
})
if exists:
print(f'Statement already exists, skip: {statement_data["text"]}')
continue
# 修改end
create_statements.append(statement_data)
if len(create_statements) > 0:
self.statements.insert_many(create_statements)
回复部分修改
对于回复部分，我们需要在回复的处理过程中，对训练部分添加的两个额外字段进行判断，将问句与特殊回答排除在随机回复的选择范围外。那么究竟要修改哪里呢？阅读源码后发现，只需要重写StorageAdapter的get_random方法就可以了：
storage_adapter.py
class MyMongoDatabaseAdapter(MongoDatabaseAdapter):
...
def get_random(self):
"""
从数据库中返回随机结果
"""
# 剔除掉is_question为True的元素
# 剔除掉is_special为True的元素
result = self.statements.aggregate([
{'$match': {'is_question': {'$ne': True}, 'is_special': {'$ne': True}}},
{'$sample': {'size': 1}}
])
r = self.mongo_to_object(list(result)[0])
return r
...
这里使用MongoDB的聚合操作，过滤掉is_question与is_special为true的记录，然后随机采样一个结果。
重新训练
至此我们对Chatterbot的扩展修改就已经完成了，接下来调整一下调用的部分。在根目录下新建一个bot.py，用来统一bot的初始化；新建一个train.py，用来训练；原有的main.py用来测试对话效果：
把原来创建bot的部分放到bot.py中：
bot.py
from chatterbot import ChatBot
from chatterbot.response_selection import get_random_response
BOT_NAME = '二队长'
MONGO_URI = 'mongodb://localhost:27017/chatterbot'
def new_bot() -> ChatBot:
return ChatBot(
BOT_NAME,
storage_adapter='mybot.MyMongoDatabaseAdapter',
database_uri=MONGO_URI,
response_selection_method=get_random_response, # 存在匹配时从候选回复中随机选取
read_only=True # 不许学习
)
在train.py编写训练代码，改用QATrainer之后，一次train方法的调用训练一组问答。实际使用时建议从文件读取训练数据，维护起来方便一些。
train.py
from typing import List
from bot import new_bot
from mybot import QATrainer
def train_qa(conversations: List[List[str]]):
bot = new_bot()
trainer = QATrainer(bot)
for conversation in conversations:
trainer.train(conversation)
print('Training finished.')
if __name__ == '__main__':
conversations = [
[
'早上好，二队长。',
'早上好呀。',
],
[
'这么早去哪儿啊？',
'别提了，宿舍马桶又堵了，通马桶去呢。',
'去执行任务。-s',
'没事，我溜达。'
],
[
'愿马桶精灵保佑你。',
'...不用了',
'你说的这个马桶精灵...是不是穿着厚厚的装甲？-s'
],
[
'祝你一路顺利。',
'谢谢。-s',
]
]
train_qa(conversations)
记得清除一下数据库，运行train.py，不出意外的话可以看到如下输出：
第二条75%并不是没有执行完，而是中途输出调试信息把进度更新打断了。训练完成后数据库中的训练结果：
调整main.py：
main.py
from bot import new_bot
# 获取bot
chatbot = new_bot()
# 对话
while True:
try:
user_input = input()
bot_response = chatbot.get_response(user_input)
print(bot_response)
except (KeyboardInterrupt, EOFError, SystemExit):
break
运行main.py，可以测试对话效果了。
运行结果：
其他一些零碎的修改就不过多记录了，总的来说Chatterbot的功能虽然不是很强大，但源码比较容易理解与扩展，从中也能学习到NLP的一些基本流程，还是非常不错的。
网友评论