107 KiB
107 KiB
In [19]:
from collections import defaultdict
import numpy as np
class FisrtVisitMC:
''' On-Policy First-Visit MC Control
'''
def __init__(self,cfg):
self.n_actions = cfg.n_actions
self.epsilon = cfg.epsilon
self.gamma = cfg.gamma
self.Q_table = defaultdict(lambda: np.zeros(cfg.n_actions))
self.returns_sum = defaultdict(float) # 保存return之和
self.returns_count = defaultdict(float)
def sample_action(self,state):
state = str(state)
if np.random.uniform(0, 1) > self.epsilon:
action = np.argmax(self.Q_table[str(state)]) # 选择Q(s,a)最大对应的动作
else:
action = np.random.choice(self.n_actions) # 随机选择动作
return action
# if state in self.Q_table.keys():
# best_action = np.argmax(self.Q_table[state])
# action_probs = np.ones(self.n_actions, dtype=float) * self.epsilon / self.n_actions
# action_probs[best_action] += (1.0 - self.epsilon)
# action = np.random.choice(np.arange(len(action_probs)), p=action_probs)
# else:
# action = np.random.randint(0,self.n_actions)
# return action
def predict_action(self,state):
state = str(state)
state = str(state)
if np.random.uniform(0, 1) > self.epsilon:
action = np.argmax(self.Q_table[str(state)]) # 选择Q(s,a)最大对应的动作
else:
action = np.random.choice(self.n_actions) # 随机选择动作
return action
# if state in self.Q_table.keys():
# best_action = np.argmax(self.Q_table[state])
# action_probs = np.ones(self.n_actions, dtype=float) * self.epsilon / self.n_actions
# action_probs[best_action] += (1.0 - self.epsilon)
# action = np.argmax(self.Q_table[state])
# else:
# action = np.random.randint(0,self.n_actions)
# return action
def update(self,one_ep_transition):
# Find all (state, action) pairs we've visited in this one_ep_transition
# We convert each state to a tuple so that we can use it as a dict key
sa_in_episode = set([(str(x[0]), x[1]) for x in one_ep_transition])
for state, action in sa_in_episode:
sa_pair = (state, action)
# Find the first occurence of the (state, action) pair in the one_ep_transition
first_occurence_idx = next(i for i,x in enumerate(one_ep_transition)
if str(x[0]) == state and x[1] == action)
# Sum up all rewards since the first occurance
G = sum([x[2]*(self.gamma**i) for i,x in enumerate(one_ep_transition[first_occurence_idx:])])
# Calculate average return for this state over all sampled episodes
self.returns_sum[sa_pair] += G
self.returns_count[sa_pair] += 1.0
self.Q_table[state][action] = self.returns_sum[sa_pair] / self.returns_count[sa_pair]In [20]:
def train(cfg,env,agent):
print('开始训练！')
print(f'环境:{cfg.env_name}, 算法:{cfg.algo_name}, 设备:{cfg.device}')
rewards = [] # 记录奖励
for i_ep in range(cfg.train_eps):
ep_reward = 0 # 记录每个回合的奖励
one_ep_transition = []
state = env.reset(seed=cfg.seed) # 重置环境,即开始新的回合
for _ in range(cfg.max_steps):
action = agent.sample_action(state) # 根据算法采样一个动作
next_state, reward, terminated, info = env.step(action) # 与环境进行一次动作交互
one_ep_transition.append((state, action, reward)) # 保存transitions
agent.update(one_ep_transition) # 更新智能体
state = next_state # 更新状态
ep_reward += reward
if terminated:
break
rewards.append(ep_reward)
if (i_ep+1)%10==0:
print(f"回合：{i_ep+1}/{cfg.train_eps}，奖励：{ep_reward:.1f}")
print('完成训练！')
return {"rewards":rewards}
def test(cfg,env,agent):
print('开始测试！')
print(f'环境：{cfg.env_name}, 算法：{cfg.algo_name}, 设备：{cfg.device}')
rewards = [] # 记录所有回合的奖励
for i_ep in range(cfg.test_eps):
ep_reward = 0 # 记录每个episode的reward
state = env.reset(seed=cfg.seed) # 重置环境, 重新开一局（即开始新的一个回合）
for _ in range(cfg.max_steps):
action = agent.predict_action(state) # 根据算法选择一个动作
next_state, reward, terminated, info = env.step(action) # 与环境进行一个交互
state = next_state # 更新状态
ep_reward += reward
if terminated:
break
rewards.append(ep_reward)
print(f"回合数：{i_ep+1}/{cfg.test_eps}, 奖励：{ep_reward:.1f}")
print('完成测试！')
return {"rewards":rewards}In [21]:
import gym
import turtle
import numpy as np
# turtle tutorial : https://docs.python.org/3.3/library/turtle.html
class CliffWalkingWapper(gym.Wrapper):
def __init__(self, env):
gym.Wrapper.__init__(self, env)
self.t = None
self.unit = 50
self.max_x = 12
self.max_y = 4
def draw_x_line(self, y, x0, x1, color='gray'):
assert x1 > x0
self.t.color(color)
self.t.setheading(0)
self.t.up()
self.t.goto(x0, y)
self.t.down()
self.t.forward(x1 - x0)
def draw_y_line(self, x, y0, y1, color='gray'):
assert y1 > y0
self.t.color(color)
self.t.setheading(90)
self.t.up()
self.t.goto(x, y0)
self.t.down()
self.t.forward(y1 - y0)
def draw_box(self, x, y, fillcolor='', line_color='gray'):
self.t.up()
self.t.goto(x * self.unit, y * self.unit)
self.t.color(line_color)
self.t.fillcolor(fillcolor)
self.t.setheading(90)
self.t.down()
self.t.begin_fill()
for i in range(4):
self.t.forward(self.unit)
self.t.right(90)
self.t.end_fill()
def move_player(self, x, y):
self.t.up()
self.t.setheading(90)
self.t.fillcolor('red')
self.t.goto((x + 0.5) * self.unit, (y + 0.5) * self.unit)
def render(self):
if self.t == None:
self.t = turtle.Turtle()
self.wn = turtle.Screen()
self.wn.setup(self.unit * self.max_x + 100,
self.unit * self.max_y + 100)
self.wn.setworldcoordinates(0, 0, self.unit * self.max_x,
self.unit * self.max_y)
self.t.shape('circle')
self.t.width(2)
self.t.speed(0)
self.t.color('gray')
for _ in range(2):
self.t.forward(self.max_x * self.unit)
self.t.left(90)
self.t.forward(self.max_y * self.unit)
self.t.left(90)
for i in range(1, self.max_y):
self.draw_x_line(
y=i * self.unit, x0=0, x1=self.max_x * self.unit)
for i in range(1, self.max_x):
self.draw_y_line(
x=i * self.unit, y0=0, y1=self.max_y * self.unit)
for i in range(1, self.max_x - 1):
self.draw_box(i, 0, 'black')
self.draw_box(self.max_x - 1, 0, 'yellow')
self.t.shape('turtle')
x_pos = self.s % self.max_x
y_pos = self.max_y - 1 - int(self.s / self.max_x)
self.move_player(x_pos, y_pos)In [22]:
import gym
import os
def all_seed(env,seed = 1):
''' omnipotent seed for RL, attention the position of seed function, you'd better put it just following the env create function
Args:
env (_type_):
seed (int, optional): _description_. Defaults to 1.
'''
import torch
import numpy as np
import random
# print(f"seed = {seed}")
env.seed(seed) # env config
np.random.seed(seed)
random.seed(seed)
torch.manual_seed(seed) # config for CPU
torch.cuda.manual_seed(seed) # config for GPU
os.environ['PYTHONHASHSEED'] = str(seed) # config for python scripts
# config for cudnn
torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False
torch.backends.cudnn.enabled = False
def env_agent_config(cfg):
'''创建环境和智能体
'''
env = gym.make(cfg.env_name,new_step_api=True) # 创建环境
env = CliffWalkingWapper(env)
if cfg.seed !=0: # set random seed
all_seed(env,seed=cfg.seed)
try: # 状态维度
n_states = env.observation_space.n # print(hasattr(env.observation_space, 'n'))
except AttributeError:
n_states = env.observation_space.shape[0]
n_actions = env.action_space.n # 动作维度
setattr(cfg, 'n_states', n_states) # 将状态维度添加到配置参数中
setattr(cfg, 'n_actions', n_actions) # 将动作维度添加到配置参数中
agent = FisrtVisitMC(cfg)
return env,agentIn [23]:
import torch
import matplotlib.pyplot as plt
import seaborn as sns
class Config:
'''配置参数
'''
def __init__(self):
self.env_name = 'CliffWalking-v0' # 环境名称
self.algo_name = "FirstVisitMC" # 算法名称
self.train_eps = 400 # 训练回合数
self.test_eps = 20 # 测试回合数
self.max_steps = 200 # 每个回合最大步数
self.epsilon = 0.1 # 贪婪度
self.gamma = 0.9 # 折扣因子
self.lr = 0.5 # 学习率
self.seed = 1 # 随机种子
# if torch.cuda.is_available(): # 是否使用GPUs
# self.device = torch.device('cuda')
# else:
# self.device = torch.device('cpu')
self.device = torch.device('cpu')
def smooth(data, weight=0.9):
'''用于平滑曲线
'''
last = data[0] # First value in the plot (first timestep)
smoothed = list()
for point in data:
smoothed_val = last * weight + (1 - weight) * point # 计算平滑值
smoothed.append(smoothed_val)
last = smoothed_val
return smoothed
def plot_rewards(rewards,title="learning curve"):
sns.set()
plt.figure() # 创建一个图形实例，方便同时多画几个图
plt.title(f"{title}")
plt.xlim(0, len(rewards), 10) # 设置x轴的范围
plt.xlabel('epsiodes')
plt.plot(rewards, label='rewards')
plt.plot(smooth(rewards), label='smoothed')
plt.legend()In [24]:
# 获取参数
cfg = Config()
# 训练
env, agent = env_agent_config(cfg)
res_dic = train(cfg, env, agent)
plot_rewards(res_dic['rewards'], title=f"training curve on {cfg.device} of {cfg.algo_name} for {cfg.env_name}")
# 测试
res_dic = test(cfg, env, agent)
plot_rewards(res_dic['rewards'], title=f"testing curve on {cfg.device} of {cfg.algo_name} for {cfg.env_name}") # 画出结果开始训练！ 环境:CliffWalking-v0, 算法:FirstVisitMC, 设备:cpu 回合：10/400，奖励：-200.0 回合：20/400，奖励：-200.0 回合：30/400，奖励：-200.0 回合：40/400，奖励：-200.0 回合：50/400，奖励：-200.0 回合：60/400，奖励：-200.0 回合：70/400，奖励：-200.0 回合：80/400，奖励：-200.0 回合：90/400，奖励：-200.0 回合：100/400，奖励：-200.0 回合：110/400，奖励：-200.0 回合：120/400，奖励：-200.0 回合：130/400，奖励：-200.0 回合：140/400，奖励：-200.0 回合：150/400，奖励：-200.0 回合：160/400，奖励：-200.0 回合：170/400，奖励：-200.0 回合：180/400，奖励：-200.0 回合：190/400，奖励：-200.0 回合：200/400，奖励：-200.0 回合：210/400，奖励：-200.0 回合：220/400，奖励：-200.0 回合：230/400，奖励：-200.0 回合：240/400，奖励：-200.0 回合：250/400，奖励：-200.0 回合：260/400，奖励：-200.0 回合：270/400，奖励：-299.0 回合：280/400，奖励：-200.0 回合：290/400，奖励：-200.0 回合：300/400，奖励：-200.0 回合：310/400，奖励：-200.0 回合：320/400，奖励：-200.0 回合：330/400，奖励：-200.0 回合：340/400，奖励：-200.0 回合：350/400，奖励：-200.0 回合：360/400，奖励：-200.0 回合：370/400，奖励：-200.0 回合：380/400，奖励：-200.0 回合：390/400，奖励：-200.0 回合：400/400，奖励：-200.0 完成训练！ 开始测试！ 环境：CliffWalking-v0, 算法：FirstVisitMC, 设备：cpu 回合数：1/20, 奖励：-200.0 回合数：2/20, 奖励：-200.0 回合数：3/20, 奖励：-200.0 回合数：4/20, 奖励：-200.0 回合数：5/20, 奖励：-200.0 回合数：6/20, 奖励：-200.0 回合数：7/20, 奖励：-200.0 回合数：8/20, 奖励：-200.0 回合数：9/20, 奖励：-200.0 回合数：10/20, 奖励：-299.0 回合数：11/20, 奖励：-200.0 回合数：12/20, 奖励：-200.0 回合数：13/20, 奖励：-200.0 回合数：14/20, 奖励：-200.0 回合数：15/20, 奖励：-200.0 回合数：16/20, 奖励：-200.0 回合数：17/20, 奖励：-200.0 回合数：18/20, 奖励：-200.0 回合数：19/20, 奖励：-200.0 回合数：20/20, 奖励：-200.0 完成测试！
