用Python实战蒙特卡洛算法:从Blackjack游戏到强化学习入门

二十一点,这个在赌场里风靡了数百年的纸牌游戏,可能比你想象中更适合作为你踏入强化学习世界的第一个沙盒。它规则清晰、状态空间有限,却又蕴含着概率与决策的微妙博弈。对于许多想动手实践强化学习的Python开发者来说,理论公式往往令人望而生畏,而一个具体的、可编程的游戏环境,则能瞬间将抽象概念转化为指尖的代码和可视化的策略图谱。今天,我们不谈复杂的数学推导,而是直接打开编辑器,用Gym库搭建一个Blackjack擂台,让蒙特卡洛方法在这里“玩”出最优策略,带你直观感受智能体是如何从零开始学会“赌术”的。

蒙特卡洛方法的核心思想异常朴素:通过大量随机采样来估计未知量。在强化学习中,这意味着我们不需要预先知道环境的全部规则(即状态转移概率和奖励函数),智能体只需一遍又一遍地玩游戏,记录下每一次的“经历”(状态、动作、奖励),然后从这些历史经验中计算平均回报,从而评估策略的好坏。这种“从实践中学习”的方式,使得蒙特卡洛方法成为解决模型未知问题的利器。本文面向的,正是那些希望绕过艰深理论、通过代码和结果来理解强化学习核心思想的实践派开发者。我们将从搭建环境开始,一步步实现蒙特卡洛预测与控制算法,并对比不同策略下的学习效果,最终让你获得一套可复用于其他场景的代码框架和实战思维。

1. 环境搭建与问题建模

在开始编写算法之前,我们需要一个标准化的“游戏场”。OpenAI的Gym库为我们提供了这样一个开箱即用的环境。它封装了游戏规则、状态转移和奖励机制,让我们可以专注于智能体策略本身。

1.1 安装与初始化Gym环境

首先,确保你的Python环境中已安装必要的库。除了gym,我们还需要numpy进行数值计算,以及matplotlibseaborn用于结果可视化。

pip install gym numpy matplotlib seaborn

安装完成后,我们可以轻松地创建一个Blackjack环境。Gym中的Blackjack-v1环境已经完美模拟了游戏规则:玩家目标是使手牌点数之和在不爆牌(超过21点)的前提下尽可能大,Ace可计为1或11,J、Q、K计为10点。

import gym
import numpy as np

# 创建二十一点环境
env = gym.make('Blackjack-v1')
# 重置环境,获得初始状态
initial_state = env.reset()
print(f"初始状态: {initial_state}")

环境返回的状态是一个三元组 (player_sum, dealer_card, usable_ace)

  • player_sum: 玩家当前手牌点数之和(12-21)。
  • dealer_card: 庄家明牌点数(1-10,1代表Ace)。
  • usable_ace: 布尔值,表示玩家手中是否有可计为11点的Ace。

动作空间是离散的:0代表“停牌”(Stand),1代表“要牌”(Hit)。每执行一步动作,环境会返回下一个状态、即时奖励、是否结束以及一些调试信息。通过与环境交互,我们可以采集到用于学习的“经验”数据。

1.2 理解状态空间与动作价值

为了后续算法的实现,我们必须清晰定义我们要估计的对象。在蒙特卡洛方法中,我们通常估计的是动作价值函数 Q(s, a),它代表在状态s下采取动作a,并在此后遵循某个策略所能获得的期望回报

对于Blackjack,其状态空间虽然有限,但直接思考所有可能性仍有些复杂。我们可以通过一个简单的表格来梳理:

状态维度 取值范围 说明
玩家点数 (player_sum) 12 至 21 点数低于12时,玩家总会选择要牌,因此不是决策点。
庄家明牌 (dealer_card) 1 (Ace) 至 10 庄家暗牌点数未知,但明牌是决策的重要依据。
是否有可用Ace (usable_ace) True 或 False 这显著影响后续爆牌概率和策略选择。

因此,总的状态数为 10(玩家点数) * 10(庄家牌) * 2(有无Ace) = 200个。对于每个状态,我们需要评估两个动作(停牌或要牌)的价值。我们的目标,就是通过让智能体反复游戏,填满这个200x2的Q值表格。

注意:Gym环境中的庄家遵循固定策略:点数总和小于17时必须拿牌,达到或超过17时则停牌。这是我们环境模型中已知的唯一规则,智能体需要学习的是如何针对庄家的这个固定策略来优化自己的决策。

2. 蒙特卡洛预测:评估一个给定策略

