最初に

https://twitter.com/shikumie に @付きでメンションするとオススメボカロ曲を教えてくれます。
こちらを試してからの方が下記記事は理解しやすいかもしれません。
開発用にアカウントを使いまわしたかったので非公開にしています。。。(2020-08-08更新)

こんな感じになる筈です

なお、返信までに最大5分程度かかります。
気長にお待ちください。

5分待っても返信が来なかったらとりこぼしていたり、なんらかの理由で落ちている可能性があります。
次の日などに試していただけるともしかしたらうまくいくかもしれません

やったこと（ざっくり）

下記を5分に1回実行するlambdaを作成した
1. メンションがきているかを確認し、リプライ先となるstatus_idを取得
2. 1.で得られた返信先ユーザの直近のツイート(最大5つ)を取得
3. ツイートに含まれる特徴量を抽出し、事前に作っておいた楽曲の特徴量と比較、最も近い物を選択
4. 3で得た楽曲情報を1.で得たstatus_idにリプライ

やったこと(詳細)

0. 事前準備

0.1 Twitter APIの準備

botにツイートしてもらったり、オススメを聞いてきた人のツイートを取得したりするのにAPIを利用します。

自分が開発を行った際は、下記記事を読んで、とりあえずツイートできるところまで進めました。
https://qiita.com/channel-techtok/items/dc5028f667a9dde17092

0.2 lambdaの準備

5分に1回、lambdaが起動するように準備しておきます。
今回はsam-cliを使用しました。

この辺りの準備に関しては、以前記事を書いているので参考にしていただければと思います。
https://qiita.com/miyatsuki/items/c221b48830db2b0a9eba#2-1%E3%81%A7%E4%BD%9C%E3%81%A3%E3%81%9F%E3%82%B9%E3%82%AF%E3%83%AA%E3%83%97%E3%83%88%E3%82%92lambda%E7%B5%8C%E7%94%B1%E3%81%A7%E5%AE%9A%E6%9C%9F%E5%AE%9F%E8%A1%8C%E3%81%95%E3%81%9B%E3%82%8B

1. メンションがきているかを確認し、リプライ先となるstatus_idを取得

メンションは下記のAPIで取得できます。
https://developer.twitter.com/en/docs/tweets/timelines/api-reference/get-statuses-mentions_timeline

メンションの投稿時刻, screen_name, ツイート文, メンションのstatus_id(リプライ先) を取得します。

get_mentions.py

def get_mentions(twitter):
    # ツイート処理
    res = twitter.get("https://api.twitter.com/1.1/statuses/mentions_timeline.json", params={"count": 200})

    # エラー処理
    if res.status_code != HTTPStatus.OK:
        print(f"Failed: {res.status_code}")
        return []

    mentions = []
    for mention in res.json():

        # 会話中の@mentionは無視する
        if mention["in_reply_to_status_id_str"]:
            continue

        unixtime = int(datetime.strptime(mention["created_at"], '%a %b %d %H:%M:%S %z %Y').timestamp())
        mentions.append((unixtime, mention["user"]["screen_name"], mention["text"], mention["id"]))

    return mentions

CONSUMER_KEY        = 'XXX'
CONSUMER_SECRET     = 'XXX'
ACCESS_TOKEN_KEY    = 'XXX'
ACCESS_TOKEN_SECRET = 'XXX'

# twitter操作用クラス
twitter = OAuth1Session(
    CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN_KEY, ACCESS_TOKEN_SECRET
)

#[[unixtime, screen_name, tweet, status_id], ...]
mentions = get_mentions(twitter)

2. 1.で得られた返信先ユーザの直近のツイート(最大5つ)を取得

各ユーザのツイートは下記APIで取得できます。
https://developer.twitter.com/en/docs/tweets/timelines/api-reference/get-statuses-user_timeline

get_new_tweets.py

def get_new_tweets(twitter, screen_name):

    # RTなども取得件数に含まれてしまうので、countは多めに取っておく
    res = twitter.get("https://api.twitter.com/1.1/statuses/user_timeline.json", params={"screen_name": screen_name, "count": 20})

    # エラー処理
    tweets = []
    if not res.status_code == HTTPStatus.OK:
        print(f"Failed: {res.status_code}")
        return []

    for res_i in res.json():
        print(res_i["text"])
        tweet = res_i["text"]

        # メンションは無視する (botへの呼びかけも含んでしまうので)
        if "@" in tweet:
            continue

        tweets.append(tweet)

        # いったん5ツイート集まったら終わりにする
        if len(tweets) >= 5:
            break

    return tweets

