AI Automated Trigger Communications (Email/SMS/Push)

Implementation of AI-Powered Trigger Communications (Email/SMS/Push)

Trigger communications — automatic messages sent in response to user behavior. Abandoned cart, view without purchase, 7-day inactivity — each trigger requires proper timing, channel, and personalized text. ML optimizes all three parameters simultaneously.

Trigger System Architecture

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:
        """Full trigger processing: channel + timing + content"""
        # Channel selection
        channel = self._select_channel(user_profile, event.event_type)

        # Optimal send time
        send_delay_hours = self._optimal_send_time(user_profile, event.event_type)

        # Content generation
        content = self._generate_content(event, user_profile, channel)

        # Communication fatigue check
        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:
        """Channel selection based on preferences and effectiveness"""
        # User preferences
        preferred = user.get('preferred_channel')
        if preferred:
            return Channel[preferred.upper()]

        # Effectiveness by event type
        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:
        """Optimal send time in hours"""
        # User activity pattern
        active_hours = user.get('active_hours', list(range(9, 22)))

        if event_type == 'abandoned_cart':
            # Abandoned cart: send in 1-3 hours
            return 1.5
        elif event_type == 'price_drop':
            # Price drop: immediately
            return 0.1
        elif event_type == 'inactivity':
            # Reactivation: next active period
            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-powered personalized content generation"""
        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:
        """Check for communication overload"""
        messages_last_7d = user.get('messages_received_7d', 0)
        opens_last_7d = user.get('messages_opened_7d', 0)

        # Too many messages
        if messages_last_7d >= 5:
            return True

        # Low open rate = user ignoring messages
        if messages_last_7d >= 3 and opens_last_7d == 0:
            return True

        return False

A/B Testing Trigger Messages

class TriggerABTest:
    """Testing variants of trigger messages"""

    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:
        """Deterministic variant assignment"""
        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

Typical trigger communication metrics: abandoned cart email in 1-3 hours — open rate 40-55%, CTR 10-20%, conversion 5-12%. Push with LLM personalization — open rate 8-15% versus 3-5% for generic. SMS (urgent triggers) — open rate 95%+ within 5 minutes. Key rule: no more than 3-4 messages per week per user.