Бот-модератор с LLM для Телеграм

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

Мы уже публиковали туториалы по созданию ботов модераторов:

• Простой бот-модератор на Aiogram 3.x

• Антиспам бот-модератор в телеграм. Защищаем группу за три минуты

Но прошлые статьи использовали классические ML-модели, что делало их менее гибкими.

В этой статье я рассмотрю пример, в котором будут использоваться нейронки для выполнения задач модерации.

Большие языковые модели (LLM) способны понимать контекст и смысл сообщения, а не просто искать запрещённые слова. И в отличии от классического ML их не нужно переобучать, достаточно просто изменить промпт. Это делает их идеальным инструментом для интеллектуальной модерации. Вместо примитивных фильтров бот с LLM анализирует семантику текста и принимает взвешенные решения. Однако просто использовать LLM недостаточно — необходимо учитывать уникальные правила, которые могут отличаться для каждого чата. Поэтому мы реализуем гибридный подход:

• LLM анализирует текст на основе контекста и установленных правил.

• Бот сверяется с правилами из базы данных для конкретного чата.

• Система принимает обоснованное решение: оставить сообщение, удалить, забанить пользователя или вынести предупреждение.

Подготовка к разработке бота модератора

Перед началом работы необходимо скачать и установить Python с официального сайта. Рекомендуется использовать версию 3.8 или выше. После установки откройте терминал (командную строку) и введите следующие команды:

pip install aiogram pip install requests

Эти команды установят необходимые для работы библиотеки. Теперь можно приступить к получению токенов для LLM и бота.

Первым делом создадим бота в Telegram. Для этого перейдите в @BotFather и введите команду /newbot. Следуйте инструкциям: укажите имя бота (например, «AI Moderator») и username (например, «aimoder_amvera_bot»).

Теперь необходимо получить токен для доступа к LLM. Регистрируемся в Amvera Cloud и получаем на баланс 111 рублей, чего вполне хватит для теста. После регистрации переходим в раздел LLM. Выбираем одну из четырёх доступных моделей и копирем токен.

Обратите внимание: для использования токена необходимо активировать подписку. В разделе LLM (Preview) найдите кнопку «Подключить пакет токенов» и выберите подходящий тарифный план. Для тестов нам подойдёт бесплатный пакет на 20 000 токенов.

Теперь, имея все необходимые токены, можно приступить к разработке.

Архитектура бота модератора

Как же будет работать система модерации? Процесс модерации организован следующим образом:

1. Пользователь отправляет сообщение в чат.

2. Бот перехватывает сообщение и отправляет его текст на проверку в LLM.

3. Модель анализирует текст на наличие нарушений, основываясь на установленных правилах беседы.

4. В зависимости от ответа модели, бот принимает решение:

   • Если сообщение нарушает правила, то бот выдает соотвествующее наказание, либо удаляет сообщение.

   • Если в сообщении не будет нарушений, то ничего делать не нужно.

Как вы могли заметить, в Amvera Cloud доступны четыре модели:

• LLaMA 3 8B — оптимальна по соотношению цены и скорости, подходит для большинства случаев использования.

• LLaMA 3 70B — более мощная модель для больших чатов и сложных сценариев модерации.

• GPT 5 и 4.1 — для самых сложных сценариев.

Для тестирования мы будем использовать LLaMA 3 8B как наиболее сбалансированный вариант по минимальной цене.

Пишем бота модератора

Создайте в удобном месте папку для проекта, в которой создайте файл main.py с кодом нашего бота и папку data для хранения базы данных.

Начнём с реализации команды /start и базового функционала:

import sqlite3 import asyncio import requests import logging import re  from contextlib import suppress from datetime import datetime, timedelta from aiogram import Bot, Dispatcher, F from aiogram.filters import Command, CommandObject from aiogram.types import Message, ChatPermissions from aiogram.client.default import DefaultBotProperties from aiogram.exceptions import TelegramBadRequest  logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s")  URL = "https://kong-proxy.yc.amvera.ru/api/v1/models/llama" BOT_TOKEN = "" # ТОКЕН ВАШЕГО БОТА AUTH_TOKEN = "" # ТОКЕН LLM В ФОРМАТЕ Bearer XXXXXXXX  logger = logging.getLogger(__name__) connection = sqlite3.connect("/data/database.db", check_same_thread=False) cursor = connection.cursor() bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode="Markdown")) dp = Dispatcher()  cursor.execute("CREATE TABLE IF NOT EXISTS rules (id INTEGER PRIMARY KEY AUTOINCREMENT, rules TEXT, group_id BIGINT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)") cursor.execute("CREATE TABLE IF NOT EXISTS warnings (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, group_id INTEGER NOT NULL, reason TEXT NOT NULL, moderator_id INTEGER, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)") connection.commit()  @dp.message(Command("start")) async def start(message: Message):     await message.answer("Привет! Я бот-модератор, основанный на LLM LLaMA 8b")  if __name__ == "__main__":     asyncio.run(dp.start_polling(bot)) 

