MiNiKou的博客

1
观测 s_t → 根据策略选 a_t → 环境执行 → 得到 r_t、s_{t+1}
2
    ↑                                      ↓
3
    └──────── 学习并更新策略或价值函数 ──────┘

1
观察状态 → ε-greedy 动作 → 环境转移 → 保存到经验池
2
                                      ↓ 随机抽样
3
在线网络 Q(s,a) ← Huber 损失 ← 目标网络构造 TD 目标
4
       ↓ 周期性复制
5
    目标网络

1
"""无第三方依赖的 Q-learning：5×5 迷宫。Python 3.10+。"""
2
import argparse
3
import json
4
import random
5
from pathlib import Path
6

7
SIZE = 5
8
START, GOAL = (0, 0), (4, 4)
9
WALLS = {(1, 1), (1, 3), (2, 3), (3, 1)}
10
MOVES = ((-1, 0), (1, 0), (0, -1), (0, 1))  # 上、下、左、右
11

12

13
def transition(state, action):
14
    """撞墙留在原地；每步付出成本，鼓励找到短路径。"""
15
    # 用 row * SIZE + col 将二维坐标编码为 Q 表的行号。
16
    row, col = divmod(state, SIZE)
17
    dr, dc = MOVES[action]
18
    nxt = (row + dr, col + dc)
19
    if not (0 <= nxt[0] < SIZE and 0 <= nxt[1] < SIZE) or nxt in WALLS:
20
        nxt = (row, col)
21
    terminated = nxt == GOAL
22
    return nxt[0] * SIZE + nxt[1], 10.0 if terminated else -1.0, terminated
23

24

25
def greedy(values, rng):
26
    # 随机打破并列，避免全零初始化时始终选择第一个动作。
27
    best = max(values)
28
    return rng.choice([i for i, v in enumerate(values) if v == best])
29

30

31
def train(episodes=3000, seed=42):
32
    if episodes <= 0:
33
        raise ValueError("episodes 必须为正整数")
34
    rng = random.Random(seed)
35
    # 每个状态占一行，四列分别对应上下左右动作的长期价值估计。
36
    q = [[0.0] * 4 for _ in range(SIZE * SIZE)]
37
    # alpha 控制 TD 更新步长；gamma 决定未来奖励的折扣程度。
38
    alpha, gamma = 0.1, 0.95
39
    for episode in range(episodes):
40
        state = 0
41
        # 随训练减少随机探索，但保留至少 5% 的探索机会。
42
        epsilon = max(0.05, 1.0 - episode / (episodes * 0.8))
43
        for _ in range(100):
44
            action = rng.randrange(4) if rng.random() < epsilon else greedy(q[state], rng)
45
            nxt, reward, terminated = transition(state, action)
46
            # 真实终止不 bootstrap；100 步是采样保护上限，不属于任务终止。
47
            target = reward + (0.0 if terminated else gamma * max(q[nxt]))
48
            # target - Q(s,a) 是 TD 误差；只更新本次访问的状态动作。
49
            q[state][action] += alpha * (target - q[state][action])
50
            state = nxt
51
            if terminated:
52
                break
53
    return q
54

55

56
def evaluate(q, seed=10042):
57
    """使用独立随机流，不探索、不更新 Q 表。"""
58
    rng = random.Random(seed)
59
    state, reward_sum = 0, 0.0
60
    route = [START]
61
    for _ in range(100):
62
        state, reward, terminated = transition(state, greedy(q[state], rng))
63
        route.append(divmod(state, SIZE))
64
        reward_sum += reward
65
        if terminated:
66
            return {"success": True, "steps": len(route) - 1,
67
                    "return": reward_sum, "route": route}
68
    return {"success": False, "steps": 100, "return": reward_sum, "route": route}
69

70

71
if __name__ == "__main__":
72
    parser = argparse.ArgumentParser()
73
    parser.add_argument("--episodes", type=int, default=3000)
