Разработка AI-системы автоматических триггерных коммуникаций (email/SMS/push)

Реализация AI-системы триггерных коммуникаций (email/SMS/push)

Триггерные коммуникации — автоматические сообщения, отправляемые в ответ на поведение пользователя. Брошенная корзина, просмотр без покупки, неактивность 7 дней — каждый триггер требует правильного момента, канала и персонализированного текста. ML оптимизирует все три параметра одновременно.

Архитектура триггерной системы

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':
            # Брошенная корзина: отправить через 1-3 часа
            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:
        """AI-генерация персонализированного контента"""
        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

        # Низкий open rate = пользователь игнорирует
        if messages_last_7d >= 3 and opens_last_7d == 0:
            return True

        return False

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

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

Типичные метрики триггерных коммуникаций: abandoned cart email через 1-3 часа — open rate 40-55%, CTR 10-20%, conversion 5-12%. Push с персонализацией через LLM — open rate 8-15% против 3-5% у generic. SMS (срочные триггеры) — open rate 95%+ в течение 5 минут. Ключевое правило: не более 3-4 сообщений в неделю на пользователя.