Slack Bot + LangChain RAGチャットボット構築実践ガイド — 社内ドキュメント検索ボットを作る

• はじめに
• アーキテクチャ概要
• プロジェクト設定
  • 依存関係のインストール
  • 環境変数
  • プロジェクト構造
• ドキュメントのインデキシング
  • ドキュメントの読み込みとチャンキング
• RAGチェーンの構築
• Slack Bot連携
  • Slackアプリの設定
  • Slack Botの実装
• 最初に起動するとログに何が出るか
• 3秒ルール — このボットの構造的な問題
• スレッドに回答が付かずチャンネルに出る理由
• 回答が切れる場所 — 4,000字、3,000字、40,000字
• 質問一つが回答になるまで — 一度だけ最後まで
• Dockerデプロイ
• パフォーマンス最適化
  • エンベディングキャッシュ
  • 会話履歴（スレッドコンテキスト）
• 2026年8月時点のバージョン点検
• 失敗事例と落とし穴
• 使わないのはどんなときか
• まとめ
• 参考資料
• クイズ

はじめに

「Confluenceのデプロイ手順ドキュメントはどこだっけ？」 「Kubernetesクラスターへのアクセス方法はどうだっけ？」

こういった質問に毎回人が答える代わりに、社内ドキュメントを検索するAIチャットボットを作りましょう。LangChain + RAG（Retrieval-Augmented Generation）+ Slack Botの組み合わせで、実践的なプロダクションレベルのチャットボットを構築します。

アーキテクチャ概要

# インデキシングパイプライン（オフライン）
# ドキュメント → チャンキング → エンベディング → ベクトルDB（ChromaDB）

# クエリパイプライン（オンライン）
# Slackメッセージ → エンベディング → ベクトル検索 → LLM生成 → Slack応答

プロジェクト設定

依存関係のインストール

mkdir slack-rag-bot && cd slack-rag-bot

# 仮想環境
python -m venv .venv
source .venv/bin/activate

# 依存関係
pip install \
  langchain==0.2.16 \
  langchain-openai==0.1.25 \
  langchain-community==0.2.16 \
  chromadb==0.5.3 \
  slack-bolt==1.20.0 \
  python-dotenv==1.0.1 \
  unstructured==0.15.0 \
  tiktoken==0.7.0

環境変数

# .env
OPENAI_API_KEY=sk-xxx
SLACK_BOT_TOKEN=xoxb-xxx
SLACK_APP_TOKEN=xapp-xxx
SLACK_SIGNING_SECRET=xxx
CHROMA_PERSIST_DIR=./chroma_db
DOCS_DIR=./documents

プロジェクト構造

slack-rag-bot/
├── .env
├── main.py              # Slack Botエントリーポイント
├── indexer.py           # ドキュメントインデキシング
├── rag_chain.py         # RAGチェーン
├── config.py            # 設定
├── documents/           # 社内ドキュメント（Markdown、PDFなど）
│   ├── deployment-guide.md
│   ├── k8s-access.md
│   └── onboarding.pdf
└── chroma_db/           # ベクトルDBストレージ

ドキュメントのインデキシング

ドキュメントの読み込みとチャンキング

# indexer.py
import os
from pathlib import Path
from langchain_community.document_loaders import (
    DirectoryLoader,
    UnstructuredMarkdownLoader,
    PyPDFLoader,
    TextLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from dotenv import load_dotenv

load_dotenv()


def load_documents(docs_dir: str):
    """さまざまな形式のドキュメントを読み込む"""
    documents = []

    # Markdownファイル
    md_loader = DirectoryLoader(
        docs_dir,
        glob="**/*.md",
        loader_cls=UnstructuredMarkdownLoader,
        show_progress=True
    )
    documents.extend(md_loader.load())

    # PDFファイル
    pdf_loader = DirectoryLoader(
        docs_dir,
        glob="**/*.pdf",
        loader_cls=PyPDFLoader,
        show_progress=True
    )
    documents.extend(pdf_loader.load())

    # テキストファイル
    txt_loader = DirectoryLoader(
        docs_dir,
        glob="**/*.txt",
        loader_cls=TextLoader,
        show_progress=True
    )
    documents.extend(txt_loader.load())

    print(f"合計{len(documents)}件のドキュメントを読み込みました")
    return documents


def split_documents(documents):
    """ドキュメントをチャンクに分割"""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        length_function=len,
        separators=["\n## ", "\n### ", "\n\n", "\n", " ", ""]
    )

    chunks = text_splitter.split_documents(documents)
    print(f"合計{len(chunks)}個のチャンクを作成しました")
    return chunks


def create_vectorstore(chunks, persist_dir: str):
    """ベクトルDBを作成"""
    embeddings = OpenAIEmbeddings(
        model="text-embedding-3-small",
        chunk_size=500
    )

    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir,
        collection_metadata={"hnsw:space": "cosine"}
    )

    print(f"ベクトルDB作成完了: {persist_dir}")
    return vectorstore


