Building an Intelligent Telegram FAQ Bot with LangChain + RAG: A Document-Based Q&A System

• Introduction
• Version Check First: The Imports in This Article Are Already Legacy
• Architecture
• Environment Setup
• Document Loading and Indexing
• What Actually Prints the First Time You Run Indexing
• RAG Chain Implementation
• Is It Retrieving, or Is the LLM Making It Up?
• Telegram Bot Implementation
• The 4096-Character Wall, and the Typing Indicator
• Conversation Memory Disappears on Restart
• What It Means to await a Vector Search on Top of asyncio
• Deploying with Docker
• Automatic Document Updates
• Performance Optimization
  • Caching
• Failure Cases and Traps
  • The Bot Goes Quiet and Conflict Shows Up in the Logs
  • The Answer Sounds Plausible but Is Not in the Documents
  • Only Long Answers Vanish Entirely
  • Conversations Reset on Restart
  • Chroma Cannot Be Imported
  • FAISS Alone Has Nowhere to Go
  • I Migrated and It Says Legacy Again
  • You Miss the Security Advisories
• When Not to Use This
• Summary
• References
• Quiz

Introduction

Rule-based chatbots can only answer predefined questions, but RAG (Retrieval-Augmented Generation)-based chatbots retrieve relevant information from documents and respond in natural language. In this article, we build a Telegram bot that answers questions based on company FAQ documents.

Version Check First: The Imports in This Article Are Already Legacy

The code in this article worked when it was written. As of August 2026 you should not copy it verbatim. python-telegram-bot is at 22.8 (2026-06-12), which supports Telegram Bot API 10.0 natively. The official documentation notes that the library has been built on Python asyncio since v20.0, and that fact comes back to bite us later.

On the LangChain side it is langchain 1.3.15 (2026-08-11), langchain-core 1.5.5, langchain-classic 1.0.8, langchain-community 0.4.2. The biggest change is the sunset of langchain-community. PyPI carries a "langchain-community is being sunset" banner, and the issue body says "This sunset will take effect immediately". The date is 2026-05-22 and no EOL date is stated, so do not build a schedule around a guessed disappearance date. The v1 langchain package keeps only agents, messages, tools, chat_models and embeddings; the chains and memory this article uses have moved to langchain-classic.

# Imports in this article → where they live as of 2026-08
langchain.chains.ConversationalRetrievalChain
  → langchain_classic.chains.conversational_retrieval.base.ConversationalRetrievalChain
    (deprecated since 0.1.17)
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   (deprecated since community 0.2.9)
langchain_community.vectorstores.FAISS
  → No migration path. The langchain-faiss package does not exist.

create_retrieval_chain, create_stuff_documents_chain and create_history_aware_retriever, the replacements people were pointed to for the legacy chains back in the v0.2 and v0.3 era, are in langchain-classic in v1 as well. The people most surprised by this are the ones who already migrated once. What v1 puts front and center is create_agent, and short-term memory is a checkpointer rather than a Memory object.

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,
)

In production you use PostgresSaver instead of InMemorySaver. The reason this shape fits Telegram so well is that you can drop the chat ID straight into the thread_id slot. Installation looks like this.

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

The code below is left in its original form, because the legacy paths still work and most codebases under maintenance look exactly like this.

Architecture

User Question
    ↓
Telegram Bot API
    ↓
LangChain RAG Pipeline
    ├── 1. Query Embedding (OpenAI)
    ├── 2. Vector Search (ChromaDB)
    ├── 3. Context Retrieval (Top-K)
    └── 4. LLM Generation (GPT-4o)
    ↓
Answer + Source Citation

Environment Setup

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 Settings
CHUNK_SIZE = 1000
CHUNK_OVERLAP = 200
TOP_K = 4
MODEL_NAME = "gpt-4o"
EMBEDDING_MODEL = "text-embedding-3-small"

Document Loading and Indexing

# 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):
    """Load documents in various formats"""
    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"):
    """Split documents into chunks and store in the vector store"""
    # Load documents
    documents = load_documents(docs_dir)
    print(f"Total documents: {len(documents)}")

    # Text splitting
    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)}")

    # Generate embeddings & save to vector store
    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")

What Actually Prints the First Time You Run Indexing

Running indexer.py for the first time gives you output like this. The numbers differ per document set, but the shape is the same.

$ 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

