用Python实战Model-based强化学习：从零实现值迭代与策略迭代

在传统强化学习教材中，数学公式和算法伪代码常常让学习者望而生畏。本文将通过Python代码实现一个完整的网格世界（Grid World）环境，并逐步构建值迭代和策略迭代算法。我们将用NumPy处理矩阵运算，Matplotlib可视化策略更新过程，最后对比两种算法的收敛特性与适用场景。

1. 搭建网格世界环境

任何强化学习实践都需要一个明确的环境模型。我们首先构建一个5x5的网格世界：

import numpy as np
import matplotlib.pyplot as plt

class GridWorld:
    def __init__(self, size=5):
        self.size = size
        self.walls = [(1,1), (2,3), (3,1)]  # 障碍物坐标
        self.goal = (4,4)  # 目标位置
        self.trap = (0,4)   # 陷阱位置
        self.actions = ['up', 'down', 'left', 'right']
        
    def step(self, state, action):
        x, y = state
        if action == 'up': x = max(0, x-1)
        elif action == 'down': x = min(self.size-1, x+1)
        elif action == 'left': y = max(0, y-1)
        elif action == 'right': y = min(self.size-1, y+1)
        
        next_state = (x, y)
        if next_state in self.walls:  # 撞墙保持原位
            next_state = state
        if next_state == self.goal:
            reward = 10
            done = True
        elif next_state == self.trap:
            reward = -10
            done = True
        else:
            reward = -1  # 每步有小惩罚
            done = False
        return next_state, reward, done

这个环境包含以下关键元素：

• 状态空间 ：5x5网格，共25个状态
• 动作空间 ：上、下、左、右四个基本动作
• 转移规则 ：碰到边界或障碍物保持原位
• 奖励机制 ：
  • 到达目标点：+10
  • 掉入陷阱：-10
  • 每移动一步：-1（鼓励高效路径）

2. 值迭代算法实现

值迭代的核心思想是通过不断更新状态价值函数来逼近最优值函数。以下是完整实现：

def value_iteration(env, gamma=0.9, theta=1e-4):
    V = np.zeros((env.size, env.size))
    policy = np.empty((env.size, env.size), dtype=object)
    
    while True:
        delta = 0
        for i in range(env.size):
            for j in range(env.size):
                if (i,j) == env.goal or (i,j) == env.trap:
                    continue
                    
                v_old = V[i,j]
                q_values = []
                for action in env.actions:
                    (next_i, next_j), reward, _ = env.step((i,j), action)
                    q = reward + gamma * V[next_i, next_j]
                    q_values.append(q)
                
                V[i,j] = max(q_values)
                delta = max(delta, abs(v_old - V[i,j]))
        
        if delta < theta:
            break
    
    # 提取最优策略
    for i in range(env.size):
        for j in range(env.size):
            if (i,j) == env.goal:
                policy[i,j] = 'goal'
            elif (i,j) == env.trap:
                policy[i,j] = 'trap'
            else:
                q_values = []
                for action in env.actions:
                    (next_i, next_j), reward, _ = env.step((i,j), action)
                    q = reward + gamma * V[next_i, next_j]
                    q_values.append(q)
                best_action = env.actions[np.argmax(q_values)]
                policy[i,j] = best_action
    
    return V, policy

关键实现细节：

1. 初始化 ：价值函数V初始化为全零矩阵
2. 循环终止条件 ：当最大价值变化小于阈值θ时停止
3. 状态价值更新 ：使用贝尔曼最优方程更新每个状态价值
4. 策略提取 ：在所有状态价值收敛后，根据最优价值函数提取策略

可视化收敛过程：

def plot_value_convergence(V_history):
    plt.figure(figsize=(10,6))
    for i in range(V_history[0].shape[0]):
        for j in range(V_history[0].shape[1]):
            plt.plot([v[i,j] for v in V_history], 
                    label=f'({i},{j})')
    plt.xlabel('Iterations')
    plt.ylabel('State Value')
    plt.title('Value Iteration Convergence')
    plt.legend()
    plt.show()

3. 策略迭代算法实现

策略迭代包含策略评估和策略提升两个交替进行的阶段：

def policy_iteration(env, gamma=0.9, theta=1e-4):
    # 初始化随机策略
    policy = np.random.choice(env.actions, size=(env.size, env.size))
    policy[env.goal] = 'goal'
    policy[env.trap] = 'trap'
    
    V = np.zeros((env.size, env.size))
    
    while True:
        # 策略评估
        while True:
            delta = 0
            for i in range(env.size):
                for j in range(env.size):
                    if (i,j) == env.goal or (i,j) == env.trap:
                        continue
                        
                    v_old = V[i,j]
                    action = policy[i,j]
                    (next_i, next_j), reward, _ = env.step((i,j), action)
                    V[i,j] = reward + gamma * V[next_i, next_j]
                    delta = max(delta, abs(v_old - V[i,j]))
            
            if delta < theta:
                break
        
        # 策略提升
        policy_stable = True
        for i in range(env.size):
            for j in range(env.size):
                if (i,j) == env.goal or (i,j) == env.trap:
                    continue
                    
                old_action = policy[i,j]
                q_values = []
                for action in env.actions:
                    (next_i, next_j), reward, _ = env.step((i,j), action)
                    q = reward + gamma * V[next_i, next_j]
                    q_values.append(q)
                
                best_action = env.actions[np.argmax(q_values)]
                policy[i,j] = best_action
                
                if old_action != best_action:
                    policy_stable = False
        
        if policy_stable:
            break
    
    return V, policy

算法特点对比：

特性	值迭代	策略迭代
计算复杂度	O(S²A) per iteration	O(S³) per evaluation
收敛速度	线性收敛	二次收敛
内存需求	只需存储V	需存储V和π
适用场景	大状态空间	小状态空间
策略更新频率	最后一次性提取	每次迭代都更新

4. 算法对比与实战技巧

在实际运行中，我们发现两种算法各有优劣：

值迭代特点 ：

• 每次迭代更新所有状态价值
• 不需要完整策略评估
• 适合状态空间较大的问题
• 代码实现相对简单

策略迭代特点 ：

• 策略评估阶段需要完全收敛
• 通常需要更少的迭代次数
• 适合可以承受每次迭代高计算量的场景
• 策略质量在每次迭代都有提升

实用调试技巧：

1. 收敛监控 ：

# 在值迭代中添加
print(f"Iteration {k}, Delta: {delta:.4f}")

# 在策略迭代中添加
print(f"Policy improved {np.sum(policy != new_policy)} states")

2. 可视化工具 ：

def plot_policy(policy, env):
    arrows = {'up':'↑', 'down':'↓', 'left':'←', 'right':'→',
             'goal':'G', 'trap':'X'}
    grid = np.empty((env.size, env.size), dtype='U2')
    for i in range(env.size):
        for j in range(env.size):
            grid[i,j] = arrows[policy[i,j]]
    print(grid)

3. 超参数调优 ：

• γ (折扣因子)：通常在0.9-0.99之间
• θ (收敛阈值)：一般设为1e-4到1e-6
• 最大迭代次数：防止无限循环

实际项目中，我通常会先用策略迭代在小规模问题上验证思路，再对大规模问题采用值迭代。当状态空间特别大时，可以考虑异步更新策略，只随机选择部分状态进行更新。