Теперь добавим базовую интеграцию с API Amvera, где бот будет проверять каждое сообщение и принимать решение об удалении или оставлении:

@dp.message(F.text) async def on_message(message: Message):     headers = {"X-Auth-Token": str(AUTH_TOKEN), "Content-Type": "application/json"}     data = {"model": "llama8b", "messages": [{"role": "user", "text": "Ты модератор, проверяющий сообщения участников чата. Отвечай строго по форме: Оставить/Удалить"}, {"role": "system", "text": message.text}]}      response = requests.post(URL, headers=headers, json=data, timeout=10)     result = response.json()     decision = result['result']['alternatives'][0]['message']['text'].lower()     if decision == "удалить":         await message.delete()         logger.info(f"УДАЛЕНО | @{message.from_user.username} (ID: {message.from_user.id}) | Текст: \"{message.text}\"")     elif decision == "оставить":         logger.info(f"ОСТАВЛЕНО | @{message.from_user.username} (ID: {message.from_user.id}) | Текст: \"{message.text}\"")     else:         logger.warning(f"ОШИБКА | Текст: \"{message.text}\" | Ответ: {result}") 

Отлично! У нас есть основа, от которой мы можем отталкиваться в дальнейшем. Приступим к реализации полноценной логики модерации.

Добавим три команды для управления правилами группы. Команда /addrules — добавляет или обновляет правила:

@dp.message(Command("addrules")) async def add_rules(message: Message):     if message.chat.type not in ["group", "supergroup"]:         await message.answer("Эта команда работает только в групповых чатах!")         return      rules = message.text.split(' ', 1)     if len(rules) < 2:         await message.answer("Пожалуйста, укажите правила. Пример:\n/addrules Не использовать мат в чате")         return      rules = rules[1].strip()     cursor.execute("SELECT id FROM rules WHERE group_id = ?", (message.chat.id,))     if cursor.fetchone():         cursor.execute("UPDATE rules SET rules = ?, created_at = ? WHERE group_id = ?", (rules, datetime.now(), message.chat.id))         await message.answer("Правила успешно обновлены!")     else:         cursor.execute("INSERT INTO rules (rules, group_id) VALUES (?, ?)", (rules, message.chat.id))         await message.answer("Правила успешно добавлены!")     connection.commit() 

Команда /rules — отображает текущие правила группы:

@dp.message(Command("rules")) async def show_rules(message: Message):     if message.chat.type not in ["group", "supergroup"]:         await message.answer("Эта команда работает только в групповых чатах!")         return      cursor.execute("SELECT rules FROM rules WHERE group_id = ?", (message.chat.id,))     if cursor.fetchone():         await message.answer(f"Правила этой группы:\n\n{cursor.fetchone()[0]}")     else:         await message.answer("Правила для этой группы еще не установлены. Используйте /addrules чтобы добавить правила.") 

И команда /clearrules — удаляет текущие правила группы:

@dp.message(Command("clearrules")) async def clear_rules(message: Message):     if message.chat.type not in ["group", "supergroup"]:         await message.answer("Эта команда работает только в групповых чатах!")         return      cursor.execute("DELETE FROM rules WHERE group_id = ?", (message.chat.id,))     connection.commit()     if cursor.rowcount > 0:         await message.answer("Правила успешно удалены!")     else:         await message.answer("Правила для этой группы не найдены.") 

Отлично! Теперь у нас есть функционал управления правилами:

Можем приступить к разработке основной логики модерации.

Для начала усовершенствуем обработку сообщений. Добавим более строгий и детализированный промпт, где чётко определим процедуру проверки, формат ответа и примеры наказаний для случаев, когда правила в группе не установлены:

