Gathers new episodes metrics tuples from the given evaluators.

def collect_episodes(local_evaluator=None,
                     remote_evaluators=[],
                     timeout_seconds=180):
    """Gathers new episodes metrics tuples from the given evaluators."""

    pending = [
        a.apply.remote(lambda ev: ev.get_metrics()) for a in remote_evaluators
    ]
    collected, _ = ray.wait(
        pending, num_returns=len(pending), timeout=timeout_seconds * 1.0)
    num_metric_batches_dropped = len(pending) - len(collected)
    if pending and len(collected) == 0:
        raise ValueError(
            "Timed out waiting for metrics from workers. You can configure "
            "this timeout with `collect_metrics_timeout`.")

    metric_lists = ray_get_and_free(collected)
    if local_evaluator:
        metric_lists.append(local_evaluator.get_metrics())
    episodes = []
    for metrics in metric_lists:
        episodes.extend(metrics)
    return episodes, num_metric_batches_dropped