Розробка AI-системи автоматизації email-маркетингу

Більшість email-розсилок працюють за жорстким сценарієм: «через 3 дні після реєстрації надіслати лист N». Ми бачимо, як 80% таких листів не відкривають, а 90% відкриттів ігнорують CTA. Проблема в тому, що статичні послідовності не враховують поведінку конкретного користувача. AI-автоматизація адаптує вміст, час відправки та шлях кожного отримувача індивідуально. Різниця: 25–40% приріст click-through rate та 2–3x покращення конверсії. Впровадження займає від 2 до 6 тижнів під ключ.

Які проблеми вирішуємо

Низька релевантність листів. Без персоналізації контенту — 70% користувачів відписуються в перший тиждень. AI аналізує історію взаємодій та підбирає теми, які точно цікаві саме цій людині.

Неоптимальний час відправки. Відправка о 10 ранку для всієї бази — минуле століття. Ми будуємо модель, яка передбачає для кожного отримувача годину та день з максимальною ймовірністю кліку. У тестах це дає +12% до CTR та економію до 30% бюджету на розсилках.

Слабкі теми листів. A/B тестування вручну — довго та неефективно. LLM генерує 5–10 варіантів теми для кожного сегмента, а алгоритм обирає переможця за прогнозом відкриття.

Як AI-персоналізація підвищує конверсію?

Ми впровадили систему для SaaS-продукту з 50 000 користувачів. Через місяць open rate виріс з 22% до 43%, CTR — з 3% до 11%. Ключовий елемент — прогнозування оптимального часу відправки на основі градієнтного бустингу (Scikit-learn documentation).

import pandas as pd
import numpy as np
from sklearn.ensemble import GradientBoostingRegressor
from anthropic import Anthropic
import json

class SendTimeOptimizer:
    """Оптимальное время отправки для каждого получателя"""

    def __init__(self):
        self.model = GradientBoostingRegressor(
            n_estimators=100, learning_rate=0.1, random_state=42
        )

    def train(self, email_history: pd.DataFrame):
        """
        email_history: user_id, sent_hour, sent_weekday, opened (bool), clicked (bool)
        """
        features = self._extract_features(email_history)
        target = email_history['clicked'].astype(float)
        self.model.fit(features, target)

    def _extract_features(self, df: pd.DataFrame) -> pd.DataFrame:
        return pd.DataFrame({
            'sent_hour': df['sent_hour'],
            'sent_weekday': df['sent_weekday'],
            'is_weekend': (df['sent_weekday'] >= 5).astype(int),
            'is_morning': ((df['sent_hour'] >= 7) & (df['sent_hour'] <= 10)).astype(int),
            'is_lunch': ((df['sent_hour'] >= 12) & (df['sent_hour'] <= 14)).astype(int),
            'is_evening': ((df['sent_hour'] >= 19) & (df['sent_hour'] <= 22)).astype(int),
        })

    def predict_best_time(self, user_id: str,
                           user_open_history: list[dict]) -> dict:
        """Лучший час и день для пользователя"""
        if not user_open_history:
            return {'best_hour': 10, 'best_weekday': 1, 'confidence': 0.3}

        opens_df = pd.DataFrame(user_open_history)
        opens_df = opens_df[opens_df['opened'] == True]

        if len(opens_df) < 5:
            return {'best_hour': 10, 'best_weekday': 1, 'confidence': 0.4}

        best_score, best_hour, best_weekday = -1, 10, 1

        for weekday in range(5):
            for hour in [8, 10, 12, 14, 17, 19]:
                features = self._extract_features(pd.DataFrame([{
                    'sent_hour': hour, 'sent_weekday': weekday
                }]))
                score = self.model.predict(features)[0]
                if score > best_score:
                    best_score, best_hour, best_weekday = score, hour, weekday

        confidence = min(0.95, 0.4 + len(opens_df) * 0.02)

        return {
            'best_hour': best_hour,
            'best_weekday': best_weekday,
            'predicted_ctr': round(best_score, 3),
            'confidence': round(confidence, 2)
        }


class EmailContentPersonalizer:
    """Персонализация содержимого писем"""

    def __init__(self):
        self.llm = Anthropic()

    def generate_personalized_email(self, template: dict,
                                     recipient: dict,
                                     campaign_type: str) -> dict:
        """Генерация персонализированного email"""
        context = {
            'name': recipient.get('first_name', 'Пользователь'),
            'company': recipient.get('company', ''),
            'industry': recipient.get('industry', ''),
            'last_product_used': recipient.get('last_feature', ''),
            'days_inactive': recipient.get('days_since_last_login', 0),
            'plan': recipient.get('subscription_plan', 'free'),
        }

        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=400,
            messages=[{
                "role": "user",
                "content": f"""Write a personalized email for this recipient.

Campaign type: {campaign_type}
Template theme: {template.get('theme', '')}
Recipient context: {json.dumps(context, ensure_ascii=False)}

Requirements:
- Subject line: compelling, 40-60 chars, personalized
- Body: 3-4 short paragraphs
- CTA: single clear action
- Tone: professional but warm
- Language: Russian
- NO generic phrases like "we hope this email finds you well"

Return JSON: {{"subject": "...", "body": "...", "cta_text": "...", "cta_url_param": "..."}}"""
            }]
        )

        try:
            return json.loads(response.content[0].text)
        except Exception:
            return {
                'subject': f"{context['name']}, специальное предложение",
                'body': template.get('default_body', ''),
                'cta_text': 'Открыть',
                'cta_url_param': ''
            }

    def generate_subject_line_variants(self, base_subject: str,
                                        audience_segment: str,
                                        n_variants: int = 5) -> list[str]:
        """A/B тестирование: несколько вариантов темы письма"""
        response = self.llm.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=200,
            messages=[{
                "role": "user",
                "content": f"""Generate {n_variants} subject line variants for A/B testing.
Base: "{base_subject}"
Audience: {audience_segment}
Language: Russian
Mix strategies: curiosity, urgency, social proof, benefit, question.
Return JSON array: ["variant1", "variant2", ...]"""
            }]
        )

        try:
            return json.loads(response.content[0].text)[:n_variants]
        except Exception:
            return [base_subject]


