Files
2026-02-27 21:17:25 +08:00

9.7 KiB

第 1 章:强化学习基本概念 (Basic Concepts)

本笔记本旨在复现 3x3 网格世界中的基础马尔可夫决策过程 (MDP) 元素。

In [1]:
# 定义状态空间 (9个格子)
states = ["s1", "s2", "s3", "s4", "s5", "s6", "s7", "s8", "s9"]

# 定义动作空间 (5种动作)
actions = ["a1", "a2", "a3", "a4", "a5"] # 分别代表:上、右、下、左、原地

print(f"状态空间大小: {len(states)}")
print(f"动作空间大小: {len(actions)}")
状态空间大小: 9
动作空间大小: 5
In [4]:
# 构建状态转移字典,格式:(当前状态, 动作): 下一个状态
transitions = {}

# s1 的转移
transitions[("s1", "a1")] = "s1" # 向上撞墙,反弹回 s1
transitions[("s1", "a2")] = "s2" # 向右进入 s2
transitions[("s1", "a3")] = "s4" # 向下进入 s4
transitions[("s1", "a4")] = "s1" # 向左撞墙,反弹回 s1
transitions[("s1", "a5")] = "s1" # 原地不动

# s2 的转移
transitions[("s2", "a1")] = "s2" # 向上撞墙,反弹回 s2
transitions[("s2", "a2")] = "s3" # 向右进入 s3
transitions[("s2", "a3")] = "s5" # 向下进入 s5
transitions[("s2", "a4")] = "s1" # 向左进入 s1
transitions[("s2", "a5")] = "s2" # 原地不动

# s3 的转移
transitions[("s3", "a1")] = "s3" # 向上撞墙
transitions[("s3", "a2")] = "s3" # 向右撞墙
transitions[("s3", "a3")] = "s6" # 向下进入 s6
transitions[("s3", "a4")] = "s2" # 向左进入 s2
transitions[("s3", "a5")] = "s3" # 原地不动

# s4 的转移
transitions[("s4", "a1")] = "s1" # 向上进入 s1
transitions[("s4", "a2")] = "s5" # 向右进入 s5
transitions[("s4", "a3")] = "s7" # 向下进入 s7
transitions[("s4", "a4")] = "s4" # 向左撞墙
transitions[("s4", "a5")] = "s4" # 原地不动

# s5 的转移
transitions[("s5", "a1")] = "s2" # 向上进入 s2
transitions[("s5", "a2")] = "s6" # 向右进入 s6
transitions[("s5", "a3")] = "s8" # 向下进入 s8
transitions[("s5", "a4")] = "s4" # 向左进入 s4
transitions[("s5", "a5")] = "s5" # 原地不动

# s6 的转移
transitions[("s6", "a1")] = "s3" # 向上进入 s3
transitions[("s6", "a2")] = "s6" # 向右撞墙
transitions[("s6", "a3")] = "s9" # 向下进入 s9
transitions[("s6", "a4")] = "s5" # 向左进入 s5
transitions[("s6", "a5")] = "s6" # 原地不动

# s7 的转移
transitions[("s7", "a1")] = "s4" # 向上进入 s4
transitions[("s7", "a2")] = "s8" # 向右进入 s8
transitions[("s7", "a3")] = "s7" # 向下撞墙
transitions[("s7", "a4")] = "s7" # 向左撞墙
transitions[("s7", "a5")] = "s7" # 原地不动

# s8 的转移
transitions[("s8", "a1")] = "s5" # 向上进入 s5
transitions[("s8", "a2")] = "s9" # 向右进入 s9
transitions[("s8", "a3")] = "s8" # 向下撞墙
transitions[("s8", "a4")] = "s7" # 向左进入 s7
transitions[("s8", "a5")] = "s8" # 原地不动

