Bayesium – LLM чат-бот. Часть 3. Создаём Telegram чат-бота

Будучи вдохновлённым статьёй Кита МакНалти [1], я решил развить изложенную в ней идею и создать чат-бота в Telegram, который будет отвечать на вопросы, опираясь исключительно на книги по байесовской статистике.

В предыдущих частях серии мы создали локальную базу данных с книгами по байесовской статистике [2] и настроили LLM с использованием RAG-фреймворка, которая даёт ответы на вопросы на основе этих книг [3]. В завершающей статье мы настроим более дружелюбный интерфейс для работы, воспользовавшись возможностями Telegram чат-ботов.

Создать бота в Telegram очень просто, для этого находим поиском BotFather и командой /newbot даём жизнь новому боту. Далее выбираем имена и получаем токен, который понадобится для дальнейшей работы. Вызовите команду /mybots, если хотите кастомизировать бота, установить логотип, дать описание и прочее.

Пример создания бота Bayesium

Теперь настала пора открыть Python и реализовать логику работы чат-бота. В этой статье я рассмотрю только вопросы настройки бота, поэтому, для краткости, код по LLM из предыдущей статьи [3] не будет повторен.

Мы будем работать с API Telegram для этого в первый раз устанавливаем, а далее подключаем необходимые пакеты.

# Установка необходимых пакетов:
# pip install python-telegram-bot


# Подключаем пакеты для работы с Telegram
import logging
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, ContextTypes, filters

# Для работы с регулярными выражениями
import re


Также для работы бота нам понадобится использовать асинхронные функции, для чего воспользуемся следующей библиотекой 

# Импорт и применение nest_asyncio для обеспечения корректной работы асинхронного кода
import nest_asyncio
nest_asyncio.apply()

Для соблюдения авторских прав и личного использования бот будет отвечать только на мои сообщения. Для этого укажем мой Telegram ID.

# ID пользователя, которому разрешено использовать бота
ALLOWED_USER_ID = 164935376

Настало время определить поведение традиционной для телеграм-бота команды /start. Напишем для этого следующую фукнцию:

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """
    Обрабатывает команду /start:
      - Проверяет, является ли пользователь авторизованным.
      - Отправляет приветственное сообщение.
    """
    user_id = update.effective_user.id
    if user_id != ALLOWED_USER_ID:
        await update.message.reply_text("Извините, но вы не можете воспользоваться этим ботом.")
        return

    await update.message.reply_text(
        "Привет! Я Байезиум - бот по байесовской статистике. 
         Задай мне вопрос, и я отвечу на основе книг Ричарда МакЭлрита и 
         Джона Крушке"
    )

Потерпев поражение в битве с markdown форматированием сообщений, я создаю дополнительную функцию, которая будет переводить текст в HTML. Это позволяет красиво отображать блоки кода (сниппеты) в сообщениях:

def format_code_for_telegram(text: str, parse_mode: str = "HTML") -> str:
    """
    Форматирует текст для Telegram с использованием HTML:
      - Экранирует специальные HTML-символы.
      - Преобразует блоки кода, выделенные с помощью ```...```, в HTML-теги <pre><code>...</code></pre>.
    """
    if parse_mode == "HTML":
        # Экранирование символов &, <, >
        text = text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")
        # Замена блоков кода на HTML-формат
        text = re.sub(r"```(\w+)?\n(.*?)```", r"<pre><code>\2</code></pre>", text, flags=re.DOTALL)
        return text
    else:
        return text

Следующая функция является ключевой: она принимает вопрос пользователя, передаёт его в LLM-модель (функция ask_question, реализованная в [3]) и возвращает ответ.

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """
    Обрабатывает входящие сообщения:
      - Проверяет авторизацию пользователя.
      - Извлекает текст вопроса из сообщения.
      - Уведомляет пользователя о начале обработки.
      - Вызывает функцию ask_question для получения ответа.
      - Форматирует ответ и отправляет его обратно пользователю.
    """
    
    user_id = update.effective_user.id
    if user_id != ALLOWED_USER_ID:
        await update.message.reply_text("Извините, но вы не можете воспользоваться этим ботом.")
        return

    question = update.message.text
    print(f"DEBUG: Received message - {question}")
    logging.info(f"DEBUG: Received message - {question}")

    # Информируем пользователя о том, что запрос обрабатывается
    await update.message.reply_text("Осуществляется обработка вашего вопроса, пожалуйста, ожидайте...")

    try:
        # Выполнение функции ask_question в отдельном потоке, чтобы не блокировать event loop
        answer = await asyncio.to_thread(ask_question, question)
        print(f"DEBUG: Answer generated - {answer}")
        logging.info(f"DEBUG: Answer generated - {answer}")

        # Форматирование ответа для корректного отображения в Telegram
        formatted_answer = format_code_for_telegram(answer, parse_mode="HTML")
    except Exception as e:
        logging.error(f"Error in ask_question: {e}")
        formatted_answer = "Извините, что-то пошло не так во время исполнения запроса"

    # Отправка отформатированного ответа пользователю
    await update.message.reply_text(formatted_answer, parse_mode="HTML")

Ниже приведён последний фрагмент кода для запуска бота:

async def main() -> None:
    """
    Основная функция:
      - Создает экземпляр приложения Telegram-бота.
      - Регистрирует обработчики команд и сообщений.
      - Запускает цикл поллинга для получения обновлений.
    """
    application = ApplicationBuilder().token("<ВАШ API KEY>").build()

    # Регистрация обработчика команды /start
    application.add_handler(CommandHandler("start", start))
    # Регистрация обработчика для всех остальных сообщений
    application.add_handler(MessageHandler(filters.ALL, handle_message))

    # Запуск бота в режиме поллинга
    await application.run_polling()

if __name__ == '__main__':
    # Настройка логирования для отслеживания событий и ошибок
    logging.basicConfig(
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO
    )

    import asyncio
    # Запуск основной функции с использованием asyncio
    asyncio.run(main())

Теперь настало время перейти в Telegram и задать вопрос. Например, можно спросить: "Как правильно учесть post-treatment эффект? Приведи пример кода на R." Бот обработает запрос и вернёт ответ, сгенерированный на основе специализированных книг.

Пример разговора с ботом Bayesium в Telegram

Для постоянной работы бота его можно запустить на сервере. Пример настройки серверного окружения для запуска Telegram-бота доступен в одной из моих статей [4].

Подводя итог серии, можно сказать, что, расширив идею Кита МакНалти, мне удалось создать чат-бота, который, опираясь на специализированные книги по байесовской статистике, даёт ответы в удобном интерфейсе Telegram. Дальнейшим развитием модели могла бы стать поддержка контекста во время диалога, однако моя серия статей закончится на этом месте, и я оставляю такую возможность другим энтузиастам.

Код доступен на моём GitHub.

Ссылки