假设我们初始给智能体一个非常保守的策略:“只要我的点数达到18或以上,我就停牌;否则,我要牌。”这个策略好还是坏?蒙特卡洛预测算法就是用来回答这个问题的。它通过让智能体用这个策略玩很多很多局游戏,来估算每个状态的价值。

2.1 首次访问与每次访问MC预测

蒙特卡洛预测有两种主要变体:首次访问型(First-Visit)和每次访问型(Every-Visit)。它们的核心区别在于如何对待同一幕(一局游戏)中重复出现的同一个状态。

  • 首次访问型MC预测:只在一幕中第一次访问到某个状态s时,用该幕后续的累计回报G来更新s的价值估计。这种方法得到的估计是无偏的,且各次更新相互独立。
  • 每次访问型MC预测:在一幕中每次访问到状态s时,都用从该次访问开始的后续累计回报来更新s的价值估计。在样本有限时,它能提供更多的更新数据,但估计值之间存在相关性。

对于像Blackjack这样的 episodic 任务,两种方法在大样本下都会收敛到真实价值。下面我们以实现首次访问型MC预测为例,展示核心代码逻辑。

from collections import defaultdict

def mc_prediction_first_visit(policy, env, num_episodes, gamma=1.0):
    """
    首次访问蒙特卡洛预测算法。
    评估给定策略在给定环境下的状态价值函数V(s)。

    参数:
        policy: 一个函数,输入状态,输出动作。
        env: Gym环境。
        num_episodes: 要运行的幕数。
        gamma: 折扣因子,Blackjack中通常为1(无折扣)。

    返回:
        V: 字典,状态 -> 估计价值。
    """
    # 初始化:回报总和 与 访问次数
    returns_sum = defaultdict(float)
    returns_count = defaultdict(float)
    # 最终的价值函数 V(s)
    V = defaultdict(float)

    for i_episode in range(1, num_episodes + 1):
        # 生成一幕序列
        episode = []
        state = env.reset()
        done = False

        while not done:
            action = policy(state)
            next_state, reward, done, _ = env.step(action)
            # 记录 (状态, 动作, 奖励)
            episode.append((state, action, reward))
            state = next_state

        # 计算回报并反向更新
        G = 0.0
        # 使用集合记录本幕中已访问过的状态(首次访问)
        visited_states_in_episode = set()

        for t in range(len(episode)-1, -1, -1): # 从后向前遍历
            state_t, action_t, reward_t = episode[t]
            G = gamma * G + reward_t  # 累计回报

            # 如果是首次访问该状态
            if state_t not in visited_states_in_episode:
                visited_states_in_episode.add(state_t)
                returns_sum[state_t] += G
                returns_count[state_t] += 1
                V[state_t] = returns_sum[state_t] / returns_count[state_t]

        # 可选:每10万幕打印一次进度
        if i_episode % 100000 == 0:
            print(f"已完成 {i_episode} 幕...")

    return V

定义一个简单的保守策略进行评估:

def simple_policy(state):
    """点数>=18则停牌,否则要牌。"""
    player_score, dealer_card, usable_ace = state
    return 0 if player_score >= 18 else 1  # 0: Stand, 1: Hit

运行100万局游戏进行预测:

# 运行预测
value_estimates = mc_prediction_first_visit(simple_policy, env, num_episodes=1000000)

2.2 可视化分析策略价值

得到价值估计后,最直观的方式是将其可视化。由于状态是三维的(玩家点数、庄家牌、有无Ace),我们可以分别绘制“有可用Ace”和“无可用Ace”两种情况下的价值曲面。

import matplotlib.pyplot as plt
from mpl_toolkits.mplot3d import Axes3D

def plot_value_surface(V, usable_ace=True):
    """绘制状态价值函数的3D曲面图。"""
    player_range = range(12, 22)
    dealer_range = range(1, 11)
    X, Y = np.meshgrid(player_range, dealer_range)
    Z = np.zeros_like(X, dtype=float)

    for i, player in enumerate(player_range):
        for j, dealer in enumerate(dealer_range):
            state = (player, dealer, usable_ace)
            Z[j, i] = V.get(state, 0.0)  # 未访问过的状态价值设为0

    fig = plt.figure(figsize=(12, 8))
    ax = fig.add_subplot(111, projection='3d')
    surf = ax.plot_surface(X, Y, Z, cmap='viridis', edgecolor='none')
    ax.set_xlabel('Player Sum')
    ax.set_ylabel('Dealer Showing')
    ax.set_zlabel('State Value')
    title = 'State Value Function (Usable Ace)' if usable_ace else 'State Value Function (No Usable Ace)'
    ax.set_title(title)
    fig.colorbar(surf, shrink=0.5, aspect=5)
    plt.show()

