Ray 一个开放源码框架，为构建分布式应用程序提供简单、通用的API

import ray ray.init()

@ray.remote def f(x):
    return x * x futures = [f.remote(i) for i in range(4)]
print(ray.get(futures))

import ray ray.init()

@ray.remote class Counter(object):
    def __init__(self):
        self.n = 0 def increment(self):
        self.n += 1 def read(self):
        return self.n counters = [Counter.remote() for i in range(4)]
[c.increment.remote() for c in counters]
futures = [c.read.remote() for c in counters]
print(ray.get(futures))

from ray import tune def objective(step, alpha, beta):
    return (0.1 + alpha * step / 100)**(-1) + beta * 0.1 def training_function(config):
    # Hyperparameters alpha, beta = config["alpha"], config["beta"]
    for step in range(10):
        # Iterative training function - can be any arbitrary training procedure. intermediate_score = objective(step, alpha, beta)
        # Feed the score back back to Tune. tune.report(mean_loss=intermediate_score)


analysis = tune.run(
    training_function,
    config={
        "alpha": tune.grid_search([0.001, 0.01, 0.1]),
        "beta": tune.choice([1, 2, 3])
    })

print("Best config: ", analysis.get_best_config(metric="mean_loss", mode="min"))

# Get a dataframe for analyzing trial results. df = analysis.results_df

import gym from gym.spaces import Discrete, Box from ray import tune class SimpleCorridor(gym.Env):
    def __init__(self, config):
        self.end_pos = config["corridor_length"]
        self.cur_pos = 0 self.action_space = Discrete(2)
        self.observation_space = Box(0.0, self.end_pos, shape=(1, ))

    def reset(self):
        self.cur_pos = 0 return [self.cur_pos]

    def step(self, action):
        if action == 0 and self.cur_pos > 0:
            self.cur_pos -= 1 elif action == 1:
            self.cur_pos += 1 done = self.cur_pos >= self.end_pos return [self.cur_pos], 1 if done else 0, done, {}

tune.run(
    "PPO",
    config={
        "env": SimpleCorridor,
        "num_workers": 4,
        "env_config": {"corridor_length": 5}})

from ray import serve import pickle import requests from sklearn.datasets import load_iris from sklearn.ensemble import GradientBoostingClassifier # Train model iris_dataset = load_iris()
model = GradientBoostingClassifier()
model.fit(iris_dataset["data"], iris_dataset["target"])

# Define Ray Serve model, class BoostingModel:
    def __init__(self):
        self.model = model self.label_list = iris_dataset["target_names"].tolist()

    def __call__(self, flask_request):
        payload = flask_request.json["vector"]
        print("Worker: received flask request with data", payload)

        prediction = self.model.predict([payload])[0]
        human_name = self.label_list[prediction]
        return {"result": human_name}


# Deploy model client = serve.start()
client.create_backend("iris:v1", BoostingModel)
client.create_endpoint("iris_classifier", backend="iris:v1", route="/iris")

# Query it! sample_request_input = {"vector": [1.2, 1.0, 1.1, 0.9]}
response = requests.get("http://localhost:8000/iris", json=sample_request_input)
print(response.text)
# Result: # { #  "result": "versicolor" # }