code: def run(
      checkpoint: str = cli.Checkpoint,
      algo: str = cli.Algo,
      env: str = cli.Env,
      local_mode: bool = cli.LocalMode,
      render: bool = cli.Render,
      steps: int = cli.Steps,
      episodes: int = cli.Episodes,
      out: str = cli.Out,
      config: str = cli.Config,
      save_info: bool = cli.SaveInfo,
      use_shelve: bool = cli.UseShelve,
      track_progress: bool = cli.TrackProgress,
  ):
      if use_shelve and not out:
          raise ValueError(
              "If you set --use-shelve, you must provide an output file via "
              "--out as well!"
          )
      if track_progress and not out:
          raise ValueError(
              "If you set --track-progress, you must provide an output file via "
              "--out as well!"
          )
      # Load configuration from checkpoint file.
      config_args = json.loads(config)
      config_path = ""
      if checkpoint:
          config_dir = os.path.dirname(checkpoint)
          config_path = os.path.join(config_dir, "params.pkl")
          # Try parent directory.
          if not os.path.exists(config_path):
              config_path = os.path.join(config_dir, "../params.pkl")
  
      # Load the config from pickled.
      if os.path.exists(config_path):
          with open(config_path, "rb") as f:
              config = cloudpickle.load(f)
      # If no pkl file found, require command line `--config`.
      else:
          # If no config in given checkpoint -> Error.
          if checkpoint:
              raise ValueError(
                  "Could not find params.pkl in either the checkpoint dir or "
                  "its parent directory AND no `--config` given on command "
                  "line!"
              )
          # Use default config for given agent.
          if not algo:
              raise ValueError(
                  "Please provide an algorithm via `--algo`."
              )
          algo_cls = get_trainable_cls(algo)
          config = algo_cls.get_default_config()
  
      # Make sure worker 0 has an Env.
      config["create_env_on_driver"] = True
  
      # Merge with `evaluation_config` (first try from command line, then from
      # pkl file).
      evaluation_config = copy.deepcopy(
          config_args.get(
              "evaluation_config", config.get("evaluation_config", {})
          )
      )
      config = merge_dicts(config, evaluation_config)
      # Merge with command line `--config` settings (if not already the same anyways).
      config = merge_dicts(config, config_args)
      if not env:
          if not config.get("env"):
              raise ValueError(
                  "You either need to provide an --env argument or pass"
                  "an `env` key with a valid environment to your `config`"
                  "argument."
              )
          env = config.get("env")
  
      # Make sure we have evaluation workers.
      if not config.get("evaluation_num_workers"):
          config["evaluation_num_workers"] = config.get(
              "num_workers", 0
          )
      if not config.get("evaluation_duration"):
          config["evaluation_duration"] = 1
  
      # Hard-override this as it raises a warning by Algorithm otherwise.
      # Makes no sense anyways, to have it set to None as we don't call
      # `Algorithm.train()` here.
      config["evaluation_interval"] = 1
  
      # Rendering settings.
      config["render_env"] = render
  
      ray.init(local_mode=local_mode)
  
      # Create the Algorithm from config.
      cls = get_trainable_cls(algo)
      algorithm = cls(config=config)
  
      # Load state from checkpoint, if provided.
      if checkpoint:
          if os.path.isdir(checkpoint):
              checkpoint_dir = checkpoint
          else:
              checkpoint_dir = str(Path(checkpoint).parent)
          print(f"Restoring algorithm from {checkpoint_dir}")
          restore_result = _TrainingResult(
              checkpoint=Checkpoint.from_directory(checkpoint_dir),
              metrics={},
          )
          algorithm.restore(restore_result)
  
      # Do the actual rollout.
      with RolloutSaver(
          outfile=out,
          use_shelve=use_shelve,
          write_update_file=track_progress,
          target_steps=steps,
          target_episodes=episodes,
          save_info=save_info,
      ) as saver:
          rollout(algorithm, env, steps, episodes, saver, not render)
      algorithm.stop()
  
imports: import collections
  import copy
  import gymnasium
  import json
  import os
  from pathlib import Path
  import shelve
  import typer
  import ray
  import ray.cloudpickle
  from ray.rllib.env import MultiAgentEnv
  from ray.rllib.env.base_env import _DUMMY_AGENT_ID
  from ray.rllib.env.env_context import EnvContext
  from ray.rllib.evaluation.worker_set import WorkerSet
  from ray.rllib.policy.sample_batch import DEFAULT_POLICY_ID
  from ray.rllib.utils.spaces.space_utils import flatten_to_single_ndarray
  from ray.rllib.common import CLIArguments
  from ray.train._checkpoint import Checkpoint
  from ray.train._internal.session import _TrainingResult
  from ray.tune.utils import merge_dicts
  from ray.tune.registry import get_trainable_cls, _global_registry, ENV_CREATOR
label_desc: Use image node_id 1 for calling a global function with example usage: run(checkpoint, algo, env, local_mode, render, steps, episodes, out, config, save_info, use_shelve, track_progress) without return types