Реализация AI-персонализации рекомендаций в email-кампаниях
Персонализация email-кампаний переходит от "подставить имя" к индивидуальному подбору товаров, времени отправки и темы письма для каждого получателя. ML-персонализация повышает Open Rate на 20-40% и CTR на 50-100% по сравнению с массовыми рассылками.
Персонализированная подборка товаров в письме
import pandas as pd
import numpy as np
from anthropic import Anthropic
import json
class EmailPersonalizationEngine:
def __init__(self, recommender, user_database):
self.recommender = recommender # Рекомендательная система
self.user_db = user_database
self.llm = Anthropic()
def generate_personalized_email(self, user_id: str,
campaign_type: str,
product_pool: list[dict]) -> dict:
"""Полностью персонализированное письмо"""
user = self.user_db.get_user(user_id)
if not user:
return self._generate_fallback_email(campaign_type, product_pool)
# Персонализированные товары
user_recs = self.recommender.recommend(user_id, n=6)
rec_ids = {r[0] for r in user_recs}
# Из пула кампании выбираем те, что релевантны пользователю
personalized_products = [
p for p in product_pool if p['id'] in rec_ids
][:3]
# Если мало персонализированных — добавляем популярные
if len(personalized_products) < 3:
popular = sorted(product_pool, key=lambda x: x.get('popularity', 0), reverse=True)
personalized_products += [p for p in popular
if p not in personalized_products][:3 - len(personalized_products)]
# Персонализированная тема письма
subject = self._generate_subject(user, personalized_products, campaign_type)
# Персонализированный текст
body_intro = self._generate_intro(user, personalized_products, campaign_type)
# Оптимальное время отправки
send_time = self._optimal_send_time(user)
return {
'user_id': user_id,
'subject': subject,
'body_intro': body_intro,
'products': personalized_products,
'cta_text': self._get_cta_text(campaign_type, user),
'send_at': send_time
}
def _generate_subject(self, user: dict, products: list,
campaign_type: str) -> str:
"""AI генерирует персонализированную тему"""
product_names = [p.get('name', '') for p in products[:2]]
top_category = products[0].get('category', '') if products else ''
response = self.llm.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=80,
messages=[{
"role": "user",
"content": f"""Write an email subject line (max 50 chars).
Customer: {user.get('first_name', 'Customer')}
Campaign: {campaign_type}
Featured: {', '.join(product_names)}
Top interest: {top_category}
Rules: personalize with name or interest, create curiosity, no spam words (FREE, !!!).
Return only the subject line."""
}]
)
return response.content[0].text.strip()
def _generate_intro(self, user: dict, products: list,
campaign_type: str) -> str:
"""Персонализированный вступительный абзац"""
user_context = f"""
First name: {user.get('first_name', 'Customer')}
Last purchase: {user.get('last_purchase_category', 'N/A')}
Member since: {user.get('member_months', 0)} months
Preferred brand: {user.get('top_brand', 'N/A')}
"""
response = self.llm.messages.create(
model="claude-3-5-sonnet-20241022",
max_tokens=100,
messages=[{
"role": "user",
"content": f"""Write a 2-sentence personalized email intro.
User: {user_context}
Campaign: {campaign_type}
Products to highlight: {[p.get('name') for p in products[:2]]}
Be warm and personal. Reference past purchase or interest if relevant."""
}]
)
return response.content[0].text
def _optimal_send_time(self, user: dict) -> str:
"""Оптимальное время отправки по паттернам пользователя"""
# Исторические данные о времени открытий
open_hours = user.get('email_open_hours', [])
if open_hours:
# Находим час с максимальным числом открытий
hour_counts = np.bincount(open_hours, minlength=24)
best_hour = int(np.argmax(hour_counts))
else:
# Default: 10:00 утра как лучшее время для большинства
best_hour = 10
# День недели (если данные есть)
open_days = user.get('email_open_days', [])
best_day = int(np.bincount(open_days, minlength=7).argmax()) if open_days else 2 # Tuesday default
from datetime import datetime, timedelta
today = datetime.now()
days_until = (best_day - today.weekday()) % 7
send_date = today + timedelta(days=days_until)
return send_date.strftime(f"%Y-%m-%d {best_hour:02d}:00:00")
def _get_cta_text(self, campaign_type: str, user: dict) -> str:
cta_map = {
'new_arrivals': 'Shop New Arrivals',
'sale': f"Get Your {user.get('loyalty_tier', '').title()} Discount",
'abandoned_cart': 'Complete Your Purchase',
'reactivation': 'Discover What\'s New',
'birthday': 'Claim Your Birthday Gift'
}
return cta_map.get(campaign_type, 'Shop Now')
def _generate_fallback_email(self, campaign_type: str,
product_pool: list[dict]) -> dict:
"""Не-персонализированное письмо как fallback"""
top_products = sorted(
product_pool, key=lambda x: x.get('popularity', 0), reverse=True
)[:3]
return {
'subject': 'Special picks just for you',
'products': top_products,
'cta_text': 'Shop Now'
}
Batch-генерация для рассылки
def generate_campaign_batch(self, user_ids: list,
campaign_type: str,
product_pool: list[dict]) -> pd.DataFrame:
"""Генерация писем для всей базы"""
results = []
for user_id in user_ids:
try:
email = self.generate_personalized_email(
user_id, campaign_type, product_pool
)
results.append({'user_id': user_id, 'status': 'generated', **email})
except Exception as e:
results.append({'user_id': user_id, 'status': 'failed', 'error': str(e)})
return pd.DataFrame(results)
Типичные результаты email-персонализации: Open Rate 22-35% (против 15-20% generic), CTR 3-8% (против 1-2%), Revenue per Email $0.15-0.40 (против $0.03-0.08). Важные технические ограничения: LLM-генерация 10K субжектов занимает 15-30 минут, поэтому для больших баз нужно предгенерировать шаблоны.