别再死记硬背MDP公式了！用Python手搓一个‘走迷宫’游戏，5分钟搞懂马尔可夫决策过程

爱燃烧

251人浏览 · 2026-04-18 09:01:50

爱燃烧 · 2026-04-18 09:01:50 发布

用Python构建迷宫游戏：5分钟直观理解马尔可夫决策过程

从游戏视角看强化学习核心概念

第一次接触强化学习中的马尔可夫决策过程（MDP）时，那些晦涩的数学公式往往让人望而生畏。但如果我们换一个角度，用编写迷宫游戏的方式来实现MDP的核心组件，一切就会变得清晰起来。本文将带您用Python构建一个简单的迷宫环境，在代码实践中理解状态、动作、转移概率和奖励这些关键概念。

想象一下，您正在设计一个2D迷宫游戏：玩家从起点出发，目标是找到出口。每走一步都会消耗体力（负奖励），找到出口获得大奖（正奖励），碰到墙壁则停留在原地。这个看似简单的游戏场景，其实包含了MDP的所有要素。让我们用代码把它实现出来。

1. 搭建迷宫游戏环境

首先创建一个5x5的迷宫网格环境。我们将使用面向对象的方式构建，便于后续扩展：

import numpy as np
from typing import Tuple, Dict

class MazeEnv:
    def __init__(self):
        # 定义迷宫布局：0=通路，1=墙，2=起点，3=终点
        self.grid = np.array([
            [2, 0, 0, 0, 0],
            [0, 1, 1, 1, 0], 
            [0, 1, 3, 0, 0],
            [0, 1, 0, 1, 0],
            [0, 0, 0, 1, 0]
        ])
        self.start_pos = (0, 0)  # 起点坐标
        self.current_pos = self.start_pos
        self.actions = ['up', 'down', 'left', 'right']  # 可用动作
        
    def reset(self) -> Tuple[int, int]:
        """重置环境到初始状态"""
        self.current_pos = self.start_pos
        return self.current_pos
    
    def step(self, action: str) -> Tuple[Tuple[int, int], float, bool, Dict]:
        """执行动作并返回(next_state, reward, done, info)"""
        x, y = self.current_pos
        reward = -0.1  # 每步小惩罚鼓励尽快找到出口
        
        # 根据动作计算新位置
        if action == 'up' and x > 0:
            new_pos = (x-1, y)
        elif action == 'down' and x < 4:
            new_pos = (x+1, y)
        elif action == 'left' and y > 0:
            new_pos = (x, y-1)
        elif action == 'right' and y < 4:
            new_pos = (x, y+1)
        else:
            new_pos = (x, y)  # 无效动作保持原位
            
        # 检查新位置属性
        cell_value = self.grid[new_pos]
        if cell_value == 1:  # 撞墙
            new_pos = (x, y)
            reward = -0.5
        elif cell_value == 3:  # 到达终点
            reward = 10.0
            done = True
        else:
            done = False
            
        self.current_pos = new_pos
        return new_pos, reward, done, {}

这个环境已经包含了MDP的四个关键要素：

状态(State): 玩家的当前位置坐标 (x,y)
动作(Action): 上下左右移动
转移概率(Transition): 在step()方法中实现的状态转换逻辑
奖励(Reward): 根据到达的格子类型给予不同奖励

2. 实现MDP五元组

现在让我们将迷宫游戏映射到MDP的数学定义上。一个完整的MDP由五元组(S, A, P, R, γ)构成：

2.1 状态空间(S)和动作空间(A)

在我们的迷宫中：

S = [(x,y) for x in range(5) for y in range(5) if env.grid[x,y] != 1]  # 所有非墙位置
A = ['up', 'down', 'left', 'right']  # 可能的动作

2.2 转移函数(P)

