- Регистрация
- 23 Август 2023
- Сообщения
- 2 822
- Лучшие ответы
- 0
- Реакции
- 0
- Баллы
- 51
Offline
Современные онлайн-чаты, особенно в Telegram, сталкиваются с серьезными проблемами токсичного контента, спама и оскорблений. Традиционные фильтры по ключевым словам давно устарели: пользователи легко обходят их с помощью замены букв, использования смайликов или специальных символов. В результате ручная модерация отнимает много времени, а автоматическая часто оказывается неточной и излишне жесткой.
Мы уже публиковали туториалы по созданию ботов модераторов:
Но прошлые статьи использовали классические ML-модели, что делало их менее гибкими.
В этой статье я рассмотрю пример, в котором будут использоваться нейронки для выполнения задач модерации.
Большие языковые модели (LLM) способны понимать контекст и смысл сообщения, а не просто искать запрещённые слова. И в отличии от классического ML их не нужно переобучать, достаточно просто изменить промпт. Это делает их идеальным инструментом для интеллектуальной модерации. Вместо примитивных фильтров бот с 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. Выбираем одну из четырёх доступных моделей и копирем токен.
Подключаем GPT 5 в Amvera Cloud
Обратите внимание: для использования токена необходимо активировать подписку. В разделе LLM (Preview) найдите кнопку «Подключить пакет токенов» и выберите подходящий тарифный план. Для тестов нам подойдёт бесплатный пакет на 20 000 токенов.
Теперь, имея все необходимые токены, можно приступить к разработке.
Архитектура бота модератора
Как же будет работать система модерации? Процесс модерации организован следующим образом:
Как вы могли заметить, в Amvera Cloud доступны четыре модели:
Для тестирования мы будем использовать 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 способа деплоя:
Для начала создадим файл зависимостей 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.
Отлично! Бот успешно запустился в обоих случаях:
Запущено через git
Запущено через интерфейс
На этом наш деплой законечен!
Заключение
Мы успешно реализовали интеллектуального Telegram-бота-модератора, который использует LLM Amvera Cloud для семантического анализа сообщений. В отличие от примитивных фильтров по ключевым словам, наша система способна понимать контекст, учитывать специфические правила каждого чата и автоматически применять адекватные санкции к нарушителям.
Такой подход позволяет:
На практике это означает, что администраторам больше не нужно постоянно мониторить чат вручную и разбирать каждый спорный случай — бот берет эту рутинную работу на себя, позволяя модераторам сосредоточиться на стратегических задачах. В результате чат становится более безопасным и комфортным пространством, а пользователи — более дисциплинированными, зная, что нарушения не останутся незамеченными.
И главное, в сравнении с классическим ML, модель не нужно переобучать, достаточно просто изменить промт.
Релевантные материалы:
Мы уже публиковали туториалы по созданию ботов модераторов:
Простой бот-модератор на 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. Выбираем одну из четырёх доступных моделей и копирем токен.
Подключаем GPT 5 в Amvera Cloud
Обратите внимание: для использования токена необходимо активировать подписку. В разделе 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"Правила этой группы:\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 > 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.
Отлично! Бот успешно запустился в обоих случаях:
Запущено через git
Запущено через интерфейс
На этом наш деплой законечен!
Заключение
Мы успешно реализовали интеллектуального Telegram-бота-модератора, который использует LLM Amvera Cloud для семантического анализа сообщений. В отличие от примитивных фильтров по ключевым словам, наша система способна понимать контекст, учитывать специфические правила каждого чата и автоматически применять адекватные санкции к нарушителям.
Такой подход позволяет:
Снизить нагрузку на администраторов
Повысить качество модерации
Гибко адаптировать поведение бота под разные сообщества
Обеспечить проактивное выявление нарушений до эскалации конфликтов
На практике это означает, что администраторам больше не нужно постоянно мониторить чат вручную и разбирать каждый спорный случай — бот берет эту рутинную работу на себя, позволяя модераторам сосредоточиться на стратегических задачах. В результате чат становится более безопасным и комфортным пространством, а пользователи — более дисциплинированными, зная, что нарушения не останутся незамеченными.
И главное, в сравнении с классическим ML, модель не нужно переобучать, достаточно просто изменить промт.
Релевантные материалы:
