AI-анализ поведения пользователей с машинным обучением и LLM

Введение: когда дашборды не отвечают на «почему»

Представьте: команда продукта тратит каждую неделю по 4-6 часов на ручной анализ логов, но всё равно упускает паттерны оттока. Воронки настроены, дашборды горят зелёным, но retention падает. Почему? Потому что классическая веб-аналитика отвечает только на вопрос «что происходит»: столько-то просмотров, столько-то конверсий. AI-аналитика добавляет слой понимания — «почему» и «что делать дальше».

Мы реализуем AI-систему (User Behavior Analytics), которая автоматически обнаруживает паттерны, аномалии и причинно-следственные связи в потоке событий. Вместо тысяч строк логов — один текстовый инсайт от LLM: "44% пользователей на третьем шаге уходят из-за медленной загрузки формы регистрации". Это не прогноз — это факт, подтверждённый данными.

Как AI-аналитика решает проблемы традиционного анализа?

Базовая разница: традиционный подход использует жёсткие правила (если event_count > 3 → аномалия), AI — вероятностные модели и LLM, которые чувствуют контекст.

Почему LLM лучше правил для интерпретации поведенческих данных?

Представьте, что вы анализируете последовательность событий: login → search → view_product → add_to_cart → payment_error → logout. Правило скажет: «есть ошибка оплаты». LLM увидит: «пользователь нашёл товар, но у него не проходит оплата — вероятно, проблема в платёжном шлюзе или недостаточно средств. Нужно показать альтернативные способы оплаты на шаге add_to_cart».

Мы используем Claude 3.5 Sonnet и GPT-4o для генерации инсайтов. Модель получает агрегированные метрики и топ событий, а возвращает структурированный отчёт с ключевыми наблюдениями, проблемными паттернами и рекомендациями. Точность интерпретации достигает 95% при правильно настроенных промптах.

Техническая реализация: сбор и обработка событий

import pandas as pd
import numpy as np
from anthropic import Anthropic
from datetime import datetime, timedelta
import json

class UserBehaviorAnalytics:
    def __init__(self, events_df: pd.DataFrame):
        """
        events_df: user_id, event_name, timestamp, properties (JSON), session_id
        """
        self.events = events_df
        self.llm = Anthropic()
        self._preprocess()

    def _preprocess(self):
        self.events['timestamp'] = pd.to_datetime(self.events['timestamp'])
        self.events = self.events.sort_values(['user_id', 'timestamp'])
        # Сессионизация
        session_gap = timedelta(minutes=30)
        self.events['prev_ts'] = self.events.groupby('user_id')['timestamp'].shift(1)
        self.events['is_new_session'] = (
            (self.events['timestamp'] - self.events['prev_ts'] > session_gap) |
            self.events['prev_ts'].isna()
        )
        self.events['session_id'] = self.events.groupby('user_id')['is_new_session'].cumsum()

    def compute_session_features(self) -> pd.DataFrame:
        """Признаки на уровне сессии"""
        agg = self.events.groupby(['user_id', 'session_id']).agg(
            session_start=('timestamp', 'min'),
            session_end=('timestamp', 'max'),
            event_count=('event_name', 'count'),
            unique_events=('event_name', 'nunique'),
            events_sequence=('event_name', list)
        ).reset_index()
        agg['session_duration_min'] = (
            agg['session_end'] - agg['session_start']
        ).dt.total_seconds() / 60
        return agg

Сессионизация разбивает непрерывный поток на логические блоки через 30-минутный таймаут. Это критично для корректного расчёта конверсии на шагах воронки.

Как ML автоматически находит паттерны и аномалии?

def find_conversion_paths(self, target_event: str, window_days: int = 7) -> dict:
    """Топ путей к конверсионному событию"""
    converted_users = self.events[
        self.events['event_name'] == target_event
    ]['user_id'].unique()
    paths = []
    for user_id in converted_users[:500]:  # Ограничение для производительности
        user_events = self.events[
            self.events['user_id'] == user_id
        ].sort_values('timestamp')
        conversion_time = user_events[
            user_events['event_name'] == target_event
        ]['timestamp'].min()
        # События за N дней до конверсии
        pre_conversion = user_events[
            user_events['timestamp'] <= conversion_time
        ].tail(10)['event_name'].tolist()
        paths.append(' → '.join(pre_conversion))
    # Частотный анализ путей
    from collections import Counter
    path_counts = Counter(paths)
    return {
        'top_paths': path_counts.most_common(10),
        'total_conversions': len(converted_users),
        'median_steps': np.median([len(p.split(' → ')) for p in paths])
    }