def index_documents():
    """全インデキシングパイプライン"""
    docs_dir = os.getenv("DOCS_DIR", "./documents")
    persist_dir = os.getenv("CHROMA_PERSIST_DIR", "./chroma_db")

    # 読み込み → チャンキング → エンベディング → 保存
    documents = load_documents(docs_dir)
    chunks = split_documents(documents)
    vectorstore = create_vectorstore(chunks, persist_dir)

    return vectorstore


if __name__ == "__main__":
    index_documents()

# インデキシングの実行
python indexer.py

RAGチェーンの構築

# rag_chain.py
import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

load_dotenv()


class RAGChain:
    def __init__(self):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.vectorstore = Chroma(
            persist_directory=os.getenv("CHROMA_PERSIST_DIR", "./chroma_db"),
            embedding_function=self.embeddings
        )
        self.retriever = self.vectorstore.as_retriever(
            search_type="mmr",  # Maximum Marginal Relevance
            search_kwargs={
                "k": 5,
                "fetch_k": 20,
                "lambda_mult": 0.7
            }
        )
        self.llm = ChatOpenAI(
            model="gpt-4o-mini",
            temperature=0.1,
            max_tokens=2000
        )
        self.chain = self._build_chain()

    def _build_chain(self):
        """RAGチェーンを構成"""
        prompt = ChatPromptTemplate.from_messages([
            ("system", """あなたは社内ドキュメントベースのQ&Aアシスタントです。
以下のコンテキストに基づいて質問に回答してください。

ルール：
1. コンテキストにある情報のみを使用してください。
2. 確信がない場合は「関連ドキュメントが見つかりませんでした」と答えてください。
3. 回答に出典ドキュメントを含めてください。
4. コードやコマンドがある場合はコードブロックでフォーマットしてください。

コンテキスト：
{context}"""),
            ("human", "{question}")
        ])

        def format_docs(docs):
            formatted = []
            for i, doc in enumerate(docs):
                source = doc.metadata.get("source", "unknown")
                formatted.append(f"[ドキュメント {i+1}] ({source})\n{doc.page_content}")
            return "\n\n---\n\n".join(formatted)

        chain = (
            {"context": self.retriever | format_docs, "question": RunnablePassthrough()}
            | prompt
            | self.llm
            | StrOutputParser()
        )

        return chain

    def ask(self, question: str) -> dict:
        """質問に回答"""
        # 関連ドキュメントを検索
        docs = self.retriever.invoke(question)

        # LLM生成
        answer = self.chain.invoke(question)

        # 出典ドキュメント情報
        sources = list(set(
            doc.metadata.get("source", "unknown") for doc in docs
        ))

        return {
            "answer": answer,
            "sources": sources,
            "num_docs": len(docs)
        }

    def refresh_index(self):
        """インデックスの更新"""
        from indexer import index_documents
        self.vectorstore = index_documents()
        self.retriever = self.vectorstore.as_retriever(
            search_type="mmr",
            search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.7}
        )
        self.chain = self._build_chain()

Slack Bot連携

Slackアプリの設定

1. https://api.slack.com/apps で新しいアプリを作成
2. Socket Modeを有効化
3. Bot Token Scopesを追加：
   - app_mentions:read
   - chat:write
   - im:history
   - im:read
   - im:write
4. Event Subscriptionsを有効化：
   - app_mention
   - message.im
5. ワークスペースにインストール

Slack Botの実装

# main.py
import os
import logging
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from rag_chain import RAGChain
from dotenv import load_dotenv

load_dotenv()
logging.basicConfig(level=logging.INFO)

# Slack Appの初期化
app = App(token=os.environ["SLACK_BOT_TOKEN"])

# RAG Chainの初期化
rag = RAGChain()


@app.event("app_mention")
def handle_mention(event, say, client):
    """@メンションで質問を受ける"""
    user = event["user"]
    text = event["text"]
    channel = event["channel"]
    thread_ts = event.get("thread_ts", event["ts"])

    # ボットメンションを除去
    question = text.split(">", 1)[-1].strip()

    if not question:
        say(
            text="質問を入力してください！例：`@DocBot デプロイ手順を教えて`",
            thread_ts=thread_ts
        )
        return

    # ローディングメッセージ
    loading_msg = client.chat_postMessage(
        channel=channel,
        thread_ts=thread_ts,
        text=":mag: ドキュメントを検索しています..."
    )

    try:
        # RAGクエリ
        result = rag.ask(question)

        # レスポンスのフォーマット
        response = f"<@{user}>\n\n{result['answer']}"

        if result["sources"]:
            sources_text = "\n".join(f"• `{s}`" for s in result["sources"])
            response += f"\n\n:page_facing_up: *参考ドキュメント:*\n{sources_text}"

        # ローディングメッセージを更新
        client.chat_update(
            channel=channel,
            ts=loading_msg["ts"],
            text=response
        )

    except Exception as e:
        logging.error(f"RAG error: {e}")
        client.chat_update(
            channel=channel,
            ts=loading_msg["ts"],
            text=f"申し訳ございません。エラーが発生しました: {str(e)}"
        )


