LangChain + RAGでインテリジェントなTelegram FAQボットを作る：ドキュメントベースのQ&Aシステム

• はじめに
• バージョン確認から：この記事のimportはすでにレガシーです
• アーキテクチャ
• 環境設定
• ドキュメントの読み込みとインデックス作成
• インデックス作成を初めて回したとき、画面に何が出るか
• RAGチェーンの実装
• 検索できているのか、LLMが作っているのか
• Telegramボットの実装
• 4096字の壁、そしてタイピング表示
• 会話メモリは再起動すると消える
• asyncioの上でベクトル検索をawaitするということ
• Dockerでのデプロイ
• ドキュメントの自動更新
• パフォーマンス最適化
  • キャッシング
• 失敗事例と落とし穴
  • ボットが静かになりログにConflictが出る
  • 回答はもっともらしいのにドキュメントにない内容だ
  • 長い回答だけがまるごと消える
  • 再起動すると会話がリセットされる
  • Chromaをimportできない
  • FAISSだけ行き先がない
  • マイグレーションしたのにまたレガシーだと言われる
  • セキュリティ告知を見逃す
• 使わないのはどんなときか
• まとめ
• 参考資料
• クイズ

はじめに

ルールベースのチャットボットは事前に定義された質問にしか答えられませんが、RAG（Retrieval-Augmented Generation） ベースのチャットボットはドキュメントから関連情報を検索し、自然言語で回答します。この記事では、会社のFAQドキュメントに基づいて質問に答えるTelegramボットを構築します。

バージョン確認から：この記事のimportはすでにレガシーです

この記事のコードは、書いた当時は動いていました。2026年8月時点では、そのままコピーしてはいけません。python-telegram-botは22.8（2026-06-12）が最新で、Telegram Bot API 10.0をネイティブにサポートします。公式ドキュメントはv20.0からPythonのasyncioの上に載っていると書いており、この事実が後で足を引っ張ります。

LangChainはlangchain 1.3.15（2026-08-11）、langchain-core 1.5.5、langchain-classic 1.0.8、langchain-community 0.4.2です。もっとも大きな変化はlangchain-communityのサンセットです。PyPIには「langchain-community is being sunset」というバナーが付き、イシューの本文は「This sunset will take effect immediately」と書いています。日付は2026-05-22で、EOLの日付は明示されていないので、消える時点を推測してスケジュールを立てないでください。v1のlangchainパッケージにはagents、messages、tools、chat_models、embeddingsだけが残り、この記事が使うチェーンとメモリはlangchain-classicへ引っ越しました。

# この記事のimport → 2026-08時点の場所
langchain.chains.ConversationalRetrievalChain
  → langchain_classic.chains.conversational_retrieval.base.ConversationalRetrievalChain
    (0.1.17からdeprecated)
langchain.chains.RetrievalQA
  → langchain_classic.chains.retrieval_qa.base.RetrievalQA
langchain.memory.ConversationBufferMemory
  → langchain_classic.memory.buffer.ConversationBufferMemory
langchain.text_splitter.RecursiveCharacterTextSplitter
  → langchain_text_splitters.RecursiveCharacterTextSplitter
langchain_community.vectorstores.Chroma
  → langchain_chroma.Chroma   (community 0.2.9からdeprecated)
langchain_community.vectorstores.FAISS
  → 移行先はない。langchain-faissというパッケージは存在しない。

v0.2とv0.3の頃にレガシーチェーンの代替として案内されたcreate_retrieval_chain、create_stuff_documents_chain、create_history_aware_retrieverも、v1ではlangchain-classicにあります。一度マイグレーションした人ほど驚くところです。v1が前面に出すのはcreate_agentで、短期メモリはMemoryオブジェクトではなくチェックポインターです。

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

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

thread_config = {"configurable": {"thread_id": "1"}}
response = agent.invoke(
    {"messages": [{"role": "user", "content": "Hi! My name is Bob."}]},
    thread_config,
)

本番ではInMemorySaverの代わりにPostgresSaverを使います。Telegramでこの構造がよく合う理由は、thread_idの位置にチャットIDをそのまま入れればよいからです。インストールはこうなります。

pip install "python-telegram-bot[rate-limiter]==22.8" \
  langchain langchain-classic langchain-openai \
  langchain-chroma langchain-text-splitters \
  chromadb tiktoken pypdf docx2txt

以下のコードは元の形のまま残しました。レガシーの経路はまだ動いており、メンテナンス中のコードベースのほとんどがこの形だからです。

アーキテクチャ

ユーザーの質問
    ↓
