diff --git a/Notebooks/C1.ipynb b/Notebooks/C1.ipynb new file mode 100644 index 0000000..b5952a3 --- /dev/null +++ b/Notebooks/C1.ipynb @@ -0,0 +1,285 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "f6b663c9", + "metadata": {}, + "source": [ + "# 第 1 章:强化学习基本概念 (Basic Concepts)\n", + "本笔记本旨在复现 3x3 网格世界中的基础马尔可夫决策过程 (MDP) 元素。" + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "id": "6771d846", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "状态空间大小: 9\n", + "动作空间大小: 5\n" + ] + } + ], + "source": [ + "# 定义状态空间 (9个格子)\n", + "states = [\"s1\", \"s2\", \"s3\", \"s4\", \"s5\", \"s6\", \"s7\", \"s8\", \"s9\"]\n", + "\n", + "# 定义动作空间 (5种动作)\n", + "actions = [\"a1\", \"a2\", \"a3\", \"a4\", \"a5\"] # 分别代表:上、右、下、左、原地\n", + "\n", + "print(f\"状态空间大小: {len(states)}\")\n", + "print(f\"动作空间大小: {len(actions)}\")" + ] + }, + { + "cell_type": "code", + "execution_count": 4, + "id": "20ed6dce", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "在 s1 采取动作 a2,下一个状态是: s2\n" + ] + } + ], + "source": [ + "# 构建状态转移字典,格式:(当前状态, 动作): 下一个状态\n", + "transitions = {}\n", + "\n", + "# s1 的转移\n", + "transitions[(\"s1\", \"a1\")] = \"s1\" # 向上撞墙,反弹回 s1\n", + "transitions[(\"s1\", \"a2\")] = \"s2\" # 向右进入 s2\n", + "transitions[(\"s1\", \"a3\")] = \"s4\" # 向下进入 s4\n", + "transitions[(\"s1\", \"a4\")] = \"s1\" # 向左撞墙,反弹回 s1\n", + "transitions[(\"s1\", \"a5\")] = \"s1\" # 原地不动\n", + "\n", + "# s2 的转移\n", + "transitions[(\"s2\", \"a1\")] = \"s2\" # 向上撞墙,反弹回 s2\n", + "transitions[(\"s2\", \"a2\")] = \"s3\" # 向右进入 s3\n", + "transitions[(\"s2\", \"a3\")] = \"s5\" # 向下进入 s5\n", + "transitions[(\"s2\", \"a4\")] = \"s1\" # 向左进入 s1\n", + "transitions[(\"s2\", \"a5\")] = \"s2\" # 原地不动\n", + "\n", + "# s3 的转移\n", + "transitions[(\"s3\", \"a1\")] = \"s3\" # 向上撞墙\n", + "transitions[(\"s3\", \"a2\")] = \"s3\" # 向右撞墙\n", + "transitions[(\"s3\", \"a3\")] = \"s6\" # 向下进入 s6\n", + "transitions[(\"s3\", \"a4\")] = \"s2\" # 向左进入 s2\n", + "transitions[(\"s3\", \"a5\")] = \"s3\" # 原地不动\n", + "\n", + "# s4 的转移\n", + "transitions[(\"s4\", \"a1\")] = \"s1\" # 向上进入 s1\n", + "transitions[(\"s4\", \"a2\")] = \"s5\" # 向右进入 s5\n", + "transitions[(\"s4\", \"a3\")] = \"s7\" # 向下进入 s7\n", + "transitions[(\"s4\", \"a4\")] = \"s4\" # 向左撞墙\n", + "transitions[(\"s4\", \"a5\")] = \"s4\" # 原地不动\n", + "\n", + "# s5 的转移\n", + "transitions[(\"s5\", \"a1\")] = \"s2\" # 向上进入 s2\n", + "transitions[(\"s5\", \"a2\")] = \"s6\" # 向右进入 s6\n", + "transitions[(\"s5\", \"a3\")] = \"s8\" # 向下进入 s8\n", + "transitions[(\"s5\", \"a4\")] = \"s4\" # 向左进入 s4\n", + "transitions[(\"s5\", \"a5\")] = \"s5\" # 原地不动\n", + "\n", + "# s6 的转移\n", + "transitions[(\"s6\", \"a1\")] = \"s3\" # 向上进入 s3\n", + "transitions[(\"s6\", \"a2\")] = \"s6\" # 向右撞墙\n", + "transitions[(\"s6\", \"a3\")] = \"s9\" # 向下进入 s9\n", + "transitions[(\"s6\", \"a4\")] = \"s5\" # 向左进入 s5\n", + "transitions[(\"s6\", \"a5\")] = \"s6\" # 原地不动\n", + "\n", + "# s7 的转移\n", + "transitions[(\"s7\", \"a1\")] = \"s4\" # 向上进入 s4\n", + "transitions[(\"s7\", \"a2\")] = \"s8\" # 向右进入 s8\n", + "transitions[(\"s7\", \"a3\")] = \"s7\" # 向下撞墙\n", + "transitions[(\"s7\", \"a4\")] = \"s7\" # 向左撞墙\n", + "transitions[(\"s7\", \"a5\")] = \"s7\" # 原地不动\n", + "\n", + "# s8 的转移\n", + "transitions[(\"s8\", \"a1\")] = \"s5\" # 向上进入 s5\n", + "transitions[(\"s8\", \"a2\")] = \"s9\" # 向右进入 s9\n", + "transitions[(\"s8\", \"a3\")] = \"s8\" # 向下撞墙\n", + "transitions[(\"s8\", \"a4\")] = \"s7\" # 向左进入 s7\n", + "transitions[(\"s8\", \"a5\")] = \"s8\" # 原地不动\n", + "\n", + "# s9 的转移\n", + "transitions[(\"s9\", \"a1\")] = \"s6\" # 向上进入 s6\n", + "transitions[(\"s9\", \"a2\")] = \"s9\" # 向右撞墙\n", + "transitions[(\"s9\", \"a3\")] = \"s9\" # 向下撞墙\n", + "transitions[(\"s9\", \"a4\")] = \"s8\" # 向左进入 s8\n", + "transitions[(\"s9\", \"a5\")] = \"s9\" # 原地不动\n", + "\n", + "def get_next_state(state, action):\n", + " return transitions.get((state, action), state) # 如果没定义,默认原地不动\n", + "\n", + "print(f\"在 s1 采取动作 a2,下一个状态是: {get_next_state('s1', 'a2')}\")" + ] + }, + { + "cell_type": "code", + "execution_count": 5, + "id": "788f4de0", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "s1撞墙奖励: -1\n", + "进入目标s9奖励: 1\n" + ] + } + ], + "source": [ + "# 定义奖励规则\n", + "def get_reward(state, action, next_state):\n", + " # 如果撞墙(当前状态和下一个状态一样,且不是原地动作)\n", + " if state == next_state and action != \"a5\":\n", + " return -1\n", + " # 如果进入目标状态 s9\n", + " elif next_state == \"s9\":\n", + " return 1\n", + " # 如果进入禁区 s6 或 s7 (根据书中图 1.2)\n", + " elif next_state in [\"s6\", \"s7\"]:\n", + " return -1\n", + " # 其他情况\n", + " else:\n", + " return 0\n", + "\n", + "# 测试一下\n", + "test_r1 = get_reward(\"s1\", \"a1\", \"s1\")\n", + "test_r2 = get_reward(\"s8\", \"a2\", \"s9\")\n", + "print(f\"s1撞墙奖励: {test_r1}\") # 应该输出 -1\n", + "print(f\"进入目标s9奖励: {test_r2}\") # 应该输出 1" + ] + }, + { + "cell_type": "code", + "execution_count": 8, + "id": "c64e665e", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "--- 运行轨迹并计算回报 ---\n", + "Step 0: s1 --a2--> s2, Reward: 0\n", + "Step 1: s2 --a3--> s5, Reward: 0\n", + "Step 2: s5 --a3--> s8, Reward: 0\n", + "Step 3: s8 --a2--> s9, Reward: 1\n", + "最终折扣回报: 0.7290000000000001\n" + ] + } + ], + "source": [ + "# 定义一条书中的轨迹 (图 1.6 左侧的好策略轨迹)\n", + "# 格式:[(状态, 动作), ...]\n", + "good_trajectory = [\n", + " (\"s1\", \"a2\"), (\"s2\", \"a3\"), (\"s5\", \"a3\"), (\"s8\", \"a2\")\n", + "]\n", + "\n", + "# 计算回报的函数\n", + "def calculate_discounted_return(trajectory, gamma=0.9):\n", + " total_return = 0.0\n", + " \n", + " for t, (state, action) in enumerate(trajectory):\n", + " next_state = get_next_state(state, action)\n", + " reward = get_reward(state, action, next_state)\n", + " \n", + " # 计算折扣回报\n", + " total_return += (gamma ** t) * reward\n", + " print(f\"Step {t}: {state} --{action}--> {next_state}, Reward: {reward}\")\n", + " \n", + " return total_return\n", + "\n", + "print(\"--- 运行轨迹并计算回报 ---\")\n", + "final_return = calculate_discounted_return(good_trajectory, gamma=0.9)\n", + "print(f\"最终折扣回报: {final_return}\")" + ] + }, + { + "cell_type": "markdown", + "id": "5de54db1", + "metadata": {}, + "source": [ + "接下来来有一些随机策略的case" + ] + }, + { + "cell_type": "code", + "execution_count": 9, + "id": "6b05c85f", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "在状态 s1 进行 10 次动作选择测试:\n", + "[np.str_('a2'), np.str_('a2'), np.str_('a2'), np.str_('a2'), np.str_('a3'), np.str_('a3'), np.str_('a3'), np.str_('a3'), np.str_('a2'), np.str_('a3')]\n", + "向右(a2)的次数: 5, 向下(a3)的次数: 5\n" + ] + } + ], + "source": [ + "import numpy as np\n", + "\n", + "# 定义一个随机策略字典\n", + "# 格式:状态: {动作: 概率, 动作: 概率}\n", + "stochastic_policy = {\n", + " \"s1\": {\"a2\": 0.5, \"a3\": 0.5}, # 图 1.5 中的设定\n", + " \"s2\": {\"a3\": 1.0}, # 其他状态假设是确定性的 100%\n", + " \"s5\": {\"a3\": 1.0},\n", + " \"s8\": {\"a2\": 1.0}\n", + "}\n", + "\n", + "def choose_action(state, policy):\n", + " # 如果策略里有这个状态的动作概率分布\n", + " if state in policy:\n", + " action_probs = policy[state]\n", + " actions = list(action_probs.keys())\n", + " probs = list(action_probs.values())\n", + " # 根据概率分布随机选择一个动作\n", + " chosen_action = np.random.choice(actions, p=probs)\n", + " return chosen_action\n", + " else:\n", + " return \"a5\" # 默认原地不动\n", + "\n", + "# 测试随机策略:让智能体在 s1 连续做 10 次决定\n", + "print(\"在状态 s1 进行 10 次动作选择测试:\")\n", + "choices = [choose_action(\"s1\", stochastic_policy) for _ in range(10)]\n", + "print(choices)\n", + "print(f\"向右(a2)的次数: {choices.count('a2')}, 向下(a3)的次数: {choices.count('a3')}\")" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "GymRL", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.13.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/Notebooks/C2.ipynb b/Notebooks/C2.ipynb new file mode 100644 index 0000000..2a88605 --- /dev/null +++ b/Notebooks/C2.ipynb @@ -0,0 +1,184 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "31a4e03b", + "metadata": {}, + "source": [ + "# 第 2 章:状态价值与贝尔曼方程\n", + "贝尔曼方程描述了所有状态价值之间的关系。\n", + "我们将使用**矩阵-向量形式** 来表示贝尔曼方程: $$v_{\\pi} = r_{\\pi} + \\gamma P_{\\pi}v_{\\pi}$$\n", + "以及使用**迭代法** 来求解它: $$v_{k+1} = r_{\\pi} + \\gamma P_{\\pi}v_{k}$$" + ] + }, + { + "cell_type": "code", + "execution_count": 4, + "id": "8e67777b", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "--- 贝尔曼方程迭代求解过程 ---\n", + "第 1 次迭代: v(s1)=-0.5000, v(s2)=1.0000, v(s3)=1.0000, v(s4)=1.0000\n", + "第 2 次迭代: v(s1)=0.4000, v(s2)=1.9000, v(s3)=1.9000, v(s4)=1.9000\n", + "第 3 次迭代: v(s1)=1.2100, v(s2)=2.7100, v(s3)=2.7100, v(s4)=2.7100\n", + "第 4 次迭代: v(s1)=1.9390, v(s2)=3.4390, v(s3)=3.4390, v(s4)=3.4390\n", + "第 5 次迭代: v(s1)=2.5951, v(s2)=4.0951, v(s3)=4.0951, v(s4)=4.0951\n", + "第 100 次迭代: v(s1)=8.4997, v(s2)=9.9997, v(s3)=9.9997, v(s4)=9.9997\n", + "\n", + "最终收敛的状态价值: [ 8.5 10. 10. 10. ]\n" + ] + } + ], + "source": [ + "import numpy as np\n", + "\n", + "# 1. 设定折扣因子\n", + "gamma = 0.9\n", + "\n", + "# 2. 定义书本图 2.5 中的期望奖励向量 r_pi (Reward vector)\n", + "# s1 的期望奖励是 0.5*0 + 0.5*(-1) = -0.5\n", + "r_pi = np.array([-0.5, 1.0, 1.0, 1.0])\n", + "\n", + "# 3. 定义状态转移矩阵 P_pi (Transition matrix)\n", + "# 每一行代表 s1, s2, s3, s4; 每一列代表转移到 s1, s2, s3, s4 的概率\n", + "P_pi = np.array([\n", + " [0.0, 0.5, 0.5, 0.0], # s1 有 0.5 概率到 s2, 0.5 概率到 s3\n", + " [0.0, 0.0, 0.0, 1.0], # s2 有 1.0 概率到 s4\n", + " [0.0, 0.0, 0.0, 1.0], # s3 有 1.0 概率到 s4\n", + " [0.0, 0.0, 0.0, 1.0] # s4 有 1.0 概率停留在 s4\n", + "])\n", + "\n", + "# 4. 迭代法求解状态价值 (Iterative solution)\n", + "v_values = np.zeros(4) # 初始时,假设所有状态价值为 0\n", + "iterations = 100\n", + "\n", + "print(\"--- 贝尔曼方程迭代求解过程 ---\")\n", + "for k in range(iterations):\n", + " # 核心公式:v_{k+1} = r_pi + gamma * P_pi * v_k\n", + " v_next = r_pi + gamma * P_pi.dot(v_values)\n", + " \n", + " if k < 5 or k == iterations - 1:\n", + " print(f\"第 {k+1} 次迭代: v(s1)={v_next[0]:.4f}, v(s2)={v_next[1]:.4f}, v(s3)={v_next[2]:.4f}, v(s4)={v_next[3]:.4f}\")\n", + " \n", + " v_values = v_next\n", + "\n", + "print(\"\\n最终收敛的状态价值:\", np.round(v_values, 2))\n", + "# 你可以将这个输出结果与书中 2.5 节手算的 8.5, 10, 10, 10 进行对比!" + ] + }, + { + "cell_type": "markdown", + "id": "2086b1ab", + "metadata": {}, + "source": [ + "# 2.8 从状态价值到动作价值 (Action Value)\n", + "动作价值 $q_\\pi(s,a)$ 评估的是在特定状态下采取特定动作的好坏。\n", + "\n", + "它由两部分组成:\n", + "1. **即时奖励的期望**:采取动作 $a$ 后立刻得到的奖励。\n", + "2. **未来奖励的期望**:进入下一个状态 $s'$ 后,未来的状态价值 $\\gamma v_\\pi(s')$。\n", + "\n", + "数学公式为:\n", + "$$q_{\\pi}(s,a) = \\sum_{r\\in\\mathcal{R}}p(r|s,a)r + \\gamma\\sum_{s'\\in\\mathcal{S}}p(s'|s,a)v_{\\pi}(s')$$\n", + "\n", + "状态价值其实就是所有可能动作价值的加权平均:\n", + "$$v_{\\pi}(s) = \\sum_{a\\in\\mathcal{A}}\\pi(a|s)q_{\\pi}(s,a)$$" + ] + }, + { + "cell_type": "code", + "execution_count": 5, + "id": "a3e76dd6", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "--- 计算在状态 s1 下,所有可能动作的价值 (Q-values) ---\n", + "策略内的动作: q(s1, a2向右) = -1 + 0.9 * v(s2) = 8.0\n", + "策略内的动作: q(s1, a3向下) = 0 + 0.9 * v(s3) = 9.0\n", + "\n", + "策略外的动作: q(s1, a1向上) = -1 + 0.9 * v(s1) = 6.65\n", + "策略外的动作: q(s1, a4向左) = -1 + 0.9 * v(s1) = 6.65\n", + "策略外的动作: q(s1, a5原地) = 0 + 0.9 * v(s1) = 7.65\n", + "\n", + "--- 验证公式:状态价值是动作价值的加权平均 ---\n", + "根据 Q 值反推的 v(s1) = 0.5 * 8.0 + 0.5 * 9.0 = 8.5\n", + "这与我们之前迭代出来的 v(s1) = 8.5 完全一致!\n" + ] + } + ], + "source": [ + "import numpy as np\n", + "\n", + "# 1. 继承之前的已知数据\n", + "gamma = 0.9\n", + "# 假设这是我们在 Cell 8 算出来的最终状态价值\n", + "v_values = {\"s1\": 8.5, \"s2\": 10.0, \"s3\": 10.0, \"s4\": 10.0}\n", + "\n", + "# 2. 定义动作价值计算函数 q(s, a)\n", + "def calculate_q_value(state, action, reward, next_state, gamma, v_values):\n", + " # q(s,a) = 立即奖励 + gamma * 下一个状态的价值\n", + " return reward + gamma * v_values[next_state]\n", + "\n", + "print(\"--- 计算在状态 s1 下,所有可能动作的价值 (Q-values) ---\")\n", + "\n", + "# --- 策略会选择的动作 (向右 a2, 向下 a3) ---\n", + "# a2: 向右进入 s2,得奖励 -1\n", + "q_s1_a2 = calculate_q_value(\"s1\", \"a2\", -1, \"s2\", gamma, v_values)\n", + "print(f\"策略内的动作: q(s1, a2向右) = -1 + 0.9 * v(s2) = {q_s1_a2}\") # 预期为 8.0\n", + "\n", + "# a3: 向下进入 s3,得奖励 0\n", + "q_s1_a3 = calculate_q_value(\"s1\", \"a3\", 0, \"s3\", gamma, v_values)\n", + "print(f\"策略内的动作: q(s1, a3向下) = 0 + 0.9 * v(s3) = {q_s1_a3}\") # 预期为 9.0\n", + "\n", + "\n", + "# --- 策略不会选择的动作 (向上 a1, 向左 a4, 原地 a5) ---\n", + "# a1: 向上撞墙,弹回 s1,得奖励 -1\n", + "q_s1_a1 = calculate_q_value(\"s1\", \"a1\", -1, \"s1\", gamma, v_values)\n", + "print(f\"\\n策略外的动作: q(s1, a1向上) = -1 + 0.9 * v(s1) = {q_s1_a1:.2f}\") # 预期为 6.65\n", + "\n", + "# a4: 向左撞墙,弹回 s1,得奖励 -1\n", + "q_s1_a4 = calculate_q_value(\"s1\", \"a4\", -1, \"s1\", gamma, v_values)\n", + "print(f\"策略外的动作: q(s1, a4向左) = -1 + 0.9 * v(s1) = {q_s1_a4:.2f}\") # 预期为 6.65\n", + "\n", + "# a5: 原地不动,停在 s1,得奖励 0\n", + "q_s1_a5 = calculate_q_value(\"s1\", \"a5\", 0, \"s1\", gamma, v_values)\n", + "print(f\"策略外的动作: q(s1, a5原地) = 0 + 0.9 * v(s1) = {q_s1_a5:.2f}\") # 预期为 7.65\n", + "\n", + "\n", + "print(\"\\n--- 验证公式:状态价值是动作价值的加权平均 ---\")\n", + "# 我们的策略是: 0.5概率选a2, 0.5概率选a3\n", + "v_s1_calculated = 0.5 * q_s1_a2 + 0.5 * q_s1_a3\n", + "print(f\"根据 Q 值反推的 v(s1) = 0.5 * {q_s1_a2} + 0.5 * {q_s1_a3} = {v_s1_calculated}\")\n", + "print(f\"这与我们之前迭代出来的 v(s1) = {v_values['s1']} 完全一致!\")" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "GymRL", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.13.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/Notebooks/C3.ipynb b/Notebooks/C3.ipynb new file mode 100644 index 0000000..4371e73 --- /dev/null +++ b/Notebooks/C3.ipynb @@ -0,0 +1,166 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "745510b1", + "metadata": {}, + "source": [ + "# 第 3 章:最优状态价值与贝尔曼最优方程\n", + "\n", + "## 1. 什么是最优策略?\n", + "如果一个策略 $\\pi^*$ 在所有状态下的价值,都大于或等于其他任何策略的价值,即对于所有 $s \\in \\mathcal{S}$,都有 $v_{\\pi^*}(s) \\ge v_\\pi(s)$,那么 $\\pi^*$ 就是最优策略。\n", + "最优策略的状态价值,被称为**最优状态价值 (Optimal State Value)** $v^*$。\n", + "\n", + "## 2. 贝尔曼最优方程 (BOE)\n", + "之前我们算的是某个固定策略的价值。现在我们要算的是“最优”价值。\n", + "BOE 的核心思想是:在某个状态下的最优价值,等于**所有可能动作中,动作价值 $q(s,a)$ 的最大值**。\n", + "\n", + "数学表达为:\n", + "$$v(s) = \\max_{a \\in \\mathcal{A}} q(s,a)$$" + ] + }, + { + "cell_type": "code", + "execution_count": 2, + "id": "f31bd6cc", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "--- 策略改进演示 ---\n", + "动作 a1 的价值: 6.2\n", + "动作 a2 的价值: 8.0\n", + "动作 a3 的价值: 9.0\n", + "动作 a4 的价值: 6.2\n", + "动作 a5 的价值: 7.2\n", + "\n", + "=> 最佳动作是 'a3',其价值为 9.0。\n", + "结论:只要我们在每个状态都选择 Q 值最大的动作,就能得到一个更好的、甚至是最优的策略!\n" + ] + } + ], + "source": [ + "import numpy as np\n", + "\n", + "# 假设我们在状态 s1 算出了 5 个动作的 Q 值 (参考书中 3.1 节的计算)\n", + "# a1: 向上, a2: 向右, a3: 向下, a4: 向左, a5: 原地不动\n", + "q_values_s1 = {\n", + " \"a1\": 6.2, # 撞墙,-1 + 0.9*v(s1)\n", + " \"a2\": 8.0, # 原策略向右,进禁区,-1 + 0.9*v(s2)\n", + " \"a3\": 9.0, # 向下避开禁区,0 + 0.9*v(s3)\n", + " \"a4\": 6.2, # 撞墙,-1 + 0.9*v(s1)\n", + " \"a5\": 7.2 # 原地不动,0 + 0.9*v(s1)\n", + "}\n", + "\n", + "# 寻找使 Q 值最大的动作 (贪心策略 Greedy Policy)\n", + "best_action = max(q_values_s1, key=q_values_s1.get)\n", + "max_q_value = q_values_s1[best_action]\n", + "\n", + "print(\"--- 策略改进演示 ---\")\n", + "for action, q_val in q_values_s1.items():\n", + " print(f\"动作 {action} 的价值: {q_val}\")\n", + "\n", + "print(f\"\\n=> 最佳动作是 '{best_action}',其价值为 {max_q_value}。\")\n", + "print(\"结论:只要我们在每个状态都选择 Q 值最大的动作,就能得到一个更好的、甚至是最优的策略!\")" + ] + }, + { + "cell_type": "markdown", + "id": "618e235d", + "metadata": {}, + "source": [ + "## 3. 求解 BOE:压缩映射与迭代法\n", + "BOE 看起来很难解,因为它方程两边都有未知数 $v$。但数学上的**压缩映射定理 (Contraction Mapping Theorem)** 告诉我们三个好消息:\n", + "1. **存在性**:BOE 一定有解。\n", + "2. **唯一性**:最优状态价值 $v^*$ 的解是唯一的。\n", + "3. **算法**:可以通过不断迭代 $v_{k+1} = \\max_\\pi(r_\\pi + \\gamma P_\\pi v_k)$ 来逼近最优解,这个过程叫**价值迭代 (Value Iteration)**=。\n", + "\n", + "## 4. 影响最优策略的因素\n", + "* **折扣因子 $\\gamma$ 的魔力**:\n", + " * $\\gamma$ 较大 (如 0.9):智能体目光长远,为了以最快速度拿到目标大奖,甚至愿意冒险穿过有惩罚的禁区。\n", + " * $\\gamma$ 较小 (如 0.5 或更小):智能体变得短视且保守,宁愿绕远路也不敢冒险。\n", + " * 此外,$\\gamma$ 天生就不鼓励“无意义的绕路”,因为绕路会让未来的奖励因折扣变得更小。\n", + "* **奖励的仿射变换不变性**:如果把所有的奖励都乘以一个正数,或者都加上一个固定的常数,算出来的最优策略**完全不会改变**。" + ] + }, + { + "cell_type": "code", + "execution_count": 3, + "id": "41d7c569", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "--- 情景 1:每走一步没有任何惩罚 (即时奖励为 0) ---\n", + "直达路径(A)的回报: 9.0\n", + "绕路路径(B)的回报: 7.290000000000001\n", + "结论:哪怕每步惩罚是 0,因为有 gamma 的折现,智能体依然知道直达路径更好!\n", + "\n", + "--- 情景 2:给每一步都增加 -1 的惩罚 (仿射变换) ---\n", + "增加惩罚后,直达路径(A)的回报: 8.00\n", + "增加惩罚后,绕路路径(B)的回报: 4.58\n", + "结论:增加统一惩罚后,直达路径(A)依然大于绕路路径(B)。相对好坏关系不变,最优策略不变。\n" + ] + } + ], + "source": [ + "# 验证:假设两条路径,路径A直达终点(2步),路径B绕路(4步)\n", + "gamma = 0.9\n", + "target_reward = 10.0\n", + "\n", + "print(\"--- 情景 1:每走一步没有任何惩罚 (即时奖励为 0) ---\")\n", + "# 路径A: 0 + 0.9 * 10\n", + "return_path_A = 0 + (gamma ** 1) * target_reward \n", + "# 路径B: 0 + 0.9*0 + 0.9^2*0 + 0.9^3 * 10\n", + "return_path_B = 0 + 0 + 0 + (gamma ** 3) * target_reward\n", + "\n", + "print(f\"直达路径(A)的回报: {return_path_A}\")\n", + "print(f\"绕路路径(B)的回报: {return_path_B}\")\n", + "print(\"结论:哪怕每步惩罚是 0,因为有 gamma 的折现,智能体依然知道直达路径更好!\\n\")\n", + "\n", + "print(\"--- 情景 2:给每一步都增加 -1 的惩罚 (仿射变换) ---\")\n", + "# 路径A: -1 + 0.9 * (10 - 1) -> 简化理解为每步 -1,最后一步拿到目标\n", + "return_path_A_penalty = -1 + (gamma ** 1) * target_reward\n", + "# 路径B: 绕远路\n", + "return_path_B_penalty = -1 + gamma*(-1) + (gamma**2)*(-1) + (gamma**3)*target_reward\n", + "\n", + "print(f\"增加惩罚后,直达路径(A)的回报: {return_path_A_penalty:.2f}\")\n", + "print(f\"增加惩罚后,绕路路径(B)的回报: {return_path_B_penalty:.2f}\")\n", + "print(\"结论:增加统一惩罚后,直达路径(A)依然大于绕路路径(B)。相对好坏关系不变,最优策略不变。\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "413fb515", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": "GymRL", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.13.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/Notebooks/C4.ipynb b/Notebooks/C4.ipynb new file mode 100644 index 0000000..5d97fdb --- /dev/null +++ b/Notebooks/C4.ipynb @@ -0,0 +1,230 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "id": "13468a42", + "metadata": {}, + "source": [ + "# 第 4 章:价值迭代与策略迭代 (Value Iteration and Policy Iteration)\n", + "\n", + "本章介绍的算法属于**动态规划 (Dynamic Programming)**,因为它们需要提前知道环境的完整模型(状态转移概率和奖励)。\n", + "\n", + "## 1. 价值迭代 (Value Iteration)\n", + "价值迭代是直接求解第三章“贝尔曼最优方程 (BOE)”的迭代算法。\n", + "它在每次迭代中包含两步:\n", + "* **策略更新 (Policy Update)**:算出 Q 值,找到当前最贪心(收益最大)的动作。\n", + "* **价值更新 (Value Update)**:把状态的价值直接更新为最大的 Q 值:$v_{k+1}(s) = \\max_a q_k(s,a)$。\n", + "* **注意**:价值迭代中间产生的 $v_k$ 仅仅是计算过程中的临时数值,它们并不代表任何特定策略的真实状态价值!\n", + "\n", + "## 2. 策略迭代 (Policy Iteration)\n", + "策略迭代的过程更像是一个稳扎稳打的“评估-改进”循环。\n", + "* **策略评估 (Policy Evaluation)**:对当前策略,死磕到底,算出它真实的、收敛的状态价值 $v_{\\pi_k}$(这通常需要内部再套一个循环)。\n", + "* **策略改进 (Policy Improvement)**:根据算出来的真实价值,选择 Q 值最大的动作,更新策略。这保证了新策略一定比老策略更好(或者一样好)。" + ] + }, + { + "cell_type": "code", + "execution_count": 1, + "id": "9ae8b2ba", + "metadata": {}, + "outputs": [], + "source": [ + "import numpy as np\n", + "\n", + "# --- 1. 定义 2x2 环境 ---\n", + "states = [\"s1\", \"s2\", \"s3\", \"s4\"]\n", + "actions = [\"up\", \"right\", \"down\", \"left\", \"stay\"]\n", + "gamma = 0.9\n", + "\n", + "# 状态转移规则 (依据 Table 4.1 逆向推导的简单网格规律)\n", + "def get_transition(state, action):\n", + " # s4 是目标(吸收态),到了就停在原地\n", + " if state == \"s4\": return \"s4\"\n", + " \n", + " if state == \"s1\":\n", + " if action == \"right\": return \"s2\"\n", + " if action == \"down\": return \"s3\"\n", + " if action == \"stay\": return \"s1\"\n", + " return \"s1\" # 撞墙反弹\n", + " elif state == \"s2\":\n", + " if action == \"left\": return \"s1\"\n", + " if action == \"down\": return \"s4\"\n", + " if action == \"stay\": return \"s2\"\n", + " return \"s2\" # 撞墙反弹\n", + " elif state == \"s3\":\n", + " if action == \"up\": return \"s1\"\n", + " if action == \"right\": return \"s4\"\n", + " if action == \"stay\": return \"s3\"\n", + " return \"s3\" # 撞墙反弹\n", + " return state\n", + "\n", + "# 奖励规则 (依据 Table 4.1 提取)\n", + "def get_reward(state, action, next_state):\n", + " if state == \"s4\": return 1 # 目标奖励\n", + " \n", + " # 判断是否撞墙 (尝试移动但留在原地)\n", + " if state == next_state and action != \"stay\":\n", + " return -1\n", + " \n", + " # 正常移动的奖励\n", + " if next_state == \"s2\": return -1 # 禁区\n", + " if next_state == \"s4\": return 1 # 目标\n", + " return 0 # 其他移动 (书中这题为 0)\n", + "\n", + "# 辅助函数:计算 Q 值 (这里假设转移是 100% 确定性的)\n", + "def compute_q_value(state, action, v_values):\n", + " next_s = get_transition(state, action)\n", + " reward = get_reward(state, action, next_s)\n", + " return reward + gamma * v_values[states.index(next_s)]" + ] + }, + { + "cell_type": "code", + "execution_count": 7, + "id": "32dc5456", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "=== 开始价值迭代 (Value Iteration) ===\n", + "迭代 1: V值 = [0. 1. 1. 1.], 策略 = ['down', 'down', 'right', 'up']\n", + "迭代 2: V值 = [0.9 1.9 1.9 1.9], 策略 = ['down', 'down', 'right', 'up']\n", + "迭代 3: V值 = [1.71 2.71 2.71 2.71], 策略 = ['down', 'down', 'right', 'up']\n", + "迭代 4: V值 = [2.44 3.44 3.44 3.44], 策略 = ['down', 'down', 'right', 'up']\n", + "迭代 5: V值 = [3.1 4.1 4.1 4.1], 策略 = ['down', 'down', 'right', 'up']\n" + ] + } + ], + "source": [ + "print(\"=== 开始价值迭代 (Value Iteration) ===\")\n", + "\n", + "# 初始化 V 值为 0\n", + "V_vi = np.zeros(len(states))\n", + "policy_vi = [\"stay\"] * len(states) \n", + "iterations = 5\n", + "\n", + "for k in range(iterations):\n", + " new_V = np.zeros(len(states))\n", + " \n", + " # 遍历所有状态\n", + " for i, s in enumerate(states):\n", + " q_values = []\n", + " # 遍历所有动作,计算 Q 值\n", + " for a in actions:\n", + " q = compute_q_value(s, a, V_vi)\n", + " q_values.append(q)\n", + " \n", + " # 核心:价值更新 (直接取最大的 Q 值)\n", + " best_q = max(q_values)\n", + " new_V[i] = best_q\n", + " \n", + " # 核心:策略更新 (记录最大 Q 值对应的动作)\n", + " best_action_idx = np.argmax(q_values)\n", + " policy_vi[i] = actions[best_action_idx]\n", + " \n", + " print(f\"迭代 {k+1}: V值 = {np.round(new_V, 2)}, 策略 = {policy_vi}\")\n", + " \n", + " # 如果价值不再变化,说明收敛了\n", + " if np.max(np.abs(new_V - V_vi)) < 1e-5:\n", + " print(f\"-> 价值迭代在第 {k+1} 步提前收敛!\")\n", + " V_vi = new_V\n", + " break\n", + " \n", + " V_vi = new_V" + ] + }, + { + "cell_type": "code", + "execution_count": 9, + "id": "9e6c5d0d", + "metadata": {}, + "outputs": [ + { + "name": "stdout", + "output_type": "stream", + "text": [ + "=== 开始策略迭代 (Policy Iteration) ===\n", + "\n", + "第 1 轮主迭代,当前策略: ['stay', 'stay', 'stay', 'stay']\n", + " 评估完成,当前策略的真实 V值 = [ 0. -10. 0. 10.]\n", + "\n", + "第 2 轮主迭代,当前策略: ['down', 'down', 'right', 'up']\n", + " 评估完成,当前策略的真实 V值 = [ 9. 10. 10. 10.]\n", + "-> 策略不再改变,策略迭代收敛!最优策略已找到。\n" + ] + } + ], + "source": [ + "print(\"=== 开始策略迭代 (Policy Iteration) ===\")\n", + "\n", + "V_pi = np.zeros(len(states))\n", + "# 初始给一个极差的策略:全部原地不动 (类似书中图 4.3 的烂策略)\n", + "current_policy = [\"stay\", \"stay\", \"stay\", \"stay\"] \n", + "\n", + "for k in range(5):\n", + " print(f\"\\n第 {k+1} 轮主迭代,当前策略: {current_policy}\")\n", + " \n", + " # --- 步骤 1: 策略评估 (Policy Evaluation) ---\n", + " # 死磕到底,一直循环直到 V 值收敛,算出当前策略的真实价值\n", + " while True:\n", + " new_V = np.zeros(len(states))\n", + " for i, s in enumerate(states):\n", + " action = current_policy[i] # 只看当前策略指定的动作\n", + " new_V[i] = compute_q_value(s, action, V_pi)\n", + " \n", + " if np.max(np.abs(new_V - V_pi)) < 1e-5:\n", + " break\n", + " V_pi = new_V\n", + " print(f\" 评估完成,当前策略的真实 V值 = {np.round(V_pi, 2)}\")\n", + " \n", + " # --- 步骤 2: 策略改进 (Policy Improvement) ---\n", + " policy_stable = True\n", + " for i, s in enumerate(states):\n", + " old_action = current_policy[i]\n", + " \n", + " # 看看有没有更好的动作\n", + " q_values = [compute_q_value(s, a, V_pi) for a in actions]\n", + " best_action = actions[np.argmax(q_values)]\n", + " \n", + " current_policy[i] = best_action\n", + " if old_action != best_action:\n", + " policy_stable = False # 策略发生了改变\n", + " \n", + " if policy_stable:\n", + " print(\"-> 策略不再改变,策略迭代收敛!最优策略已找到。\")\n", + " break" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5282df09", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": "GymRL", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.13.9" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +}