Розробка 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

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