@app.event("message")
def handle_dm(event, say):
    """DMで質問を受ける"""
    if event.get("channel_type") != "im":
        return
    if event.get("bot_id"):
        return

    question = event["text"]

    try:
        result = rag.ask(question)

        response = result["answer"]
        if result["sources"]:
            sources_text = "\n".join(f"• `{s}`" for s in result["sources"])
            response += f"\n\n:page_facing_up: *参考ドキュメント:*\n{sources_text}"

        say(text=response)

    except Exception as e:
        say(text=f"エラーが発生しました: {str(e)}")


@app.command("/docbot-reindex")
def handle_reindex(ack, say):
    """スラッシュコマンドでインデックスを更新"""
    ack()
    say("インデックスを更新しています... :hourglass_flowing_sand:")

    try:
        rag.refresh_index()
        say("インデックスの更新が完了しました！ :white_check_mark:")
    except Exception as e:
        say(f"インデックスの更新に失敗しました: {str(e)}")


if __name__ == "__main__":
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    print("Slack RAG Bot started!")
    handler.start()

最初に起動するとログに何が出るか

アプリをワークスペースにインストールしてpython main.pyを実行したあとからが本当の始まりなのに、チュートリアルはたいていちょうどここで終わります。

Socket Modeのハンドラーはアプリレベルトークンでapps.connections.openを呼び出してWebSocketのURLを受け取り、そのソケットでイベントを受け取ります。公開URLもインバウンドポートも要りません。ドキュメントの表現そのままで「When using Socket Mode, your app does not need a Request URL to use the Events API.」です。

トークンが三つあって紛らわしいです。SLACK_BOT_TOKENはxoxb-、SLACK_APP_TOKENはxapp-で始まり、SLACK_SIGNING_SECRETはHTTPモードでのみ使います。アプリレベルトークンにconnections:writeがなければソケット自体が開かず、app_mentionを受け取るにはapp_mentions:readが、回答を上げるにはchat:writeが必要です。

# 実行
python main.py

# 自分たちで出したログがこの順に出れば正常です。
# （Boltが自前で出力する文言はバージョンごとに違うので基準にしないでください。）
[INFO] socket mode handler started
[INFO] event=app_mention channel=C123ABC456 user=U061F7AUR ts=1515449522.000016
[INFO] question='デプロイ手順を教えて' thread_ts=None
[INFO] retrieved 5 chunks in 0.42s
[INFO] llm answered in 6.1s len=1842
[INFO] chat_update ok

このログを出すには、リスナーの引数を一つ増やすだけで済みます。Boltはリスナー関数の引数を名前で注入するので、必要なものだけ書けばよいのです。

ack       Slackサーバーに確認応答を返す
say       関連づけられたチャンネルIDでchat.postMessageを呼び出す
respond   関連づけられたresponse_urlを利用する
body      パースされたリクエストボディ全体
payload   リクエストボディから中心的なデータだけを取り出したもの
event / message / command / action / shortcut / view / options
          各リスナーにおけるpayloadの別名
client    有効なトークンの入ったWebClientインスタンス
logger    ロガー
context   BoltContextインスタンス
next      ミドルウェアチェーンの次の段階へ

# 引数は名前で注入されます。順序も、全部受け取る必要もありません。
@app.event("app_mention")
def handle_mention(event, say, client, logger):
    logger.info(
        "event=app_mention channel=%s user=%s ts=%s thread_ts=%s",
        event["channel"],
        event["user"],
        event["ts"],
        event.get("thread_ts"),
    )

この一行があるだけで、後に出てくる問題のほとんどはログを見るだけで判別できます。とくに最後の項目、thread_tsがNoneと出るのか値が出るのかが、次の二つの節の核心です。

3秒ルール — このボットの構造的な問題

まず症状から。ボットが同じ質問に二回、ひどいときは四回答えます。しかも重複がリズムを刻みます。最初の重複はほぼ即座に、次はおよそ1分後、最後はおよそ5分後。この間隔を見た瞬間に原因が一つに絞られます。

Events APIのドキュメントはこう書いています。「Your app should respond to the event request with an HTTP 2xx within three seconds.」Boltのドキュメントはもっと直截です。「We recommend calling ack() right away before initiating any time-consuming processes… since you only have 3 seconds to respond before Slack registers a timeout error.」

RAGパイプラインは3秒では終わりません。クエリのエンベディング、ベクトル検索、LLMの生成。最後の一つだけでも普通は3秒から15秒です。この記事の構造はデフォルトのままにしておけばタイムアウトが出るようにできており、だから偶然のバグではなく構造的な問題です。

タイムアウトになるとSlackは再試行します。ドキュメントの表現そのままで「retrying a failed request up to 3 times in a gradually increasing timetable」であり、最初の再試行はほぼ即座に、二回目は1分後、最後は5分後です。再試行のリクエストにはx-slack-retry-numヘッダーが付き、値は「1」「2」「3」のいずれか、理由はx-slack-retry-reasonに入ります。もう受け取りたくなければ200以外の応答にx-slack-no-retry: 1を載せればよく、ドキュメントはこれを「we'll understand it to mean you'd rather this specific event not be re-delivered」と説明しています。