Telegram Bot API
    ↓
LangChain RAG Pipeline
    ├── 1. Query Embedding (OpenAI)
    ├── 2. Vector Search (ChromaDB)
    ├── 3. Context Retrieval (Top-K)
    └── 4. LLM Generation (GPT-4o)
    ↓
回答 + ソース表示

環境設定

pip install langchain langchain-openai langchain-community \
  chromadb python-telegram-bot tiktoken \
  pypdf docx2txt unstructured

# config.py
import os

TELEGRAM_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]

# RAG設定
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 200
TOP_K = 4
MODEL_NAME = "gpt-4o"
EMBEDDING_MODEL = "text-embedding-3-small"

ドキュメントの読み込みとインデックス作成

# indexer.py
from langchain_community.document_loaders import (
    DirectoryLoader,
    PyPDFLoader,
    TextLoader,
    Docx2txtLoader,
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

def load_documents(docs_dir: str):
    """様々な形式のドキュメントを読み込む"""
    loaders = {
        "**/*.pdf": PyPDFLoader,
        "**/*.txt": TextLoader,
        "**/*.md": TextLoader,
        "**/*.docx": Docx2txtLoader,
    }

    all_docs = []
    for glob_pattern, loader_cls in loaders.items():
        loader = DirectoryLoader(
            docs_dir,
            glob=glob_pattern,
            loader_cls=loader_cls,
            show_progress=True,
        )
        docs = loader.load()
        all_docs.extend(docs)
        print(f"Loaded {len(docs)} docs from {glob_pattern}")

    return all_docs

def create_vector_store(docs_dir: str, persist_dir: str = "./chroma_db"):
    """ドキュメントをチャンクに分割し、ベクトルストアに保存"""
    # ドキュメントの読み込み
    documents = load_documents(docs_dir)
    print(f"Total documents: {len(documents)}")

    # テキスト分割
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200,
        separators=["\n\n", "\n", ".", "!", "?", ",", " "],
    )
    chunks = text_splitter.split_documents(documents)
    print(f"Total chunks: {len(chunks)}")

    # エンベディング生成 & ベクトルストアに保存
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir,
        collection_metadata={"hnsw:space": "cosine"},
    )

    print(f"Vector store created at {persist_dir}")
    return vectorstore

if __name__ == "__main__":
    create_vector_store("./docs")

インデックス作成を初めて回したとき、画面に何が出るか

indexer.pyを初めて回すと、こういう出力が出ます。数字はドキュメントごとに違いますが、形は同じです。

$ python indexer.py
100%|█████████████████████████| 12/12 [00:04<00:00,  2.71it/s]
Loaded 47 docs from **/*.pdf
100%|█████████████████████████| 31/31 [00:00<00:00, 240.11it/s]
Loaded 31 docs from **/*.txt
Loaded 9 docs from **/*.md
Loaded 3 docs from **/*.docx
Total documents: 90
Total chunks: 412
Vector store created at ./chroma_db

見るべき数字は三つです。一つ目はTotal documentsです。PyPDFLoaderはページ単位でDocumentを作るので、PDFを12個入れて47が出れば正常で、12が出たらページ分割ができていません。二つ目はTotal chunksです。chunk_size 1000にオーバーラップ200なら、チャンク一つが消化する新しいテキストは800字ほどなので、全体の文字数を800で割った値と近くなるはずです。ドキュメント90個にチャンク95個なら、ほとんどが1000字未満でチャンキングが何もしていません。三つ目はchroma_dbディレクトリのサイズです。数十KBにとどまるなら、エンベディングが静かに失敗したか、チャンクが空です。スキャンしたPDFのようにテキストレイヤーがないファイルは、ローダーが成功してpage_contentだけが空文字列になるのですが、このときインデックス作成は最後までエラーなく回ります。

ボットを起動する前に、検索だけを別に確認するスクリプトを回してください。

# smoke_test.py — ボットなしで検索だけを確認する
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
store = Chroma(
    embedding_function=embeddings,
    persist_directory="./chroma_db",
)

QUERIES = [
    "年次有給休暇はいつから使えますか",   # ドキュメントにある質問
    "リモートワークの申請手順",           # ドキュメントにある質問
    "会社の犬の名前は何ですか",           # ドキュメントにない質問 = 陰性対照
]

for q in QUERIES:
    docs = store.similarity_search(q, k=3)
    print(f"\nQ: {q}  -> {len(docs)} hits")
    for d in docs:
        src = d.metadata.get("source", "?")
        print(f"   {src}: {d.page_content[:60]}...")

