🤖

🤖 Slack × FastAPI で作る！社内 AI アシスタント基盤

2025/07/22に公開

Build an In-House AI Assistant Platform with Slack and FastAPI!

 1. 概要 (Overview)社内向け AI Slack ボットを作りたい！ 🤖
・・・ということは無いでしょうか？
特に社内ナレッジを扱う RAG 機能を備えた、業務利用の AI Slack ボットのニーズが多いと思います。
今回のこの記事では RAG 機能の実装までは触れませんが、それを組み込むためのベースとなる

AI Slack ボットの実装方法について、私が実装したものを元に紹介します。
アプリは簡単に言うと "ChatGPT の Slack 版" みたいなもので、ボットを AI にした Slack ボットです。
この記事では Python やアプリ開発にはある程度慣れた方向けに、

以下の具体的な実装の解説をします。
▶ LangChain を利用して OpenAI の AI モデルを扱う部分
▶ FastAPI と Slack Bolt により Slack ボットを構築する部分
▶ Slack Bolt から LLM を呼び出す部分
アプリの全ソースコードと実行のさせ方などは GitHub で公開しています。
https://github.com/Bubbles877/ai-slack-bot

 2. アプリの構成 (App Architecture)AI モデルは OpenAI、Azure OpenAI Service を利用
AI モデルの取り回しは LangChain を利用
Slack のインタラクションは Slack Bolt を利用
HTTP サーバーの構築は FastAPI を利用

 メッセージの流れ
 3. 実装例 (Implementation Example)※以降は Python 3.12.10、langchain 0.3.25、langchain-core 0.3.59、

　 langchain-openai 0.3.18、fastapi 0.115.12、slack-bolt 1.23.0、uvicorn 0.34.2、

　 redis 5.2.0 で動作を確認しています

 3.1. Slack アプリの構成Slack のサイトで Slack アプリを構成してください。 (参考: Bolt 入門ガイド | Bolt for Python)
以下のボットイベントを購読してください。
message.channels
message.groups
message.im
message.mpim
ボットトークンには以下のスコープが必要です。
channels:history
groups:history
im:history
mpim:history
chat:write
reactions:write
アプリレベルトークンには以下のスコープが必要です。
connections:write

 3.2. LangChain を利用して AI モデルを扱うLangChain を利用して OpenAI の AI モデルを扱う例です。
以下のようにモデルを初期化します。
from langchain_openai import ChatOpenAI
from pydantic import SecretStr

llm = ChatOpenAI(
    model="gpt-4.1-mini",                # LLM のモデル名
    base_url="https://api.openai.com/",  # OpenAI のエンドポイント
    api_key=SecretStr("***"),            # OpenAI の API のキー
    temperature=0.8,                     # LLM の出力の多様性
)
※パラメータは実践では .env で指定する想定です

※base_url を明示的に指定しない場合は OpenAI デフォルトの https://api.openai.com/ になります
Azure OpenAI Service の AI モデルの場合は以下を参考にしてください。

似たように書け、基底クラスの BaseChatModel で扱うことで、以降の処理は共通になります。
ai-slack-bot/util/llm_utils.py at main · Bubbles877/ai-slack-bot
次に以下のように LLM を呼び出します。
from langchain_core.messages import AIMessage, AnyMessage, HumanMessage, SystemMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

# システムプロンプト
instructions = "You are a helpful assistant."
# 会話履歴
history: list[AnyMessage] = [
    HumanMessage(content="What is the capital of France?"),
    AIMessage(content="The capital of France is Paris."),
]
# ユーザーメッセージ
user_message = "What is the capital of Japan?"

messages: list[AnyMessage] = [SystemMessage(content=instructions)]
messages.extend(history)
messages.append(HumanMessage(content=user_message))

# プロンプトと LLM、パーサーを連結する
prompt = ChatPromptTemplate.from_messages(messages)
chain = prompt | llm | StrOutputParser()

# LLM を呼び出す
response = chain.invoke({})
print(response)  # e.g. The capital of Japan is Tokyo.
[システムプロンプト + 会話履歴 + 新規ユーザーメッセージ] の組み合わせで LLM に渡して呼び出し、

その応答を取得しています。

 3.3. FastAPI と Slack Bolt により Slack ボットを構築する
 3.3.1. FastAPI と Slack Bolt の連携例えば以下のように FastAPI と Slack Bolt を連携できます。
from fastapi import FastAPI, Request, Response
from slack_bolt.adapter.fastapi.async_handler import AsyncSlackRequestHandler
from slack_bolt.async_app import AsyncApp, AsyncSay