74
    parser.add_argument("--seed", type=int, default=42)
75
    parser.add_argument("--output", type=Path, default=Path("q_table.json"))
76
    args = parser.parse_args()
77
    table = train(args.episodes, args.seed)
78
    # 用独立的随机流打破并列动作，不让评估消耗训练的随机序列。
79
    result = evaluate(table, args.seed + 10000)
80
    args.output.write_text(json.dumps({"seed": args.seed, "q": table,
81
                                      "evaluation": result}, indent=2), encoding="utf-8")
82
    print(json.dumps(result, ensure_ascii=False))

1
"""教学版 DQN / Double DQN；状态为连续向量，动作只有左右两个。"""
2
import argparse
3
import json
4
import random
5
from collections import deque
6
from pathlib import Path
7

8
import gymnasium as gym
9
import numpy as np
10
import torch
11
from torch import nn
12

13

14
class QNetwork(nn.Module):
15
    def __init__(self, obs_dim, actions):
16
        super().__init__()
17
        # 输出是无界 Q 值，不是概率，最后一层不能使用 softmax。
18
        self.layers = nn.Sequential(nn.Linear(obs_dim, 64), nn.ReLU(),
19
                                    nn.Linear(64, 64), nn.ReLU(),
20
                                    nn.Linear(64, actions))
21

22
    def forward(self, states):
23
        return self.layers(states)
24

25

26
def evaluate(net, device, seed, episodes=20):
27
    env = gym.make("CartPole-v1")  # 单独的评估环境，避免改变训练状态。
28
    scores = []
29
    net.eval()
30
    with torch.no_grad():
31
        for episode in range(episodes):
32
            obs, _ = env.reset(seed=seed + episode)
33
            score = 0.0
34
            while True:
35
                x = torch.as_tensor(obs, dtype=torch.float32, device=device).unsqueeze(0)
36
                action = net(x).argmax(dim=1).item()  # 评估关闭 epsilon 探索。
37
                obs, reward, terminated, truncated, _ = env.step(action)
38
                score += reward
39
                if terminated or truncated:
40
                    break
41
            scores.append(score)
42
    env.close()
43
    net.train()
44
    return {"mean": float(np.mean(scores)), "std": float(np.std(scores)), "scores": scores}
45

46

47
def main(args):
48
    if args.steps <= 0:
49
        raise ValueError("steps 必须为正整数")
50
    random.seed(args.seed)
51
    np.random.seed(args.seed)
52
    torch.manual_seed(args.seed)
53
    torch.set_num_threads(1)  # 小网络在 CPU 上通常不需要大量线程。
54
    device = torch.device(args.device)
55
    env = gym.make("CartPole-v1")
56
    env.action_space.seed(args.seed)
57
    obs, _ = env.reset(seed=args.seed)
58
    online = QNetwork(4, 2).to(device)
59
    target = QNetwork(4, 2).to(device)
60
    # 两个网络从相同参数出发；只有 online 的参数交给优化器更新。
61
    target.load_state_dict(online.state_dict())
62
    target.eval()
63
    optimizer = torch.optim.Adam(online.parameters(), lr=1e-3)
64
    # 有界经验池自动淘汰最旧转移，限制内存并复用历史交互数据。
65
    replay = deque(maxlen=50000)
66
    batch_size, gamma = 64, 0.99
67
    history = []
68
    episode_return = 0.0
69
    for step in range(1, args.steps + 1):
70
        # 前 20000 个环境步逐渐从随机探索转向利用，之后保持 5% 探索。
71
        epsilon = max(0.05, 1.0 - 0.95 * step / 20000)
72
        if random.random() < epsilon:
73
            action = env.action_space.sample()
74
        else:
75
            with torch.no_grad():
76
                action = online(torch.as_tensor(obs, dtype=torch.float32,
77
                                device=device).unsqueeze(0)).argmax(dim=1).item()