出力はこういう形です。

Q: 年次有給休暇はいつから使えますか  -> 3 hits
   docs/hr-policy.pdf: 年次有給休暇は、入社日から1年間で80パーセント以上出勤した労働者に...
   docs/hr-policy.pdf: 年次有給休暇の取得促進制度により、未消化の休暇は毎年12月31日を基準に...
   docs/onboarding.md: 入社初年度は、1か月皆勤ごとに1日の有給休暇が発生し...

Q: 会社の犬の名前は何ですか  -> 3 hits
   docs/office-guide.md: オフィスの入館カードは1階の受付で受け取り、紛失した場合は...
   docs/hr-policy.pdf: 慶弔休暇は次の基準で付与します。本人の結婚は5日...
   docs/onboarding.md: Slackチャンネルの案内は次のとおりです。全社のお知らせはgeneralチャンネル...

三つ目のクエリが、このスクリプトのすべてです。ドキュメントにない質問を入れても、similarity_searchはいつもk個を返します。類似度がどれだけ低くても、いちばん近いものを埋めて返します。検索ができているかを確認するには、できるケースではなくできないケースを見なければならない理由がこれです。上の出力の二つ目のブロックに出た三つのチャンクは質問と何の関係もないのに、そのままLLMのコンテキストに入ります。ボットが作り出す回答のかなりの部分は、生成の段階ではなくこの地点ですでに決まっています。

RAGチェーンの実装

# rag_chain.py
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferWindowMemory
from langchain.prompts import ChatPromptTemplate, SystemMessagePromptTemplate, HumanMessagePromptTemplate

SYSTEM_PROMPT = """あなたは会社のFAQアシスタントです。提供されたコンテキストに基づいて質問に回答してください。

ルール：
1. コンテキストにある情報のみを使用してください。
2. 確信が持てない場合は「提供されたドキュメントにはその情報が見つかりません」と答えてください。
3. 回答の最後に参照したドキュメントのソースを表示してください。
4. 簡潔かつ明確に回答してください。

コンテキスト：
{context}"""

def create_rag_chain(persist_dir: str = "./chroma_db"):
    """RAGチェーンを作成"""
    # ベクトルストアの読み込み
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma(
        persist_directory=persist_dir,
        embedding_function=embeddings,
    )

    # リトリーバーの設定
    retriever = vectorstore.as_retriever(
        search_type="mmr",  # Maximal Marginal Relevance
        search_kwargs={
            "k": 4,
            "fetch_k": 10,
            "lambda_mult": 0.7,
        },
    )

    # LLM
    llm = ChatOpenAI(
        model="gpt-4o",
        temperature=0.1,
        max_tokens=1024,
    )

    # 会話メモリ（直近5ターン）
    memory = ConversationBufferWindowMemory(
        k=5,
        memory_key="chat_history",
        return_messages=True,
        output_key="answer",
    )

    # プロンプト
    prompt = ChatPromptTemplate.from_messages([
        SystemMessagePromptTemplate.from_template(SYSTEM_PROMPT),
        HumanMessagePromptTemplate.from_template("{question}"),
    ])

    # チェーンの作成
    chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=retriever,
        memory=memory,
        return_source_documents=True,
        combine_docs_chain_kwargs={"prompt": prompt},
        verbose=False,
    )

    return chain

class RAGBot:
    """ユーザーごとの会話コンテキストを管理するRAGボット"""

    def __init__(self, persist_dir: str = "./chroma_db"):
        self.persist_dir = persist_dir
        self.user_chains: dict[int, ConversationalRetrievalChain] = {}

    def get_chain(self, user_id: int):
        """ユーザーごとのチェーン（会話メモリを分離）"""
        if user_id not in self.user_chains:
            self.user_chains[user_id] = create_rag_chain(self.persist_dir)
        return self.user_chains[user_id]

    async def ask(self, user_id: int, question: str) -> tuple[str, list[str]]:
        """質問に回答し、ソースを返す"""
        chain = self.get_chain(user_id)
        result = chain.invoke({"question": question})

        answer = result["answer"]
        sources = []
        for doc in result.get("source_documents", []):
            source = doc.metadata.get("source", "Unknown")
            page = doc.metadata.get("page", "")
            if page:
                sources.append(f"{source} (p.{page})")
            else:
                sources.append(source)

        # 重複を除去
        sources = list(dict.fromkeys(sources))
        return answer, sources

    def reset_memory(self, user_id: int):
        """ユーザーの会話メモリをリセット"""
        if user_id in self.user_chains:
            del self.user_chains[user_id]

