あらまし
住み着いているDiscordサーバで対話のできるBOTが動かせたら面白いなと思って、とりあえず簡単に動きそうなマルコフ連鎖を使ったBOTを作ってみた。
入力されたテキストを含むマルコフ連鎖のモデルを作って、それを元に返信する単純な学習機能を入れてます。
環境
私はUbuntuServerでscreenの中にvenvを建てて動かしてます。
こいつらを入れて動かします。
ライブラリのインストール
python3 -m pip install -U discord.py
py -3 -m pip install -U discord.py #Windowsの場合はこっち
pip install mecab-python3
pip install unidic-lite
pip install markovify
中身
BOTを動かしたいサーバのチャンネルIDとBOTのトークンは適宜書き換えてください。
discordbot.pyと同ディレクトリにm.txtを置いて動かします。m.txtにはある程度文章を入れておきます。
マルコフ連鎖に使うデータが無いと返信が帰ってこないのでモデル生成の部分は state_size=1 としています。会話を進めてある程度m.txtに文章量が集まったら1ずつ増やすといいです。
discordbot.py
import discord
import MeCab
import unidic_lite
import markovify
import random
CHANNELID = XXXXXXXXXXXXXXXXXXX #チャンネルID
TOKEN = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX' #BOTのトークン
intents = discord.Intents.default()
intents.message_content = True
intents.messages = True
client = discord.Client(intents=discord.Intents.all())
@client.event
async def on_ready():
print('---------------------------------------------------')
print(' 起動しました ')
print('---------------------------------------------------')
@client.event
async def on_message(message):
if message.channel.id != CHANNELID:
return
if not message.author.bot:
channel = client.get_channel(CHANNELID)
m1 = (message.content)
# 末尾を"。"に定める
matubi = (m1[-1])
if matubi == "。":
m2 = m1
else:
m2 = m1 + "。"
# 学習データに書き込み
f = open('m.txt', 'a')
f.write(m2)
f.write("\n")
f.close()
# 学習データを読み込み
with open('m.txt', 'r') as f:
kotoba = f.read().split("\n")
print(m2)
# 話題になりそうなワードを抽出
mecab = MeCab.Tagger()
result = mecab.parse(m2)
l1 = [line.split()[0] for line in mecab.parse(m2).splitlines()
if '0' in line.split()[-1]]
l2 = [line.split()[0] for line in mecab.parse(m2).splitlines()
if '1' in line.split()[-1]]
l3 = l1 + l2
#リストを作成できなかった場合冒頭の単語を取る
kazu = len(l3)
if kazu < 1:
words = mecab.parse(m2).split()
#print(words)
wadai = words[0]
else:
wadai = random.choice(l3)
# 文章の処理
breaking_chars = ['(', ')', '[', ']', '"', "'"]
splitted_kotoba = ''
for line in kotoba:
parsed_nodes = mecab.parseToNode(line)
while parsed_nodes:
try:
if parsed_nodes.surface not in breaking_chars:
splitted_kotoba += parsed_nodes.surface
if parsed_nodes.surface != '。' and parsed_nodes.surface != '、':
splitted_kotoba += ' '
if parsed_nodes.surface == '。':
splitted_kotoba += '\n'
except UnicodeDecodeError as error:
print('Error : ', line)
finally:
parsed_nodes = parsed_nodes.next
# モデル作成
model = markovify.NewlineText(splitted_kotoba,well_formed=False,state_size=1)
# wadaiから続く文章を生成
sentence = model.make_sentence_with_start(beginning=wadai,strict=False)
if sentence is not None:
#。を除く
out = (''.join(sentence.split()))
out = out[:-1]
print(out)
else:
print('None')
out = 'None'
# discordに送信
await message.channel.send(out)
client.run(TOKEN)
雑にコピペしたりして書いたので蛇足な部分があるかも。
おわりに
偶に偶然人間らしく振る舞ったりしてなかなか面白いです。
追記
ぼちぼちと手を加えました。
画像をTransformerで読んで(それで喋れよ)そこから返答したり、モデルの元になってる文章からワードクラウドを作ったり、あと知らない言葉を受け取ったときにエラーを吐かなくなりました。
Blip2を使ってるのでVRAMが4GBくらいでCUDAの使えるGPUが必要です。
キャプションの翻訳はDeeplのAPIが無料なので適当に入れました。
discordbot.py
import discord
import MeCab
import unidic_lite
import markovify
import random
from datetime import datetime
from pytz import timezone
from discord.ext import tasks
from wordcloud import WordCloud
import ipadic
import deepl
import torch
from PIL import Image
from transformers import Blip2Processor, Blip2ForConditionalGeneration
T_model = None
processor = None
CHANNELID = XXXXXXXXXXXXXXXXXXXXXX #ここ書き換えて
TOKEN = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX' #discordのやつ ここ書き換えて
TIME = '00:00'
intents = discord.Intents.default()
intents.message_content = True
intents.messages = True
client = discord.Client(intents=discord.Intents.all())
@client.event
async def on_ready():
global processor
global T_model
processor = Blip2Processor.from_pretrained("Salesforce/blip2-opt-2.7b")
T_model = Blip2ForConditionalGeneration.from_pretrained("Salesforce/blip2-opt-2.7b", torch_dtype=torch.float16, device_map="auto")
time_loop.start()
print('---------------------------------------------------')
print(' 起動しました ')
print('---------------------------------------------------')
#チャットと画像に対する返答
@client.event
async def on_message(message):
global processor
global T_model
if message.channel.id != CHANNELID:
return
if not message.author.bot:
channel = client.get_channel(CHANNELID)
#画像が貼られた場合
if message.attachments:
print("---画像の添付を検知")
for attachment in message.attachments:
await attachment.save("attachment.png")
await message.channel.send("......!")
image = Image.open("/Your/Directory/attachment.png") #ここ書き換えて
inputs = processor(images=image, return_tensors="pt").to("cuda", torch.float16)
generated_ids = T_model.generate(**inputs)
generated_text = processor.batch_decode(generated_ids, skip_special_tokens=True)[0].strip()
print(generated_text)
print("---画像からキャプション生成")
API_KEY = 'XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX' #DeepLのやつ ここ書き換えて
text = generated_text
source_lang = 'EN'
target_lang = 'JA'
# イニシャライズ
translator = deepl.Translator(API_KEY)
# 翻訳を実行
result = translator.translate_text(text, source_lang=source_lang, target_lang=target_lang)
translated_text = result.text
# ファイルに書き込む
with open("caption.txt", "w") as file:
file.write(translated_text)
caption = ("デイジーはかぷかぷと笑いあなたの側にいるその恐ろしいものに触れて弾けた。")
#画像からキャプションを抽出
f = open('caption.txt', 'r')
caption = f.read()
f.close()
m1 = caption
# 末尾を"。"に定める
matubi = (m1[-1])
if matubi == "。":
m2 = m1
else:
m2 = m1 + "。"
# 学習データに書き込み
f = open('m.txt', 'a')
f.write(m2)
f.write("\n")
f.close()
# 学習データを読み込み
with open('m.txt', 'r') as f:
kotoba = f.read().split("\n")
print(m2)
# 話題になりそうなワードを抽出
mecab = MeCab.Tagger()
result = mecab.parse(m2)
l1 = [line.split()[0] for line in mecab.parse(m2).splitlines()
if '0' in line.split()[-1]]
l2 = [line.split()[0] for line in mecab.parse(m2).splitlines()
if '1' in line.split()[-1]]
l3 = l1 + l2
#リストを作成できなかった場合冒頭の単語を取る
kazu = len(l3)
if kazu < 1:
words = mecab.parse(m2).split()
#print(words)
wadai = words[0]
else:
wadai = random.choice(l3)
#文章が送られた場合
else:
m1 = (message.content)
# 末尾を"。"に定める
matubi = (m1[-1])
if matubi == "。":
m2 = m1
else:
m2 = m1 + "。"
# 学習データに書き込み
f = open('m.txt', 'a')
f.write(m2)
f.write("\n")
f.close()
# 学習データを読み込み
with open('m.txt', 'r') as f:
kotoba = f.read().split("\n")
print(m2)
# 話題になりそうなワードを抽出
mecab = MeCab.Tagger()
result = mecab.parse(m2)
l1 = [line.split()[0] for line in mecab.parse(m2).splitlines()
if '0' in line.split()[-1]]
l2 = [line.split()[0] for line in mecab.parse(m2).splitlines()
if '1' in line.split()[-1]]
l3 = l1 + l2
#リストを作成できなかった場合冒頭の単語を取る
kazu = len(l3)
if kazu < 1:
words = mecab.parse(m2).split()
#print(words)
wadai = words[0]
else:
wadai = random.choice(l3)
# 文章の処理
breaking_chars = ['(', ')', '[', ']', '"', "'"]
splitted_kotoba = ''
for line in kotoba:
parsed_nodes = mecab.parseToNode(line)
while parsed_nodes:
try:
if parsed_nodes.surface not in breaking_chars:
splitted_kotoba += parsed_nodes.surface
if parsed_nodes.surface != '。' and parsed_nodes.surface != '、':
splitted_kotoba += ' '
if parsed_nodes.surface == '。':
splitted_kotoba += '\n'
except UnicodeDecodeError as error:
print('Error : ', line)
finally:
parsed_nodes = parsed_nodes.next
# モデル作成
model = markovify.NewlineText(splitted_kotoba,well_formed=False,state_size=2)
# wadaiから続く文章を生成
try:
sentence = model.make_sentence_with_start(beginning=wadai,strict=False)
if sentence is not None:
#。を除く
out = (''.join(sentence.split()))
out = out[:-1]
print(out)
else:
print('None')
out = 'None'
except markovify.text.ParamError:
out = (wadai + ' のことわかんないかも')
# discordに送信
await message.channel.send(out)
#0時にワードクラウドを表示
@tasks.loop(seconds=59)
async def time_loop():
now = datetime.now(timezone('Asia/Tokyo')).strftime('%H:%M')
# 現在の時刻を取得
if now == TIME:
print('---------------------------------------------------')
print(' ワードクラウドを投稿 ')
print('---------------------------------------------------')
FONT_PATH = "./XXXXXXX.ttf" #ここ書き換えて
# テキストファイル読み込み
read_text = open("m.txt", encoding="utf8").read()
mecab = MeCab.Tagger()
result = mecab.parse(read_text)
node = mecab.parseToNode(read_text)
output = []
#除外リスト
stoplist =['(',')']
# 品詞に分解して、助詞や接続詞などは除外している。
while node:
word_type = node.feature.split(",")[0]
if word_type in ["名詞","形容詞","動詞"]:
if not node.surface in stoplist and not node.surface.isdigit():
output.append(node.surface.upper())
node = node.next
# 文字列取得
text = ' '.join(output)
# wordcloudで可視化
def color_func(word, font_size, position, orientation, random_state, font_path):
return 'black'
wordcloud = WordCloud(background_color="white",color_func=color_func, font_path=FONT_PATH ,width=1024,height=1024,max_font_size=500,min_font_size=15,max_words=400).generate(text)
wordcloud.to_file("./wordcloud.png")
channel = client.get_channel(CHANNELID)
file = discord.File("/Your/Directory/wordcloud.png", filename="wordcloud.png") #ここここ書き換えて
embed = discord.Embed()
embed.set_image(url="attachment://wordcloud.png")
await channel.send(file=file, embed=embed)
client.run(TOKEN)