@dp.message(F.text) async def on_message(message: Message):     member = await bot.get_chat_member(message.chat.id, message.from_user.id)     if member.status in ['administrator', 'creator']:         return      cursor.execute("SELECT rules FROM rules WHERE group_id = ?", (message.chat.id,))     rules_result = cursor.fetchone()     rules = rules_result[0] if rules_result else "Стандартные правила поведения в чатах"     headers = {"X-Auth-Token": str(AUTH_TOKEN), "Content-Type": "application/json"}     data = { "model": "llama8b", "messages": [ {"role": "system", "text": f"""Ты — строгий модератор чата. Анализируй сообщения на основе правил этой группы.  ПРАВИЛА ДАННОЙ ГРУППЫ: {rules}  ПРОЦЕДУРА АНАЛИЗА: 1. ОЧЕНЬ внимательно прочитай правила группы выше 2. Сравни сообщение пользователя с этими правилами 3. Если есть нарушение - определи какое именно правило нарушено 4. Назначь наказание согласно предусмотренному в правилах 5. Если в правилах не указано конкретное наказание - примени стандартное для типа нарушения  ФОРМАТ ОТВЕТА ТОЛЬКО ОДНА СТРОКА БЕЗ КАВЫЧЕК: Удалить, действие: [бан/мут/кик/предупреждение], длительность: [секунды], причина: [конкретное правило и нарушение]  - ЕСЛИ НАРУШЕНИЕ Оставить, причина: -  - ЕСЛИ НЕТ НАРУШЕНИЯ  ПРИМЕРЫ ОТВЕТОВ ДЛЯ РАЗНЫХ ПРАВИЛ: Если в правилах: "За оскорбления - бан" Сообщение: "иди на хуй" → "Удалить, действие: бан, длительность: 0, причина: оскорбление (нарушение правила про оскорбления)" Если в правилах: "Спам - мут на 1 час" Сообщение: "купите товар" → "Удалить, действие: мут, длительность: 3600, причина: спам (нарушение правила про рекламу)" Если в правилах: "Флуд - предупреждение" Сообщение: "спам спам спам" → "Удалить, действие: предупреждение, длительность: 0, причина: флуд (нарушение правила про флуд)" Если нарушение не описано в правилах явно, но противорит духу правил: Сообщение: "угрозы" → "Удалить, действие: бан, длительность: 0, причина: угрозы (нарушение общего правила поведения)" Сообщение без нарушений: "привет" → "Оставить, причина: -" Сообщение для анализа:"""             },             {                 "role": "user", "text": message.text             }         ], "temperature": 0.1, "max_tokens": 70     }      response = requests.post(URL, headers=headers, json=data, timeout=10)     result = response.json()     decision = result['result']['alternatives'][0]['message']['text'].lower().strip()      delete_match = re.match(r"удалить,\s*действие:\s*(\w+),\s*длительность:\s*(\d+),\s*причина:\s*(.+)",decision)     leave_match = re.match(r"оставить,\s*причина:\s*-", decision)      if delete_match:         action = delete_match.group(1)         duration = int(delete_match.group(2))         reason = delete_match.group(3).strip()         await apply_punish(message, action, duration, reason)         await message.delete()         logger.info(f"УДАЛЕНО | Правила: {rules[:50]}... | Действие: {action} | Длительность: {duration}с | Пользователь: @{message.from_user.username} | Причина: {reason} | Текст: {message.text}")     elif leave_match:         logger.info(f"ОСТАВЛЕНО | @{message.from_user.username} | Правила: {rules[:30]}... | Текст: {message.text}")     else:         logger.warning(f"НЕПОНЯТНЫЙ ОТВЕТ: '{decision}' | Правила: {rules[:30]}... | Текст: {message.text}") 

Теперь реализуем асинхронную функцию apply_punish, которая отвечает за применение наказаний:

async def apply_punish(message: Message, action: str, duration: int, reason: str) -> str:     user_id = message.from_user.id     chat_id = message.chat.id      if action == "бан":         await message.bot.ban_chat_member(chat_id, user_id)         text = "Бан (навсегда)"      elif action == "мут":         until_date = datetime.now() + timedelta(seconds=duration)         await message.bot.restrict_chat_member(chat_id, user_id, until_date = until_date, permissions = ChatPermissions(can_send_messages=False, can_send_media_messages=False, can_send_other_messages=False, can_add_web_page_previews=False))         if duration == 0:             time_str = "навсегда"         elif duration < 60:             time_str = f"{duration} с"         elif duration < 3600:             time_str = f"{duration//60} м"         elif duration < 86400:             time_str = f"{duration // 3600} ч"         else:             days = duration // 86400             time_str = f"{days} дн"         text = f"Мут ({time_str})"      elif action == "кик":         await message.bot.ban_chat_member(chat_id, user_id, until_date=datetime.now() + timedelta(seconds=30))         text = "Кик"      elif action == "предупреждение":         cursor.execute("INSERT INTO warnings (user_id, group_id, reason, moderator_id) VALUES (?, ?, ?, ?)", (user_id, chat_id, reason, message.bot.id))         connection.commit()         text = "Предупреждение"     return text 

