102 KiB
102 KiB
In [1]:
import numpy as np
import random
from collections import defaultdict
# 1. 简单的 1D 走廊环境 (黑盒)
# 状态: 0, 1, 2, 3 (3 是目标宝箱)
# 动作: 0 (向左), 1 (向右)
def step(state, action):
if state == 3: # 已经在终点
return 3, 0, True
if action == 1: # 向右走
next_state = state + 1
else: # 向左走
next_state = max(0, state - 1)
reward = 10 if next_state == 3 else -1
done = (next_state == 3) # 是否结束回合
return next_state, reward, done
# 2. 定义 epsilon-greedy 策略
def epsilon_greedy_policy(state, Q, epsilon, n_actions=2):
# 以 epsilon 的概率随机探索
if random.uniform(0, 1) < epsilon:
return random.choice(range(n_actions))
# 以 1 - epsilon 的概率贪心利用 (选择 Q 值最大的动作)
else:
# 如果 Q 值全是 0,也会默认选第一个,所以用 argmax
return np.argmax(Q[state])In [3]:
print("=== 开始蒙特卡洛控制 (Monte Carlo Control) ===")
# 初始化 Q 表 (状态数目为4,动作为2)
# 使用 defaultdict 方便处理没见过的状态
Q = defaultdict(lambda: np.zeros(2))
# 用于记录每个 (状态, 动作) 组合被访问了多少次,以及获得的总回报
returns_sum = defaultdict(float)
returns_count = defaultdict(float)
num_episodes = 500
gamma = 0.9
epsilon = 0.2
for i in range(num_episodes):
# --- 1. 生成一个完整的回合 (Episode) ---
episode = []
state = 0 # 每次都从起点开始
# 智能体开始在黑盒里凭感觉走,直到碰壁或找到宝箱
while True:
action = epsilon_greedy_policy(state, Q, epsilon)
next_state, reward, done = step(state, action)
# 记录下这一步的“经验”: (状态, 动作, 奖励)
episode.append((state, action, reward))
state = next_state
if done:
break
# --- 2. 回合结束后,从后往前算回报并更新 Q 表 ---
G = 0.0 # G 代表累计回报 Return
# 从轨迹的最后一步倒着往前算
for t in reversed(range(len(episode))):
state, action, reward = episode[t]
# 计算折扣回报
G = gamma * G + reward
# First-Visit MC (初次访问蒙特卡洛):
# 只在回合中首次遇到这个 (状态,动作) 时才更新
state_action_pairs_before_t = [(x[0], x[1]) for x in episode[:t]]
if (state, action) not in state_action_pairs_before_t:
# 记录总回报并计数
returns_sum[(state, action)] += G
returns_count[(state, action)] += 1.0
# 平均值法更新 Q 表: Q = sum(G) / count
Q[state][action] = returns_sum[(state, action)] / returns_count[(state, action)]
# 打印部分训练过程
if (i + 1) % 100 == 0:
print(f"完成第 {i + 1} 个回合训练...")
print("\n--- 训练结束!揭晓学到的 Q 表 ---")
for s in range(3):
print(f"状态 {s}: 向左 Q={Q[s][0]:.2f}, 向右 Q={Q[s][1]:.2f} -> 最优动作: {'向右' if np.argmax(Q[s])==1 else '向左'}")
print("\n结论:即使不知道环境具体规则,智能体仅凭不断试错取平均,也学会了一直向右走才是通关秘籍!")=== 开始蒙特卡洛控制 (Monte Carlo Control) === 完成第 100 个回合训练... 完成第 200 个回合训练... 完成第 300 个回合训练... 完成第 400 个回合训练... 完成第 500 个回合训练... --- 训练结束!揭晓学到的 Q 表 --- 状态 0: 向左 Q=3.44, 向右 Q=5.38 -> 最优动作: 向右 状态 1: 向左 Q=3.03, 向右 Q=7.49 -> 最优动作: 向右 状态 2: 向左 Q=5.10, 向右 Q=10.00 -> 最优动作: 向右 结论:即使不知道环境具体规则,智能体仅凭不断试错取平均,也学会了一直向右走才是通关秘籍!
In [ ]:
import numpy as np
import matplotlib.pyplot as plt
import random
from collections import defaultdict
# 1. 简单的 1D 走廊黑盒环境
def step(state, action):
if state == 3:
return 3, 0, True
if action == 1:
next_state = state + 1
else:
next_state = max(0, state - 1)
reward = 10 if next_state == 3 else -1
done = (next_state == 3)
return next_state, reward, done
# 2. epsilon-greedy 策略
def epsilon_greedy_policy(state, Q, epsilon, n_actions=2):
if random.uniform(0, 1) < epsilon:
return random.choice(range(n_actions))
else:
return np.argmax(Q[state])
# 3. 封装好的蒙特卡洛控制算法
def run_mc_control(num_episodes, gamma, initial_epsilon, decay_epsilon=False):
Q = defaultdict(lambda: np.zeros(2))
returns_sum = defaultdict(float)
returns_count = defaultdict(float)
episode_returns = [] # 记录每一局的最终回报,用来画图
# 循环执行多次回合
for i in range(num_episodes):
# --- 【彩蛋核心逻辑:Epsilon 衰减】 ---
# 如果开启衰减,每一局的 epsilon 都会按照比例减小,但最低不低于 0.01
if decay_epsilon:
epsilon = max(0.01, initial_epsilon * (1 - i / num_episodes))
else:
epsilon = initial_epsilon
episode = []
state = 0
episode_return = 0
# 跑完一个完整的回合
while True:
action = epsilon_greedy_policy(state, Q, epsilon)
next_state, reward, done = step(state, action)
episode.append((state, action, reward))
episode_return += reward
state = next_state
if done: break
episode_returns.append(episode_return)
# 从最后一步倒算回报并更新 Q 表 (初次访问 MC)
G = 0.0
for t in reversed(range(len(episode))):
s, a, r = episode[t]
G = gamma * G + r
# 检查是否初次访问
is_first_visit = True
for prev_t in range(t):
if episode[prev_t][0] == s and episode[prev_t][1] == a:
is_first_visit = False
break
if is_first_visit:
returns_sum[(s, a)] += G
returns_count[(s, a)] += 1.0
Q[s][a] = returns_sum[(s, a)] / returns_count[(s, a)]
return episode_returns
# --- 4. 运行对比实验并绘图 ---
num_episodes = 500
gamma = 0.9
# 实验一:恒定 epsilon = 0.2
returns_const = run_mc_control(num_episodes, gamma, initial_epsilon=0.2, decay_epsilon=False)
# 实验二:衰减 epsilon (初始值大胆设为 0.5,慢慢降到 0.01)
returns_decay = run_mc_control(num_episodes, gamma, initial_epsilon=0.5, decay_epsilon=True)
# 计算滑动平均 (平滑曲线,看起来更直观)
def moving_average(a, n=20):
ret = np.cumsum(a, dtype=float)
ret[n:] = ret[n:] - ret[:-n]
return ret[n - 1:] / n
# 绘图
plt.figure(figsize=(10, 6))
plt.plot(moving_average(returns_const), label='Constant Epsilon (0.2)')
plt.plot(moving_average(returns_decay), label='Decaying Epsilon (0.5 -> 0.01)')
plt.title('Monte Carlo Control: Constant vs Decaying Epsilon')
plt.xlabel('Episodes (Smoothed over 20 episodes)')
plt.ylabel('Average Return per Episode')
plt.legend()
plt.grid(True)
plt.show()[1;31m在当前单元格或上一个单元格中执行代码时 Kernel 崩溃。 [1;31m请查看单元格中的代码,以确定故障的可能原因。 [1;31m单击<a href='https://aka.ms/vscodeJupyterKernelCrash'>此处</a>了解详细信息。 [1;31m有关更多详细信息,请查看 Jupyter <a href='command:jupyter.viewOutput'>log</a>。