Telegram-бот с LLM: персональный AI-ассистент в мессенджере

ChatGPT в Telegram — удобно, но зависишь от OpenAI. Свой бот с локальной LLM — бесплатно, приватно, и можно кастомизировать как угодно. Пишем за вечер.

Архитектура

Telegram → Bot API → Python (aiogram) → Ollama → LLM
                                       ↘ ChromaDB (RAG)

Компоненты:

• aiogram 3 — асинхронная библиотека для Telegram Bot API
• Ollama — запуск LLM локально
• ChromaDB — опционально, для RAG

Минимальный бот

import asyncio
from aiogram import Bot, Dispatcher, types
from aiogram.filters import CommandStart
import ollama

BOT_TOKEN = "your-bot-token"
MODEL = "llama3.1:8b"
SYSTEM_PROMPT = "Ты — полезный ассистент. Отвечай кратко и по делу. Язык — русский."

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

# Хранилище контекста диалогов
conversations: dict[int, list[dict]] = {}
MAX_HISTORY = 20

@dp.message(CommandStart())
async def start(message: types.Message):
    conversations[message.from_user.id] = []
    await message.answer("Привет! Я AI-ассистент. Задавай вопросы.")

@dp.message()
async def handle_message(message: types.Message):
    user_id = message.from_user.id

    if user_id not in conversations:
        conversations[user_id] = []

    # Добавляем сообщение пользователя
    conversations[user_id].append({
        "role": "user",
        "content": message.text
    })

    # Обрезаем историю
    if len(conversations[user_id]) > MAX_HISTORY:
        conversations[user_id] = conversations[user_id][-MAX_HISTORY:]

    # Показываем "печатает..."
    await bot.send_chat_action(message.chat.id, "typing")

    # Запрос к LLM
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    messages.extend(conversations[user_id])

    response = ollama.chat(model=MODEL, messages=messages)
    answer = response["message"]["content"]

    # Сохраняем ответ в контексте
    conversations[user_id].append({
        "role": "assistant",
        "content": answer
    })

    await message.answer(answer)

async def main():
    await dp.start_polling(bot)

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

Добавляем RAG

Бот, который отвечает по загруженным документам:

import chromadb

chroma = chromadb.HttpClient(host="localhost", port=8000)
collection = chroma.get_or_create_collection("bot_docs")

@dp.message(lambda m: m.document is not None)
async def handle_document(message: types.Message):
    """Обработка загруженных документов."""
    file = await bot.get_file(message.document.file_id)
    content = await bot.download_file(file.file_path)
    text = content.read().decode("utf-8")

    # Чанкинг
    chunks = [text[i:i+1000] for i in range(0, len(text), 800)]

    # Индексируем
    for i, chunk in enumerate(chunks):
        emb = ollama.embed(model="nomic-embed-text", input=chunk)
        collection.upsert(
            ids=[f"{message.document.file_name}_{i}"],
            embeddings=[emb["embeddings"][0]],
            documents=[chunk],
            metadatas=[{"filename": message.document.file_name}]
        )

    await message.answer(
        f"Проиндексировал {len(chunks)} фрагментов из {message.document.file_name}. "
        "Теперь можешь задавать вопросы по документу."
    )

@dp.message()
async def handle_with_rag(message: types.Message):
    """Ответ с учётом RAG-контекста."""
    # Поиск релевантных чанков
    query_emb = ollama.embed(
        model="nomic-embed-text",
        input=message.text
    )["embeddings"][0]

    results = collection.query(
        query_embeddings=[query_emb],
        n_results=3
    )

    context = ""
    if results["documents"][0]:
        context = "\n\n".join(results["documents"][0])

    # Промпт с контекстом
    if context:
        prompt = f"""Используй контекст для ответа на вопрос.
Если ответа нет в контексте, скажи об этом.

Контекст:
{context}

Вопрос: {message.text}"""
    else:
        prompt = message.text

    await bot.send_chat_action(message.chat.id, "typing")

    response = ollama.chat(
        model="llama3.1:8b",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": prompt}
        ]
    )

    await message.answer(response["message"]["content"])

Стриминг ответа

Длинные ответы лучше отправлять по частям — пользователь не будет ждать:

@dp.message()
async def handle_streaming(message: types.Message):
    await bot.send_chat_action(message.chat.id, "typing")

    # Стриминг от Ollama
    full_response = ""
    sent_message = None

    stream = ollama.chat(
        model=MODEL,
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": message.text}
        ],
        stream=True
    )

    buffer = ""
    for chunk in stream:
        buffer += chunk["message"]["content"]

        # Обновляем сообщение каждые ~100 символов
        if len(buffer) - len(full_response) > 100:
            full_response = buffer
            if sent_message:
                await sent_message.edit_text(full_response)
            else:
                sent_message = await message.answer(full_response)

    # Финальное обновление
    if sent_message and buffer != full_response:
        await sent_message.edit_text(buffer)
    elif not sent_message:
        await message.answer(buffer)

Деплой

Systemd-сервис:

# /etc/systemd/system/ai-bot.service
[Unit]
Description=Telegram AI Bot
After=network.target ollama.service

[Service]
Type=simple
User=bot
WorkingDirectory=/opt/ai-bot
ExecStart=/opt/ai-bot/venv/bin/python bot.py
Restart=always
RestartSec=10
Environment=BOT_TOKEN=your-token-here

[Install]
WantedBy=multi-user.target

sudo systemctl enable --now ai-bot

Ограничения и советы

• Rate limiting: ограничьте количество запросов на пользователя (Ollama может не справиться)
• Whitelist: если бот приватный — проверяйте user_id перед обработкой
• Таймаут: LLM может думать долго, ставьте таймаут 60 сек
• Память: контекст диалога хранится в RAM, для продакшена используйте Redis

Итого

Telegram-бот с LLM — это пара сотен строк на Python. aiogram + Ollama + опционально ChromaDB для RAG. Деплоится как обычный systemd-сервис. Работает на том же сервере, где крутится Ollama.