code: def create_remote_object(
      self, name: str, object_cls, *args, **kwargs
  ):
      task_state_actor = self._get_task_state_actor()
      r = task_state_actor.create_remote_object.remote(
          name, object_cls, *args, **kwargs
      )
      # Make sure the actor is created. The remote object may not be created
      # when get_remote_object from worker because the callers of
      # create_remote_object and get_remote_object are not in the same worker.
      # Use sync Ray actor requires this `ray.get`, too.
      ray.get(r)
      return _RayRemoteObjectWrapper(task_state_actor, name)
  
imports: import logging
  from dataclasses import asdict
  from typing import Dict, List, Callable
  from .....core.context import Context
  from .....storage.base import StorageLevel
  from .....typing import ChunkType
  from .....utils import implements, lazy_import, sync_to_async
  from ....context import ThreadedServiceContext
  from .config import RayExecutionConfig
label_desc: Use image node_id 3 for calling the _RayRemoteObjectContext obj's underlying member method code with example usage: obj.create_remote_object(name, object_cls) and returns: _RayRemoteObjectWrapper