# 绘制结果
plot_value_surface(value_estimates, usable_ace=True)
plot_value_surface(value_estimates, usable_ace=False)

通过观察3D曲面,你可以清晰地看到:当玩家点数较高(如19-21)且庄家明牌较小时,状态价值最高(接近+0.5),因为此时玩家胜率很高。而当玩家点数刚好是12-16这种“尴尬”点数,且庄家明牌是7-Ace这种强牌时,状态价值往往为负,因为玩家很容易爆牌或输给庄家。这个可视化结果直观地验证了我们保守策略的优劣区域。

3. 蒙特卡洛控制:寻找最优策略

预测是评估一个既定策略的好坏,而控制的目标是找到最优策略。这需要通过“策略评估”和“策略改进”的不断迭代来完成,即广义策略迭代(GPI)。蒙特卡洛控制方法的核心挑战在于“探索-利用”困境:如果一直遵循当前认为最好的策略(贪心),可能永远无法发现某些状态下更好的动作。

3.1 ε-贪心策略与同轨策略控制

最直接的解决方案是采用ε-贪心策略。在绝大多数时候(概率为1-ε),智能体选择当前估计价值最高的动作(利用);但有ε的小概率,它会随机选择一个动作(探索)。在控制算法中,我们同时用这个ε-贪心策略作为行为策略(生成样本)和目标策略(被评估和改进),这就是同轨策略(On-policy) 控制。

以下是同轨策略蒙特卡洛控制(每次访问型)的核心实现。我们这次直接估计动作价值函数Q(s, a)。

def mc_control_on_policy(env, num_episodes, epsilon=0.1, gamma=1.0):
    """
    同轨策略(ε-贪心)蒙特卡洛控制算法。
    返回估计的最优动作价值函数Q和最优策略π。

    参数:
        env: 环境。
        num_episodes: 总幕数。
        epsilon: 探索概率。
        gamma: 折扣因子。

    返回:
        Q: 字典,(状态, 动作) -> 估计价值。
        policy: 字典,状态 -> 最优动作(贪心)。
    """
    # 初始化
    # 为每个(状态, 动作)对维护回报总和与访问次数
    returns_sum = defaultdict(float)
    returns_count = defaultdict(float)
    Q = defaultdict(float)  # 动作价值函数
    # 策略初始化为任意值,例如总是要牌
    policy = defaultdict(int)  # 默认动作是0(停牌)

    for i_episode in range(1, num_episodes + 1):
        # 使用当前ε-贪心策略生成一幕
        episode = []
        state = env.reset()
        done = False

        while not done:
            # ε-贪心动作选择
            if np.random.random() > epsilon:
                # 贪心选择:选择Q值最大的动作
                action = 0 if Q[(state, 0)] >= Q[(state, 1)] else 1
            else:
                # 随机探索
                action = env.action_space.sample()

            next_state, reward, done, _ = env.step(action)
            episode.append((state, action, reward))
            state = next_state

        # 反向遍历,更新Q值
        G = 0.0
        for t in range(len(episode)-1, -1, -1):
            state_t, action_t, reward_t = episode[t]
            G = gamma * G + reward_t
            # 每次访问型更新
            returns_sum[(state_t, action_t)] += G
            returns_count[(state_t, action_t)] += 1
            Q[(state_t, action_t)] = returns_sum[(state_t, action_t)] / returns_count[(state_t, action_t)]

            # 策略改进:更新当前状态的策略为贪心策略
            # 比较该状态下两个动作的Q值
            if Q[(state_t, 0)] >= Q[(state_t, 1)]:
                policy[state_t] = 0
            else:
                policy[state_t] = 1

        # 可选:动态衰减epsilon,后期减少探索
        # epsilon = max(0.01, epsilon * 0.9999)

    return Q, policy

运行这个控制算法,比如500万局游戏后,我们不仅能得到最优策略,还能看到Q函数是如何逐渐收敛的。

3.2 结果解读与策略可视化

得到最优策略policy后,我们可以用一个热力图来直观展示这个策略矩阵。横轴是庄家明牌,纵轴是玩家点数,颜色代表动作(例如,黄色代表“要牌”,蓝色代表“停牌”)。

import seaborn as sns

