import gymnasium as gym
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
import json
class ActorCritic(nn.Module):
def __init__(self, sd, ad, h=128):
super().__init__()
self.shared = nn.Sequential(nn.Linear(sd, h), nn.ReLU())
self.actor = nn.Sequential(nn.Linear(h, h//2), nn.ReLU(), nn.Linear(h//2, ad), nn.Softmax(dim=-1))
self.critic = nn.Sequential(nn.Linear(h, h//2), nn.ReLU(), nn.Linear(h//2, 1))
def forward(self, x):
shared = self.shared(x)
return self.actor(shared), self.critic(shared)
def train_ac(env, n_episodes=600, gamma=0.99, lr=1e-3, entropy_coef=0.01):
sd = env.observation_space.shape[0]; ad = env.action_space.n
model = ActorCritic(sd, ad)
opt = optim.Adam(model.parameters(), lr=lr)
history = []
for ep in range(n_episodes):
s, _ = env.reset(); done = False; total = 0
values = []; log_probs = []; rewards = []; entropies = []
while not done:
s_t = torch.FloatTensor(s)
probs, v = model(s_t)
dist = torch.distributions.Categorical(probs)
a = dist.sample()
ns, r, t, tr, _ = env.step(a.item())
log_probs.append(dist.log_prob(a))
values.append(v.squeeze())
rewards.append(r)
entropies.append(dist.entropy())
s = ns; total += r; done = t or tr
# 计算损失
R = 0; returns = []
for r in reversed(rewards):
R = r + gamma * R; returns.insert(0, R)
returns = torch.FloatTensor(returns)
returns = (returns - returns.mean()) / (returns.std() + 1e-8)
policy_loss = 0; value_loss = 0; entropy_loss = 0
for i in range(len(rewards)):
advantage = returns[i] - values[i].detach()
policy_loss -= log_probs[i] * advantage
value_loss += nn.MSELoss()(values[i], returns[i])
entropy_loss -= entropies[i]
loss = policy_loss + 0.5 * value_loss - entropy_coef * entropy_loss
opt.zero_grad(); loss.backward(); opt.step()
history.append(total)
if (ep+1) % 100 == 0: print(f"AC Ep{ep+1}: avg={np.mean(history[-100:]):.1f}")
return model, history
env = gym.make('CartPole-v1')
print("=== Actor-Critic 训练 ===")
model, rewards = train_ac(env, n_episodes=500)
# 测试
test_r = []
for ep in range(100):
s, _ = env.reset(seed=ep+8000); done = False; total = 0
while not done:
with torch.no_grad():
probs, _ = model(torch.FloatTensor(s))
a = probs.argmax().item()
s, r, t, tr, _ = env.step(a); total += r; done = t or tr
test_r.append(total)
w = 50
smooth = [np.mean(rewards[max(0,i-w):i+1]) for i in range(len(rewards))]
print(f"\\n训练最终50回合: {np.mean(rewards[-50:]):.1f}")
print(f"测试平均: {np.mean(test_r):.1f}")
# 分析Actor和Critic输出
s, _ = env.reset()
with torch.no_grad():
probs, v = model(torch.FloatTensor(s))
print(f"初始状态: 动作概率={[round(p,3) for p in probs.numpy()]}, 状态值={v.item():.3f}")
result = {
"train_final": round(float(np.mean(rewards[-50:])),1),
"test_avg": round(float(np.mean(test_r)),1),
"smooth": [round(v,1) for v in smooth[::50]],
"initial_probs": [round(float(p),3) for p in probs],
"initial_value": round(float(v),3)
}
with open("/var/www/ttl/rl/lesson20_result.json", "w") as f:
json.dump(result, f)
print("✅验证通过 - Actor-Critic有效结合策略梯度和值函数学习")
env.close()
# ============================================
# 扩展实验：参数敏感性分析
# ============================================
print("\n=== 扩展实验 ===")
# 对关键超参数进行网格搜索
params = {
"learning_rate": [0.001, 0.01, 0.1],
"epsilon": [0.05, 0.1, 0.2],
"gamma": [0.9, 0.95, 0.99]
}
print("超参数搜索空间:")
for k, v in params.items():
print(f" {k}: {v}")
print("共{}种组合".format(1))
for k, v in params.items():
print(f" {k}: {len(v)}种选择")
total = 1
for k, v in params.items():
total *= len(v)
print(f"总计: {total}种超参数组合")
print("扩展实验框架验证成功 - ✅")
📝 算法伪代码：Actor-Critic
Actor-Critic核心步骤:
1. 初始化参数/网络
2. FOR episode = 1 TO N:
3. 初始化环境状态 s
4. WHILE NOT done:
5. 根据当前策略选择动作 a
6. 执行动作, 观察奖励 r 和新状态 s'
7. 存储经验 (s, a, r, s')
8. 采样mini-batch更新参数
9. s = s'
10. END WHILE
11. 更新探索率/目标网络(如适用)
12. END FOR
13. RETURN 训练好的策略/值函数