Современные онлайн-чаты, особенно в Telegram, сталкиваются с серьезными проблемами токсичного контента, спама и оскорблений. Традиционные фильтры по ключевым словам давно устарели: пользователи легко обходят их с помощью замены букв, использования смайликов или специальных символов. В результате ручная модерация отнимает много времени, а автоматическая часто оказывается неточной и излишне жесткой.
Мы уже публиковали туториалы по созданию ботов модераторов:
Но прошлые статьи использовали классические 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 токенов.
Теперь, имея все необходимые токены, можно приступить к разработке.
Архитектура бота модератора
Как же будет работать система модерации? Процесс модерации организован следующим образом:
-
Пользователь отправляет сообщение в чат.
-
Бот перехватывает сообщение и отправляет его текст на проверку в LLM.
-
Модель анализирует текст на наличие нарушений, основываясь на установленных правилах беседы.
-
В зависимости от ответа модели, бот принимает решение:
-
Если сообщение нарушает правила, то бот выдает соотвествующее наказание, либо удаляет сообщение.
-
Если в сообщении не будет нарушений, то ничего делать не нужно.
-
Как вы могли заметить, в 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"Правила этой группы:nn{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. В отличии от , Amvera предоставляет движек приложений, максимально упрощающий развёртывание и администрирование проекта.
Мы рассмотрим 2 способа деплоя:
-
Через веб-интерфейс Amvera
-
Через Git push
Для начала создадим файл зависимостей requirements.txt:
aiogram==3.21.0
requests==2.32.4Это минимальный вариант файла — дополнительные зависимости будут установлены автоматически. Вы также можете создать полный файл зависимостей с помощью команды:
pip freeze > requirements.txtТеперь создадим конфигурационный файл 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, модель не нужно переобучать, достаточно просто изменить промт.
Релевантные материалы:
Автор: NikitinIlya