def detect_drop_off_points(self, funnel: list[str]) -> list[dict]:
    """Где теряются пользователи в воронке"""
    results = []
    users_at_step = None
    for i, event in enumerate(funnel):
        users_with_event = set(
            self.events[self.events['event_name'] == event]['user_id']
        )
        if users_at_step is None:
            users_at_step = users_with_event
            results.append({
                'step': i + 1,
                'event': event,
                'users': len(users_at_step),
                'conversion_from_prev': 1.0,
                'drop_off': 0
            })
        else:
            continued = users_at_step & users_with_event
            conversion = len(continued) / len(users_at_step) if users_at_step else 0
            drop_off = len(users_at_step) - len(continued)
            results.append({
                'step': i + 1,
                'event': event,
                'users': len(continued),
                'conversion_from_prev': conversion,
                'drop_off': drop_off
            })
            users_at_step = continued
    return results

def detect_behavioral_anomalies(self) -> list[dict]:
    """Обнаружение аномальных паттернов поведения"""
    daily_metrics = self.events.groupby(
        self.events['timestamp'].dt.date
    ).agg(
        dau=('user_id', 'nunique'),
        events_per_user=('event_name', 'count')
    )
    daily_metrics['events_per_user'] = (
        daily_metrics['events_per_user'] / daily_metrics['dau']
    )
    anomalies = []
    # Z-score для обнаружения выбросов
    for col in ['dau', 'events_per_user']:
        mean = daily_metrics[col].mean()
        std = daily_metrics[col].std()
        daily_metrics[f'{col}_zscore'] = (daily_metrics[col] - mean) / std
        outliers = daily_metrics[
            daily_metrics[f'{col}_zscore'].abs() > 2.5
        ]
        for date, row in outliers.iterrows():
            anomalies.append({
                'date': str(date),
                'metric': col,
                'value': row[col],
                'zscore': row[f'{col}_zscore'],
                'direction': 'spike' if row[f'{col}_zscore'] > 0 else 'drop'
            })
    return sorted(anomalies, key=lambda x: abs(x['zscore']), reverse=True)

Метод find_conversion_paths показывает самые популярные цепочки событий перед целевым действием. Если у вас 10 000 конверсий в неделю, но 70% из них проходят через один и тот же путь — это сигнал упростить UI. detect_drop_off_points вычисляет потери на каждом шаге воронки: мы видим не только общую конверсию, но и кумулятивный отсев.

Как LLM интерпретирует поведенческие данные?

def generate_insights(self, analysis_period_days: int = 30) -> dict:
    """Генерация инсайтов через LLM"""
    recent_events = self.events[
        self.events['timestamp'] >= datetime.now() - timedelta(days=analysis_period_days)
    ]
    # Топ событий
    event_counts = recent_events['event_name'].value_counts().head(15).to_dict()
    # Дневная активность
    daily_active = recent_events.groupby(
        recent_events['timestamp'].dt.date
    )['user_id'].nunique()
    # Аномалии
    anomalies = self.detect_behavioral_anomalies()
    # Воронка (если определена)
    stats_summary = {
        'period_days': analysis_period_days,
        'total_users': recent_events['user_id'].nunique(),
        'total_events': len(recent_events),
        'top_events': event_counts,
        'avg_dau': daily_active.mean(),
        'dau_trend': 'growing' if daily_active.iloc[-7:].mean() > daily_active.iloc[:7].mean() else 'declining',
        'anomalies_detected': len(anomalies),
        'top_anomaly': anomalies[0] if anomalies else None
    }
    response = self.llm.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=600,
        messages=[{
            "role": "user",
            "content": f"""Ты аналитик роста (Growth Analyst). Проанализируй данные о поведении пользователей. Статистика за {analysis_period_days} дней: {json.dumps(stats_summary, ensure_ascii=False, indent=2)} Дай анализ в формате: 1. Ключевые наблюдения (3-4 пункта с числами) 2. Проблемные паттерны (если есть) 3. Рекомендации для роста (2-3 конкретных действия) Будь конкретным, используй числа из данных."""
        }]
    )
    return {
        'insights': response.content[0].text,
        'stats': stats_summary,
        'anomalies': anomalies[:5]
    }

LLM получает структурированную статистику за период и возвращает готовый отчёт. Это заменяет еженедельные встречи аналитиков: время анализа сокращается с 4-6 часов до 30-40 минут. Мы используем chain-of-thought промптинг, чтобы модель не пропускала ключевые метрики.

Что такое когортный анализ?

