第20章 强化学习与Bandit推荐¶

import numpy as np

class EpsilonGreedy:
    """ε-Greedy策略"""
    def __init__(self, n_arms, epsilon=0.1):
        self.n_arms = n_arms
        self.epsilon = epsilon
        self.counts = np.zeros(n_arms)     # 每个臂被选次数
        self.values = np.zeros(n_arms)     # 每个臂的估计价值

    def select_arm(self):
        if np.random.random() < self.epsilon:
            return np.random.randint(self.n_arms)  # 探索
        else:
            return np.argmax(self.values)            # 利用

    def update(self, arm, reward):
        self.counts[arm] += 1
        n = self.counts[arm]
        # 增量更新均值: Q_new = Q_old + (r - Q_old) / n
        self.values[arm] += (reward - self.values[arm]) / n

class UCB:
    """Upper Confidence Bound策略"""
    def __init__(self, n_arms, c=2.0):
        self.n_arms = n_arms
        self.c = c
        self.counts = np.zeros(n_arms)
        self.values = np.zeros(n_arms)
        self.t = 0

    def select_arm(self):
        self.t += 1
        # 先让每个臂至少被选一次
        for arm in range(self.n_arms):
            if self.counts[arm] == 0:
                return arm

        ucb_values = self.values + self.c * np.sqrt(
            np.log(self.t) / self.counts
        )
        return np.argmax(ucb_values)

    def update(self, arm, reward):
        self.counts[arm] += 1
        n = self.counts[arm]
        self.values[arm] += (reward - self.values[arm]) / n

class ThompsonSampling:
    """Thompson Sampling (Beta-Bernoulli)"""
    def __init__(self, n_arms):
        self.n_arms = n_arms
        self.alpha = np.ones(n_arms)  # 成功次数 + 1
        self.beta = np.ones(n_arms)   # 失败次数 + 1

    def select_arm(self):
        # 从每个臂的后验Beta分布中采样
        samples = np.random.beta(self.alpha, self.beta)
        return np.argmax(samples)

    def update(self, arm, reward):
        # reward为0或1
        if reward == 1:
            self.alpha[arm] += 1
        else:
            self.beta[arm] += 1

import numpy as np
import matplotlib.pyplot as plt
# EpsilonGreedy, UCB, ThompsonSampling 类定义同上

class BanditEnv:
    """模拟K臂老虎机环境"""
    def __init__(self, true_probs):
        self.true_probs = np.array(true_probs)  # np.array创建NumPy数组
        self.n_arms = len(true_probs)
        self.best_arm = np.argmax(true_probs)
        self.best_prob = true_probs[self.best_arm]

    def pull(self, arm):
        return 1 if np.random.random() < self.true_probs[arm] else 0

# ----- 运行实验 -----
def run_experiment(env, policy, n_rounds=5000):
    """运行一次实验，返回累积遗憾"""
    regrets = []
    cumulative_regret = 0
    for t in range(n_rounds):
        arm = policy.select_arm()
        reward = env.pull(arm)
        policy.update(arm, reward)
        instant_regret = env.best_prob - env.true_probs[arm]
        cumulative_regret += instant_regret
        regrets.append(cumulative_regret)
    return regrets

# 设置环境：10个臂的Bandit
np.random.seed(42)
true_probs = [0.05, 0.08, 0.12, 0.15, 0.20, 0.25, 0.30, 0.35, 0.10, 0.45]
env = BanditEnv(true_probs)
n_rounds = 5000
n_experiments = 20  # 多次实验取平均

policies = {
    'ε-Greedy (ε=0.1)': lambda: EpsilonGreedy(env.n_arms, epsilon=0.1),
    'UCB (c=2)': lambda: UCB(env.n_arms, c=2.0),
    'Thompson Sampling': lambda: ThompsonSampling(env.n_arms),
}

plt.figure(figsize=(10, 6))
for name, policy_fn in policies.items():
    all_regrets = []
    for _ in range(n_experiments):
        policy = policy_fn()
        regrets = run_experiment(env, policy, n_rounds)
        all_regrets.append(regrets)
    mean_regrets = np.mean(all_regrets, axis=0)
    plt.plot(mean_regrets, label=name)

plt.xlabel('Round')
plt.ylabel('Cumulative Regret')
plt.title('Bandit策略对比 - 累积遗憾曲线')
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('bandit_comparison.png', dpi=150)
plt.show()
print(f"最优臂: arm {env.best_arm}, 真实概率: {env.best_prob}")