14:02:10.114  event=app_mention ts=1515449522.000016   回答生成の開始
14:02:10.140  event=app_mention ts=1515449522.000016   <- 再試行1
14:03:10.203  event=app_mention ts=1515449522.000016   <- 再試行2
14:08:10.377  event=app_mention ts=1515449522.000016   <- 再試行3

# tsが全部同じです。ユーザーは一度尋ね、私たちは四回答えました。

一つ押さえておきます。ドキュメントが必ずack()で確認応答しなければならないと明言している対象は、actions、commands、shortcuts、options requests、view submissionsです。イベントはその一覧になく、イベントの確認応答はフレームワークが処理します。だからといって3秒が消えるわけではありません。3秒はBoltではなくEvents API側のルールだからです。確実な方法は一つだけです。リスナーを即座に返し、実際の作業は別のスレッドで回すこと。

import threading

@app.event("app_mention")
def handle_mention(event, client, logger):
    # リスナーは即座に返します。RAGはここで回しません。
    threading.Thread(
        target=answer_in_background,
        args=(event, client, logger),
        daemon=True,
    ).start()


@app.command("/docbot-reindex")
def handle_reindex(ack, say):
    ack()  # スラッシュコマンドはドキュメントがack()を明示的に要求します。
    threading.Thread(target=reindex_in_background, args=(say,), daemon=True).start()

ここまでやっても、すでに出てしまった再試行は止められません。LLMが遅い日には依然として重複が来るので、もう一枚重ねます。

from collections import OrderedDict
import threading

_seen = OrderedDict()
_seen_lock = threading.Lock()


def already_handled(body) -> bool:
    """同じevent_idがまた来たらTrue。Slackの再送を濾し取ります。"""
    event_id = body.get("event_id")
    if not event_id:
        return False
    with _seen_lock:
        if event_id in _seen:
            return True
        _seen[event_id] = True
        while len(_seen) > 5000:
            _seen.popitem(last=False)
    return False


@app.event("app_mention")
def handle_mention(body, event, client, logger):
    if already_handled(body):
        logger.info("duplicate event_id=%s skipped", body.get("event_id"))
        return
    threading.Thread(
        target=answer_in_background, args=(event, client, logger), daemon=True
    ).start()

正直に書いておきます。再試行のスケジュールとヘッダー名はドキュメントに明記された動作ですが、event_idで濾し取る上のコードはBoltドキュメントの公式レシピではなく、私が使っている慣用句です。HTTPモードでx-slack-retry-numの付いたリクエストをまるごと飛ばす、より粗い方法についても同じです。そしてSocket Modeの再送が上の表とまったく同じに動くのかは、ドキュメントで確認できませんでした。event_id基準の防御は両方のモードで安全です。

プロセスを複数立てるなら、メモリ上の辞書では足りません。Redisのような共用ストレージにevent_idをTTLとともに入れてください。

スレッドに回答が付かずチャンネルに出る理由

app_mentionのペイロードはこういう形です。ドキュメントに載っている例そのままです。

{
  "type": "app_mention",
  "user": "U061F7AUR",
  "text": "<@U0LAN0Z89> is it everything a river should be?",
  "ts": "1515449522.000016",
  "channel": "C123ABC456",
  "event_ts": "1515449522000016"
}

thread_tsキーがないという点が重要です。チャンネルの最上位でメンションするとスレッドという概念自体がないのでキーがそもそも来ず、だからevent["thread_ts"]と書けばその場でKeyErrorになります。ログには例外だけが残り、ユーザー側には何の反応もありません。ボットが死んだようだ、という報告はたいていここから出ます。

逆に、すでにあるスレッドの中でメンションするとthread_tsに親メッセージのtsが入って来ます。だから慣用句はこの形に固まりました。

# 最上位のメンションなら、このメッセージ自身がスレッドの起点になります。
# スレッド内のメンションなら、親のtsがすでにthread_tsに入っています。
thread_ts = event.get("thread_ts") or event["ts"]

say(text=answer, thread_ts=thread_ts)

chat.postMessageのthread_tsのドキュメントはこう書いています。「Provide another message's ts value to make this message a reply. Avoid using a reply's ts value; use its parent instead.」上の一行がまさにそのルールを守ります。スレッドの中で返信自身のタイムスタンプをそのまま渡すと、返信にまたスレッドを掘ろうとすることになり、ユーザーが期待した場所に回答が付きません。元のコードのevent.get("thread_ts", event["ts"])も結果は同じです。

一つ明かしておくことがあります。sayにthread_tsを渡す正確な呼び出しの形を、Boltのドキュメントでそのまま見つけることはできませんでした。sayが「calls chat.postMessage API with the associated channel ID」だという説明と上のルールを合わせると出てくる慣用句です。

回答が切れる場所 — 4,000字、3,000字、40,000字

RAGの回答は長いです。出典の一覧まで付ければさらに長くなります。ところがSlackには互いに異なる上限が三つあります。

• chat.postMessage — 最良の結果のためにtextを4,000字に制限するよう勧めています。40,000字を超えるとSlackが切り捨てます。
• chat.update — エラーの文字列が明言します。「Message text is too long. The text field cannot exceed 4,000 characters.」markdown_textは12,000字までです。
• Block Kitのsectionブロック — textの最小の長さは1、最大は3,000字。fields配列の項目はそれぞれ2,000字です。

