How to Build a Self-Hosted RAG Telegram AI Bot with Python & LlamaIndex (2026 Guide)

# Create project directory
mkdir rag-telegram-bot && cd rag-telegram-bot

# Initialize Python virtual environment
python3 -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Upgrade pip and install core dependencies
pip install --upgrade pip
pip install aiogram==3.15.0 llama-index==0.11.0 chromadb==0.5.5 python-dotenv pypdf

BOT_TOKEN=8123456789:AAEexampleTokenFromBotFather
OPENAI_API_KEY=sk-proj-your-actual-openai-api-key-here
CHROMA_PATH=./chroma_db
DOCUMENTS_DIR=./docs

import os
from dotenv import load_dotenv

load_dotenv()

BOT_TOKEN = os.getenv("BOT_TOKEN")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
CHROMA_PATH = os.getenv("CHROMA_PATH", "./chroma_db")
DOCUMENTS_DIR = os.getenv("DOCUMENTS_DIR", "./docs")

if not BOT_TOKEN or not OPENAI_API_KEY:
    raise ValueError("Missing BOT_TOKEN or OPENAI_API_KEY in environment settings!")

import os
import chromadb
from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, StorageContext, Settings
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.llms.openai import OpenAI
from config import CHROMA_PATH, DOCUMENTS_DIR, OPENAI_API_KEY

# Set global LLM and Embedding models
Settings.llm = OpenAI(model="gpt-4o-mini", temperature=0.2, api_key=OPENAI_API_KEY)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small", api_key=OPENAI_API_KEY)

class TelegramRAGEngine:
    def __init__(self):
        os.makedirs(DOCUMENTS_DIR, exist_ok=True)
        self.chroma_client = chromadb.PersistentClient(path=CHROMA_PATH)
        self.chroma_collection = self.chroma_client.get_or_create_collection("telegram_kb")
        self.vector_store = ChromaVectorStore(chroma_collection=self.chroma_collection)
        self.storage_context = StorageContext.from_defaults(vector_store=self.vector_store)
        self.index = self._load_or_create_index()

    def _load_or_create_index(self):
        try:
            return VectorStoreIndex.from_vector_store(
                self.vector_store,
                storage_context=self.storage_context
            )
        except Exception:
            documents = SimpleDirectoryReader(DOCUMENTS_DIR).load_data()
            return VectorStoreIndex.from_documents(
                documents,
                storage_context=self.storage_context
            )

    def ingest_file(self, file_path: str) -> int:
        reader = SimpleDirectoryReader(input_files=[file_path])
        new_docs = reader.load_data()
        for doc in new_docs:
            self.index.insert(doc)
        return len(new_docs)

    def query(self, user_query: str) -> dict:
        query_engine = self.index.as_query_engine(similarity_top_k=3)
        response = query_engine.query(user_query)
        
        sources = []
        for node in response.source_nodes:
            file_name = node.metadata.get("file_name", "Unknown")
            score = round(node.score if node.score else 0.0, 2)
            sources.append(f"📄 {file_name} (Relevance: {score})")
            
        return {
            "answer": str(response),
            "sources": sources
        }

import asyncio
import logging
import os
from aiogram import Bot, Dispatcher, F, types
from aiogram.filters import CommandStart
from aiogram.enums import ParseMode
from config import BOT_TOKEN, DOCUMENTS_DIR
from rag_engine import TelegramRAGEngine

logging.basicConfig(level=logging.INFO)

bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()
rag_engine = TelegramRAGEngine()

@dp.message(CommandStart())
async def cmd_start(message: types.Message):
    welcome_text = (
        "🤖 Welcome to RAG AI Knowledge Assistant!\n\n"
        "I can answer questions grounded directly in your custom documents.\n\n"
        "How to use:\n"
        "1. Send me a .pdf or .txt file to index into the knowledge base.\n"
        "2. Type any question, and I will search the vector index and answer with citations."
    )
    await message.answer(welcome_text, parse_mode=ParseMode.HTML)

@dp.message(F.document)
async def handle_document_upload(message: types.Message):
    doc = message.document
    file_ext = os.path.splitext(doc.file_name)[1].lower()
    
    if file_ext not in [".pdf", ".txt", ".md"]:
        await message.answer("⚠️ Unsupported file type. Please upload .pdf, .txt, or .md files.")
        return

    status_msg = await message.answer(f"📥 Downloading {doc.file_name}...", parse_mode=ParseMode.HTML)
    
    local_path = os.path.join(DOCUMENTS_DIR, doc.file_name)
    file_info = await bot.get_file(doc.file_id)
    await bot.download_file(file_info.file_path, local_path)
    
    await status_msg.edit_text(f"⚙️ Indexing {doc.file_name} into ChromaDB vector store...", parse_mode=ParseMode.HTML)
    
    num_chunks = rag_engine.ingest_file(local_path)
    await status_msg.edit_text(
        f"✅ Successfully Indexed!\n"
        f"File: {doc.file_name}\n"
        f"Parsed Chunks: {num_chunks}\n\n"
        f"You can now ask questions about this document!",
        parse_mode=ParseMode.HTML
    )

@dp.message(F.text)
async def handle_user_query(message: types.Message):
    if message.text.startswith("/"):
        return
        
    await bot.send_chat_action(chat_id=message.chat.id, action="typing")
    
    result = rag_engine.query(message.text)
    
    response_text = f"💡 Answer:\n{result['answer']}\n\n📚 Citations & Sources:\n"
    response_text += "\n".join(result['sources']) if result['sources'] else "No direct sources match."
    
    await message.answer(response_text, parse_mode=ParseMode.HTML)

async def main():
    logging.info("Starting Telegram RAG Bot polling...")
    await dp.start_polling(bot)

if __name__ == "__main__":
    asyncio.run(main())

FROM python:3.12-slim

WORKDIR /app

RUN apt-get update && apt-get install -y build-essential && rm -rf /var/lib/apt/lists/*

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

COPY . .

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

version: '3.8'

services:
  rag_telegram_bot:
    build: .
    container_name: rag_telegram_bot
    restart: always
    env_file:
      - .env
    volumes:
      - ./chroma_db:/app/chroma_db
      - ./docs:/app/docs

Can I run this RAG bot completely offline without OpenAI?

Yes! You can swap OpenAI with Ollama (running local models like DeepSeek-R1, Llama 3.2, or Mistral) and use HuggingFace BAAI/bge-small-en-v1.5 for local vector embeddings. This makes your Telegram bot 100% private and free of API fees.

How many documents can ChromaDB store on a basic VPS?

ChromaDB is exceptionally lightweight. A standard 2GB RAM Linux VPS can easily index and query tens of thousands of PDF pages without latency degradation.

Does this work in Telegram Group Chats?

Yes. You can disable privacy mode in BotFather (/setprivacy -> Disable), allowing the bot to listen to questions in group chats and automatically answer when mentioned or queried.

Where can I find more verified Telegram bots and developer tools?

Browse Telekit's Bots & Apps Directory to discover verified productivity bots, developer utilities, and AI assistants.