Python初心者に送る「人工知能の作り方」 Part6 後編

前回は「マルコフ連鎖とはなんぞや」に始まり、それをプログラムの設計に落とし込むところまで行きました。今回はいよいよ実装に移ります。

sandmark.hateblo.jp

Python的には『オブジェクトのコピー』『オブジェクトの永続化』を扱っていきます。今回のソースコードはこちら

坊っちゃん、「ぞなもし」と言う

---

まずはインポート

markov.pyを新しく作りましょう。文字通りマルコフ連鎖を用いた学習・生成を行うモジュールで、それなりに複雑な処理を行います。これをdictionary.pyやresponder.pyにまるごと組み込んでしまうと見通しが悪くなってしまうため、別モジュールとして作成し、インポートしてもらうことにします。

markov.pyではどんな機能を使うのでしょうか。インポート文を見ていきましょう。

import os
import sys
from random import choice
from collections import defaultdict
import re
import copy
import dill
import morph

osとsysはテスト用です。辞書ファイルの存在チェックと、プログラム引数の取得に使用します。

choice, defaultdict, re, morphはこれまでにも登場しました。見慣れないのはcopyとdillです。copyに関しては後述します。

dillは「Pythonのデータをファイルに保存したり、読み込む」ためのパッケージで、pip経由でインストールできます。

pip install dill

今回のマルコフモデルは、テキストデータに変換しようとすると結構な労力がかかります。読み込むときも同様です。そこでエンバグしてはつまらないので、Pythonの辞書データをそのままファイルへ保存したいところです。そしてそれを読み込んだときに、そのままデータが扱えたら便利ですよね。

これをオブジェクトの永続化と言います。ノリ的にはハイバネートが近く、「OS起動しっぱなしだけど電源落とせるよ。電源ボタン押すと元の画面に戻るよ」という機能をデータで行うだけです。

同じような組み込みの機能にmarshalとpickleがありますが、今回の辞書をそれらでファイルにダンプするには一工夫必要なので、dillを使って楽をします。

Markovクラス

辞書の構造

まずはクラス定数と__init__メソッドを見ていきます。

class Markov:
    """マルコフ連鎖による文章の学習・生成を行う。

    クラス定数:
    ENDMARK -- 文章の終わりを表す記号
    CHAIN_MAX -- 連鎖を行う最大値
    """
    ENDMARK = '%END%'
    CHAIN_MAX = 30

    def __init__(self):
        """インスタンス変数の初期化。
        self._dic -- マルコフ辞書。 _dic['prefix1']['prefix2'] == ['suffixes']
        self._starts -- 文章が始まる単語の数。 _starts['prefix'] == count
        """
        self._dic = defaultdict(lambda: defaultdict(lambda: []))
        self._starts = defaultdict(lambda: 0)

CHAIN_MAXは文章の生成時に使用され、文中に含まれる単語の最大個数を30に設定しています。ENDMARKが出てくるまで文章を生成し続けると非常に長い文章になる可能性があるためですが、マルコフ連鎖をCHAIN_MAX回で打ち切るというのも、そこで文章が途切れてしまいます。悩ましいところですが、とりあえず30に設定しています。

self._dicの初期化にはdefaultdict関数が使われています。今回のデータは

{
    'あたし': {
        'は': ['おしゃべり'],
        'が': ['好き'],
    },
    'は': {
        'おしゃべり': ['が', 'と'],
    },
    'おしゃべり': {
        'が': ['好き'],
        'と': ['月餅'],
    },
    'が': {
        '好き': ['な', 'な'],
    },
    '好き': {
        'な': ['プログラム', 'の'],
    },
    ...後略...
}

という構造になっています。存在しないキーを指定した場合は空のハッシュを、さらに空のハッシュに存在しないキーを指定した場合は空のリストを返す、と定義しておくことで、キーの存在チェックやリストの作成を行う処理を省略することができます。

この構造に対して、例えば「あたし は」というprefixに対応するsuffixのリストは

self._dic['あたし']['は']

と表すことができます。「が 好き -> な」のように、同じprefixとsuffixの組み合わせが複数回登場することがあれば、suffixを重複して登録します。このため、ランダムに選択するときは重複している単語が選ばれやすくなり、学習した文章の傾向に沿った出力が得られるはずです。

学習処理