# s9 的转移
transitions[("s9", "a1")] = "s6" # 向上进入 s6
transitions[("s9", "a2")] = "s9" # 向右撞墙
transitions[("s9", "a3")] = "s9" # 向下撞墙
transitions[("s9", "a4")] = "s8" # 向左进入 s8
transitions[("s9", "a5")] = "s9" # 原地不动

def get_next_state(state, action):
    return transitions.get((state, action), state) # 如果没定义,默认原地不动

print(f"在 s1 采取动作 a2,下一个状态是: {get_next_state('s1', 'a2')}")
在 s1 采取动作 a2,下一个状态是: s2
In [5]:
# 定义奖励规则
def get_reward(state, action, next_state):
    # 如果撞墙(当前状态和下一个状态一样,且不是原地动作)
    if state == next_state and action != "a5":
        return -1
    # 如果进入目标状态 s9
    elif next_state == "s9":
        return 1
    # 如果进入禁区 s6 或 s7 (根据书中图 1.2)
    elif next_state in ["s6", "s7"]:
        return -1
    # 其他情况
    else:
        return 0

# 测试一下
test_r1 = get_reward("s1", "a1", "s1")
test_r2 = get_reward("s8", "a2", "s9")
print(f"s1撞墙奖励: {test_r1}") # 应该输出 -1
print(f"进入目标s9奖励: {test_r2}") # 应该输出 1
s1撞墙奖励: -1
进入目标s9奖励: 1
In [8]:
# 定义一条书中的轨迹 (图 1.6 左侧的好策略轨迹)
# 格式:[(状态, 动作), ...]
good_trajectory = [
    ("s1", "a2"), ("s2", "a3"), ("s5", "a3"), ("s8", "a2")
]

# 计算回报的函数
def calculate_discounted_return(trajectory, gamma=0.9):
    total_return = 0.0
    
    for t, (state, action) in enumerate(trajectory):
        next_state = get_next_state(state, action)
        reward = get_reward(state, action, next_state)
        
        # 计算折扣回报
        total_return += (gamma ** t) * reward
        print(f"Step {t}: {state} --{action}--> {next_state}, Reward: {reward}")
        
    return total_return

print("--- 运行轨迹并计算回报 ---")
final_return = calculate_discounted_return(good_trajectory, gamma=0.9)
print(f"最终折扣回报: {final_return}")
--- 运行轨迹并计算回报 ---
Step 0: s1 --a2--> s2, Reward: 0
Step 1: s2 --a3--> s5, Reward: 0
Step 2: s5 --a3--> s8, Reward: 0
Step 3: s8 --a2--> s9, Reward: 1
最终折扣回报: 0.7290000000000001

接下来来有一些随机策略的case

In [9]:
import numpy as np

# 定义一个随机策略字典
# 格式:状态: {动作: 概率, 动作: 概率}
stochastic_policy = {
    "s1": {"a2": 0.5, "a3": 0.5}, # 图 1.5 中的设定
    "s2": {"a3": 1.0},            # 其他状态假设是确定性的 100%
    "s5": {"a3": 1.0},
    "s8": {"a2": 1.0}
}

def choose_action(state, policy):
    # 如果策略里有这个状态的动作概率分布
    if state in policy:
        action_probs = policy[state]
        actions = list(action_probs.keys())
        probs = list(action_probs.values())
        # 根据概率分布随机选择一个动作
        chosen_action = np.random.choice(actions, p=probs)
        return chosen_action
    else:
        return "a5" # 默认原地不动

# 测试随机策略:让智能体在 s1 连续做 10 次决定
print("在状态 s1 进行 10 次动作选择测试:")
choices = [choose_action("s1", stochastic_policy) for _ in range(10)]
print(choices)
print(f"向右(a2)的次数: {choices.count('a2')}, 向下(a3)的次数: {choices.count('a3')}")
在状态 s1 进行 10 次动作选择测试:
[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')]
向右(a2)的次数: 5, 向下(a3)的次数: 5