There are three numbers to look at. First, Total documents. PyPDFLoader creates one Document per page, so feeding in 12 PDFs and getting 47 is normal; getting 12 means the pages were not split. Second, Total chunks. With chunk_size 1000 and an overlap of 200, each chunk digests roughly 800 characters of new text, so the number should be close to the total character count divided by 800. Ninety documents producing 95 chunks means most of them are under 1000 characters and chunking is doing nothing. Third, the size of the chroma_db directory. If it stops at a few dozen KB, the embeddings failed quietly or the chunks are empty. A file with no text layer, such as a scanned PDF, makes the loader succeed with page_content as an empty string, and indexing then runs to the end without a single error.

Before you start the bot, run a script that checks retrieval on its own.

# smoke_test.py — check retrieval without the bot
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 = [
    "when can I start using annual leave",   # a question the docs answer
    "remote work application procedure",     # a question the docs answer
    "what is the company dog's name",        # not in the docs = negative control
]

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]}...")

The output looks like this.

Q: when can I start using annual leave  -> 3 hits
   docs/hr-policy.pdf: Paid annual leave is granted to employees who have worked at le...
   docs/hr-policy.pdf: Under the leave usage promotion policy, unused leave is measure...
   docs/onboarding.md: In your first year, one day of paid leave accrues for each full...

Q: what is the company dog's name  -> 3 hits
   docs/office-guide.md: Office access cards are issued at the first-floor reception des...
   docs/hr-policy.pdf: Family event leave is granted on the following basis. Five days...
   docs/onboarding.md: The Slack channel guide is as follows. Company-wide announcemen...

The third query is the whole point of this script. Even for a question the documents do not answer, similarity_search always returns k results. However low the similarity, it fills the list with the nearest things it has. That is why checking whether retrieval works means looking at the case that should fail, not the case that should succeed. The three chunks in the second block above have nothing to do with the question, and they go into the LLM context exactly as they are. A large share of what a bot invents is decided here, not in the generation step.

RAG Chain Implementation

# 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 = """You are a company FAQ assistant. Answer questions based on the provided context.

Rules:
1. Only use information from the context.
2. If unsure, respond with "I could not find that information in the provided documents."
3. Include the source documents referenced at the end of your answer.
4. Keep answers concise and clear.

Context:
{context}"""

def create_rag_chain(persist_dir: str = "./chroma_db"):
    """Create the RAG chain"""
    # Load vector store
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    vectorstore = Chroma(
        persist_directory=persist_dir,
        embedding_function=embeddings,
    )

    # Retriever configuration
    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,
    )

    # Conversation memory (last 5 turns)
    memory = ConversationBufferWindowMemory(
        k=5,
        memory_key="chat_history",
        return_messages=True,
        output_key="answer",
    )

    # Prompt
    prompt = ChatPromptTemplate.from_messages([
        SystemMessagePromptTemplate.from_template(SYSTEM_PROMPT),
        HumanMessagePromptTemplate.from_template("{question}"),
    ])

    # Create chain
    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 bot that manages per-user conversation context"""

    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):
        """Per-user chain (separate conversation memory)"""
        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]]:
        """Answer a question and return sources"""
        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)

        # Remove duplicates
        sources = list(dict.fromkeys(sources))
        return answer, sources

    def reset_memory(self, user_id: int):
        """Reset conversation memory for a user"""
        if user_id in self.user_chains:
            del self.user_chains[user_id]

Is It Retrieving, or Is the LLM Making It Up?

Bring the bot up and an answer comes out either way. Whether that answer came from the documents or from what the model already knew is not something the chat window distinguishes. Three things, turned on, make the distinction.

First, make an empty result possible. The search_type on as_retriever has exactly three values: the default similarity, mmr, and similarity_score_threshold. The first two never hand back an empty list; only the third can express no relevant documents.

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

docs = retriever.invoke(question)
if not docs:
    # Do not call the LLM at all. Ending here is the correct answer.
    return "I could not find that information in the provided documents.", []

The threshold is not a magic number. The same 0.5 means different things depending on the embedding model and the distance function. Pick it somewhere between the score of the negative control in the smoke test above and the scores of the working questions. Whether the score is a distance or a similarity also varies by implementation. Check the exact API in the docs for the version you are using.

Second, log the retrieved chunks.

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],
)

This is the first line you look at when a report comes in. If hits is 0 and an answer still went out, the prompt guard was breached; if hits is 4 and the sources are all wrong, it is a retrieval problem and not an LLM problem. Without that distinction you spend weeks editing prompts.

Third, add a retrieval-only command. In CommandHandler(command, callback, filters=None, block=True, has_args=None), the arguments that follow the command arrive as CallbackContext.args.

