AI-система триггерных коммуникаций: email, SMS, push

Проблема: упущенные продажи из-за отсутствия своевременных коммуникаций

Интернет-магазин с 50 000 посетителей в день теряет до 70% потенциальных продаж из-за задержки коммуникаций. Каждый день десятки брошенных корзин, сотни пользователей уходят без покупки. Ручные рассылки не успевают — момент упущен. Мы разрабатываем AI-систему триггерных коммуникаций, которая анализирует поведение каждого пользователя в real-time и отправляет персонализированное сообщение в нужный канал в оптимальное время. Результат — конверсия из триггеров растёт в 2-4 раза, retention увеличивается на 30%.

Триггерная система на основе ML решает три задачи: выбор канала (email/SMS/push), определение времени отправки и генерацию контента. В отличие от статичных правил, AI адаптируется под каждого пользователя — его паттерны активности, предпочтения и историю покупок. Мы используем LLM (Claude, GPT-4) для создания текста, который звучит как от реального менеджера, а не шаблонная рассылка. Система демонстрирует open rate до 55% для брошенных корзин и CTR 15-25% для персонализированных push.

Как ML-модель выбирает канал и время отправки?

ML-модель учитывает множество факторов: историческую эффективность канала для данного типа события, предпочтения пользователя, текущую нагрузку на канал. Например, для брошенной корзины email работает лучше всего (open rate 40-55%), а для срочных уведомлений об изменении цены — push или SMS. Время отправки определяется на основе паттернов активности: если пользователь чаще открывает письма вечером, система отправит сообщение к 18:00. Модель обучается на исторических данных о взаимодействиях, используя признаки: час дня, день недели, тип события, предыдущие отклики. Мы применяем градиентный бустинг (CatBoost) для задачи регрессии времени отправки и классификацию канала.

Код: базовый класс системы с выбором канала и времени

from anthropic import Anthropic
import pandas as pd
import numpy as np
from dataclasses import dataclass
from enum import Enum
import json

class Channel(Enum):
    EMAIL = "email"
    SMS = "sms"
    PUSH = "push"
    IN_APP = "in_app"

@dataclass
class TriggerEvent:
    user_id: str
    event_type: str
    event_data: dict
    timestamp: float

class TriggerCommunicationSystem:
    def __init__(self):
        self.llm = Anthropic()
        self.send_time_model = None
        self.channel_model = None

    def process_trigger(self, event: TriggerEvent, user_profile: dict) -> dict:
        """Полная обработка триггера: канал + время + контент"""
        channel = self._select_channel(user_profile, event.event_type)
        send_delay_hours = self._optimal_send_time(user_profile, event.event_type)
        content = self._generate_content(event, user_profile, channel)
        if self._is_communication_fatigue(user_profile):
            return {'send': False, 'reason': 'communication_fatigue'}
        return {
            'send': True,
            'channel': channel.value,
            'send_delay_hours': send_delay_hours,
            'content': content,
            'event_type': event.event_type
        }

    def _select_channel(self, user: dict, event_type: str) -> Channel:
        preferred = user.get('preferred_channel')
        if preferred:
            return Channel[preferred.upper()]
        channel_effectiveness = {
            'abandoned_cart': {'email': 0.15, 'push': 0.08, 'sms': 0.12},
            'inactivity': {'email': 0.05, 'push': 0.06, 'sms': 0.04},
            'price_drop': {'push': 0.12, 'email': 0.10, 'sms': 0.08},
            'order_shipped': {'sms': 0.25, 'email': 0.20, 'push': 0.18},
        }
        effectiveness = channel_effectiveness.get(event_type, {'email': 0.1})
        best_channel = max(effectiveness, key=effectiveness.get)
        return Channel[best_channel.upper()]

    def _optimal_send_time(self, user: dict, event_type: str) -> float:
        active_hours = user.get('active_hours', list(range(9, 22)))
        if event_type == 'abandoned_cart':
            return 1.5
        elif event_type == 'price_drop':
            return 0.1
        elif event_type == 'inactivity':
            return 24 if 9 in active_hours else 48
        else:
            return 2.0

    def _generate_content(self, event: TriggerEvent, user: dict, channel: Channel) -> dict:
        channel_constraints = {
            Channel.SMS: {'max_chars': 160, 'format': 'plain'},
            Channel.PUSH: {'max_chars': 100, 'format': 'title+body'},
            Channel.EMAIL: {'max_chars': 2000, 'format': 'html'},
            Channel.IN_APP: {'max_chars': 200, 'format': 'markdown'},
        }
        constraint = channel_constraints[channel]
        event_context = json.dumps(event.event_data, ensure_ascii=False)[:300]
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=300,
            messages=[{
                "role": "user",
                "content": f"""Generate a {channel.value} message for trigger event.

Event: {event.event_type}
Event data: {event_context}
User name: {user.get('first_name', 'Customer')}
User purchase history: {user.get('total_orders', 0)} orders, avg order ${user.get('avg_order_value', 0):.0f}
Channel: {channel.value} (max {constraint['max_chars']} chars)

Return JSON:
{{
  "subject": "email subject or push title",
  "body": "message body",
  "cta": "call to action text",
  "cta_url": "URL path"
}}

Tone: friendly, personal. Mention specific item if available. No generic marketing language."""
            }]
        )
        try:
            return json.loads(response.content[0].text)
        except Exception:
            return {
                'subject': f"We have something for you, {user.get('first_name', '')}!",
                'body': response.content[0].text[:constraint['max_chars']],
                'cta': 'View Now'
            }

    def _is_communication_fatigue(self, user: dict) -> bool:
        messages_last_7d = user.get('messages_received_7d', 0)
        opens_last_7d = user.get('messages_opened_7d', 0)
        if messages_last_7d >= 5:
            return True
        if messages_last_7d >= 3 and opens_last_7d == 0:
            return True
        return False

A/B тест триггерных сообщений: сравнение вариантов

Для подбора оптимальных стратегий мы запускаем A/B тесты по разным каналам, времени и содержанию. Результаты фиксируются в реальном времени — можно оперативно изменять параметры. Например, тест может показать, что для вашей аудитории push с эмодзи даёт CTR на 20% выше, чем без них.

class TriggerABTest:
    """Тестирование вариантов триггерных сообщений"""

    def __init__(self, test_name: str, variants: list[dict]):
        self.test_name = test_name
        self.variants = variants
        self.results = {v['name']: {'sent': 0, 'opened': 0, 'clicked': 0, 'converted': 0}
                        for v in variants}

    def assign_variant(self, user_id: str) -> dict:
        """Детерминированное назначение варианта"""
        idx = hash(f"{self.test_name}_{user_id}") % len(self.variants)
        return self.variants[idx]

    def compute_results(self) -> dict:
        results_summary = {}
        for variant_name, stats in self.results.items():
            sent = stats['sent']
            if sent == 0:
                continue
            results_summary[variant_name] = {
                'open_rate': stats['opened'] / sent,
                'click_rate': stats['clicked'] / sent,
                'conversion_rate': stats['converted'] / sent,
                'sample_size': sent
            }
        return results_summary

Сравнение эффективности каналов

Rule-based vs ML: что эффективнее?

Rule-based подходит для простых сценариев с низкой вариативностью. ML — для сложных, где важна персонализация. На практике мы часто комбинируем: rule-based как fallback, ML как основной движок.

Ключевые преимущества AI-системы

Персонализация контента с помощью LLM — модель генерирует текст с учётом истории покупок, текущего просмотра и имени пользователя. Open rate в 2-3 раза выше, чем у массовых рассылок. ML-оптимизация времени — модель анализирует, когда конкретный пользователь активен, и отправляет сообщение в момент максимальной вероятности прочтения. Контроль частоты — встроенный детектор коммуникационной усталости не даёт отправить более 3-4 сообщений в неделю, предотвращая отписки. В результате retention растёт, а количество жалоб на спам снижается в 5-10 раз.

Что входит в разработку системы?

1. Аналитика — аудит текущих коммуникаций, сбор паттернов активности пользователей.
2. Проектирование — архитектура event-driven системы, выбор стека (PyTorch, Hugging Face, ChromaDB).
3. Разработка — реализация ML-модулей выбора канала и времени, генерация контента через LLM.
4. Интеграция — подключение к CRM (Bitrix24, AmoCRM), ESP (SendGrid, Unisender) и SMS-шлюзам.
5. A/B тестирование — запуск экспериментов для подбора оптимальных стратегий.
6. Документация и обучение — API-документация, инструкции для маркетологов.

Мы имеем 5+ лет опыта в ML-системах для e-commerce, реализовали более 30 проектов. Используем современный стек: PyTorch, LangChain, Triton Inference Server. Гарантируем измеримый результат — фиксируем метрики open rate, conversion и сравниваем с baseline. Внедрение окупается в среднем за 2-3 месяца, принося дополнительную выручку от $50 000 до $200 000 в год для среднего e-commerce проекта.

Оценим ваш проект за 1-2 дня — свяжитесь с нами для консультации. Получите расчёт стоимости и сроков под ваши задачи. Дополнительная выручка от AI-системы может превысить затраты в 5-10 раз в первый же год.

Разработка рекомендательных систем: от collaborative filtering до real-time serving

На одном проекте для e-commerce с каталогом 300k SKU мы подняли CTR с 1,8% до 4,4% — в 2,4 раза. Первый рывок дала коллаборативная фильтрация вместо «популярное за последние 7 дней», второй — добавление контентных признаков и re-ranking. Разница между «показываем популярное» и «показываем персонализированное» — измеримая и существенная. Ниже — инженерный опыт, который помог это сделать, и архитектуры, которые реально работают в продакшене.