落とし穴は最後の行です。回答をきれいに見せようとsectionブロックに入れた瞬間、上限が3,000に下がります。ごく普通のRAGの回答が3,000字を超えることはとても多いので、テキストで送るときは平気だった回答が、見栄えをよくした瞬間から切れはじめます。

SECTION_LIMIT = 3000   # Block Kitのsectionブロックのtextの上限
TEXT_LIMIT = 4000      # chat.postMessage / chat.updateの推奨上限


def chunk_for_slack(answer: str, limit: int = TEXT_LIMIT) -> list[str]:
    """段落の境界で切ります。どこででも切るとコードブロックが壊れます。"""
    parts, buf = [], ""
    for para in answer.split("\n\n"):
        if len(buf) + len(para) + 2 > limit:
            if buf:
                parts.append(buf)
                buf = ""
            while len(para) > limit:
                parts.append(para[:limit])
                para = para[limit:]
            buf = para
        else:
            buf = f"{buf}\n\n{para}" if buf else para
    if buf:
        parts.append(buf)
    return parts

その次は呼び出しの頻度です。chat.postMessageはドキュメント上「generally allows posting one message per second per channel」であり、ワークスペース全体の上限も一緒にかかります。chat.updateはTier 3、つまり毎分50回以上です（ティアは1が毎分1回、2が20回、3が50回、4が100回以上）。ローディングメッセージを上げてchat.updateで差し替えるパターン自体は良いのですが、ストリーミングのように見せようとトークンごとに呼ぶとTier 3に正面からぶつかります。超えるとドキュメントの表現そのままで「Slack will return a HTTP 429 Too Many Requests error, and a Retry-After HTTP header containing the number of seconds until you can retry.」です。

症状    回答が文の途中でぷつりと切れる
確認    len(answer) をログに出す  ->  3214
診断    Block Kitのsectionブロック（3,000）に入れた。textフィールド（4,000）なら通った長さだ。

症状    回答は出るのに数秒後に更新が止まる
確認    応答コードとヘッダーを出す  ->  429 / Retry-After: 12
診断    chat.update を毎秒何度も呼んだ。Tier 3（毎分50回以上）を超えた。

Boltにはsay_streamやWebClient.chat_streamのようなストリーミング用の面もあります。ただし全部の引数と動作までは確認できていないので、正確なAPIは使用中のバージョンのドキュメントで確認してください。

質問一つが回答になるまで — 一度だけ最後まで

ここまでに出てきたものを一つのハンドラーにまとめるとこうなります。新しい概念はなく、順序がすべてです。

[チャンネル #dev-help]
ジウ     @DocBot ステージングのデプロイ手順を教えて        14:02:10

DocBot   :mag: ドキュメントを検索しています...             14:02:10
         （同じメッセージが6秒後に回答へ変わります）

DocBot   @ジウ                                             14:02:16
         ステージングのデプロイは次の順序です。
         1. mainブランチにマージ
         2. CIの通過を確認
         3. deploy-stagingワークフローを手動で実行

         :page_facing_up: 参考ドキュメント:
         - documents/deployment-guide.md
         - documents/ci-cd.md

# main.py — 確認応答 / 重複除去 / スレッド / 長さまで反映した形
import os
import threading
import logging

from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
from rag_chain import RAGChain

logging.basicConfig(level=logging.INFO)
app = App(token=os.environ["SLACK_BOT_TOKEN"])
rag = RAGChain()


def answer_in_background(event, client, logger):
    channel = event["channel"]
    thread_ts = event.get("thread_ts") or event["ts"]
    question = event["text"].split(">", 1)[-1].strip()
    logger.info("question=%r thread_ts=%s", question, event.get("thread_ts"))

    placeholder = client.chat_postMessage(
        channel=channel,
        thread_ts=thread_ts,
        text=":mag: ドキュメントを検索しています...",
    )
    try:
        result = rag.ask(question)
        body = f"<@{event['user']}>\n\n{result['answer']}"
        if result["sources"]:
            lines = "\n".join(f"- `{s}`" for s in result["sources"])
            body += f"\n\n:page_facing_up: *参考ドキュメント:*\n{lines}"

        parts = chunk_for_slack(body)
        client.chat_update(channel=channel, ts=placeholder["ts"], text=parts[0])
        for extra in parts[1:]:
            client.chat_postMessage(channel=channel, thread_ts=thread_ts, text=extra)
        logger.info("answered len=%d parts=%d", len(body), len(parts))
    except Exception:
        # 例外の文字列をそのままチャンネルに上げません。詳しくはログへ。
        logger.exception("rag failed")
        client.chat_update(
            channel=channel,
            ts=placeholder["ts"],
            text="回答を作れませんでした。しばらくしてからもう一度お試しください。",
        )


@app.event("app_mention")
def handle_mention(body, event, client, logger):
    if already_handled(body):
        return
    threading.Thread(
        target=answer_in_background, args=(event, client, logger), daemon=True
    ).start()


if __name__ == "__main__":
    SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"]).start()

同じ質問一回に対して、ログはこう残ります。

[INFO] event=app_mention channel=C123ABC456 user=U061F7AUR ts=1515449522.000016 thread_ts=None
[INFO] question='ステージングのデプロイ手順を教えて' thread_ts=None
[INFO] retrieved 5 chunks in 0.42s
[INFO] llm answered in 6.1s
[INFO] answered len=1842 parts=1

thread_ts=Noneなら最上位のメンションだという意味で、だからそのメッセージ自身がスレッドの起点になります。値が出ていればスレッドの中で尋ねたということで、回答もそのスレッドに付きます。この一行があれば、前の節の問題が起きたかどうかが即座に分かります。

時間を分けて見ることも重要です。検索が0.4秒で生成が6.1秒なら、ボトルネックは明らかに生成のほうです。検索が2秒を超えるならkやfetch_kから疑ってください。parts=1は回答が一度で入ったという意味で、2以上が多いならプロンプトで回答の長さを制限するほうがよいです。

Dockerデプロイ

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# インデキシング後にボットを起動
CMD ["python", "main.py"]

# docker-compose.yml
version: '3.8'

services:
  slack-rag-bot:
    build: .
    env_file: .env
    volumes:
      - ./documents:/app/documents
      - ./chroma_db:/app/chroma_db
    restart: unless-stopped

# ビルドと実行
docker compose up -d

# ログの確認
docker compose logs -f

パフォーマンス最適化

エンベディングキャッシュ

from langchain.storage import LocalFileStore
from langchain.embeddings import CacheBackedEmbeddings

store = LocalFileStore("./embedding_cache")
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    document_embedding_cache=store,
    namespace="text-embedding-3-small"
)

