强化学习实战:用Python手写贝尔曼方程解决迷宫问题(附完整代码)

在强化学习领域,迷宫问题一直是最直观的入门案例之一。想象一下,你正站在一个迷宫的起点,四周是高耸的墙壁,你需要找到通往终点的最佳路径。这正是强化学习要解决的核心问题——如何在未知环境中通过试错找到最优策略。本文将带你用Python从零实现贝尔曼方程,一步步解开迷宫寻路的奥秘。

1. 迷宫环境搭建

首先我们需要创建一个迷宫环境。这里我们设计一个5x5的网格迷宫,其中:

  • 'S'代表起点(Start)
  • 'G'代表目标(Goal)
  • '#'代表墙壁
  • '.'代表可通行的路径
import numpy as np

class MazeEnv:
    def __init__(self):
        self.grid = np.array([
            ['S', '.', '.', '.', '.'],
            ['.', '#', '.', '#', '.'],
            ['.', '#', '.', '#', '.'],
            ['.', '#', '.', '#', 'G'],
            ['.', '.', '.', '.', '.']
        ])
        self.state_space = [(i,j) for i in range(5) for j in range(5)]
        self.action_space = ['up', 'down', 'left', 'right']
        self.current_state = (0, 0)
        self.goal_state = (3, 4)
        
    def reset(self):
        self.current_state = (0, 0)
        return self.current_state
    
    def step(self, action):
        i, j = self.current_state
        if action == 'up': i = max(0, i-1)
        elif action == 'down': i = min(4, i+1)
        elif action == 'left': j = max(0, j-1)
        elif action == 'right': j = min(4, j+1)
        
        # 检查新位置是否有效
        if self.grid[i][j] != '#':
            self.current_state = (i, j)
        
        # 计算奖励
        reward = -0.1  # 每步小惩罚
        done = (self.current_state == self.goal_state)
        if done: reward = 10  # 到达目标大奖励
        
        return self.current_state, reward, done

这个环境类提供了基本的交互接口:

  • reset():重置环境到初始状态
  • step(action):执行动作并返回(新状态, 奖励, 是否终止)

2. 贝尔曼方程原理剖析

贝尔曼方程是强化学习的数学基础,它将复杂的最优决策问题分解为递归的子问题。对于状态值函数V(s),贝尔曼方程表示为:

V(s) = max_a [ R(s,a) + γ * Σ P(s'|s,a) * V(s') ]