78
        next_obs, reward, terminated, truncated, _ = env.step(action)
79
        # 保存 reset 之前的 next_obs。只用 terminated 掩蔽 bootstrap。
80
        replay.append((obs.copy(), action, reward, next_obs.copy(), terminated))
81
        episode_return += reward
82
        obs = next_obs
83
        if terminated or truncated:
84
            history.append(episode_return)
85
            episode_return = 0.0
86
            obs, _ = env.reset()  # 不要每回合都用相同 seed 重置。
87
        # 先积累 1000 条经验；之后每四个环境步做一次小批量更新。
88
        if step >= 1000 and step % 4 == 0:
89
            # 随机抽样降低连续转移的相关性；B 表示 batch_size。
90
            states, actions, rewards, next_states, terminals = zip(*random.sample(replay, batch_size))
91
            states = torch.as_tensor(np.array(states), dtype=torch.float32, device=device)
92
            next_states = torch.as_tensor(np.array(next_states), dtype=torch.float32, device=device)
93
            # actions 为 [B,1] 的整数索引，供 gather 选择每条经验的动作列。
94
            actions = torch.as_tensor(actions, dtype=torch.int64, device=device).unsqueeze(1)
95
            rewards = torch.as_tensor(rewards, dtype=torch.float32, device=device)
96
            terminals = torch.as_tensor(terminals, dtype=torch.float32, device=device)
97
            # 网络输出 [B,2]；取实际执行动作的 Q 值，得到 [B]。
98
            prediction = online(states).gather(1, actions).squeeze(1)
99
            # TD 目标是固定监督信号，不对下一状态的网络输出反向传播。
100
            with torch.no_grad():
101
                if args.double:
102
                    # Double DQN：在线网络选动作，目标网络评价该动作。
103
                    next_actions = online(next_states).argmax(dim=1, keepdim=True)
104
                    next_q = target(next_states).gather(1, next_actions).squeeze(1)
105
                else:
106
                    next_q = target(next_states).max(dim=1).values
107
                td_target = rewards + gamma * (1.0 - terminals) * next_q
108
            # Huber 损失减弱大 TD 误差的影响，比平方误差更不易受异常值支配。
109
            loss = nn.functional.smooth_l1_loss(prediction, td_target)
110
            optimizer.zero_grad()
111
            loss.backward()
112
            # 限制梯度范数，降低单次更新过大的风险；不裁剪 Q 值本身。
113
            nn.utils.clip_grad_norm_(online.parameters(), 10.0)
114
            optimizer.step()
115
        if step % 1000 == 0:  # 此处单位为环境步，而不是梯度更新次数。
116
            target.load_state_dict(online.state_dict())
117
        if step % 10000 == 0:
118
            print(f"step={step} epsilon={epsilon:.3f} train_mean={np.mean(history[-20:]):.1f}")
119
    # 最终评估关闭随机探索；训练回报含探索噪声，两者应分别记录。
120
    result = evaluate(online, device, args.seed + 10000)
121
    args.output.mkdir(parents=True, exist_ok=True)
122
    # 这里只保存推理权重；恢复训练还需优化器、经验池及随机状态。
123
    torch.save(online.state_dict(), args.output / "q_network.pt")
124
    (args.output / "metrics.json").write_text(json.dumps({
125
        "algorithm": "Double DQN" if args.double else "DQN", "seed": args.seed,
126
        "env": "CartPole-v1", "steps": args.steps, "device": str(device),
127
        "torch": torch.__version__, "gymnasium": gym.__version__,
128
        "config": {"lr": 1e-3, "gamma": gamma, "batch": batch_size,
129
                   "buffer": 50000, "warmup": 1000, "train_freq": 4,
130
                   "target_interval": 1000, "epsilon_decay_steps": 20000},
131
        "train_returns": history, "evaluation": result}, indent=2), encoding="utf-8")
132
    env.close()