検索できているのか、LLMが作っているのか

ボットを立ち上げれば、とにかく回答は出ます。その回答がドキュメントから出たものなのか、モデルがもともと知っていたものなのかは、チャット画面を見るだけでは区別できません。三つを有効にしておけば区別がつきます。

一つ目、空の結果を作れるようにします。as_retrieverのsearch_typeはデフォルトのsimilarity、mmr、similarity_score_thresholdの三つだけです。前の二つは決して空のリストを返さず、三つ目だけが関連ドキュメントなしを表現できます。

retriever = store.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"k": 4, "score_threshold": 0.5},
)

docs = retriever.invoke(question)
if not docs:
    # LLMをそもそも呼び出さない。ここで終えるのが正解だ。
    return "提供されたドキュメントにはその情報が見つかりません。", []

しきい値はマジックナンバーではありません。エンベディングモデルと距離関数によって、同じ0.5が違う意味になります。先のスモークテストの陰性対照のスコアと、正常な質問のスコアの間で選んでください。スコアが距離なのか類似度なのかも実装ごとに違います。正確なAPIは、使用中のバージョンのドキュメントで確認してください。

二つ目、検索されたチャンクをログに残します。

logger.info(
    "retrieval chat_id=%s q=%r hits=%d sources=%s",
    chat_id,
    question[:80],
    len(docs),
    [d.metadata.get("source") for d in docs],
)

報告を受けたとき、まっさきに見る行です。hitsが0なのに回答が出ているならプロンプトのガードが破られたということで、hitsが4なのにsourcesが全部見当違いなら、それは検索の問題でLLMの問題ではありません。この区別がつかないと、プロンプトだけを何週間も直すことになります。

三つ目、検索専用のコマンドを付けます。CommandHandler(command, callback, filters=None, block=True, has_args=None)で、コマンドの後ろに付いた引数はCallbackContext.argsに入ります。

async def debug(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query = " ".join(context.args)
    if not query:
        await update.message.reply_text("使い方: /debug 検索語")
        return

    docs = retriever.invoke(query)
    if not docs:
        await update.message.reply_text("検索結果なし（しきい値未満）")
        return

    lines = []
    for i, d in enumerate(docs, 1):
        src = d.metadata.get("source", "?")
        lines.append(f"{i}. {src}\n   {d.page_content[:120]}")
    await update.message.reply_text("\n".join(lines))

application.add_handler(CommandHandler("debug", debug))

LLMを通さずに検索結果だけを見せるコマンドです。回答が間違っているという報告の文をそのまま入れてみれば、検索が間違ったのか生成が間違ったのかが一度で分かれます。

最後にソース表示についての誤解を一つ。ソースが付いているというのは、そのチャンクがコンテキストに入ったという意味であって、回答の文がそこから出たという保証ではありません。逆に、ソースが空なのに自信のある回答が出たなら、それはドキュメントではなくモデルが話しています。

Telegramボットの実装

# bot.py
import logging
from telegram import Update, BotCommand
from telegram.ext import (
    Application,
    CommandHandler,
    MessageHandler,
    filters,
    ContextTypes,
)
from rag_chain import RAGBot
from config import TELEGRAM_TOKEN

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

rag_bot = RAGBot()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """開始コマンド"""
    welcome = (
        "こんにちは！FAQアシスタントです。\n\n"
        "お気軽にご質問ください。\n"
        "会社のドキュメントに基づいてお答えします。\n\n"
        "コマンド：\n"
        "/reset - 会話をリセット\n"
        "/sources - 検索可能なドキュメント一覧"
    )
    await update.message.reply_text(welcome)

async def reset(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """会話メモリをリセット"""
    user_id = update.effective_user.id
    rag_bot.reset_memory(user_id)
    await update.message.reply_text("会話がリセットされました。")

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """一般メッセージの処理"""
    user_id = update.effective_user.id
    question = update.message.text

    # タイピング表示
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id,
        action="typing"
    )

    try:
        answer, sources = await rag_bot.ask(user_id, question)

        # 回答のフォーマット
        response = answer
        if sources:
            response += "\n\n参考ドキュメント：\n"
            for src in sources[:3]:
                response += f"  - {src}\n"

        await update.message.reply_text(response)

    except Exception as e:
        logger.error(f"Error: {e}")
        await update.message.reply_text(
            "申し訳ありません。回答の生成中にエラーが発生しました。"
        )

async def post_init(application: Application):
    """ボット起動時にコマンドを登録"""
    commands = [
        BotCommand("start", "ボットを開始"),
        BotCommand("reset", "会話をリセット"),
        BotCommand("sources", "検索可能なドキュメント一覧"),
    ]
    await application.bot.set_my_commands(commands)

def main():
    app = Application.builder().token(TELEGRAM_TOKEN).post_init(post_init).build()

    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("reset", reset))
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    logger.info("Bot started")
    app.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