import numpy as np

class LinUCB:
    """
    Disjoint LinUCB算法
    每个臂有独立的参数向量θ_a
    """
    def __init__(self, n_arms, d, alpha=1.0):
        """
        Args:
            n_arms: 候选臂数量
            d: 上下文特征维度
            alpha: 探索系数，控制置信区间宽度
        """
        self.n_arms = n_arms
        self.d = d
        self.alpha = alpha

        # 每个臂的参数
        self.A = {}  # A_a = I_d + Σ x_s x_s^T
        self.b = {}  # b_a = Σ r_s * x_s

        for a in range(n_arms):
            self.A[a] = np.eye(d)        # d×d 单位矩阵
            self.b[a] = np.zeros(d)       # d维零向量

    def select_arm(self, context_vectors):
        """
        选择一个臂
        Args:
            context_vectors: dict or list, {arm_id: context_vector}
                             或 shape=(n_arms, d) 的数组
        Returns:
            chosen_arm: 选择的臂编号
        """
        ucb_values = np.zeros(self.n_arms)

        for a in range(self.n_arms):
            if isinstance(context_vectors, dict):  # isinstance检查类型
                x = context_vectors[a]
            else:
                x = context_vectors[a]

            A_inv = np.linalg.inv(self.A[a])  # np.linalg线性代数运算
            theta_hat = A_inv @ self.b[a]        # 参数估计

            # UCB = x^T θ_hat + α * sqrt(x^T A^{-1} x)
            pred = x @ theta_hat                   # 预测奖励
            confidence = self.alpha * np.sqrt(x @ A_inv @ x)  # 置信宽度
            ucb_values[a] = pred + confidence

        return np.argmax(ucb_values)

    def update(self, arm, context, reward):
        """
        更新臂的参数
        Args:
            arm: 选择的臂
            context: 上下文向量 shape=(d,)
            reward: 获得的奖励
        """
        self.A[arm] += np.outer(context, context)  # A_a += x x^T
        self.b[arm] += reward * context              # b_a += r * x

# ----- LinUCB 模拟实验 -----
def simulate_linucb():
    """模拟新闻推荐场景"""
    np.random.seed(42)

    n_arms = 5       # 5篇候选新闻
    d = 6            # 6维用户特征
    n_rounds = 3000
    alpha = 0.5

    # 每个臂的真实参数（未知，模拟用）
    true_theta = np.random.randn(n_arms, d) * 0.5

    agent = LinUCB(n_arms, d, alpha=alpha)

    cumulative_reward = 0
    rewards_history = []
    optimal_rewards = []

    for t in range(n_rounds):
        # 生成随机用户上下文
        user_context = np.random.randn(d)
        user_context = user_context / np.linalg.norm(user_context)  # 归一化

        # 构建每个臂的上下文（此处简化为共享用户特征）
        contexts = {a: user_context for a in range(n_arms)}

        # 计算真实期望奖励
        true_rewards = [user_context @ true_theta[a] for a in range(n_arms)]
        optimal_arm = np.argmax(true_rewards)

        # LinUCB选择
        chosen_arm = agent.select_arm(contexts)

        # 生成奖励（真实期望 + 噪声）
        reward = user_context @ true_theta[chosen_arm] + np.random.randn() * 0.1
        reward = max(0, min(1, reward))  # 裁剪到[0,1]

        # 更新
        agent.update(chosen_arm, user_context, reward)

        cumulative_reward += reward
        rewards_history.append(cumulative_reward / (t + 1))
        optimal_rewards.append(
            max(0, min(1, true_rewards[optimal_arm]))
        )

    # 绘制平均奖励曲线
    plt.figure(figsize=(10, 5))
    plt.plot(rewards_history, label='LinUCB Avg Reward', alpha=0.8)
    window = 100
    rolling_optimal = np.convolve(
        optimal_rewards, np.ones(window)/window, mode='valid'
    )
    plt.plot(range(window-1, n_rounds), rolling_optimal,
             label='Optimal (rolling avg)', alpha=0.8, linestyle='--')
    plt.xlabel('Round')
    plt.ylabel('Average Reward')
    plt.title('LinUCB 新闻推荐模拟')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig('linucb_simulation.png', dpi=150)
    plt.show()
    print(f"最终平均奖励: {rewards_history[-1]:.4f}")  # [-1]负索引取最后元素

