wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-06 05:16:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

[rllib, tune] TrainingResult -> Dict, Removes C408 from flake8 (#2565)

Browse Source
This commit is contained in:
Richard Liaw
2018-08-07 12:17:44 -07:00
committed by GitHub
parent a3202f581c
commit bb44456f6f
43 changed files with 252 additions and 301 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/ray/rllib/agents/a3c/a3c.py
+2 -2
View File
@@ -103,8 +103,8 @@ class A3CAgent(Agent):
FilterManager.synchronize(self.local_evaluator.filters,
self.remote_evaluators)
result = self.optimizer.collect_metrics()
result = result._replace(
timesteps_this_iter=self.optimizer.num_steps_sampled - prev_steps)
result.update(timesteps_this_iter=self.optimizer.num_steps_sampled -
prev_steps)
return result
def _stop(self):
python/ray/rllib/agents/agent.py
+3 -4
View File
@@ -12,7 +12,6 @@ import tensorflow as tf
from ray.rllib.evaluation.policy_evaluator import PolicyEvaluator
from ray.rllib.utils import deep_update
from ray.tune.registry import ENV_CREATOR, _global_registry
from ray.tune.result import TrainingResult
from ray.tune.trainable import Trainable
COMMON_CONFIG = {
@@ -266,7 +265,7 @@ class _MockAgent(Agent):
if self.config["mock_error"] and self.iteration == 1 \
and (self.config["persistent_error"] or not self.restored):
raise Exception("mock error")
return TrainingResult(
return dict(
episode_reward_mean=10,
episode_len_mean=10,
timesteps_this_iter=10,
@@ -310,7 +309,7 @@ class _SigmoidFakeData(_MockAgent):
i = max(0, self.iteration - self.config["offset"])
v = np.tanh(float(i) / self.config["width"])
v *= self.config["height"]
return TrainingResult(
return dict(
episode_reward_mean=v,
episode_len_mean=v,
timesteps_this_iter=self.config["iter_timesteps"],
@@ -330,7 +329,7 @@ class _ParameterTuningAgent(_MockAgent):
}
def _train(self):
return TrainingResult(
return dict(
episode_reward_mean=self.config["reward_amt"] * self.iteration,
episode_len_mean=self.config["reward_amt"],
timesteps_this_iter=self.config["iter_timesteps"],
python/ray/rllib/agents/bc/bc.py
+1 -2
View File
@@ -8,7 +8,6 @@ from ray.rllib.agents.bc.bc_evaluator import BCEvaluator, \
GPURemoteBCEvaluator, RemoteBCEvaluator
from ray.rllib.optimizers import AsyncGradientsOptimizer
from ray.rllib.utils import merge_dicts
from ray.tune.result import TrainingResult
from ray.tune.trial import Resources
DEFAULT_CONFIG = {
@@ -89,7 +88,7 @@ class BCAgent(Agent):
for m in ray.get(metrics):
total_samples += m["num_samples"]
total_loss += m["loss"]
result = TrainingResult(
result = dict(
mean_loss=total_loss / total_samples,
timesteps_this_iter=total_samples,
)
python/ray/rllib/agents/dqn/dqn.py
+2 -1
View File
@@ -203,13 +203,14 @@ class DQNAgent(Agent):
result = collect_metrics(self.local_evaluator,
self.remote_evaluators)
return result._replace(
result.update(
timesteps_this_iter=self.global_timestep - start_timestep,
info=dict({
"min_exploration": min(exp_vals),
"max_exploration": max(exp_vals),
"num_target_updates": self.num_target_updates,
}, **self.optimizer.stats()))
return result
def _stop(self):
# workaround for https://github.com/ray-project/ray/issues/1516
python/ray/rllib/agents/es/es.py
+1 -1
View File
@@ -300,7 +300,7 @@ class ESAgent(Agent):
"time_elapsed": step_tend - self.tstart
}
result = ray.tune.result.TrainingResult(
result = dict(
episode_reward_mean=eval_returns.mean(),
episode_len_mean=eval_lengths.mean(),
timesteps_this_iter=noisy_lengths.sum(),
python/ray/rllib/agents/impala/impala.py
+2 -2
View File
@@ -93,8 +93,8 @@ class ImpalaAgent(Agent):
FilterManager.synchronize(self.local_evaluator.filters,
self.remote_evaluators)
result = self.optimizer.collect_metrics()
result = result._replace(
timesteps_this_iter=self.optimizer.num_steps_sampled - prev_steps)
result.update(timesteps_this_iter=self.optimizer.num_steps_sampled -
prev_steps)
return result
def _stop(self):
python/ray/rllib/agents/pg/pg.py
+4 -2
View File
@@ -50,5 +50,7 @@ class PGAgent(Agent):
def _train(self):
prev_steps = self.optimizer.num_steps_sampled
self.optimizer.step()
return self.optimizer.collect_metrics()._replace(
timesteps_this_iter=self.optimizer.num_steps_sampled - prev_steps)
result = self.optimizer.collect_metrics()
result.update(timesteps_this_iter=self.optimizer.num_steps_sampled -
prev_steps)
return result
python/ray/rllib/agents/ppo/ppo.py
+2 -2
View File
@@ -112,9 +112,9 @@ class PPOAgent(Agent):
FilterManager.synchronize(self.local_evaluator.filters,
self.remote_evaluators)
res = self.optimizer.collect_metrics()
res = res._replace(
res.update(
timesteps_this_iter=self.optimizer.num_steps_sampled - prev_steps,
info=dict(fetches, **res.info))
info=dict(fetches, **res.get("info", {})))
return res
def _stop(self):
python/ray/rllib/evaluation/metrics.py
+1 -2
View File
@@ -6,7 +6,6 @@ import numpy as np
import collections
import ray
from ray.tune.result import TrainingResult
def collect_metrics(local_evaluator, remote_evaluators=[]):
@@ -38,7 +37,7 @@ def collect_metrics(local_evaluator, remote_evaluators=[]):
for policy_id, rewards in policy_rewards.copy().items():
policy_rewards[policy_id] = np.mean(rewards)
return TrainingResult(
return dict(
episode_reward_max=max_reward,
episode_reward_min=min_reward,
episode_reward_mean=avg_reward,
python/ray/rllib/optimizers/policy_optimizer.py
+2 -2
View File
@@ -82,11 +82,11 @@ class PolicyOptimizer(object):
"""Returns evaluator and optimizer stats.
Returns:
res (TrainingResult): TrainingResult from evaluator metrics with
res (dict): A training result dict from evaluator metrics with
`info` replaced with stats from self.
"""
res = collect_metrics(self.local_evaluator, self.remote_evaluators)
res = res._replace(info=self.stats())
res.update(info=self.stats())
return res
def save(self):
python/ray/rllib/test/test_multi_agent_env.py
+11 -9
View File
@@ -313,8 +313,8 @@ class TestMultiAgentEnv(unittest.TestCase):
for i in range(100):
result = pg.train()
print("Iteration {}, reward {}, timesteps {}".format(
i, result.episode_reward_mean, result.timesteps_total))
if result.episode_reward_mean >= 50 * n:
i, result["episode_reward_mean"], result["timesteps_total"]))
if result["episode_reward_mean"] >= 50 * n:
return
raise Exception("failed to improve reward")
@@ -349,7 +349,7 @@ class TestMultiAgentEnv(unittest.TestCase):
for i in range(10):
result = pg.train()
print("Iteration {}, reward {}, timesteps {}".format(
i, result.episode_reward_mean, result.timesteps_total))
i, result["episode_reward_mean"], result["timesteps_total"]))
self.assertTrue(
pg.compute_action([0, 0, 0, 0], policy_id="policy_1") in [0, 1])
self.assertTrue(
@@ -407,9 +407,10 @@ class TestMultiAgentEnv(unittest.TestCase):
ev.foreach_policy(
lambda p, _: p.update_target()
if isinstance(p, DQNPolicyGraph) else None)
print("Iter {}, rew {}".format(i, result.policy_reward_mean))
print("Total reward", result.episode_reward_mean)
if result.episode_reward_mean >= 25 * n:
print("Iter {}, rew {}".format(i,
result["policy_reward_mean"]))
print("Total reward", result["episode_reward_mean"])
if result["episode_reward_mean"] >= 25 * n:
return
print(result)
raise Exception("failed to improve reward")
@@ -442,9 +443,10 @@ class TestMultiAgentEnv(unittest.TestCase):
for i in range(100):
optimizer.step()
result = collect_metrics(ev)
print("Iteration {}, rew {}".format(i, result.policy_reward_mean))
print("Total reward", result.episode_reward_mean)
if result.episode_reward_mean >= 25 * n:
print("Iteration {}, rew {}".format(i,
result["policy_reward_mean"]))
print("Total reward", result["episode_reward_mean"])
if result["episode_reward_mean"] >= 25 * n:
return
raise Exception("failed to improve reward")
python/ray/rllib/test/test_policy_evaluator.py
+8 -8
View File
@@ -124,8 +124,8 @@ class TestPolicyEvaluator(unittest.TestCase):
ev.sample()
ray.get(remote_ev.sample.remote())
result = collect_metrics(ev, [remote_ev])
self.assertEqual(result.episodes_total, 20)
self.assertEqual(result.episode_reward_mean, 10)
self.assertEqual(result["episodes_total"], 20)
self.assertEqual(result["episode_reward_mean"], 10)
def testAsync(self):
ev = PolicyEvaluator(
@@ -160,12 +160,12 @@ class TestPolicyEvaluator(unittest.TestCase):
batch = ev.sample()
self.assertEqual(batch.count, 16)
result = collect_metrics(ev, [])
self.assertEqual(result.episodes_total, 0)
self.assertEqual(result["episodes_total"], 0)
for _ in range(8):
batch = ev.sample()
self.assertEqual(batch.count, 16)
result = collect_metrics(ev, [])
self.assertEqual(result.episodes_total, 8)
self.assertEqual(result["episodes_total"], 8)
indices = []
for env in ev.async_env.vector_env.envs:
self.assertEqual(env.unwrapped.config.worker_index, 0)
@@ -191,10 +191,10 @@ class TestPolicyEvaluator(unittest.TestCase):
batch = ev.sample()
self.assertEqual(batch.count, 16)
result = collect_metrics(ev, [])
self.assertEqual(result.episodes_total, 0)
self.assertEqual(result["episodes_total"], 0)
batch = ev.sample()
result = collect_metrics(ev, [])
self.assertEqual(result.episodes_total, 4)
self.assertEqual(result["episodes_total"], 4)
def testVectorEnvSupport(self):
ev = PolicyEvaluator(
@@ -206,12 +206,12 @@ class TestPolicyEvaluator(unittest.TestCase):
batch = ev.sample()
self.assertEqual(batch.count, 10)
result = collect_metrics(ev, [])
self.assertEqual(result.episodes_total, 0)
self.assertEqual(result["episodes_total"], 0)
for _ in range(8):
batch = ev.sample()
self.assertEqual(batch.count, 10)
result = collect_metrics(ev, [])
self.assertEqual(result.episodes_total, 8)
self.assertEqual(result["episodes_total"], 8)
def testTruncateEpisodes(self):
ev = PolicyEvaluator(
python/ray/rllib/test/test_serving_env.py
+6 -6
View File
@@ -157,8 +157,8 @@ class TestServingEnv(unittest.TestCase):
for i in range(100):
result = dqn.train()
print("Iteration {}, reward {}, timesteps {}".format(
i, result.episode_reward_mean, result.timesteps_total))
if result.episode_reward_mean >= 100:
i, result["episode_reward_mean"], result["timesteps_total"]))
if result["episode_reward_mean"] >= 100:
return
raise Exception("failed to improve reward")
@@ -168,8 +168,8 @@ class TestServingEnv(unittest.TestCase):
for i in range(100):
result = pg.train()
print("Iteration {}, reward {}, timesteps {}".format(
i, result.episode_reward_mean, result.timesteps_total))
if result.episode_reward_mean >= 100:
i, result["episode_reward_mean"], result["timesteps_total"]))
if result["episode_reward_mean"] >= 100:
return
raise Exception("failed to improve reward")
@@ -180,8 +180,8 @@ class TestServingEnv(unittest.TestCase):
for i in range(100):
result = pg.train()
print("Iteration {}, reward {}, timesteps {}".format(
i, result.episode_reward_mean, result.timesteps_total))
if result.episode_reward_mean >= 100:
i, result["episode_reward_mean"], result["timesteps_total"]))
if result["episode_reward_mean"] >= 100:
return
raise Exception("failed to improve reward")
python/ray/rllib/tuned_examples/regression_tests/regression_test.py
+5 -3
View File
@@ -27,9 +27,11 @@ def _evaulate_config(filename):
trials = tune.run_experiments(experiments)
results = defaultdict(list)
for t in trials:
results["time_total_s"] += [t.last_result.time_total_s]
results["episode_reward_mean"] += [t.last_result.episode_reward_mean]
results["training_iteration"] += [t.last_result.training_iteration]
results["time_total_s"] += [t.last_result["time_total_s"]]
results["episode_reward_mean"] += [
t.last_result["episode_reward_mean"]
]
results["training_iteration"] += [t.last_result["training_iteration"]]
return {k: np.median(v) for k, v in results.items()}
python/ray/rllib/tuned_examples/run_regression_tests.py
+1 -1
View File
@@ -23,7 +23,7 @@ if __name__ == '__main__':
num_failures = 0
for t in trials:
if (t.last_result.episode_reward_mean <
if (t.last_result["episode_reward_mean"] <
t.stopping_criterion["episode_reward_mean"]):
num_failures += 1
python/ray/rllib/utils/__init__.py
+1 -6
View File
@@ -5,12 +5,7 @@ from ray.rllib.utils.filter import Filter
from ray.rllib.utils.policy_client import PolicyClient
from ray.rllib.utils.policy_server import PolicyServer
__all__ = [
"Filter",
"FilterManager",
"PolicyClient",
"PolicyServer",
]
__all__ = ["Filter", "FilterManager", "PolicyClient", "PolicyServer"]
def merge_dicts(d1, d2):
python/ray/tune/ParallelCoordinatesVisualization.ipynb
-1
View File
@@ -70,7 +70,6 @@
"source": [
"GOOD_FIELDS = ['experiment_id',\n",
" 'num_sgd_iter',\n",
" 'timesteps_total',\n",
" 'episode_len_mean',\n",
" 'episode_reward_mean']\n",
"\n",
python/ray/tune/__init__.py
+1 -2
View File
@@ -6,11 +6,10 @@ from ray.tune.error import TuneError
from ray.tune.tune import run_experiments
from ray.tune.experiment import Experiment
from ray.tune.registry import register_env, register_trainable
from ray.tune.result import TrainingResult
from ray.tune.trainable import Trainable
from ray.tune.suggest import grid_search
__all__ = [
"Trainable", "TrainingResult", "TuneError", "grid_search", "register_env",
"Trainable", "TuneError", "grid_search", "register_env",
"register_trainable", "run_experiments", "Experiment"
]
python/ray/tune/async_hyperband.py
+7 -7
View File
@@ -18,11 +18,11 @@ class AsyncHyperBandScheduler(FIFOScheduler):
See https://openreview.net/forum?id=S1Y7OOlRZ
Args:
time_attr (str): The TrainingResult attr to use for comparing time.
time_attr (str): A training result attr to use for comparing time.
Note that you can pass in something non-temporal such as
`training_iteration` as a measure of progress, the only requirement
is that the attribute should increase monotonically.
reward_attr (str): The TrainingResult objective value attribute. As
reward_attr (str): The training result objective value attribute. As
with `time_attr`, this may refer to any objective value. Stopping
procedures will use this attribute.
max_t (float): max time units per trial. Trials will be stopped after
@@ -72,20 +72,20 @@ class AsyncHyperBandScheduler(FIFOScheduler):
def on_trial_result(self, trial_runner, trial, result):
action = TrialScheduler.CONTINUE
if getattr(result, self._time_attr) >= self._max_t:
if result[self._time_attr] >= self._max_t:
action = TrialScheduler.STOP
else:
bracket = self._trial_info[trial.trial_id]
action = bracket.on_result(trial, getattr(result, self._time_attr),
getattr(result, self._reward_attr))
action = bracket.on_result(trial, result[self._time_attr],
result[self._reward_attr])
if action == TrialScheduler.STOP:
self._num_stopped += 1
return action
def on_trial_complete(self, trial_runner, trial, result):
bracket = self._trial_info[trial.trial_id]
bracket.on_result(trial, getattr(result, self._time_attr),
getattr(result, self._reward_attr))
bracket.on_result(trial, result[self._time_attr],
result[self._reward_attr])
del self._trial_info[trial.trial_id]
def on_trial_remove(self, trial_runner, trial):
python/ray/tune/config_parser.py
+3 -3
View File
@@ -70,9 +70,9 @@ def make_parser(parser_creator=None, **kwargs):
default="{}",
type=json.loads,
help="The stopping criteria, specified in JSON. The keys may be any "
"field in TrainingResult, e.g. "
"'{\"time_total_s\": 600, \"timesteps_total\": 100000}' to stop "
"after 600 seconds or 100k timesteps, whichever is reached first.")
"field returned by 'train()' e.g. "
"'{\"time_total_s\": 600, \"training_iteration\": 100000}' to stop "
"after 600 seconds or 100k iterations, whichever is reached first.")
parser.add_argument(
"--config",
default="{}",
python/ray/tune/examples/async_hyperband_example.py
+6 -5
View File
@@ -12,7 +12,7 @@ import random
import numpy as np
import ray
from ray.tune import Trainable, TrainingResult, register_trainable, \
from ray.tune import Trainable, register_trainable, \
run_experiments
from ray.tune.async_hyperband import AsyncHyperBandScheduler
@@ -33,8 +33,8 @@ class MyTrainableClass(Trainable):
v *= self.config["height"]
# Here we use `episode_reward_mean`, but you can also report other
# objectives such as loss or accuracy (see tune/result.py).
return TrainingResult(episode_reward_mean=v, timesteps_this_iter=1)
# objectives such as loss or accuracy.
return {"episode_reward_mean": v}
def _save(self, checkpoint_dir):
path = os.path.join(checkpoint_dir, "checkpoint")
@@ -58,9 +58,10 @@ if __name__ == "__main__":
# asynchronous hyperband early stopping, configured with
# `episode_reward_mean` as the
# objective and `timesteps_total` as the time unit.
# objective and `training_iteration` as the time unit,
# which is automatically filled by Tune.
ahb = AsyncHyperBandScheduler(
time_attr="timesteps_total",
time_attr="training_iteration",
reward_attr="episode_reward_mean",
grace_period=5,
max_t=100)
python/ray/tune/examples/hyperband_example.py
+6 -5
View File
@@ -12,7 +12,7 @@ import random
import numpy as np
import ray
from ray.tune import Trainable, TrainingResult, register_trainable, \
from ray.tune import Trainable, register_trainable, \
run_experiments, Experiment
from ray.tune.hyperband import HyperBandScheduler
@@ -33,8 +33,8 @@ class MyTrainableClass(Trainable):
v *= self.config["height"]
# Here we use `episode_reward_mean`, but you can also report other
# objectives such as loss or accuracy (see tune/result.py).
return TrainingResult(episode_reward_mean=v, timesteps_this_iter=1)
# objectives such as loss or accuracy.
return {"episode_reward_mean": v}
def _save(self, checkpoint_dir):
path = os.path.join(checkpoint_dir, "checkpoint")
@@ -57,9 +57,10 @@ if __name__ == "__main__":
ray.init()
# Hyperband early stopping, configured with `episode_reward_mean` as the
# objective and `timesteps_total` as the time unit.
# objective and `training_iteration` as the time unit,
# which is automatically filled by Tune.
hyperband = HyperBandScheduler(
time_attr="timesteps_total",
time_attr="training_iteration",
reward_attr="episode_reward_mean",
max_t=100)
python/ray/tune/examples/hyperopt_example.py
+2 -1
View File
@@ -20,7 +20,8 @@ def easy_objective(config, reporter):
for i in range(100):
reporter(
timesteps_total=i,
mean_loss=((config["height"] - 14)**2 + abs(config["width"] - 3)))
neg_mean_loss=-(config["height"] - 14)**2 +
abs(config["width"] - 3))
time.sleep(0.02)
python/ray/tune/examples/pbt_example.py
+3 -5
View File
@@ -11,8 +11,7 @@ import random
import time
import ray
from ray.tune import Trainable, TrainingResult, register_trainable, \
run_experiments
from ray.tune import Trainable, register_trainable, run_experiments
from ray.tune.pbt import PopulationBasedTraining
@@ -35,9 +34,8 @@ class MyTrainableClass(Trainable):
self.current_value += random.gauss(self.config["factor_2"], 1.0)
# Here we use `episode_reward_mean`, but you can also report other
# objectives such as loss or accuracy (see tune/result.py).
return TrainingResult(
episode_reward_mean=self.current_value, timesteps_this_iter=1)
# objectives such as loss or accuracy.
return {"episode_reward_mean": self.current_value}
def _save(self, checkpoint_dir):
path = os.path.join(checkpoint_dir, "checkpoint")
python/ray/tune/examples/pbt_tune_cifar10_with_keras.py
+3 -4
View File
@@ -26,7 +26,6 @@ import ray
from ray.tune import grid_search, run_experiments
from ray.tune import register_trainable
from ray.tune import Trainable
from ray.tune import TrainingResult
from ray.tune.pbt import PopulationBasedTraining
num_classes = 10
@@ -157,7 +156,7 @@ class Cifar10Model(Trainable):
# loss, accuracy
_, accuracy = self.model.evaluate(x_test, y_test, verbose=0)
return TrainingResult(timesteps_this_iter=10, mean_accuracy=accuracy)
return {"mean_accuracy": accuracy}
def _save(self, checkpoint_dir):
file_path = checkpoint_dir + "/model"
@@ -189,7 +188,7 @@ if __name__ == "__main__":
},
"stop": {
"mean_accuracy": 0.80,
"timesteps_total": 300,
"training_iteration": 30,
},
"config": {
"epochs": 1,
@@ -208,7 +207,7 @@ if __name__ == "__main__":
ray.init()
pbt = PopulationBasedTraining(
time_attr="timesteps_total",
time_attr="training_iteration",
reward_attr="mean_accuracy",
perturbation_interval=10,
hyperparam_mutations={
python/ray/tune/examples/tune_mnist_ray_hyperband.py
+3 -4
View File
@@ -31,7 +31,7 @@ import time
import ray
from ray.tune import grid_search, run_experiments, register_trainable, \
Trainable, TrainingResult
Trainable
from ray.tune.hyperband import HyperBandScheduler
from tensorflow.examples.tutorials.mnist import input_data
@@ -196,8 +196,7 @@ class TrainMNIST(Trainable):
})
self.iterations += 1
return TrainingResult(
timesteps_this_iter=10, mean_accuracy=train_accuracy)
return {"mean_accuracy": train_accuracy}
def _save(self, checkpoint_dir):
return self.saver.save(
@@ -234,6 +233,6 @@ if __name__ == '__main__':
ray.init()
hyperband = HyperBandScheduler(
time_attr="timesteps_total", reward_attr="mean_accuracy", max_t=100)
time_attr="training_iteration", reward_attr="mean_accuracy", max_t=10)
run_experiments({'mnist_hyperband_test': mnist_spec}, scheduler=hyperband)
python/ray/tune/experiment.py
+2 -2
View File
@@ -16,8 +16,8 @@ class Experiment(object):
user-defined trainable function or class
registered in the tune registry.
stop (dict): The stopping criteria. The keys may be any field in
TrainingResult, whichever is reached first. Defaults to
empty dict.
the return result of 'train()', whichever is reached first.
Defaults to empty dict.
config (dict): Algorithm-specific configuration
(e.g. env, hyperparams). Defaults to empty dict.
trial_resources (dict): Machine resources to allocate per trial,
python/ray/tune/function_runner.py
+10 -12
View File
@@ -8,7 +8,7 @@ import traceback
from ray.tune import TuneError
from ray.tune.trainable import Trainable
from ray.tune.result import TrainingResult
from ray.tune.result import TIMESTEPS_TOTAL
from ray.tune.util import _serve_get_pin_requests
@@ -26,13 +26,11 @@ class StatusReporter(object):
"""Report updated training status.
Args:
kwargs (TrainingResult): Latest training result status. You must
at least define `timesteps_total`, but probably want to report
some of the other metrics as well.
kwargs: Latest training result status.
"""
with self._lock:
self._latest_result = self._last_result = TrainingResult(**kwargs)
self._latest_result = self._last_result = kwargs.copy()
def _get_and_clear_status(self):
if self._error:
@@ -40,7 +38,8 @@ class StatusReporter(object):
if self._done and not self._latest_result:
if not self._last_result:
raise TuneError("Trial finished without reporting result!")
return self._last_result._replace(done=True)
self._last_result.update(done=True)
return self._last_result
with self._lock:
res = self._latest_result
self._latest_result = None
@@ -112,13 +111,12 @@ class FunctionRunner(Trainable):
_serve_get_pin_requests()
time.sleep(1)
result = self._status_reporter._get_and_clear_status()
if result.timesteps_total is None:
raise TuneError("Must specify timesteps_total in result", result)
result = result._replace(
timesteps_this_iter=(
result.timesteps_total - self._last_reported_timestep))
self._last_reported_timestep = result.timesteps_total
curr_ts_total = result.get(TIMESTEPS_TOTAL,
self._last_reported_timestep)
result.update(
timesteps_this_iter=(curr_ts_total - self._last_reported_timestep))
self._last_reported_timestep = curr_ts_total
return result
python/ray/tune/hyperband.py
+4 -5
View File
@@ -52,11 +52,11 @@ class HyperBandScheduler(FIFOScheduler):
See also: https://people.eecs.berkeley.edu/~kjamieson/hyperband.html
Args:
time_attr (str): The TrainingResult attr to use for comparing time.
time_attr (str): The training result attr to use for comparing time.
Note that you can pass in something non-temporal such as
`training_iteration` as a measure of progress, the only requirement
is that the attribute should increase monotonically.
reward_attr (str): The TrainingResult objective value attribute. As
reward_attr (str): The training result objective value attribute. As
with `time_attr`, this may refer to any objective value. Stopping
procedures will use this attribute.
max_t (int): max time units per trial. Trials will be stopped after
@@ -323,8 +323,7 @@ class Bracket():
self._r = int(min(self._r, self._max_t_attr - self._cumul_r))
self._cumul_r += self._r
sorted_trials = sorted(
self._live_trials,
key=lambda t: getattr(self._live_trials[t], reward_attr))
self._live_trials, key=lambda t: self._live_trials[t][reward_attr])
good, bad = sorted_trials[-self._n:], sorted_trials[:-self._n]
return good, bad
@@ -376,7 +375,7 @@ class Bracket():
def _get_result_time(self, result):
if result is None:
return 0
return getattr(result, self._time_attr)
return result[self._time_attr]
def _calculate_total_work(self, n, r, s):
work = 0
python/ray/tune/logger.py
+20 -16
View File
@@ -8,8 +8,8 @@ import numpy as np
import os
import yaml
from ray.tune.result import TrainingResult
from ray.tune.log_sync import get_syncer
from ray.tune.result import NODE_IP, TRAINING_ITERATION, TIME_TOTAL_S
try:
import tensorflow as tf
@@ -67,7 +67,7 @@ class UnifiedLogger(Logger):
def on_result(self, result):
for logger in self._loggers:
logger.on_result(result)
self._log_syncer.set_worker_ip(result.node_ip)
self._log_syncer.set_worker_ip(result.get(NODE_IP))
self._log_syncer.sync_if_needed()
def close(self):
@@ -96,7 +96,7 @@ class _JsonLogger(Logger):
self.local_out = open(local_file, "w")
def on_result(self, result):
json.dump(result._asdict(), self, cls=_SafeFallbackEncoder)
json.dump(result, self, cls=_SafeFallbackEncoder)
self.write("\n")
def write(self, b):
@@ -125,20 +125,20 @@ class _TFLogger(Logger):
self._file_writer = tf.summary.FileWriter(self.logdir)
def on_result(self, result):
tmp = result._asdict()
tmp = result.copy()
for k in [
"config", "pid", "timestamp", "time_total_s", "timesteps_total"
"config", "pid", "timestamp", TIME_TOTAL_S, TRAINING_ITERATION
]:
del tmp[k] # not useful to tf log these
values = to_tf_values(tmp, ["ray", "tune"])
train_stats = tf.Summary(value=values)
self._file_writer.add_summary(train_stats, result.timesteps_total)
timesteps_value = to_tf_values({
"timesteps_total": result.timesteps_total
self._file_writer.add_summary(train_stats, result[TRAINING_ITERATION])
iteration_value = to_tf_values({
"training_iteration": result[TRAINING_ITERATION]
}, ["ray", "tune"])
timesteps_stats = tf.Summary(value=timesteps_value)
self._file_writer.add_summary(timesteps_stats,
result.training_iteration)
iteration_stats = tf.Summary(value=iteration_value)
self._file_writer.add_summary(iteration_stats,
result[TRAINING_ITERATION])
def flush(self):
self._file_writer.flush()
@@ -149,13 +149,16 @@ class _TFLogger(Logger):
class _VisKitLogger(Logger):
def _init(self):
"""CSV outputted with Headers as first set of results."""
# Note that we assume params.json was already created by JsonLogger
self._file = open(os.path.join(self.logdir, "progress.csv"), "w")
self._csv_out = csv.DictWriter(self._file, TrainingResult._fields)
self._csv_out.writeheader()
self._csv_out = None
def on_result(self, result):
self._csv_out.writerow(result._asdict())
if self._csv_out is None:
self._csv_out = csv.DictWriter(self._file, result.keys())
self._csv_out.writeheader()
self._csv_out.writerow(result.copy())
def close(self):
self._file.close()
@@ -194,9 +197,10 @@ class _SafeFallbackEncoder(json.JSONEncoder):
def pretty_print(result):
result = result._replace(config=None) # drop config from pretty print
result = result.copy()
result.update(config=None) # drop config from pretty print
out = {}
for k, v in result._asdict().items():
for k, v in result.items():
if v is not None:
out[k] = v
python/ray/tune/median_stopping_rule.py
+6 -6
View File
@@ -15,11 +15,11 @@ class MedianStoppingRule(FIFOScheduler):
https://research.google.com/pubs/pub46180.html
Args:
time_attr (str): The TrainingResult attr to use for comparing time.
time_attr (str): The training result attr to use for comparing time.
Note that you can pass in something non-temporal such as
`training_iteration` as a measure of progress, the only requirement
is that the attribute should increase monotonically.
reward_attr (str): The TrainingResult objective value attribute. As
reward_attr (str): The training result objective value attribute. As
with `time_attr`, this may refer to any objective value that
is supposed to increase with time.
grace_period (float): Only stop trials at least this old in time.
@@ -62,7 +62,7 @@ class MedianStoppingRule(FIFOScheduler):
assert not self._hard_stop
return TrialScheduler.CONTINUE # fall back to FIFO
time = getattr(result, self._time_attr)
time = result[self._time_attr]
self._results[trial].append(result)
median_result = self._get_median_result(time)
best_result = self._best_result(trial)
@@ -107,10 +107,10 @@ class MedianStoppingRule(FIFOScheduler):
# TODO(ekl) we could do interpolation to be more precise, but for now
# assume len(results) is large and the time diffs are roughly equal
return np.mean([
getattr(r, self._reward_attr) for r in results
if getattr(r, self._time_attr) <= t_max
r[self._reward_attr] for r in results
if r[self._time_attr] <= t_max
])
def _best_result(self, trial):
results = self._results[trial]
return max(getattr(r, self._reward_attr) for r in results)
return max(r[self._reward_attr] for r in results)
python/ray/tune/pbt.py
+4 -4
View File
@@ -103,11 +103,11 @@ class PopulationBasedTraining(FIFOScheduler):
population.
Args:
time_attr (str): The TrainingResult attr to use for comparing time.
time_attr (str): The training result attr to use for comparing time.
Note that you can pass in something non-temporal such as
`training_iteration` as a measure of progress, the only requirement
is that the attribute should increase monotonically.
reward_attr (str): The TrainingResult objective value attribute. As
reward_attr (str): The training result objective value attribute. As
with `time_attr`, this may refer to any objective value. Stopping
procedures will use this attribute.
perturbation_interval (float): Models will be considered for
@@ -175,13 +175,13 @@ class PopulationBasedTraining(FIFOScheduler):
self._trial_state[trial] = PBTTrialState(trial)
def on_trial_result(self, trial_runner, trial, result):
time = getattr(result, self._time_attr)
time = result[self._time_attr]
state = self._trial_state[trial]
if time - state.last_perturbation_time < self._perturbation_interval:
return TrialScheduler.CONTINUE # avoid checkpoint overhead
score = getattr(result, self._reward_attr)
score = result[self._reward_attr]
state.last_score = score
state.last_perturbation_time = time
lower_quantile, upper_quantile = self._quantiles()
python/ray/tune/result.py
+27 -93
View File
@@ -2,101 +2,35 @@ from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from collections import namedtuple
import os
"""
When using ray.tune with custom training scripts, you must periodically report
training status back to Ray by calling reporter(result).
Most of the fields are optional, the only required one is timesteps_total.
# (Optional/Auto-filled) training is terminated. Filled only if not provided.
DONE = "done"
In RLlib, the supplied algorithms fill in TrainingResult for you.
"""
# (Auto-filled) The hostname of the machine hosting the training process.
HOSTNAME = "hostname"
# Where ray.tune writes result files by default
# (Auto-filled) The node ip of the machine hosting the training process.
NODE_IP = "node_ip"
# (Auto-filled) The pid of the training process.
PID = "pid"
# Number of timesteps in this iteration.
TIMESTEPS_THIS_ITER = "timesteps_this_iter"
# (Optional/Auto-filled) Accumulated time in seconds for this experiment.
TIMESTEPS_TOTAL = "timesteps_total"
# (Auto-filled) Time in seconds this iteration took to run.
# This may be overriden to override the system-computed time difference.
TIME_THIS_ITER_S = "time_this_iter_s"
# (Auto-filled) Accumulated time in seconds for this entire experiment.
TIME_TOTAL_S = "time_total_s"
# (Auto-filled) The index of thistraining iteration.
TRAINING_ITERATION = "training_iteration"
# Where Tune writes result files by default
DEFAULT_RESULTS_DIR = os.path.expanduser("~/ray_results")
TrainingResult = namedtuple(
"TrainingResult",
[
# (Required) Accumulated timesteps for this entire experiment.
"timesteps_total",
# (Optional) If training is terminated.
"done",
# (Optional) Custom metadata to report for this iteration.
"info",
# (Optional) The mean episode reward if applicable.
"episode_reward_mean",
# (Optional) The min episode reward if applicable.
"episode_reward_min",
# (Optional) The max episode reward if applicable.
"episode_reward_max",
# (Optional) The mean episode length if applicable.
"episode_len_mean",
# (Optional) The number of episodes total.
"episodes_total",
# (Optional) Per-policy reward information in multi-agent RL.
"policy_reward_mean",
# (Optional) The current training accuracy if applicable.
"mean_accuracy",
# (Optional) The current validation accuracy if applicable.
"mean_validation_accuracy",
# (Optional) The current training loss if applicable.
"mean_loss",
# (Auto-filled) The negated current training loss.
"neg_mean_loss",
# (Auto-filled) Unique string identifier for this experiment.
# This id is preserved across checkpoint / restore calls.
"experiment_id",
# (Auto-filled) The index of this training iteration,
# e.g. call to train().
"training_iteration",
# (Auto-filled) Number of timesteps in the simulator
# in this iteration.
"timesteps_this_iter",
# (Auto-filled) Time in seconds this iteration took to run. This may
# be overriden in order to override the system-computed
# time difference.
"time_this_iter_s",
# (Auto-filled) Accumulated time in seconds for this entire experiment.
"time_total_s",
# (Auto-filled) The pid of the training process.
"pid",
# (Auto-filled) A formatted date of when the result was processed.
"date",
# (Auto-filled) A UNIX timestamp of when the result was processed.
"timestamp",
# (Auto-filled) The hostname of the machine hosting the
# training process.
"hostname",
# (Auto-filled) The node ip of the machine hosting the
# training process.
"node_ip",
# (Auto=filled) The current hyperparameter configuration.
"config",
])
TrainingResult.__new__.__defaults__ = (None, ) * len(TrainingResult._fields)
python/ray/tune/suggest/hyperopt.py
+2 -2
View File
@@ -29,7 +29,7 @@ class HyperOptSearch(SuggestionAlgorithm):
parameters generated in the variant generation process.
max_concurrent (int): Number of maximum concurrent trials. Defaults
to 10.
reward_attr (str): The TrainingResult objective value attribute.
reward_attr (str): The training result objective value attribute.
This refers to an increasing value, which is internally negated
when interacting with HyperOpt so that HyperOpt can "maximize"
this value.
@@ -111,7 +111,7 @@ class HyperOptSearch(SuggestionAlgorithm):
del self._live_trial_mapping[trial_id]
def _to_hyperopt_result(self, result):
return {"loss": -getattr(result, self._reward_attr), "status": "ok"}
return {"loss": -result[self._reward_attr], "status": "ok"}
def _get_hyperopt_trial(self, trial_id):
if trial_id not in self._live_trial_mapping:
python/ray/tune/suggest/search.py
+1 -1
View File
@@ -43,7 +43,7 @@ class SearchAlgorithm(object):
Arguments:
trial_id: Identifier for the trial.
result (TrainingResult): Defaults to None. A TrainingResult will
result (dict): Defaults to None. A dict will
be provided with this notification when the trial is in
the RUNNING state AND either completes naturally or
by manual termination.
python/ray/tune/test/trial_runner_test.py
+17 -14
View File
@@ -13,7 +13,7 @@ from ray.tune import Trainable, TuneError
from ray.tune import register_env, register_trainable, run_experiments
from ray.tune.trial_scheduler import TrialScheduler, FIFOScheduler
from ray.tune.registry import _global_registry, TRAINABLE_CLASS
from ray.tune.result import DEFAULT_RESULTS_DIR, TrainingResult
from ray.tune.result import DEFAULT_RESULTS_DIR, TIMESTEPS_TOTAL, DONE
from ray.tune.util import pin_in_object_store, get_pinned_object
from ray.tune.experiment import Experiment
from ray.tune.trial import Trial, Resources
@@ -57,7 +57,7 @@ class TrainableFunctionApiTest(unittest.TestCase):
}
})
self.assertEqual(trial.status, Trial.TERMINATED)
self.assertEqual(trial.last_result.timesteps_total, 100)
self.assertEqual(trial.last_result[TIMESTEPS_TOTAL], 100)
def testRegisterEnv(self):
register_env("foo", lambda: None)
@@ -81,7 +81,7 @@ class TrainableFunctionApiTest(unittest.TestCase):
}
})
self.assertEqual(trial.status, Trial.TERMINATED)
self.assertEqual(trial.last_result.timesteps_total, 200)
self.assertEqual(trial.last_result[TIMESTEPS_TOTAL], 200)
def testRegisterTrainable(self):
def train(config, reporter):
@@ -105,7 +105,7 @@ class TrainableFunctionApiTest(unittest.TestCase):
return Resources(cpu=config["cpu"], gpu=config["gpu"])
def _train(self):
return TrainingResult(timesteps_this_iter=1, done=True)
return dict(timesteps_this_iter=1, done=True)
register_trainable("B", B)
@@ -276,7 +276,7 @@ class TrainableFunctionApiTest(unittest.TestCase):
self.assertRaises(TuneError, f)
def testBadReturn(self):
def testBadStoppingReturn(self):
def train(config, reporter):
reporter()
@@ -286,6 +286,9 @@ class TrainableFunctionApiTest(unittest.TestCase):
run_experiments({
"foo": {
"run": "f1",
"stop": {
"time": 10
},
"config": {
"script_min_iter_time_s": 0,
},
@@ -309,7 +312,7 @@ class TrainableFunctionApiTest(unittest.TestCase):
}
})
self.assertEqual(trial.status, Trial.TERMINATED)
self.assertEqual(trial.last_result.timesteps_total, 100)
self.assertEqual(trial.last_result[TIMESTEPS_TOTAL], 100)
def testAbruptReturn(self):
def train(config, reporter):
@@ -325,7 +328,7 @@ class TrainableFunctionApiTest(unittest.TestCase):
}
})
self.assertEqual(trial.status, Trial.TERMINATED)
self.assertEqual(trial.last_result.timesteps_total, 100)
self.assertEqual(trial.last_result[TIMESTEPS_TOTAL], 100)
def testErrorReturn(self):
def train(config, reporter):
@@ -360,7 +363,7 @@ class TrainableFunctionApiTest(unittest.TestCase):
}
})
self.assertEqual(trial.status, Trial.TERMINATED)
self.assertEqual(trial.last_result.timesteps_total, 99)
self.assertEqual(trial.last_result[TIMESTEPS_TOTAL], 99)
class RunExperimentTest(unittest.TestCase):
@@ -393,7 +396,7 @@ class RunExperimentTest(unittest.TestCase):
})
for trial in trials:
self.assertEqual(trial.status, Trial.TERMINATED)
self.assertEqual(trial.last_result.timesteps_total, 99)
self.assertEqual(trial.last_result[TIMESTEPS_TOTAL], 99)
def testExperiment(self):
def train(config, reporter):
@@ -410,7 +413,7 @@ class RunExperimentTest(unittest.TestCase):
})
[trial] = run_experiments(exp1)
self.assertEqual(trial.status, Trial.TERMINATED)
self.assertEqual(trial.last_result.timesteps_total, 99)
self.assertEqual(trial.last_result[TIMESTEPS_TOTAL], 99)
def testExperimentList(self):
def train(config, reporter):
@@ -435,7 +438,7 @@ class RunExperimentTest(unittest.TestCase):
trials = run_experiments([exp1, exp2])
for trial in trials:
self.assertEqual(trial.status, Trial.TERMINATED)
self.assertEqual(trial.last_result.timesteps_total, 99)
self.assertEqual(trial.last_result[TIMESTEPS_TOTAL], 99)
def testSpecifyAlgorithm(self):
"""Tests run_experiments works without specifying experiment."""
@@ -457,7 +460,7 @@ class RunExperimentTest(unittest.TestCase):
trials = run_experiments(search_alg=alg)
for trial in trials:
self.assertEqual(trial.status, Trial.TERMINATED)
self.assertEqual(trial.last_result.timesteps_total, 99)
self.assertEqual(trial.last_result[TIMESTEPS_TOTAL], 99)
class VariantGeneratorTest(unittest.TestCase):
@@ -903,9 +906,9 @@ class TrialRunnerTest(unittest.TestCase):
runner.step()
self.assertEqual(trials[0].status, Trial.RUNNING)
runner.step()
self.assertNotEqual(trials[0].last_result.done, True)
self.assertNotEqual(trials[0].last_result[DONE], True)
runner.step()
self.assertEqual(trials[0].last_result.done, True)
self.assertEqual(trials[0].last_result[DONE], True)
def testPauseThenResume(self):
ray.init(num_cpus=1, num_gpus=1)
python/ray/tune/test/trial_scheduler_test.py
+4 -5
View File
@@ -11,7 +11,6 @@ from ray.tune.hyperband import HyperBandScheduler
from ray.tune.async_hyperband import AsyncHyperBandScheduler
from ray.tune.pbt import PopulationBasedTraining, explore
from ray.tune.median_stopping_rule import MedianStoppingRule
from ray.tune.result import TrainingResult
from ray.tune.trial import Trial, Resources
from ray.tune.trial_scheduler import TrialScheduler
@@ -20,7 +19,7 @@ _register_all()
def result(t, rew):
return TrainingResult(
return dict(
time_total_s=t, episode_reward_mean=rew, training_iteration=int(t))
@@ -126,7 +125,7 @@ class EarlyStoppingSuite(unittest.TestCase):
def testAlternateMetrics(self):
def result2(t, rew):
return TrainingResult(training_iteration=t, neg_mean_loss=rew)
return dict(training_iteration=t, neg_mean_loss=rew)
rule = MedianStoppingRule(
grace_period=0,
@@ -465,7 +464,7 @@ class HyperbandSuite(unittest.TestCase):
"""Checking that alternate metrics will pass."""
def result2(t, rew):
return TrainingResult(time_total_s=t, neg_mean_loss=rew)
return dict(time_total_s=t, neg_mean_loss=rew)
sched = HyperBandScheduler(
time_attr='time_total_s', reward_attr='neg_mean_loss')
@@ -855,7 +854,7 @@ class AsyncHyperBandSuite(unittest.TestCase):
def testAlternateMetrics(self):
def result2(t, rew):
return TrainingResult(training_iteration=t, neg_mean_loss=rew)
return dict(training_iteration=t, neg_mean_loss=rew)
scheduler = AsyncHyperBandScheduler(
grace_period=1,
python/ray/tune/trainable.py
+43 -22
View File
@@ -14,9 +14,9 @@ import time
import uuid
import ray
from ray.tune import TuneError
from ray.tune.logger import UnifiedLogger
from ray.tune.result import DEFAULT_RESULTS_DIR
from ray.tune.result import (DEFAULT_RESULTS_DIR, TIME_THIS_ITER_S,
TIMESTEPS_THIS_ITER, DONE, TIMESTEPS_TOTAL)
from ray.tune.trial import Resources
@@ -102,11 +102,31 @@ class Trainable(object):
"""Runs one logical iteration of training.
Subclasses should override ``_train()`` instead to return results.
This method auto-fills many fields, so only ``timesteps_this_iter``
is required to be present.
This class automatically fills the following fields in the result:
done (bool): training is terminated. Filled only if not provided.
time_this_iter_s (float): Time in seconds
this iteration took to run. This may be overriden in order to
override the system-computed time difference.
time_total_s (float): Accumulated time in seconds
for this entire experiment.
experiment_id (str): Unique string identifier
for this experiment. This id is preserved
across checkpoint / restore calls.
training_iteration (int): The index of this
training iteration, e.g. call to train().
pid (str): The pid of the training process.
date (str): A formatted date of
when the result was processed.
timestamp (str): A UNIX timestamp of
when the result was processed.
hostname (str): The hostname of the machine
hosting the training process.
node_ip (str): The node ip of the machine
hosting the training process.
Returns:
A TrainingResult that describes training progress.
A dict that describes training progress.
"""
if not self._initialize_ok:
@@ -115,35 +135,33 @@ class Trainable(object):
start = time.time()
result = self._train()
result = result.copy()
self._iteration += 1
if result.time_this_iter_s is not None:
time_this_iter = result.time_this_iter_s
if result.get(TIME_THIS_ITER_S) is not None:
time_this_iter = result[TIME_THIS_ITER_S]
else:
time_this_iter = time.time() - start
if result.timesteps_this_iter is None:
raise TuneError("Must specify timesteps_this_iter in result",
result)
self._time_total += time_this_iter
self._timesteps_total += result.timesteps_this_iter
# Include the negative loss to use as a stopping condition
if result.mean_loss is not None:
neg_loss = -result.mean_loss
else:
neg_loss = result.neg_mean_loss
self._timesteps_total += result.get(TIMESTEPS_THIS_ITER, 0)
result.setdefault(DONE, False)
result.setdefault(TIMESTEPS_TOTAL, self._timesteps_total)
# Provides auto-filled neg_mean_loss for avoiding regressions
if result.get("mean_loss"):
result.setdefault("neg_mean_loss", -result["mean_loss"])
now = datetime.today()
result = result._replace(
result.update(
experiment_id=self._experiment_id,
date=now.strftime("%Y-%m-%d_%H-%M-%S"),
timestamp=int(time.mktime(now.timetuple())),
training_iteration=self._iteration,
timesteps_total=self._timesteps_total,
time_this_iter_s=time_this_iter,
time_total_s=self._time_total,
neg_mean_loss=neg_loss,
pid=os.getpid(),
hostname=os.uname()[1],
node_ip=self._local_ip,
@@ -247,7 +265,10 @@ class Trainable(object):
self._stop()
def _train(self):
"""Subclasses should override this to implement train()."""
"""Subclasses should override this to implement train().
Returns:
A dict that describes training progress."""
raise NotImplementedError
python/ray/tune/trial.py
+19 -23
View File
@@ -16,7 +16,8 @@ from ray.tune.logger import NoopLogger, UnifiedLogger, pretty_print
# need because there are cyclic imports that may cause specific names to not
# have been defined yet. See https://github.com/ray-project/ray/issues/1716.
import ray.tune.registry
from ray.tune.result import TrainingResult, DEFAULT_RESULTS_DIR
from ray.tune.result import (DEFAULT_RESULTS_DIR, DONE, HOSTNAME, PID,
TIME_TOTAL_S, TRAINING_ITERATION)
from ray.utils import random_string, binary_to_hex
DEBUG_PRINT_INTERVAL = 5
@@ -101,13 +102,6 @@ class Trial(object):
if not has_trainable(trainable_name):
raise TuneError("Unknown trainable: " + trainable_name)
if stopping_criterion:
for k in stopping_criterion:
if k not in TrainingResult._fields:
raise TuneError(
"Stopping condition key `{}` must be one of {}".format(
k, TrainingResult._fields))
# Trial config
self.trainable_name = trainable_name
self.config = config or {}
@@ -237,11 +231,13 @@ class Trial(object):
def should_stop(self, result):
"""Whether the given result meets this trial's stopping criteria."""
if result.done:
if result.get(DONE):
return True
for criteria, stop_value in self.stopping_criterion.items():
if getattr(result, criteria) >= stop_value:
if criteria not in result:
raise TuneError("Stopping Criteria not provided in result.")
if result[criteria] >= stop_value:
return True
return False
@@ -252,7 +248,7 @@ class Trial(object):
if not self.checkpoint_freq:
return False
return self.last_result.training_iteration % self.checkpoint_freq == 0
return self.last_result[TRAINING_ITERATION] % self.checkpoint_freq == 0
def progress_string(self):
"""Returns a progress message for printing out to the console."""
@@ -269,23 +265,23 @@ class Trial(object):
pieces = [
'{} [{}]'.format(
self._status_string(),
location_string(self.last_result.hostname,
self.last_result.pid)),
'{} s'.format(int(self.last_result.time_total_s)), '{} ts'.format(
int(self.last_result.timesteps_total))
location_string(
self.last_result.get(HOSTNAME),
self.last_result.get(PID))),
'{} s'.format(int(self.last_result.get(TIME_TOTAL_S))),
]
if self.last_result.episode_reward_mean is not None:
if self.last_result.get("episode_reward_mean") is not None:
pieces.append('{} rew'.format(
format(self.last_result.episode_reward_mean, '.3g')))
format(self.last_result["episode_reward_mean"], '.3g')))
if self.last_result.mean_loss is not None:
if self.last_result.get("mean_loss") is not None:
pieces.append('{} loss'.format(
format(self.last_result.mean_loss, '.3g')))
format(self.last_result["mean_loss"], '.3g')))
if self.last_result.mean_accuracy is not None:
if self.last_result.get("mean_accuracy") is not None:
pieces.append('{} acc'.format(
format(self.last_result.mean_accuracy, '.3g')))
format(self.last_result["mean_accuracy"], '.3g')))
return ', '.join(pieces)
@@ -350,10 +346,10 @@ class Trial(object):
def update_last_result(self, result, terminate=False):
if terminate:
result = result._replace(done=True)
result.update(done=True)
if self.verbose and (terminate or time.time() - self.last_debug >
DEBUG_PRINT_INTERVAL):
print("TrainingResult for {}:".format(self))
print("Result for {}:".format(self))
print(" {}".format(pretty_print(result).replace("\n", "\n ")))
self.last_debug = time.time()
self.last_result = result
python/ray/tune/trial_runner.py
+2 -1
View File
@@ -9,6 +9,7 @@ import time
import traceback
from ray.tune import TuneError
from ray.tune.result import TIME_THIS_ITER_S
from ray.tune.web_server import TuneServer
from ray.tune.trial import Trial, Resources
from ray.tune.trial_scheduler import FIFOScheduler, TrialScheduler
@@ -262,7 +263,7 @@ class TrialRunner(object):
trial = self._running.pop(result_id)
try:
result = ray.get(result_id)
self._total_time += result.time_this_iter_s
self._total_time += result[TIME_THIS_ITER_S]
if trial.should_stop(result):
# Hook into scheduler
python/ray/tune/web_server.py
+1 -1
View File
@@ -87,7 +87,7 @@ def RunnerHandler(runner):
def trial_info(self, trial):
if trial.last_result:
result = trial.last_result._asdict()
result = trial.last_result.copy()
else:
result = None
info_dict = {