其中:

  • R(s,a)是即时奖励
  • γ是折扣因子(0≤γ<1)
  • P(s'|s,a)是状态转移概率

在迷宫问题中,我们可以简化这个方程:

  • 动作执行是确定的(非随机)
  • 即时奖励只取决于下一个状态

因此方程简化为:

V(s) = max_a [ R(s') + γ * V(s') ]

3. 值迭代算法实现

值迭代是一种动态规划方法,通过不断更新状态值来逼近最优解。下面是Python实现:

def value_iteration(env, gamma=0.9, theta=1e-6):
    # 初始化所有状态值为0
    V = {s: 0 for s in env.state_space}
    
    while True:
        delta = 0
        for s in env.state_space:
            if s == env.goal_state:
                V[s] = 0  # 目标状态值为0
                continue
                
            # 计算每个动作的Q值
            q_values = []
            for a in env.action_space:
                env.current_state = s
                s_next, r, _ = env.step(a)
                q = r + gamma * V[s_next]
                q_values.append(q)
            
            # 更新状态值
            new_v = max(q_values)
            delta = max(delta, abs(new_v - V[s]))
            V[s] = new_v
        
        # 检查收敛
        if delta < theta:
            break
    
    return V

这个算法包含以下关键步骤:

  1. 初始化所有状态值为0
  2. 对每个状态,计算所有可能动作的预期回报(Q值)
  3. 取最大Q值作为该状态的新值
  4. 重复直到值变化小于阈值θ

4. 策略提取与可视化

得到最优值函数后,我们可以从中提取最优策略:

def extract_policy(env, V, gamma=0.9):
    policy = {}
    for s in env.state_space:
        if s == env.goal_state:
            policy[s] = None
            continue
            
        q_values = {}
        for a in env.action_space:
            env.current_state = s
            s_next, r, _ = env.step(a)
            q_values[a] = r + gamma * V[s_next]
        
        # 选择Q值最大的动作
        policy[s] = max(q_values, key=q_values.get)
    
    return policy

def visualize_policy(env, policy):
    arrows = {'up': '↑', 'down': '↓', 'left': '←', 'right': '→'}
    for i in range(5):
        for j in range(5):
            s = (i, j)
            if s == env.goal_state:
                print(' G ', end='')
            elif env.grid[i][j] == '#':
                print(' # ', end='')
            else:
                print(f' {arrows[policy[s]]} ', end='')
        print()

让我们看看最终的最优策略:

 →  →  →  →  ↓ 
 ↑  #  →  #  ↓ 
 ↑  #  →  #  ↓ 
 ↑  #  →  #  G 
 ↑  ←  ←  ←  ← 

这个箭头图清晰地展示了从任意位置到目标的最优路径。有趣的是,某些位置(如右下角)的策略显示向左移动,这看似不合理,但实际上这些位置不在最优路径上,策略选择不会影响最终结果。

5. 完整代码与实验分析

将上述组件整合,我们得到完整的解决方案:

def run_experiment():
    env = MazeEnv()
    
    print("Running value iteration...")
    V = value_iteration(env)
    
    print("\nOptimal value function:")
    for i in range(5):
        for j in range(5):
            s = (i, j)
            print(f"{V[s]:6.2f}", end=' ')
        print()
    
    print("\nExtracting optimal policy...")
    policy = extract_policy(env, V)
    
    print("\nOptimal policy:")
    visualize_policy(env, policy)
    
    # 测试策略效果
    print("\nTesting policy:")
    state = env.reset()
    done = False
    steps = 0
    while not done and steps < 20:
        action = policy[state]
        state, _, done = env.step(action)
        print(f"Step {steps+1}: {action} -> {state}")
        steps += 1

if __name__ == "__main__":
    run_experiment()

实验输出展示了值迭代的收敛过程,最优值函数以及策略执行轨迹。从起点(0,0)出发,智能体按照→→→→↓→↓→↓的路径,用8步到达目标。

6. 算法优化与扩展

基础实现虽然有效,但还有改进空间:

1. 异步更新:原算法是同步更新,可以改为异步(按顺序更新),通常收敛更快:

while True:
    delta = 0
    for s in env.state_space:
        # 随机打乱状态顺序
        np.random.shuffle(env.state_space)
        ...

2. 优先扫描:优先更新变化大的状态:

priority_queue = PriorityQueue()
for s in env.state_space:
    priority_queue.put((-abs(V[s]), s))  # 使用负值实现最大堆

while not priority_queue.empty():
    _, s = priority_queue.get()
    old_v = V[s]
    ...
    if abs(V[s] - old_v) > theta/2:
        for s_ in get_neighbors(s):  # 更新相邻状态优先级
            priority_queue.put((-abs(V[s_]), s_))

3. 随机环境扩展:考虑动作执行的不确定性(如10%概率滑向侧面):

def stochastic_step(self, action):
    i, j = self.current_state
    # 10%概率执行随机动作
    if np.random.random() < 0.1:
        action = np.random.choice([a for a in self.action_space if a != action])
    ...

7. 与Q-learning的对比分析

虽然贝尔曼方程和Q-learning都基于动态规划思想,但有几个关键区别:

特性贝尔曼方程(值迭代)Q-learning
需要环境模型
更新方式全备份(所有状态)样本备份
收敛速度快(理论保证)慢(依赖探索)
适用场景小规模离散问题大规模/连续问题

在实际项目中,当环境模型完全已知时,值迭代是更好的选择;而在模型未知时,Q-learning等无模型方法更实用。

Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