第3章:有限马尔可夫决策过程(Finite Markov Decision Processes)
一、导读
1.1 本章学习目标
第3章是强化学习的数学基础。马尔可夫决策过程(MDP)为强化学习问题提供了严格的形式化框架,几乎所有强化学习理论都可以被理解为 MDP 的特例或推广。本章将建立 MDP 的完整数学体系,包括状态转移、策略、价值函数、贝尔曼方程等核心概念。
本章的核心学习目标包括:
- 理解马尔可夫性质:未来只依赖于当前状态,与历史无关
- 掌握 MDP 的完整定义:状态集合、动作集合、转移概率、奖励函数、折扣因子
- 理解策略(policy)的定义和分类:确定性策略 vs 随机策略
- 掌握价值函数的数学定义:状态价值函数 $v_\pi$ 和动作价值函数 $q_\pi$
- 深入理解贝尔曼方程:价值函数的递归分解,是几乎所有强化学习算法的理论基础
- 理解最优价值函数和最优策略:贝尔曼最优方程及其解的存在性
1.2 为什么本章如此重要
MDP 是强化学习的"语言"。正如概率论为机器学习提供了数学基础,MDP 为强化学习提供了形式化框架。没有 MDP,我们就无法严格地定义"什么是最优策略"、"如何衡量策略的好坏"、"算法是否收敛"等关键问题。
贝尔曼方程是本章最重要的结论。它将一个看似复杂的全局优化问题分解为局部的递归关系,使得我们可以用动态规划、蒙特卡洛方法、时序差分等算法来求解。可以说,后续所有章节的算法都是贝尔曼方程的不同近似或推广。
1.3 核心问题引导
在阅读本章时,请带着以下问题思考:
什么是马尔可夫性质?为什么这个假设对强化学习如此重要?现实中的问题是否都满足马尔可夫性?
状态价值函数和动作价值函数有什么区别?它们之间如何相互转换?
贝尔曼方程为什么能将无限步的累积奖励分解为递归形式?
最优策略是否总是确定性的?在什么情况下随机策略可能更优?
如何从贝尔曼最优方程出发设计算法来求解最优策略?
二、核心概念详解
3.1 马尔可夫性质与 MDP 定义
马尔可夫性质:状态信号 $S_t$ 满足马尔可夫性质,当且仅当:
$$P(S_{t+1} | S_t, S_{t-1}, \ldots, S_0) = P(S_{t+1} | S_t)$$
即未来只依赖于当前状态,与历史无关。这个假设极大地简化了问题:智能体不需要记住整个历史,只需要根据当前状态做出决策。
MDP 的完整定义:一个有限马尔可夫决策过程由五元组 $(\mathcal{S}, \mathcal{A}, P, R, \gamma)$ 定义:
- $\mathcal{S}$:有限的状态集合
- $\mathcal{A}$:有限的动作集合
- $P(s'|s,a)$:状态转移概率,表示在状态 $s$ 执行动作 $a$ 后转移到状态 $s'$ 的概率
- $R(s,a,s')$:奖励函数,表示在状态 $s$ 执行动作 $a$ 转移到 $s'$ 时获得的奖励
- $\gamma \in [0, 1]$:折扣因子
import numpy as np
class SimpleMDP:
def __init__(self):
self.states = [0, 1, 2, 3]
self.actions = ['left', 'right']
self.gamma = 0.9
self.transitions = {
(0, 'right'): {1: 1.0},
(0, 'left'): {0: 1.0},
(1, 'right'): {2: 0.8, 0: 0.2},
(1, 'left'): {0: 0.8, 2: 0.2},
(2, 'right'): {3: 1.0},
(2, 'left'): {1: 1.0},
(3, 'right'): {3: 1.0},
(3, 'left'): {3: 1.0},
}
self.rewards = {
(0, 'right', 1): 0,
(1, 'right', 2): 0,
(1, 'right', 0): 0,
(2, 'right', 3): 1,
(2, 'left', 1): 0,
}
def get_transition_prob(self, s, a, next_s):
return self.transitions.get((s, a), {}).get(next_s, 0.0)
def get_reward(self, s, a, next_s):
return self.rewards.get((s, a, next_s), 0.0)
mdp = SimpleMDP()
print(f"States: {mdp.states}")
print(f"Actions: {mdp.actions}")
print(f"P(2|1,right) = {mdp.get_transition_prob(1, 'right', 2)}")
print(f"R(2,right,3) = {mdp.get_reward(2, 'right', 3)}")3.2 策略
策略 $\pi$ 是从状态到动作概率分布的映射。确定性策略 $\pi(s) = a$ 直接将状态映射到动作;随机策略 $\pi(a|s)$ 给出在状态 $s$ 下选择动作 $a$ 的概率。
class Policy:
def __init__(self, states, actions):
self.states = states
self.actions = actions
n_s = len(states)
n_a = len(actions)
self.pi = np.ones((n_s, n_a)) / n_a
def get_action_prob(self, state_idx, action_idx):
return self.pi[state_idx][action_idx]
def set_deterministic(self, state_idx, action_idx):
self.pi[state_idx] = 0
self.pi[state_idx][action_idx] = 1.0
def get_action(self, state_idx):
return np.random.choice(len(self.actions), p=self.pi[state_idx])
policy = Policy([0, 1, 2, 3], ['left', 'right'])
policy.set_deterministic(0, 1)
policy.set_deterministic(1, 1)
policy.set_deterministic(2, 1)
policy.set_deterministic(3, 0)
print("Policy:")
for s in range(4):
probs = policy.pi[s]
print(f" State {s}: left={probs[0]:.2f}, right={probs[1]:.2f}")3.3 价值函数
状态价值函数 $v_\pi(s)$ 表示在策略 $\pi$ 下从状态 $s$ 出发能获得的期望回报:
$$v_\pi(s) = \mathbb{E}_\pi[G_t | S_t = s] = \mathbb{E}_\pi\left[\sum_{k=0}^{\infty} \gamma^k R_{t+k+1} \Big| S_t = s\right]$$
动作价值函数 $q_\pi(s, a)$ 表示在策略 $\pi$ 下从状态 $s$ 执行动作 $a$ 后能获得的期望回报:
$$q_\pi(s, a) = \mathbb{E}_\pi[G_t | S_t = s, A_t = a] = \mathbb{E}_\pi\left[\sum_{k=0}^{\infty} \gamma^k R_{t+k+1} \Big| S_t = s, A_t = a\right]$$
两者之间的关系:
$$v_\pi(s) = \sum_a \pi(a|s) q_\pi(s, a)$$
$$q_\pi(s, a) = \sum_{s'} P(s'|s,a) \left[ R(s,a,s') + \gamma v_\pi(s') \right]$$
def compute_value_functions(mdp, policy, gamma=0.9, theta=1e-6):
n_states = len(mdp.states)
n_actions = len(mdp.actions)
V = np.zeros(n_states)
Q = np.zeros((n_states, n_actions))
while True:
delta = 0
for s_idx, s in enumerate(mdp.states):
v = V[s_idx]
new_v = 0
for a_idx, a in enumerate(mdp.actions):
prob = policy.get_action_prob(s_idx, a_idx)
if prob == 0:
continue
expected_return = 0
for next_s_idx, next_s in enumerate(mdp.states):
p = mdp.get_transition_prob(s, a, next_s)
r = mdp.get_reward(s, a, next_s)
expected_return += p * (r + gamma * V[next_s_idx])
Q[s_idx][a_idx] = expected_return
new_v += prob * expected_return
V[s_idx] = new_v
delta = max(delta, abs(v - V[s_idx]))
if delta < theta:
break
return V, Q
V, Q = compute_value_functions(mdp, policy, gamma=0.9)
print("State values:", np.round(V, 3))
print("Action values:")
for s_idx in range(4):
print(f" State {s_idx}: left={Q[s_idx][0]:.3f}, right={Q[s_idx][1]:.3f}")3.4 贝尔曼方程
贝尔曼方程是价值函数的递归分解。它将"从当前状态出发的期望回报"分解为"即时奖励 + 折扣后的下一状态价值":
$$v_\pi(s) = \sum_a \pi(a|s) \sum_{s'} P(s'|s,a) \left[ R(s,a,s') + \gamma v_\pi(s') \right]$$
这个方程的深刻之处在于:它将一个无限步的问题分解为一步决策 + 剩余子问题的递归结构。这使得我们可以用迭代法求解价值函数。
def bellman_backup(mdp, V, policy, gamma=0.9):
n_states = len(mdp.states)
n_actions = len(mdp.actions)
new_V = np.zeros(n_states)
for s_idx, s in enumerate(mdp.states):
for a_idx, a in enumerate(mdp.actions):
prob = policy.get_action_prob(s_idx, a_idx)
if prob == 0:
continue
expected_return = 0
for next_s_idx, next_s in enumerate(mdp.states):
p = mdp.get_transition_prob(s, a, next_s)
r = mdp.get_reward(s, a, next_s)
expected_return += p * (r + gamma * V[next_s_idx])
new_V[s_idx] += prob * expected_return
return new_V
V_init = np.zeros(4)
for i in range(20):
V_init = bellman_backup(mdp, V_init, policy, gamma=0.9)
if i % 5 == 0:
print(f"Iteration {i}: V = {np.round(V_init, 3)}")
print(f"Final V = {np.round(V_init, 3)}")3.5 最优价值函数与最优策略
最优状态价值函数 $v_(s) = \max_\pi v_\pi(s)$,**最优动作价值函数** $q_(s,a) = \max_\pi q_\pi(s,a)$。
贝尔曼最优方程:
$$v_(s) = \max_a \sum_{s'} P(s'|s,a) \left[ R(s,a,s') + \gamma v_(s') \right]$$
$$q_(s,a) = \sum_{s'} P(s'|s,a) \left[ R(s,a,s') + \gamma \max_{a'} q_(s',a') \right]$$
最优策略可以从最优价值函数直接得到:
$$\pi_(s) = \arg\max_a q_(s, a)$$
def value_iteration(mdp, gamma=0.9, theta=1e-6):
n_states = len(mdp.states)
n_actions = len(mdp.actions)
V = np.zeros(n_states)
while True:
delta = 0
for s_idx, s in enumerate(mdp.states):
v = V[s_idx]
max_val = -float('inf')
for a_idx, a in enumerate(mdp.actions):
expected_return = 0
for next_s_idx, next_s in enumerate(mdp.states):
p = mdp.get_transition_prob(s, a, next_s)
r = mdp.get_reward(s, a, next_s)
expected_return += p * (r + gamma * V[next_s_idx])
max_val = max(max_val, expected_return)
V[s_idx] = max_val
delta = max(delta, abs(v - V[s_idx]))
if delta < theta:
break
optimal_policy = np.zeros((n_states, n_actions))
for s_idx, s in enumerate(mdp.states):
best_action = -1
best_value = -float('inf')
for a_idx, a in enumerate(mdp.actions):
expected_return = 0
for next_s_idx, next_s in enumerate(mdp.states):
p = mdp.get_transition_prob(s, a, next_s)
r = mdp.get_reward(s, a, next_s)
expected_return += p * (r + gamma * V[next_s_idx])
if expected_return > best_value:
best_value = expected_return
best_action = a_idx
optimal_policy[s_idx][best_action] = 1.0
return V, optimal_policy
V_opt, pi_opt = value_iteration(mdp, gamma=0.9)
print("Optimal state values:", np.round(V_opt, 3))
print("Optimal policy:")
for s_idx in range(4):
action_idx = np.argmax(pi_opt[s_idx])
print(f" State {s_idx}: {mdp.actions[action_idx]}")3.6 马尔可夫性质的实际意义
马尔可夫性质是一个强假设。现实中的很多问题并不严格满足马尔可夫性——围棋的打劫规则需要记住之前的棋局,自动驾驶需要了解之前的交通状况。但在实践中,我们可以通过状态表示的设计来恢复马尔可夫性:将历史信息编码到状态中。
class NonMarkovianEnv:
def __init__(self):
self.history = []
self.secret_state = 0
def step(self, action):
self.history.append(action)
if len(self.history) >= 3 and self.history[-3:] == [1, 0, 1]:
reward = 10
else:
reward = 0
return self.secret_state, reward
class MarkovianWrapper:
def __init__(self, env, history_length=3):
self.env = env
self.history_length = history_length
self.action_history = []
def step(self, action):
self.action_history.append(action)
if len(self.action_history) > self.history_length:
self.action_history = self.action_history[-self.history_length:]
state = tuple(self.action_history)
_, reward = self.env.step(action)
return state, reward
def reset(self):
self.action_history = []
self.env.history = []
return tuple(self.action_history)
env = NonMarkovianEnv()
wrapped = MarkovianWrapper(env, history_length=3)
state = wrapped.reset()
for _ in range(5):
action = np.random.randint(2)
state, reward = wrapped.step(action)
print(f"State: {state}, Reward: {reward}")三、实战案例
3.1 案例一:Grid World 的完整 MDP 建模
class GridWorldMDP:
def __init__(self, width=4, height=3):
self.width = width
self.height = height
self.n_states = width * height
self.n_actions = 4
self.gamma = 0.9
self.actions = [(0, 1), (0, -1), (1, 0), (-1, 0)]
self.goal = (height - 1, width - 1)
self.trap = (0, width - 1)
def state_to_pos(self, s):
return (s // self.width, s % self.width)
def pos_to_state(self, pos):
return pos[0] * self.width + pos[1]
def get_transition_prob(self, s, a):
r, c = self.state_to_pos(s)
dr, dc = self.actions[a]
new_r = max(0, min(self.height - 1, r + dr))
new_c = max(0, min(self.width - 1, c + dc))
next_s = self.pos_to_state((new_r, new_c))
return {next_s: 1.0}
def get_reward(self, s, a, next_s):
pos = self.state_to_pos(next_s)
if pos == self.goal:
return 1.0
if pos == self.trap:
return -1.0
return 0.0
def solve_value_iteration(self, theta=1e-6):
V = np.zeros(self.n_states)
while True:
delta = 0
for s in range(self.n_states):
if self.state_to_pos(s) in [self.goal, self.trap]:
continue
v = V[s]
max_val = -float('inf')
for a in range(self.n_actions):
transitions = self.get_transition_prob(s, a)
expected = 0
for next_s, p in transitions.items():
r = self.get_reward(s, a, next_s)
expected += p * (r + self.gamma * V[next_s])
max_val = max(max_val, expected)
V[s] = max_val
delta = max(delta, abs(v - V[s]))
if delta < theta:
break
return V
def extract_policy(self, V):
policy = np.zeros(self.n_states, dtype=int)
for s in range(self.n_states):
if self.state_to_pos(s) in [self.goal, self.trap]:
continue
best_a = -1
best_val = -float('inf')
for a in range(self.n_actions):
transitions = self.get_transition_prob(s, a)
expected = 0
for next_s, p in transitions.items():
r = self.get_reward(s, a, next_s)
expected += p * (r + self.gamma * V[next_s])
if expected > best_val:
best_val = expected
best_a = a
policy[s] = best_a
return policy
mdp = GridWorldMDP(width=4, height=3)
V = mdp.solve_value_iteration()
policy = mdp.extract_policy(V)
action_names = ['right', 'left', 'down', 'up']
print("Optimal values:")
for r in range(3):
row = []
for c in range(4):
s = r * 4 + c
row.append(f"{V[s]:.2f}")
print(" " + " ".join(row))
print("\nOptimal policy:")
for r in range(3):
row = []
for c in range(4):
s = r * 4 + c
if (r, c) == mdp.goal:
row.append("G")
elif (r, c) == mdp.trap:
row.append("T")
else:
row.append(action_names[policy[s]][0].upper())
print(" " + " ".join(row))3.2 案例二:策略评估的迭代方法
def policy_evaluation_demo():
mdp = GridWorldMDP(width=4, height=3)
random_policy = np.ones((mdp.n_states, mdp.n_actions)) / mdp.n_actions
V = np.zeros(mdp.n_states)
gamma = 0.9
theta = 1e-6
print("Policy evaluation iterations:")
for iteration in range(100):
delta = 0
for s in range(mdp.n_states):
if mdp.state_to_pos(s) in [mdp.goal, mdp.trap]:
continue
v = V[s]
new_v = 0
for a in range(mdp.n_actions):
transitions = mdp.get_transition_prob(s, a)
expected = 0
for next_s, p in transitions.items():
r = mdp.get_reward(s, a, next_s)
expected += p * (r + gamma * V[next_s])
new_v += random_policy[s][a] * expected
V[s] = new_v
delta = max(delta, abs(v - V[s]))
if iteration < 5 or iteration % 20 == 0:
print(f" Iteration {iteration}: max change = {delta:.6f}")
if delta < theta:
print(f" Converged at iteration {iteration}")
break
print("\nFinal values under random policy:")
for r in range(3):
row = []
for c in range(4):
s = r * 4 + c
row.append(f"{V[s]:.3f}")
print(" " + " ".join(row))
policy_evaluation_demo()3.3 案例三:策略迭代的完整流程
def policy_iteration(mdp, gamma=0.9, theta=1e-6):
n_states = mdp.n_states
n_actions = mdp.n_actions
policy = np.zeros(n_states, dtype=int)
iteration = 0
while True:
V = np.zeros(n_states)
while True:
delta = 0
for s in range(n_states):
if mdp.state_to_pos(s) in [mdp.goal, mdp.trap]:
continue
v = V[s]
a = policy[s]
transitions = mdp.get_transition_prob(s, a)
expected = 0
for next_s, p in transitions.items():
r = mdp.get_reward(s, a, next_s)
expected += p * (r + gamma * V[next_s])
V[s] = expected
delta = max(delta, abs(v - V[s]))
if delta < theta:
break
policy_stable = True
for s in range(n_states):
if mdp.state_to_pos(s) in [mdp.goal, mdp.trap]:
continue
old_action = policy[s]
best_action = -1
best_value = -float('inf')
for a in range(n_actions):
transitions = mdp.get_transition_prob(s, a)
expected = 0
for next_s, p in transitions.items():
r = mdp.get_reward(s, a, next_s)
expected += p * (r + gamma * V[next_s])
if expected > best_value:
best_value = expected
best_action = a
policy[s] = best_action
if old_action != best_action:
policy_stable = False
iteration += 1
print(f"Policy iteration {iteration}: stable={policy_stable}")
if policy_stable:
break
return V, policy
mdp = GridWorldMDP(width=4, height=3)
V, policy = policy_iteration(mdp)
print("Policy iteration complete!")
action_names = ['right', 'left', 'down', 'up']
for r in range(3):
row = []
for c in range(4):
s = r * 4 + c
if (r, c) == mdp.goal:
row.append("G")
elif (r, c) == mdp.trap:
row.append("T")
else:
row.append(action_names[policy[s]][0].upper())
print(" " + " ".join(row))四、本章小结
第3章建立了强化学习的数学基础——有限马尔可夫决策过程,核心要点如下:
马尔可夫性质:未来只依赖于当前状态。这个假设使得问题可以被分解为局部的递归结构。
MDP 的五元组定义:$(\mathcal{S}, \mathcal{A}, P, R, \gamma)$ 完整描述了强化学习环境。
策略与价值函数:策略决定行为,价值函数评估行为的好坏。状态价值函数和动作价值函数可以相互转换。
贝尔曼方程:价值函数的递归分解,是几乎所有强化学习算法的理论基础。
最优策略:通过贝尔曼最优方程定义,可以通过值迭代和策略迭代求解。
算法框架:策略评估(给定策略求价值)和策略改进(给定价值改进策略)构成了策略迭代的核心循环。
五、思考与练习
思考题
考虑一个扑克游戏,当前手牌相同但对手的历史行为不同。这个环境是否满足马尔可夫性?如何设计状态表示来恢复马尔可夫性?
证明:对于任意策略 $\pi$,其价值函数 $v_\pi$ 是贝尔曼方程的唯一解。
值迭代和策略迭代哪个收敛更快?在什么场景下各有优势?
如果折扣因子 $\gamma = 1$,贝尔曼方程是否仍然有唯一解?为什么?
考虑一个有 1000 个状态和 10 个动作的 MDP,值迭代的时间复杂度是多少?如何加速?
编程练习
实现策略迭代:在第3.1节的 Grid World 上实现完整的策略迭代算法,与值迭代对比收敛速度。
随机环境测试:将 Grid World 的动作改为有 10% 概率执行随机方向,观察最优策略的变化。
不同折扣因子的影响:在 Grid World 上分别设置 $\gamma = 0.5, 0.9, 0.99$,比较最优策略的差异。