def plot_optimal_policy(policy, usable_ace=True):
    """绘制最优策略热力图。"""
    policy_matrix = np.zeros((10, 10))  # 玩家12-21,庄家1-10
    for i, player in enumerate(range(12, 22)):
        for j, dealer in enumerate(range(1, 11)):
            state = (player, dealer, usable_ace)
            policy_matrix[i, j] = policy.get(state, 0)  # 默认为停牌

    plt.figure(figsize=(10, 8))
    ax = sns.heatmap(policy_matrix, cmap='YlOrRd', cbar_kws={'label': 'Action (0=Stand, 1=Hit)'},
                     xticklabels=range(1, 11), yticklabels=range(12, 22))
    ax.invert_yaxis()  # 让玩家点数从下往上增加
    ax.set_xlabel('Dealer Showing Card')
    ax.set_ylabel('Player Sum')
    title = f'Optimal Policy (Usable Ace: {usable_ace})'
    ax.set_title(title)
    plt.show()

# 假设我们已经运行算法得到了optimal_policy
# plot_optimal_policy(optimal_policy, usable_ace=True)
# plot_optimal_policy(optimal_policy, usable_ace=False)

你会发现,学到的策略非常符合人类高手的直觉:

  • 有可用Ace时:策略非常激进。因为Ace的灵活性大大降低了爆牌风险,所以即使玩家点数达到18、19,面对庄家强牌(7、8、9、10、A)时,仍然会选择要牌,以期拿到更好的点数。
  • 无可用Ace时:策略则保守得多。当玩家点数为12-16时,如果庄家明牌是7-Ace(强牌),玩家会选择要牌,因为停牌几乎必输;如果庄家明牌是2-6(弱牌),玩家则会选择停牌,让庄家去冒爆牌的风险。这就是著名的“基本策略”的核心部分。

4. 高级话题:离轨策略控制与重要性采样

同轨策略要求行为策略和目标策略相同,都是ε-贪心的。但有时我们希望用更富探索性的行为策略(比如完全随机)来收集数据,用以评估和改进一个完全贪心的目标策略。这就是离轨策略(Off-policy) 学习。其关键挑战在于,行为策略下采集的回报不能直接用于估计目标策略的价值,需要用重要性采样(Importance Sampling) 进行修正。

4.1 加权重要性采样原理

重要性采样比(Importance Sampling Ratio)是目标策略和行为策略在给定轨迹下发生概率的比值。对于一幕序列,其重要性采样比ρ为:

ρ = Π_{k=t}^{T-1} [π(A_k|S_k) / b(A_k|S_k)]

其中π是目标策略(我们想评估的策略),b是行为策略(实际采样的策略)。在离轨策略蒙特卡洛控制中,我们常用加权重要性采样来更新Q值,因为它方差更小:

Q(S_t, A_t) ← Q(S_t, A_t) + (ρ_t / C(S_t, A_t)) * (G_t - Q(S_t, A_t))
C(S_t, A_t) ← C(S_t, A_t) + ρ_t

这里C(s, a)是累积权重和。算法流程中,我们从一幕的末尾向前遍历,不断累积重要性采样比ρ。如果某个时刻目标策略选择的动作与行为策略选择的动作不同(即π(A_t|S_t) = 0),则这幕后续的经验对学习目标策略没有用处,可以提前终止该幕的更新。

4.2 离轨策略MC控制代码实现

以下是一个简化的离轨策略蒙特卡洛控制实现。我们假设行为策略b是完全随机的(等概率选择要牌或停牌),而目标策略π是贪心的。

def mc_control_off_policy(env, num_episodes, gamma=1.0):
    """
    离轨策略蒙特卡洛控制算法(加权重要性采样)。
    行为策略b为随机策略,目标策略π为贪心策略。

    返回:
        Q: 最优动作价值函数。
        policy: 贪心目标策略。
    """
    # 初始化
    Q = defaultdict(float)  # 动作价值
    C = defaultdict(float)  # 累积权重和
    # 目标策略初始化为任意值,例如总是停牌
    policy = defaultdict(int)

    # 行为策略:完全随机
    def behavior_policy(state):
        return env.action_space.sample()  # 以0.5概率返回0或1

    for i_episode in range(1, num_episodes + 1):
        # 使用行为策略b生成一幕
        episode = []
        state = env.reset()
        done = False
        while not done:
            action = behavior_policy(state)
            next_state, reward, done, _ = env.step(action)
            episode.append((state, action, reward))
            state = next_state

        # 初始化累计回报G和权重W
        G = 0.0
        W = 1.0

        # 从后向前遍历
        for t in range(len(episode)-1, -1, -1):
            state_t, action_t, reward_t = episode[t]
            G = gamma * G + reward_t
            C[(state_t, action_t)] += W
            # 增量式更新Q值
            Q[(state_t, action_t)] += (W / C[(state_t, action_t)]) * (G - Q[(state_t, action_t)])

            # 策略改进:更新目标策略为对当前Q的贪心策略
            if Q[(state_t, 0)] >= Q[(state_t, 1)]:
                policy[state_t] = 0
            else:
                policy[state_t] = 1

            # 如果行为策略选择的动作不是目标策略选择的动作,则提前终止本幕的更新
            if action_t != policy[state_t]:
                break

            # 更新重要性采样权重W
            # 目标策略π是确定性的(贪心),所以π(action_t|state_t) = 1
            # 行为策略b是均匀随机,所以b(action_t|state_t) = 0.5
            W = W * 1.0 / 0.5  # 即 W = W * 2

    return Q, policy

