From ff0aafdb1e25bdfd72adf107d8abd8bbf82cacf1 Mon Sep 17 00:00:00 2001 From: mrleo1nid <43583385+mrleo1nid@users.noreply.github.com> Date: Tue, 20 Oct 2026 22:31:24 +0300 Subject: [PATCH] feat(recommender): Thompson-sampling bandit over candidate sources (M6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- docs/recommendations.md | 5 ++ services/recommender/pipu_rec/train/bandit.py | 68 +++++++++++++++++++ .../recommender/pipu_rec/train/pipeline.py | 6 +- services/recommender/pipu_rec/wave/engine.py | 13 ++++ services/recommender/pipu_rec/wave/store.py | 4 +- services/recommender/tests/test_train.py | 36 ++++++++++ 6 files changed, 130 insertions(+), 2 deletions(-) create mode 100644 services/recommender/pipu_rec/train/bandit.py diff --git a/docs/recommendations.md b/docs/recommendations.md index 3441345..534fc99 100644 --- a/docs/recommendations.md +++ b/docs/recommendations.md @@ -191,6 +191,11 @@ score = 0.35·cos(e, session_vec) + 0.25·max_c cos(e, centroid_c) + 0.15·artis что одному пользователю лучше заходит `fresh`, а другому — `rediscover`. Параметры затухают (×0.99 в день), чтобы подстраиваться под смену вкуса. +Реализация (`pipu_rec/train/bandit.py`): параметры пересчитываются ночью по `wave_impressions` за 90 дней +(награда — `complete` / `like`) в `user_profiles_rec.bandit`; при сборке пачки источник получает бонус +`0.2·(θ − 0.5)`, где θ ~ Beta(α, β) (источник без наблюдений — нейтральные 0.5). `exploration` следует за +успехом `fresh` относительно остальных источников (0.05–0.4). + ### 6.7 Реакция в реальном времени - Клиент держит в очереди 2 трека вперёд и запрашивает новую пачку, когда осталось ≤ 2. diff --git a/services/recommender/pipu_rec/train/bandit.py b/services/recommender/pipu_rec/train/bandit.py new file mode 100644 index 0000000..0d70455 --- /dev/null +++ b/services/recommender/pipu_rec/train/bandit.py @@ -0,0 +1,68 @@ +"""Бандит по источникам кандидатов (docs/recommendations.md#6.6): Beta(α, β) на пару (пользователь, источник). + +Награда — трек из источника доигран или лайкнут; скип и дизлайк — неудача. Наблюдения затухают ×0.99 в день, +чтобы бандит подстраивался под смену вкуса. Доля «нового» (exploration) следует за успехом источника `fresh`. +""" + +from __future__ import annotations + +import json +from collections import defaultdict +from typing import Any + +import psycopg + +DAILY_DECAY = 0.99 +REWARD = {"complete": 1.0, "like": 1.0, "skip": 0.0, "skip_early": 0.0, "dislike": 0.0} +EXPLORATION_MIN, EXPLORATION_MAX = 0.05, 0.4 + +Params = dict[str, list[float]] # источник → [α, β] + + +def compute(rows: list[tuple[str, str, str, float]], now: float) -> dict[str, Params]: + """rows: (user, source, outcome, served_at) → user → source → [α, β] с приором Beta(1, 1).""" + out: dict[str, Params] = defaultdict(dict) + for user, source, outcome, ts in rows: + r = REWARD.get(outcome) + if r is None: + continue + w = DAILY_DECAY ** max((now - ts) / 86400.0, 0.0) + ab = out[user].setdefault(source, [1.0, 1.0]) + ab[0] += w * r + ab[1] += w * (1.0 - r) + return {u: {s: [round(a, 4), round(b, 4)] for s, (a, b) in p.items()} for u, p in out.items()} + + +def exploration(params: Params, default: float = 0.15) -> float: + """Доля «нового»: успешнее `fresh` относительно остальных источников — больше исследования.""" + fresh = params.get("fresh") + others = [a / (a + b) for s, (a, b) in params.items() if s != "fresh"] + if fresh is None or not others: + return default + rate = fresh[0] / (fresh[0] + fresh[1]) + base = sum(others) / len(others) + value = default * (rate / base if base > 0 else 1.0) + return round(min(max(value, EXPLORATION_MIN), EXPLORATION_MAX), 3) + + +def update_bandits(conn: psycopg.Connection[Any], now: float) -> int: + """Пересчитывает параметры по показам волны за 90 дней. Возвращает число пользователей.""" + rows = conn.execute( + """ + SELECT ws.user_id::text, wi.candidate_source, wi.outcome, extract(epoch FROM wi.served_at)::float8 + FROM wave_impressions wi JOIN wave_sessions ws ON ws.id = wi.wave_session_id + WHERE wi.outcome IS NOT NULL AND wi.served_at > now() - interval '90 days' + """ + ).fetchall() + params = compute([(u, s, o, float(ts)) for u, s, o, ts in rows], now) + with conn.transaction(), conn.cursor() as cur: + cur.executemany( + """ + INSERT INTO user_profiles_rec (user_id, bandit, exploration, updated_at) + VALUES (%s, %s, %s, now()) + ON CONFLICT (user_id) DO UPDATE SET bandit = EXCLUDED.bandit, exploration = EXCLUDED.exploration, + updated_at = now() + """, + [(u, json.dumps(p), exploration(p)) for u, p in params.items()], + ) + return len(params) diff --git a/services/recommender/pipu_rec/train/pipeline.py b/services/recommender/pipu_rec/train/pipeline.py index 6b7699e..38e76b0 100644 --- a/services/recommender/pipu_rec/train/pipeline.py +++ b/services/recommender/pipu_rec/train/pipeline.py @@ -15,6 +15,7 @@ import numpy as np import psycopg from pipu_rec.train.als import ALSModel, normalize, train_als +from pipu_rec.train.bandit import update_bandits from pipu_rec.train.metrics import evaluate from pipu_rec.train.signals import DISLIKE, LIKE, Interaction, event_weight, implicit_scores @@ -157,9 +158,11 @@ def train_all(dsn: str, now: float | None = None) -> dict[str, Any]: if not conn.execute("SELECT pg_try_advisory_lock(%s)", (ADVISORY_LOCK,)).fetchone()[0]: # type: ignore[index] return {"skipped": "another training is running"} try: + # Бандит считается всегда — ему хватает и пары сессий волны, в отличие от ALS. + bandit_users = update_bandits(conn, now) interactions = load_interactions(conn) if len(interactions) < MIN_INTERACTIONS: - return {"skipped": f"only {len(interactions)} interactions"} + return {"skipped": f"only {len(interactions)} interactions", "bandit_users": bandit_users} started = time.monotonic() offline = offline_metrics(interactions, now) model = train_als(implicit_scores(interactions, now)) @@ -171,6 +174,7 @@ def train_all(dsn: str, now: float | None = None) -> dict[str, Any]: "users": len(model.users), "items": len(model.items), "interactions": len(interactions), + "bandit_users": bandit_users, "train_s": round(time.monotonic() - started, 1), } if not should_activate(offline, active_metrics(conn, "als")): diff --git a/services/recommender/pipu_rec/wave/engine.py b/services/recommender/pipu_rec/wave/engine.py index d1c9130..5ab2193 100644 --- a/services/recommender/pipu_rec/wave/engine.py +++ b/services/recommender/pipu_rec/wave/engine.py @@ -93,6 +93,9 @@ class Profile: 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) @@ -113,6 +116,9 @@ class Pick: 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 @@ -284,6 +290,13 @@ def build_batch( ) 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: diff --git a/services/recommender/pipu_rec/wave/store.py b/services/recommender/pipu_rec/wave/store.py index a699266..4030171 100644 --- a/services/recommender/pipu_rec/wave/store.py +++ b/services/recommender/pipu_rec/wave/store.py @@ -156,10 +156,11 @@ def load_profile(conn: psycopg.Connection[Any], user_id: str) -> Profile: ) } row = conn.execute( - "SELECT exploration, taste_cf::text FROM user_profiles_rec WHERE user_id = %s", (user_id,) + "SELECT exploration, taste_cf::text, bandit FROM user_profiles_rec WHERE user_id = %s", (user_id,) ).fetchone() exploration = float(row[0]) if row else 0.15 taste_cf = parse_vector(row[1]) if row else None + bandit = {str(k): [float(x) for x in v] for k, v in (row[2] or {}).items()} if row else {} if len(played) < 30: exploration = max(exploration, 0.4) # холодный старт: больше исследования на первых треках day = 86400 @@ -173,6 +174,7 @@ def load_profile(conn: psycopg.Connection[Any], user_id: str) -> Profile: rediscover={t for t, s in scores.items() if s > 1.5 and now - last.get(t, now) > 60 * day}, exploration=exploration, taste_cf=taste_cf.tolist() if taste_cf is not None else None, + bandit=bandit, recent_likes=recent_likes, ) diff --git a/services/recommender/tests/test_train.py b/services/recommender/tests/test_train.py index a04c129..819a21d 100644 --- a/services/recommender/tests/test_train.py +++ b/services/recommender/tests/test_train.py @@ -116,3 +116,39 @@ def test_cf_vectors_are_padded_to_column_size() -> None: v = np.ones((3, 5), dtype=np.float32) assert pad(v).shape == (3, CF_DIM) assert pad(np.ones((2, 80), dtype=np.float32)).shape == (2, CF_DIM) + + +def test_bandit_params_and_exploration() -> None: + from pipu_rec.train.bandit import compute, exploration + + now = 100 * DAY + rows = [("u", "fresh", "complete", now)] * 8 + [("u", "fresh", "skip", now)] * 2 + rows += [("u", "audio_sim", "complete", now)] * 2 + [("u", "audio_sim", "skip_early", now)] * 8 + rows += [("u", "cf", "complete", now - 69 * DAY)] # затухание: ~0.5 + p = compute(rows, now)["u"] + assert p["fresh"] == [9.0, 3.0] + assert p["audio_sim"] == [3.0, 9.0] + assert 1.4 < p["cf"][0] < 1.6 + assert exploration(p) > 0.15 # «новое» заходит лучше остального — исследования больше + assert exploration({"fresh": [1.0, 9.0], "audio_sim": [9.0, 1.0]}) == 0.05 + + +def test_bandit_favours_successful_source() -> None: + n = 60 + ids = [f"t{i}" for i in range(n)] + index = CatalogIndex.build( + ids=ids, artist=[f"a{i}" for i in range(n)], raw_emb=[None] * n, features={}, plays=[0] * n + ) + good = Profile( + scores={}, + disliked=set(), + disliked_artists=set(), + recent=set(), + played=set(), + liked_recent=set(), + rediscover=set(), + exploration=0.0, + bandit={"popular": [50.0, 1.0], "explore": [1.0, 50.0]}, + ) + picks = build_batch(index, good, Session(), [], n=5, rng=np.random.default_rng(1)) + assert sum(p.source == "popular" for p in picks) >= 4