286 lines
9.7 KiB
Plaintext
286 lines
9.7 KiB
Plaintext
{
|
|
"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
|
|
}
|