Разработка рекомендательной системы для музыкального сервиса

Реализация рекомендательной системы для музыкального сервиса

Музыкальные рекомендации — один из наиболее технически зрелых доменов: Spotify Discover Weekly, Apple Music Radio, Last.fm — у каждого своя архитектура. Специфика: сессионный контекст ("настроение"), дискретные события (skip после 10 секунд = дизлайк), аудио-признаки трека как дополнение к поведенческим данным.

Audio-Based Feature Extraction

import librosa
import numpy as np

class AudioFeatureExtractor:
    """Извлечение аудио-признаков через librosa"""

    def extract(self, audio_path: str) -> dict:
        """30-секундный превью → вектор признаков"""
        y, sr = librosa.load(audio_path, duration=30, sr=22050)

        features = {}

        # Темп и ритм
        tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
        features['tempo'] = float(tempo)
        features['tempo_std'] = float(librosa.beat.beat_track(y=y, sr=sr, trim=False)[0])

        # Энергия и громкость
        rms = librosa.feature.rms(y=y)[0]
        features['energy_mean'] = float(rms.mean())
        features['energy_std'] = float(rms.std())

        # Тональность
        chroma = librosa.feature.chroma_stft(y=y, sr=sr)
        features['chroma_mean'] = chroma.mean(axis=1).tolist()

        # MFCC (тембр)
        mfcc = librosa.feature.mfcc(y=y, sr=sr, n_mfcc=13)
        features['mfcc_mean'] = mfcc.mean(axis=1).tolist()
        features['mfcc_std'] = mfcc.std(axis=1).tolist()

        # Spectral features
        spectral_centroid = librosa.feature.spectral_centroid(y=y, sr=sr)[0]
        features['spectral_centroid'] = float(spectral_centroid.mean())

        rolloff = librosa.feature.spectral_rolloff(y=y, sr=sr)[0]
        features['spectral_rolloff'] = float(rolloff.mean())

        # Валентность/ароузал (proxy)
        features['danceability'] = float(min(tempo / 180, 1.0) * features['energy_mean'])

        return features

    def to_vector(self, features: dict) -> np.ndarray:
        """Конвертация в numpy вектор для индексирования"""
        vector = (
            [features['tempo'] / 200, features['energy_mean']] +
            features['mfcc_mean'] +
            features['chroma_mean'] +
            [features['spectral_centroid'] / 5000,
             features['spectral_rolloff'] / 10000]
        )
        return np.array(vector, dtype=np.float32)

Сессионные рекомендации с учётом настроения

from collections import deque

class SessionAwareMusicRecommender:
    """Рекомендации с учётом текущей сессии"""

    def __init__(self, track_index, audio_features: dict):
        self.track_index = track_index
        self.audio_features = audio_features
        self.session_history = {}

    def update_session(self, user_id: str, track_id: str,
                        played_seconds: int, total_seconds: int):
        """Обновление контекста сессии"""
        if user_id not in self.session_history:
            self.session_history[user_id] = deque(maxlen=10)

        completion = played_seconds / max(total_seconds, 1)
        signal = 1.0 if completion > 0.8 else (0.5 if completion > 0.4 else -0.5)

        self.session_history[user_id].append({
            'track_id': track_id,
            'signal': signal,
            'completion': completion
        })

    def get_session_context_vector(self, user_id: str) -> np.ndarray:
        """Средний аудио-вектор последних положительных треков"""
        history = self.session_history.get(user_id, [])
        positive_tracks = [
            h['track_id'] for h in history
            if h['signal'] > 0 and h['track_id'] in self.audio_features
        ]

        if not positive_tracks:
            return None

        vectors = [self.audio_features[t] for t in positive_tracks[-5:]]
        return np.mean(vectors, axis=0)

    def recommend_next(self, user_id: str,
                        long_term_profile: np.ndarray,
                        n: int = 5,
                        session_weight: float = 0.6) -> list[tuple]:
        """Следующий трек: смесь долгосрочных предпочтений и текущей сессии"""
        session_context = self.get_session_context_vector(user_id)

        if session_context is not None:
            # Смешиваем профиль с сессионным контекстом
            query_vector = (
                session_weight * session_context +
                (1 - session_weight) * long_term_profile
            )
        else:
            query_vector = long_term_profile

        # Нормализация
        norm = np.linalg.norm(query_vector)
        query_vector = query_vector / (norm + 1e-10)

        # Исключаем только что прослушанные в сессии
        recent_tracks = {h['track_id'] for h in self.session_history.get(user_id, [])}
        candidates = self.track_index.search(query_vector, k=50)

        results = [
            (tid, score) for tid, score in candidates
            if tid not in recent_tracks
        ][:n]

        return results

Skip сигналы как неявные оценки

def process_skip_signals(plays_df: pd.DataFrame) -> pd.DataFrame:
    """Преобразование пропусков в взвешенные сигналы"""
    plays_df['completion_rate'] = plays_df['played_seconds'] / plays_df['duration_seconds'].clip(1)

    # Веса на основе completion rate
    plays_df['implicit_rating'] = np.where(
        plays_df['completion_rate'] >= 0.80, 1.0,    # Дослушал — нравится
        np.where(
            plays_df['completion_rate'] >= 0.50, 0.5,  # Частично — нейтрально
            np.where(
                plays_df['completion_rate'] <= 0.10, -1.0,  # Сразу пропустил — не нравится
                0.0  # Неопределённо
            )
        )
    )

    # Повторные прослушивания = сильный позитив
    repeat_plays = plays_df.groupby(['user_id', 'track_id']).size().reset_index(name='play_count')
    plays_df = plays_df.merge(repeat_plays, on=['user_id', 'track_id'])
    plays_df['implicit_rating'] += np.log1p(plays_df['play_count'] - 1) * 0.3

    return plays_df[plays_df['implicit_rating'] != 0]

Spotify Discover Weekly достигает 30% лайков (кнопка сердечко) — это отличный показатель для рекомендаций нового контента. Ключевые метрики: long-term retention (пользователи через 6 месяцев), skip rate первых 30 секунд, diversity score (разнообразие жанров), serendipity (обнаружение контента вне обычного вкуса).