Collaborative Filtering: матричная факторизация и нейронные подходы

Matrix Factorization — классика для implicit feedback (клики, просмотры, покупки без явного рейтинга). ALS (Alternating Least Squares) в библиотеке Implicit обрабатывает матрицы user×item с сотнями миллионов ненулевых значений за минуты на GPU. Latent factors 64–256, регуляризация λ=0.01–0.1 — стартовые параметры. Проблема cold start: для нового пользователя или товара нет истории — классический CF беспомощен, нужны контентные признаки или гибрид.

Neural Collaborative Filtering (NCF) заменяет скалярное произведение на нейросеть. На практике выигрыш над хорошо настроенным ALS умеренный, но NCF проще расширять дополнительными признаками (возраст, категория, время суток). Sequence-aware модели (SASRec, BERT4Rec) учитывают порядок взаимодействий — state-of-the-art для сессионных рекомендаций.

Как выбрать архитектуру рекомендательной системы?

Ответ зависит от данных, нагрузки и требований к холодному старту. Ниже — три основных подхода с критериями выбора.

Гибридная архитектура на 20–40% эффективнее чистого CF по покрытию long-tail — проверено на каталогах от 100k SKU.

Content-Based Filtering: когда истории взаимодействий мало

Content-based рекомендует на основе характеристик товаров, а не поведения других пользователей — решает cold start для новых items. Текстовые эмбеддинги через sentence-transformers (multilingual-e5-base, BGE-M3) → поиск похожих через FAISS IndexFlatIP — запрос за <5 ms на 100k товаров. Item2Vec (Word2Vec на последовательностях просмотров) даёт интерпретируемые «похожие товары» за пару часов обучения.

Структурированные признаки (категория, бренд, цена) подаются через embedding layers или в gradient boosting — CatBoost работает с категориями без ручного кодирования.

Почему гибридные модели работают лучше?

Production-системы почти всегда двухуровневые. Stage 1 (Retrieval) — быстрый отбор 100–500 кандидатов из 300k товаров через ALS или Two-Tower модель с векторным поиском (FAISS, Qdrant). Stage 2 (Ranking) — тяжёлый ранжировщик на LightGBM или нейросети с cross-features, временем, устройством и контекстом сессии. LightFM — хорошая отправная точка для среднего масштаба без тяжёлой инфраструктуры. Наша практика показывает: переход от single-stage к two-stage даёт прирост точности на 15–25% при росте latency всего на 20–30 мс.

Real-Time Serving: архитектура под нагрузку

Latency SLA — 50–100 ms при тысячах запросов в секунду. Base-рекомендации precompute (batch job раз в час) → Redis по user_id → <5 ms. Real-time re-ranking через Kafka для событий (клики, добавления в корзину) → обновление контекстных признаков. Feature serving — Redis с TTL (число просмотров за 24 часа, последний кликнутый item). При нагрузке 10k req/s ставим Redis Cluster с репликацией.

A/B тестирование — единственный достоверный способ оценить улучшения. Офлайн-метрики коррелируют с онлайн не всегда. Kohavi et al., «Online Controlled Experiments at Large Scale» (KDD 2013) — обязательное чтение для команды. Тест с 5–10% трафика, мониторинг CTR, конверсии, revenue per session. Одна из наших клиентских систем после гибридизации увеличила выручку на 18% за месяц A/B.

Сроки разработки рекомендательной системы

Этапы и типичные временные затраты — в таблице ниже. Стоимость рассчитывается индивидуально под масштаб каталога и требования к latency.

Что входит в разработку под ключ

1. Аудит данных — плотность матрицы user×item (обычно <0,1%), распределение активности, temporal паттерны, cold start статистика.
2. Baseline — «популярное» как простой порог, который часто трудно обогнать.
3. Итеративное улучшение — ALS → контентные признаки → two-stage → sequence-aware. Каждый шаг с A/B.
4. Инфраструктура serving — batch precomputation, Redis, real-time re-ranking, мониторинг в Grafana.
5. Документация — model card с метриками, инструкция по деплою, описание признаков.
6. Обучение команды — сессия по интерпретации результатов и дообучению модели.
7. Поддержка — 1 месяц после запуска (фикс инцидентов, донастройка pipeline).

Мы — команда с 7+ годами опыта в рекомендательных системах, реализовали более 30 проектов для e-commerce и медиа. Гарантируем прозрачное A/B‑тестирование и фиксацию улучшения метрик.

Хотите оценить потенциал роста вашего каталога? Свяжитесь с нами для бесплатного аудита данных. Закажите разработку рекомендательной системы — первый прототип в течение двух недель.

from implicit.als import AlternatingLeastSquares

model = AlternatingLeastSquares(
    factors=64,
    regularization=0.05,
    iterations=15,
    use_gpu=True
)
model.fit(user_item_matrix)

Больше о математике рекомендательных систем — в Wikipedia.