AI-автоответы на комментарии и личные сообщения
SMM-менеджер тратит 2–3 часа в день на однотипные ответы: «спасибо за комментарий», «напишите нам на почту», «да, доставляем в ваш регион». При объёме 100+ сообщений в день — это механическая работа, которую AI закрывает полностью. Задача не в том, чтобы заменить общение, а в том, чтобы убрать 70–80% рутины и дать человеку заниматься сложными случаями.
Классификация и роутинг сообщений
Прежде чем отвечать — нужно понять, что перед нами: простой вопрос о продукте, жалоба, токсичный комментарий или вопрос, требующий экспертизы.
from anthropic import Anthropic
import json
from enum import Enum
client = Anthropic()
class MessageCategory(str, Enum):
PRODUCT_QUESTION = "product_question" # вопрос о продукте/услуге
ORDER_STATUS = "order_status" # статус заказа
COMPLAINT = "complaint" # жалоба
PRAISE = "praise" # похвала/благодарность
TOXIC = "toxic" # агрессия/токсичность
IRRELEVANT = "irrelevant" # спам, нерелевантно
REQUIRES_HUMAN = "requires_human" # сложный кейс, нужен человек
SALES_INQUIRY = "sales_inquiry" # интерес к покупке
class MessageClassifier:
CLASSIFY_PROMPT = """Классифицируй сообщение клиента в социальных сетях.
Верни JSON:
{
"category": "product_question|order_status|complaint|praise|toxic|irrelevant|requires_human|sales_inquiry",
"sentiment": "positive|neutral|negative",
"urgency": "low|medium|high",
"language": "ru|en|uk|other",
"auto_reply_possible": true/false,
"extracted_entities": {
"product": "упомянутый продукт или null",
"order_number": "номер заказа или null",
"location": "город/регион или null"
}
}
Только JSON."""
def classify(self, message: str, context: dict = {}) -> dict:
"""Классифицирует входящее сообщение"""
context_text = ""
if context.get("post_text"):
context_text = f"\nКонтекст поста: {context['post_text'][:200]}"
if context.get("username"):
context_text += f"\nОт пользователя: {context['username']}"
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=256,
messages=[{
"role": "user",
"content": f"{self.CLASSIFY_PROMPT}{context_text}\n\nСообщение: {message}"
}],
)
text = response.content[0].text
return json.loads(text[text.find("{"):text.rfind("}") + 1])
class AutoReplyGenerator:
"""Генератор автоответов с учётом бренд-голоса"""
def __init__(self, brand_config: dict):
"""
brand_config: {
name: "Название компании",
tone: "дружелюбный и неформальный" | "профессиональный",
products: ["продукт 1", "продукт 2"],
faq: {"вопрос": "ответ"},
escalation_contact: "email или телефон",
do_not_say: ["список запрещённых фраз"],
}
"""
self.brand = brand_config
def generate_reply(
self,
message: str,
category: str,
platform: str = "instagram",
context: dict = {},
) -> dict:
"""Генерирует ответ с учётом категории и платформы"""
# Ограничения по платформе
max_length = {
"instagram": 2200,
"vk": 4096,
"telegram": 4096,
"twitter": 280,
"facebook": 8000,
}.get(platform, 1000)
faq_text = "\n".join([f"Q: {q}\nA: {a}" for q, a in self.brand.get("faq", {}).items()])
system = f"""Ты — SMM-менеджер бренда «{self.brand['name']}».
Тон общения: {self.brand.get('tone', 'дружелюбный')}.
Платформа: {platform} (максимум {max_length} символов).
Продукты компании: {', '.join(self.brand.get('products', []))}.
Контакт для эскалации: {self.brand.get('escalation_contact', '')}.
FAQ:
{faq_text}
НЕ ГОВОРИ: {', '.join(self.brand.get('do_not_say', []))}.
Правила:
- Будь конкретным, не шаблонным
- Обращайся к пользователю на "вы" если не знаешь его возраст
- При жалобах — сначала признай проблему, потом предлагай решение
- Не обещай то, что не можешь выполнить
- Для сложных вопросов — перенаправляй в личку или на email"""
messages_list = []
if context.get("post_text"):
messages_list.append({
"role": "user",
"content": f"Контекст поста: {context['post_text']}"
})
messages_list.append({
"role": "assistant",
"content": "Понял контекст поста, готов отвечать на комментарии."
})
messages_list.append({
"role": "user",
"content": f"Категория: {category}\nСообщение: {message}\n\nНапиши ответ."
})
response = client.messages.create(
model="claude-haiku-4-5",
max_tokens=512,
system=system,
messages=messages_list,
)
reply = response.content[0].text.strip()
# Проверяем длину и обрезаем если нужно
if len(reply) > max_length:
reply = reply[:max_length - 3] + "..."
return {
"reply": reply,
"length": len(reply),
"platform": platform,
"requires_review": category in ("complaint", "requires_human", "toxic"),
}
Полный pipeline обработки
import asyncio
from datetime import datetime
class SocialMediaAutoReply:
"""Полная система автоответов для соцсетей"""
def __init__(self, brand_config: dict):
self.classifier = MessageClassifier()
self.generator = AutoReplyGenerator(brand_config)
self.pending_review: list = [] # очередь на ручную проверку
async def process_message(self, message_data: dict) -> dict:
"""
message_data: {
id, text, platform, post_text, username, timestamp
}
"""
text = message_data["text"]
platform = message_data.get("platform", "instagram")
# 1. Классификация
classification = self.classifier.classify(
text,
context={"post_text": message_data.get("post_text"), "username": message_data.get("username")}
)
result = {
"message_id": message_data["id"],
"classification": classification,
"action": None,
"reply": None,
"processed_at": datetime.now().isoformat(),
}
# 2. Роутинг
if classification["category"] == "toxic":
result["action"] = "hide" # скрыть комментарий
result["reply"] = None
elif classification["category"] == "irrelevant":
result["action"] = "skip" # игнорировать
elif not classification.get("auto_reply_possible", True):
result["action"] = "escalate_to_human"
self.pending_review.append({**message_data, "classification": classification})
else:
# 3. Генерация ответа
reply_data = self.generator.generate_reply(
message=text,
category=classification["category"],
platform=platform,
context={"post_text": message_data.get("post_text")},
)
if reply_data.get("requires_review"):
result["action"] = "reply_with_review"
self.pending_review.append({**message_data, "suggested_reply": reply_data["reply"]})
else:
result["action"] = "auto_reply"
result["reply"] = reply_data["reply"]
return result
async def process_batch(self, messages: list[dict]) -> list[dict]:
"""Обрабатывает пачку сообщений параллельно"""
semaphore = asyncio.Semaphore(10)
async def process_one(msg):
async with semaphore:
return await self.process_message(msg)
return await asyncio.gather(*[process_one(m) for m in messages])
Интеграция с VK API и Instagram
import vk_api
import requests
class VKAutoReply:
"""Интеграция с ВКонтакте"""
def __init__(self, vk_token: str, brand_config: dict):
vk_session = vk_api.VkApi(token=vk_token)
self.vk = vk_session.get_api()
self.auto_reply = SocialMediaAutoReply(brand_config)
async def process_new_comments(self, group_id: int, since_timestamp: int):
"""Обрабатывает новые комментарии к постам группы"""
posts = self.vk.wall.get(owner_id=-group_id, count=10, filter="owner")
for post in posts["items"]:
comments = self.vk.wall.getComments(
owner_id=-group_id,
post_id=post["id"],
filter="all",
count=100,
)
new_comments = [
c for c in comments["items"]
if c["date"] > since_timestamp and c.get("from_id", 0) > 0 # не от группы
]
for comment in new_comments:
result = await self.auto_reply.process_message({
"id": comment["id"],
"text": comment["text"],
"platform": "vk",
"post_text": post.get("text", "")[:300],
"username": str(comment["from_id"]),
})
if result["action"] == "auto_reply" and result["reply"]:
self.vk.wall.createComment(
owner_id=-group_id,
post_id=post["id"],
reply_to_comment=comment["id"],
message=result["reply"],
)
elif result["action"] == "hide":
self.vk.wall.deleteComment(
owner_id=-group_id,
comment_id=comment["id"],
)
Практический кейс: интернет-магазин косметики
Ситуация: 3 площадки (Instagram, ВКонтакте, Telegram-канал). Ежедневно 180–250 комментариев и DM. Один SMM-специалист + 0.5 ставки помощника.
Результаты за 6 недель:
- Автоматически закрыто 73% сообщений (ответы на вопросы о составе, доставке, наличии)
- Среднее время ответа: 4–6 часов → 3 минуты
- Жалобы с задержкой ответа >2 часов: снизились на 89%
- SMM-специалист переключился на создание контента, перестав тонуть в ответах
Нюанс: первые 2 недели пришлось вручную редактировать ~15% ответов — модель путалась с конкретными артикулами и ценами. Решили добавив FAQ с актуальными ценами и перечнем SKU в brand_config.
Сроки
- Классификатор + генератор ответов: 3–5 дней
- Интеграция VK / Instagram API: 3–5 дней каждая
- Дашборд модерации для команды: 1 неделя
- Полная система с аналитикой: 3–4 недели