code: def run_api_experiment(
      config,
      dataset,
      backend_config,
      predict=False,
      evaluate=True,
      skip_save_processed_input=True,
      skip_save_predictions=True,
      required_metrics=None,
  ):
      # Sanity check that we get 4 slots over 1 host
      kwargs = get_trainer_kwargs()
      if torch.cuda.device_count() > 0:
          assert (
              kwargs.get("num_workers") == torch.cuda.device_count()
          ), kwargs
          assert kwargs.get("use_gpu"), kwargs
      else:
          assert kwargs.get("num_workers") == 1, kwargs
          assert not kwargs.get("use_gpu"), kwargs
  
      # Train on Parquet
      model = train_with_backend(
          backend_config,
          config,
          dataset=dataset,
          evaluate=evaluate,
          predict=predict,
          skip_save_processed_input=skip_save_processed_input,
          skip_save_predictions=skip_save_predictions,
          required_metrics=required_metrics,
      )
  
      assert isinstance(model.backend, RayBackend)
      if isinstance(model.backend.df_engine, DaskEngine):
          assert (
              model.backend.df_engine.parallelism
              == backend_config["processor"]["parallelism"]
          )
  
      return model
  
imports: import copy
  import os
  import tempfile
  from typing import Literal, Optional, Union
  import numpy
  import pandas
  import pytest
  import torch
  from ludwig.api import LudwigModel
  from ludwig.backend import create_ray_backend, initialize_backend, LOCAL_BACKEND
  from ludwig.constants import AUDIO, BAG, BALANCE_PERCENTAGE_TOLERANCE, BATCH_SIZE, BFILL, BINARY, CATEGORY, COLUMN, DATE, H3, IMAGE, MAX_BATCH_SIZE_DATASET_FRACTION, NAME, NUMBER, PREPROCESSING, SEQUENCE, SET, SPLIT, TEXT, TIMESERIES, TRAINER, VECTOR
  from ludwig.data.preprocessing import balance_data
  from ludwig.data.split import DEFAULT_PROBABILITIES
  from ludwig.utils.data_utils import read_parquet
  from ludwig.utils.misc_utils import merge_dict
  from tests.integration_tests.utils import audio_feature, augment_dataset_with_none, bag_feature, binary_feature, category_feature, create_data_set_to_use, date_feature, generate_data, h3_feature, image_feature, number_feature, RAY_BACKEND_CONFIG, sequence_feature, set_feature, text_feature, timeseries_feature, train_with_backend, vector_feature
  import dask
  import ray
  import ray.exceptions
  from ray.air.config import DatasetConfig
  from ray.data import Dataset, DatasetPipeline
  from ray.train._internal.dataset_spec import DataParallelIngestSpec
  from ludwig.backend.ray import get_trainer_kwargs, RayBackend
  from ludwig.data.dataframe.dask import DaskEngine
label_desc: Use image node_id 3 for calling a global function with example usage: run_api_experiment(config, dataset, backend_config, predict, evaluate, skip_save_processed_input, skip_save_predictions, required_metrics) and returns: model