用Python玩转强化学习：手把手教你用NumPy实现赌徒问题的MDP求解
在人工智能领域，强化学习正以其独特的决策能力改变着我们解决问题的思维方式。而马尔科夫决策过程（MDP）作为强化学习的理论基础，其重要性不言而喻。今天，我们将通过一个经典案例——赌徒问题，来深入理解MDP的核心概念，并掌握两种经典算法：策略迭代（Policy Iteration）和值迭代（Value Iteration）的Python实现。
赌徒问题看似简单，却完美展现了MDP的所有关键要素：状态、动作、奖励和策略。通过这个案例，我们不仅能理解强化学习的基本原理，还能获得宝贵的编程实践经验。本文将使用NumPy和Matplotlib这两个Python基础库，从零开始构建完整的求解器，并通过可视化手段直观展示算法的收敛过程和最优策略。
1. 赌徒问题与MDP基础
赌徒问题描述了一个简单的赌博场景：玩家每次可以选择下注一定金额，有概率ph赢得等额赌注，也有概率(1-ph)输掉赌注。游戏在玩家资产达到100美元或归零时结束。这个看似简单的设定，却包含了MDP的所有核心要素：
- 状态（State）：当前持有的资本金额，s ∈ {0,1,2,...,99}
- 动作（Action）：每次下注的金额，a ∈ {1,2,...,min(s,100-s)}
- 奖励（Reward）：只有达到100美元时获得+1奖励，其他情况为0
- 折扣因子（γ）：这里设为1，表示不考虑未来奖励的折扣
理解这些基本概念是后续算法实现的基础。值得注意的是，赌徒问题的状态空间和动作空间都是离散且有限的，这使得它成为学习MDP的理想案例。
# 基本参数设置示例 GOAL = 100 # 目标金额 ph = 0.4 # 硬币正面朝上的概率 states = np.arange(GOAL + 1) # 所有可能的状态2. 策略迭代算法实现
策略迭代是解决MDP问题的经典方法之一，它通过交替进行策略评估和策略改进来逐步优化策略。下面我们详细分解实现步骤：
2.1 策略评估
策略评估的目标是计算当前策略下的状态价值函数。我们采用迭代法，直到价值函数收敛（变化小于阈值θ）：
def policy_evaluation(self): while True: old_state_values = self.state_values.copy() self.sweeps_history.append(old_state_values) # 记录迭代历史 for s in self.states[1:self.goal]: # 跳过0和100状态 actions = np.arange(min(s, self.goal - s)) + 1 action_returns = [] for a in actions: # 计算动作a的期望回报 ret = self.ph * (self.gamma * self.state_values[s + a]) + \ (1 - self.ph) * (self.gamma * self.state_values[s - a]) action_returns.append(ret) # 使用当前策略选择动作（这里假设策略是平均选择） self.state_values[s] = np.mean(action_returns) # 检查收敛 delta = abs(self.state_values - old_state_values).max() if delta <= self.theta: break2.2 策略改进
在得到准确的价值函数后，我们可以改进策略，使其在每个状态都选择能带来最大期望回报的动作：
def policy_improvement(self): policy_stable = True for s in self.states[1:self.goal]: actions = np.arange(min(s, self.goal - s)) + 1 old_a = self.policy[s] action_returns = [] for a in actions: ret = self.ph * (self.gamma * self.state_values[s + a]) + \ (1 - self.ph) * (self.gamma * self.state_values[s - a]) action_returns.append(ret) # 选择回报最大的动作 argmax_a = actions[np.argmax(np.round(action_returns, 5))] self.policy[s] = argmax_a if policy_stable and argmax_a != old_a: policy_stable = False return policy_stable2.3 完整策略迭代流程
将策略评估和策略改进结合起来，就构成了完整的策略迭代算法：
class PolicyIteration: def __init__(self, goal=100, proba_h=0.4, theta=1e-9, gamma=1): self.ph = proba_h self.gamma = gamma self.goal = goal self.theta = theta self.states = np.arange(self.goal + 1) self.state_values = np.zeros(self.goal + 1) self.state_values[self.goal] = 1.0 # 达到目标的价值为1 self.policy = np.zeros(self.goal + 1) self.sweeps_history = [] def iterate(self): while True: self.policy_evaluation() policy_stable = self.policy_improvement() if policy_stable: break3. 值迭代算法实现
值迭代是另一种解决MDP的重要方法，它将策略评估和策略改进结合在一个步骤中，通常收敛更快。
3.1 值迭代核心算法
值迭代直接寻找最优价值函数，然后在收敛后提取最优策略：
def truncated_policy_evaluation(self): while True: old_state_values = self.state_values.copy() self.sweeps_history.append(old_state_values) for s in self.states[1:self.goal]: actions = np.arange(min(s, self.goal - s)) + 1 action_returns = [] for a in actions: ret = self.ph * (self.gamma * self.state_values[s + a]) + \ (1 - self.ph) * (self.gamma * self.state_values[s - a]) action_returns.append(ret) # 关键区别：直接取最大值而不是平均值 self.state_values[s] = np.max(action_returns) delta = abs(self.state_values - old_state_values).max() if delta <= self.theta: break3.2 提取最优策略
在值函数收敛后，我们可以提取出最优策略：
def optimal_policy(self): for s in self.states[1:self.goal]: actions = np.arange(min(s, self.goal - s)) + 1 action_returns = [] for a in actions: ret = self.ph * (self.gamma * self.state_values[s + a]) + \ (1 - self.ph) * (self.gamma * self.state_values[s - a]) action_returns.append(ret) argmax_a = actions[np.argmax(np.round(action_returns, 5))] self.policy[s] = argmax_a3.3 完整值迭代实现
class ValueIteration: def __init__(self, goal=100, proba_h=0.4, theta=1e-9, gamma=1): self.ph = proba_h self.gamma = gamma self.goal = goal self.theta = theta self.states = np.arange(self.goal + 1) self.state_values = np.zeros(self.goal + 1) self.state_values[self.goal] = 1.0 self.policy = np.zeros(self.goal + 1) self.sweeps_history = [] def solve(self): self.truncated_policy_evaluation() self.optimal_policy()4. 结果分析与可视化
实现算法后，我们需要对结果进行分析和可视化，以深入理解两种算法的特点和差异。
4.1 值函数收敛过程
我们可以绘制值函数在迭代过程中的变化：
def plot_value_convergence(sweeps_history, goal=100): plt.figure(figsize=(10, 6)) for i, sweep in enumerate(sweeps_history): if i % 5 == 0: # 每隔5次迭代绘制一次 plt.plot(sweep, alpha=0.8, label=f'Iteration {i}') plt.xlabel('Capital') plt.ylabel('Value Estimates') plt.title('Value Function Convergence') plt.legend() plt.show()4.2 最优策略可视化
最优策略展示了在不同资本状态下应该下注多少：
def plot_policy(policy, goal=100): plt.figure(figsize=(10, 4)) plt.bar(range(goal), policy[:goal]) plt.xlabel('Capital') plt.ylabel('Optimal Stake') plt.title('Optimal Policy') plt.show()4.3 算法对比
策略迭代和值迭代在ph=0.4时的表现对比：
| 特性 | 策略迭代 | 值迭代 |
|---|---|---|
| 收敛速度 | 较慢 | 较快 |
| 每次迭代复杂度 | 较高 | 较低 |
| 策略特性 | 分段式激进与保守 | 周期性激进与保守 |
| 适用场景 | 策略变化不大的问题 | 大规模状态空间问题 |
从实现角度看，值迭代通常更适合大规模问题，而策略迭代在策略空间较小时可能更有效。在赌徒问题中，当ph=0.4时，两种算法得到的最优策略存在显著差异：
- 策略迭代产生的策略更加分段化，在某些资本范围内采取激进策略，在另一些范围则保守
- 值迭代产生的策略则呈现出周期性模式，激进与保守交替出现
这种差异源于两种算法在更新方式上的本质区别：策略迭代在每次迭代中完全评估当前策略，而值迭代则始终朝着最优价值函数方向更新。
