Pythonでマルコフ連鎖を用いた文章生成に挑戦してみた。

Python
2023.04.04

マルコフ連鎖（Markov Chain）とは、確率に基づいて未来の状態を予測するための数学的モデルの一つです。天気予報や自然言語処理、ゲームAIなど多くの分野で活用されています。

本記事では、マルコフ連鎖の基本概念からPythonでの簡単な実装まで、初心者にもわかりやすく解説していきます。

スポンサーリンク
スポンサーリンク

そもそも、マルコフ連鎖とは？

マルコフ連鎖とは、確率過程の一種で、過去の状態に依存して次の状態を決定するような過程のことを指します。具体的には、ある状態にあるシステムが次の状態に遷移する確率が、現在の状態にのみ依存し、それ以前の状態の履歴には依存しないという性質を持ちます。

テキスト生成においては、マルコフ連鎖を用いて、過去の単語の履歴に基づいて、次の単語を予測することができます。たとえば、「私は」という単語が与えられたときに、その次に来る単語が「学生です」や「ご飯が好きです」などのように、ある程度自然な文章を生成することができます。このように、マルコフ連鎖を用いることで、自然言語処理において文章生成や文章修正などのタスクが行われます。

たとえば、ある日の天気が「晴れ」か「雨」かのいずれかで、明日の天気は今日の天気だけに影響されるとしましょう。

このように、履歴を持たず現在の状態だけから次の状態が決まる特性を「マルコフ性」と呼びます。

図：天気のマルコフ連鎖モデル

          +--------+       0.8        +--------+
          |  晴れ  |----------------->|  晴れ  |
          +--------+                 +--------+
              |                         ^
           0.2|                         |
              v                         |0.6
          +--------+       0.4        +--------+
          |  雨    |----------------->|  雨    |
          +--------+                 +--------+

この図は、晴れと雨の状態間の遷移確率を表しています。

  • 晴れから晴れになる確率：0.8
  • 晴れから雨になる確率：0.2
  • 雨から晴れになる確率：0.4
  • 雨から雨になる確率：0.6

マルコフ連鎖の用語解説

状態（State）

システムがとりうるすべての状態です。上記の例では「晴れ」「雨」が状態です。

遷移確率（Transition Probability）

ある状態から別の状態へ移る確率。

遷移行列（Transition Matrix）

全ての遷移確率を表す行列。例えば：

晴れ雨
晴れ0.80.2
雨0.40.6

マルコフ連鎖の利用例

(1) 天気予報

次の日の天気を予測するモデル。

(2) 文章生成

自然言語処理では、前の単語に基づいて次の単語を選ぶモデル。

(3) 経済予測

株価や消費行動などの状態変化モデル。

(4) ボードゲームのAI

チェスや将棋の局面遷移を予測。

Pythonによるマルコフ連鎖の実装（基本編）

以下のコードでは、天気（晴れ・雨）のマルコフ連鎖をPythonでシミュレーションします。

import random

states = ["晴れ", "雨"]
transition_matrix = {
    "晴れ": {"晴れ": 0.8, "雨": 0.2},
    "雨": {"晴れ": 0.4, "雨": 0.6}
}

def next_state(current):
    next_prob = transition_matrix[current]
    rand = random.random()
    cumulative = 0.0
    for state, prob in next_prob.items():
        cumulative += prob
        if rand < cumulative:
            return state

# シミュレーション
current = "晴れ"
sequence = [current]
for _ in range(10):
    current = next_state(current)
    sequence.append(current)

print("→".join(sequence))

出力例：

晴れ→晴れ→晴れ→雨→雨→晴れ→晴れ→晴れ→晴れ→雨→雨

このように、確率に基づいて状態が変化する様子が確認できます。

文章の生成（応用編）

import random
import re

#1---テキストファイルを読み込む
try:
    with open('test.txt', encoding='utf-8') as f:
        text = f.read()
except FileNotFoundError:
    print('テキストファイルが存在しません')
    exit()

#2---テキストを単語に分割する
words = re.findall(r'\w+', text)

#3---2次元辞書を作成する
chain = {}
prev_word = ''
for word in words:
    if prev_word not in chain:
        chain[prev_word] = {}
    if word not in chain[prev_word]:
        chain[prev_word][word] = 0
    chain[prev_word][word] += 1
    prev_word = word

#4---初めの単語を選択する
word1 = ''
while not word1:
    word1 = random.choice(words)

#5---前の単語、その後の出現単語と出現回数の出力
word2 = ''
while not word2:
    try:
        word2 = random.choice(list(chain[word1].keys()))
    except KeyError:
        word1 = ''
        while not word1:
            word1 = random.choice(words)
#6---出力先ファイルの指定
open_file = open("test.txt",mode = "w",encoding = "utf-8")
endflag =0
#7---マルコフ連鎖で文章を生成する
while True:
    try:
        if isinstance(chain[word1][word2], dict):
            weight_sum = sum(chain[word1][word2].values())
            word3 = random.choices(list(chain[word1][word2].keys()), weights=[count/weight_sum for count in chain[word1][word2].values()])[0]
        else:
            word3 = word2
#8---KeyError時の処理
   except KeyError:
        word1 = ''
        while not word1:
            word1 = random.choice(words)
        word2 = ''
        while not word2:
            try:
                word2 = random.choice(list(chain[word1].keys()))
            except KeyError:
                word1 = ''
                while not word1:
                    word1 = random.choice(words)
        continue
    word1, word2 = word2, word3
    endflag = endflag + 1