class EmailSequenceOrchestrator:
    """Оркестрация email-последовательностей"""

    def __init__(self):
        self.send_time_optimizer = SendTimeOptimizer()
        self.content_personalizer = EmailContentPersonalizer()

    def process_trigger_event(self, event: dict,
                               recipient: dict,
                               sequences: dict) -> dict:
        """Обработка триггерного события → выбор и запуск последовательности"""
        event_type = event.get('type')

        sequence_map = {
            'signup': 'onboarding',
            'trial_start': 'trial_nurture',
            'trial_expire_soon': 'conversion_push',
            'purchase': 'post_purchase',
            'inactivity_7d': 'reactivation',
            'feature_not_used': 'feature_adoption',
        }

        sequence_name = sequence_map.get(event_type)
        if not sequence_name or sequence_name not in sequences:
            return {'action': 'skip', 'reason': f'No sequence for event: {event_type}'}

        sequence = sequences[sequence_name]
        first_email = sequence[0]

        email_content = self.content_personalizer.generate_personalized_email(
            first_email, recipient, sequence_name
        )

        send_time = self.send_time_optimizer.predict_best_time(
            recipient['id'],
            recipient.get('email_history', [])
        )

        return {
            'sequence': sequence_name,
            'email': email_content,
            'send_at': {
                'hour': send_time['best_hour'],
                'weekday': send_time['best_weekday']
            },
            'remaining_steps': len(sequence) - 1,
            'tracking_params': {
                'campaign': sequence_name,
                'user_id': recipient['id']
            }
        }

    def evaluate_sequence_performance(self,
                                       sequence_logs: pd.DataFrame) -> pd.DataFrame:
        """Метрики эффективности последовательности"""
        return sequence_logs.groupby(['sequence_name', 'step_number']).agg(
            sent=('email_id', 'count'),
            open_rate=('opened', 'mean'),
            ctr=('clicked', 'mean'),
            conversion_rate=('converted', 'mean'),
            unsubscribe_rate=('unsubscribed', 'mean')
        ).round(3)

Чому важливо визначати найкращий час відправки?

Кожен користувач має свої звички: хтось перевіряє пошту зранку, хтось — в обід. Якщо надіслати лист у неактивний час, він загубиться. Наш алгоритм на основі градієнтного бустингу знаходить персональний sweet spot. Впевненість прогнозу зростає з обсягом даних: від 30% при відсутності історії до 95% після 30+ відкриттів.

Як ми це робимо

Використовуємо зв'язку:

• Python + scikit-learn для предиктивних моделей,
• Anthropic Claude 3.5 Sonnet для генерації контенту (з можливістю RAG-розширення),
• PostgreSQL з розширенням pgvector для зберігання ембедінгів,
• RabbitMQ для асинхронної відправки листів.

Моделі навчаються на історичних даних (мінімум 5000 подій) і донавчаються щотижня. Для холодного старту застосовуємо few-shot підхід — беремо патерни схожих користувачів.

Процес роботи

Що входить у роботу

• Вихідний код всіх модулів (SendTimeOptimizer, EmailContentPersonalizer, EmailSequenceOrchestrator)
• Документація API у форматі OpenAPI
• Навчання вашої команди (2–3 години онлайн)
• Доступ до репозиторію та моніторингу (Grafana + Prometheus)
• Гарантія на коректну роботу 3 місяці
• Економія бюджету на email-маркетинг до 40% за рахунок автоматизації

Порівняння традиційної та AI-автоматизації

Типові помилки на старті

Гіперперсоналізація без достатнього обсягу даних — листи виходять дивними, знижується довіра. Ми починаємо із загальних патернів і поступово нарощуємо індивідуальність. Ще одна помилка — ігнорувати guardrails: LLM може згенерувати офтоп або некоректний контент. Наші промпти містять чіткі обмеження.

Досвід і гарантії

5 років на ринку розробки AI-рішень для email-маркетингу. Реалізували понад 20 проектів для e-commerce та SaaS. Надаємо гарантію на стабільну роботу системи протягом 3 місяців після деплою.

Замовте розробку AI-системи для вашого email-маркетингу — зв'яжіться з нами для оцінки вашого проекту. Отримайте консультацію та попередній розрахунок вартості.

Розробка рекомендаційних систем: від 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.