学习一下rl-agents的项目结构以及代码实现思路。

source: https://github.com/eleurent/rl-agents

rl-agent Get start

Installation

pip install --user git+https://github.com/eleurent/rl-agents

Usage

rl-agents中的大部分例子可以通过cd到scripts文件夹 cd scripts，执行 python experiments.py命令实现。

Usage:experiments evaluate <environment> <agent> (--train|--test)[--episodes <count>][--seed <str>][--analyze]experiments benchmark <benchmark> (--train|--test)[--processes <count>][--episodes <count>][--seed <str>]experiments -h | --helpOptions:-h --help            Show this screen.--analyze            Automatically analyze the experiment results.--episodes <count>   Number of episodes [default: 5].--processes <count>  Number of running processes [default: 4].--seed <str>         Seed the environments and agents.--train              Train the agent.--test               Test the agent.

evaluate命令允许在给定的环境中评估给定的agent。例如，

# Train a DQN agent on the CartPole-v0 environment
$ python3 experiments.py evaluate configs/CartPoleEnv/env.json configs/CartPoleEnv/DQNAgent.json --train --episodes=200

每个agent都按照标准接口与环境交互：

action = agent.act(state)
next_state, reward, done, info = env.step(action)
agent.record(state, action, reward, next_state, done, info)

环境的配置文件

{"id": "intersection-v0","import_module": "highway_env","observation": {"type": "Kinematics","vehicles_count": 15,"features": ["presence", "x", "y", "vx", "vy", "cos_h", "sin_h"],"features_range": {"x": [-100, 100],"y": [-100, 100],"vx": [-20, 20],"vy": [-20, 20]},"absolute": true,"order": "shuffled"},"destination": "o1"
}

agent的配置文件，核心就是"__class__": "<class 'rl_agents.agents.deep_q_network.pytorch.DQNAgent'>"，利用agent_factory进行agent的创建。

{"__class__": "<class 'rl_agents.agents.deep_q_network.pytorch.DQNAgent'>","model": {"type": "MultiLayerPerceptron","layers": [128, 128]},"gamma": 0.95,"n_steps": 1,"batch_size": 64,"memory_capacity": 15000,"target_update": 512,"exploration": {"method": "EpsilonGreedy","tau": 15000,"temperature": 1.0,"final_temperature": 0.05}
}

如果部分key缺失的话，会使用默认的值agent.default_config()。

最后，可以在基准（baseline）测试中安排一批实验。然后在几个进程上并行执行所有实验。

# Run a benchmark of several agents interacting with environments
$ python3 experiments.py benchmark cartpole_benchmark.json --test --processes=4

基准配置文件包含环境配置列表和agent配置列表。

{"environments": ["configs/CartPoleEnv/env.json"],"agents": ["configs/CartPoleEnv/DQNAgent.json","configs/CartPoleEnv/LinearAgent.json","configs/CartPoleEnv/MCTSAgent.json"]
}

Monitoring

有几种工具可用于监控agent性能：

• Run metadata：为了可重复性，将运行所用的环境和agent配置合并，并保存到metadata.*.json文件中。
• Gym Monitor：每次运行的主要统计数据（episode rewards, lengths, seeds）都会记录到episode_batch.*.stats.json文件中。可以通过运行scripts/analyze.py来自动可视化这些数据。
• Logging：agent可以通过标准的Python日志记录库发送消息。默认情况下，所有日志级别为INFO的消息都会保存到logging.*.lo文件中。要保存日志级别为DEBUG的消息，请添加选项scripts/experiments.py --verbose。
• Tensorboard：默认情况下，一个tensoboard writer会记录有关有用标量、图像和模型图的信息到运行目录。可以通过运行以下命令来进行可视化：tensorboard --logdir <path-to-runs-dir>。

具体代码

rl-agents核心代码集中在rl-agents文件夹和scripts文件夹中，其中，rl-agents主要实现相关的算法，scripts为相应的配置文件。

experiments.py为入口程序，先从它看起，其相应的用法如下：