async def debug(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    query = " ".join(context.args)
    if not query:
        await update.message.reply_text("Usage: /debug <search term>")
        return

    docs = retriever.invoke(query)
    if not docs:
        await update.message.reply_text("No results (below threshold)")
        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))

This command shows retrieval results without going through the LLM. Paste in the exact sentence from a report that an answer was wrong and it separates a retrieval failure from a generation failure in one shot.

One last misconception about source citation. A source attached to an answer means that chunk went into the context; it is not a guarantee that the sentence came from it. Conversely, if the sources are empty and a confident answer came out, that is the model talking, not the documents.

Telegram Bot Implementation

# 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):
    """Start command"""
    welcome = (
        "Hello! I'm the FAQ assistant.\n\n"
        "Feel free to ask me anything.\n"
        "I'll answer based on company documents.\n\n"
        "Commands:\n"
        "/reset - Reset conversation\n"
        "/sources - List searchable documents"
    )
    await update.message.reply_text(welcome)

async def reset(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Reset conversation memory"""
    user_id = update.effective_user.id
    rag_bot.reset_memory(user_id)
    await update.message.reply_text("Conversation has been reset.")

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle general messages"""
    user_id = update.effective_user.id
    question = update.message.text

    # Show typing indicator
    await context.bot.send_chat_action(
        chat_id=update.effective_chat.id,
        action="typing"
    )

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

        # Format response
        response = answer
        if sources:
            response += "\n\nReference Documents:\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(
            "Sorry, an error occurred while generating the answer."
        )

async def post_init(application: Application):
    """Register commands on bot startup"""
    commands = [
        BotCommand("start", "Start the bot"),
        BotCommand("reset", "Reset conversation"),
        BotCommand("sources", "List searchable documents"),
    ]
    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()

The 4096-Character Wall, and the Typing Indicator

Answers from a RAG bot get long. Feed in four chunks and ask for an explanation with evidence and 2,000 characters goes by easily. But telegram.constants.MessageLimit.MAX_TEXT_LENGTH is 4096. Go over it and the send fails; I will not claim which exception comes up. The description of telegram.error.BadRequest is only a sentence about the request not being processed, with no mention of length. Check the exact exception type in the docs for the version you are using. What matters is not the exception name but cutting the text beforehand.

from telegram.constants import MessageLimit

def split_for_telegram(text: str, limit: int = MessageLimit.MAX_TEXT_LENGTH) -> list[str]:
    """Split on paragraph boundaries -> line boundaries -> hard cut, in that order."""
    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)

There is a reason paragraph boundaries come first. Cut an answer containing a code block mechanically at 4096 characters and the opening backticks and the closing backticks land in different messages, and if you turned parse_mode on the whole send fails with a formatting error. It does not arrive truncated; it does not arrive at all.

The typing indicator is not optional in a RAG bot. Embedding plus vector search plus LLM generation adds up to a perceived delay of three to ten seconds. Use the constant rather than a string. The path is telegram.constants.ChatAction, with TYPING, UPLOAD_PHOTO, UPLOAD_DOCUMENT, CHOOSE_STICKER, FIND_LOCATION and others.

from telegram.constants import ChatAction

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

Watch the duration. In the wording of the Telegram documentation, "The status is set for 5 seconds or less". If generation takes 12 seconds, the user stares at a screen with no reaction for 7 of them. Put a background task in place that calls it again every four seconds until generation finishes. Check the exact parameter list for send_chat_action in the docs for the version you are using. Rate limits follow along too. These are sentences from the Telegram bot FAQ.

# 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."

The 4096-character split and the rate limit are the same problem stuck together. Cut a long answer into four pieces and fire them back to back and you throw four messages in one second, past the single-chat guidance, and in a group you also hit the 20-per-minute figure. The peculiar thing about a RAG bot is that it is not the bot with many users but the bot with long answers that gets caught first. The defaults for telegram.ext.AIORateLimiter are overall_max_rate=30, overall_time_period=1, group_max_rate=20, group_time_period=60 and max_retries=0, which reflect those numbers directly. It needs a separate extra to install.

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()
)

Keep in mind that max_retries defaults to 0. With the default settings, receiving a RetryAfter does not trigger a retry. RetryAfter has a retry_after attribute, so either raise the value or handle it yourself. For reference, Application.builder() is a static method that returns an ApplicationBuilder, and token and bot are mutually exclusive.

Conversation Memory Disappears on Restart

