Files
pipu-music/services/recommender/pipu_rec/wave/engine.py
T
mrleo1nidandClaude Opus 5.5 ff0aafdb1e
ci / recommender / recommender (push) Failing after 4s
ci / db / db (push) Failing after 2m21s
feat(recommender): Thompson-sampling bandit over candidate sources (M6)
Nightly Beta(α, β) per user and source from wave impressions (reward:
complete or like, ×0.99 daily decay) stored in user_profiles_rec.bandit;
the wave adds 0.2·(θ − 0.5) per candidate source, sources without data
stay neutral. The user's exploration share now follows how well `fresh`
does against the other sources.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-20 22:31:24 +03:00

366 lines
17 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Сборка пачки «Моей волны» (v1: эвристика). Чистые функции над массивами — без БД и сети."""
from __future__ import annotations
from dataclasses import dataclass, field
import numpy as np
from numpy.typing import NDArray
from pipu_rec.wave.contexts import FEATURES, Rule, fit
MODEL_VER = "heuristic-v1"
# С коллаборативным сигналом (ALS) — суффикс к версии модели в показах.
CF_SUFFIX = "+cf"
Vec = NDArray[np.float32]
@dataclass
class CatalogIndex:
"""Каталог в памяти: строка i — трек ids[i]."""
ids: list[str]
artist: list[str] # основной исполнитель
emb: Vec # (n, d), стандартизирован и нормирован; нулевой — нет анализа
has_emb: NDArray[np.bool_]
features: dict[str, NDArray[np.float64]] # NaN — нет признака
popularity: NDArray[np.float64] # 0..1
cf: Vec = field(default_factory=lambda: np.zeros((0, 0), dtype=np.float32)) # (n, k) ALS, нормирован
has_cf: NDArray[np.bool_] = field(default_factory=lambda: np.zeros(0, dtype=bool))
pos: dict[str, int] = field(default_factory=dict)
def __post_init__(self) -> None:
self.pos = {t: i for i, t in enumerate(self.ids)}
if self.has_cf.shape[0] != len(self.ids):
self.cf = np.zeros((len(self.ids), 1), dtype=np.float32)
self.has_cf = np.zeros(len(self.ids), dtype=bool)
def __len__(self) -> int:
return len(self.ids)
@staticmethod
def build(
ids: list[str],
artist: list[str],
raw_emb: list[Vec | None],
features: dict[str, list[float | None]],
plays: list[int],
dim: int = 256,
raw_cf: list[Vec | None] | None = None,
) -> CatalogIndex:
n = len(ids)
emb = np.zeros((n, dim), dtype=np.float32)
has = np.zeros(n, dtype=bool)
for i, e in enumerate(raw_emb):
if e is not None:
emb[i], has[i] = e, True
if has.sum() >= 2:
# Стандартизация по библиотеке: косинус сравнивает отличия треков, а не общий «звук библиотеки».
mu = emb[has].mean(axis=0)
sd = emb[has].std(axis=0) + 1e-6
emb[has] = (emb[has] - mu) / sd
emb[has] /= np.linalg.norm(emb[has], axis=1, keepdims=True) + 1e-9
feats = {
k: np.array([np.nan if v is None else v for v in features.get(k, [None] * n)], dtype=np.float64)
for k in FEATURES
}
p = np.log1p(np.array(plays, dtype=np.float64))
pop = p / p.max() if n and p.max() > 0 else np.zeros(n)
cf_dim = next((len(v) for v in raw_cf or [] if v is not None), 0)
cf = np.zeros((n, max(cf_dim, 1)), dtype=np.float32)
has_cf = np.zeros(n, dtype=bool)
for i, v in enumerate(raw_cf or []):
if v is not None and len(v) == cf_dim:
norm = float(np.linalg.norm(v))
if norm > 0:
cf[i], has_cf[i] = v / norm, True
return CatalogIndex(
ids=ids, artist=artist, emb=emb, has_emb=has, features=feats, popularity=pop, cf=cf, has_cf=has_cf
)
@dataclass
class Profile:
"""Вкус пользователя: оценки треков и производные."""
scores: dict[str, float] # implicit_score > 0 — нравится
disliked: set[str]
disliked_artists: set[str]
recent: set[str] # сыграны за последние 3 часа
played: set[str] # слышал когда-либо
liked_recent: set[str] # лайкнутые, игравшие за 24 ч — штраф
rediscover: set[str] # любимые, не игравшие > 60 дней
exploration: float = 0.15
taste_cf: list[float] | None = None # фактор пользователя ALS
bandit: dict[str, list[float]] = field(
default_factory=dict
) # источник → [α, β] (docs/recommendations.md#6.6)
recent_likes: list[str] = field(default_factory=list) # последние лайки, новые первыми (для item_cf)
@dataclass
class Session:
"""Состояние сессии волны (Redis)."""
vec: list[float] | None = None
served: list[str] = field(default_factory=list)
recent_artists: list[str] = field(default_factory=list)
skips_in_row: int = 0
@dataclass(frozen=True)
class Pick:
id: str
source: str
score: float
# Вес бонуса бандита к оценке (θ − 0.5 ∈ [−0.5, 0.5]).
BANDIT_WEIGHT = 0.2
# Вклад реакции в сессионный вектор: s ← normalize(s + η·r·e).
REACTION_WEIGHT = {"complete": 1.0, "like": 2.0, "skip_early": -1.0, "dislike": -2.0, "skip": -0.5}
ETA = 0.3
def taste_centroids(index: CatalogIndex, profile: Profile, max_k: int = 6, seed: int = 0) -> Vec:
"""Центроиды вкуса: k-means по эмбеддингам треков с положительной оценкой (веса = оценка).
k растёт с историей: 1 центроид на ~8 любимых треков, не больше max_k."""
rows = [(index.pos[t], s) for t, s in profile.scores.items() if s > 0 and t in index.pos]
rows = [(i, s) for i, s in rows if index.has_emb[i]]
if not rows:
return np.zeros((0, index.emb.shape[1]), dtype=np.float32)
x = index.emb[[i for i, _ in rows]]
w = np.array([s for _, s in rows])
k = int(min(max_k, max(1, len(rows) // 8)))
rng = np.random.default_rng(seed)
centers = x[rng.choice(len(x), size=k, replace=False)]
for _ in range(10):
assign = np.argmax(x @ centers.T, axis=1)
for c in range(k):
m = assign == c
if m.any():
v = (x[m] * w[m, None]).sum(axis=0)
centers[c] = v / (np.linalg.norm(v) + 1e-9)
return centers.astype(np.float32)
def update_session(
index: CatalogIndex, session: Session, feedback: list[tuple[str, str]], centroids: Vec
) -> None:
"""Обновляет сессионный вектор и счётчик скипов по реакциям на выданные треки.
Три скипа подряд — «смена курса»: вектор переводится к наименее похожему центроиду вкуса."""
d = index.emb.shape[1]
vec = np.array(session.vec, dtype=np.float32) if session.vec else np.zeros(d, dtype=np.float32)
for track_id, outcome in feedback:
i = index.pos.get(track_id)
r = REACTION_WEIGHT.get(outcome)
if r is None:
continue
session.skips_in_row = session.skips_in_row + 1 if outcome in ("skip", "skip_early", "dislike") else 0
if i is None or not index.has_emb[i]:
continue
vec = vec + ETA * r * index.emb[i]
norm = float(np.linalg.norm(vec))
vec = vec / norm if norm > 0 else vec
if session.skips_in_row >= 3 and len(centroids):
sims = centroids @ vec if np.any(vec) else np.zeros(len(centroids))
vec = centroids[int(np.argmin(sims))].copy()
session.skips_in_row = 0
session.vec = vec.tolist() if np.any(vec) else None
def cf_scores(index: CatalogIndex, profile: Profile) -> NDArray[np.float64] | None:
"""Косинус треков к фактору пользователя ALS; None — нет модели или пользователь в неё не попал."""
if profile.taste_cf is None or not index.has_cf.any():
return None
u = np.array(profile.taste_cf, dtype=np.float32)
if u.shape[0] != index.cf.shape[1]:
return None
u = u / (float(np.linalg.norm(u)) or 1.0)
return np.where(index.has_cf, index.cf @ u, 0.0).astype(np.float64)
def model_version(index: CatalogIndex, profile: Profile) -> str:
return MODEL_VER + (CF_SUFFIX if cf_scores(index, profile) is not None else "")
def _top(scores: NDArray[np.float64], mask: NDArray[np.bool_], k: int) -> NDArray[np.intp]:
idx = np.nonzero(mask)[0]
if len(idx) <= k:
return idx
part = idx[np.argpartition(-scores[idx], k)[:k]]
return part
def build_batch(
index: CatalogIndex,
profile: Profile,
session: Session,
rules: list[Rule],
n: int = 5,
exclude: set[str] | None = None,
seed_ids: list[str] | None = None,
rng: np.random.Generator | None = None,
) -> list[Pick]:
"""Кандидаты → фильтры → оценка → соответствие контексту → MMR с правилами разнообразия."""
rng = rng or np.random.default_rng()
size = len(index)
if size == 0:
return []
centroids = taste_centroids(index, profile)
svec = np.array(session.vec, dtype=np.float32) if session.vec else None
# 2. Фильтры.
excluded = profile.disliked | profile.recent | set(session.served) | (exclude or set())
allowed = np.array(
[
t not in excluded and a not in profile.disliked_artists
for t, a in zip(index.ids, index.artist, strict=True)
]
)
if not allowed.any():
allowed = np.array([t not in profile.disliked for t in index.ids])
if not allowed.any():
return []
# Сходства со вкусом и сессией.
sim_taste = (index.emb @ centroids.T).max(axis=1) if len(centroids) else np.zeros(size)
sim_session = index.emb @ svec if svec is not None else np.zeros(size)
sim_taste = np.where(index.has_emb, sim_taste, 0.0)
sim_session = np.where(index.has_emb, sim_session, 0.0)
cf_score = cf_scores(index, profile)
# Привязанность к исполнителю — сглаженное среднее оценок его треков.
totals: dict[str, list[float]] = {}
for t, s in profile.scores.items():
i = index.pos.get(t)
if i is not None:
totals.setdefault(index.artist[i], []).append(s)
affinity = {a: sum(v) / (len(v) + 2) for a, v in totals.items()}
artist_aff = np.array([np.tanh(affinity.get(a, 0.0) / 3) for a in index.artist])
# 1. Источники кандидатов.
sources: dict[int, str] = {}
def add(idxs: NDArray[np.intp] | list[int], source: str) -> None:
for i in idxs:
sources.setdefault(int(i), source)
if seed_ids:
seeds = [index.pos[s] for s in seed_ids if s in index.pos and index.has_emb[index.pos[s]]]
if seeds:
q = index.emb[seeds].mean(axis=0)
add(_top((index.emb @ q).astype(np.float64), allowed & index.has_emb, 100), "seed")
query = np.maximum(sim_session, sim_taste) if svec is not None or len(centroids) else None
if query is not None:
# «Новое» помечаем раньше audio_sim: неслышанный трек рядом со вкусом считается исследованием.
never = np.array([t not in profile.played for t in index.ids])
add(_top(sim_taste + 0.3 * sim_session, allowed & index.has_emb & never, 50), "fresh")
add(_top(query, allowed & index.has_emb, 150), "audio_sim")
# Коллективный опыт (ALS): похожие по вкусу других слушателей и «с этим слушают» к последним лайкам.
if cf_score is not None:
add(_top(cf_score, allowed & index.has_cf, 100), "cf")
likes = [
index.pos[t] for t in profile.recent_likes[:10] if t in index.pos and index.has_cf[index.pos[t]]
]
if likes:
item_sim = (index.cf @ index.cf[likes].T).max(axis=1).astype(np.float64)
add(_top(np.where(index.has_cf, item_sim, -1.0), allowed & index.has_cf, 60), "item_cf")
liked_artists = {a for a, v in affinity.items() if v > 0.3}
if liked_artists:
by_artist = [int(i) for i in np.nonzero(allowed)[0] if index.artist[i] in liked_artists]
add(rng.permutation(by_artist)[:60].tolist() if by_artist else [], "artist")
redis_ = [index.pos[t] for t in profile.rediscover if t in index.pos and allowed[index.pos[t]]]
add(rng.permutation(redis_)[:30].tolist() if redis_ else [], "rediscover")
add(_top(index.popularity + rng.random(size) * 0.3, allowed, 30), "popular")
# Холодный старт и исследование: случайные треки из разных частей библиотеки.
add(rng.permutation(np.nonzero(allowed)[0])[:40].tolist(), "explore")
cand = np.array(sorted(sources))
# 3. Оценка v1.
score = (
0.35 * sim_session[cand]
+ 0.25 * sim_taste[cand]
+ 0.15 * artist_aff[cand]
+ 0.05 * index.popularity[cand]
+ (0.10 * cf_score[cand] if cf_score is not None else 0.0)
+ 0.02 * rng.random(len(cand)) # разбивает ничьи, особенно при холодном старте
)
liked_recent = np.array([index.ids[i] in profile.liked_recent for i in cand])
score = score - 0.2 * liked_recent
# Бандит: семплирование Томпсона по источникам — удачные для пользователя источники получают бонус.
# Источник без наблюдений — нейтральные 0.5 (без шума, иначе холодный старт ранжировался бы случайно).
theta: dict[str, float] = {}
for src in set(sources.values()):
ab = profile.bandit.get(src)
theta[src] = float(rng.beta(ab[0], ab[1])) if ab else 0.5
score = score + BANDIT_WEIGHT * (np.array([theta[sources[int(i)]] for i in cand]) - 0.5)
# 4. Соответствие контексту; если отсеялось слишком много — порог ослабляется.
if rules:
f = fit({k: v[cand] for k, v in index.features.items()}, rules)
# Нужен запас для разнообразия, но не больше 50 и не больше четверти пула.
enough = min(50, max(3 * n, len(cand) // 4))
for threshold in (0.35, 0.15, 0.0):
keep = f >= threshold
if keep.sum() >= enough or threshold == 0.0:
break
cand, score, f = cand[keep], score[keep], f[keep]
score = score + 0.5 * f # контекст важнее вкуса, когда выбран явно
if len(cand) == 0:
return []
# 5. MMR + правила: 1 трек исполнителя на пачку, не больше 2 на последние 10; доля «нового».
picks: list[Pick] = []
chosen: list[int] = []
recent_artists = session.recent_artists[-10:]
want_fresh = max(0, round(n * profile.exploration))
order = list(range(len(cand)))
lam = 0.75
while len(picks) < n and order:
best_j, best_v = -1, -np.inf
for j in order:
i = int(cand[j])
a = index.artist[i]
if any(index.artist[c] == a for c in chosen) or recent_artists.count(a) >= 2:
continue
redundancy = max((float(index.emb[i] @ index.emb[c]) for c in chosen), default=0.0)
v = lam * float(score[j]) - (1 - lam) * redundancy
if v > best_v:
best_j, best_v = j, v
if best_j < 0:
# Правила разнообразия не дают набрать пачку — ослабляем их, но не повторяем треки.
if recent_artists:
recent_artists = []
continue
if chosen:
chosen_artists = {index.artist[c] for c in chosen}
rest = [j for j in order if index.artist[int(cand[j])] in chosen_artists]
if not rest:
break
best_j = max(rest, key=lambda j: score[j])
else:
break
i = int(cand[best_j])
chosen.append(i)
order.remove(best_j)
picks.append(Pick(id=index.ids[i], source=sources[i], score=round(float(score[best_j]), 4)))
# Исследование: «новых» (неслышанных) меньше доли exploration — заменяем ими слабейшие выборы.
fresh_left = sorted(
(j for j in order if sources[int(cand[j])] == "fresh"), key=lambda j: score[j], reverse=True
)
while sum(p.source == "fresh" for p in picks) < want_fresh and fresh_left:
j = fresh_left.pop(0)
i = int(cand[j])
others = [k for k, p in enumerate(picks) if p.source != "fresh"]
if not others:
break
weakest = min(others, key=lambda k: picks[k].score)
if any(index.artist[index.pos[p.id]] == index.artist[i] for k, p in enumerate(picks) if k != weakest):
continue
picks[weakest] = Pick(id=index.ids[i], source="fresh", score=round(float(score[j]), 4))
return picks