4096字の壁、そしてタイピング表示

RAGボットの回答は長くなります。チャンクを四つ入れて根拠を挙げて説明しろと指示すれば、2,000字は簡単に超えます。ところがtelegram.constants.MessageLimit.MAX_TEXT_LENGTHは4096です。超えると送信が失敗しますが、どの例外が上がるのかは断定しません。telegram.error.BadRequestの説明はリクエストを処理できなかったという文だけで、長さへの言及はありません。正確な例外の型は、使用中のバージョンのドキュメントで確認してください。重要なのは例外の名前ではなく、あらかじめ切っておくことです。

from telegram.constants import MessageLimit

def split_for_telegram(text: str, limit: int = MessageLimit.MAX_TEXT_LENGTH) -> list[str]:
    """段落の境界 -> 行の境界 -> 強制切断の順に切る。"""
    if len(text) <= limit:
        return [text]

    parts: list[str] = []
    buf = ""
    for block in text.split("\n\n"):
        if len(block) > limit:
            for line in block.split("\n"):
                while len(line) > limit:
                    parts.append(line[:limit])
                    line = line[limit:]
                if len(buf) + len(line) + 1 > limit:
                    parts.append(buf)
                    buf = line
                else:
                    buf = f"{buf}\n{line}" if buf else line
            continue
        if len(buf) + len(block) + 2 > limit:
            parts.append(buf)
            buf = block
        else:
            buf = f"{buf}\n\n{block}" if buf else block
    if buf:
        parts.append(buf)
    return parts

async def reply_long(update: Update, text: str) -> None:
    for part in split_for_telegram(text):
        await update.message.reply_text(part)

段落の境界を優先する理由があります。コードブロックが混じった回答を4096字で機械的に切ると、開きのバッククォートと閉じのバッククォートが別のメッセージに分かれ、parse_modeを有効にしていればフォーマットのエラーで送信がまるごと失敗します。切れて届くのではなく、そもそも届きません。

タイピング表示は、RAGボットでは選択肢ではありません。エンベディングとベクトル検索にLLMの生成まで足すと、体感の遅延は3秒から10秒です。値は文字列ではなく定数を使ってください。パスはtelegram.constants.ChatActionで、TYPING、UPLOAD_PHOTO、UPLOAD_DOCUMENT、CHOOSE_STICKER、FIND_LOCATIONなどがあります。

from telegram.constants import ChatAction

await context.bot.send_chat_action(
    chat_id=update.effective_chat.id,
    action=ChatAction.TYPING,
)

注意すべきは持続時間です。Telegramのドキュメントの表現では「The status is set for 5 seconds or less」です。生成に12秒かかれば、ユーザーは7秒のあいだ何の反応もない画面を見ます。生成が終わるまで4秒間隔で呼び直すバックグラウンドタスクを置いてください。send_chat_actionの正確なパラメータ一覧は、使用中のバージョンのドキュメントで確認してください。レートリミットも付いてきます。Telegramボットのよくある質問の文です。

# https://core.telegram.org/bots/faq
"In a single chat, avoid sending more than one message per second."
"In a group, bots are not be able to send more than 20 messages per minute."
"For bulk notifications, bots are not able to broadcast more than about
 30 messages per second, unless they enable paid broadcasts."

4096字の分割とレートリミットはくっついた問題です。長い回答を四つに切って続けて送れば1秒に四つ投げることになり、単一チャットの推奨を超え、グループなら毎分20個にも引っかかります。ユーザーの多いボットではなく回答の長いボットが先に引っかかるところが、RAGボットの変わった点です。telegram.ext.AIORateLimiterのデフォルト値はoverall_max_rate=30、overall_time_period=1、group_max_rate=20、group_time_period=60、max_retries=0で、上の数字をそのまま反映しています。別途extraのインストールが必要です。

pip install "python-telegram-bot[rate-limiter]"

from telegram.ext import AIORateLimiter, ApplicationBuilder

application = (
    ApplicationBuilder()
    .token(TELEGRAM_TOKEN)
    .rate_limiter(AIORateLimiter())
    .concurrent_updates(True)
    .build()
)

