'''demo:
import torch.nn as nn
import torch.nn.functional as F
class Model(nn.Module):
def __init__(self):
super().__init__()
self.conv1 = nn.Conv2d(1, 20, 5)
self.conv2 = nn.Conv2d(20, 20, 5)
def forward(self, x):
x = F.relu(self.conv1(x))
return F.relu(self.conv2(x)
'''
class FCN(nn.Module):
def __init__(self, states, actions, hidden_dim=128):
super(FCN, self).__init__() ##
self.fc1 = nn.Linear(states, hidden_dim)
self.fc2 = nn.Linear(hidden_dim, hidden_dim)
self.fc3 = nn.Linear(hidden_dim, actions)
def forward(self, x):
x = F.relu(self.fc1(x))
x = F.relu(self.fc2(x))
return self.fc3(x)
经验回放
class replybuffer:
def __init__(self, capacity):
self.capacity = capacity # 容量
self.buffer = [] # 缓冲区，队列
self.position = 0 # 列表中的位置
def store(self, state, action, reward, next_action, done): # 存储
if len(self.buffer) < self.capacity: # 容量超出 去掉开始存入的转移？？
self.buffer.append(None)
self.buffer[self.position] = (state, action, reward, next_action, done) # （）元组
self.position = (self.position + 1) % self.capacity # ?? %取模运算，5%2=1
''' #zip() 函数:用于将可迭代的对象作为参数，将对象中对应的元素打包成一个个元组，
然后返然后返回由这些元组组成的对象. # 如果各个迭代器的元素个数不一致，则返回列表长度
利用 * 号操作符，可以将元组解压为列表。'''
def sample(self, batch_size): # 采样
batch = random.sample(self.buffer, batch_size)
state, action, reward, next_action, done = zip(*batch) # 解压
return state, action, reward, next_action, done
def __len__(self):
return len(self.buffer)
DQN
class DQN:
def __init__(self, state_n, action_n, cfg):
self.actions = action_n
self.epsilon = 0
self.sample_count = 0
self.epsilon_start = cfg.epsilon_start
self.epsilon_end = cfg.epsilon_end
self.epsilon_delay = cfg.epsilon_delay
self.gamma = cfg.gamma # 折扣因子
self.device = cfg.device # 设备 cpu或者gpu 为什么要有？？
self.policy_net = FCN(state_n, action_n).to(self.device) #
self.target_net = FCN(state_n, action_n).to(self.device)
for target_param, parma in zip(self.target_net.parameters(), self.policy_net.parameters()): # 目标网络的参数是从更新网络copy
target_param.data.copy_(parma.data) # !!dir()方法查看类的所有属性 ,copy_ ,没有copy
self.batch_size = cfg.batch_size
self.memory = replybuffer(cfg.memory_capacity) # 经验回放
self.optimizer = optim.Adam(self.policy_net.parameters(), lr=cfg.lr)
# self.epsilon =lambda sample_count: cfg.epsilon_end + (cfg.epsilon_start - cfg.epsilon_end) *
# math.exp(-1 * sample_count / cfg.epsilon_delay)
def choose_action(self, state):
self.sample_count += 1
self.epsilon = self.epsilon_end + (self.epsilon_start - self.epsilon_end) *
math.exp(-1 * self.sample_count / self.epsilon_delay)
#if np.random.uniform(0, 1) > self.epsilon: ##
if random.random() > self.epsilon:
with torch.no_grad(): # 无梯度
state = torch.tensor(state, device=self.device, dtype=torch.float32).unsqueeze(dim=0) # ??
#print('动作选择时的状态', state)
q_values = self.policy_net(state)
action = q_values.max(1)[1].item()
# action = np.argmax() # 区别Q-learning 不是用的Q-table
else:
# action = np.random.choice(self.actions) # 对比一下两个区别
action = random.randrange(self.actions)
return action
def update(self):
if len(self.memory) < self.batch_size: # 当经验值不够一个批量的时候，就不更新策略
return
state_batch, action_batch, reward_batch, next_state_batch, done_batch
= self.memory.sample(self.batch_size) # 经验回放采样的一批buffer
state_batch = torch.tensor(np.array(state_batch), device=self.device, dtype=torch.float)
action_batch = torch.tensor(action_batch, device=self.device).unsqueeze(1)
reward_batch = torch.tensor(reward_batch, device=self.device, dtype=torch.float)
next_state_batch = torch.tensor(np.array(next_state_batch), device=self.device, dtype=torch.float)
done_batch = torch.tensor(np.float32(done_batch), device=self.device)
q_values = self.policy_net(state_batch).gather(dim=1, index=action_batch)
next_q_values = self.target_net(next_state_batch).max(1)[0].detach()
# if done_batch:
# expected_q_values = reward_batch
# else:
# expected_q_values = reward_batch + self.gamma * next_q_values
expected_q_values = reward_batch + self.gamma * next_q_values*(1-done_batch)
loss = nn.MSELoss()(q_values, expected_q_values.unsqueeze(1)) # 均方根误差
# 优化更新模型
self.optimizer.zero_grad()
loss.backward()
for param in self.policy_net.parameters(): # clip防止梯度爆炸
param.grad.data.clamp_(-1, 1)
self.optimizer.step()
训练（）
参数设置
class Config:
def __init__(self):
self.lr = 0.0001
self.gamma = 0.95
self.epsilon_start = 0.90
self.epsilon_end = 0.01
self.epsilon_delay = 500
self.batch_size = 64
self.memory_capacity = 100000 ## ^ 不是次方， 是按位或 运算符 10^3 =9
self.train_episode = 200
self.target_update = 4
self.device = 'cpu'
self.hidden_dim = 256
与环境交互（差不多一样的部分）
def play_environment(agent, env, cfg ,render = False): # 交互是 环境env和智能体 agent
print('training!')
rewards = []
steps=[]
for i_ep in range(cfg.train_episode):
ep_reward = 0
ep_step = 0
state = env.reset()
while True:
if render:
env.render()
ep_step += 1
action = agent.choose_action(state)
next_state, reward, done, _ = env.step(action)
agent.memory.store(state, action, reward, next_state, done)
state = next_state
agent.update()
ep_reward += reward
if done:
break
if (i_ep + 1) % cfg.target_update == 0: ## 更新目标网络!!!! ##
agent.target_net.load_state_dict(agent.policy_net.state_dict()) ##
rewards.append(ep_reward)
steps.append(ep_step)
print("回合数：{}/{}，奖励{:.1f},步数{:.1f}".format(i_ep + 1, cfg.train_episode, ep_reward, ep_step))
# env.close()
return rewards
主函数（环境、智能体创建、运行）
env = gym.make('CartPole-v0')
states = env.observation_space.shape[0] ## shape[0] 读取行数 'Box' object has no attribute 'n'
actions = env.action_space.n
cfg = Config()
agent = DQN(states, actions, cfg)
env.seed(10)
rewards = play_environment(agent, env, cfg , render=True) ## render=True 可视化界面
print()
env.close()
plt.plot(rewards)
plt.show()
结果
占坑（1.改进网络CNN，2.DQN变种）