Можем попробовать запустить бота. Вводим команду:

python main.py

В консоли мы увидим следующий вывод:

Это означает, что бот успешно запустился и готов к работе. Создадим тестовую группу, добавим бота с правами администратора и пригласим пользователя для тестирования нарушений.

Вот так выглядит вывод в логах, когда сообщение было оставлено:

А так выглядит вывод при обнаружении нарушения правил:

Как можно видеть, бот успешно заблокировал пользователя и удалил его сообщение:

За кадром мы также протестировали другие виды наказаний, и все они работают корректно.

Однако ограничиваться только автоматической выдачей наказаний нельзя, поэтому добавим команды для ручного управления: /ban,/unban, /mute и /unmute.

Для начала реализуем функцию для парсинга времени:

def pars_time(time: str | None) -> datetime | None:     if not time:         return None      match = re.match(r"(\d+)([a-z])", time.lower())     if match:         value = int(match.group(1))         unit = match.group(2)         match unit:             case "h": delta = timedelta(hours=value)             case "d": delta = timedelta(days=value)             case _: return None     else:         return None      new = datetime.now() + delta     return new 

Теперь последовательно добавим команды модерации:

Команда /ban — блокировка пользователя:

@dp.message(Command("ban")) async def mute(message: Message, command: CommandObject | None = None) -> None:     if message.chat.type not in ["group", "supergroup"]:         await message.answer("Эта команда работает только в групповых чатах!")         return      reply = message.reply_to_message     if not reply:         return None      date = pars_time(command.args)     with suppress(TelegramBadRequest):         await bot.ban_chat_member(chat_id=message.chat.зid, user_id=reply.from_user.id, until_date=date)         await message.answer(f"Модератор {message.from_user.mention_markdown(message.from_user.username)} заблокировал пользователя {reply.from_user.mention_markdown(reply.from_user.username)}") 

Команда /unban — разблокировка пользователя:

@dp.message(Command("unban")) async def mute(message: Message):     if message.chat.type not in ["group", "supergroup"]:         await message.answer("Эта команда работает только в групповых чатах!")         return      reply = message.reply_to_message     if not reply:         return None      with suppress(TelegramBadRequest):         await bot.unban_chat_member(chat_id=message.chat.id, user_id=reply.from_user.id)         await message.answer(f"Модератор {message.from_user.mention_markdown(message.from_user.username)} разблокировал пользователя {reply.from_user.mention_markdown(reply.from_user.username)}") 

Команда /mute — ограничение возможности отправки сообщений:

@dp.message(Command("mute")) async def mute(message: Message, command: CommandObject | None = None) -> None:     if message.chat.type not in ["group", "supergroup"]:         await message.answer("Эта команда работает только в групповых чатах!")         return      reply = message.reply_to_message     if not reply:         return None      date = pars_time(command.args)     with suppress(TelegramBadRequest):         await bot.restrict_chat_member(chat_id=message.chat.id, user_id=reply.from_user.id, until_date=date, permissions=ChatPermissions(can_send_messages=False))         await message.answer(f"Модератор {message.from_user.mention_markdown(message.from_user.username)} выдал мут пользователю {reply.from_user.mention_markdown(reply.from_user.username)}") 

Команда /unmute — снятие ограничений:

@dp.message(Command("unmute")) async def mute(message: Message, command: CommandObject | None = None) -> None:     if message.chat.type not in ["group", "supergroup"]:         await message.answer("Эта команда работает только в групповых чатах!")         return      reply = message.reply_to_message     if not reply:         return None      with suppress(TelegramBadRequest):         await bot.restrict_chat_member(chat_id=message.chat.id, user_id=reply.from_user.id, permissions=ChatPermissions(can_send_messages=True))         await message.answer(f"Модератор {message.from_user.mention_markdown(message.from_user.username)} снял мут пользователю {reply.from_user.mention_markdown(reply.from_user.username)}") 