CONSUMER_KEY        = 'XXX'
CONSUMER_SECRET     = 'XXX'
ACCESS_TOKEN_KEY    = 'XXX'
ACCESS_TOKEN_SECRET = 'XXX'

# twitter操作用クラス
twitter = OAuth1Session(
    CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN_KEY, ACCESS_TOKEN_SECRET
)

user_tweets_map = {}

# screen_namesはツイートを取得したいscreen_nameを格納したiteratable
for screen_name in screen_names:
    tweets =  get_new_tweets(twitter, screen_name)
    if tweets:
        user_tweets_map[screen_name] = tweets

3. ツイートに含まれる特徴量を抽出し、事前に作っておいた楽曲の特徴量と比較、最も近い物を選択

3.1. 楽曲から特徴量を取得

3.1.0 歌詞データの収集

今回は推薦対象の楽曲の歌詞データから特徴量を取得します。
その前提として、推薦する楽曲の歌詞データを事前に取得しておく必要がありますが、ここはすでにあるものとして話を進めます。
実際にはスクレイピングしてデータを集めていますが、その部分の詳細についてはここでは触れません。

3.1.1 特徴量の計算(学習)

今回は「ドキュメント内の文字を漢字のみに絞った」上で各漢字をwordとみなしてword2vecを計算することで漢字ごとにベクトルを算出、ドキュメント内に含まれる漢字ベクトルの平均値をドキュメントの特徴量としました。

普通に考えると「既存のword2vec」のモデルを流用した方が良さそうな気もするのですが、それができない、異なるレイヤーの問題が存在します

歌詞という文脈における単語の使われ方と既存のニュース記事やwikipediaにおける単語の使われ方が一致していないような気がする
- 例えば「太陽」という単語はニュース記事では天文分野の他の単語（惑星）などが近い気がするが、歌詞の文脈だと比喩表現としてもっと近い単語があるように思える（情熱とか）
そもそも推論フェーズで形態素解析が動かない可能性があるので、形態素ベースの既存モデルが使えない
- lambdaの場合、使用できるライブラリの上限が250MBという制約がある
  - 厳密にはlambdaにアップロードできるファイルサイズの上限
- 例えばpure pythonの形態素解析ライブラリであるjanomeはこれだけで100MB程度を占有する
  - janome: https://mocobeta.github.io/janome/
- さらに、実行時もメモリ量ごとに従量課金額が変動するので可能な限り使用するライブラリを節約しメモリ量を削減したい

ということで、推論時の負荷を可能な限り下げ、かつ、歌詞の文脈における情報も可能な限り持たせる方法ということで漢字のみのword2vecを使用することにしました。

学習時のコードは下記の通りです。
この部分のコードは事前に計算しておく前提なので、numpyやgensimを使っています。

なお、漢字だけを抜き出す部分については下記を参考にしました
https://qiita.com/mocha_xx/items/00c5a968f7069d8e092c

train.py


### 歌詞から漢字の部分だけを抜き出して、学習用データ作成
import re

re_kanji = re.compile(r'^[\u4E00-\u9FD0]$')

corpus = []
title_list = []

### lyrics_mapは lyrics_map["title"] = "lyrics" であるdict
for title, lyric in lyrics_map.items():
    token_list = []

    for line in lyric.split("\n"):
        for token in line:
            if re_kanji.fullmatch(token):
                token_list.append(token)

    if len(token_list) > 20:    
        title_list.append(title)
        corpus.append(token_list)

### gensimのword2vecで漢字ごとのベクトルを計算
from gensim.models import word2vec
model = word2vec.Word2Vec(corpus, size=200, min_count=20, window=10)

### 推論時はgensimのモデルを使えないので、普通のdictにしておく
word_wv_map = {}
print(len(model.wv.index2word))
for word in model.wv.index2word:
    wv = [0.0] * len(model.wv[word])

    for i, val in enumerate(model.wv[word]):
        wv[i] = float(val)

    word_wv_map[word] = wv

### 歌詞全体のベクトルは、歌詞に含まれる漢字ベクトルの平均とする
import numpy as np
import json
import math

title_wv_map = {}