The RAGBot above holds per-user chains in a user_chains dictionary. That is process memory, so a single deploy wipes all of it, and objects pile up in proportion to the number of users with no code cleaning them out. A hundred people inside the company is fine, but attach it to an open channel and it is a straight leak.

The library already provides this slot. context.user_data is a dictionary mapped per user ID, chat_data is keyed by chat ID, and bot_data is a single one for the whole bot.

A design choice appears here. If you only take one-to-one conversations, user_data and chat_data are the same thing, but the moment the bot is invited into a group they diverge. user_data is the shape where one person carries context across several groups; chat_data is the shape where a whole group shares one context. For an internal FAQ bot, chat_data is usually the natural one.

By default this is process memory too. To put it on disk, attach PicklePersistence(filepath, store_data=None, single_file=True, on_flush=False, update_interval=60, context_types=None). It stores user_data, chat_data, bot_data, callback_data and conversations.

from telegram.ext import ApplicationBuilder, PicklePersistence

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

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

With Docker, bot_state.pickle absolutely has to sit on a volume. Inside the image it resets every time the container is recreated, which is the same as not having the persistence you think you attached. Add one next to the chroma-data volume in the docker-compose.yml below.

In the v1 style a checkpointer takes this slot, and you drop the chat ID straight into thread_id.

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

# The docs advise using PostgresSaver instead of InMemorySaver in production.
# Check the exact import path in the docs for the version you are using.
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,
    )
    # The return structure varies by version. Check the exact API in the docs for your version.
    await reply_long(update, extract_answer(result))

Those few lines replace get_chain, user_chains and reset_memory in RAGBot entirely.

What It Means to await a Vector Search on Top of asyncio

python-telegram-bot has been asyncio-based since v20. A single event loop handles every update, so a blocking call inside a handler stops the whole bot. So far this is familiar.

The problem is that attaching await is not enough to relax. Most of the async methods on a LangChain VectorStore are not genuinely asynchronous but thread-pool wrappers. The langchain-core source looks like this.

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)

The source comments call these methods "temporary workarounds" and note that the proper fix is for the vector store implementation to become asynchronous itself. And Chroma in langchain_chroma has not a single async def. It defines only synchronous methods and inherits every method beginning with a from the base class.

So await retriever.ainvoke(question) looks like it releases the event loop, while in reality it occupies one thread-pool worker. The worker count is finite, and once concurrent questions exceed it, queuing starts. To the user it looks like the bot is sometimes slow, and the cause is easy to misread as LLM API latency.

There is one more mistake in the same vein. run_polling and run_webhook are blocking synchronous methods, not coroutines. Attaching await inside an async def main does not work; you have to call them from an ordinary def main. The original code above having a def main is correct.

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)

With the default settings, updates are processed sequentially. The next turn only comes once the previous user's RAG call finishes. Turning on concurrent_updates gives you concurrent processing, but the moment you turn it on, the thread-pool problem above and the rate limits from the previous section both become real at once. Do not decide the three separately; look at them together.

Deploying with Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

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

COPY . .

# Index documents
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

Automatic Document Updates

# watcher.py - Detect document changes and auto-reindex
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
        # Debounce (prevent duplicates within 5 seconds)
        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

Performance Optimization

Caching

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

Failure Cases and Traps

The symptom comes first, with the diagnosis attached.

The Bot Goes Quiet and Conflict Shows Up in the Logs

The description of telegram.error.Conflict is "Raised when a long poll or webhook conflicts with another one". It means more than one process is polling with the same token. The overwhelmingly common case is that the bot on the server is alive while you ran python bot.py locally; next is a redeploy where the previous container has not died yet. The fix is to create a separate development token in BotFather. Errors in this library branch out under TelegramError into NetworkError, BadRequest, TimedOut, Forbidden, InvalidToken, EndPointNotFound, ChatMigrated, RetryAfter, Conflict and PassportDecryptionError.

The Answer Sounds Plausible but Is Not in the Documents

Work through it in order. If the sources in the retrieval log are wrong, it is a retrieval problem. If the files are right, look at the actual text of those chunks. The PDF loader may have mangled a table, or headers and footers may have been mixed into the body. Only once the chunks check out is it a prompt problem. Skip that order and you spend weeks trying to fix a retrieval problem with prompts.

Only Long Answers Vanish Entirely

Short questions work, and only questions asking for an explanation get no answer. That is 4096 characters. Swallow it with except Exception and print only the error message and the logs keep no trace of the cause either. Record the stack with logger.exception and attach split_for_telegram.