import matplotlib.pyplot as plt
simulate_linucb()

import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from collections import deque
import random
import matplotlib.pyplot as plt

class QNetwork(nn.Module):  # 继承nn.Module定义网络层
    """Q网络：输入状态，输出每个动作的Q值"""
    def __init__(self, state_dim, n_actions, hidden_dim=128):
        super().__init__()  # super()调用父类方法
        self.net = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, n_actions)
        )

    def forward(self, x):
        return self.net(x)

class DQNRecommender:
    """基于DQN的推荐Agent"""
    def __init__(self, state_dim, n_items, hidden_dim=128,
                 lr=1e-3, gamma=0.99, epsilon_start=1.0,
                 epsilon_end=0.01, epsilon_decay=0.995,
                 buffer_size=10000, batch_size=64,
                 target_update_freq=100):
        self.state_dim = state_dim
        self.n_items = n_items
        self.gamma = gamma
        self.epsilon = epsilon_start
        self.epsilon_end = epsilon_end
        self.epsilon_decay = epsilon_decay
        self.batch_size = batch_size
        self.target_update_freq = target_update_freq
        self.step_count = 0

        # 主网络和目标网络
        self.q_net = QNetwork(state_dim, n_items, hidden_dim)
        self.target_net = QNetwork(state_dim, n_items, hidden_dim)
        self.target_net.load_state_dict(self.q_net.state_dict())
        self.target_net.eval()

        self.optimizer = optim.Adam(self.q_net.parameters(), lr=lr)
        self.loss_fn = nn.SmoothL1Loss()  # Huber Loss

        # 经验回放
        self.replay_buffer = deque(maxlen=buffer_size)

    def select_item(self, state):
        """ε-Greedy选择推荐物品"""
        if random.random() < self.epsilon:
            return random.randint(0, self.n_items - 1)

        with torch.no_grad():  # 禁用梯度计算，节省内存
            state_t = torch.FloatTensor(state).unsqueeze(0)  # unsqueeze增加一个维度
            q_values = self.q_net(state_t)
            return q_values.argmax(dim=1).item()  # 将单元素张量转为Python数值

    def store_transition(self, state, action, reward, next_state, done):
        """存入经验回放"""
        self.replay_buffer.append((state, action, reward, next_state, done))

    def train_step(self):
        """从经验回放中采样并训练"""
        if len(self.replay_buffer) < self.batch_size:
            return None

        batch = random.sample(self.replay_buffer, self.batch_size)
        states, actions, rewards, next_states, dones = zip(*batch)  # zip按位置配对

        states = torch.FloatTensor(np.array(states))
        actions = torch.LongTensor(actions).unsqueeze(1)
        rewards = torch.FloatTensor(rewards).unsqueeze(1)
        next_states = torch.FloatTensor(np.array(next_states))
        dones = torch.FloatTensor(dones).unsqueeze(1)

        # 当前Q值
        q_values = self.q_net(states).gather(1, actions)

        # 目标Q值 (Double DQN)
        with torch.no_grad():
            next_actions = self.q_net(next_states).argmax(dim=1, keepdim=True)
            next_q = self.target_net(next_states).gather(1, next_actions)
            target_q = rewards + self.gamma * next_q * (1 - dones)

        loss = self.loss_fn(q_values, target_q)

        self.optimizer.zero_grad()  # 清零梯度
        loss.backward()  # 反向传播计算梯度
        torch.nn.utils.clip_grad_norm_(self.q_net.parameters(), 1.0)
        self.optimizer.step()  # 更新参数

        # 更新目标网络
        self.step_count += 1
        if self.step_count % self.target_update_freq == 0:
            self.target_net.load_state_dict(self.q_net.state_dict())

        # 衰减ε
        self.epsilon = max(self.epsilon_end,
                          self.epsilon * self.epsilon_decay)

        return loss.item()