Сохраняем наш файл и проверяем работу, предварительно запустив бота командой:

python main.py

Проверим выдачу и снятие мута:

И проверим выдачу и снятие блокировки:

Отлично! Наш бот полностью готов к использованию. Осталось задеплоить его на хостинг Amvera Cloud.

Деплой бота модератора

Развернём наш скрипт для модерации в Amvera, воспользовавшись встроенным CI/CD. В отличии от VPS, Amvera предоставляет движек приложений, максимально упрощающий развёртывание и администрирование проекта.

Мы рассмотрим 2 способа деплоя:

• Через веб-интерфейс Amvera

• Через Git push

Для начала создадим файл зависимостей requirements.txt:

aiogram==3.21.0 requests==2.32.4

Это минимальный вариант файла — дополнительные зависимости будут установлены автоматически. Вы также можете создать полный файл зависимостей с помощью команды:

pip freeze

Теперь создадим конфигурационный файл amvera.yml:

version: null meta:   environment: python   toolchain:     name: pip     version: "3.12" build:   requirementsPath: requirements.txt run:   scriptName: main.py   command: null   persistenceMount: /data   containerPort: 80 

Сделать это можно в разделе Конфигурация, выбрав нужные параметры.

Также обязательно убедитесь, что в файле main.py используется правильный путь к базе данных:

connection = sqlite3.connect("/data/database.db", check_same_thread=False)

Это критически важно, так как при неправильном пути бот не сможет найти базу данных и не запустится.

Теперь приступим непосредственно к деплою. Начнём с деплоя через веб-интерфейс.

Идем на сайт Amvera Cloud заходим в свой личный кабиент и идем в раздел Приложения. Создайте новый проект, укажите название и выберите подходящий тарифный план. На этапе загрузки данных выберите опцию «Через интерфейс» и загрузите необходимые файлы (не включая папку data).

На этапе настройки переменных окружения вы можете добавить переменные для токена бота и токена LLM. Для этого предварительно удалите из файла main.py прямые указания токенов:

BOT_TOKEN = "" # ТОКЕН ВАШЕГО БОТА AUTH_TOKEN = "" # ТОКЕН LLM В ФОРМАТЕ Bearer XXXXXXXX

Теперь рассмотрим вариант деплоя через Git.

Откройте терминал в папке с проектом и выполните команду:

git init

Создайте приложение, но на этапе загрузки файлов нажмите «Отменить».

Перейдите в раздел созданного приложения и в разделе «Репозиторий» найдите команду «Подключить к существующему репозиторию». Скопируйте и выполните ее в терминале. Команда будет иметь следующий формат:

git remote add amvera https://git.amvera.ru/имя пользователя/имя проекта;

Эта команда подключит удаленный репозиторий проекта к вашему локальному репозиторию. Теперь добавьте файлы в отслеживаемые:

git add amvera.yml main.py requirements.txt

Выполните коммит изменений:

git commit -m "Обновление 1.0"

Теперь отправьте изменения на сервер Amvera (сборка начнется автоматически):

git push amvera master

Появится запрос на ввод учетных данных:

Введите имя пользователя и пароль от вашего аккаунта Amvera Cloud.

Отлично! Бот успешно запустился в обоих случаях:

На этом наш деплой законечен!

Заключение

Мы успешно реализовали интеллектуального Telegram-бота-модератора, который использует LLM Amvera Cloud для семантического анализа сообщений. В отличие от примитивных фильтров по ключевым словам, наша система способна понимать контекст, учитывать специфические правила каждого чата и автоматически применять адекватные санкции к нарушителям.

Такой подход позволяет:

• Снизить нагрузку на администраторов

• Повысить качество модерации

• Гибко адаптировать поведение бота под разные сообщества

• Обеспечить проактивное выявление нарушений до эскалации конфликтов

На практике это означает, что администраторам больше не нужно постоянно мониторить чат вручную и разбирать каждый спорный случай — бот берет эту рутинную работу на себя, позволяя модераторам сосредоточиться на стратегических задачах. В результате чат становится более безопасным и комфортным пространством, а пользователи — более дисциплинированными, зная, что нарушения не останутся незамеченными.

И главное, в сравнении с классическим ML, модель не нужно переобучать, достаточно просто изменить промт.

Код проекта доступен по ссылке на GitHub.

---

Релевантные материалы:

• Простой бот-модератор на Aiogram 3.x

• Антиспам бот-модератор в телеграм. Защищаем группу за три минуты