今回は、自然言語処理AIとtwitterAPIを使って定期的にツイート・リプライに返信していきましょう。
この記事にぴったりな人
ちなみに、私は「凪ノなぎ」というAIbotの開発・運用をしており、その実際のコードの公開・簡単な説明をしていきたいと思います。
twitterアカウントはこちら。
twitter.com
みなさんもぜひフォローとリプライをお願いします。
ちょっとやってみたいけど、完成形がどのようになるのか分からなくて心配という方もぜひ一度「凪ノなぎ」のツイートやリプを見て、その後にやってみるかどうかを決めるのもありだと思います。
「凪ノなぎ」の要点は、主に3つ
- 自然言語処理（AI）
- 定期ツイート
- 定期リプライ処理
「凪ノなぎ」の自然言語処理部分は、無料公開されているrinna/japanese-gpt2-mediumをファインチューニングし、それを「凪ノなぎ」として運用しています。
以下のブログ内のコードは、全てgithubで公開しているので気になった方は是非試してみてください。
github.com
開発の準備
- twitterAPIの取得
twitterBotには、twitterAPIが必要になります。
もちろんtwitterBotにしたくない方は無視していだいて結構です。
わかりやすかったサイト様
【0からわかる！？】Twitter API利用申請＆基本操作コードまとめ【超入門】 | - hugging faceを使用するための環境を整える
リンク先の説明に沿ってインストールをお願いします。 🤗 Transformers
ローカルでなくて、google colaboratoryでも出来ると思います。
実行環境
- Python 3.10.6
- Pytorch 1.12.1+cu116
- 12th Gen Intel(R) Core(TM) i7-12700F 2.10 GHz
- RAM 32.0 GB
- GPU RTX 3070
自然言語処理AIについて
huggingfaceでは、学習済みの機械学習モデルやデータセットなどを公開されており、それをライセンスに応じて使うことが出来ます。
また、本来自然言語処理では、形態素解析→構文解析→意味解析→文脈解析という順序で行い、それぞれの処理を行うために様々なライブラリを使わなければならないのですがhuggingfaceで公開されているモデルは、これらの処理をはるかに短くわかりやすいコードで使うことができるのでとっても便利です。感謝してます。
今回使わせていただいているrinnaモデルはrinna株式会社さんが開発されたもので、主にwikipedia/cc100をトレーニングに使用したようです。
しかし、公開されているモデルはtwitter向きではないので、ファインチューニングを行い、女の子らしさとtwitterらしさを出すことにしました。
ファインチューニングとは、学習済みデータを自分の使いたいように調整することです。
女の子らしさが欲しかったのでネットの海をさまよい、女の子のセリフを軽くまとめました。
実際にファインチューニングする言葉は、train.txtにまとめてあります。
それらを実際にtrain_rinnna.pyでファインチューニングしました。
簡単にファインチューニングのコードを見ていきましょう。
自然言語処理についてはこちらの方のYoutubeがとってもわかりやすいのでおすすめです。
youtu.be
ファインチューニング
githubのコードはこちら。
では、コードの簡単な説明をしていきます。
これらは、huggingfaceで公開されているモデルを使うときのおまじないです。
とりあえず、書いておきましょう。train_rinnna.py#L9-L12
tokenizer = T5Tokenizer.from_pretrained("rinna/japanese-gpt2-medium") tokenizer.do_lower_case = True # due to some bug of tokenizer config loading model = AutoModelForCausalLM.from_pretrained("rinna/japanese-gpt2-medium")
ファインチューニングするコードは、大体定型文で決まっていますが、場合に応じて変えないといけないのは下のtrain.txtの部分です。
train.txtには、自分がファインチューニングした言葉を
つめたテキストファイルを入れます。
ファインチューニングは、いわゆる教師あり学習です。なので、trainでトレーニングし、
その結果があっているかどうかをtestで指定したテキストファイルで検証し、正解であるtrain.txt内の言葉に近づけていくようにしていきます。train_rinnna.py#L17-L19
data_files={
"train":"train.txt",
"test":"train.txt"
ここではどれくらいトレーニングの設定をします。
・output_dir：レーニング結果の保存場所
・num_train_epochs : 訓練のエポック数
・per_device_train_batch_size : バッチサイズ。（指定したテキストファイルをいくつに分けてトレーニングを行うか）
・load_best_model_at_end=True：学習後に最良のモデルを自動でロード
・evaluation_strategy：指定した値のステップ数ごとにバリデーション（検証）
train_rinnna.py#L30-L37
training_args = TrainingArguments(
output_dir="output",
num_train_epochs=3,
per_device_train_batch_size=2,
per_device_eval_batch_size=2,
load_best_model_at_end=True,
evaluation_strategy="steps"
)
上のtrainerで、設定したものをまとめており、それを実行する。train_rinnna.py#L54
trainer.train()
定期ツイート
上のファインチューニングでトレーニングしたものを使用して、定期的にツイートするようにしていきます。
定期ツイートは、2段階に分けることが出来ます。
・pythonでツイート
・windowsのタスクスケジューラーで定期的に実行
要はpython単体でツイートするコードを書くことができれば、あとは他のプログラムやサービスに任せれば定期的に実行することができるということです。
*pythonのsleep関数で定期的に実行しようとするのはあまりおすすめしません。
なので、今回はpythonでツイートするコードを書いて行きましょう。
githubのコードはこちら。
では、コードの簡単な説明をしていきます。
pythonでツイート
twitterAPIの設定
twitterbotではtwitterAPIが必要になり、それをここに貼り付けて設定します。
変えなければならないのは、CONSUMER_KEY・CONSUMER_SECRET・ACCESS_TOKEN・ACCESS_SECRETのみで、後はコピペで良いと思います。
Autotweet.py#L11-L33
#Twitter各種設定 #取得したtwitterAPIを貼り付ける CONSUMER_KEY = '' CONSUMER_SECRET = '' ACCESS_TOKEN = '' ACCESS_SECRET = '' #twitter各種設定 auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET) auth.set_access_token(ACCESS_TOKEN, ACCESS_SECRET) api = tweepy.API(auth,wait_on_rate_limit=True) # OAuthの認証オブジェクトの作成 oauth = twitter.OAuth(ACCESS_TOKEN, ACCESS_SECRET, CONSUMER_KEY, CONSUMER_SECRET) # Twitter REST APIを利用するためのオブジェクト twitter_api = twitter.Twitter(auth=oauth) # Twitter Streaming APIを利用するためのオブジェクト twitter_stream = twitter.TwitterStream(auth=oauth)
ツイート処理
お願い：voc_list_kiso_kansei.csvをjapanese_words.csvとしてnagino_nagi_tweetBot_publicのディレクトリ内に入れてください。
では、次に実際にtweetを行う処理について見ていきます。
まず、何についてAIが文を作るのかの単語を指定しなければならないため、日本語の単語を収録したjapanese.csvの中から単語をランダムに１つ選ぶ処理をします。
#csvから単語のリストを読み込む csv_file = open("japanese_words.csv", "r", encoding="shift_jis") f = csv.reader(csv_file, delimiter=",", doublequote=True, lineterminator="\r\n", quotechar='"', skipinitialspace=True) header = next(f) #単語を１文字取る japanese_word=[] for row in f: #rowはList japanese_word.append(row[2]) tweet_seed = random.choice(japanese_word)
次に、ファインチューニングしたモデルを設定します。
ここでは、modelをoutput\checkpoint-5500としていますが、
modelには、上のファインチューニングしたモデル内でcheckpointの数字が一番大きいディレクトリを指定してください。
ファインチューニングを行っていない場合は、modelに「rinna/japanese-gpt2-medium」を指定していただければと思います。
Autotweet.py#L54-L57
# トークナイザーとモデルの準備 tokenizer = T5Tokenizer.from_pretrained("rinna/japanese-gpt2-medium") tokenizer.do_lower_case = True # due to some bug of tokenizer config loading model = AutoModelForCausalLM.from_pretrained("output\checkpoint-5500")
上でランダムに選んだ単語の中に全角が紛れていることがあるかもしれないので、全角を半角に直す。
また、大文字のアルファベットは小文字に直しています。
Autotweet.py#L63
#.transrate()で全角を半角に直す #.lower()でアルファベットを小文字に直す word = tweet_seed.translate(str.maketrans({chr(0xFF01 + i): chr(0x21 + i) for i in range(94)})).lower()
実際に推論を行っています。
また、時折、存在しない写真のurl（pic.twitter）が生成される事があるため3文生成し、リストに入れています。
生成後、それらを除くような処理をしています。
何も生成されなかった場合、「...」とするようにしています。また、特殊文字・記号は取り除いています。
Autotweet.py#L65-L89
# 推論の実行 input = tokenizer.encode(word,return_tensors="pt",) #do_sample=True、ランダムな自然な文を生成するため。repetition_penalty = 1.1、同じ言葉が反復して出てこないようにするため。num_return_sequences、文の数 output = model.generate(input, do_sample=True, max_length=100, num_return_sequences=3, repetition_penalty =1.0) #skip_special_tokens、いらない文字を無くすため become_voice_list=[] become_voice_list = tokenizer.batch_decode(output,skip_special_tokens=True) #推論された文字を調整 #pic.twitterが入ってる場合は、次の配列に移る for i in become_voice_list: if 'pic' in i: print('スキップ:'+i) continue else: become_voice_str = "".join(i) #become_voice_str=become_voice_str.replace(word,'') if become_voice_str == '': become_voice_str = '...' #文章的に直すところ voice = become_voice_str.replace("[" , "").replace("@" , "").replace("ω" , "").replace("_","").replace('&','')
設定したtwitterAPIを使用してツイートします。
#リストに含まれるツイート内容をランダムでツイート print(voice) api.update_status(voice)
定期実行
windowsタスクスケジューラーを使っていくのですが、直接.pyをpythonで実行しようとしてもうまく動きませんでした。
なので、以下のバッチファイルを書き、バッチファイルをタスクスケジューラーで設定し30分ごとに実行させています。
@if not "%~0"=="%~dp0.\%~nx0" start /min cmd /c,"%~dp0.\%~nx0" %* & goto :eof python 場所/Autotweet.py
定期リプライ
ファインチューニングでトレーニングしたものを使用して、定期的にツイートするようにしていきます。
定期ツイートは、3段階に分けることが出来ます。
・タイムラインからリプライを取得
・リプライ
・定期実行
が、定期ツイートと重なる部分も多くあるので、重なる部分の説明は省きます。
githubのコードはこちら
タイムライン取得
目的は、「自分のツイートに対するリプライを取得したい。」
そもそも、twitterAPIにおけるタイムラインは、2つに分けられます。
・user_timeline
特定したユーザーのタイムラインを取得できる。
・home_timeline
自分のタイムラインを取得できる。
自分のツイートに対するリプライは、home_timelineに表示されるのでhome_timelineかを取得し、
リプライを探していきます。
home_timelineから3200のツイートを取得します。
その中で自分のアカウント名がある場合は、そのツイートを取得しデータを整形します。
下のコードで処理は出来ますが、より深くタイムラインの取得について詳しく知りたい方は、
タイムラインの取得と整形は奥が深いのでこちらの記事におまかせします。
twitter_Auto_reply.py#L86-L107
for i, tweet in enumerate(tweepy.Cursor(api.home_timeline).items(3200)): #.find()が-1のときはその中に無い if tweet.text.find('@アカウント名') != -1: #print("ツイート本文:" +tweet.text+'投稿者:'+str(tweet.user.name)+str(tweet.user.screen_name)+"ツイートid:"+str(tweet.id)) before_seed = tweet.text.split(" ") tweet_seed = before_seed[-1].replace("\n","") #.transrate()で全角を半角に直す #.lower()でアルファベットを小文字に直す word = tweet_seed.translate(str.maketrans({chr(0xFF01 + i): chr(0x21 + i) for i in range(94)})) print(word)
次に取得したツイートに対してのリプライがすでに行われているかどうかを調べる事が必要になります。
仮に、この確かめを行わなかった場合、リプライしていただいたツイートに何回もリプライを返してしまうことになります。
すでにリプライが行われた他者の自分に対するリプライは、すべてauto_rep.txtに保存します。
そのため、このauto_rep.txtにツイートがあるかを確かめ、なかった場合は、リプライ処理を行うようになっています。
twitter_Auto_reply.py#L109-L129
#テキストファイルにあるかどうか if word+'\n' in datafile: print('済み') else: print('リプライ処理') #.txtに書き込み f = open('auto_rep.txt', 'a',encoding='utf-8') f.write(word+'\n') f.close() voice = NaginoNagi_GPT(word) # 特定のツイートへのリプライの場合 reply_text = '@'+tweet.user.screen_name +str(before_seed[0:-2]).replace("[",' ').replace("]",'').replace("'",'')+ '\n' reply_text += voice print("reply->>>>"+reply_text) #リプライ twitter_api.statuses.update(status=reply_text, in_reply_to_status_id=tweet.id)
後は、上のツイート処理と特に変わりがないため割愛させていただきます。
質問など訂正があった場合は、コメント・twitterにお願い致します。
