強化学習ライブラリtianshou——DQN使用
61181 ワード
強化学習ライブラリtianshou——DQN使用
Tianshouは清華大学の学生がオープンソースで編纂した強化学習庫である.本人は試合の都合で強化学習に使用していたが、緊張しすぎて強化学習のコードを迅速に再現しようとしなかったため、良好な成績を得られなかったため、ライブラリで迅速な再現を試みた.
前にもparlなどのライブラリを試してみましたが、parlはドキュメントなどではtianshouに及ばないような気がして、性能的には菜鳥として評価しにくいです.Tianshouの公式ドキュメントも久しぶりに更新され、上には実行できないコードもあり、最新版tianshouのgithubのコードケースを使って学習し、関連注釈が記録されています.
上はすべてスクリプト式の動作で、パラメータをargsの中で相対的に便利に定義して私たちがパラメータの修正を行うことができて、論理を修正する時、比較的に複雑で、時間が十分な情況の下で、対象に向かう思想を使うことを考慮することができます.
問題を問題クラスとアルゴリズムクラスに分解し、問題クラスは問題の記述に集中し、問題パラメータを提供し、アルゴリズムクラスはアルゴリズムの記述に集中し、アルゴリズムパラメータを提供する.
その後,本問題におけるアルゴリズムクラスの一部の属性が問題クラスの属性に依存することを考慮して,アルゴリズムクラスに直接問題クラスのインスタンスを転送することができる.この方法は考えやすい.ここでは、3番目のクラスを提供し、3番目のクラスは問題クラスとアルゴリズムクラスを調整するために使用され、3番目のクラスで呼び出し解決とパラメータインタラクションが行われます.どちらがいいかは説明できませんが、自分の試みとしか言えません.
ここで、policyは、探索ポリシーの実装を含むポリシーを記述するために使用され、dqnは確率的貪欲ポリシーであり、maskパラメータは動作を遮蔽することができる.policyがforwardメソッドを実装する際に探索戦略の実装を行い,予測値と探索動作を返す.
再現のために、シード設定は環境構築後、netなどの他のパラメータの前に置く必要があります.netは初期化時にnpの乱数を呼び出し、前に置くと再現が保証されます.この点は長い間調べていましたが...
更新の策略については、理解できませんでした.ドキュメントの評価を見ると、tianshouはすべての強化学習ライブラリよりも速いが、機能的にはまだ十分ではなく、マルチスマートなどのアルゴリズムが実現していないため、rayに転向することを考えているかもしれない.rayは分布式の枠組みとして、sparkとmllibが私を潰す日子を思い出させた.
Tianshouは清華大学の学生がオープンソースで編纂した強化学習庫である.本人は試合の都合で強化学習に使用していたが、緊張しすぎて強化学習のコードを迅速に再現しようとしなかったため、良好な成績を得られなかったため、ライブラリで迅速な再現を試みた.
前にもparlなどのライブラリを試してみましたが、parlはドキュメントなどではtianshouに及ばないような気がして、性能的には菜鳥として評価しにくいです.Tianshouの公式ドキュメントも久しぶりに更新され、上には実行できないコードもあり、最新版tianshouのgithubのコードケースを使って学習し、関連注釈が記録されています.
import os
import gym
import torch
import pickle
import pprint
import argparse
import numpy as np
from torch.utils.tensorboard import SummaryWriter
from tianshou.policy import DQNPolicy
from tianshou.env import DummyVectorEnv
from tianshou.utils.net.common import Net
from tianshou.trainer import offpolicy_trainer
from tianshou.data import Collector, ReplayBuffer, PrioritizedReplayBuffer
def get_args():
'''
max_epoch： ， （ stop_fn ）
step_per_epoch： epoch
collect_per_step： 。 ， 10
episode_per_test： rollout
batch_size：
train_fn： epoch ， epoch env step 。 ， epsilon 0.1
test_fn： epoch ， epoch env step 。 ， epsilon 0.05
stop_fn： ， （the average undiscounted returns），
writer： TensorBoard， ：
:return:
'''
parser = argparse.ArgumentParser()
parser.add_argument('--task', type=str, default='CartPole-v0') #
parser.add_argument('--seed', type=int, default=1626) #
parser.add_argument('--eps-test', type=float, default=0.05) #
parser.add_argument('--eps-train', type=float, default=0.1) #
parser.add_argument('--buffer-size', type=int, default=20000) #
parser.add_argument('--lr', type=float, default=1e-3) #
parser.add_argument('--gamma', type=float, default=0.9) #
parser.add_argument('--n-step', type=int, default=3) #
parser.add_argument('--target-update-freq', type=int, default=320) # ， freq ，0
parser.add_argument('--epoch', type=int, default=10) #
parser.add_argument('--step-per-epoch', type=int, default=1000) #
parser.add_argument('--collect-per-step', type=int, default=10) #
parser.add_argument('--batch-size', type=int, default=64) #
parser.add_argument('--hidden-sizes', type=int,
nargs='*', default=[128, 128, 128, 128]) #
parser.add_argument('--training-num', type=int, default=8) #
parser.add_argument('--test-num', type=int, default=100) #
parser.add_argument('--logdir', type=str, default='log')
parser.add_argument('--render', type=float, default=0.)
parser.add_argument('--prioritized-replay',
action="store_true", default=False) #
parser.add_argument('--alpha', type=float, default=0.6) # ，
parser.add_argument('--beta', type=float, default=0.4) # ， ， ，
parser.add_argument(
'--save-buffer-name', type=str,
default="./expert_DQN_CartPole-v0.pkl")
parser.add_argument(
'--device', type=str,
default='cuda' if torch.cuda.is_available() else 'cpu')
args = parser.parse_known_args()[0]
return args
def test_dqn(args=get_args()):
env = gym.make(args.task) # env
#
args.state_shape = env.observation_space.shape or env.observation_space.n
#
args.action_shape = env.action_space.shape or env.action_space.n
# train_envs = gym.make(args.task)
# envs，dummyvectorenv for subpro
# you can also use tianshou.env.SubprocVectorEnv
train_envs = DummyVectorEnv(
[lambda: gym.make(args.task) for _ in range(args.training_num)])
# test_envs = gym.make(args.task)
test_envs = DummyVectorEnv(
[lambda: gym.make(args.task) for _ in range(args.test_num)])
# seed
np.random.seed(args.seed)
torch.manual_seed(args.seed)
train_envs.seed(args.seed)
test_envs.seed(args.seed)
# Q_param = V_param = {"hidden_sizes": [128]}
# model
# ，Net
net = Net(args.state_shape, args.action_shape,
hidden_sizes=args.hidden_sizes, device=args.device,
# dueling=(Q_param, V_param),
).to(args.device)
#
optim = torch.optim.Adam(net.parameters(), lr=args.lr)
#
policy = DQNPolicy(
net, optim, args.gamma, args.n_step,
target_update_freq=args.target_update_freq)
# buffer
if args.prioritized_replay:
buf = PrioritizedReplayBuffer(
args.buffer_size, alpha=args.alpha, beta=args.beta)
else:
buf = ReplayBuffer(args.buffer_size)
# collector ，
train_collector = Collector(policy, train_envs, buf)
test_collector = Collector(policy, test_envs)
# policy.set_eps(1)
# batchsize ， batchsize
# batchsize
# ， ， ，
train_collector.collect(n_step=args.batch_size, no_grad=False)
# log
log_path = os.path.join(args.logdir, args.task, 'dqn')
writer = SummaryWriter(log_path)
def save_fn(policy):
torch.save(policy.state_dict(), os.path.join(log_path, 'policy.pth'))
#
def stop_fn(mean_rewards):
return mean_rewards >= env.spec.reward_threshold
#
# epoch ， epoch env step 。
# eps( )
def train_fn(epoch, env_step):
# eps annnealing, just a demo
if env_step <= 10000:
policy.set_eps(args.eps_train)
elif env_step <= 50000:
eps = args.eps_train - (env_step - 10000) / \
40000 * (0.9 * args.eps_train)
policy.set_eps(eps)
else:
policy.set_eps(0.1 * args.eps_train)
def test_fn(epoch, env_step):
policy.set_eps(args.eps_test)
# trainer
#
result = offpolicy_trainer(
policy, train_collector, test_collector, args.epoch,
args.step_per_epoch, args.collect_per_step, args.test_num,
args.batch_size, train_fn=train_fn, test_fn=test_fn,
stop_fn=stop_fn, save_fn=save_fn, writer=writer)
assert stop_fn(result['best_reward'])
if __name__ == '__main__':
pprint.pprint(result)
# Let's watch its performance!
env = gym.make(args.task)
policy.eval()
policy.set_eps(args.eps_test)
collector = Collector(policy, env)
result = collector.collect(n_episode=1, render=args.render)
print(f'Final reward: {result["rew"]}, length: {result["len"]}')
# save buffer in pickle format, for imitation learning unittest
buf = ReplayBuffer(args.buffer_size)
collector = Collector(policy, test_envs, buf)
# ， ， ，
#n_step ，
collector.collect(n_step=args.buffer_size)
pickle.dump(buf, open(args.save_buffer_name, "wb"))
def test_pdqn(args=get_args()):
args.prioritized_replay = True
args.gamma = .95
args.seed = 1
test_dqn(args)
if __name__ == '__main__':
test_dqn(get_args())
上はすべてスクリプト式の動作で、パラメータをargsの中で相対的に便利に定義して私たちがパラメータの修正を行うことができて、論理を修正する時、比較的に複雑で、時間が十分な情況の下で、対象に向かう思想を使うことを考慮することができます.
問題を問題クラスとアルゴリズムクラスに分解し、問題クラスは問題の記述に集中し、問題パラメータを提供し、アルゴリズムクラスはアルゴリズムの記述に集中し、アルゴリズムパラメータを提供する.
その後,本問題におけるアルゴリズムクラスの一部の属性が問題クラスの属性に依存することを考慮して,アルゴリズムクラスに直接問題クラスのインスタンスを転送することができる.この方法は考えやすい.ここでは、3番目のクラスを提供し、3番目のクラスは問題クラスとアルゴリズムクラスを調整するために使用され、3番目のクラスで呼び出し解決とパラメータインタラクションが行われます.どちらがいいかは説明できませんが、自分の試みとしか言えません.
import pprint
import gym
import torch
from tianshou.policy import DQNPolicy
from tianshou.env import DummyVectorEnv
from tianshou.utils.net.common import Net
from tianshou.trainer import offpolicy_trainer
from tianshou.data import Collector, ReplayBuffer, PrioritizedReplayBuffer
import numpy as np
class question:
def __init__(self, gamename='CartPole-v0'):
self.gamename = gamename
env = gym.make(gamename)
self.env = env
#
self.state_shape = env.observation_space.shape or env.observation_space.n
#
self.action_shape = env.action_space.shape or env.action_space.n
class program:
def __init__(self, eps_train, eps_test, epoch, hidden_sizes,
buffer_size, gamma=0.9, n_step=3,
device='cpu', lr=1e-3, target_update_freq=320,
training_num=1, test_num=1, batch_size=64,
step_per_epoch=1, collect_per_step=10):
self.net = None #
self.optim = None #
self.policy = None #
self.eps_train = eps_train
self.eps_test = eps_test
self.buf = ReplayBuffer(buffer_size) #
self.train_collector = None
self.test_collector = None
self.reward_threshold = None
self.epoch = epoch
self.hidden_sizes = hidden_sizes
self.device = device
self.lr = lr
self.step_per_epoch = step_per_epoch
self.collect_per_step = collect_per_step
self.training_num = training_num
self.test_num = test_num
self.batch_size = batch_size
self.ready = False
self.gamma = gamma
self.n_step = n_step
self.target_update_freq = target_update_freq
#
def stop_fn(self, reward_threshold):
return lambda mean_rewards: mean_rewards >= reward_threshold
#
# epoch ， epoch env step 。
# eps( )
def train_fn(self, epoch, env_step):
# eps annnealing, just a demo
if env_step <= 10000:
self.policy.set_eps(self.eps_train)
elif env_step <= 50000:
eps = self.eps_train - (env_step - 10000) / \
40000 * (0.9 * self.eps_train)
self.policy.set_eps(eps)
else:
self.policy.set_eps(0.1 * self.eps_train)
def test_fn(self, epoch, env_step):
self.policy.set_eps(self.eps_test)
def sovle(self):
if self.ready:
return offpolicy_trainer(
self.policy, self.train_collector, self.test_collector, self.epoch,
self.step_per_epoch, self.collect_per_step, self.test_num,
self.batch_size, train_fn=self.train_fn, test_fn=self.test_fn,
stop_fn=self.stop_fn(self.reward_threshold))
else:
raise Exception('unkown error ,maybe you should use init() in class resolve')
class resolve:
def __init__(self, que: question, prg: program,seed=None):
self.que = que
self.prg = prg
self.prg.train_envs = DummyVectorEnv(
[lambda: gym.make(self.que.gamename) for _ in range(self.prg.training_num)])
# test_envs = gym.make(args.task)
self.prg.test_envs = DummyVectorEnv(
[lambda: gym.make(self.que.gamename) for _ in range(self.prg.test_num)])
self.set_seed(seed)
# ， ，net
# net np ，
self.prg.net = Net(self.que.state_shape, self.que.action_shape,
hidden_sizes=self.prg.hidden_sizes, device=self.prg.device,
# dueling=(Q_param, V_param),
).to(self.prg.device) #
self.prg.optim = torch.optim.Adam(self.prg.net.parameters(), lr=self.prg.lr) #
self.prg.policy = DQNPolicy(
self.prg.net, self.prg.optim, self.prg.gamma, self.prg.n_step,
target_update_freq=self.prg.target_update_freq) #
self.prg.train_collector = Collector(self.prg.policy, self.prg.train_envs, self.prg.buf)
self.prg.test_collector = Collector(self.prg.policy, self.prg.test_envs)
self.prg.reward_threshold = self.que.env.spec.reward_threshold
self.prg.ready = True
def set_seed(self, seed):
np.random.seed(seed)
torch.manual_seed(seed)
self.prg.train_envs.seed(seed)
self.prg.test_envs.seed(seed)
def solve(self):
return self.prg.sovle()
def main():
reslv = resolve(question(), program(0.1, 0.05, 10,
[128, 128, 128, 128], 20000,
training_num=8, test_num=100,
step_per_epoch=1000),
seed=1626
)
result=reslv.solve()
pprint.pprint(result)
if __name__ == '__main__':
main()
ここで、policyは、探索ポリシーの実装を含むポリシーを記述するために使用され、dqnは確率的貪欲ポリシーであり、maskパラメータは動作を遮蔽することができる.policyがforwardメソッドを実装する際に探索戦略の実装を行い,予測値と探索動作を返す.
再現のために、シード設定は環境構築後、netなどの他のパラメータの前に置く必要があります.netは初期化時にnpの乱数を呼び出し、前に置くと再現が保証されます.この点は長い間調べていましたが...
更新の策略については、理解できませんでした.ドキュメントの評価を見ると、tianshouはすべての強化学習ライブラリよりも速いが、機能的にはまだ十分ではなく、マルチスマートなどのアルゴリズムが実現していないため、rayに転向することを考えているかもしれない.rayは分布式の枠組みとして、sparkとmllibが私を潰す日子を思い出させた.