Современные онлайн-чаты, особенно в 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

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