会話履歴（スレッドコンテキスト）

from langchain.memory import ConversationBufferWindowMemory

# スレッドごとのメモリ管理
thread_memories = {}

def get_memory(thread_ts: str) -> ConversationBufferWindowMemory:
    if thread_ts not in thread_memories:
        thread_memories[thread_ts] = ConversationBufferWindowMemory(
            k=5,
            memory_key="chat_history",
            return_messages=True
        )
    return thread_memories[thread_ts]

2026年8月時点のバージョン点検

この記事のコードは2026年3月時点のものです。いまそのままインストールすると半分は別の場所にあります。slack_boltは1.30.0が2026年7月15日に出ており、Python 3.7から3.14までをサポートします。LangChain側はlangchain 1.3.15（2026年8月11日）、langchain-core 1.5.5、langchain-classic 1.0.8、langchain-community 0.4.2です。

問題はlangchain-communityです。PyPIのバナーに「langchain-community is being sunset. See #674 for details and guidance.」が付き、イシューの本文は「We are making the decision to sunset the langchain-community package… This sunset will take effect immediately.」です。2026年5月22日付で終了日は明記されていないので、ありもしない締め切りで日程を組まないでください。v1のlangchainはagents、messages、tools、chat_models、embeddingsに減り、以前のチェーンとメモリはlangchain-classicへ行きました。

# 2026-08時点で入れ直すなら
pip install \
  "langchain==1.3.15" \
  "langchain-core==1.5.5" \
  "langchain-classic==1.0.8" \
  "langchain-openai" \
  "langchain-text-splitters" \
  "langchain-chroma>=0.1.2" \
  "slack-bolt==1.30.0" \
  "python-dotenv"

# この記事に出てくるimport -> 2026-08時点の場所
# langchain.memory.ConversationBufferMemory
#   -> langchain_classic.memory.buffer.ConversationBufferMemory
# langchain.embeddings.CacheBackedEmbeddings
#   -> langchain_classic.embeddings.cache.CacheBackedEmbeddings
# langchain.storage.LocalFileStore
#   -> langchain_classic.storage.file_system.LocalFileStore
# langchain.chains.RetrievalQA
#   -> langchain_classic.chains.retrieval_qa.base.RetrievalQA

# テキスト分割器は専用パッケージとして出ました。
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    add_start_index=True,
)

# Chromaも専用パッケージが正式です。
# (langchain_community.vectorstores.Chroma はcommunity 0.2.9からdeprecated)
from langchain_chroma import Chroma

# LCELの部品はlangchain-coreにそのまま残っています。
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.documents import Document

# 検索器のオプション整理
#   search_type   : 'similarity'(デフォルト) / 'mmr' / 'similarity_score_threshold'
#   search_kwargs : k(デフォルト4), score_threshold, fetch_k(デフォルト20),
#                   lambda_mult(デフォルト0.5), filter
#   この記事が使ったlambda_mult=0.7は多様性より関連度に重きを置いた値です。

一度マイグレーションしてみた人のほうが驚きます。v0.2とv0.3の頃に標準だと案内されたcreate_retrieval_chain、create_stuff_documents_chain、create_history_aware_retrieverも、v1ではlangchain-classicにあります。FAISSは事情がもっと悪いです。依然としてfrom langchain_community.vectorstores import FAISSであり、langchain-faissというパッケージはありません。日没が予告されたパッケージの中に、移住先なしに残った唯一の経路です。

