まえがき
お久しぶりです。まぁじです。
今回はマルコフ連鎖で僕のツイートを学習させてbotを作った話をしようかと思います。
技術系のブログじゃなくて「マルコフ連鎖をPython3とherokuを使って作る時に便利なサイトがまとめられるブログ」ぐらいで考えていただけるとありがたいです。
それでは暇を持て余した石油王の皆様よろしくお願いします。
はじめに
僕は@lock_ai0が大好きです。三途の川より大好きです。大好き加減は以下のモーメントを見ていただけるとわかると思います。
twitter.com
その真似事をしてみようかと思い立ち、いろいろ調べてやってみることにしました。
あと、コードは全部ここに上がってます。(いろんなサイトをごちゃまぜにしているので滅茶苦茶な可能性があります)
github.com
0.準備
シリーズ構成
- マルコフ連鎖をやってみる→これ(1/3)
- ツイート取得＆ツイート→https://halss.hatenablog.com/entry/hals_ai2
- 定期的にツイートする→https://halss.hatenablog.com/entry/hals_ai3
やっておくこと
1.マルコフ連鎖をやってみる
マルコフ連鎖とは？
マルコフ連鎖は、未来の挙動が現在の値だけで決定され、過去の挙動と無関係である（マルコフ性）。各時刻において起こる状態変化（遷移または推移）に関して、マルコフ連鎖は遷移確率が過去の状態によらず、現在の状態のみによる系列である。
マルコフ連鎖 - Wikipedia
この場合で言うと、文章生成が文脈によるのではなく選ばれた単語によって次の単語が選ばれる、という流れで文章が生成されます。
例
「私は神です」と「彼の家はピーマンです」があったとき、「私/は/神/です」と「彼/の/家/は/ピーマン/です」という風に分かち書きし、「私・は」「ピーマン・です」のようなチェーンを作って保存します。次に何か単語が与えられたときにその単語が初めに来るチェーンを選び、その終わりの単語が初めに来るチェーンを選び…と終わりが来るまで繰り返していきます。
その結果「私はピーマンです」みたいな文章が出来上がります。これを使います。
プログラム
Python2で書かれたプログラムがあったのでこれを適宜修正しつつ使います。
ブログ→Pythonリハビリのために文章自動生成プログラムを作ってみた - [[ともっくす alloc] init]
GitHub→GitHub - ohshige15/TextGenerator: マルコフ連鎖を使った文章自動生成プログラム
まず空っぽのchain.dbと以下のようなschema.sqlを作成します。これで作ったチェーンをchain.dbに格納する準備ができました。
drop table if exists chain_freqs; create table chain_freqs ( id integer primary key autoincrement not null, prefix1 text not null, prefix2 text not null, suffix text not null, freq integer not null );
次にdata.txtというファイルから文章を読みだし、分かち書き・チェーン作成をしてchain.dbに保存するPrepareChain.pyを作成します。
# -*- coding: utf-8 -*- import unittest import re import MeCab import sqlite3 from collections import defaultdict class PrepareChain(object): """ チェーンを作成してdbに保存するクラス """ BEGIN = u"__BEGIN_SENTENCE__" END = u"__END_SENTENCE__" DB_PATH = "chain.db" DB_SCHEMA_PATH = "schema.sql" def __init__(self, text): """ 初期化メソッド """ self.text=text # 形態素解析用タガー self.tagger = MeCab.Tagger('-Ochasen') def make_triplet_freqs(self): """ 形態素解析から3つ組の出現回数まで """ # 長い文章をセンテンス毎に分割 sentences = self._divide(self.text) # 3つ組の出現回数 triplet_freqs = defaultdict(int) # センテンス毎に3つ組にする for sentence in sentences: # 形態素解析 morphemes = self._morphological_analysis(sentence) # 3つ組をつくる triplets = self._make_triplet(morphemes) # 出現回数を加算 for (triplet, n) in triplets.items(): triplet_freqs[triplet] += n return triplet_freqs def _divide(self, text): """ 「。」や改行などで区切られた長い文章を一文ずつに分ける """ # 改行文字以外の分割文字（正規表現表記） delimiter = u"。|．|\." # 全ての分割文字を改行文字に置換（splitしたときに「。」などの情報を無くさないため） text = re.sub(r"({})".format(delimiter), r"\1\n", text) # 改行文字で分割 sentences = text.splitlines() # 前後の空白文字を削除 sentences = [sentence.strip() for sentence in sentences] return sentences def _morphological_analysis(self, sentence): """ 一文を形態素解析する """ morphemes = [] node = self.tagger.parseToNode(sentence) while node: if node.posid != 0: morpheme = node.surface morphemes.append(morpheme) node = node.next return morphemes def _make_triplet(self, morphemes): """ 形態素解析で分割された配列を、形態素毎に3つ組にしてその出現回数を数える """ # 3つ組をつくれない場合は終える if len(morphemes) < 3: return {} # 出現回数の辞書 triplet_freqs = defaultdict(int) # 繰り返し for i in range(len(morphemes) - 2): triplet = tuple(morphemes[i:i + 3]) triplet_freqs[triplet] += 1 # beginを追加 triplet = (PrepareChain.BEGIN, morphemes[0], morphemes[1]) triplet_freqs[triplet] = 1 # endを追加 triplet = (morphemes[-2], morphemes[-1], PrepareChain.END) triplet_freqs[triplet] = 1 return triplet_freqs def save(self, triplet_freqs, init=False): """ 3つ組毎に出現回数をDBに保存 """ # DBオープン con = sqlite3.connect(PrepareChain.DB_PATH) # 初期化から始める場合 if init: # DBの初期化 with open(PrepareChain.DB_SCHEMA_PATH, "r") as f: schema = f.read() con.executescript(schema) # データ整形 datas = [(triplet[0], triplet[1], triplet[2], freq) for (triplet, freq) in triplet_freqs.items()] # データ挿入 p_statement = u"insert into chain_freqs (prefix1, prefix2, suffix, freq) values (?, ?, ?, ?)" con.executemany(p_statement, datas) # コミットしてクローズ con.commit() con.close() def show(self, triplet_freqs): """ 3つ組毎の出現回数を出力する """ for triplet in triplet_freqs: print("|".join(triplet), "\t", triplet_freqs[triplet]) if __name__ == "__main__": f = open("data.txt",encoding="utf-8") text = f.read() f.close() chain = PrepareChain(text) triplet_freqs = chain.make_triplet_freqs() chain.save(triplet_freqs, True)
最後に、chain.dbからチェーンを呼び出し、文章を生成して出力するGenerateText.pyを作成します。
# -*- coding: utf-8 -*- import os.path import sqlite3 import random from PrepareChain import PrepareChain class GenerateText(object): """ 文章作成用のクラス """ # 短い文を作りたいのでn=1とする def __init__(self, n=1): """ 初期化メソッド """ self.n = n def generate(self): """ 文章生成 """ generated_text = u"" # dbが見つからないときの例外処理 if not os.path.exists(PrepareChain.DB_PATH): raise IOError("DBファイルが存在しません") # dbを開く conn = sqlite3.connect(PrepareChain.DB_PATH) conn.row_factory = sqlite3.Row for i in range(self.n): text = self._generate_sentence(conn) generated_text += text # dbを閉じる conn.close() return generated_text def _generate_sentence(self, conn): """ ランダムに1文生成する """ morphemes = [] # 始まりを取得 first_triplet = self._get_first_triplet(conn) morphemes.append(first_triplet[1]) morphemes.append(first_triplet[2]) # 文章を繋げる while morphemes[-1] != PrepareChain.END: prefix1, prefix2 = morphemes[-2], morphemes[-1] triplet = self._get_triplet(conn, prefix1, prefix2) morphemes += [triplet[2]] # 連結して返す return "".join(morphemes[:-1]) def _get_chain_from_DB(self, conn, prefixes): """ チェーンの情報をDBから取得する """ sql = u"select prefix1, prefix2, suffix, freq from chain_freqs where prefix1 = ?" # prefixが2つなら条件に加える if len(prefixes) == 2: sql += u" and prefix2 = ?" # 結果 result = [] # dbから取得 cursor = conn.execute(sql, prefixes) for row in cursor: result.append(dict(row)) return result def _get_first_triplet(self, conn): """ 文章のはじまりの3つ組をランダムに取得する """ # BEGINをprefix1としてチェーンを取得 prefixes = (PrepareChain.BEGIN,) # チェーン情報を取得 chains = self._get_chain_from_DB(conn, prefixes) # 取得したチェーンから、確率的に1つ選ぶ triplet = self._get_probable_triplet(chains) return (triplet["prefix1"], triplet["prefix2"], triplet["suffix"]) def _get_triplet(self, conn, prefix1, prefix2): """ prefix1とprefix2からsuffixをランダムに取得する """ # BEGINをprefix1としてチェーンを取得 prefixes = (prefix1, prefix2) # チェーン情報を取得 chains = self._get_chain_from_DB(conn, prefixes) # 取得したチェーンから、確率的に1つ選ぶ triplet = self._get_probable_triplet(chains) return (triplet["prefix1"], triplet["prefix2"], triplet["suffix"]) def _get_probable_triplet(self, chains): """ チェーンの配列の中から確率的に1つを返す """ probability = [] # 確率に合うように、インデックスを入れる for (index, chain) in enumerate(chains): for j in range(chain["freq"]): probability.append(index) # ランダムに1つを選ぶ chain_index = random.choice(probability) return chains[chain_index] if __name__ == "__main__": generator = GenerateText() print(generator.generate())
マルコフ連鎖で実際に文章を生成しよう
これで文章から学習して文を生成することができるようになりました。実際にやってみましょう。
青空文庫から「ドグラ・マグラ」をダウンロードし、ルビを削除してdata.txtとして保存します。
https://www.aozora.gr.jp/cards/000096/card2093.html#download
import re f=open("dogura_magura.txt",encoding="utf-8") text=f.read() text=re.sub("［＃\S+］","",text) text=re.sub("《\S+》","",text) f.close() f=open("data.txt",mode="w",encoding="utf-8") f.write(text) f.close()
ここで生成したdata.txtをPrepareChain.pyとGenerateText.pyがあるフォルダに移動し、コマンドプロンプトで以下のように打ちます。
$ python PrepareChain.py $ python GenerateText.py すべて精神病者との間の皺が寄って行かれたらしく又も行衛も知らない人間でない事が君に話したら直ぐに貴女にお相手に見えたにも。
こんな感じでちょっと変な文章が生成されたらOKです。ご覧いただきありがとうございました。
次：ツイート取得＆ツイートhttps://halss.hatenablog.com/entry/hals_ai2
