code: def load_model(
      bento_model: str | Tag | bentoml.Model,
      device_id: str | torch.device | None = None,
      pipeline_class: str
      | type[
          diffusers.pipelines.DiffusionPipeline
      ] = diffusers.DiffusionPipeline,
      device_map: str
      | dict[str, int | str | torch.device]
      | None = None,
      custom_pipeline: str | None = None,
      scheduler_class: type[diffusers.SchedulerMixin] | None = None,
      torch_dtype: str | torch.dtype | None = None,
      low_cpu_mem_usage: bool | None = None,
      enable_xformers: bool = False,
      enable_attention_slicing: int | str | None = None,
      enable_model_cpu_offload: bool | None = None,
      enable_sequential_cpu_offload: bool | None = None,
      enable_torch_compile: bool | None = None,
      variant: str | None = None,
      lora_weights: LoraOptionType | list[LoraOptionType] | None = None,
      textual_inversions: TextualInversionOptionType
      | list[TextualInversionOptionType]
      | None = None,
      load_pretrained_extra_kwargs: dict[str, t.Any] | None = None,
  ) -> diffusers.DiffusionPipeline:
      """
      Load a Diffusion model and convert it to diffusers `Pipeline <https://huggingface.co/docs/diffusers/api/pipelines/overview>`_
      with the given tag from the local BentoML model store.
  
      Args:
          bento_model:
              Either the tag of the model to get from the store, or a BentoML
              ``~bentoml.Model`` instance to load the model from.
          device_id (:code:`str`, `optional`, default to :code:`None`):
              Optional devices to put the given model on. Refer to `device attributes <https://pytorch.org/docs/stable/tensor_attributes.html#torch.torch.device>`_.
          pipeline_class (:code:`type[diffusers.DiffusionPipeline]`, `optional`):
              DiffusionPipeline Class use to load the saved diffusion model, default to
              ``diffusers.DiffusionPipeline``. For more pipeline types, refer to
              `Pipeline Overview <https://huggingface.co/docs/diffusers/api/pipelines/overview>`_
          device_map (:code:`None | str | Dict[str, Union[int, str, torch.device]]`, `optional`):
              A map that specifies where each submodule should go. For more information, refer to
              `device_map <https://huggingface.co/docs/diffusers/main/en/api/diffusion_pipeline#diffusers.DiffusionPipeline.from_pretrained.device_map>`_
          custom_pipeline (:code:`None | str`, `optional`):
              An identifier of custom pipeline hosted on github. For a list of community
              maintained custom piplines, refer to https://github.com/huggingface/diffusers/tree/main/examples/community
          scheduler_class (:code:`type[diffusers.SchedulerMixin]`, `optional`):
              Scheduler Class to be used by DiffusionPipeline
          torch_dtype (:code:`str | torch.dtype`, `optional`):
              Override the default `torch.dtype` and load the model under this dtype.
          low_cpu_mem_usage (:code:`bool`, `optional`):
              Speed up model loading by not initializing the weights and only loading the
              pre-trained weights. defaults to `True` if torch version >= 1.9.0 else `False`
          enable_xformers (:code:`bool`, `optional`):
              Use xformers optimization if it's available. For more info, refer to
              https://github.com/facebookresearch/xformers
          variant (:code:`str`, *optional*):
              If specified load weights from `variant` filename, *e.g.* pytorch_model.<variant>.bin.
          lora_weights (:code:`LoraOptionType | list[LoraOptionType]` *optional*):
              lora weights to be loaded. :code:`LoraOptionType` can be either a string or a dictionary.
              When it's a string, it represents a path to the weight file. When it's a dictionary, it
              contains a key :code`"model_name"` pointing to a huggingface repository or a local directory,
              a key :code:`weight_name` pointing the weight file and other keys that will be passed to
              pipeline's :code:`load_lora_weights` method.
          textual_inversions (:code:`TextualInversionOptionType | list[TextualInversionOptionType]` *optional*):
              Textual inversions to be loaded. :code:`TextualInversionOptionType` can be either a string or a dictionary.
              When it's a string, it represents a path to the weight file. When it's a dictionary, it
              contains a key :code`"model_name"` pointing to a huggingface repository or a local directory,
              a key :code:`weight_name` pointing the weight file and other keys that will be passed to
              pipeline's :code:`load_lora_weights` method.
          load_pretrained_extra_kwargs: (:code:`dict[str, t.Any]`, *optional*):
              Extra kwargs passed to Pipeline class's :code:`from_pretrained` method
  
      Returns:
          The Diffusion model loaded as diffusers pipeline from the BentoML model store.
  
      Example:
  
      .. code-block:: python
  
          import bentoml
          pipeline = bentoml.diffusers.load_model('my_diffusers_model:latest')
          pipeline(prompt)
      """  # noqa
      if not isinstance(bento_model, bentoml.Model):
          bento_model = get(bento_model)
  
      if bento_model.info.module not in (MODULE_NAME, __name__):
          raise NotFound(
              f"Model {bento_model.tag} was saved with module {bento_model.info.module}, not loading with {MODULE_NAME}."
          )
  
      if isinstance(pipeline_class, str):
          pipeline_class = t.cast(
              type[diffusers.DiffusionPipeline],
              _str2cls(pipeline_class),
          )
  
      if lora_weights:
          if not issubclass(pipeline_class, LoraLoaderMixin):
              raise NotImplementedError(
                  f"Class {pipeline_class} is not a subclass of LoraLoaderMixin, cannot load textual inversions"
              )
  
      if textual_inversions:
          if not issubclass(
              pipeline_class, TextualInversionLoaderMixin
          ):
              raise NotImplementedError(
                  f"Class {pipeline_class} is not a subclass of TextualInversionLoaderMixin, cannot load lora weights"
              )
  
      diffusion_model_dir = bento_model.path_of(DIFFUSION_MODEL_FOLDER)
  
      if low_cpu_mem_usage is None:
          if (
              is_torch_version(">=", "1.9.0")
              and is_accelerate_available()
          ):
              low_cpu_mem_usage = True
          else:
              low_cpu_mem_usage = False
  
      load_pretrained_extra_kwargs = load_pretrained_extra_kwargs or {}
      pipeline: diffusers.DiffusionPipeline = (
          pipeline_class.from_pretrained(
              diffusion_model_dir,
              torch_dtype=torch_dtype,
              low_cpu_mem_usage=low_cpu_mem_usage,
              device_map=device_map,
              custom_pipeline=custom_pipeline,
              variant=variant,
              **load_pretrained_extra_kwargs,
          )
      )
  
      if scheduler_class:
          scheduler: diffusers.SchedulerMixin = (
              scheduler_class.from_config(pipeline.scheduler.config)
          )
          pipeline.scheduler = scheduler
  
      if device_id is not None:
          move_model_to_device = True
  
          if str(device_id).lower().startswith("cuda"):
              # when device_map is not None, we should not move the
              # pipeline to gpu again see
              # https://github.com/huggingface/diffusers/issues/2782
              if device_map is not None:
                  move_model_to_device = False
              if enable_sequential_cpu_offload:
                  move_model_to_device = False
              if enable_model_cpu_offload:
                  move_model_to_device = False
  
          if move_model_to_device:
              pipeline = pipeline.to(device_id)
  
      if enable_xformers:
          pipeline.enable_xformers_memory_efficient_attention()
  
      if enable_sequential_cpu_offload:
          pipeline.enable_sequential_cpu_offload()
  
      if enable_model_cpu_offload:
          pipeline.enable_model_cpu_offload()
  
      if enable_attention_slicing is not None:
          pipeline.enable_attention_slicing(enable_attention_slicing)
  
      if enable_torch_compile:
          logger.info("Run torch compile on unet")
          pipeline.unet = torch.compile(
              pipeline.unet, mode="reduce-overhead", fullgraph=True
          )
  
      if lora_weights:
          _load_lora_weights_to_pipeline(pipeline, lora_weights)
  
      if textual_inversions:
          if not isinstance(textual_inversions, list):
              textual_inversions = [textual_inversions]
  
          for textual_inversion in textual_inversions:
              model_name, kwargs = _prepare_textual_inversion_args(
                  textual_inversion
              )
              pipeline.load_textual_inversion(model_name, **kwargs)
  
      return pipeline
  
imports: from __future__ import annotations
  import logging
  import os
  import shutil
  import typing
  from typing import TYPE_CHECKING
  import attr
  import bentoml
  from bentoml import Tag
  from bentoml.exceptions import BentoMLException
  from bentoml.exceptions import MissingDependencyException
  from bentoml.exceptions import NotFound
  from bentoml.models import ModelContext
  from ..models.model import PartialKwargsModelOptions
  from .utils.transformers import extract_commit_hash
label_desc: Use image node_id 6 for calling a global function with example usage: load_model(bento_model, device_id, pipeline_class, device_map, custom_pipeline, scheduler_class, torch_dtype, low_cpu_mem_usage, enable_xformers, enable_attention_slicing, enable_model_cpu_offload, enable_sequential_cpu_offload, enable_torch_compile, variant, lora_weights, textual_inversions, load_pretrained_extra_kwargs) and returns: pipeline