LCELの部品はlangchain-coreにそのまま生きており、Runnableのドキュメントも「Any chain constructed this way will automatically have sync, async, batch, and streaming support.」と書いています。ただしv1のドキュメントから「LCEL」という名前は消えました。廃止されたのではなく、ドキュメントがもうその名前で教えていないだけです。

v1が前面に出すのはcreate_agentです。「create_agent is the standard way to build agents」であり、メモリはチェックポインターになりました。「To add short-term memory (thread-level persistence) to an agent, you need to specify a checkpointer when creating an agent.」

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="openai:gpt-5.5",
    tools=[search_internal_docs],
    checkpointer=InMemorySaver(),
)

# ここがSlackボットと正確に噛み合う地点です。
# スレッド一つ = 会話一つ。thread_tsをそのままthread_idとして使います。
thread_ts = event.get("thread_ts") or event["ts"]
thread_config = {"configurable": {"thread_id": thread_ts}}

対応がきれいです。thread_memoriesという辞書がやっていた仕事をチェックポインターが代わりに務め、キーはすでにSlackが渡してくれています。運用ではPostgresSaverのような永続チェックポインターを使ってください。

非同期についても一つ。VectorStoreの非同期メソッドは本当の非同期ではなくスレッドプールのラッパーです。

# langchain-coreのVectorStore実装の一部
async def asimilarity_search(self, query: str, k: int = 4, **kwargs: Any) -> list[Document]:
    return await run_in_executor(None, self.similarity_search, query, k=k, **kwargs)


# Boltの非同期側のインポート経路も一緒に書いておきます。
# from slack_bolt.app.async_app import AsyncApp          (短い別名: slack_bolt.async_app)
# from slack_bolt.adapter.socket_mode.async_handler import AsyncSocketModeHandler
#   (slack_bolt.adapter.socket_mode.aiohttp にもあります)
# await handler.start_async()

ソースのコメントはこれを「temporary workarounds」と呼び、「The proper solution is to make the similarity search asynchronous in the vector store implementations.」と書いてあります。langchain_chromaのChromaにはasync defが一つもないので、非同期に変えても検索が速くなるわけではありません。

最後にCacheBackedEmbeddings.from_bytes_storeの全引数の一覧は、現在のlangchain-classicのリファレンスでは確認できませんでした。そのページに作成中の表示が付いています。正確なAPIは使用中のバージョンのドキュメントで確認してください。

失敗事例と落とし穴

症状を先に書き、診断の順序を付けます。

ボットが自分の回答にまた答える — @app.event("message")ハンドラーはボットが上げたメッセージも受け取ります。ガードがなければ自分の回答を新しい質問として読むループが回ります。元のコードのif event.get("bot_id"): returnという二行がそのガードです。消さないでください。

メンション一回で回答が二回来る — 再試行ではなく二重購読の可能性があります。app_mentionとmessageを両方購読すると、チャンネルのメンション一つが二つのリスナーを起こします。ログにイベント名を出しておけば分かれます。同じ名前が二回なら再試行、異なる名前が一回ずつなら二重購読です。

インデキシングはできたはずなのに検索結果が空になる — 三つを順に見ます。インデキシングのときとクエリのときのエンベディングモデルが同じか（違えばベクトル空間が異なり類似度が無意味になります）、persist_directoryが同じ場所か（ボリュームを付けずコンテナ内の空ディレクトリを見ている場合がもっとも多いです）、コレクション名が同じか。この三つより先にチャンキングやプロンプトを触らないでください。

いつも見当違いのドキュメントを持ってくる — チャンクの境界を疑います。chunk_size=1000にchunk_overlap=200なら、表やコードブロックが途中で切れます。切れた断片は意味がないのに、エンベディングはきちんとできてしまいます。add_start_index=Trueを有効にしておけば、原文のどの位置で切れたのかがメタデータとして残ります。

再インデキシングのコマンドを呼ぶとボット全体が止まる — スラッシュコマンドも同じ3秒ルールの下にあります。元のコードがack()を先に呼ぶのは正しいのですが、全体の再インデキシングを同じスレッドで同期に回すのが問題です。3分かかるワークスペースなら、その3分のあいだに入ってきた質問は全部タイムアウトし、再試行され、ずいぶん後にまとめて回答として返ってきます。

再インデキシング中に入ってきた質問が変な回答を受け取る — refresh_index()がベクトルストア、検索器、チェーンを順に入れ替えているあいだ、他のスレッドがそのオブジェクトを読みます。新しいインデックスに古いチェーンが混ざります。一時的な場所に全部作ってから参照を一つだけ差し替えてください。

請求書が静かに大きくなる — k=5でチャンクが1,000字なら、質問ごとに5,000字以上がプロンプトに入ります。履歴まで付けばさらに増えます。ログにプロンプトの長さを出しておけば、月末ではなく当日に見えます。

例外の文字列がそのままチャンネルに貼られる — 元のコードは例外を文字列にしてメッセージに上げます。コネクション文字列やキーの断片が混ざっていれば、それが公開チャンネルに残ります。チャンネルには一般的な案内だけを、詳しい内容はログへ。