# Slack Bolt アプリを初期化する
slack_app = AsyncApp(token="xoxb-***", signing_secret="***")
slack_request_handler = AsyncSlackRequestHandler(slack_app)


# Slack Bolt: message イベントのハンドリングを追加する
@slack_app.event("message")
async def handle_message(body: dict, say: AsyncSay) -> None:
    event: dict = body.get("event", {})
    user_id: str = event.get("user", "")

    if not user_id:
        return

    ts: str = event.get("ts", "")
    thread_ts: str = event.get("thread_ts", ts)

    res = f"Hello <@{user_id}>!"
    await say(text=res, thread_ts=thread_ts)


# FastAPI アプリを初期化する
server_app = FastAPI()


# FastAPI: Slack イベントのエンドポイントを追加する
@server_app.post("/slack/events")
async def handle_events(request: Request) -> Response:
    # リクエストを Slack のイベントハンドラーへ渡す
    return await slack_request_handler.handle(request)


# FastAPI: 必要に応じて他のエンドポイントを追加する
@server_app.get("/status")
async def handle_status(request: Request) -> dict:
    return {"status": "healthy"}
FastAPI により /slack/events への HTTP リクエストをルーティングして、

Slack Bolt のリクエストハンドラーに流します。

そして Slack Bolt 側でのイベント処理が呼ばれます。
ちなみに、Slack イベントのハンドリング、ルーティングは以下のように登録することもできます。

クラス化・メソッド化する場合にはこのやり方で対応できます。
slack_app.event("message")(handle_message)
server_app.add_api_route("/slack/events", handle_events, methods=["POST"])
server_app.add_api_route("/status", handle_status, methods=["GET"])

 3.3.2. Uvicorn による HTTP サーバーの管理Uvicorn を利用して、例えば以下のように FastAPI のアプリを実行できます。
import uvicorn
from fastapi import FastAPI, Request

from app.settings import Settings

server_app = FastAPI()


@server_app.get("/status")
async def handle_status(request: Request) -> dict:
    return {"status": "healthy"}


if __name__ == "__main__":
    # 注: Settings() の中では環境変数を読み込んでいると仮定します
    settings = Settings()

    # Uvicorn でサーバーアプリを起動する
    uvicorn.run(
        app="main:server_app",           # モジュール名:サーバーアプリのインスタンス名
        host="0.0.0.0",                  # 全てのネットワークインターフェースにバインド
        port=settings.port,              # ポート番号
        reload=settings.is_development,  # 開発時は自動リロードを有効
    )
これは例えば以下のように実行できます。
python app/main.py
Uvicorn のコマンドを利用する場合は、例えば以下のように実行できます。
uvicorn app.main:server_app --port 8000 --reload

 3.3.3. 本番のサーバー環境での実行クラウドにデプロイした場合など、本番のサーバー環境では Gunicorn を利用して、

例えば以下のように実行できます。
gunicorn "app.main:server_app" \
  --worker-class uvicorn.workers.UvicornWorker \
  --workers 4 \
  --log-level warning \
  --access-logfile - \
  --error-logfile -
※ポート番号を明示的に指定しない場合は、Gunicorn はデフォルトで環境変数 PORT を参照します
これは Gunicorn が 4 つの Uvicorn ワーカーを実行して負荷分散をしています。

 3.4. Slack Bolt から LLM を呼び出す
 3.4.1. Slack イベントのハンドリングと LLM の呼び出しSlack イベントのハンドリングから、例えば以下のように LLM を呼び出します。
bot_user_id: str = ""


async def setup() -> None:
    global bot_user_id

    # Slack ボット自身のユーザー ID を取得する
    res = await slack_app.client.auth_test()
    bot_user_id = res.get("user_id", "")


@slack_app.event("message")
async def handle_message(body: dict, say: AsyncSay) -> None:
    event: dict = body.get("event", {})
    user_id: str = event.get("user", "")
    bot_id: str = event.get("bot_id", "")

    if bot_id or not user_id:
        return

    ts: str = event.get("ts", "")
    thread_ts: str = event.get("thread_ts", ts)
    text: str = event.get("text", "")
    channel_id: str = event.get("channel", "")

    try:
        # 応答に時間が掛かることを想定し、先ず即時にリアクションだけする
        await slack_app.client.reactions_add(
            channel=channel_id, name="eyes", timestamp=ts
        )
    except Exception as e:
        print(f"Add reaction error: {e}")

    history = []

    if thread_ts != ts:
        # スレッドの場合はメッセージ履歴を取得する
        history = await get_thread_history(channel_id, thread_ts, ts)

    # LLM を呼び出す
    # 注: invoke_llm() の中では前述のように LLM を呼び出していると仮定します
    ai_res = await invoke_llm(user_message=text, history=history)
    await say(text=ai_res, thread_ts=thread_ts)


