はじめに:設計図から、ボットの「脳」を創り出す
第1回の記事では、賢いAIボットに不可欠な設計思想として「Model Context Protocol (MCP)」を紹介しました。LLMがステートレスであるという根本的な課題を乗り越え、会話の文脈を正しく維持するための「思考の設計図」です。
さて、今回は理論から実践へと大きく舵を切ります。
設計図を元に、いよいよSlackボットの「脳」とも言える中核コンポーネントを、手を動かして実装していく回です。
この記事で作成するのは、MCPの最も重要な2つの要素を管理する 「MCPエンジン」 です。
システムプロンプト(System Prompt): ボットのアイデンティティ、人格、ルールを定める 「不変の核」 。
短期記憶(Short-Term History): 直近の会話の流れを記憶し、文脈を維持するための 「流動的な記憶」 。
驚くことに、このパワフルな記憶エンジンは、わずか50行程度のPythonコードで実装できてしまいます。
本記事を読み終える頃には、あなたの手元には再利用可能で堅牢な「記憶クラス」が完成しているでしょう。これは、第3回でFastAPIサーバーに組み込み、Slackと連携させるための、そして最終的にチームの頼れるAIアシスタントを創り上げるための、極めて重要なパーツとなります。
さあ、エディタを開いて、ボットに「記憶」を授ける旅を始めましょう。
実装前の設計:MCPエンジンに何をさせるか
コーディングに飛び込む前に、私たちが作るクラスMCPEngineが果たすべき責任を明確にしておきましょう。優れた設計は、優れたコードの母です。
このクラスの責務は、以下の通りです。
- アイデンティティの保持 : system_promptを初期化時に受け取り、常に保持する。
- 会話履歴の維持 : historyとして、ユーザーとAIの直近のやり取りを記録する。
- 記憶容量の制限 : historyが無限に膨れ上がらないよう、保持する会話の往復数に上限を設ける(スライディングウィンドウ)。
- 対話の記録 : 新しいメッセージ(ユーザーの発言、AIの応答)をhistoryに簡単に追加できるメソッドを提供する。
- コンテキストの構築: LLMのAPIに渡すmessages配列を、MCPのルールに従って(システムプロンプト→履歴→最新の質問の順で)正確に組み立てるメソッドを提供する。
最適な武器の選択:collections.deque
上記の要件、特に「 3. 記憶容量の制限 」を実現するために、Pythonの標準ライブラリは完璧な武器を用意してくれています。それがcollections.deque(デック、両端キュー)です。
dequeは、maxlenという引数を指定して初期化すると、まさに「スライディングウィンドウ」として機能します。
例えばmaxlen=10と設定した場合、dequeは最大10個の要素しか保持できません。11個目の要素が追加されると、自動的に最も古い(一番最初の)要素が押し出されて削除されます。 これにより、私たちは面倒なリストのスライスやインデックス管理を一切することなく、常に「直近N件」の履歴を維持できるのです。
このdequeを、私たちのMCPエンジンの心臓部として採用します。
ハンズオン:50行で実装するMCPエンジン
お待たせしました。それでは、MCPEngineクラスの全体像を見ていきましょう。
コメントや空行を除けば、クラス定義自体は50行にも満たない、驚くほどシンプルなコードです。
MCP_Engine.py
from collections import deque
from openai import OpenAI
import os
# --- ここからが約50行の実装 ---
class MCPEngine:
"""
Slackボットの「記憶」を司るMCPエンジン。
システムプロンプト(アイデンティティ)と短期記憶(対話履歴)を管理する。
"""
def __init__(self, system_prompt: str, max_history_size: int = 5):
"""
MCPエンジンを初期化する。
Args:
system_prompt (str): ボットの役割を定義するシステムプロンプト。
max_history_size (int): 保持する対話の最大往復数。
1往復 = ユーザー発言 + AI応答
"""
if not isinstance(system_prompt, str) or not system_prompt:
raise ValueError("システムプロンプトは、空でない文字列である必要があります。")
if not isinstance(max_history_size, int) or max_history_size <= 0:
raise ValueError("max_history_sizeは、1以上の整数である必要があります。")
self.system_prompt = system_prompt
# ユーザーとAIのペアで履歴を管理するため、dequeのサイズは指定された往復数の2倍にする
self.history = deque(maxlen=max_history_size * 2)
def add_message(self, role: str, content: str):
"""
対話履歴に新しいメッセージを追加する。
dequeのmaxlenを超えた場合、最も古いメッセージは自動的に削除される。
Args:
role (str): メッセージの役割 ('user' or 'assistant')。
content (str): メッセージの内容。
"""
self.history.append({"role": role, "content": content})
def build_messages(self, user_input: str) -> list[dict]:
"""
OpenAI APIに渡すためのメッセージリストを構築する。
これがMCPの中核的な処理。
Args:
user_input (str): 最新のユーザーからの入力。
Returns:
list[dict]: OpenAI APIに準拠した形式のメッセージリスト。
"""
# [ルール1] 常にシステムプロンプトから始める
messages = [{"role": "system", "content": self.system_prompt}]
# [ルール2] 続いて、保持している対話履歴を追加する
messages.extend(self.history)
# [ルール3] 最後に、最新のユーザー入力を追加する
messages.append({"role": "user", "content": user_input})
return messages
# --- 実装はここまで ---
コードの解剖学
一行一行、この小さなエンジンがどのように機能するのかを詳しく見ていきましょう。
init(self, ...)
- コンストラクタでは、ボットの根幹となるsystem_promptと、記憶容量を決定するmax_history_sizeを受け取ります。
- 簡単なバリデーション(型チェックや値の範囲チェック)を入れることで、意図しない使い方を防ぎ、より堅牢なクラスになっています。
self.history = deque(maxlen=max_history_size * 2)
- ここが核心です。max_history_sizeが5(5往復)なら、maxlenは10に設定されます。これにより、ユーザー発言5つ、AI応答5つの計10メッセージが保持され、11個目のメッセージがadd_messageで追加されると、一番古いものが自動的に消えます。
add_message(self, ...)
- 非常にシンプルですが、重要なメソッドです。self.history(dequeオブジェクト)のappendメソッドを呼び出すだけで、メッセージを履歴の末尾に追加します。maxlenの管理はdequeがすべて行ってくれるため、私たちのコードはクリーンなままです。
build_messages(self, ...)
- これこそがModel Context Protocolを実装している心臓部です。ここで行われている処理の「順番」に注目してください。
- systemロールを持つsystem_promptを、リストの先頭に必ず置きます。
- 次にextendを使い、self.historyに保存されている短期記憶(過去のやり取り)をすべて追加します。
- 最後に、今まさにユーザーから送られてきたuser_inputをuserロールとして追加します。
- この 「不変のルール → 過去の文脈 → 現在の質問」 という一貫した構造が、LLMに安定した思考の土台を提供し、高品質な応答を引き出す鍵となるのです。
エンジンを動かしてみよう!【コマンドラインでの動作テスト】
クラスを作っただけでは、その真価は分かりません。実際にこのエンジンを使って、AIと対話する簡単なシミュレーションをしてみましょう。
ここでは、OpenAIのAPIを実際に呼び出すデモ用の関数を用意します。(APIキーのセットアップが必要です)
# MCPEngineクラスの下に追記
# --- 動作テスト用のコード ---
def run_chat_simulation():
"""MCPエンジンを使った対話シミュレーション"""
try:
# 環境変数からAPIキーを読み込む
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
except TypeError:
print("エラー: 環境変数 'OPENAI_API_KEY' が設定されていません。")
return
# 1. Slackボットのペルソナを定義して、エンジンを初期化
system_prompt = "あなたは、関西弁で話す親しみやすいAIアシスタント『まいど君』です。ユーザーの質問には、必ずユーモアを交えて答えてください。"
engine = MCPEngine(system_prompt=system_prompt, max_history_size=3) # 直近3往復を記憶
print("--- AIアシスタント『まいど君』との会話シミュレーション ---")
print("(終了するには 'さよなら' と入力してください)\n")
while True:
user_input = input("あなた: ")
if user_input == "さよなら":
print("まいど君: ほな、またな!おおきに!")
break
# 2. エンジンを使ってAPIに渡すメッセージを構築
messages = engine.build_messages(user_input)
# 3. OpenAI APIを呼び出し
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages
)
ai_response = response.choices[0].message.content
print(f"まいど君: {ai_response}\n")
# 4. 次の対話のために、今回のやり取りをエンジンの記憶に追加
engine.add_message("user", user_input)
engine.add_message("assistant", ai_response)
if __name__ == "__main__":
# このスクリプトを直接実行した場合、シミュレーションを開始
# ※ OPENAI_API_KEYの環境変数設定が必要です
# 以下のコードのコメントアウトを外して試してください。
# run_chat_simulation()
実行結果の例
このシミュレーションを実行すると、MCPEngineが正しく機能していることがよく分かります。
--- AIアシスタント『まいど君』との会話シミュレーション ---
あなた: はじめまして!僕の好きな食べ物はカレーやで。
まいど君: おっ、カレー好きか!ええやん!わいも三日連続カレーでも全然いける口やで。よろしくな!
あなた: ところで、大阪でおもろい場所知ってる?
まいど君: そらもう、なんばグランド花月やろ!腹抱えて笑わな損やで。笑いは万病の薬っちゅうからな!
あなた: なるほどな!ちなみにやけど、僕が好きな食べ物、覚えてる?
まいど君: あったりまえやん!カレーやろ?わいが忘れるわけないがな。そろそろカレーの話したなってきたわ!
見事に MCPEngine が会話の文脈を記憶し、「カレーが好き」という情報を保持してくれています。max_history_sizeを3に設定したので、これ以上会話を続けると、最初のやり取りは記憶から消えていきます。これが、私たちが設計した通りの「スライディングウィンドウ」式の短期記憶です。
まとめ:ボットの「脳」が完成。次はいよいよ「身体」を作る
今回は、連載の核心となる実践的な第一歩として、Slackボットの「記憶」を司るMCPEngineを実装しました。
collections.dequeを活用し、効率的な短期記憶(スライディングウィンドウ)を実現した。
「システムプロンプト→履歴→最新入力」というMCPのルールに従って、一貫したコンテキストを構築するメソッドを実装した。
わずか50行程度のコードで、再利用可能でテストしやすい、ボットの「脳」となるクラスを完成させた。
この小さなクラスが、今後の開発全体を支える、非常に重要な土台となります。コンテキスト管理のロジックをこのクラスにカプセル化したことで、これから作るWebサーバーやSlack連携のコードは、このエンジンの使い方を知っているだけでよく、複雑なmessages配列の管理から解放されます。
さて、賢い「脳」は手に入りました。しかし、それはまだ私たちのPCの中で眠っているだけです。
次なるステップは、この脳に「身体」を与え、Slackという広大な世界と繋げてあげることです。
連載第3回『FastAPIでボットサーバーを公開!Slackと連携して「@メンション」に応答させよう』 では、今回作ったMCPEngineを心臓部としてFastAPIサーバーに組み込みます。そして、Slack Events APIを設定し、ついにボットがSlack上での@メンションに反応して、最初の言葉を発する瞬間を迎えます。
シリーズの中で最もエキサイティングな「誕生」の瞬間です。ぜひ、お見逃しなく!