离轨策略方法的优势在于,行为策略可以持续探索,不受目标策略限制,理论上能更全面地覆盖状态-动作空间。但代价是引入了重要性采样,增加了算法的复杂性,并且学习速度可能较慢,因为许多样本的权重W会变得很小(当行为策略与目标策略差异大时)。

在实际运行中,你会发现离轨策略控制算法同样能收敛到与同轨策略相似的最优策略,但收敛所需的幕数可能更多,因为随机行为策略产生的数据效率相对较低。不过,它在某些探索要求极高的场景下具有不可替代的价值。

5. 工程实践:优化、调试与扩展

将算法跑通只是第一步。要让代码高效、健壮并易于扩展到其他问题,还需要考虑一些工程细节。

5.1 增量更新与性能优化

在前面的代码中,我们使用了returns_sumreturns_count来存储历史回报并计算平均值。这种方式直观但内存效率不高。更优雅的方式是使用增量式更新公式,无需存储所有历史数据:

Q(s, a) ← Q(s, a) + α * (G - Q(s, a))

其中α是学习率。在蒙特卡洛中,常使用随时间递减的学习率,例如α = 1 / N(s, a),其中N(s, a)是访问次数。这等价于计算算术平均。我们可以修改更新步骤:

# 替换原来的平均值计算
returns_count[(state_t, action_t)] += 1
N = returns_count[(state_t, action_t)]
Q[(state_t, action_t)] += (G - Q[(state_t, action_t)]) / N

对于加权重要性采样,增量更新公式已在前面算法中给出。使用增量更新不仅能节省内存,还能让代码更简洁。

5.2 调试与收敛性检查

在运行大量幕数的学习过程中,如何知道算法是否在正确学习?除了最终的可视化,我们还可以监控一些中间指标:

  1. 跟踪平均奖励:每完成一定幕数(如1万幕),计算这段时间内的平均单局奖励。随着学习进行,这个平均值应该呈现上升趋势并最终稳定在一个正值附近(对于Blackjack,最优策略的期望收益约为+0.04左右)。
  2. 检查策略稳定性:定期(如每10万幕)保存一次策略,比较相邻周期策略的差异。当策略不再变化或变化极小时,可以认为已收敛。
  3. 验证特定状态:选择几个关键状态(如(player=16, dealer=10, usable_ace=False)),跟踪其Q值或最优动作的变化过程,观察其是否收敛到合理值。

5.3 扩展到其他Gym环境

蒙特卡洛方法的代码框架具有很好的通用性。要将其应用到其他Gym的离散环境(如FrozenLake, CliffWalking),你通常只需要修改以下几点:

  • 状态和动作的表示:确保你的Q字典或数组能正确索引到环境返回的状态。有些环境状态可能是整数,而不是Blackjack这样的元组。
  • 策略函数接口:保持policy(state)的输入输出格式一致。
  • 折扣因子γ:对于非回合制或带有折扣的任务,需要设置gamma < 1
  • 探索策略:根据问题调整ε值或探索策略。对于状态空间更大的问题,可能需要更复杂的探索机制。

例如,将其迁移到FrozenLake-v1环境(一个4x4的网格世界),你主要需要处理状态是0-15的整数,并调整可视化部分。算法的核心循环几乎无需改动。

通过这个从Blackjack游戏出发的完整实战,我们不仅实现了蒙特卡洛预测和控制,还深入到了同轨与离轨策略的差异、重要性采样的原理以及工程实现的优化技巧。强化学习并非遥不可及的理论,它始于这样一段段与环境交互、从经验中学习的代码。当你看到热力图中那个由算法自动学出的、符合人类专家直觉的最优策略时,你便能真切体会到智能体“学习”的过程。接下来,你可以尝试调整探索率ε,观察它对学习速度和最终策略的影响;或者将环境换成FrozenLake,看看蒙特卡洛方法如何解决序列决策问题。记住,在强化学习中,没有比亲手实现并观察其运行更能加深理解的方式了。

Logo

这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容,并连接云服务、办公空间等稀缺资源,助你专注创造,无忧运营。

更多推荐