Conversations Reset on Restart

The state lives only in memory. Attach PicklePersistence or a checkpointer, and check that the file is on a volume as well.

Chroma Cannot Be Imported

langchain_community.vectorstores.Chroma has been deprecated since community 0.2.9 and the current path is langchain_chroma. Install langchain-chroma at 0.1.2 or above and change the import. That said, I could not confirm whether it was physically removed in 0.4.2, so run it and judge whether you get an ImportError or a DeprecationWarning.

FAISS Alone Has Nowhere to Go

The langchain-faiss package does not exist, and FAISS in langchain_community.vectorstores is still the only path. It is sitting inside a package with an announced sunset and no clean destination, so know this when you pick a vector store.

I Migrated and It Says Legacy Again

create_retrieval_chain, create_stuff_documents_chain and create_history_aware_retriever went to langchain-classic in v1. The things you learned as the modern replacements for the legacy chains are now in the same legacy package. LCEL itself is not dead. RunnablePassthrough and RunnableLambda from langchain_core.runnables, StrOutputParser, ChatPromptTemplate and Document are all alive and the pipe operator still works. It is just that the Runnable page in the v1 reference never uses the term LCEL and there is no dedicated documentation page for it. It was not discarded; the documentation simply no longer teaches it that way.

You Miss the Security Advisories

This one is dangerous because it has no symptom. Secondary sources report CVE-2025-68664 (CVSS 9.3, deserialization), CVE-2026-34070 (7.5, path traversal in the prompt loading API) and CVE-2025-67644 (7.3, SQL injection in the LangGraph SQLite checkpoint). Fixed versions are langchain-core 0.3.81 or above or 1.2.22 or above, langgraph-checkpoint 3.0 or above, and langgraph-checkpoint-sqlite 3.0.1 or above. These are vulnerability-database secondary sources rather than vendor documentation, so check the security advisories for your own distribution directly before acting.

When Not to Use This

There are clear cases where this structure does not fit.

When the documents are small — if the whole FAQ is ten pages of A4, you need neither embeddings nor a vector store. Put the full text into the prompt. With no retrieval step, the failure mode of retrieval picking the wrong thing disappears entirely, and the indexing pipeline, the watcher and the threshold tuning all go away with it. RAG is what you use when it does not fit in the prompt.

When the documents need permissions — the most important item. Telegram has no concept of per-document access control. If the bot is in a group, everyone in that room gets the same answer to the same question. Put documents like performance reviews or salary tables, where the answer must depend on who is asking, into the index and the vector store will not make that distinction for you. You can imitate it by putting a filter into the search_kwargs of as_retriever, but it only holds on the assumption that the bot code identifies people correctly, and missing it in one place is a leak. If authorization is a requirement, put it somewhere you can design authorization first.

When you need an exact value — questions like meeting room availability, remaining leave days or order shipping status are not RAG problems. Vector similarity is an approximation, and these are questions you must not answer with an approximation. It is more accurate to get the answer from keyword search or a database query and let the bot only turn that result into a sentence. When in doubt, ask it this way: when the answer is wrong, is it a question where "well, it found something similar" is survivable, or is it just wrong?

When the content is regulated — in areas where an answer carries liability, such as medicine, law and finance, source citation is not a disclaimer. A source is a record that a chunk entered the context, not a guarantee that the answer came from it. Usually it is better to show the retrieved text as it stands instead of a generated sentence.

When documents change often and accuracy matters — while reindexing runs, the vector store holds a mix of old and new chunks, and chunks from deleted documents keep being retrieved unless you delete them explicitly. This is where the incident of a bot confidently citing an already-repealed policy comes from. If changes are frequent, build a new collection and swap it wholesale instead of updating incrementally.

Summary

We built an intelligent FAQ bot using LangChain + RAG + Telegram:

• Document-based answers: Provides only accurate information, minimizing hallucinations
• Conversation memory: Maintains per-user context
• Source citation: Transparently presents the reference documents for answers
• MMR search: Balances diversity and relevance in search results
• Auto-update: Automatically re-indexes when documents change

One more item belongs on the operations list: knowing which version of each package you are actually running.

References

All checked as of 2026-08-16.

• python-telegram-bot official documentation
• python-telegram-bot on PyPI
• echobot example
• Telegram Bot FAQ
• langchain-community on PyPI
• Sunsetting langchain-community issue
• langchain-classic chains reference
• LangChain short-term memory documentation

Quiz