133
    print(result)
134

135

136
if __name__ == "__main__":
137
    parser = argparse.ArgumentParser()
138
    parser.add_argument("--steps", type=int, default=100000)
139
    parser.add_argument("--seed", type=int, default=42)
140
    parser.add_argument("--device", default="cpu")
141
    parser.add_argument("--double", action="store_true")
142
    parser.add_argument("--output", type=Path, default=Path("runs/dqn"))
143
    main(parser.parse_args())

1
"""从零实现离散动作 PPO-Clip + GAE；便于与 DQN 在同一环境比较。"""
2
import argparse
3
import json
4
from pathlib import Path
5

6
import gymnasium as gym
7
import numpy as np
8
import torch
9
from torch import nn
10
from torch.distributions import Categorical
11

12

13
class ActorCritic(nn.Module):
14
    def __init__(self):
15
        super().__init__()
16
        # Actor 输出两个动作的 logits；Critic 输出一个状态价值，参数分开学习。
17
        self.actor = nn.Sequential(nn.Linear(4, 64), nn.Tanh(),
18
                                   nn.Linear(64, 64), nn.Tanh(), nn.Linear(64, 2))
19
        self.critic = nn.Sequential(nn.Linear(4, 64), nn.Tanh(),
20
                                    nn.Linear(64, 64), nn.Tanh(), nn.Linear(64, 1))
21

22
    def forward(self, states):
23
        # logits 交给 Categorical，内部稳定地计算归一化概率。
24
        return Categorical(logits=self.actor(states)), self.critic(states).squeeze(-1)
25

26

27
def gae(rewards, values, next_values, terminals, boundaries, gamma=0.99, lam=0.95):
28
    """两个掩码：TD 在真实终止停止；GAE 在终止或截断停止跨回合传播。"""
29
    advantage = np.zeros(len(rewards), dtype=np.float32)
30
    # 从采样块末端反向递推；末步的 delta 已通过 next_values 保留 bootstrap。
31
    carry = 0.0
32
    for t in reversed(range(len(rewards))):
33
        # 真实终止将下一状态价值置零；纯时间截断仍使用其价值。
34
        delta = rewards[t] + gamma * (1.0 - terminals[t]) * next_values[t] - values[t]
35
        # lam 调节多步误差权重；边界掩码防止传播到 reset 后的新回合。
36
        carry = delta + gamma * lam * (1.0 - boundaries[t]) * carry
37
        advantage[t] = carry
38
    # 用未归一化优势加旧价值估计，构造 Critic 的固定回归目标。
39
    return advantage, advantage + np.asarray(values, dtype=np.float32)
40

41

42
def evaluate(model, device, seed, episodes=20):
43
    env = gym.make("CartPole-v1")
44
    scores = []
45
    with torch.no_grad():
46
        for episode in range(episodes):
47
            obs, _ = env.reset(seed=seed + episode)
48
            score = 0.0
49
            while True:
50
                x = torch.as_tensor(obs, dtype=torch.float32, device=device).unsqueeze(0)
51
                distribution, _ = model(x)
52
                # 评估选择最高概率动作；训练时则从分布采样以保留探索。
53
                action = distribution.probs.argmax(dim=1).item()
54
                obs, reward, terminated, truncated, _ = env.step(action)
55
                score += reward
56
                if terminated or truncated:
57
                    break
58
            scores.append(score)
59
    env.close()
60
    return {"mean": float(np.mean(scores)), "std": float(np.std(scores)), "scores": scores}
61

62

63
def main(args):
64
    if args.steps <= 0:
65
        raise ValueError("steps 必须为正整数")
66
    np.random.seed(args.seed)
67
    torch.manual_seed(args.seed)
68
    torch.set_num_threads(1)
69
    device = torch.device(args.device)
70
    env = gym.make("CartPole-v1")
71
    obs, _ = env.reset(seed=args.seed)
