前回までのあらすじと今回の目的
その1とその2を通して、Python(Flask) でラインボットを作るに至りました。友達登録すると応答メッセージを送ってくれ、メッセージを送るとユーザーネーム付きでオウム返ししてくれるというものです。
が、そんなボットに需要があるわけがありません。これはあくまでテストアプリであり、前回も述べましたが本来作りたいのは「ビートルズの情報検索ツール」です。ということで、テストアプリのコードを本番ボット用のものに書き換えていこうと思います。
作成するボットは、「楽曲名からYouTube動画の検索」「楽曲名から歌詞の取得」などが目的ですが、そのためには受信メッセージから楽曲名を特定できるようにするのが第一段階となります。これには自然言語処理や情報検索 (infomation retrieval) の理論が大きく関わってきます。今回はそこに重点を置いて解説しようと思います。
ちなみにまだまだ現在進行形ですが、作成したボットはこちらになります。ぜひ友達登録して使ってみて下さい。
LINE BOT ID : @711mvjit
曖昧検索
検索する際に、検索ワード(query) とのミスマッチをどの程度まで許容するかは非常に重要です。例えば、 A Hard Day’s Night という曲を検索するときに、query として
a hard day’s nighta hard dayhard days nighthard nightharrd da
のどの文字列でもヒットさせたいと考えるとします（最後は少し無理矢理感ありますが）。
(1) の例は文字列自体は完全一致し、大文字小文字の違いのみです。これは .lower() などで簡単に処理できます。
(2) は部分一致と言われるものです。これは単純に query in ... などで調べられます。
一方の (3) は、部分一致しているように見えてアポストロフィの有無・スペースの数という違いがあるため、 そのままでは in でマッチしません。このため、aharddaysnight のようにスペースや punctuation (コロンやピリオドなどの文字) を削除した文字列と比較することで可能になります。
(4) の場合は、day's がないので部分一致ではありませんが、含まれている単語は全て一致します。ですので、単語を一つ一つチェックして全て含まれているかどうか、という処理をすると良さそうです。
(5) は、ミスタイプがあり単語は全く一致しませんが、含まれている文字は似ています。ですので、文字列の「類似度」を測ることでヒットさせることができそうです。
inverted index
上記の (4) の例に関連する話です。情報検索における重要な概念として、inverted index (転置インデックス、逆引き検索) というものがあります。インデックス(key)を指定して中身(value)をもってくるのとは逆に、「中身の一部（今回の場合は単語）を key とし、その中身を含むインデックスを value とする」というものです。例を挙げると以下のようになります。
文書データ
| document ID | sentence |
|---|---|
| 0 | I love you |
| 1 | You love dog |
| 2 | She hates you |
inverted index
| word | document ID |
|---|---|
| I | [0] |
| love | [0, 1] |
| you | [0, 1, 2] |
| dog | [1] |
| she | [2] |
| hates | [2] |
このような「単語ごとの document ID のリスト」をあらかじめ作成しておくことで、全件検索を行うことなく高速にデータ取得が可能になります。コーパス(大規模言語使用記録)なんかでは当然のように使われる技術で、私が作成したコーパスもこれなしではとてつもない時間がかかってしまいます。
とはいえ、今回のボットの場合はビートルズの曲名や歌詞検索が目的で、たかだか200数十曲しかないので、inverted index を使わずに全曲検索をしてもすぐ終わるでしょうが、せっかく紹介したので実装したいと思います。
レーベンシュタイン距離
編集距離とも呼ばれます。上述の (5) に対応する手法で、2つの文字列がどのくらい類似しているかを「挿入・削除・置換」の回数によって表したものになります。たとえば、
dogs と dog → 削除1回
get と wet → 置換1回
のような感じです。この回数が少ないほど、似ている文字だと言えます。3種類の操作の重みを変えることもあり、例えば置換操作は挿入・削除に対して2倍とする、というのも見かけます。
幸にして Python には Levenshtein というライブラリがありこれを使うことで簡単に実装できます。
ただ、外部ライブラリの管理が面倒 or 無料枠なので少しでも容量節約したい or コードに依存関係を持たせたくない、という方のために、Pandas や numpy さえも一切使わない、私の「車輪の再発明」コードを載せておきますので使って下さい。と言っても、この後のコードで Pandas は使うことになる上、このライブラリの重さなんてたかが知れてるんですけどね...
def levenshtein(str1, str2):
str1 = '^' + str1
str2 = '^' + str2
distances = [list(range(len(str2)))]
for i in range(1, len(str1)):
row = [i]
for j in range(1, len(str2)):
insert = row[j-1] + 1
delete = distances[i-1][j] + 1
replace = distances[i-1][j-1] + int(str1[i]!=str2[j])
row.append(min(insert, delete, replace))
distances.append(row)
return distances[-1][-1]
レーベンシュタイン距離以外にも文字列の類似度を測る方法はいくつかありますので、興味があったら調べてみて下さい。個人的には Jaro-winkler 距離をよく使いますが、これも上述の Levenstein ライブラリに入っています。
参考
https://ja.wikipedia.org/wiki/レーベンシュタイン距離
https://ja.wikipedia.org/wiki/ジャロ・ウィンクラー距離
形態素解析 tokenization
自然言語処理で頻繁に出てくるワードです。要するに、文章を語 (token) ごとに分割する手法です。スペースのない（分かち書きをしない）日本語とは異なり、英語の場合は単純にスペースでスプリットするだけでもそれなりにできます。ピリオド . やハイフン - やアポストロフィ ' などがある場合、さらなる処理が必要になります。
また、活用形や複数形などを、辞書の見出し語の形に直す lemmatization という処理もあります。went → go や、dogs → dog などです。ただ、今回は曲名検索や歌詞検索だけですので、あまり複雑な処理はやらないでおきます。何より自然言語処理系のライブラリ（spacy, nltk など）は重いので…
各手法の優先順位
精度の高い検索結果を得るためには、各手法をどのような順序で組み合わせるかが非常に重要になります。今回は以下の順序でいきます。
- 完全一致・部分一致するか
- 単語の一致数
- 最小のレーベンシュタイン距離
ただ、いくつか注意が必要です。まず部分一致においては、複数該当する可能性があります。例えば "get" で検索した場合、
- Get Back
- When I Get Home
- Getting Better
- Got To Get You Into My Life
- I’ll Get You
などがヒットします。この場合は絞りきれないので、すべてを返すようにするか、ランダムで一つ選ぶなどの工夫が必要です。
単語に関しても同様で、"you love" で検索した場合、
- Love You To
- P.S. I Love You
- All You Need Is Love
などがヒットします。まあこの複数マッチはプログラムの問題というよりは、検索クエリ自体の問題なのでどうしようもないのですが。
レーベンシュタイン距離は使い方がさらに難しくなります。クエリとタイトルをそのまま比較すると、長さの違いによって大きな値が出てしまいます。例えば、検索クエリが "yest" の時、
Yesterday → erday の5字挿入
Help → y s t の3字置換
と、Yesterday よりも Help の方が類似しているという結果になってしまいます。これを解決する方法として、クエリ文字列の長さの窓を用意して substring を前から走査していき、最小のレーベンシュタイン距離を探すという方法があります。たとえば今回の場合、
クエリ : "yest"
窓の長さ : 4
| 開始位置 | substring | レーベンシュタイン距離 |
|---|---|---|
| 0 | yest | 0 |
| 1 | este | 2 |
| 2 | ster | 4 |
| 3 | terd | 3 |
| 4 | erda | 4 |
| 5 | rday | 4 |
のようにチェックしていくということですね。最小レーベンシュタイン距離が0 というのは、部分一致と全く同じことです。ちなみに、同じ長さの文字列同士のレーベンシュタイン距離は、ハミング距離というものと同等です。
実装
まず曲名検索をするための関数 search_song_title(query) を作ります。こちらのレポジトリから、曲名がプライマリキーとなっているファイル beatles.csv をダウンロードして data フォルダに入れておきます。
テーブルファイルを扱う上で Pandas なしでは流石に不便なので、ここでやむなくインストールします。多少は容量を食いますが、これでもまだ十分無料枠内で収まります。
(venv) linebot $ pip install pandas
その上で、「ファイルの読み込み」「曲名のプライマリーキー化」「曲名リストの作成」をするために以下のコードを app.py 内に記載しておきます。
import pandas as pd
DATA = pd.read_csv('data/beatles.csv', index_col='song') ## 曲名をプライマリキーに設定
SONGS = DATA.index ## 曲名のリスト
1. 完全一致・部分一致
これは単に in を使うだけですが、上述したようにスペースや punctuation の問題があります。そこで、「オリジナルの曲名」「スペースや puctuation を取り除いた曲名」の2つと比較して一致するか確認します。完全一致は必ず一つしかないですが、部分一致は複数あり得るので、全曲確認の必要があります。複数マッチした場合、ランダムで一つを選ぶことにします。
また、前後の余計なスペースを取り除いておいたり、全てを小文字化しておきます。スペースが2個以上ある場合などにそれを1個に短縮するために、正規表現を使います。
import re, random
## function to remove whitespaces/punctuations
## e.g. A Hard Day's Night -> aharddaysnight
def remove_punct(text:str) -> str:
return ''.join(c for c in text if c.isalnum()).lower()
def search_song_title(query:str) -> str:
query_lower = re.sub(r'\s\s+', ' ', query).strip().lower() ## shrink space & lowercase
query_no_punct = remove_punct(query_lower) ## a hard day's -> aharddays
partial_match = []
for song in SONGS:
song_lower = song.lower() ## lowercase song title
song_no_punct = remove_punct(song) ## remove punct
### 1. EXACT MATCH ###
if query_lower==song_lower or query_no_punct==song_no_punct:
return song
### 2. PARTIAL MATCH ###
if query_lower in song_lower or query_no_punct in song_no_punct:
partial_match.append(song)
if len(partial_match) > 0:
return random.sample(partial_match, 1)[0]
else:
return None
コードテストのために、notebook を作ってその中にコードをコピーして実行してみましょう（仮想環境でない Python Kernel を選択）。search_song_title('get') としてみると、Get Back や I'll Get You などに混じって、Come Together などもヒットしてしまいます。確かに部分一致はしていますが、単語一致ではないのでこの結果を上位に持ってくるのはあまり良くなさそうです。ということで、単語一致の方も作りましょう。
2. Inverted Index
単語の索引を作る前に、まず単語の分かち書きをする関数 tokenize() を定義します。正規表現で単語境界文字を複数設定し、re.split(r'[\s;:,\-\(\)\"!?]', text) などとすれば、大半の英単語は分かち書きできます。ただし、この場合空文字 '' もできてしまうので、それを除去する作業が必要です。
この正規表現のパターンに . と ' を含めなかったのには理由があります。. は文末だけではなく、Mr. や P.S. のような略語としても使われるからです。少なくとも曲名においては、文末のピリオドとしての . は使われることはないので、単語境界文字としては含めないでおきます。
' については、don't のように正しく入力する人もいれば、dont のように省略する人もいることを考慮しての結果です。また、day's は day だけで一単語と考えることもできそうです。そこで、こういったコードにしてみます。
def tokenize(text:str) -> set:
rough_tokens = re.split(r'[\s;:,\-\(\)\"!?]', text.lower())
tokens = set()
for token in rough_tokens:
if token == '':
continue
if token.endswith("\'s"):
tokens.add(token.replace("\'s", '')) ## day's - > day
if "\'" in token:
tokens.add(token.replace("\'", '')) ## don't -> dont
tokens.add(token)
return tokens
day's は、day's, days, day の3単語として登録されることになります。同じ単語の重複を省くためセット型にし、単語順序は考慮していません。自然言語処理では Bag of Words (BoW) と呼ばれる手法です。この関数を使って inverted index を作成します。以下のコードを追加します。
INVERTED_INDEX_SONGTITLE = {}
for song in SONGS:
for token in tokenize(song):
INVERTED_INDEX_SONGTITLE[token] = INVERTED_INDEX_SONGTITLE.get(token, set()) | {song}
INVERTED_INDEX_SONGTITLE は、{単語 : {その単語を含む曲名のセット}} という辞書になっています。最後の2行は、各曲のタイトルを単語に分割してその単語が辞書にあれば追加、無ければ空セットを用意して追加、という意味です。| は or （和集合）なので、実質的に .add() メソッドと同じ働きをします。
出来上がった INVERTED_INDEX_SONGTITLE はこのような感じです。
{'there': {'Here, There And Everywhere',
'I Saw Her Standing There',
"There's A Place",
'Till There Was You'},
'her': {'And I Love Her',
'Devil In Her Heart',
'Her Majesty',
'I Saw Her Standing There'},
'saw': {'I Saw Her Standing There'},
'standing': {'I Saw Her Standing There'},
'i': {'And I Love Her',
'I Am The Walrus',
'I Call Your Name',
...
}
これを使って、先ほどの search_song_title() 関数を修正してみましょう。
import re, random
## function to tokenize text
def tokenize(text:str) -> set:
rough_tokens = re.split(r'[\s;:,\-\(\)\"!?]', text.lower())
tokens = set()
for token in rough_tokens:
if token == '':
continue
if token.endswith("\'s"):
tokens.add(token.replace("\'s", '')) ## day's - > day
if "\'" in token:
tokens.add(token.replace("\'", '')) ## don't -> dont
tokens.add(token)
return tokens
## function to remove whitespaces/punctuations
## e.g. A Hard Day's Night -> aharddaysnight
def remove_punct(text:str) -> str:
return ''.join(c for c in text if c.isalnum()).lower()
INVERTED_INDEX_SONGTITLE = {}
for song in SONGS:
for token in tokenize(song):
INVERTED_INDEX_SONGTITLE[token] = INVERTED_INDEX_SONGTITLE.get(token, set()) | {song}
def search_song_title(query:str) -> str:
query_lower = re.sub(r'\s\s+', ' ', query).strip().lower() ## shrink space & lowercase
query_no_punct = remove_punct(query_lower) ## a hard day's -> aharddays
query_tokens = query_lower.split(' ')
partial_match = []
for song in SONGS:
song_lower = song.lower() ## lowercase song title
song_no_punct = remove_punct(song) ## remove punct
### 1. EXACT MATCH ###
if query_lower==song_lower or query_no_punct==song_no_punct:
return song
### 2. PARTIAL MATCH ###
if query_lower in song_lower or query_no_punct in song_no_punct:
partial_match.append(song)
### 3. TOKEN MATCH ###
song_intersection = set(SONGS)
for token in query_tokens:
song_intersection &= INVERTED_INDEX_SONGTITLE.get(token, set())
if len(partial_match) == 1:
return partial_match[0]
elif len(song_intersection) > 0:
return random.sample(list(song_intersection), 1)[0]
elif len(partial_match) > 0:
return random.sample(partial_match, 1)[0]
return None
大分長くなってしまいました。アルゴリズムは、
- 検索文字列の小文字化、puctuation 除去、単語分割を行う
- 部分一致した曲名を格納するためのリスト用意
- 全曲名を for ループ。その際に、曲名も小文字化、puctuation 除去する
- もし検索文字列と完全一致した場合、その曲名を返してループ終了
- もし検索文字列と部分一致した場合、曲名リストに格納しておく
- 検索文字列内の単語を for ループし、それぞれの単語を含む曲名のセットの intersection (積集合)を求める。つまり、「検索文字列内の全ての単語を含む曲名」のセットを作る。
- もし部分一致した曲名が1曲だけなら、その曲を返す
- もし「検索文字列内の全ての単語を含む曲名」が存在するなら、その中のうち一つをランダムで返す
- もし「検索文字列内の全ての単語を含む曲名」が存在せず、かつ部分一致が2曲以上あるなら、その中のうち一つをランダムで返す
ということになっています。複数部分一致よりも単語一致が上に来ているので、"get" の時に Come Together がヒットする心配はありません。また、"lovey" だと単語一致はありませんが、スペース除去での部分一致はするので、Love You To か P.S. I Love You の一方がランダムで返されます。
ボット返信メッセージの変更
本来はここにレーベンシュタイン距離も追加したかったのですが、長くなりすぎたので後編に回そうと思います。
とりあえずオウム返しメッセージを削除し、曲名を検索してその YouTube リンクを返す仕様に変更しようと思います。ヒットしない場合は NOT FOUND と返信します。前回の handle_message をこう修正します。
## 返信メッセージ
@handler.add(MessageEvent, message=TextMessageContent)
def handle_message(event):
## APIインスタンス化
with ApiClient(configuration) as api_client:
line_bot_api = MessagingApi(api_client)
## 受信メッセージの中身を取得
received_message = event.message.text
## 曲名取得
songtitle = search_song_title(received_message)
## 返信
if songtitle != None:
reply = 'https://www.youtube.com/watch?v=' + DATA.loc[songtitle, 'official_youtube']
else:
reply = 'NOT FOUND'
line_bot_api.reply_message(ReplyMessageRequest(
replyToken=event.reply_token,
messages=[TextMessage(text=reply)]
))
さらに、仮想環境を有効化して $ pip freeze > requirements.txt を実行し、ライブラリ一覧を更新してから GitHub にプッシュします。Render は新コミットがあると、自動でアプリをデプロイし直してくれます。ログを確認して、デプロイが成功しているか確認しましょう。
このようにボットも正常に動作しています。
それでは次回は、やり残したレーベンシュタイン距離をやっていこうと思います。
app.py 全コード
# -*- coding: utf-8 -*-
from flask import Flask, request, abort
from linebot.v3 import WebhookHandler
from linebot.v3.exceptions import InvalidSignatureError
from linebot.v3.messaging import (
ApiClient, Configuration, MessagingApi,
ReplyMessageRequest, PushMessageRequest,
TextMessage, PostbackAction
)
from linebot.v3.webhooks import (
FollowEvent, MessageEvent, PostbackEvent, TextMessageContent
)
import os, random, re
import pandas as pd
## .env ファイル読み込み
from dotenv import load_dotenv
load_dotenv()
## 環境変数を変数に割り当て
CHANNEL_ACCESS_TOKEN = os.environ["CHANNEL_ACCESS_TOKEN"]
CHANNEL_SECRET = os.environ["CHANNEL_SECRET"]
## データ読み込み
DATA = pd.read_csv('data/beatles.csv', index_col='song') ## 曲名をプライマリキーに設定
SONGS = DATA.index ## 曲名のリスト
## 使用関数もろもろ
def levenshtein(str1, str2):
str1 = '^' + str1
str2 = '^' + str2
distances = [list(range(len(str2)))]
for i in range(1, len(str1)):
row = [i]
for j in range(1, len(str2)):
insert = row[j-1] + 1
delete = distances[i-1][j] + 1
replace = distances[i-1][j-1] + int(str1[i]!=str2[j])
row.append(min(insert, delete, replace))
distances.append(row)
return distances[-1][-1]
## function to tokenize text
def tokenize(text:str) -> set:
rough_tokens = re.split(r'[\s;:,\-\(\)\"!?]', text.lower())
tokens = set()
for token in rough_tokens:
if token == '':
continue
if token.endswith("\'s"):
tokens.add(token.replace("\'s", '')) ## day's - > day
if "\'" in token:
tokens.add(token.replace("\'", '')) ## don't -> dont
tokens.add(token)
return tokens
## function to remove whitespaces/punctuations
## e.g. A Hard Day's Night -> aharddaysnight
def remove_punct(text:str) -> str:
return ''.join(c for c in text if c.isalnum()).lower()
INVERTED_INDEX_SONGTITLE = {}
for song in SONGS:
for token in tokenize(song):
INVERTED_INDEX_SONGTITLE[token] = INVERTED_INDEX_SONGTITLE.get(token, set()) | {song}
def search_song_title(query:str) -> str:
query_lower = re.sub(r'\s\s+', ' ', query).strip().lower() ## shrink space & lowercase
query_no_punct = remove_punct(query_lower) ## a hard day's -> aharddays
query_tokens = query_lower.split(' ')
partial_match = []
for song in SONGS:
song_lower = song.lower() ## lowercase song title
song_no_punct = remove_punct(song) ## remove punct
### 1. EXACT MATCH ###
if query_lower==song_lower or query_no_punct==song_no_punct:
return song
### 2. PARTIAL MATCH ###
if query_lower in song_lower or query_no_punct in song_no_punct:
partial_match.append(song)
### 3. TOKEN MATCH ###
song_intersection = set(SONGS)
for token in query_tokens:
song_intersection &= INVERTED_INDEX_SONGTITLE.get(token, set())
if len(partial_match) == 1:
return partial_match[0]
elif len(song_intersection) > 0:
return random.sample(list(song_intersection), 1)[0]
elif len(partial_match) > 0:
return random.sample(partial_match, 1)[0]
return None
## Flask アプリのインスタンス化
app = Flask(__name__)
## LINE のアクセストークン読み込み
configuration = Configuration(access_token=CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(CHANNEL_SECRET)
## コールバックのおまじない
@app.route("/callback", methods=['POST'])
def callback():
# get X-Line-Signature header value
signature = request.headers['X-Line-Signature']
# get request body as text
body = request.get_data(as_text=True)
app.logger.info("Request body: " + body)
# handle webhook body
try:
handler.handle(body, signature)
except InvalidSignatureError:
app.logger.info("Invalid signature. Please check your channel access token/channel secret.")
abort(400)
return 'OK'
## 友達追加時のメッセージ送信
@handler.add(FollowEvent)
def handle_follow(event):
## APIインスタンス化
with ApiClient(configuration) as api_client:
line_bot_api = MessagingApi(api_client)
## 返信
line_bot_api.reply_message(ReplyMessageRequest(
replyToken=event.reply_token,
messages=[TextMessage(text='Thank You!')]
))
## 返信メッセージ
@handler.add(MessageEvent, message=TextMessageContent)
def handle_message(event):
## APIインスタンス化
with ApiClient(configuration) as api_client:
line_bot_api = MessagingApi(api_client)
## 受信メッセージの中身を取得
received_message = event.message.text
## 曲名取得
songtitle = search_song_title(received_message)
## 返信
if songtitle != None:
reply = 'https://www.youtube.com/watch?v=' + DATA.loc[songtitle, 'official_youtube']
else:
reply = 'NOT FOUND'
line_bot_api.reply_message(ReplyMessageRequest(
replyToken=event.reply_token,
messages=[TextMessage(text=reply)]
))
## 起動確認用ウェブサイトのトップページ
@app.route('/', methods=['GET'])
def toppage():
return 'Hello world!'
## ボット起動時のコード
if __name__ == "__main__":
## ローカルでテストする時のために、`debug=True` にしておく
app.run(host="0.0.0.0", port=8000, debug=True)
目次 : [無料][2024年版] LINE Messaging API v3 + Python(Flask) でボットを作る
GitHub レポジトリ
older_version 内に、各回時点での app.py と requirements.txt があります。