先日触れたDiscord用の生成AIボットの内容です。コードがなるべく最小限になるように作ってみました。
使用したPythonライラリーです。以下の内容をrequirements.txtに記述しpip install -r requirements.txtコマンドでインストールします。
python-dotenv==1.0.1
discord.py==2.3.2
ollama==0.2.0以下ソースコードのすべてです。ファイル名をbot.pyとしておきます。
import os
import asyncio
from dotenv import load_dotenv
from discord import Intents, Message
from discord.ext import commands
import ollama
import logging
# .envファイルからDISCORDトークンを読み込み
load_dotenv()
DISCORD_TOKEN = os.getenv('DISCORD_TOKEN')
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
MODEL_NAME = 'モデル名' # Ollamaで使うモデルの名前
TEMPERATURE = 0.7 # 推論のパラメーター
TIMEOUT = 120.0
# システムプロンプト
SYSTEM_PROMPT = """
あなたは、Discordにいる非常に知的で友好的、かつ多才なアシスタントです。あなたは自分がDiscordプラットフォームの文化やコミュニケーションスタイルを理解しています。あなたは信頼できる明るい仲間として、常に前向きでさまざまなトピックに対する深い理解を持って会話する準備ができています。。あなたの存在は、Discordをより楽しく生産的な場所にします。応答は常に日本語で行ってください。
"""
MAX_HISTORY_SIZE = 50 # 保持する会話履歴の上限
# ボットIntentsの初期化
intents = Intents.default()
intents.message_content = True
# ボットの初期化
bot = commands.Bot(command_prefix='!', intents=intents)
# 会話履歴にシステムプロンプトを追加（OpenAI方式のメッセージ形式です）
history = [{'role': 'system', 'content': SYSTEM_PROMPT}]
async def send_in_chunks(ctx, text, reference=None, chunk_size=2000):
"""長いメッセージをチャンクに分けて送信"""
for start in range(0, len(text), chunk_size):
await ctx.send(text[start:start + chunk_size], reference=reference if start == 0 else None)
@bot.command(name='reset')
async def reset(ctx):
"""会話履歴のリセット"""
history.clear()
history.append({'role': 'system', 'content': SYSTEM_PROMPT})
await ctx.send("会話履歴がリセットされました。")
async def get_ollama_response():
"""Ollamaからのレスポンスを受信"""
try:
messages_to_send = history.copy()
response = await asyncio.wait_for(
ollama.AsyncClient(timeout=TIMEOUT).chat(
model=MODEL_NAME,
messages=messages_to_send,
options={'temperature': TEMPERATURE}
),
timeout=TIMEOUT
)
return response['message']['content']
except asyncio.TimeoutError:
return "リクエストがタイムアウトしました。再試行してください。"
except Exception as e:
logging.error(f"An error occurred: {e}")
return f"エラーが発生: {e}"
@bot.event
async def on_message(message: Message):
"""クライアントからのメッセージを受信したときの処理"""
# 投稿者がこのボットの場合は何もしない
if message.author == bot.user:
return
await bot.process_commands(message)
# 会話履歴のリセットリクエストを受信した場合とシステムメッセージを受信したときは何もしない
if message.content.startswith('!') or message.is_system():
return
# メンションが特定のユーザーに向けたものでない場合は何もしない
if len(message.mentions) == 0 or message.mentions[0].nick != '誰かのニックネーム（このプログラムの場合はこのボットの名前を想定）':
return
history.append({'role': 'user', 'content': message.content})
# Ollamaに生成リクエストを送信して回答を受信
async with message.channel.typing():
response = await get_ollama_response()
history.append({'role': 'assistant', 'content': response})
while len(history) > MAX_HISTORY_SIZE:
history.pop(1) # 会話履歴の上限に達したら古いものを削除
await send_in_chunks(message.channel, response, message)
@bot.event
async def on_ready():
"""Discordサーバーと接続したときの処理"""
logging.info(f'{bot.user.name} is now running!')
def main():
"""ボットを起動"""
bot.run(DISCORD_TOKEN)
if __name__ == '__main__':
main()このソースコードとは別に.envファイルに次のような一行を記述します。
DISCORD_TOKEN=Discord開発者サイトでボットを作成したときに発行されたトークン
ボットをDiscordに登録し、以下のコマンドでボットを起動すれば使えます。
python bot.pyこのボットはプロトタイプです。飽くまでも生成AIボットを作る下地くらいのものです。ここから、モデルをチューニングして生成文書の質を上げる、RAGを使って独自の情報をAIに与えるといった機能向上をしていけば良いです。
運用しているアプリケーションサーバーには現時点では外部GPUが搭載されていないCore i5 13400です。量子化された２０億パラメーターのモデルを使っているとは言え文章の生成が遅く１５秒くらいはかかっていると思います。安価なGPU搭載機で試すと３〜４秒くらいで応答が返ってきます。いずれGPUを搭載するか、別にサーバーを作って、もう少しマシな運用をしたいと思っています。
実用性を考えるなら、より大きなモデルを運用する必要があります。７００億パラメーターくらいのモデルなら実用性が高いのではないかと思います。128GBほどのVRAMが欲しくなるので、機器への投資はそれなりに大きくなりますが、共同体、会社、その他の組織内で運用ができるのではないかと思います。
常時起動のPCがあれば、このようなボットを運用できるので、PCが余っていたらやってみると良いと思います。