推荐系统

一句话概述

在推荐系统中应用强化学习代表着从「预测用户会点什么」（监督学习范式）到「优化用户的长期体验」（决策范式）的转变。核心思路是将用户与推荐系统的交互建模为MDP——系统推荐一个物品列表、用户产生反馈（点击、购买、停留时间）、系统获得奖励并更新策略。RL能够自然地处理推荐中的探索-利用权衡、长时域效果优化和多样性的主动探索。

教学与演示

从监督学习到RL的范式转变

传统推荐系统基于监督学习：收集大量 (用户特征, 物品特征, 点击标签) 数据，训练模型预测「用户会不会点击这个物品」，然后推荐预测分数最高的物品。

问题在哪？

监督学习推荐的几个根本缺陷：

undefined

RL推荐系统的思路完全不同：把推荐建模为序列决策问题。

目标是最大化用户的生命周期总价值，而非单次点击。

大白话 传统推荐像快餐店——「今天你想吃啥我就给你啥，只要给得多就赚得多」。RL推荐像营养师——「综合你的长期健康来搭配饮食，偶尔推荐你没吃过但可能喜欢的新菜品」。短期的点击率可能略低，但用户用得更久。

RL推荐系统的建模

将推荐建模为RL问题：

状态s_t：用户的历史行为序列、人口统计特征、当前会话上下文（时间、设备、位置）。通常用RNN/Transformer编码为稠密向量：

动作a_t：推荐一个物品或一组物品（slate）。动作空间通常是离散的（物品ID）且极大（百万级）。

奖励r_t：用户的即时反馈——点击、购买、收藏、转发、停留时间。

状态转移：用户的行为历史更新——新点击的物品加入历史序列，状态编码器重新编码。

大白话 把用户在APP里的每一次滑动、每一次点击、每一次购买都看作一条「轨迹」。RL的目标不是让每一步的奖励最大，而是让整条轨迹的总奖励（用户用得久、买得多、满意度高）最大。这就是「长期主义」的本质。

关键挑战与技术方案

挑战一：超大规模离散动作空间

推荐系统中动作空间是百万物品级，不能用常规的softmax输出层。

解决：基于策略梯度方法，用物品的embedding向量来参数化策略：

只对少量候选物品（如检索Top-K）计算softmax，实际中用负采样近似。

挑战二：离线评估与在线学习的平衡

无法像游戏一样「重来一次试试不同推荐」，因为用户感受是真实的。

三大评估方案：

undefined

挑战三：长时域信用分配

用户在第1次点击「好评电影」到第7天续费会员之间，因果链长达数百步。

使用序列化奖励建模：不仅用即时反馈，还用未来行为的汇总信号（如「这次会话后用户是否7天内回来了」作为额外奖励）。

大白话 推荐RL最难的地方不是算法，而是「你怎么知道推荐做得好不好」。用户的即时点赞可能是因为标题吸引人（标题党），长期取关可能是因为推荐在慢慢变差。RL的价值就在于——它能区分「一时爽」和「一直爽」。

离线强化学习推荐

在真实推荐场景中，与用户实时交互探索（在线RL）代价极高——一次糟糕的推荐可能导致用户永久流失。

离线RL（Offline RL / Batch RL）提供了一条中间道路：仅在历史日志数据上训练，不与环境实时交互。