def add_sentence(self, parts):
    """形態素解析結果partsを分解し、学習を行う。"""
    # 実装を簡単にするため、3単語以上で構成された文章のみ学習する
    if len(parts) < 3:
        return

    # 呼び出し元の値を変更しないように`copy`する
    parts = copy.copy(parts)

    # prefix1, prefix2 には文章の先頭の2単語が入る
    prefix1, prefix2 = parts.pop(0)[0], parts.pop(0)[0]

    # 文章の開始点を記録する
    # 文章生成時に「どの単語から文章を作るか」の参考にするため
    self.__add_start(prefix1)

    # `prefix`と`suffix`をスライドさせながら`__add_suffix`で学習させる
    # 品詞情報は必要ないため、`_`で使わないことを明示する
    # すべての単語を登録したら、最後にENDMARKを追加する
    for suffix, _ in parts:
        self.__add_suffix(prefix1, prefix2, suffix)
        prefix1, prefix2 = prefix2, suffix
    self.__add_suffix(prefix1, prefix2, Markov.ENDMARK)

def __add_suffix(self, prefix1, prefix2, suffix):
    self._dic[prefix1][prefix2].append(suffix)

def __add_start(self, prefix1):
    self._starts[prefix1] += 1

ヘルパーメソッドである__add_suffixと__add_startはプライベートメソッドになっています。もしself._dicのデフォルト値がなかったら、ここでキーの存在チェックと、必要に応じたデータの作成を行うことになります。

__add_startメソッドは本来文頭に来るprefixのリストを保持すればいいのですが、試験的に「prefixが何回文頭に来たか」をカウントしています。この値がクラス内で使用されることはありませんが、「どんな単語がもっとも文頭に来やすいか」を測る指標になる……かもしれません。

copyについて

copy.copy関数はオブジェクトを複製します。ここではpartsという引数に対して.popメソッドを呼び出していますが、このメソッドは呼び出された側のオブジェクト（parts）の中身を変更してしまいます。Pythonシェルで確認してみましょう。

>>> a = [1,2,3]
>>> b = a
>>> b
[1, 2, 3]
>>> one = a.pop(0)  # aの先頭の要素を取り出し、削除する
>>> one
1
>>> a
[2, 3]
>>> b
[2, 3]  # ！？

aの値を変更しただけのつもりが、bの中身まで変わってしまっています。これはb = aという代入があくまでも「bはaを参照しろ」という意味であるためで、参照先であるaに変更が加われば、それを見ているbを呼び出したときも、当然変わってしまうわけです。

引数として渡されるデータも同様で、「add_sentenceに渡した形態素解析結果を使い回そうとして別のメソッドに渡そう」などと考えていると、add_sentenceの中で.popが呼び出され、他のメソッドに渡すときには中身が減っている、という怖ろしい現象が起こります。

それを防ぐため、add_sentenceがpartsの参照先を変更しないよう、オブジェクトを複製しなければなりません。

>>> import copy
>>> a = [1,2,3]
>>> b = copy.copy(a)
>>> a.pop(0)
1
>>> b
[1, 2, 3]

なぜこんな仕様なのかというと、速度を重視するためです。もし仮にすべてのオブジェクトを複製していたら、メモリ消費量もオーバーヘッドも凄まじいことになります。

例えばUnmoはDictionaryを持っていますが、仮にrandom.txtのサイズが10MBだったら、Dictionaryのメモリ上のサイズは少なくとも10MB以上ということになります。各Responderに渡すときに毎回メモリ上で10MBのオブジェクトが複製されていたら、あっという間にメモリを食いつぶしてしまいます。

生成処理

Markov.generateメソッドはこのようになっています。

def generate(self, keyword):
    """keywordをprefix1とし、そこから始まる文章を生成して返す。"""
    # 辞書が空である場合はNoneを返す
    if not self._dic:
        return None

    # keywordがprefix1として登録されていない場合、_startsからランダムに選択する
    prefix1 = keyword if self._dic[keyword] else choice(list(self._starts.keys()))

    # prefix1をもとにprefix2をランダムに選択する
    prefix2 = choice(list(self._dic[prefix1].keys()))

    # 文章の始めの単語2つをwordsに設定する
    words = [prefix1, prefix2]

    # 最大CHAIN_MAX回のループを回し、単語を選択してwordsを拡張していく
    # ランダムに選択したsuffixがENDMARKであれば終了し、単語であればwordsに追加する
    # その後prefix1, prefix2をスライドさせて始めに戻る
    for _ in range(Markov.CHAIN_MAX):
        suffix = choice(self._dic[prefix1][prefix2])
        if suffix == Markov.ENDMARK:
            break
        words.append(suffix)
        prefix1, prefix2 = prefix2, suffix

    return ''.join(words)

ハッシュに対する.keys()メソッドは、キーのイテレータを返します。イテレータというのは「繰り返し処理をデータにしたもの」とでも言うべきもので、今回詳しくは触れません。

Python2系では.keys()メソッドはリストを返していたのですが、Python3になってからイテレータを返すよう変更されました。このため、イテレータ型をリスト型に変換するようlistを呼び出しています。

value1 if (condition) else value2

