分散深層強化学習ライブラリHandyRLをコンペで使ってみた。
Sponsored Links
皆さんこんにちは
お元気ですか。ブログ書きながら、当チームのガチョウを見守っています。
最近までHungryGeeseに参加しており、このコンペでHandyRLライブラリには大変お世話になりました。
このコンペでHandyRLを改造して使ったので、そのポイントを記録として残しておきます。
HandyRLとは
一言で言えば、PyTorchで利用できる軽量な深層分散強化学習用のフレームワークです。
実際に使ってみた感想としても、これまでの強化学習のFWより直感的に理解しやすいものでした。
分散深層強化学習ではこの2つの処理を同時に行っています。
1. エージェントを自己対戦させてエピソードを生成する処理
2. 生成したエピソードを使ってモデルを学習させる処理
これを自前で実装するのは非常に骨が折れるものですが、このあたりをきれいに実装されており、使いやすいものになってました。
HandyRLの使い方
Tips
コンペで勝つには、上記の基本だけだと対応しづらい部分があります。
そのため、他コンペでも利用できる改造ポイントをTipsを書いておきます。
自己対戦以外でエージェント作成
HandyRLのデフォルトエージェントは、自己対戦になります。
しかし、多様性の観点からある程度、様々なエージェントと対戦させたほうが良いです。
自己対戦以外のエージェントを利用する場合、train.pyとworker.pyを変更する必要があります。
今回は変更の例とその解説を簡単に行います。
train.pyのL587付近
for p in self.env.players(): if random.random() > 0.5: args['model_id'][p] = -1 else: args['model_id'][p] = self.model_era is_self_model = True players.append(p) # 自己対戦リスト if is_self_model is False: p = random.choice(self.env.players()) args['model_id'][p] = self.model_era players.append(p)
入力の値が-1であれば、事前に学習済のエージェントを使ってエピソードを生成する、
また、self.model_era（＝学習回数,Epoch）であれば、自己対戦用エージェントを利用する。といった方式を想定して入力しています。
入力されたデータによってどのモデルかは、worker.pyで選択するようにします。
worker.py
class Worker: def __init__(self, args, conn, wid): # 省略、以下を追加、モデルをPoolしておく処理、get_modelsでエージェントとなるモデルの一覧を取得する。 self.trained_models = [ModelWrapper(model, ts=1) for model in get_models()] + [ModelWrapper(model, ts=2) for model in get_2ts_models()] def _gather_models(self, model_ids, role="g"): model_pool = {} for model_id in model_ids: if model_id not in model_pool: if model_id < 0: if role == "g": model_pool[model_id] = self.trained_models # Poolしたモデルを利用する。 elif model_id == self.latest_model[0]: model_pool[model_id] = self.latest_model[1] else: model_pool[model_id] = ModelWrapper(pickle.loads(send_recv(self.conn, ('model', model_id))), ts=-1) if model_id > self.latest_model[0]: self.latest_model = model_id, model_pool[model_id] return model_pool def run(self): while True: args = send_recv(self.conn, ('args', None)) role = args['role'] models = {} if 'model_id' in args: model_ids = list(args['model_id'].values()) model_pool = self._gather_models(model_ids, role=role) # make dict of models for p, model_id in args['model_id'].items(): if model_id < 0: models[p] = deepcopy(random.choice(model_pool[model_id])) # Poolの中からランダムに選択する。 else: models[p] = model_pool[model_id]
学習済モデルを利用する実装です。いわばPoolのようなもので、学習済モデルをこの中から選択する方式です。
上記コードでは次のことを行っています。
・学習済モデルをコンストラクタで初期化
・_gather_modelsで、エピソード生成時にmodel_idが-1の場合にモデルPoolを選択
・runでPoolの中から一つモデルを選択し、対戦させる。
様々なエージェントで評価
初期値だとランダム動作での評価になる（HungryGeeseだけ？）ので、正直ある程度学習させると弱すぎてほぼ勝利し意味のある評価ができなくなります。
そのため、多様性がありつつも、ある程度強いエージェントで評価が必要です。
こちらもtrain.py/worker.pyを修正する必要があります。
train.pyのL596付近
elif args['role'] == 'e': # evaluation configuration args['player'] = [self.env.players()[self.num_results % len(self.env.players())]] for p in self.env.players(): if p in args['player']: args['model_id'][p] = self.model_era else: args['model_id'][p] = -1 self.num_results += 1
train.pyでは、一つ必ず、学習しているモデルを選択し、残りを評価として選択するためのモデルとして設定します。
worker.py
class Worker: def __init__(self, args, conn, wid): self.evaluate_models = [ModelWrapper(model, ts=1) for model in get_models()] + [ModelWrapper(model, ts=2) for model in get_2ts_models()] def _gather_models(self, model_ids, role="g"): model_pool = {} for model_id in model_ids: if model_id not in model_pool: if model_id < 0: if role == "g": # 略 else: model_pool[model_id] = self.evaluate_models
roleが"g"であれば、エピソード生成、"e"であれば、評価になります。
そのため、評価をする場合は定義しておいたものからランダムに選択する実装にしています。（選択部は学習と同じ選択仕様）
自己対戦モデルを一定期間保存する
一定期間のEpisodeを保存しておき、そのエージェントと対戦させたいといったケースが考えられます。
これにより学習を進めた結果、今までのモデルに負けるといったことを防ぐ取り組みです。
例えば、AlphaStarでは過去学習させたエージェントも保存しておき、対戦するようなこともされました。
この方式もworker.pyを変更することで実現しました。
class Worker: def __init__(self, args, conn, wid): # 省略、以下を追加、自己対戦モデル用のPool self.self_fight_models = [] def _gather_models(self, model_ids, role="g"): model_pool = {} for model_id in model_ids: if model_id not in model_pool: if model_id < 0: if role == "g": if len(self.self_fight_models) == 0: model_pool[model_id] = self.trained_models else: model_pool[model_id] = \ random.choices([self.self_fight_models, self.trained_models], k=1, weights=[1, 1])[0] # 自己対戦モデル or 学習済エージェント else: model_pool[model_id] = self.evaluate_models elif model_id == self.latest_model[0]: # 略 else: model_pool[model_id] = ModelWrapper(pickle.loads(send_recv(self.conn, ('model', model_id))), ts=-1) if model_id > self.latest_model[0]: self.latest_model = model_id, model_pool[model_id] if model_id % 300 == 0: # 300学習につき一つ対戦用のPoolに追加する。 self.self_fight_models.append(model_pool[model_id]) if len(self.self_fight_models) > 100: # 100件以上作られれば最初のを捨てる。 self.self_fight_models = self.self_fight_models[1:]
学習済モデルの実行
事前に学習したモデルを使いたいといった場面があります。
例えば、Kaggleでは、上位LBなど他エージェントのエピソードを獲得できます。
それを用いて勝利モデルの行動を事前に模倣学習することで、ある一定の強さのエージェントから学習を開始でき、学習の高速化を見込めます。
ソースをほぼ変更しなくとも具体的には次の方式で実現可能です。
1. modelsディレクトリを作成し、モデルを「1.pth」して配置する。
2. 設定の「config.yaml」の「restart_epoch」を1にする。
3. 学習を実行する。
HandyRLはepisodeの設定を1以上でmodels配下のモデルを利用する方式を採用しています。
また、デフォルトの読み込み（pytorchのload_state_dict)のStrict=Falseとなっているため、若干の構造変化には対応できます。
決定的動作で動かす
HandyRLのエピソード生成でのエージェントのアクションはモデルの出力の分布からサンプリングになります。
しかし、一番確度が高そうな行動を決定的に動かしたほうがエージェントとしては（おそらく）強くなります。
train.pyのL596付近
for p in self.env.players(): if random.random() > 0.5: args['model_id'][p] = -1 else: args['model_id'][p] = self.model_era is_self_model = True players.append(p) if random.random() < 0.05: determistics.append(True) else: determistics.append(False) args['determistics'] = determistics
これによりdetermisticsに各プレイヤーが決定的に動くかどうかを入力しています。
（実装例は5%の確率で決定的に動作するエージェントになる）
また、generation.pyのL53の次の箇所を修正します。
if args['determistics'][player]: action = legal_actions[np.argmax(p[legal_actions])] else: action = random.choices(legal_actions, weights=softmax(p[legal_actions]))[0]
これにより、determisticsフラグがTrueであれば、決定的に動作するようにしています。
感想
今まで強化学習を行うとなれば、複雑なライブラリや実装を読み解く必要があり、少し改造するにも大変だと感じていました。
しかし、HandyRLは非常にシンプルな構成で実装も読みやすく、扱いやすかったです。
シュミレーションコンペなどで強化学習を行う場合はこのライブラリを積極的に使うのが今後良さそうに思えます。
最後にこれを利用したのはHandyRLチームが積極的にHungryGeeseで学習の条件やNotebookを公開いただき、シュミレーションコンペと強化学習への障壁を下げてもらったからでした。
今回、HungryGeeseでフルに活用させていただき、最新の強化学習のキャッチアップ含め、勉強になりました。ありがとうございました。