max_retriesのデフォルト値が0であることは覚えておいてください。デフォルトの設定ではRetryAfterを受け取っても再試行しません。RetryAfterにはretry_after属性があるので、値を上げるか自分で処理する必要があります。ちなみにApplication.builder()は静的メソッドでApplicationBuilderを返し、tokenとbotは排他的です。

会話メモリは再起動すると消える

上のRAGBotはuser_chainsという辞書にユーザーごとのチェーンを持っています。プロセスのメモリなのでデプロイ一度で全部飛び、ユーザーの数だけオブジェクトが積み上がるのに、片付けるコードがありません。社内の100人なら大丈夫ですが、オープンなチャンネルに付ければそのままリークです。

ライブラリがすでにこの場所を用意しています。context.user_dataはユーザーIDごとにマッピングされる辞書で、chat_dataはチャットID基準、bot_dataはボット全体で一つです。

ここで設計の選択が一つ生まれます。1対1の会話だけを受けるならuser_dataとchat_dataは同じですが、グループに招待された瞬間に分かれます。user_dataは同じ人が複数のグループで文脈をつないでいく形で、chat_dataはグループ全体が一つの文脈を共有する形です。社内のFAQボットなら、たいていchat_dataが自然です。

デフォルトではこれもプロセスのメモリです。ディスクに下ろすにはPicklePersistence(filepath, store_data=None, single_file=True, on_flush=False, update_interval=60, context_types=None)を付けます。user_data、chat_data、bot_data、callback_data、conversationsを保存します。

from telegram.ext import ApplicationBuilder, PicklePersistence

persistence = PicklePersistence(filepath="bot_state.pickle")

application = (
    ApplicationBuilder()
    .token(TELEGRAM_TOKEN)
    .persistence(persistence)
    .build()
)

Dockerならbot_state.pickleは必ずボリュームの上になければなりません。イメージの中にあるとコンテナを作り直すたびに初期化され、付けたつもりの永続性がないのと同じになります。下のdocker-compose.ymlのchroma-dataボリュームの横にもう一つ追加してください。

v1スタイルなら、この場所をチェックポインターが代わりに務め、thread_idにチャットIDをそのまま入れます。

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

# 本番ではInMemorySaverの代わりにPostgresSaverを使うようドキュメントが案内している。
# 正確なimportのパスは、使用中のバージョンのドキュメントで確認すること。
agent = create_agent(model="openai:gpt-5.5", tools=[], checkpointer=InMemorySaver())

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    config = {"configurable": {"thread_id": str(update.effective_chat.id)}}
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": update.message.text}]},
        config,
    )
    # 戻り値の構造はバージョンによって違う。正確なAPIは使用中のバージョンのドキュメントで確認すること。
    await reply_long(update, extract_answer(result))

この数行が、RAGBotのget_chain、user_chains、reset_memoryを全部代わりに務めます。

asyncioの上でベクトル検索をawaitするということ

python-telegram-botはv20からasyncioベースです。イベントループ一つがすべてのアップデートを処理するので、ハンドラーでブロッキング呼び出しをすればボット全体が止まります。ここまでは見慣れた話です。

問題は、awaitを付けたからといって安心できない点です。LangChainのVectorStoreの非同期メソッドは、ほとんどが本当の非同期ではなくスレッドプールのラッピングです。langchain-coreのソースはこういう形をしています。

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)

ソースのコメントはこれらのメソッドを「temporary workarounds」と呼び、まともな解法はベクトルストアの実装そのものが非同期になることだと書いてあります。そしてlangchain_chromaのChromaにはasync defが一つもありません。同期メソッドだけを定義し、aで始まるメソッドは全部ベースから継承しています。

なのでawait retriever.ainvoke(question)はイベントループを手放しているように見えて、実際にはスレッドプールのワーカーを一つ占有します。ワーカーの数は有限で、同時の質問がその大きさを超えるとキューイングが始まります。ユーザーの目にはときどき遅くなると映り、原因をLLM APIの遅延と誤解しやすくなります。

同じ文脈の間違いがもう一つあります。run_pollingとrun_webhookはコルーチンではなく、ブロッキングの同期メソッドです。async def mainの中でawaitを付けても動かず、普通のdef mainで呼び出さなければなりません。上の元のコードがdef mainになっているのが正しいです。

run_polling(poll_interval=0.0, timeout=datetime.timedelta(seconds=10),
            bootstrap_retries=0, allowed_updates=None, drop_pending_updates=None,
            close_loop=True, stop_signals=None)

run_webhook(listen='127.0.0.1', port=80, url_path='', cert=None, key=None,
            bootstrap_retries=0, webhook_url=None, allowed_updates=None,
            drop_pending_updates=None, ip_address=None, max_connections=40,
            close_loop=True, stop_signals=None, secret_token=None, unix=None)