# ----- 简化模拟 -----
class SimpleRecEnv:
    """简化的推荐环境模拟"""
    def __init__(self, n_items=20, state_dim=16):
        self.n_items = n_items
        self.state_dim = state_dim
        # 模拟用户偏好
        self.user_pref = np.random.randn(state_dim)
        self.item_embeddings = np.random.randn(n_items, state_dim) * 0.5
        self.step = 0
        self.max_steps = 50

    def reset(self):
        self.user_pref = np.random.randn(self.state_dim)
        self.step = 0
        return self._get_state()

    def _get_state(self):
        return self.user_pref.copy()

    def act(self, item_id):
        """模拟用户对推荐物品的反应"""
        # 相似度越高，奖励越高
        sim = np.dot(self.user_pref, self.item_embeddings[item_id])  # np.dot矩阵/向量点乘
        prob_click = 1 / (1 + np.exp(-sim))  # sigmoid
        clicked = np.random.random() < prob_click
        reward = 1.0 if clicked else 0.0

        # 更新用户状态（模拟兴趣漂移）
        if clicked:
            self.user_pref = 0.9 * self.user_pref + \
                             0.1 * self.item_embeddings[item_id]

        self.step += 1
        done = self.step >= self.max_steps
        return self._get_state(), reward, done

def train_dqn_recommender():
    """训练DQN推荐Agent"""
    env = SimpleRecEnv(n_items=20, state_dim=16)
    agent = DQNRecommender(
        state_dim=16, n_items=20,
        lr=1e-3, gamma=0.95, buffer_size=5000, batch_size=32
    )

    n_episodes = 300
    episode_rewards = []

    for ep in range(n_episodes):
        state = env.reset()
        total_reward = 0

        while True:
            action = agent.select_item(state)
            next_state, reward, done = env.act(action)
            agent.store_transition(state, action, reward, next_state, done)

            loss = agent.train_step()
            state = next_state
            total_reward += reward

            if done:
                break

        episode_rewards.append(total_reward)

        if (ep + 1) % 50 == 0:
            avg = np.mean(episode_rewards[-50:])
            print(f"Episode {ep+1}, Avg Reward: {avg:.2f}, "
                  f"Epsilon: {agent.epsilon:.3f}")

    # 绘制学习曲线
    plt.figure(figsize=(10, 5))
    window = 20
    smoothed = np.convolve(episode_rewards,
                           np.ones(window)/window, mode='valid')
    plt.plot(smoothed)
    plt.xlabel('Episode')
    plt.ylabel('Total Reward (smoothed)')
    plt.title('DQN推荐Agent学习曲线')
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig('dqn_recommender.png', dpi=150)
    plt.show()

train_dqn_recommender()

import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np

class SlatePolicy(nn.Module):
    """
    推荐列表策略网络
    输入用户状态，输出每个候选物品的选择概率
    """
    def __init__(self, state_dim, n_candidates, hidden_dim=64):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(state_dim, hidden_dim),
            nn.ReLU(),
            nn.Linear(hidden_dim, hidden_dim),
            nn.ReLU(),
        )
        self.scorer = nn.Linear(hidden_dim, n_candidates)

    def forward(self, state):
        h = self.encoder(state)
        logits = self.scorer(h)
        return logits

    def select_slate(self, state, slate_size, greedy=False):
        """
        选择一个推荐列表（无放回采样）
        Returns: (slate, log_probs)
        """
        logits = self.forward(state)
        slate = []
        log_probs = []
        mask = torch.zeros_like(logits)  # 已选物品的mask

        for _ in range(slate_size):
            masked_logits = logits + mask
            probs = F.softmax(masked_logits, dim=-1)  # F.xxx PyTorch函数式API

            if greedy:
                idx = probs.argmax(dim=-1)
            else:
                dist = torch.distributions.Categorical(probs)
                idx = dist.sample()
                log_probs.append(dist.log_prob(idx))

            slate.append(idx.item())
            mask[0, idx] = float('-inf')  # 屏蔽已选物品

        return slate, log_probs