转移函数P(s'|s,a)定义了在状态s执行动作a后转移到状态s'的概率。在我们的确定性环境中，动作执行结果完全可预测：

def transition_prob(s: Tuple[int,int], a: str, s_next: Tuple[int,int]) -> float:
    """计算P(s'|s,a)"""
    x, y = s
    if a == 'up':
        expected = (x-1, y)
    elif a == 'down':
        expected = (x+1, y)
    elif a == 'left':
        expected = (x, y-1)
    elif a == 'right':
        expected = (x, y+1)
    
    # 检查是否撞墙
    if s_next == expected and env.grid[s_next] != 1:
        return 1.0
    elif s_next == s and (env.grid[expected] == 1 or expected[0]<0 or expected[0]>4 or expected[1]<0 or expected[1]>4):
        return 1.0
    return 0.0

2.3 奖励函数(R)

奖励函数R(s,a,s')定义了在状态s执行动作a到达状态s'时获得的即时奖励：

def reward_func(s: Tuple[int,int], a: str, s_next: Tuple[int,int]) -> float:
    """计算R(s,a,s')"""
    if env.grid[s_next] == 3:  # 到达终点
        return 10.0
    elif s_next == s:  # 撞墙或无效移动
        return -0.5
    else:  # 普通移动
        return -0.1

2.4 折扣因子(γ)

折扣因子γ∈[0,1]决定了未来奖励的现值。我们暂时设为0.9：

gamma = 0.9  # 重视近期奖励

3. 可视化状态转移与价值

理解MDP的关键是看到状态之间的转移关系。我们可以用networkx库绘制状态转移图：

import networkx as nx
import matplotlib.pyplot as plt

def build_transition_graph():
    G = nx.DiGraph()
    
    for s in S:
        for a in A:
            for s_next in S:
                prob = transition_prob(s, a, s_next)
                if prob > 0:
                    r = reward_func(s, a, s_next)
                    G.add_edge(s, s_next, action=a, prob=prob, reward=r)
    
    return G

# 绘制状态转移图
G = build_transition_graph()
pos = {(x,y): (y,-x) for x,y in G.nodes()}  # 保持迷宫布局

plt.figure(figsize=(10,8))
nx.draw(G, pos, with_labels=True, node_color='lightblue')
edge_labels = {(u,v): f"{d['action']}\nR:{d['reward']:.1f}" 
               for u,v,d in G.edges(data=True)}
nx.draw_networkx_edge_labels(G, pos, edge_labels=edge_labels, font_size=8)
plt.title("迷宫状态转移图")
plt.show()

这张图清晰地展示了：

每个状态节点
动作导致的转移边
每条边上的即时奖励
撞墙时的自循环

4. 实现随机策略与价值计算

策略π(a|s)定义了在状态s下选择动作a的概率分布。我们先实现一个随机策略：

def random_policy(state: Tuple[int,int]) -> Dict[str, float]:
    """均等随机选择动作"""
    return {a: 0.25 for a in A}

根据Bellman方程，状态价值函数V(s)可以递归计算：

def calculate_value_function(policy, theta=0.0001):
    """迭代计算给定策略下的状态价值函数"""
    V = {s: 0 for s in S}  # 初始化价值函数
    
    while True:
        delta = 0
        for s in S:
            v = V[s]
            total = 0
            for a in A:
                action_prob = policy(s)[a]
                for s_next in S:
                    prob = transition_prob(s, a, s_next)
                    r = reward_func(s, a, s_next)
                    total += action_prob * prob * (r + gamma * V[s_next])
            V[s] = total
            delta = max(delta, abs(v - V[s]))
        
        if delta < theta:  # 收敛判断
            break
    
    return V

让我们计算并可视化随机策略下的状态价值：

random_v = calculate_value_function(random_policy)

# 可视化价值函数
grid_v = np.full((5,5), np.nan)
for (x,y), v in random_v.items():
    grid_v[x,y] = v

plt.imshow(grid_v, cmap='viridis')
plt.colorbar(label='State Value')
for (x,y), v in random_v.items():
    plt.text(y, x, f"{v:.1f}", ha='center', va='center')
plt.title("随机策略下的状态价值")
plt.axis('off')
plt.show()

从价值图中可以看到：

靠近终点的状态价值较高
被墙阻挡的区域价值较低
随机策略下整体价值不高

5. 寻找最优策略

MDP的最终目标是找到最优策略π*，使得每个状态的价值最大化。我们可以使用策略迭代算法：

def policy_iteration():
    # 1. 初始化随机策略
    policy = {s: {a: 0.25 for a in A} for s in S}
    
    while True:
        # 2. 策略评估
        V = calculate_value_function(lambda s: policy[s])
        
        # 3. 策略改进
        policy_stable = True
        for s in S:
            old_action_probs = policy[s].copy()
            
            # 对每个状态寻找最优动作
            action_values = {}
            for a in A:
                total = 0
                for s_next in S:
                    prob = transition_prob(s, a, s_next)
                    r = reward_func(s, a, s_next)
                    total += prob * (r + gamma * V[s_next])
                action_values[a] = total
            
            # 选择价值最高的动作
            best_action = max(action_values, key=action_values.get)
            policy[s] = {a: 1.0 if a == best_action else 0.0 for a in A}
            
            if old_action_probs != policy[s]:
                policy_stable = False
        
        if policy_stable:
            return policy, V

运行策略迭代并可视化最优策略：

optimal_policy, optimal_v = policy_iteration()

# 可视化最优策略
grid_policy = np.full((5,5), '')
for (x,y) in optimal_policy:
    if env.grid[x,y] == 3:  # 终点
        grid_policy[x,y] = 'G'
    else:
        best_a = max(optimal_policy[(x,y)], key=optimal_policy[(x,y)].get)
        grid_policy[x,y] = best_a[0].upper()  # 取动作首字母

plt.imshow(grid_v, cmap='viridis')
plt.colorbar(label='State Value')
for (x,y) in optimal_policy:
    plt.text(y, x, grid_policy[x,y], ha='center', va='center', fontsize=12)
plt.title("最优策略(字母表示动作方向)")
plt.axis('off')
plt.show()

最终我们得到了：

每个状态下的最优动作
一条从起点到终点的最优路径
比随机策略高得多的状态价值

6. 从游戏到通用MDP框架

通过这个迷宫游戏示例，我们直观地理解了MDP的核心组件：

游戏概念	MDP数学概念	代码实现
玩家位置	状态s	`current_pos`
移动方向	动作a	`actions`列表
移动规则	转移概率P(s'\|s,a)	`transition_prob()`
得分机制	奖励R(s,a,s')	`reward_func()`
策略选择	策略π(a\|s)	`random_policy()`
长期得分	价值V(s)	`calculate_value_function()`