#9---書込み
    open_file.write(word3)
#10---単語の区切り指定
    if word3.endswith('。') or word3.endswith('？') or word3.endswith('！') or word3.endswith('」'):
        break
#11---終了Flag
    if endflag == 500:
        break
open_file.close()

上記のプログラムは、任意のテキストファイルを読込みんで、マルコフ連鎖で出力した結果をテキストファイルに保存するものです。

以下は解説になります。

#1---テキストファイルを読み込む
try:
    with open('test.txt', encoding='utf-8') as f:
        text = f.read()
except FileNotFoundError:
    print('テキストファイルが存在しません')
    exit()

1の部分では読込むファイルを指定しています。

#2---テキストを単語に分割する
words = re.findall(r'\w+', text)

2の部分ではテキストを単語に分割しています。

#3---2次元辞書を作成する
chain = {}
prev_word = ''
for word in words:
    if prev_word not in chain:
        chain[prev_word] = {}
    if word not in chain[prev_word]:
        chain[prev_word][word] = 0
    chain[prev_word][word] += 1
    prev_word = word

3の部分では2次元辞書を作成しています。分割した単語をそれぞれ格納しています。

#4---初めの単語を選択する
word1 = ''
while not word1:
    word1 = random.choice(words)

4の部分では初めの単語をchoice関数で取り出しています。

#5---前の単語、その後の出現単語と出現回数の出力
word2 = ''
while not word2:
    try:
        word2 = random.choice(list(chain[word1].keys()))
    except KeyError:
        word1 = ''
        while not word1:
            word1 = random.choice(words)
#6---出力先ファイルの指定
open_file = open("test.txt",mode = "w",encoding = "utf-8")
endflag =0

5の部分では辞書のキーは前の単語、値はその後に出現した単語とその出現回数を出力しています。

続けて6の部分では出力先ファイルの指定。「endflag」は後ほど文章生成をしたときに任意の値になったら処理を終了させるFlagです。

#7---マルコフ連鎖で文章を生成する
while True:
    try:
        if isinstance(chain[word1][word2], dict):
            weight_sum = sum(chain[word1][word2].values())
            word3 = random.choices(list(chain[word1][word2].keys()), weights=[count/weight_sum for count in chain[word1][word2].values()])[0]
        else:
            word3 = word2

7の部分ではマルコフ連鎖で文章を生成しています。While構文で条件が満たすまで実行しています。try構文の「isinstance」で辞書型であるかどうかを確認しています。辞書である場合にのみ、次の単語をランダムに選択しています。辞書でない場合は、前の単語 word2を次の単語word3として使用しています。

#8---KeyError時の処理
   except KeyError:
        word1 = ''
        while not word1:
            word1 = random.choice(words)
        word2 = ''
        while not word2:
            try:
                word2 = random.choice(list(chain[word1].keys()))
            except KeyError:
                word1 = ''
                while not word1:
                    word1 = random.choice(words)
        continue
    word1, word2 = word2, word3

8の部分では、文章の生成中に、2つ前の単語が存在しない場合には、初めの2つの単語を再度選択しています。

#9---書込み
    open_file.write(word3)

9の部分ではテキストをファイルに書き込んでいます。

#10---単語の区切り指定
    if word3.endswith('。') or word3.endswith('？') or word3.endswith('！') or word3.endswith('」'):
        break

10の部分では、単語の区切りに記号を指定しています。

#11---終了Flag
    if endflag == 500:
        break
open_file.close()

11の部分ではフラグで条件を満たしたらBreakして抜けて、ファイルを閉じて終了させています。

結果

の内にの内にあきらめの朝まわが歌しばらく東京で無為徒食してわびしさの堪えか日に虫食われゆき何かとでもいったようなものを何か声を失い自信に似たものを得て忍従の夜わかさとなむ何か声を失い陋巷わびしさの堪えかまえから腹案していた長い小説に取りかかったそのおのれの作品に依って知らされ何かとでもいったようなものを歌でなくあきらめの努めか見つけし歌

結果はとある青空文庫の文章を読み込んだものになります。

よくある疑問

Q1
マルコフ連鎖は未来予測に使える？
A

使えますが、あくまで確率に基づく予測です。複雑な環境や長期予測には適していないこともあります。

Q2
マルコフ連鎖の精度を上げるには？
A

状態の定義を細かくしたり、遷移確率を大量のデータから算出することで精度向上が見込めます。

おわりに

マルコフ連鎖は、シンプルながら強力な確率モデルです。初学者にとってもPythonで簡単に扱えるため、実験しながら学ぶには最適です。

今後の発展としては、以下のようなステップが考えられます：

  • 高次のマルコフモデル（例：2-gram, 3-gram）
  • 確率の正規化や平滑化手法
  • マルコフ決定過程（MDP）への拡張

ぜひ、身の回りの変化をマルコフ連鎖で表現してみてください！

【基本情報】基数変換-16進数(小数点)→10進数の分数-
【Windowsバッチファイル入門】Windowsバッチファイルを用いたファイルコピー方法
【初心者向け】Raspberry PiでローカルWebサーバーを構築する方法
バッチファイルでダウンロードしたファイルのブロックをまとめて解除する方法
Pythonでシリアル通信をやってみた-環境構築と使用例-
スポンサーリンク
タイトルとURLをコピーしました