症状                             まず見るもの            その次
同じ回答が2〜4回                 event_idの重複          3秒超過 -> 再試行
メンション一つに回答が2回        ログのイベント名        app_mention + message の二重購読
検索結果が空になる               エンベディングモデル    persist_directory / コレクション名
回答が文の途中で切れる           回答の長さ              3,000(Block Kit) 対 4,000(text)
スレッドの外に回答が出る         thread_ts のログ値      返信の ts をそのまま使っていないか

# ask()がプロンプトの長さも一緒に返すよう一行だけ増やしておけば
# コストの問題は請求書ではなくログで先に見えます。
try:
    result = rag.ask(question)
    logger.info("prompt_chars=%d chunks=%d", result["prompt_chars"], result["num_docs"])
except Exception:
    logger.exception("rag failed question=%r", question[:200])
    client.chat_update(
        channel=channel,
        ts=placeholder["ts"],
        text="回答を作れませんでした。しばらくしてからもう一度お試しください。",
    )

使わないのはどんなときか

この組み合わせが答えにならない場合がはっきりとあります。

ドキュメントごとに閲覧権限が異なる場合。これがいちばん危険です。Slackボットは質問した人の権限ではなく、ボット自身のドキュメントアクセス権限で答えます。インデキシングのスクリプトが読めたドキュメントは全部ボットの知識になり、ワークスペースの誰でもメンション一回でその知識を取り出せます。人事評価のドキュメントのフォルダが一つ、誤ってドキュメントディレクトリに入っていれば、権限のない人も尋ねるだけで回答を受け取ります。ベクトルDBには原文の断片がそのまま保存されるので、要約しか出ないから大丈夫だ、という慰めも成り立ちません。

# 人がドキュメントを開くとき
ユーザー -> ドキュメントストア -> 権限の確認 -> 閲覧の許可または拒否

# ボットが答えるとき
ユーザー -> ボット -> ボットの権限ですでにインデキシングされたベクトルDB -> 回答
                  （質問した人が誰なのかはここで何の影響もありません）

権限が分かれるコーパスなら、インデックスを等級別に分けるか、検索の段階でユーザーごとのフィルターをかける設計が先です。それが負担なら、このボットには誰が見てもよいドキュメントだけを入れてください。

貼り付ければ済む分量。ドキュメントが十ページほどなら、ベクトルDBを立てる理由がありません。全部プロンプトに入れてください。インデキシングのパイプライン、再インデキシングのコマンド、ボリュームのマウント、エンベディングのコストが一度に消えます。RAGはコンテキストに全部入りきらないときに使う道具です。

Slackの検索のほうが得意な質問。あのスレッドはどこへ行ったのか、といった質問はRAGではなくSlack自身の検索の領域です。会話の記録はドキュメントではなく、エンベディングして要約するとむしろ文脈が飛びます。

Marketplaceに載せる予定がある場合。Socket Modeのドキュメントにはこう書かれています。「Apps using Socket Mode are not currently allowed in the public Slack Marketplace.」社内用なら問題ありませんが、製品として売るつもりなら最初からHTTPモードで行ってください。署名シークレットを渡してRequest URLを公開する形に変わりますが、あとから変えるのはなかなか面倒な決定です。

もう一つ。規程や法務のように間違うと困る領域では、回答を作らずに出典のリンクだけを返すボットのほうがよいです。コンテキストにある情報だけを使えと書いてもLLMはときどき破りますし、そのときどきが監査の対象なら、道具の形そのものを変えなければなりません。

まとめ

Slack RAGチャットボットのキーポイント：

1. ドキュメントチャンキング：RecursiveCharacterTextSplitterで意味単位の分割
2. ベクトル検索：MMR（Maximum Marginal Relevance）で多様なドキュメント検索
3. プロンプト：出典明記＋不確実な場合は正直に答えるよう設計
4. Slack連携：Socket Mode + app_mention/DMイベント処理
5. 再インデキシング：スラッシュコマンドでドキュメント更新を反映
6. 3秒ルール：リスナーは即座に返し、RAGは別スレッドで。再試行の重複はevent_idで防御
7. 長さの上限：テキストは4,000字、Block Kitのsectionブロックは3,000字。切れる場所が違います
8. バージョン：langchain-communityは日没が予告され、以前のチェーンはlangchain-classicへ移動

参考資料

すべて2026-08-16確認時点のものです。

• Bolt for Python — Socket Mode — 初期化とハンドラーの起動
• Bolt for Python — Acknowledging requests — 確認応答の推奨
• Bolt for Python — リスナーの引数 — 上の一覧の出典
• Slack Events API — 3秒ルール、再試行のスケジュールとヘッダー
• Using Socket Mode — Request URL不要、Marketplaceの制約
• app_mentionイベント — ペイロードの例
• Web API rate limits — ティアと429応答
• langchain-classicのリファレンス — 移った先のモジュールパス

古いアドレスのtools.slack.dev/bolt-python/はdocs.slack.dev/tools/bolt-python/へ301リダイレクトされます。社内Wikiに残っているなら直しておいてください。メッセージの長さの上限とBlock Kitのフィールドの仕様は、docs.slack.devの該当メソッドとブロックのドキュメントが正確です。

クイズ