72
    model = ActorCritic().to(device)
73
    optimizer = torch.optim.Adam(model.parameters(), lr=3e-4)
74
    # 一轮最多采集 1024 步，每批约 64 条；同一轮轨迹最多复用 10 遍。
75
    rollout_size, minibatch, epochs = 1024, 64, 10
76
    clip, target_kl = 0.2, 0.02
77
    total_steps, episode_return = 0, 0.0
78
    history, updates = [], []
79
    while total_steps < args.steps:
80
        # 每轮建立新 buffer；更新结束后丢弃，不能充当长期经验回放。
81
        states, actions, old_logs = [], [], []
82
        rewards, values, next_values, terminals, boundaries = [], [], [], [], []
83
        for _ in range(min(rollout_size, args.steps - total_steps)):
84
            x = torch.as_tensor(obs, dtype=torch.float32, device=device).unsqueeze(0)
85
            # 采样阶段不建梯度图；策略在整轮采样期间保持不变。
86
            with torch.no_grad():
87
                distribution, value = model(x)
88
                action = distribution.sample()
89
                log_prob = distribution.log_prob(action)
90
            next_obs, reward, terminated, truncated, _ = env.step(action.item())
91
            with torch.no_grad():
92
                # 必须在 reset 前评估下一状态；超时仍保留下一状态价值。
93
                _, next_value = model(torch.as_tensor(next_obs, dtype=torch.float32,
94
                                                     device=device).unsqueeze(0))
95
            states.append(obs.copy())
96
            actions.append(action.item())
97
            # 保存采样时的旧 log_prob；本轮优化时始终作为固定的比率分母。
98
            old_logs.append(log_prob.item())
99
            rewards.append(reward)
100
            values.append(value.item())
101
            next_values.append(next_value.item())
102
            terminals.append(float(terminated))
103
            boundaries.append(float(terminated or truncated))
104
            total_steps += 1
105
            episode_return += reward
106
            obs = next_obs
107
            if terminated or truncated:
108
                history.append(episode_return)
109
                episode_return = 0.0
110
                obs, _ = env.reset()
111
        adv, returns = gae(rewards, values, next_values, terminals, boundaries)
112
        # 先算价值回归目标，再归一化优势，不能用归一化优势构造 returns。
113
        adv = (adv - adv.mean()) / (adv.std() + 1e-8)
114
        s = torch.as_tensor(np.array(states), dtype=torch.float32, device=device)
115
        a = torch.as_tensor(actions, dtype=torch.int64, device=device)
116
        old_log = torch.as_tensor(old_logs, dtype=torch.float32, device=device)
117
        advantages = torch.as_tensor(adv, device=device)
118
        targets = torch.as_tensor(returns, device=device)
119
        stop = False
120
        # 每遍重新打乱样本顺序；优势、价值目标与旧 log_prob 都不重算。
121
        for _ in range(epochs):
