Ray：AI的分布式系统

# 定义两个远程函数。 调用这些函数创造任务
# 这是远程执行的。

@ray.remote
def multiply(x, y):
    return np.dot(x, y)

@ray.remote
def zeros(size):
    return np.zeros(size)

# 并行启动两个任务。这些立即返回预期结果，且
# 任务在后台执行。
x_id  zeros.remote((100, 100))
y_id = zeros.remote((100, 100))

# 开始第三个任务。在前两个任务完成之前，不会安排这一任务.
z_id = multiply.remote(x_id, y_id)

# 得到结果。直到第三个任务完成才可以得到。
z = ray.get(z_id)

import gym

@ray.remote
class Simulator(object):
    def __init__(self):
        self.env = gym.make("Pong-v0")
        self.env.reset()

    def step(self, action):
        return self.env.step(action)

# 创建一个模拟器,这将启动一个远程进程为
# 该actor运行所有方法。
simuator = Simulator.remote()

observations = []
for _ in range(4):
    # 在模拟器中执行操作0.此调用不阻塞并且
    # 返回一个预期结果
    observations.append(simulator.step.remote(0))

@ray.remote
class ParameterServer(object):
    def __init__(self, keys, values):
        # 这些值将会发生改变，因此我们需要创建一个本地副本。
        values = [value.copy() for value in values]
        self.parameters = dict(zip(keys, values))

    def get(self, keys):
        return [self.parameters[key] for key in keys]

    def update(self, keys, values):
        # 这个更新函数使现有的值增加，但是更新
        # 函数可以被任意定义
        for key, value in zip(keys, values):
            self.parameters[key] += value

parameter_server = ParameterServer.remote(initial_keys, initial_values)

@ray.remote
def worker_task(parameter_server):
    while True:
        keys = ['key1', 'key2', 'key3']
        # 获取最新的参数。
        values = ray.get(parameter_server.get.remote(keys))
        # 计算一些需更新参数。
        updates = …
        # 更新参数。
        parameter_server.update.remote(keys, updates)

# 启动四个长期任务。
for _ in range(4):
    worker_task.remote(parameter_server)

from ray.tune import register_trainable, grid_search, run_experiments

# 优化功能。超参数是在配置
# 参数。
def my_func(config, reporter):
    import time, numpy as np
    i = 0
    while True:
        reporter(timesteps_total=i, mean_accuracy=(i ** config['alpha']))
        i += config['beta']
        time.sleep(0.01)

register_trainable('my_func', my_func)

run_experiments({
    'my_experiment': {
        'run': 'my_func',
        'resources': {'cpu': 1, 'gpu': 0},
        'stop': {'mean_accuracy': 100},
        'config': {
            'alpha': grid_search([0.2, 0.4, 0.6]),
            'beta': grid_search([1, 2]),
        },
    }
})