核心思想：给定的历史数据D = {(s, a, r, s')}来自行为策略π_b（通常是当前线上模型），离线RL要学出一个比π_b更好的策略π_θ，但永远不实际与环境交互。

保守Q-Learning (CQL) 是最流行的离线RL推荐算法：

undefined

大白话 离线RL就像只看「过去的点菜记录」来设计新菜单——不能试菜（不能上线测试），只能从历史数据中推断「哪些搭配会受欢迎」。关键技巧是「保守」——对没见过的东西保持怀疑，不轻易冒险。

什么用

RL在推荐系统中的实际应用：

undefined

大白话 RL在推荐领域的应用已经从「学术玩具」变成「工业标配」。如今的短视频平台的推荐系统几乎都在不同程度地使用RL——这是少数几个RL在生产环境中产生数十亿美元商业价值的领域。

哪些坑

核心代码演示

"""
推荐系统RL - 基于策略梯度的推荐
简化演示探索-利用权衡和长时域优化
"""
import numpy as np

# ===== 推荐环境 =====
class RecommendationEnv:
    """
    简化推荐环境
    用户有「兴趣向量」，推荐「物品向量」
    两者点积越高，点击概率越大
    """
    def __init__(self, n_items=100, feature_dim=10):
        self.n_items = n_items
        self.feature_dim = feature_dim
        
        # 随机生成物品特征
        self.item_features = np.random.randn(n_items, feature_dim)
        self.item_features = self.item_features / np.linalg.norm(self.item_features, axis=1, keepdims=True)
        
        self.reset()
    
    def reset(self):
        """新用户或新会话开始"""
        # 用户兴趣向量（会随推荐内容变化）
        self.user_interest = np.random.randn(self.feature_dim)
        self.user_interest = self.user_interest / np.linalg.norm(self.user_interest)
        self.session_items = []  # 本次会话推荐的物品
        self.satisfaction = 0.5  # 用户满意度（长期指标）
        return self.user_interest.copy()
    
    def step(self, action_item_id):
        """
        推荐一个物品，观察用户反馈
        
        返回：(用户新状态, 奖励, 是否结束会话)
        """
        item_vec = self.item_features[action_item_id]
        
        # 点击概率 = sigmoid(兴趣·物品特征)
        relevance = np.dot(self.user_interest, item_vec)
        click_prob = 1.0 / (1.0 + np.exp(-5.0 * relevance))
        
        # 模拟用户行为
        clicked = np.random.random() < click_prob
        if clicked:
            # 用户点击了 → 兴趣略微向这个物品偏移（兴趣漂移）
            self.user_interest = 0.8 * self.user_interest + 0.2 * item_vec
            self.user_interest = self.user_interest / np.linalg.norm(self.user_interest)
            self.satisfaction = min(1.0, self.satisfaction + 0.05)
        
        # Session-level多样性探索
        diversity = 1.0  # 多样性奖励
        if action_item_id in self.session_items:
            diversity = 0.3  # 重复推荐 → 多样性惩罚
        
        self.session_items.append(action_item_id)
        
        # 组合奖励：即时点击 + 多样性 + 长期满意度
        immediate_reward = 1.0 if clicked else 0.0
        long_term_reward = self.satisfaction * 0.2
        reward = immediate_reward + 0.1 * diversity + long_term_reward
        
        # 会话结束条件
        done = len(self.session_items) >= 10 or (not clicked and len(self.session_items) >= 3)
        
        return self.user_interest.copy(), reward, done

# ===== 策略梯度推荐智能体 =====
class PolicyGradientRecommender:
    """
    基于REINFORCE的推荐策略
    学习用户兴趣到推荐物品的映射
    """
    def __init__(self, n_items, feature_dim, lr=0.01):
        self.n_items = n_items
        self.feature_dim = feature_dim
        self.lr = lr
        
        # 策略参数：用户兴趣 → 物品embedding的映射矩阵
        # 实际中这是物品embedding网络的一部分
        self.policy_weights = np.random.randn(feature_dim, feature_dim) * 0.1
    
    def get_item_scores(self, user_state, item_features):
        """计算所有物品的推荐得分"""
        # 用户兴趣经过策略网络变换
        policy_vector = user_state @ self.policy_weights
        # 与物品特征的点积作为得分
        scores = item_features @ policy_vector
        return scores
    
    def recommend(self, user_state, item_features, epsilon=0.1):
        """
        推荐一个物品
        ε-greedy探索 vs 贪心利用
        """
        if np.random.random() < epsilon:
            # 探索：随机推荐（发现新兴趣）
            return np.random.randint(self.n_items), True
        
        # 利用：推荐得分最高的
        scores = self.get_item_scores(user_state, item_features)
        return np.argmax(scores), False

# ===== 训练演示 =====
env = RecommendationEnv(n_items=100, feature_dim=10)
agent = PolicyGradientRecommender(100, 10)

print("=== RL推荐系统演示 ===\n")

n_episodes = 30
for ep in range(n_episodes):
    state = env.reset()
    session_rewards = []
    done = False
    
    while not done:
        # 推荐 + 探索
        item_id, is_explore = agent.recommend(state, env.item_features, epsilon=0.15)
        next_state, reward, done = env.step(item_id)
        
        session_rewards.append(reward)
        state = next_state
    
    total_reward = sum(session_rewards)
    
    if ep % 5 == 0:
        print(f"Episode {ep:2d} | 总奖励: {total_reward:.2f} | "
              f"会话长度: {len(session_rewards)} | "
              f"最终满意度: {env.satisfaction:.2f}")

print(f"\n最终用户满意度: {env.satisfaction:.2f}")
print(f"\nRL推荐的优势：")
print("1. 自动平衡探索(发现新兴趣) vs 利用(推荐已知偏好)")
print("2. 长期满意度指标引导策略向「留住用户」方向优化")
print("3. 多样性奖励防止推荐变得单调重复")

"""
离线RL推荐 - CQL (Conservative Q-Learning) 思路演示
在历史数据上训练，保守估计Q值
"""
import numpy as np

def conservative_q_learning_demo():
    """
    演示CQL的核心思想：
    Q(s,a)的更新中，对「历史数据中未出现过的动作」加上惩罚
    防止过高估计未见过物品的Q值
    """
    n_items = 10
    # 当前Q值估计
    q_values = np.random.randn(n_items) * 0.5
    
    # 历史数据中出现过的物品（行为策略推荐过的）
    observed_items = {2, 5, 7, 8}
    
    # 标准Q-learning更新
    td_target = 2.0  # 假设TD目标=2.0
    
    # === CQL保守更新 ===
    alpha = 0.5  # 保守系数
    new_q = q_values.copy()
    
    for i in range(n_items):
        if i in observed_items:
            # 常见物品：正常Q-learning更新
            new_q[i] = q_values[i] + 0.1 * (td_target - q_values[i])
        else:
            # 罕见物品：保守更新 + 惩罚
            conservative_penalty = alpha * 1.0  # 惩罚项
            # CQL: Q = Q + lr * (TD_target - Q - α)
            new_q[i] = q_values[i] + 0.1 * (td_target - q_values[i] - conservative_penalty)
    
    print("=== CQL保守Q-Learning ===\n")
    print(f"{'物品ID':<8} {'出现?':<8} {'原Q值':<10} {'新Q值':<10} {'变化':<10}")
    print("-" * 46)
    for i in range(n_items):
        obs = "是" if i in observed_items else "否"
        diff = new_q[i] - q_values[i]
        marker = " ← 被惩罚" if i not in observed_items else ""
        print(f"{i:<8} {obs:<8} {q_values[i]:>8.3f}  {new_q[i]:>8.3f}  {diff:>+8.3f}{marker}")
    
    print(f"\nCQL的效果：")
    print(f"- 物品{list(observed_items)}（历史中出现过）→ 正常更新Q值")
    print(f"- 其他物品（未见过）→ Q值被压制")
    print(f"- 策略倾向于推荐历史中出现的物品（安全+保守）")
    
    print(f"\n为什么推荐系统需要这种保守性？")
    print("直接推荐一个从未展示过的物品是高风险操作——")
    print("用户可能很喜欢（好），也可能觉得莫名其妙（用户流失）。")
    print("CQL确保系统在「有证据」时才大胆推荐。")

conservative_q_learning_demo()

概念关系图谱

重点答疑

大白话 推荐RL的核心思维转变是：「用户不是来点东西的，用户是来获得满足感的」。你的目标不是预测他会点什么（监督学习），而是给他推荐「最能让他满意、用得最久」的东西（强化学习）。

💡 核心要点：推荐系统RL最独特的三个挑战——（1）动作空间百万级，不能用标准softmax（用embedding+策略梯度）；（2）不能在线试错，一次烂推荐用户就跑了（用离线RL/CQL保守学习）；（3）奖励信号极其稀疏和延迟（用汇总的长期奖励信号如留存率）。

undefined

章节单词汇总