という構文はif文を1行で書いたもので、今回のケースでは

if self._dic[keyword]:
    prefix1 = keyword
else:
    prefix1 = choice(list(self._starts.keys()))

と等価です。どちらが読みやすいかは好みですが、個人的にはprefix1をひとつ書くだけでいいので、一行で書くほうを採用しました。

辞書の保存と読み込み

前述した通り、dillを使って行います。Pythonのデータ型を（関数でさえも）そのままファイルに保存できるというのは魅力的ですが、裏を返せばPython以外のプログラムからは扱えません。かといって暗号化されているわけでもなく、ファイルサイズが小さくなるわけでもありません。

もしマルコフ辞書を他のプログラム（メモ帳など）から扱いたいと思ったら、JSONなどの形式にシリアライズすることをお勧めします。

def load(self, filename):
    """ファイルfilenameから辞書データを読み込む。"""
    with open(filename, 'rb') as f:
        self._dic, self._starts = dill.load(f)

def save(self, filename):
    """ファイルfilenameへ辞書データを書き込む。"""
    with open(filename, 'wb') as f:
        dill.dump((self._dic, self._starts), f)

dill.dump, dill.saveにはファイルディスクリプタ（f）を渡します。ファイル名ではないことに注意してください。また、テキストではなくバイナリを読み書きするので、open関数には'rb'または'wb'を指定します。

今回保存したいのは_self._dicと_self._startsという2つのデータなので、一時的にタプルに包んで保存しています。読み出すときは多重代入を使えば簡単です。

テスト

以上でMarkovクラスの定義は終わりです。しかし、テスト用のコードを埋め込んでおきましょう。Unmoから使う前にひとつのスクリプトとしてテストできるようにしておけば、ソースを読んだ人にも使い方がわかって一石二鳥です。

def main():
    markov = Markov()
    sep = r'[。?？!！ 　]+'
    filename = sys.argv[1]
    dicfile = '{}.dat'.format(filename)
    if os.path.exists(dicfile):
        markov.load(dicfile)
    else:
        with open(filename, encoding='utf-8') as f:
            sentences = []
            for line in f:
                sentences.extend(re.split(sep, line.strip()))
        for sentence in sentences:
            if sentence:
                markov.add_sentence(morph.analyze(sentence))
                print('.', end='')
                sys.stdout.flush()
        markov.save(dicfile)
    print('\n')

    while True:
        line = input('> ')
        if not line:
            break
        parts = morph.analyze(line)
        keyword = next((word for word, part in parts if morph.is_keyword(part)), '')
        print(markov.generate(keyword))


if __name__ == '__main__':
    main()

指定されたファイルから文章を読み取ってマルコフ辞書に登録し、ユーザーからの入力に対して辞書を使った文章を生成して応答する、というものです。

if __name__ == '__main__'というイディオムは以前に説明しましたが、コードがスクリプトとして実行された場合のみ実行され、ライブラリとしてインポートされたときは無視されるという挙動をします。

引数はsys.argvから参照します。読み込んだ後はsaveメソッドを呼び出し、textfile.txt.datという名前で保存しています。一度生成してしまえばこのキャッシュをloadするので、学習処理は省略されます。

今回とりあえず学習してもらうのは、吾輩は猫であるに引き続き、青空文庫からダウンロードした「坊っちゃん」のテキストファイルbocchan.txtです。日本語の文章以外のものがあまり含まれておらず、かつ文章のバリエーションを出すために豊富なサンプルが含まれているという意味で、充分なテストが行えるはずです。

$ python markov.py bocchan.txt

大量の.が表示されますが、文章をひとつ学習するたびに.を出力しています。しばらくすると学習が終わって>というプロンプトが表示されるので、坊っちゃんと会話してみましょう。

[f:id:sandmark:20171013001115p:plain]

ぞなもし……？

tqdm

ドットが表示されるのもいかにも「アナログな学習中」って感じでいいのですが、tqdmというモジュールを使うと、モダンな感じのプログレスバーを自動で表示してくれます。使い方は簡単なので紹介しておきます。

pip install tqdm

import tqdm
...中略...
for sentence in tqdm.tqdm(sentences):
    if sentence:
    markov.add_sentence(morph.analyze(sentence))
    # print('.', end='')  # この2つの行が必要なくなる
    # sys.stdout.flush()  #

文章のリストsentencesをtqdm.tqdm()で囲んだだけです。実行してみるとわかりますが、これだけで詳細なプログレスバーが表示されるので非常に便利。詳細はこちらの記事が参考になるかと思います。

次の課題

いよいよmarkov.pyをAI本体に組み込みます。このモジュールだけでもそれなりの会話ができていると言っても過言ではないので、AIの応答の幅もぐっと広がることでしょう。

---