122
            for indices in np.array_split(np.random.permutation(len(states)),
123
                                           max(1, (len(states) + minibatch - 1) // minibatch)):
124
                idx = torch.as_tensor(indices, dtype=torch.int64, device=device)
125
                distribution, predicted_values = model(s[idx])
126
                # 用对数概率差计算新旧概率比，避免直接相除小概率。
127
                log_ratio = distribution.log_prob(a[idx]) - old_log[idx]
128
                ratio = log_ratio.exp()
129
                with torch.no_grad():
130
                    # 当前小批量上估计旧策略到新策略的 KL，用于提前停止。
131
                    approximate_kl = ((ratio - 1) - log_ratio).mean().item()
132
                if approximate_kl > 1.5 * target_kl:
133
                    stop = True  # 辅助保护；clip 本身不能严格保证 KL 上界。
134
                    break
135
                unclipped = ratio * advantages[idx]
136
                clipped = ratio.clamp(1 - clip, 1 + clip) * advantages[idx]
137
                # 优化器做最小化；加负号才对应最大化 PPO 的裁剪目标。
138
                policy_loss = -torch.minimum(unclipped, clipped).mean()
139
                value_loss = (predicted_values - targets[idx]).square().mean()
140
                entropy = distribution.entropy().mean()
141
                # 价值误差要最小化；熵项取负，鼓励策略保留适当探索。
142
                loss = policy_loss + 0.5 * value_loss - 0.01 * entropy
143
                optimizer.zero_grad()
144
                loss.backward()
145
                nn.utils.clip_grad_norm_(model.parameters(), 0.5)
146
                optimizer.step()
147
            # KL 超阈值时同时退出剩余 epochs，然后收集新策略的数据。
148
            if stop:
149
                break
150
        updates.append({"steps": total_steps, "approx_kl": approximate_kl,
151
                        "early_stop": stop})
152
        if len(updates) % 10 == 0:
153
            print(f"step={total_steps} train_mean={np.mean(history[-20:]):.1f} kl={approximate_kl:.4f}")
154
    result = evaluate(model, device, args.seed + 10000)
155
    args.output.mkdir(parents=True, exist_ok=True)
156
    # 保存 Actor/Critic 权重与下方实验配置；此文件不含续训所需的优化器状态。
157
    torch.save(model.state_dict(), args.output / "actor_critic.pt")
158
    (args.output / "metrics.json").write_text(json.dumps({
159
        "algorithm": "PPO-Clip", "seed": args.seed, "env": "CartPole-v1",
160
        "steps": total_steps, "device": str(device), "torch": torch.__version__,
161
        "gymnasium": gym.__version__, "config": {"lr": 3e-4, "gamma": 0.99,
162
        "gae_lambda": 0.95, "rollout": rollout_size, "batch": minibatch,
163
        "epochs": epochs, "clip": clip, "target_kl": target_kl,
164
        "vf_coef": 0.5, "ent_coef": 0.01, "max_grad_norm": 0.5},
165
        "train_returns": history, "updates": updates, "evaluation": result},
166
        indent=2), encoding="utf-8")
167
    env.close()
168
    print(result)
169

170

171
if __name__ == "__main__":
172
    parser = argparse.ArgumentParser()
173
    parser.add_argument("--steps", type=int, default=100000)
174
    parser.add_argument("--seed", type=int, default=42)
175
    parser.add_argument("--device", default="cpu")
176
    parser.add_argument("--output", type=Path, default=Path("runs/ppo"))
177
    main(parser.parse_args())

1
import gymnasium as gym
2
from stable_baselines3 import PPO
3
from stable_baselines3.common.evaluation import evaluate_policy
4

5
# 显式创建训练环境，记录模型和环境版本。
6
train_env = gym.make("CartPole-v1")
7
# MlpPolicy 适用于向量观测；下列参数与手写 PPO 的关键配置保持可比较。
8
model = PPO("MlpPolicy", train_env, seed=42, device="cpu",
9
            learning_rate=3e-4, n_steps=1024, batch_size=64,
10
            n_epochs=10, gamma=0.99, gae_lambda=0.95,
11
            clip_range=0.2, ent_coef=0.01, target_kl=0.02, verbose=1)
12
# total_timesteps 是环境交互预算；库可能为完成最后一批轨迹而略超预算。
13
model.learn(total_timesteps=100000)
14
# 独立环境评估，避免评估改变训练环境的状态。
15
eval_env = gym.make("CartPole-v1")
16
eval_env.reset(seed=10042)
17
# deterministic=True 关闭动作采样；std 是回合间波动，不是训练种子间波动。
18
mean, std = evaluate_policy(model, eval_env, n_eval_episodes=20,
19
                            deterministic=True, warn=False)
20
print(f"evaluation: {mean:.1f} ± {std:.1f}")
21
model.save("ppo_cartpole_sb3")
22
eval_env.close()
23
train_env.close()