介绍强化学习中动态规划的核心概念，包括马尔可夫决策过程、价值函数及贝尔曼方程。详细阐述了策略评估、策略迭代和价值迭代算法的数学原理与步骤，并通过 3x3 网格世界的 Python 代码示例演示了具体实现与收敛验证。
一个马尔可夫决策过程由五元组构成：$\text{MDP} = (S, A, P, R, \gamma)$，其中：
这里的状态转移概率是一个比较反直觉的点。理论上来说从一个状态出发做出一个动作，下一时刻的状态似乎是确定的，然而在 MDP 中通常涉及概率分布。
策略 $\pi$ 是状态到动作概率分布的映射： $$\pi(a \mid s) = \text{Pr}(A_t=a \mid S_t=s)$$
从时间 $t$ 开始的累积折扣奖励： $$G_t = R_{t+1} + \gamma R_{t+2} + \gamma^2 R_{t+3} + \cdots = \sum_{k=0}^{\infty} \gamma^k R_{t+k+1}$$
在策略 $\pi$ 下，状态 $s$ 的价值是期望回报： $$V^\pi(s) = \mathbb{E}_\pi(G_t \mid S_t = s)$$
在状态 $s$ 执行动作 $a$ 后遵循策略 $\pi$ 的期望回报： $$Q^\pi(s,a) = \mathbb{E}_\pi(G_t \mid S_t = s, A_t = a)$$
$$V^\pi(s) = \sum_{a \in A} \pi(a \mid s) \sum_{s' \in S} P(s' \mid s,a) \left[ R(s,a,s') + \gamma V^\pi(s') \right]$$
推导： 从定义出发：$V^\pi(s) = \mathbb{E}\pi(G_t \mid S_t = s)$。 展开 $G_t = R{t+1} + \gamma G_{t+1}$，代入得： $$V^\pi(s) = \mathbb{E}\pi(R{t+1} + \gamma G_{t+1} \mid S_t = s)$$ 利用期望线性性拆分： $$V^\pi(s) = \mathbb{E}\pi(R{t+1} \mid S_t = s) + \gamma \mathbb{E}\pi(G{t+1} \mid S_t = s)$$ 第一项为期望即时奖励： $$\mathbb{E}\pi(R{t+1} \mid S_t = s) = \sum_{a} \pi(a \mid s) \sum_{s'} P(s' \mid s,a) R(s,a,s')$$ 第二项利用马尔可夫性转换为下一状态价值： $$\mathbb{E}\pi\left[G{t+1} \mid S_t = s\right] = \sum_{a} \pi(a \mid s) \sum_{s'} P(s' \mid s,a) V^\pi(s') = \mathbb{E}\left[V^\pi(S_{t+1}) \mid S_t = s\right]$$ 合并即得贝尔曼期望方程。
贝尔曼期望方程用一句话来讲那就是描述了不同状态的状态价值之间的关系。
这里对贝尔曼期望方程做一个直观理解：贝尔曼期望方程即状态价值函数是某个状态下的期望回报。第一项表达的是从这个状态出发能够拿到的即时回报，具体而言，从这个状态出发，在不同的策略下将会执行不同的动作，执行不同的动作会到达不同的状态（即使从相同的状态出发，执行相同的动作，到达的状态也可能会不同）。结合期望的定义，我们可以很容易理解即时回报就是考虑在这个状态下能够做出所有动作，然后考虑在这个动作下会到达的状态及其对应的奖励，相乘起来就得到了即时回报。 第二项则也同样好理解，从某个状态出发，考虑在这个状态下能够做出所有动作，然后考虑在这个动作下会到达的状态及其在这个状态下的状态价值函数，这个就是未来回报。
注：上面的贝尔曼期望方程，其实是贝尔曼方程的另一种表达，在学习 TD learning 时会看到这种表示形式，大家一般看见的贝尔曼方程的形式： $$V^\pi(s) = \sum_{a \in A} \pi(a \mid s) \left[ \sum_{r \in R} p(r \mid s,a) r + \gamma \sum_{s' \in S} P(s' \mid s,a) V^\pi(s') \right]$$ 二者其实是等价的。
$$Q^\pi(s,a) = \sum_{s' \in S} P(s' \mid s,a) \left[ R(s,a,s') + \gamma \sum_{a' \in A} \pi(a' \mid s') Q^\pi(s',a') \right]$$
动作价值函数与状态价值函数的关系： $$V^\pi(s) = \sum_{a \in A} \pi(a \mid s) \cdot Q^\pi(s,a)$$
问题：给定策略 $\pi$，计算其对应的状态价值函数 $V^\pi$。 方法：迭代应用贝尔曼期望方程。 算法步骤：
收敛性：贝尔曼期望算子是 $\gamma$-收缩映射，必收敛到唯一不动点 $V^\pi$。
问题：给定价值函数 $V^\pi$，找到更优的策略 $\pi'$。 核心定理（策略改进定理）： 对于任意两个确定性策略 $\pi$ 和 $\pi'$，若对所有状态 $s$ 满足： $$Q^\pi(s, \pi'(s)) \ge V^\pi(s)$$ 则对所有状态 $s$ 有： $$V^{\pi'}(s) \ge V^\pi(s)$$
算法流程：
即：对每个状态，选择能够最大化动作价值的动作。 算法流程：先定策略，再评估，再改进。 策略评估部分在数学上求解需要求解方程组（或者矩阵计算求逆），在工程上的大规模问题这种方式不方便求解，在工程上一般使用迭代算法来求解。
若 $\pi' = \pi$，则收敛到最优策略 $\pi^* = \pi$；否则令 $\pi = \pi'$，回到「策略评估」继续迭代。
核心思想：直接迭代最优价值函数，将策略评估的迭代次数简化为 1 次。 算法步骤：
算法流程：直接算出最优值，再提取策略。
环境设定：
以角落状态 $s_{00}$（编号 0）为例，随机策略下每个动作概率 0.25： $$\begin{align*} V^\pi(s_{00}) &= 0.25 \times [-1 + 0.9V^\pi(s_{00})] + 0.25 \times [-1 + 0.9V^\pi(s_{10})] \ &+ 0.25 \times [-1 + 0.9V^\pi(s_{01})] + 0.25 \times [-1 + 0.9V^\pi(s_{00})] \end{align*}$$ 简化得： $$V^\pi(s_{00}) = \frac{-1 + 0.225V^\pi(s_{10}) + 0.225V^\pi(s_{01})}{0.55}$$
初始 $V_0(s_{11}) = 0$：
import numpy as np
class GridWorldMDP:
"""3×3 网格世界 MDP 实现"""
def __init__(self, size=3, goal_reward=10, step_cost=-1, gamma=0.9):
self.size = size
self.n_states = size * size
self.n_actions = 4 # 0:上，1:下，2:左，3:右
self.goal_reward = goal_reward
self.step_cost = step_cost
self.gamma = gamma
# 初始化转移概率 P(s'|s,a) 和奖励函数 R(s,a,s')
self.P = np.zeros((self.n_states, self.n_actions, self.n_states))
self.R = np.zeros((self.n_states, self.n_actions, self.n_states))
self._build_transitions()
def _build_transitions(self):
"""构建确定性转移和奖励函数"""
goal_state = self.n_states - 1 # 右下角为目标
for s in range(self.n_states):
row, col = divmod(s, self.size)
for a in range(self.n_actions):
# 计算下一个状态（撞墙时不变）
if a == 0:
next_row, next_col = max(row-1, 0), col
elif a == 1:
next_row, next_col = min(row+1, self.size-1), col
elif a == 2:
next_row, next_col = row, max(col-1, 0)
else:
next_row, next_col = row, min(col+1, self.size-1)
s_next = next_row * self.size + next_col
self.P[s, a, s_next] = 1.0 # 确定性转移
# 设置奖励
if s == goal_state:
self.R[s, a, s_next] = 0
elif s_next == goal_state:
self.R[s, a, s_next] = self.goal_reward
else:
self.R[s, a, s_next] = self.step_cost
def policy_evaluation(self, policy, max_iter=1000, theta=1e-6):
"""策略评估：迭代求解 V^π"""
V = np.zeros(self.n_states)
for i in range(max_iter):
delta = 0
V_new = np.zeros(self.n_states)
for s in range(self.n_states):
v = 0
for a in range(self.n_actions):
pi_a_s = policy[s, a]
expected_value = 0
for s_next in range(self.n_states):
p = self.P[s, a, s_next]
r = self.R[s, a, s_next]
expected_value += p * (r + self.gamma * V[s_next])
v += pi_a_s * expected_value
V_new[s] = v
delta = max(delta, abs(v - V[s]))
V = V_new
if delta < theta:
print(f"策略评估在{i+1}次迭代后收敛")
break
return V
def policy_improvement(self, V):
"""策略改进：贪心策略更新"""
new_policy = np.zeros((self.n_states, self.n_actions))
for s in range(self.n_states):
# 计算所有动作的 Q 值
q_values = np.zeros(self.n_actions)
for a in range(self.n_actions):
q = 0
for s_next in range(self.n_states):
p = self.P[s, a, s_next]
r = self.R[s, a, s_next]
q += p * (r + self.gamma * V[s_next])
q_values[a] = q
# 选择最优动作（多最优动作时平均分配概率）
best_actions = np.where(q_values == np.max(q_values))[0]
for a in best_actions:
new_policy[s, a] = 1.0 / len(best_actions)
return new_policy
def policy_iteration(self, max_iter=100):
"""策略迭代算法"""
# 初始化随机策略
policy = np.ones((self.n_states, self.n_actions)) / self.n_actions
for i in range(max_iter):
print(f"\n=== 策略迭代第{i+1}轮 ===")
# 策略评估
V = self.policy_evaluation(policy)
# 策略改进
new_policy = self.policy_improvement(V)
# 检查收敛
if np.allclose(policy, new_policy):
print(f"策略在{i+1}次迭代后收敛到最优")
return V, new_policy
policy = new_policy
print("达到最大迭代次数")
return V, policy
def value_iteration(self, max_iter=1000, theta=1e-6):
"""价值迭代算法"""
V = np.zeros(self.n_states)
for i in range(max_iter):
delta = 0
V_new = np.zeros(self.n_states)
for s in range(self.n_states):
# 计算所有动作的期望价值
action_values = np.zeros(self.n_actions)
for a in range(self.n_actions):
q = 0
for s_next in range(self.n_states):
p = self.P[s, a, s_next]
r = self.R[s, a, s_next]
q += p * (r + self.gamma * V[s_next])
action_values[a] = q
# 取最大值
V_new[s] = np.max(action_values)
delta = max(delta, abs(V_new[s] - V[s]))
V = V_new
if delta < theta:
print(f"价值迭代在{i+1}次迭代后收敛")
break
# 提取最优策略
optimal_policy = self.policy_improvement(V)
return V, optimal_policy
# 运行示例
if __name__ == "__main__":
print("="*60)
print("3×3 网格世界 MDP 示例")
print("="*60)
# 创建 MDP 实例
mdp = GridWorldMDP(size=3, goal_reward=10, step_cost=-1, gamma=0.9)
# 1. 策略迭代
print("\n1. 策略迭代:")
V_pi, policy_pi = mdp.policy_iteration(max_iter=10)
print(f"最优价值函数（策略迭代）:\n{V_pi.reshape(3,3).round(4)}")
# 2. 价值迭代
print("\n" + "="*60)
print("\n2. 价值迭代:")
V_vi, policy_vi = mdp.value_iteration()
print(f"最优价值函数（价值迭代）:\n{V_vi.reshape(3,3).round(4)}")
# 比较结果
print("\n" + "="*60)
print("策略迭代 vs 价值迭代:")
print(f"价值函数最大差异：{np.max(np.abs(V_pi - V_vi)):.6f}")
print(f"策略最大差异：{np.max(np.abs(policy_pi - policy_vi)):.6f}")
# 验证贝尔曼最优方程
print("\n" + "="*60)
print("验证贝尔曼最优方程:")
for s in range(mdp.n_states):
lhs = V_vi[s]
# 计算右侧
action_values = np.zeros(mdp.n_actions)
for a in range(mdp.n_actions):
q = 0
for s_next in range(mdp.n_states):
p = mdp.P[s, a, s_next]
r = mdp.R[s, a, s_next]
q += p * (r + mdp.gamma * V_vi[s_next])
action_values[a] = q
rhs = np.max(action_values)
print(f"状态{s}: LHS={lhs:.4f}, RHS={rhs:.4f}, 差异={abs(lhs-rhs):.6f}")
from Policy_iteration import GridWorldMDP
import numpy as np
class ValueIterationMDP(GridWorldMDP):
def value_iteration(self, theta=1e-6):
V = np.zeros(self.n_states)
print("="*60)
print("价值迭代算法开始")
print("="*60)
print(f"初始价值函数:\n{V.reshape(self.size, self.size)}\n")
for iteration in range(1000):
delta = 0
V_new = np.zeros(self.n_states)
for s in range(self.n_states):
action_value = []
for a in range(self.n_actions):
q = 0
# 计算每个动作 a 的完整 Q 值（求和所有转移）
for next_state in np.where(self.P[s, a]>0)[0]:
q += self.P[s, a, next_state]*(self.R[s, a, next_state]+ self.gamma * V[next_state])
action_value.append(q)
V_new[s] = max(action_value) if action_value else 0.0
delta = max(delta, abs(V[s]- V_new[s]))
V = V_new
if iteration % 5 == 0:
print(f"第{iteration +1}轮迭代 | 本轮最大误差：{delta:.6f}")
print(f"当前价值函数:\n{V.reshape(self.size, self.size)}\n")
# 收敛判断
if delta < theta:
print(f"\n✅ 价值迭代在第{iteration +1}次迭代收敛！")
print(f"最终收敛误差：{delta:.8f}")
break
# 提取最优策略
policy = self._extract_policy(V)
return V, policy
def _extract_policy(self, V):
policy = np.zeros((self.n_states, self.n_actions))
for s in range(self.n_states):
q_value = []
for a in range(self.n_actions):
q = 0
for next_state in np.where(self.P[s, a]>0)[0]:
q += self.P[s, a, next_state]*(self.R[s, a, next_state]+ self.gamma * V[next_state])
q_value.append(q)
max_q = max(q_value)
# 筛选所有最优动作（处理多最优动作均分概率）
best_action = [a for a, q in enumerate(q_value) if abs(q - max_q)<1e-6]
for a in best_action:
policy[s, a] = 1.0/len(best_action)
return policy
if __name__ == "__main__":
mdp_vi = ValueIterationMDP()
print("价值迭代算法完整演示")
print("="*60)
# 运行价值迭代
V_vi, policy_vi = mdp_vi.value_iteration()
# 1. 展示最终收敛的最优价值函数
print("\n" + "="*60)
print("最终收敛 - 最优价值函数")
print("="*60)
V_grid = V_vi.reshape(mdp_vi.size, mdp_vi.size)
print(V_grid)
# 2. 展示最终收敛的最优策略矩阵
print("\n" + "="*60)
print("最终收敛 - 最优策略矩阵 (状态数×动作数)")
print("="*60)
print(f"策略矩阵形状：{policy_vi.shape}")
print("【矩阵含义：行=状态，列=动作，值=选择该动作的概率】")
print(policy_vi)
# 3. 策略可视化
print("\n" + "="*60)
print(" 最优策略 - 网格可视化")
print("="*60)
# 动作映射（根据你的 GridWorldMDP 动作定义，通用上下左右映射）
action_map = {0:"↑",1:"↓",2:"←",3:"→"}
policy_grid = np.zeros((mdp_vi.size, mdp_vi.size), dtype=object)
for i in range(mdp_vi.size):
for j in range(mdp_vi.size):
s = i * mdp_vi.size + j # 网格坐标转状态编号
best_acts = np.where(policy_vi[s]>0)[0] # 筛选最优动作
act_str = "".join([action_map[a] for a in best_acts])
policy_grid[i, j] = act_str if act_str else "终端"
print("【每个网格值：当前位置的最优动作，多动作表示等概率选择】")
print(policy_grid)
补充说明：
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online