自从上次弄明白了qq bot机器人的架构之后,就一直很想真正的写一个小机器人,但真正弄的时候发现就算QQ bot可以用Nonebot/Http Coolq通信之后,其实想要写一个可以对话的机器人还是比较困难。

最直接的方法当时是直接用Tuling Api,这个免费版本一天只有100次的家伙。

但除了上限不说,这个最大的问题在于封闭的model提供的API完全没有成长性(尊贵豪华版的可能有?),总之就算你辛辛苦苦教他说话,也没用的。

唯一可行的就是自己从头写一个Model进行训练了,这方面很显然需要涉及到NLP,数据库等等,以及如何和Nonebot集成等等,工作量想象就是很大,一直也懒得去弄(毕竟这就是个娱乐

然后趁着三天放假的时间,花心思看了一下上次看到的那个ChatterBot 库,然后集合Spacy和Jieba的框架,弄了一个可以训练的Model,结合Nonebot,虽然还是看起来很蠢,但总之看到了如果长期训练下去——会稍微聪明一点的AI。

下面简单说一下难点。

  1. ChatterBot 用pip安装的时候,因为我在Vultr上面的主机上内存不够,总是因为spacy 2.1.9的问题挂掉了,上面显示的不是Memory Error 而是Killed

解决方法:

在安装的目录下面吧Spacy>=2.1 <2.2的requirement 删掉,变成Spacy。

这样他就会直接安装Spacy,而Spacy 2.2.3解决了Memory Issue,哪怕memory不够也可以直接安装

然后是用的时候会发现Chatterbot没办法Create,这是因为Spacy的model直接load的是“en”,我们需要下一个2.2版本兼容的en_web_sm model,然后link到en上面

·

pip install https://github.com/explosion/spacy-models/releases/download/en\_core\_web\_sm-2.2.0/en_core_web_sm-2.2.0.tar.gz#egg=en_core_web_sm `
pip -m spacy en\_core\_web\_sm en

然后就可以 create了。(但其实这一步没弄好也无所谓,因为我们用的是Jieba和Spacy内置的Chinese model 不需要这个训练好的en模型)

  1. 总之在解决了安装之后,按照教程走,一个简单的能回答的robot肯定可以实现了。但仔细观察后发现这个robot对于中文的回答非常的蠢,只有一模一样的回答才能给出答案
    类似于 训练“你的名字是什么” 回答 “千夏”。
    然后遇到了“你的名字是”,就完全不知道怎么回答了。
    稍微思考了一下,这应该就是NLP的问题了。观察了一下Chattbot Source code里面的tagger(https://github.com/gunthercox/ChatterBot/blob/master/chatterbot/tagging.py),
    self.nlp = spacy.load(self.language.ISO_639_1.lower())`
    		document = self.nlp(text)
    		
    这两行就是真正进行NLP的地方,对于一句话,这会用来标注每个词的词性,然后去掉Stop word(https://en.wikipedia.org/wiki/Stop_words)后才会进行数据库进行index。
    下一次搜索的时候,不会根据一模一样的原文,单纯是根据这个除去stop word以后剩余词的词性和意思的mapping来搜索。
    所以“I like to eat pasta”
a.storage.tagger.get_text_index_string("I want to eat a paste")
'VERB:eat VERB:paste`
a.storage.tagger.get_text_index_string("I wish to eat a paste")
'VERB:eat VERB:paste'

可以看到对于上面两句话,进过NLP去掉stop word和词性分析后,两句话完全一样,所以这个model对于这两句话的回复也会一模一样,因为search_text完全一样。

所以为什么中文不行就知道了吧?Chatterbot 或者说Spacy完全不知道中文的词性处理和终止词是哪些。

所以想要用Chatterbot进行中文的处理的话,一定要自己下一个Tag,把Jieba里的词性分析和Spacy的match上,然后用ChatterBot就可以了

import string
from chatterbot import languages
from spacy.lang.zh import Chinese
from spacy.tokens import Doc
import jieba.posseg as pseg
import re
from chatterbot import languages

class JiebaChinese(Chinese):
    def __init__(self, pseg):
        super(JiebaChinese, self).__init__()
        self.pseg = pseg
	
	def custom_spacy_pipe(self, vocab, words, flags):
        doc = Doc(vocab, words=words, spaces=[False]*len(words))
        for ix, token in enumerate(doc):
            token.tag_ = flags[ix]
        doc.is_tagged = True
        return doc

	def jieba_tokenize(self, text):
        text = text.strip().replace(' ', '')
        words, flags = zip(*self.pseg.lcut(text))
        return self.custom_spacy_pipe(self.vocab, words, flags)

	def make_doc(self, text):
        return self.jieba_tokenize(text)

def remove_punctuation(text):
  return re.sub("[\s+\.\!\/_,$%^*(+\"\']+|[+——!,。?、~@#¥%……&*():;《)《
                "", text)

class ChineseTagger(object):
  def __init__(self, language):
    self.language = language or languages.ZHS

    self.nlp = JiebaChinese(pseg)
    print("Initialized Chinese Tagger")

  def get_text_index_string(self, text):
    bigram_pairs = []
    if len(text) <= 2:
      text_without_punctuation = remove_punctuation(text)
      if len(text_without_punctuation) >= 2:
        text = text_without_punctuation

	document = self.nlp(text)
    if len(text) <= 2:
      bigram_pairs = [ token.lemma_.lower() for token in document ]
    else:
      tokens = [ token for token in document
                if token.is_alpha and not token.is_stop ]
      if len(tokens) < 2:
        tokens = [ token for token in document if token.is_alpha ]

      if tokens[0].is_alpha and not tokens[0].is_stop:
        bigram_pairs.append('{}:{}'.format(
            tokens[0].tag_, tokens[0].lemma_.lower()))

      for index in range(1, len(tokens)):
        bigram_pairs.append('{}:{}'.format(
              tokens[index - 1].tag_,
              tokens[index].lemma_.lower()
        ))

    if not bigram_pairs:
      bigram_pairs = [token.lemma_.lower() for token in document]
    return ' '.join(bigram_pairs)

这样的结果如下

a.storage.tagger.get_text_index_string("我想吃一个披萨")
'v:想 v:吃 v:披萨'

a.storage.tagger.get_text_index_string("我很想吃一个披萨了")
'v:想 v:吃 v:披萨'

(聪明的孩子可能注意到了,披萨不是v啊,这是因为每个词的意思是和前一个词的词性map,这貌似也是NLP中比较常用的方法貌似)

可以看到,这样的结果就对了,不需要每个字都一模一样,差不多相似也能给出相同的match结果。

  1. 除了上面这两个不大不小的问题之外,其他的就是一些我想要实现的功能方面的架构稍微说一下。

首先,因为完全没有词库肯定刚开始基本上啥话都说不出来,我用Nonebot实现了一个当Chatterbot不能回复,就pass给图灵API,用图灵API回答,然后chatterbot偷学他的答复的策略。

@on_command('tuling')
async def tuling(session: CommandSession):
    message = session.state.get('message')

    reply_chatter = await call_chatter(session, message)
    if reply_chatter:
      print("Get reply from chatter bot for ", message)
      await session.send(escape(reply_chatter))
      return

    reply_tuling = await call_tuling_api(session, message)
    if reply_tuling:
      print("Get reply from tuling for ", message)
      await session.send(escape(reply_tuling))
      return

    else:
      print("Didn't got reply from tuling for ", message)
      reply = EXPR_DONT_UNDERSTAND
      reply.append(message)
      await session.send(render_expression(reply))
      return

效果还挺好的,有时候基本发现不了到底是图灵API回复还是小机器人回复(毕竟小机器人也是偷学图灵的嘛)

稍微需要说的一点,虽然现在图灵API每个机器人上限很低,但我们完全可以多弄几个小机器人,每次random选择一个机器人send requst就好了啊。我弄了5个机器人,也就是500每天了,基本上完全够用了。

另一个需要注意的就是因为机器人天天偷学图灵API的回答,有些回答真的蠢的难以直视。我就忍不住在SQL数据库的Schema上加了一条Boolean作为Verified。我自己添加的语料是Verified,其他所有的都是Not Verified,然后我可以在QQ定期让小机器人列出所有的Not Verified的东西,然后一条条核对需要不需要。

这方面基本上不是很难,就是地方比较多比较麻烦。需要对ChatterBot的code比较熟悉才行。最后的改动主要是在storage/sql_storage.py, ext/sqlalchemy_app/models.py, 以及conversation.py.

变动如下

models.py

verified = Column(
        Boolean,
        nullable=False,
        default=False
    )

conversation.py

statement_field_names = [
        ...
        'verified',
    ]

__slots__ = (
        ...
        'verified',
    )

self.verified = kwargs.get('verified', False)

sql_storage.py

def remove_id(self, statement_id):
        Statement = self.get_model('statement')
        session = self.Session()
        query = session.query(Statement).filter_by(id=statement_id)
        record = query.first()
        session.delete(record)
        self._session_finish(session)

def update(self, statement):
	...
	record.search_text = self.tagger.get_text_index_string(statement.text)
	record.verified = statement.verified
	...

然后在nonebot里用async写一个可以maintain session state的command就行。

最终效果如图

训练过程 (手动训练,也可以自动根据群里大家的聊天偷窥学习)

检查数据库,看从群里偷学的和从图灵偷学的靠谱么

总之,长期下去肯定会变得越来越可爱的> <

总体来说,这个小机器人难度不是很大,但因为有比较强的可成长性,还是蛮好玩的(两天时间也不算白花了hhh

因为有时候总是忍不住在机子上面做乱七八糟的测试,最后总是变得一团糟之后忍不住想要重装。

————

在Vultr上面重装的过程,应该有更简单的方法但我不太清楚……

  1. Destroy Instance,Recreate Instance
  2. 将 IP address 重新Link到Domain Name上 (我是Namesilo 的Managed Domain)
  3. Server Security简单设置
    Disable Root Login
    Change default ssh port
    https://tekeye.uk/vps/change-ssh-port-centos-vps
  4. 重装宝塔
    https://medium.com/@jackme256/wordpress-%E5%BB%BA%E7%AB%99%E6%95%99%E7%A8%8B-%E6%96%B0%E6%89%8B%E6%90%AD%E5%BB%BA-wordpress%E4%B8%AA%E4%BA%BA%E5%8D%9A%E5%AE%A2%E5%9B%BE%E6%96%87%E6%95%99%E7%A8%8B-%E5%AE%8C%E5%85%A8%E7%89%88-62ffd287f756
    yum install -y wget && wget -O install.sh http://download.bt.cn/install/install.sh && sh install.sh
  5. 重装Wordpress
  6. 在网站上重新设置SSL(Let’s Encrypt),否则就不能打开网页了
  7. 用Wordpress 的backup重新Backup WordPress

应该就差不多了吧,不过感觉重新Link domain的时候总要等上几个小时(最多48小时)才能正常使用,这也是DNS正常流程,没办法

总之因为太麻烦觉得以后少折腾是最好,实在想折腾其他的还是开新机子吧。