デフォルトの設定ではアップデートが順番に処理されます。前のユーザーのRAG呼び出しが終わってから次の順番が来ます。concurrent_updatesを有効にすれば同時処理になりますが、有効にした瞬間に上のスレッドプールの問題と前の節のレートリミットが同時に現実になります。三つを別々に決めず、一緒に見てください。

Dockerでのデプロイ

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

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

COPY . .

# ドキュメントのインデックス作成
RUN python indexer.py

CMD ["python", "bot.py"]

# docker-compose.yml
services:
  faq-bot:
    build: .
    environment:
      - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./docs:/app/docs
      - chroma-data:/app/chroma_db
    restart: unless-stopped

volumes:
  chroma-data:

docker-compose up -d

ドキュメントの自動更新

# watcher.py - ドキュメント変更を検知して自動再インデックス
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class DocChangeHandler(FileSystemEventHandler):
    def __init__(self, indexer_fn):
        self.indexer_fn = indexer_fn
        self.last_indexed = 0

    def on_modified(self, event):
        if event.is_directory:
            return
        # デバウンス（5秒以内の重複を防止）
        now = time.time()
        if now - self.last_indexed < 5:
            return
        self.last_indexed = now

        print(f"Document changed: {event.src_path}")
        self.indexer_fn()

def watch_docs(docs_dir, indexer_fn):
    handler = DocChangeHandler(indexer_fn)
    observer = Observer()
    observer.schedule(handler, docs_dir, recursive=True)
    observer.start()
    return observer

パフォーマンス最適化

キャッシング

from functools import lru_cache
import hashlib

class CachedRAGBot(RAGBot):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.cache: dict[str, tuple[str, list[str]]] = {}

    async def ask(self, user_id: int, question: str):
        cache_key = hashlib.md5(question.lower().strip().encode()).hexdigest()

        if cache_key in self.cache:
            return self.cache[cache_key]

        answer, sources = await super().ask(user_id, question)
        self.cache[cache_key] = (answer, sources)
        return answer, sources

失敗事例と落とし穴

症状を先に書き、診断を付けました。

ボットが静かになりログにConflictが出る

telegram.error.Conflictの説明は「Raised when a long poll or webhook conflicts with another one」です。同じトークンでポーリングしているプロセスが二つ以上あるという意味です。サーバーのボットが生きているのにローカルでpython bot.pyを回した場合が圧倒的に多く、次が再デプロイのときに以前のコンテナがまだ死んでいない状態です。解決はBotFatherで開発用のトークンを別に作ることです。このライブラリのエラーはTelegramErrorの下でNetworkError、BadRequest、TimedOut、Forbidden、InvalidToken、EndPointNotFound、ChatMigrated、RetryAfter、Conflict、PassportDecryptionErrorに分かれます。

回答はもっともらしいのにドキュメントにない内容だ

順番に見ます。検索ログのsourcesが見当違いなら検索の問題です。ファイルが合っているなら、そのチャンクの実際のテキストを見ます。PDFのローダーが表を潰したか、ヘッダーとフッターが本文に混ざった可能性があります。チャンクまで合っていて、そこではじめてプロンプトの問題です。この順番を飛ばすと、検索が原因の問題をプロンプトで直そうとして何週間も使います。

長い回答だけがまるごと消える

短い質問は通るのに、説明を求める質問にだけ答えが来ません。4096字です。except Exceptionで飲み込んでエラーメッセージだけを撒くと、ログにも原因が残りません。logger.exceptionでスタックを残し、split_for_telegramを付けてください。

再起動すると会話がリセットされる

メモリにしかない状態です。PicklePersistenceかチェックポインターを付け、そのファイルがボリュームの上にあるかまで確認します。

Chromaをimportできない

langchain_community.vectorstores.Chromaはcommunity 0.2.9からdeprecatedで、現在のパスはlangchain_chromaです。langchain-chromaを0.1.2以上でインストールし、importを変えてください。ただし0.4.2で物理的に削除されたかどうかは確認できていないので、ImportErrorなのかDeprecationWarningなのかは動かして判断してください。

FAISSだけ行き先がない

langchain-faissというパッケージは存在せず、langchain_community.vectorstoresのFAISSが依然として唯一の経路です。サンセットが予告されたパッケージの中に、きれいな移行先なしに残っているわけなので、ベクトルストアを選ぶときはこの点を知ったうえで選んでください。

マイグレーションしたのにまたレガシーだと言われる