async def get_thread_history(channel_id: str, thread_ts: str, message_ts: str) -> list:
    history = []

    try:
        res = await slack_app.client.conversations_replies(
            channel=channel_id,
            ts=thread_ts,
            latest=message_ts,  # イベントで受け取ったメッセージは含めずに履歴を取得する
            limit=15,
        )

        msgs: list[dict] = res.get("messages", [])
        for msg in msgs:
            user_id: str = msg.get("user", "")
            bot_id: str = msg.get("bot_id", "")
            text: str = msg.get("text", "")

            if user_id == bot_user_id:
                role = "bot"
            elif bot_id:
                # 他の Slack ボット
                continue
            elif user_id:
                role = "user"

            history.append({"role": role, "content": text})

        return history
    except Exception:
        return history
Slack からメッセージを受信したら、スレッドの場合はメッセージ履歴を取得して、

その 新規メッセージ と スレッド内のメッセージ履歴 を LLM に渡して呼び出し、

LLM の応答を Slack に送り返しています。

 3.4.2. Redis を利用した複数ワーカー間での情報共有複数ワーカーで実行する場合、例えば Redis 利用して以下のように複数ワーカー間での情報共有ができます。
from redis.asyncio import Redis

# Redis クライアントを初期化する
redis_client = Redis.from_url("redis://localhost:6379/0")

# 監視スレッドのキーのプレフィックス
ACTIVE_THREAD_KEY_PREFIX = "slack_bot:active_thread:"
# スレッドの有効期限 (1 時間)
THREAD_TTL = 3600


async def add_active_thread(thread_ts: str) -> None:
    # 監視対象スレッドに登録する
    key = f"{ACTIVE_THREAD_KEY_PREFIX}{thread_ts}"
    await redis_client.setex(key, THREAD_TTL, "1")


async def is_active_thread(thread_ts: str) -> bool:
    # 監視対象スレッドかどうかを取得する
    key = f"{ACTIVE_THREAD_KEY_PREFIX}{thread_ts}"
    return await redis_client.exists(key) == 1
メンションの有無に関わらず、Slack ボットが参加しているチャンネルのメッセージは全て送られてきます。

私は以下のように Slack の "監視対象スレッド" を管理して、処理を効率化するようにしています。

message イベントをハンドリングする (app_mention イベントは見ない)
Slack ボットがメンションされている場合 or ダイレクトメッセージの場合、

監視対象スレッドに登録されていなければ登録する
監視対象スレッドの message イベントであれば返答する
それ以外は無視する
この "監視対象スレッド" の情報を、Redis を利用して複数ワーカー間で共有します。
具体的な実装については以下を参考にしてください。
ai-slack-bot/app/slack_bot.py at main · Bubbles877/ai-slack-bot
なお、Redis は Linux 環境では以下のようにインストール・起動ができます。
sudo apt update
sudo apt install redis-server
例えば Azure では Azure Cache for Redis や Redis Sidecar Extension (Azure App Service の拡張機能) が利用できます。

 4. まとめ (Conclusion)AI Slack ボットの実装方法について、私が実装したものを元に紹介しました。
LangChain を利用して OpenAI の AI モデルを扱い、

FastAPI と Slack Bolt により Slack ボットを構築し、

Slack Bolt から LLM の呼び出しをした、AI Slack ボットの作り方を解説しました。
これをベースに、例えば業務利用の特定ドメインの情報を扱う RAG 機能などを追加実装して行けます 😄

 関連・参考 (References)Bubbles877/ai-slack-bot: AI Slack Bot / AI Slack ボット
LangChain
Slack Bolt
FastAPI
Uvicorn
Gunicorn
Redis
お気軽にいいね 👍 とフォローよろです 🍌 基本フォロバしますので輪を拡げましょう

記事のリンク・引用自由 👌

1. 概要 (Overview)

2. アプリの構成 (App Architecture)

メッセージの流れ

3. 実装例 (Implementation Example)

3.1. Slack アプリの構成

3.2. LangChain を利用して AI モデルを扱う

3.3. FastAPI と Slack Bolt により Slack ボットを構築する

3.3.1. FastAPI と Slack Bolt の連携

3.3.2. Uvicorn による HTTP サーバーの管理

3.3.3. 本番のサーバー環境での実行

3.4. Slack Bolt から LLM を呼び出す

3.4.1. Slack イベントのハンドリングと LLM の呼び出し

3.4.2. Redis を利用した複数ワーカー間での情報共有

4. まとめ (Conclusion)

関連・参考 (References)

Discussion