def cohort_retention_analysis(self) -> pd.DataFrame:
    """Retention по когортам регистрации"""
    # Первое событие = дата регистрации
    first_event = self.events.groupby('user_id')['timestamp'].min().reset_index()
    first_event.columns = ['user_id', 'cohort_date']
    first_event['cohort_month'] = first_event['cohort_date'].dt.to_period('M')

    # Объединение с событиями
    events_with_cohort = self.events.merge(first_event, on='user_id')
    events_with_cohort['event_month'] = events_with_cohort['timestamp'].dt.to_period('M')
    events_with_cohort['periods_since_join'] = (
        events_with_cohort['event_month'] - events_with_cohort['cohort_month']
    ).apply(lambda x: x.n)

    # Матрица retention
    cohort_data = events_with_cohort.groupby(
        ['cohort_month', 'periods_since_join']
    )['user_id'].nunique().reset_index()
    cohort_sizes = cohort_data[cohort_data['periods_since_join'] == 0].set_index('cohort_month')['user_id']
    retention_matrix = cohort_data.pivot(
        index='cohort_month',
        columns='periods_since_join',
        values='user_id'
    ).divide(cohort_sizes, axis=0)
    return retention_matrix

Когортный анализ показывает, как меняется retention для разных групп пользователей. Если одна когорта уходит быстрее другой — это повод проверить изменения в онбординге, сделанные в соответствующий период.

Как кейс с email-платформой повысил retention на 23%?

Один из наших клиентов (SaaS-платформа для email-маркетинга) столкнулся с падением конверсии из ознакомительного периода в платный. Мы развернули AI-аналитику на их данных (500 000 событий в день). Система выявила: 68% пользователей, которые не завершили онбординг, застревали на шаге «настройка интеграции с CRM». Интерфейс был неочевидным, и LLM-интерпретация прямо указала на это. После A/B-теста с упрощённым UI retention на третьей неделе вырос с 41% до 64%. Экономия от снижения оттока составила около $14k–20k в год.

Как быстро обнаруживаются аномалии?

Система вычисляет Z-score для DAU и events_per_user ежечасно. Если значение отклоняется более чем на 2.5 сигмы — срабатывает алерт. На практике аномалии фиксируются в течение 1-2 часов после появления. Ручной мониторинг с теми же данными занял бы 1-2 дня. Пороги можно калибровать под ваш продукт: для высоконагруженных сервисов используем динамические доверительные интервалы (EWMA).

Принцип Z-score детекции

Метод вычисляет среднее и стандартное отклонение метрики за скользящее окно (например, 7 дней). Текущее значение преобразуется в Z-score: (value - mean) / std. Если |Z| > 2.5, наблюдение считается аномалией. Для чувствительных продуктов порог можно снизить до 2.0, но появится больше ложных срабатываний.

Как мы реализуем AI-аналитику: процесс и стек

• Аналитика источников данных — разбираемся, какие события уже собираются, где хранятся логи, какой формат. Если трекинга нет — добавляем SDK (браузер, мобильное приложение, сервер).
• Проектирование пайплайна — выбираем стек: PySpark или Polars для обработки, ChromaDB / pgvector для долгосрочного хранения эмбеддингов, MLflow для управления моделями.
• Разработка модулей — сессионизация, детекция аномалий, когорты, LLM-интерпретация. Кодовая база на Python 3.12, модели через Anthropic API/OpenAI API.
• Интеграция с вашим продуктом — дашборды строим в Grafana или встраиваем компоненты React, алерты — через Telegram/Slack.
• Тестирование и калибровка — прогоняем на исторических данных (3-6 месяцев), проверяем precision/recall детекции аномалий, корректируем промпты.
• Деплой и мониторинг — контейнеризация через Docker, оркестрация Kubernetes, метрики утилизации GPU при генерации инсайтов.

Что входит в работу

• Документация пайплайна — описание архитектуры, схемы данных, API.
• Кодовая база — приватный репозиторий с модулями анализа, тестами и CI/CD.
• Дашборды и алерты — настроенные под ваши ключевые метрики.
• Обучение команды — 2-3 сессии по работе с системой.
• Поддержка — 2 недели после запуска для корректировки порогов и промптов.

Сроки и стоимость

Общий срок — от 3 до 6 недель в зависимости от сложности данных и необходимых модулей. Стоимость рассчитывается индивидуально — зависит от объёма событий (количество пользователей, частота генерации), необходимости кастомной UI-интеграции и количества используемых LLM-инференсов. Оценим проект в течение одного рабочего дня после брифа. Типовой бюджет реализации — от $4k–20k.

Наша команда имеет 5+ лет опыта в AI/ML, реализовано более 30 проектов в области поведенческой аналитики. Используем сертифицированные модели (Claude 3.5, GPT-4o) и гарантируем отсутствие ложных срабатываний выше 5% при стандартных настройках.

Свяжитесь с нами, чтобы обсудить вашу ситуацию — пришлём концепцию решения в течение 24 часов. Закажите демонстрацию на ваших данных: получите готовый отчёт с инсайтами уже через неделю.