<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 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