はじめに - プロトタイプから、頼れる「AIアシスタント」へ
第3回では、私たちが作り上げたボットがSlack上で初めて産声を上げました。@メンションに反応し、会話の文脈を短期的に記憶するその姿は、まさに生命の誕生の瞬間でした。
しかし、今の彼はまだ「賢いプロトタイプ」。チームの仕事を任せられる、真に「実用的なAIアシスタント」と呼ぶには、致命的な弱点が2つ残されています。
- 再起動で記憶を失う「揮発性の記憶」
現在、会話の記憶はサーバーのメモリ上にしかありません。サーバーを再起動、あるいはデプロイし直すたびに、全ての会話履歴は綺麗さっぱり消え去ってしまいます。 - 途中参加できない「表面的なスレッド理解」
ボットは自分が参加しているスレッドの文脈は追えますが、既に会話が進んでいるスレッドの途中でメンションされた場合、それまでの文脈を全く知らず、見当違いの応答をしてしまいます。
今回の最終回では、この2つの大きな壁を乗り越え、私たちのボットを本物の「チームアシスタント」へと育成するための、最後の仕上げを行います。
具体的には、以下の2つの大きなアップグレードを実装します。
-
記憶の永続化
- ボットの記憶装置を、揮発性のメモリから永続的なインメモリデータベース**「Redis」**へと換装し、サーバーを再起動しても記憶が失われないようにします。
-
完全なスレッド対応
- ボットがスレッドの途中から呼ばれても、そのスレッドの過去ログを自ら読み込み、「話の流れ」を完全に理解してから応答できるようにします。
この最終ステップを終えたとき、あなたのボットはもはや単なるおもちゃではありません。チームの知識を集約し、日々の業務を力強くサポートする、信頼に足るパートナーへと進化を遂げているはずです。さあ、最後の仕上げに取り掛かりましょう。
ステップ1：Redisによる記憶の永続化 - ボットに「本当の記憶」を
最初の課題は、サーバーを再起動すると記憶が消えてしまう問題です。これを解決するために、高速なインメモリ・キーバリューストアであるRedisを導入します。
なぜRedisなのか？
Redisは、今回の用途に最適な特徴をいくつも持っています。
-
超高速
- データをメモリ上で扱うため、読み書きが非常に高速です。
-
永続化対応
- メモリ上のデータをディスクに保存する機能があり、サーバーが落ちてもデータを復元できます。
-
手軽さ
- セッション管理やキャッシュなど、Webアプリケーションの様々な場面で使われる業界標準ツールであり、情報も豊富です。
Redisの準備（Dockerを使った簡単セットアップ）
ローカル環境で最も手軽にRedisを試す方法は、Dockerを使うことです。ターミナルで以下のコマンドを実行してください。
docker run -d --name mcp-redis -p 6379:6379 redis/redis-stack:latest
これで、localhost:6379でアクセスできるRedisサーバーが起動します。
次に、PythonからRedisを操作するためのライブラリをインストールします。
pip install redis
MCPEngineのシリアライズ対応
RedisにはPythonのオブジェクトをそのまま保存できません。そのため、MCPEngineクラスのインスタンスを、Redisに保存できる形式（今回はJSON文字列）に変換（シリアライズ）し、またその逆（デシリアライズ）を行う仕組みが必要です。
mcp_engine.pyに、以下の2つのメソッドを追加します。
- mcp_engine.py （追記・修正）
# from collections import deque を追記
from collections import deque
import json # jsonライブラリをインポート
class MCPEngine:
# __init__, add_message, build_messages は変更なし
# --- 以下を追記 ---
def to_json(self) -> str:
"""エンジンの状態をJSON文字列にシリアライズする"""
state = {
"system_prompt": self.system_prompt,
"max_history_size": self.history.maxlen // 2, # 往復数に戻す
# dequeは直接JSON化できないためlistに変換
"history": list(self.history)
}
return json.dumps(state)
@classmethod
def from_json(cls, json_str: str) -> "MCPEngine":
"""JSON文字列からエンジンの状態を復元（デシリアライズ）する"""
state = json.loads(json_str)
engine = cls(
system_prompt=state["system_prompt"],
max_history_size=state["max_history_size"]
)
# dequeに履歴をロード
engine.history.extend(state["history"])
return engine
main.pyのアップグレード：記憶装置の換装
いよいよサーバーの記憶装置を、揮発性の辞書から永続的なRedisに置き換えます。
main.pyを以下のように修正します。
- main.py（修正版）
import os
import asyncio
from dotenv import load_dotenv
from fastapi import FastAPI
from openai import AsyncOpenAI
from slack_bolt.async_app import AsyncApp
from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
import redis.asyncio as redis # 非同期版redisをインポート
from mcp_engine import MCPEngine
load_dotenv()
# --- 1. 初期化セクションの変更 ---
app = AsyncApp(token=os.environ.get("SLACK_BOT_TOKEN"))
openai_client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
# Redisクライアントを非同期モードで初期化
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# mcp_engines辞書は不要になったので削除！
# mcp_engines: dict[str, MCPEngine] = {}
# --- 2. イベントリスナーの実装（Redisを使うように変更） ---
@app.event("app_mention")
async def handle_app_mention_events(body: dict, say, client): # clientを追加
user_input = body["event"]["text"]
channel_id = body["event"]["channel"]
thread_ts = body["event"].get("thread_ts", body["event"]["ts"])
session_id = f"{channel_id}-{thread_ts}"
# Redisからセッションデータを取得
engine_json = await redis_client.get(session_id)
if engine_json:
# 既存セッションの場合、JSONからエンジンを復元
engine = MCPEngine.from_json(engine_json)
print(f"既存セッションをRedisからロードしました: {session_id}")
else:
# 新規セッションの場合、新しいエンジンを作成
system_prompt = "あなたは、関西弁で話す親しみやすいAIアシスタント『まいど君』です。ユーザーの質問には、必ずユーモアを交えて答えてください。"
engine = MCPEngine(system_prompt=system_prompt)
print(f"新しいセッションを開始しました: {session_id}")
# ... (LLMへの問い合わせロジックは変更なし) ...
try:
messages = engine.build_messages(user_input)
response = await openai_client.chat.completions.create(
model="gpt-4o-mini", messages=messages
)
ai_response = response.choices[0].message.content
engine.add_message("user", user_input)
engine.add_message("assistant", ai_response)
# ★重要：処理後にエンジンの最新状態をRedisに保存
await redis_client.set(session_id, engine.to_json())
except Exception as e:
print(f"エラーが発生しました: {e}")
ai_response = "すまん、ちょっと調子が悪いみたいや…。"
await say(text=ai_response, thread_ts=thread_ts)
# --- 3. FastAPIとSocket Modeの連携（変更なし） ---
api = FastAPI()
socket_handler = AsyncSocketModeHandler(app, os.environ.get("SLACK_APP_TOKEN"))
@api.on_event("startup")
async def startup():
asyncio.create_task(socket_handler.start_async())
これで記憶の永続化は完了です！
サーバーを起動し、ボットと少し会話してみてください。その後、Ctrl+Cでサーバーを停止し、再度起動してから同じスレッドで会話を続けてみましょう。ボットが以前の会話を覚えているはずです。ボットはついに、本当の記憶を手に入れました。
ステップ2：完全なスレッド対応 - ボットに「空気を読む力」を
残る最後の課題は、ボットが参加していなかった会話の文脈を理解できないことです。これを解決するには、メンションされた際に、そのスレッドの過去ログを自ら取得して「予習」する機能を実装します。
幸い、slack_boltライブラリを使えば、この処理も簡単です。イベントハンドラにclientオブジェクトを渡すことで、任意のSlack APIを呼び出せます。
handle_app_mention_events関数に、スレッド履歴を読み込むロジックを追加しましょう。
- main.py（最終版）
# (import文や初期化部分は変更なし)
@app.event("app_mention")
async def handle_app_mention_events(body: dict, say, client):
user_input = body["event"]["text"]
channel_id = body["event"]["channel"]
thread_ts = body["event"].get("thread_ts", body["event"]["ts"])
session_id = f"{channel_id}-{thread_ts}"
engine_json = await redis_client.get(session_id)
if engine_json:
engine = MCPEngine.from_json(engine_json)
print(f"既存セッションをRedisからロードしました: {session_id}")
else:
# --- ここからがスレッド履歴読み込みの追加ロジック ---
print(f"新規セッションです: {session_id}。スレッド履歴を読み込みます。")
system_prompt = "あなたは、関西弁で話す親しみやすいAIアシスタント『まいど君』です。ユーザーの質問には、必ずユーモアを交えて答えてください。"
engine = MCPEngine(system_prompt=system_prompt, max_history_size=10) # 履歴保持数を少し増やす
try:
# conversations.replies APIでスレッドの会話履歴を取得
history_response = await client.conversations_replies(
channel=channel_id,
ts=thread_ts
)
messages = history_response.get("messages", [])
# 古いメッセージから順に、最大保持数までエンジンに「予習」させる
# (最後のメッセージは現在のメンションなので除外)
for msg in messages[:-1]:
# ボット自身の発言か、ユーザーの発言かを判定
role = "assistant" if msg.get("bot_id") else "user"
content = msg.get("text", "")
engine.add_message(role=role, content=content)
print(f"{len(engine.history)}件の過去ログを読み込みました。")
except Exception as e:
print(f"スレッド履歴の読み込みに失敗しました: {e}")
# --- 追加ロジックはここまで ---
# ... (LLMへの問い合わせとRedisへの保存、Slackへの応答ロジックは変更なし) ...
try:
# (略)
except Exception as e:
# (略)
await say(text=ai_response, thread_ts=thread_ts)
# (FastAPIとSocket Modeの連携部分も変更なし)
会話をした時にSlackに関連するエラーになった場合の対応
(venv) ➜ pythontest python -m uvicorn main:api --reload
INFO: Will watch for changes in these directories: ['/Users/akira/Documents/pythontest']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [16118] using WatchFiles
INFO: Started server process [16120]
INFO: Waiting for application startup.
INFO: Application startup complete.
⚡️ Bolt app is running!
新規セッションです: C093V1FNKU6-1751439723.786869。スレッド履歴を読み込みます。
スレッド履歴の読み込みに失敗しました: The request to the Slack API failed. (url: https://slack.com/api/conversations.replies, status: 200)
The server responded with: {'ok': False, 'error': 'missing_scope', 'needed': 'channels:history,groups:history,mpim:history,im:history', 'provided': 'app_mentions:read,chat:write'}
このエラーmissing_scopeは、Slackアプリが必要な権限（スコープ）を持っていないことを意味します。
具体的には、スレッドの履歴を読み込むAPI (conversations.replies) を呼び出そうとしましたが、アプリにはその操作を実行する権限がありませんでした。
原因
エラーメッセージに原因が示されています。
必要な権限 (needed):
- channels:history (パブリックチャンネルの履歴読み取り)
- groups:history (プライベートチャンネルの履歴読み取り)
- mpim:history (グループDMの履歴読み取り)
- im:history (DMの履歴読み取り)
アプリが持っている権限 (provided):
- app_mentions:read (アプリへのメンション読み取り)
- chat:write (メッセージの投稿)
つまり、アプリはメッセージを投稿する権限はありますが、チャンネルのメッセージ履歴を読み取る権限がないため、エラーとなっています。
解決策 ✅
Slackアプリに不足している権限（スコープ）を追加する必要があります。
- Slack APIサイトにアクセスし、該当するアプリを選択します。
- 左側のメニューから「OAuth & Permissions」をクリックします。
- 「Scopes」のセクションまでスクロールします。
- 「Bot Token Scopes」または「User Token Scopes」の下にある「Add an OAuth Scope」ボタンをクリックします。
- 入力フィールドに、エラーメッセージで要求されていたスコープ（例: channels:history, groups:history, im:history など）を追加します。どのタイプの会話でも対応できるよう、必要なものを全て追加するのがお勧めです。
- スコープを追加したら、ページ上部に表示される指示に従ってアプリをワークスペースに再インストールします。
- アプリを再インストールすると新しい権限が有効になり、エラーが解消されます。
最終テスト：ボットの「空気を読む力」を見る
この最終版サーバーを起動し、以下のシナリオを試してみてください。
- あなたと同僚が、あるスレッドで10回ほど普通に会話をします（ボットはメンションしない）。
- 会話が盛り上がってきたところで、あなたがボットにメンションします。 「@My-MCP-Bot ここまでの議論を3行でまとめてくれへん？」
- ボットは、メンションされて初めてそのスレッドに参加したにもかかわらず、conversations.replies APIでそれまでの文脈をすべて「予習」し、的確な要約を返してくれるはずです。
この機能こそ、ボットが単なる応答機械ではなく、議論の流れを理解し、途中からでも的確にサポートできる**「賢いファシリテーター」**へと進化した証です。
シリーズ完結：あなたのAIアシスタント、任務開始
全4回にわたる長い旅路、お疲れ様でした。私たちはついに、実用レベルで機能する賢いAIアシスタントを、自らの手で作り上げました。
このシリーズを通して、あなたのボットは以下の能力を身につけました。
- 強固なアイデンティティ（System Prompt）
- 文脈を追う短期記憶（Sliding Window）
- 再起動しても忘れない永続記憶（Redis）
- 議論の流れを読むスレッド対応能力
このボットは、もはやあなたのPCの中だけの存在ではありません。チームのSlackワークスペースの一員として、日々の業務を助け、生産性を向上させる、頼もしい仲間となる準備ができています。
ここから先の道
このシリーズはゴールであると同時に、新たなスタート地点でもあります。ここから先、あなたのボットをさらに育成していくためのアイデアは無限にあります。
-
RAGによる知識拡張
- Vector Databaseと連携し、社内ドキュメントを読み込ませ、チームの「生き字引」にする。
-
インタラクティブ機能
- ボタンやモーダルを追加し、ユーザーがより直感的にボットを操作できるようにする。
-
本格的なデプロイ
- RenderやFly.io、AWSなどのクラウドプラットフォームにデプロイし、24時間365日稼働させる。
-
ロギングと監視
- ボットとの対話ログを分析し、プロンプトや機能を継続的に改善していく。
この連載で得た知識と経験が、あなたがAIと共に新しい価値を創造していく上での、確かな礎となることを心から願っています。
さあ、あなたの育てたAIアシスタントに、最初の任務を与えてあげてください！
この記事が、皆さんのAIアプリケーション開発の安定化に少しでも貢献できれば嬉しいです。役に立ったと感じたら、ぜひ Like をお願いします！
Written by A.H.