13 · 冷启动与探索利用

     新物品没有交互数据
              ↓
        模型给它低分
              ↓
         得不到曝光
              ↓
     永远没有交互数据  ←──┘  💀 无限循环

  新视频
    ├─ 标题文本  → 文本 Embedding (BERT / 多语言模型)
    ├─ 封面图    → 图像 Embedding (CLIP / ViT)
    ├─ 视频内容  → 视频 Embedding
    ├─ 音频      → 音频 Embedding
    └─ 元数据    → 类目、标签、作者、时长

          ↓ 全部拼起来
    内容向量  →  在向量空间里找相似的老物品
          ↓
    继承它们的用户群 / 统计特征作为先验

def allocate_slots(candidates, n_slots=10, explore_ratio=0.1):
    """每 10 个位置留 1 个给冷启动物品"""
    n_explore = max(1, int(n_slots * explore_ratio))
    cold = [c for c in candidates if c.impressions < 1000]
    warm = [c for c in candidates if c.impressions >= 1000]

    result = warm[:n_slots - n_explore]
    result += sorted(cold, key=lambda c: c.content_score)[:n_explore]
    return result

   新物品发布
        ↓
   【池 1】给 500 次曝光  →  CTR 达标？──否──→ 淘汰
        ↓ 是
   【池 2】给 5000 次曝光 →  CTR 达标？──否──→ 淘汰
        ↓ 是
   【池 3】给 5万次曝光   →  完播率达标？──否─→ 淘汰
        ↓ 是
   进入正常推荐池（爆款诞生）

   你面前有 5 台老虎机，不知道哪台中奖率高。
   有 1000 次机会，怎么赢最多钱？

   策略 A：全部投同一台（纯利用 Exploit）
      → 万一选错了，白白亏 1000 次

   策略 B：每台平均投 200 次（纯探索 Explore）
      → 知道了哪台好，但已经没机会了

   策略 C：先试探，逐渐把钱压到表现好的那台 ✅
      → 这就是 E&E 要解决的问题

import random

def epsilon_greedy(candidates, model_scores, epsilon=0.1):
    if random.random() < epsilon:
        return random.choice(candidates)        # 10% 随机探索
    return candidates[int(np.argmax(model_scores))]   # 90% 选最优

def ucb_score(mean_reward, n_pulls, total_pulls, c=2.0):
    if n_pulls == 0:
        return float('inf')                     # 没试过的必须先试
    return mean_reward + c * np.sqrt(2 * np.log(total_pulls) / n_pulls)

  物品A：曝光 10000 次，点击 500 次
         → 分布很窄，几乎确定 CTR ≈ 5%
              ╱▔▔╲
         ────╱────╲──────

  物品B：曝光 10 次，点击 1 次
         → 分布很宽，CTR 可能 1% 也可能 30%
            ╱▔▔▔▔▔▔▔▔╲
         ──╱──────────╲──

  每次从各自分布里抽一个数 → B 有机会抽到高值 → 得到曝光机会
  B 曝光多了，分布自然收窄 → 探索自动减少 ✅

import numpy as np

class ThompsonSampling:
    """Beta-Bernoulli Thompson Sampling —— 30 行的工业级 E&E"""
    def __init__(self, n_items, prior_a=1.0, prior_b=1.0):
        # Beta 分布的两个参数：alpha ≈ 点击数, beta ≈ 未点击数
        self.a = np.full(n_items, prior_a)
        self.b = np.full(n_items, prior_b)

    def select(self, candidates, k=10):
        """从每个候选的 Beta 分布采样，取采样值最高的 k 个"""
        samples = np.random.beta(self.a[candidates], self.b[candidates])
        top = np.argsort(-samples)[:k]
        return [candidates[i] for i in top]

    def update(self, item, clicked):
        if clicked: self.a[item] += 1
        else:       self.b[item] += 1

    def set_prior_from_content(self, item, predicted_ctr, strength=20):
        """⭐ 用内容模型的预估当先验，冷启动更快"""
        self.a[item] = predicted_ctr * strength
        self.b[item] = (1 - predicted_ctr) * strength