for title in title_list:
    token_count = 0
    wv = np.array([0.0] * len(model.wv[word]))
    wv_list = [0] * len(wv)

    lyric = lyrics_map[title]

    for line in lyric.split("\n"):
        for token in line:
            if token in model.wv:
                wv += model.wv[token]
                token_count += 1

    # 平均した後L2正規化する
    if token_count > 20:
        wv /= token_count
        wv /= math.sqrt(sum(wv*wv))

        for i, val in enumerate(wv):
            wv_list[i] = float(val)

        title_wv_map[title] = wv_list

### 推論用に保存
with open("features_wv.json", "w") as f:
    json.dump(word_wv_map, f)

with open("title_wv.json", "w") as f:
    json.dump(title_wv_map, f)

3.1.2 特徴量の計算(推論)

3.1.1で作ったモデルを使って推論＝最もその人のツイートに近い楽曲の推薦を行います。
この部分はlambda上で動かすので、numpyやgensimは使わずにpure pythonで書いています。

predict.py


def get_wv_from_tweets(feature_wv_map, tweets):
    wv_length = 200
    wv = [0.0] * wv_length

    token_count = 0
    for tweet in tweets:
        for token in tweet:
            if token in feature_wv_map:
                for i, val in enumerate(feature_wv_map[token]):
                    wv[i] += val
                    token_count += 1

    if token_count == 0:
        return

    # 平均を計算する
    for i in range(len(wv)):
        wv[i] /= token_count

    # L2正規化
    l2_acc = 0
    for val in wv:
        l2_acc += val * val

    for i in range(len(wv)):
        wv[i] /= math.sqrt(l2_acc)

    return wv


def get_nearest_title(title_vector_map, wv):
    max_sim = 0
    max_title = None

    for title, vec in title_vector_map.items():
        sim = 0
        for val_wv, val_test in zip(vec, wv):
            sim += val_wv * val_test

        if sim > max_sim:
            print(title, sim)
            max_sim = sim
            max_title = title

    if sim == 0:
        return
    else:
        return max_title

with open("features_wv.json") as f:
    feature_wv_map = json.load(f)

with open("title_wv.json") as f:
    title_wv_map = json.load(f)

### tweets = ["tweet1", "tweet2", ... ] というiteratable
wv = get_wv_from_tweets(feature_wv_map, tweets)
nearest_title = get_nearest_title(title_wv_map, wv)

4. 3で得た楽曲情報を1.で得たstatus_idにリプライ

tweet.py

def post_tweet(twitter, body, reply_id):
    res = twitter.post(
        "https://api.twitter.com/1.1/statuses/update.json",
        params={"status": body, "in_reply_to_status_id": reply_id},
    )
    print(res)

    # エラー処理
    if res.status_code == HTTPStatus.OK:
        print("Successfuly posted: ", body)
    else:
        print(f"Failed: {res.status_code}")
        print(body)

CONSUMER_KEY        = 'XXX'
CONSUMER_SECRET     = 'XXX'
ACCESS_TOKEN_KEY    = 'XXX'
ACCESS_TOKEN_SECRET = 'XXX'

# twitter操作用クラス
twitter = OAuth1Session(
    CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN_KEY, ACCESS_TOKEN_SECRET
)

### 「本文」には3.で求めた楽曲のタイトルとかをよしなに入れてください
### reply_idは1.で求めてたstatus_idです
post_tweet(twitter, "本文", reply_id)

所感

lambdaで機械学習系のプロダクトを作るのはかなり厳しいなあと思いました
- 実行時のメモリはお金を積めば無理やり解決できるが、アップロード制限を回避する方法がないため
今回はいったん形にすることを最優先にして「漢字だけのword2vec」という謎のアプローチを採用しましたが、ちゃんとEC2立てて普通にword2vec使った方がいいんだろうなあという気がしています
ただ、（自分がTwitterでそこまで感情を吐く方ではないので、）楽曲の推薦結果が妥当かどうかの判断が主観的にすら難しいなあと思いました
- ので、誰かが使ってくれるとうれしいです。

再掲

https://twitter.com/shikumie に @付きでメンションするとオススメボカロ曲を教えてくれます。
ここまで書いた仕組みで動いていますので、興味がある方はお試しください。

AWS lambda (python) でツイート分析して楽曲推薦するTwitter botを作った

最初に