def train_slate_reinforce():
    """REINFORCE训练推荐列表策略"""
    state_dim = 16
    n_candidates = 50
    slate_size = 5
    n_episodes = 500
    lr = 1e-3
    gamma = 0.99

    policy = SlatePolicy(state_dim, n_candidates)
    optimizer = torch.optim.Adam(policy.parameters(), lr=lr)

    # 模拟物品质量（未知）
    item_quality = np.random.rand(n_candidates)

    episode_rewards = []

    for ep in range(n_episodes):
        # 随机用户状态
        state = torch.randn(1, state_dim)
        slate, log_probs = policy.select_slate(state, slate_size)

        # 模拟用户反馈：位置衰减 + 物品质量
        rewards = []
        for pos, item_id in enumerate(slate):  # enumerate同时获取索引和元素
            position_bias = 1.0 / np.log2(pos + 2)  # DCG位置权重
            click_prob = item_quality[item_id] * position_bias
            reward = 1.0 if np.random.random() < click_prob else 0.0
            rewards.append(reward)

        # 计算折扣回报
        returns = []
        G = 0
        for r in reversed(rewards):
            G = r + gamma * G
            returns.insert(0, G)
        returns = torch.FloatTensor(returns)
        returns = (returns - returns.mean()) / (returns.std() + 1e-8)

        # REINFORCE更新
        policy_loss = 0
        for lp, G in zip(log_probs, returns):
            policy_loss -= lp * G

        optimizer.zero_grad()
        policy_loss.backward()
        optimizer.step()

        episode_rewards.append(sum(rewards))

        if (ep + 1) % 100 == 0:
            avg_r = np.mean(episode_rewards[-100:])
            print(f"Episode {ep+1}, Avg Reward: {avg_r:.3f}")

    print(f"\n学到的Top物品: {slate}")
    print(f"真实最优Top-{slate_size}: "
          f"{np.argsort(item_quality)[-slate_size:][::-1].tolist()}")

train_slate_reinforce()

def ips_estimator(rewards, pi_e_probs, pi_b_probs, clip=None):
    """
    IPS离线策略评估
    Args:
        rewards: 观测到的奖励
        pi_e_probs: 评估策略在观测动作上的概率 π_e(a|x)
        pi_b_probs: 行为策略在观测动作上的概率 π_b(a|x)
        clip: 重要性权重裁剪阈值
    """
    weights = pi_e_probs / (pi_b_probs + 1e-10)
    if clip is not None:
        weights = np.clip(weights, 0, clip)
    return np.mean(weights * rewards)

def doubly_robust_estimator(rewards, pi_e_probs, pi_b_probs,
                             reward_model_preds, pi_e_expected_reward):
    """
    Doubly Robust离线策略评估
    Args:
        reward_model_preds: 奖励模型对(x,a)的预测 r_hat(x,a)
        pi_e_expected_reward: 奖励模型在评估策略下的期望 E_π_e[r_hat(x,a)]
    """
    weights = pi_e_probs / (pi_b_probs + 1e-10)
    residuals = rewards - reward_model_preds
    return np.mean(pi_e_expected_reward + weights * residuals)

# 示例
np.random.seed(42)
n = 1000
rewards = np.random.binomial(1, 0.3, n).astype(float)
pi_b_probs = np.ones(n) * 0.2       # 行为策略：均匀分布（5选1）
pi_e_probs = np.random.uniform(0.1, 0.5, n)  # 评估策略
reward_preds = np.ones(n) * 0.3     # 简单奖励模型
pi_e_expected = np.ones(n) * 0.3

v_ips = ips_estimator(rewards, pi_e_probs, pi_b_probs, clip=10)
v_dr = doubly_robust_estimator(
    rewards, pi_e_probs, pi_b_probs, reward_preds, pi_e_expected
)
print(f"IPS估计: {v_ips:.4f}")
print(f"DR估计:  {v_dr:.4f}")
print(f"真实均值: {rewards.mean():.4f}")

def shaped_reward(action_type, duration=0, **kwargs):  # *args接收任意位置参数，**kwargs接收任意关键字参数
    """
    推荐系统中的Reward Shaping示例

    将稀疏的长期奖励转化为密集的短期信号
    """
    reward_map = {
        'impression': 0.0,         # 曝光：无奖励
        'click': 1.0,              # 点击
        'like': 2.0,               # 点赞
        'comment': 3.0,            # 评论
        'share': 5.0,              # 分享（高价值行为）
        'follow': 4.0,             # 关注创作者
        'purchase': 10.0,          # 购买
        'skip': -0.5,              # 快速跳过（负反馈）
        'dislike': -2.0,           # 不喜欢
        'report': -5.0,            # 举报
    }

    base_reward = reward_map.get(action_type, 0.0)

    # 观看时长加成（视频推荐场景）
    if action_type == 'click' and duration > 0:
        # 完播给额外奖励
        completion_ratio = min(duration / kwargs.get('video_len', 60), 1.0)
        base_reward += completion_ratio * 2.0

    return base_reward