create_retrieval_chain、create_stuff_documents_chain、create_history_aware_retrieverはv1でlangchain-classicへ行きました。レガシーチェーンの現代的な代替だと教わったものが、いまは同じレガシーパッケージにあります。LCEL自体は死んでいません。langchain_core.runnablesのRunnablePassthroughとRunnableLambda、StrOutputParser、ChatPromptTemplate、Documentは全部生きていて、パイプ演算子もそのままです。ただしv1のリファレンスのRunnableのページにはLCELという用語が登場せず、専用のドキュメントページもありません。廃止されたのではなく、ドキュメントがもうその方式で教えていないということです。

セキュリティ告知を見逃す

症状がないので危険な項目です。二次情報源によると、CVE-2025-68664（CVSS 9.3、デシリアライゼーション）、CVE-2026-34070（7.5、プロンプト読み込みAPIのパストラバーサル）、CVE-2025-67644（7.3、LangGraph SQLiteチェックポイントのSQLインジェクション）が報告されています。修正バージョンはlangchain-core 0.3.81以上または1.2.22以上、langgraph-checkpoint 3.0以上、langgraph-checkpoint-sqlite 3.0.1以上です。ベンダーのドキュメントではなく脆弱性データベース系の二次情報源なので、対応の前に使用中のディストリビューションのセキュリティ告知を直接確認してください。

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

この構造が合わない場合がはっきりとあります。

ドキュメントが小さいとき — FAQ全体がA4で10枚なら、エンベディングもベクトルストアも要りません。全文をそのままプロンプトに入れてください。検索の段階がないので、検索が間違って選んだという失敗モード自体が消え、インデックス作成のパイプラインもウォッチャーもしきい値のチューニングも全部なくなります。RAGはプロンプトに入らないときに使うものです。

権限が必要なドキュメントのとき — もっとも重要な項目です。Telegramにはドキュメント単位のアクセス制御という概念がありません。ボットがグループに入っていれば、その部屋の全員が同じ質問に同じ答えを受け取ります。人事評価や給与テーブルのように、誰が尋ねるかによって答えが変わるべきドキュメントをインデックスに入れても、ベクトルストアはその区別をしてくれません。as_retrieverのsearch_kwargsにfilterを入れて真似ることはできますが、ボットのコードが身元を正確に判別するという前提の上でのみ成り立ち、一か所でも漏らせばそのまま流出です。認可が要件なら、認可を先に設計できる場所に置いてください。

正確な値を照会するとき — 会議室の予約状況、残りの有給日数、注文の配送状況のような質問はRAGの問題ではありません。ベクトルの類似度は近似であり、近似で答えてはいけない質問です。キーワード検索やデータベースの照会で答えを得て、ボットはその結果を文にする役割だけを担うほうが正確です。曖昧なときはこう尋ねてください。間違った答えが出たとき、似たものは見つけたね、で済ませられる質問なのか、それともただ間違いなのか。

規制対象の内容のとき — 医療、法律、金融のように回答に責任が伴う領域では、ソース表示が免責になりません。ソース表示はコンテキストに入ったという記録であって、回答がそこから出たという保証ではないからです。たいていは生成された文の代わりに、検索結果の原文をそのまま見せるほうが良いです。

ドキュメントが頻繁に変わるのに正確性が重要なとき — 再インデックスが回っているあいだ、ベクトルストアには古いチャンクと新しいチャンクが混ざっており、削除されたドキュメントのチャンクは明示的に消さなければ検索され続けます。すでに廃止された規程をボットが自信たっぷりに引用する事故がここから出ます。変更が多いなら、増分更新の代わりに新しいコレクションを作ってまるごと入れ替えてください。

まとめ

LangChain + RAG + Telegramでインテリジェントなメモリを構築しました：

• ドキュメントベースの回答：正確な情報のみ提供し、ハルシネーションを最小化
• 会話メモリ：ユーザーごとのコンテキストを維持
• ソース表示：回答の根拠となるドキュメントを透明に提示
• MMR検索：多様性と関連性をバランスよく検索
• 自動更新：ドキュメント変更時に自動で再インデックス

運用に必要なリストには、ここにもう一つ付きます。いま使っているパッケージがどのバージョンなのかを知っていることです。

参考資料

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

• python-telegram-bot公式ドキュメント
• python-telegram-bot PyPI
• echobotのサンプル
• Telegram Bot FAQ
• langchain-community PyPI
• Sunsetting langchain-communityのイシュー
• langchain-classic chainsのリファレンス
• LangChainの短期メモリのドキュメント

クイズ