3强化学习:导论(第2版)

有限马尔可夫决策过程

MDP框架

MDP状态转移贝尔曼方程

🎯 AI 时代的学习建议:不要只读理论。理解这一章的设计动机(为什么需要这个概念?)和边界条件(什么情况下会失效?)。当 AI 生成的代码涉及这些概念时,你能判断它是否用对了场景。AI 能帮你总结内容,但无法替代你亲手阅读、思考、建立直觉的过程。

第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]$:折扣因子
python
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$ 的概率。

python
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]$$

python
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]$$

这个方程的深刻之处在于:它将一个无限步的问题分解为一步决策 + 剩余子问题的递归结构。这使得我们可以用迭代法求解价值函数。

python
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)$$

python
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 马尔可夫性质的实际意义

马尔可夫性质是一个强假设。现实中的很多问题并不严格满足马尔可夫性——围棋的打劫规则需要记住之前的棋局,自动驾驶需要了解之前的交通状况。但在实践中,我们可以通过状态表示的设计来恢复马尔可夫性:将历史信息编码到状态中。

python
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 建模

python
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 案例二:策略评估的迭代方法

python
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 案例三:策略迭代的完整流程

python
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$,比较最优策略的差异。

3/11有限马尔
0%