Usage:experiments evaluate <environment> <agent> (--train|--test) [options]experiments benchmark <benchmark> (--train|--test) [options]experiments -h | --helpOptions:-h --help              Show this screen.--episodes <count>     Number of episodes [default: 5].--no-display           Disable environment, agent, and rewards rendering.--name-from-config     Name the output folder from the corresponding config files--processes <count>    Number of running processes [default: 4].--recover              Load model from the latest checkpoint.--recover-from <file>  Load model from a given checkpoint.--seed <str>           Seed the environments and agents.--train                Train the agent.--test                 Test the agent.--verbose              Set log level to debug instead of info.--repeat <times>       Repeat several times [default: 1].

首先从main函数开始，根据evaluate或者benchmark执行相应的任务。暂且先从evaluate入手。

def main():opts = docopt(__doc__)if opts['evaluate']:for _ in range(int(opts['--repeat'])):evaluate(opts['<environment>'], opts['<agent>'], opts)elif opts['benchmark']:benchmark(opts)

evaluate主要完成env、agent的创建以及evaluation 对象的创建，再根据选择train或test执行不同的程序。

def evaluate(environment_config, agent_config, options):"""Evaluate an agent interacting with an environment.:param environment_config: the path of the environment configuration file:param agent_config: the path of the agent configuration file:param options: the evaluation options"""logger.configure(LOGGING_CONFIG)if options['--verbose']:logger.configure(VERBOSE_CONFIG)env = load_environment(environment_config)agent = load_agent(agent_config, env)run_directory = Noneif options['--name-from-config']:run_directory = "{}_{}_{}".format(Path(agent_config).with_suffix('').name,datetime.datetime.now().strftime('%Y%m%d-%H%M%S'),os.getpid())options['--seed'] = int(options['--seed']) if options['--seed'] is not None else Noneevaluation = Evaluation(env,agent,run_directory=run_directory,num_episodes=int(options['--episodes']),sim_seed=options['--seed'],recover=options['--recover'] or options['--recover-from'],display_env=not options['--no-display'],display_agent=not options['--no-display'],display_rewards=not options['--no-display'])if options['--train']:evaluation.train()elif options['--test']:evaluation.test()else:evaluation.close()return os.path.relpath(evaluation.run_directory)

Evaluation类中主要包含以下函数：

__init__的一些参数说明

参数	描述
env	要解决的环境，可能是包装了AbstractEnv的环境
agent	解决环境的AbstractAgent agent
directory	工作空间目录路径
run_directory	运行目录路径
num_episodes	运行的episode数
training	agent是处于训练模式还是测试模式
sim_seed	环境/agent随机性源的种子
recover	从文件中恢复agent参数。如果为True，则使用默认的最新保存。如果为字符串，则将其用作路径。
display_env	渲染环境，并有一个监视器录制其视频
display_agent	如果支持，将agent图形添加到环境查看器中
display_rewards	通过episodes显示agent的性能
close_env	当评估结束时，是否应该关闭环境
step_callback_fn	在每个环境步骤之后调用的回调函数。它接受以下参数：(episode, env, agent, transition, writer)。

首先看一下train，根据agent是否有batched属性，分为run_batched_episodes和run_episodes

    def train(self):self.training = Trueif getattr(self.agent, "batched", False):self.run_batched_episodes()else:self.run_episodes()self.close()

run_episodes就是一般强化学习的基本过程，注意其中的reset step 等函数都是经过封装的。实现自己的算法时需要注意。run_batched_episodes则主要实现一些并行计算的任务，这一部分等之后再详细介绍。

    def run_episodes(self):for self.episode in range(self.num_episodes):# Run episodeterminal = Falseself.reset(seed=self.episode)rewards = []start_time = time.time()while not terminal:# Step until a terminal step is reachedreward, terminal = self.step()rewards.append(reward)# Catch interruptionstry:if self.env.unwrapped.done:breakexcept AttributeError:pass# End of episodeduration = time.time() - start_timeself.after_all_episodes(self.episode, rewards, duration)self.after_some_episodes(self.episode, rewards)

test为模型测试部分

    def test(self):"""Test the agent.If applicable, the agent model should be loaded before using the recover option."""self.training = Falseif self.display_env:self.wrapped_env.episode_trigger = lambda e: Truetry:self.agent.eval()except AttributeError:passself.run_episodes()self.close()

其中eval也需要进行重写。

    def eval(self):"""Set to testing mode. Disable any unnecessary exploration."""pass