128 KiB
128 KiB
In [1]:
import numpy as np
# --- 1. 定义特征提取器 (Feature Extractor) ---
def get_features(state_x, state_y):
# 按照书中 8.2.4 的建议，将 x 和 y 归一化到 [-1, 1] 区间
# 假设网格是 5x5 (x, y 在 0 到 4 之间)
norm_x = (state_x - 2.0) / 2.0
norm_y = (state_y - 2.0) / 2.0
# 提取简单的 3D 特征向量: [1, x, y]^T
return np.array([1.0, norm_x, norm_y])
# --- 2. TD-Linear 参数初始化 ---
# 权重向量 w 初始设为 0 (对应特征维度 3)
w = np.zeros(3)
alpha = 0.01
gamma = 0.9
print("--- 体验 TD-Linear 的一次更新过程 ---")
# 假设智能体走了一步：从 (1, 1) 走到 (2, 1)，获得奖励 -1
s_t_x, s_t_y = 1, 1
s_next_x, s_next_y = 2, 1
reward = -1.0
# 1. 计算当前状态和下一个状态的特征
phi_t = get_features(s_t_x, s_t_y)
phi_next = get_features(s_next_x, s_next_y)
# 2. 用当前权重 w 计算预测价值 v = w^T * phi
v_t = np.dot(w, phi_t)
v_next = np.dot(w, phi_next)
# 3. 计算 TD 误差 (TD Error)
td_target = reward + gamma * v_next
td_error = td_target - v_t
# 4. 更新权重 w
w = w + alpha * td_error * phi_t
print(f"提取的当前特征 phi_t: {phi_t}")
print(f"TD 误差: {td_error}")
print(f"更新后的权重 w: {w}")
print("结论：我们没有单独记录状态 (1,1) 的价值，而是更新了统管全局的参数 w！这就是函数近似的精髓。")--- 体验 TD-Linear 的一次更新过程 --- 提取的当前特征 phi_t: [ 1. -0.5 -0.5] TD 误差: -1.0 更新后的权重 w: [-0.01 0.005 0.005] 结论：我们没有单独记录状态 (1,1) 的价值，而是更新了统管全局的参数 w！这就是函数近似的精髓。
In [4]:
import numpy as np
import matplotlib.pyplot as plt
# --- 1. 定义环境 ---
def step(state):
# 随机向左或向右走
action = np.random.choice([-1, 1])
next_state = state + action
# 到达最右侧(6)奖励为1，其余为0
reward = 1.0 if next_state == 6 else 0.0
done = next_state == 0 or next_state == 6
return next_state, reward, done
# --- 2. 定义特征提取器 (Feature Extractor) ---
# 对应书中公式：phi(s) = [1, x]^T
def get_features(state):
# 将状态归一化到 [0, 1] 之间，这是机器学习的好习惯
norm_s = state / 6.0
return np.array([1.0, norm_s])
# --- 3. 运行 TD-Linear 算法 ---
# 初始化参数 w 为 0 (只有两个参数！不管走廊有多长，都只需要2个参数)
w = np.zeros(2)
alpha = 0.1 # 学习率
gamma = 1.0 # 无折扣
# 记录每个 episode 后的状态价值，用于画图
history_v = []
num_episodes = 200
for ep in range(num_episodes):
state = 3 # 从中间开始
while True:
next_state, reward, done = step(state)
# 提取当前状态和下一个状态的特征 phi
phi_t = get_features(state)
phi_next = get_features(next_state)
# 计算当前的预测价值 v = w^T * phi
v_t = np.dot(w, phi_t)
# 如果游戏结束，下一个状态的价值必定是 0
v_next = 0.0 if done else np.dot(w, phi_next)
# 计算 TD 误差
td_target = reward + gamma * v_next
td_error = td_target - v_t
# 核心：使用 TD-Linear 公式更新参数 w
# w_{t+1} = w_t + alpha * (TD_target - v_t) * phi(s_t)
w = w + alpha * td_error * phi_t
state = next_state
if done:
break
# 每局结束后，把当前 w 眼中的“所有状态价值”记录下来
current_estimated_v = [np.dot(w, get_features(s)) for s in range(1, 6)]
history_v.append(current_estimated_v)
# --- 4. 可视化学习过程 ---
true_values = [1/6, 2/6, 3/6, 4/6, 5/6]
plt.figure(figsize=(10, 6))
plt.plot(range(1, 6), true_values, 'r-o', linewidth=2, label='True Values')
# 画出第 10, 50, 199 局的拟合结果
for ep in [10, 50, 199]:
plt.plot(range(1, 6), history_v[ep], '--', label=f'Estimated Values @ Ep {ep}')
plt.title('TD-Linear: Learning State Values with a Straight Line')
plt.xlabel('State')
plt.ylabel('Value Estimate')
plt.legend()
plt.grid(True)
plt.show()
print(f"最终学到的参数 w: {np.round(w, 3)}")
最终学到的参数 w: [0.177 0.888]
In [5]:
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import random
from collections import deque
import matplotlib.pyplot as plt
# --- 1. 定义非线性环境 ---
class NonLinearCorridor:
def __init__(self):
self.state = 0
def step(self, action):
# action: 0向左, 1向右
move = -1 if action == 0 else 1
self.state = max(0, min(5, self.state + move))
# 奖励机制：状态 5 是宝箱，状态 2 是陷阱
if self.state == 5:
return self.state, 10.0, True
elif self.state == 2:
return self.state, -10.0, False
else:
return self.state, 0.0, False
def reset(self):
self.state = 0
return self.state
# --- 2. 定义神经网络 Q-Network ---
# 输入是状态(1维)，隐藏层提取非线性特征，输出是2个动作的Q值
class QNetwork(nn.Module):
def __init__(self):
super(QNetwork, self).__init__()
# 浅层网络足以解决简单的网格世界问题
self.fc1 = nn.Linear(1, 16)
self.relu = nn.ReLU()
self.fc2 = nn.Linear(16, 2)
def forward(self, x):
x = self.relu(self.fc1(x))
return self.fc2(x)
# --- 3. 经验回放缓冲区 (Experience Replay)---
class ReplayBuffer:
def __init__(self, capacity=1000):
self.buffer = deque(maxlen=capacity)
def add(self, state, action, reward, next_state, done):
self.buffer.append((state, action, reward, next_state, done))
def sample(self, batch_size):
# 均匀随机抽样，打破数据的时间相关性
return random.sample(self.buffer, batch_size)
def __len__(self):
return len(self.buffer)
# --- 4. DQN 核心训练逻辑 ---
env = NonLinearCorridor()
# 初始化主网络和目标网络，并让它们参数一致
main_net = QNetwork()
target_net = QNetwork()
target_net.load_state_dict(main_net.state_dict())
optimizer = optim.Adam(main_net.parameters(), lr=0.01)
loss_fn = nn.MSELoss()
buffer = ReplayBuffer(capacity=2000)
batch_size = 32
gamma = 0.9
epsilon = 0.3 # 探索率
update_target_every = 20 # 每隔 C 步更新一次目标网络
episodes = 300
loss_history = []
step_count = 0
print("开始训练 DQN，让神经网络去感受陷阱和宝箱...")
for ep in range(episodes):
state = env.reset()
done = False
while not done:
step_count += 1
# --- 策略：Epsilon-Greedy ---
if random.random() < epsilon:
action = random.choice([0, 1])
else:
state_tensor = torch.FloatTensor([[state]])
with torch.no_grad():
q_values = main_net(state_tensor)
action = torch.argmax(q_values).item()
next_state, reward, done = env.step(action)
# 存入经验回放池
buffer.add(state, action, reward, next_state, done)
state = next_state
# --- 训练阶段 ---
if len(buffer) >= batch_size:
# 1. 抽取 Mini-batch
batch = buffer.sample(batch_size)
b_s = torch.FloatTensor([[x[0]] for x in batch])
b_a = torch.LongTensor([[x[1]] for x in batch])
b_r = torch.FloatTensor([[x[2]] for x in batch])
b_ns = torch.FloatTensor([[x[3]] for x in batch])
b_d = torch.FloatTensor([[x[4]] for x in batch])
# 2. 计算当前 Q 值预测: main_net(S)[A]
q_pred = main_net(b_s).gather(1, b_a)
# 3. 计算目标 Q 值 (使用 Target Network)
# y_T = R + gamma * max_a Q_target(S', a)
with torch.no_grad():
max_q_next = target_net(b_ns).max(1, keepdim=True)[0]
q_target = b_r + gamma * max_q_next * (1 - b_d)
# 4. 反向传播更新网络
loss = loss_fn(q_pred, q_target)
optimizer.zero_grad()
loss.backward()
optimizer.step()
loss_history.append(loss.item())
# 5. 定期同步目标网络
if step_count % update_target_every == 0:
target_net.load_state_dict(main_net.state_dict())
print("训练完成！")
# --- 5. 验证神经网络学到了什么 ---
print("\n--- 揭晓神经网络眼中的 Q 值 ---")
states_to_test = torch.FloatTensor([[s] for s in range(6)])
with torch.no_grad():
learned_q = main_net(states_to_test).numpy()
for s in range(6):
q_left, q_right = learned_q[s]
best_act = "向左" if q_left > q_right else "向右"
print(f"状态 {s}: Q(向左)={q_left:6.2f}, Q(向右)={q_right:6.2f} => 最优决策: {best_act}")
# 画出 Loss 曲线 (证明它在收敛)
plt.plot(loss_history)
plt.title("DQN Training Loss")
plt.xlabel("Training Steps")
plt.ylabel("Loss (MSE)")
plt.show()开始训练 DQN，让神经网络去感受陷阱和宝箱... 训练完成！ --- 揭晓神经网络眼中的 Q 值 --- 状态 0: Q(向左)= 0.34, Q(向右)= 0.13 => 最优决策: 向左 状态 1: Q(向左)= 0.32, Q(向右)= -3.04 => 最优决策: 向左 状态 2: Q(向左)= -2.48, Q(向右)= 8.08 => 最优决策: 向右 状态 3: Q(向左)= -0.92, Q(向右)= 8.95 => 最优决策: 向右 状态 4: Q(向左)= 4.09, Q(向右)= 9.81 => 最优决策: 向右 状态 5: Q(向左)= 9.18, Q(向右)= 10.66 => 最优决策: 向右
