We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Load local artefacts in HuggingFace runtime
Support loading artifacts from provided model-URI

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `runtimes/huggingface/mlserver_huggingface/common.py`
Content:
```
1 import json
2 import numpy as np
3 
4 from typing import Callable
5 from functools import partial
6 from mlserver.settings import ModelSettings
7 
8 import torch
9 import tensorflow as tf
10 
11 from optimum.pipelines import pipeline as opt_pipeline
12 from transformers.pipelines import pipeline as trf_pipeline
13 from transformers.pipelines.base import Pipeline
14 
15 from .settings import HuggingFaceSettings
16 
17 
18 OPTIMUM_ACCELERATOR = "ort"
19 
20 _PipelineConstructor = Callable[..., Pipeline]
21 
22 
23 def load_pipeline_from_settings(
24     hf_settings: HuggingFaceSettings, settings: ModelSettings
25 ) -> Pipeline:
26     # TODO: Support URI for locally downloaded artifacts
27     # uri = model_parameters.uri
28     pipeline = _get_pipeline_class(hf_settings)
29 
30     batch_size = 1
31     if settings.max_batch_size:
32         batch_size = settings.max_batch_size
33 
34     tokenizer = hf_settings.pretrained_tokenizer
35     if not tokenizer:
36         tokenizer = hf_settings.pretrained_model
37     if hf_settings.framework == "tf":
38         if hf_settings.inter_op_threads is not None:
39             tf.config.threading.set_inter_op_parallelism_threads(
40                 hf_settings.inter_op_threads
41             )
42         if hf_settings.intra_op_threads is not None:
43             tf.config.threading.set_intra_op_parallelism_threads(
44                 hf_settings.intra_op_threads
45             )
46     elif hf_settings.framework == "pt":
47         if hf_settings.inter_op_threads is not None:
48             torch.set_num_interop_threads(hf_settings.inter_op_threads)
49         if hf_settings.intra_op_threads is not None:
50             torch.set_num_threads(hf_settings.intra_op_threads)
51 
52     hf_pipeline = pipeline(
53         hf_settings.task_name,
54         model=hf_settings.pretrained_model,
55         tokenizer=tokenizer,
56         device=hf_settings.device,
57         batch_size=batch_size,
58         framework=hf_settings.framework,
59     )
60 
61     # If max_batch_size > 0 we need to ensure tokens are padded
62     if settings.max_batch_size:
63         model = hf_pipeline.model
64         eos_token_id = model.config.eos_token_id
65         hf_pipeline.tokenizer.pad_token_id = [str(eos_token_id)]  # type: ignore
66 
67     return hf_pipeline
68 
69 
70 def _get_pipeline_class(hf_settings: HuggingFaceSettings) -> _PipelineConstructor:
71     if hf_settings.optimum_model:
72         return partial(opt_pipeline, accelerator=OPTIMUM_ACCELERATOR)
73 
74     return trf_pipeline
75 
76 
77 class NumpyEncoder(json.JSONEncoder):
78     def default(self, obj):
79         if isinstance(obj, np.ndarray):
80             return obj.tolist()
81         return json.JSONEncoder.default(self, obj)
82
```

Path: `runtimes/huggingface/mlserver_huggingface/settings.py`
Content:
```
1 import os
2 import orjson
3 
4 from typing import Optional, Dict, Union, NewType
5 from pydantic import BaseSettings
6 from distutils.util import strtobool
7 from transformers.pipelines import SUPPORTED_TASKS
8 
9 try:
10     # Optimum 1.7 changed the import name from `SUPPORTED_TASKS` to
11     # `ORT_SUPPORTED_TASKS`.
12     # We'll try to import the more recent one, falling back to the previous
13     # import name if not present.
14     # https://github.com/huggingface/optimum/blob/987b02e4f6e2a1c9325b364ff764da2e57e89902/optimum/pipelines/__init__.py#L18
15     from optimum.pipelines import ORT_SUPPORTED_TASKS as SUPPORTED_OPTIMUM_TASKS
16 except ImportError:
17     from optimum.pipelines import SUPPORTED_TASKS as SUPPORTED_OPTIMUM_TASKS
18 
19 from mlserver.settings import ModelSettings
20 
21 from .errors import (
22     MissingHuggingFaceSettings,
23     InvalidTransformersTask,
24     InvalidOptimumTask,
25     InvalidModelParameter,
26     InvalidModelParameterType,
27 )
28 
29 ENV_PREFIX_HUGGINGFACE_SETTINGS = "MLSERVER_MODEL_HUGGINGFACE_"
30 PARAMETERS_ENV_NAME = "PREDICTIVE_UNIT_PARAMETERS"
31 
32 
33 class HuggingFaceSettings(BaseSettings):
34     """
35     Parameters that apply only to HuggingFace models
36     """
37 
38     class Config:
39         env_prefix = ENV_PREFIX_HUGGINGFACE_SETTINGS
40 
41     # TODO: Document fields
42     task: str = ""
43     """
44     Pipeline task to load.
45     You can see the available Optimum and Transformers tasks available in the
46     links below:
47 
48     - `Optimum Tasks <https://huggingface.co/docs/optimum/onnxruntime/usage_guides/pipelines#inference-pipelines-with-the-onnx-runtime-accelerator>`_
49     - `Transformer Tasks <https://huggingface.co/docs/transformers/task_summary>`_
50     """  # noqa: E501
51 
52     task_suffix: str = ""
53     """
54     Suffix to append to the base task name.
55     Useful for, e.g. translation tasks which require a suffix on the task name
56     to specify source and target.
57     """
58 
59     pretrained_model: Optional[str] = None
60     """
61     Name of the model that should be loaded in the pipeline.
62     """
63 
64     pretrained_tokenizer: Optional[str] = None
65     """
66     Name of the tokenizer that should be loaded in the pipeline.
67     """
68 
69     framework: Optional[str] = None
70     """
71     The framework to use, either "pt" for PyTorch or "tf" for TensorFlow.
72     """
73 
74     optimum_model: bool = False
75     """
76     Flag to decide whether the pipeline should use a Optimum-optimised model or
77     the standard Transformers model.
78     Under the hood, this will enable the model to use the optimised ONNX
79     runtime.
80     """
81 
82     device: int = -1
83     """
84     Device in which this pipeline will be loaded (e.g., "cpu", "cuda:1", "mps",
85     or a GPU ordinal rank like 1).
86     """
87 
88     inter_op_threads: Optional[int] = None
89     """
90     Threads used for parallelism between independent operations.
91     PyTorch:
92     https://pytorch.org/docs/stable/notes/cpu_threading_torchscript_inference.html
93     Tensorflow:
94     https://www.tensorflow.org/api_docs/python/tf/config/threading/set_inter_op_parallelism_threads
95     """
96 
97     intra_op_threads: Optional[int] = None
98     """
99     Threads used within an individual op for parallelism.
100     PyTorch:
101     https://pytorch.org/docs/stable/notes/cpu_threading_torchscript_inference.html
102     Tensorflow:
103     https://www.tensorflow.org/api_docs/python/tf/config/threading/set_intra_op_parallelism_threads
104     """
105 
106     @property
107     def task_name(self):
108         if self.task == "translation":
109             return f"{self.task}{self.task_suffix}"
110         return self.task
111 
112 
113 EXTRA_TYPE_DICT = {
114     "INT": int,
115     "FLOAT": float,
116     "DOUBLE": float,
117     "STRING": str,
118     "BOOL": bool,
119 }
120 
121 ExtraDict = NewType("ExtraDict", Dict[str, Union[str, bool, float, int]])
122 
123 
124 def parse_parameters_from_env() -> ExtraDict:
125     """
126     This method parses the environment variables injected via SCv1.
127 
128     At least an empty dict is always returned.
129     """
130     # TODO: Once support for SCv1 is deprecated, we should remove this method and rely
131     # purely on settings coming via the `model-settings.json` file.
132     parameters = orjson.loads(os.environ.get(PARAMETERS_ENV_NAME, "[]"))
133 
134     parsed_parameters: ExtraDict = ExtraDict({})
135 
136     # Guard: Exit early if there's no parameters
137     if len(parameters) == 0:
138         return parsed_parameters
139 
140     for param in parameters:
141         name = param.get("name")
142         value = param.get("value")
143         type_ = param.get("type")
144         if type_ == "BOOL":
145             parsed_parameters[name] = bool(strtobool(value))
146         else:
147             try:
148                 parsed_parameters[name] = EXTRA_TYPE_DICT[type_](value)
149             except ValueError:
150                 raise InvalidModelParameter(name, value, type_)
151             except KeyError:
152                 raise InvalidModelParameterType(type_)
153 
154     return parsed_parameters
155 
156 
157 def get_huggingface_settings(model_settings: ModelSettings) -> HuggingFaceSettings:
158     """Get the HuggingFace settings provided to the runtime"""
159 
160     env_params = parse_parameters_from_env()
161     extra = merge_huggingface_settings_extra(model_settings, env_params)
162     hf_settings = HuggingFaceSettings(**extra)  # type: ignore
163 
164     if hf_settings.task not in SUPPORTED_TASKS:
165         raise InvalidTransformersTask(hf_settings.task, SUPPORTED_TASKS.keys())
166 
167     if hf_settings.optimum_model:
168         if hf_settings.task not in SUPPORTED_OPTIMUM_TASKS:
169             raise InvalidOptimumTask(hf_settings.task, SUPPORTED_OPTIMUM_TASKS.keys())
170 
171     return hf_settings
172 
173 
174 def merge_huggingface_settings_extra(
175     model_settings: ModelSettings, env_params: ExtraDict
176 ) -> ExtraDict:
177     """
178     This function returns the Extra field of the Settings.
179 
180     It merges them, iff they're both present, from the
181     environment AND model settings file. Precedence is
182     giving to the environment.
183     """
184 
185     # Both `parameters` and `extra` are Optional, so we
186     # need to get the value, or nothing.
187     settings_params = (
188         model_settings.parameters.extra
189         if model_settings.parameters is not None
190         else None
191     )
192 
193     if settings_params is None and env_params == {}:
194         # There must be settings provided by at least the environment OR model settings
195         raise MissingHuggingFaceSettings()
196 
197     # Set the default value
198     settings_params = settings_params or {}
199 
200     # Overwrite any conflicting keys, giving precedence to the environment
201     settings_params.update(env_params)
202 
203     return ExtraDict(settings_params)
204
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/runtimes/huggingface/mlserver_huggingface/common.py b/runtimes/huggingface/mlserver_huggingface/common.py
--- a/runtimes/huggingface/mlserver_huggingface/common.py
+++ b/runtimes/huggingface/mlserver_huggingface/common.py
@@ -23,14 +23,15 @@
 def load_pipeline_from_settings(
     hf_settings: HuggingFaceSettings, settings: ModelSettings
 ) -> Pipeline:
-    # TODO: Support URI for locally downloaded artifacts
-    # uri = model_parameters.uri
     pipeline = _get_pipeline_class(hf_settings)
 
     batch_size = 1
     if settings.max_batch_size:
         batch_size = settings.max_batch_size
 
+    model = hf_settings.pretrained_model
+    if not model:
+        model = settings.parameters.uri  # type: ignore
     tokenizer = hf_settings.pretrained_tokenizer
     if not tokenizer:
         tokenizer = hf_settings.pretrained_model
@@ -51,7 +52,7 @@
 
     hf_pipeline = pipeline(
         hf_settings.task_name,
-        model=hf_settings.pretrained_model,
+        model=model,
         tokenizer=tokenizer,
         device=hf_settings.device,
         batch_size=batch_size,
@@ -61,7 +62,7 @@
     # If max_batch_size > 0 we need to ensure tokens are padded
     if settings.max_batch_size:
         model = hf_pipeline.model
-        eos_token_id = model.config.eos_token_id
+        eos_token_id = model.config.eos_token_id  # type: ignore
         hf_pipeline.tokenizer.pad_token_id = [str(eos_token_id)]  # type: ignore
 
     return hf_pipeline
diff --git a/runtimes/huggingface/mlserver_huggingface/settings.py b/runtimes/huggingface/mlserver_huggingface/settings.py
--- a/runtimes/huggingface/mlserver_huggingface/settings.py
+++ b/runtimes/huggingface/mlserver_huggingface/settings.py
@@ -2,7 +2,7 @@
 import orjson
 
 from typing import Optional, Dict, Union, NewType
-from pydantic import BaseSettings
+from pydantic import BaseSettings, Extra
 from distutils.util import strtobool
 from transformers.pipelines import SUPPORTED_TASKS
 
@@ -37,6 +37,7 @@
 
     class Config:
         env_prefix = ENV_PREFIX_HUGGINGFACE_SETTINGS
+        extra = Extra.ignore
 
     # TODO: Document fields
     task: str = ""

{"golden_diff": "diff --git a/runtimes/huggingface/mlserver_huggingface/common.py b/runtimes/huggingface/mlserver_huggingface/common.py\n--- a/runtimes/huggingface/mlserver_huggingface/common.py\n+++ b/runtimes/huggingface/mlserver_huggingface/common.py\n@@ -23,14 +23,15 @@\n def load_pipeline_from_settings(\n     hf_settings: HuggingFaceSettings, settings: ModelSettings\n ) -> Pipeline:\n-    # TODO: Support URI for locally downloaded artifacts\n-    # uri = model_parameters.uri\n     pipeline = _get_pipeline_class(hf_settings)\n \n     batch_size = 1\n     if settings.max_batch_size:\n         batch_size = settings.max_batch_size\n \n+    model = hf_settings.pretrained_model\n+    if not model:\n+        model = settings.parameters.uri  # type: ignore\n     tokenizer = hf_settings.pretrained_tokenizer\n     if not tokenizer:\n         tokenizer = hf_settings.pretrained_model\n@@ -51,7 +52,7 @@\n \n     hf_pipeline = pipeline(\n         hf_settings.task_name,\n-        model=hf_settings.pretrained_model,\n+        model=model,\n         tokenizer=tokenizer,\n         device=hf_settings.device,\n         batch_size=batch_size,\n@@ -61,7 +62,7 @@\n     # If max_batch_size > 0 we need to ensure tokens are padded\n     if settings.max_batch_size:\n         model = hf_pipeline.model\n-        eos_token_id = model.config.eos_token_id\n+        eos_token_id = model.config.eos_token_id  # type: ignore\n         hf_pipeline.tokenizer.pad_token_id = [str(eos_token_id)]  # type: ignore\n \n     return hf_pipeline\ndiff --git a/runtimes/huggingface/mlserver_huggingface/settings.py b/runtimes/huggingface/mlserver_huggingface/settings.py\n--- a/runtimes/huggingface/mlserver_huggingface/settings.py\n+++ b/runtimes/huggingface/mlserver_huggingface/settings.py\n@@ -2,7 +2,7 @@\n import orjson\n \n from typing import Optional, Dict, Union, NewType\n-from pydantic import BaseSettings\n+from pydantic import BaseSettings, Extra\n from distutils.util import strtobool\n from transformers.pipelines import SUPPORTED_TASKS\n \n@@ -37,6 +37,7 @@\n \n     class Config:\n         env_prefix = ENV_PREFIX_HUGGINGFACE_SETTINGS\n+        extra = Extra.ignore\n \n     # TODO: Document fields\n     task: str = \"\"\n", "issue": "Load local artefacts in HuggingFace runtime\nSupport loading artifacts from provided model-URI\n", "before_files": [{"content": "import json\nimport numpy as np\n\nfrom typing import Callable\nfrom functools import partial\nfrom mlserver.settings import ModelSettings\n\nimport torch\nimport tensorflow as tf\n\nfrom optimum.pipelines import pipeline as opt_pipeline\nfrom transformers.pipelines import pipeline as trf_pipeline\nfrom transformers.pipelines.base import Pipeline\n\nfrom .settings import HuggingFaceSettings\n\n\nOPTIMUM_ACCELERATOR = \"ort\"\n\n_PipelineConstructor = Callable[..., Pipeline]\n\n\ndef load_pipeline_from_settings(\n    hf_settings: HuggingFaceSettings, settings: ModelSettings\n) -> Pipeline:\n    # TODO: Support URI for locally downloaded artifacts\n    # uri = model_parameters.uri\n    pipeline = _get_pipeline_class(hf_settings)\n\n    batch_size = 1\n    if settings.max_batch_size:\n        batch_size = settings.max_batch_size\n\n    tokenizer = hf_settings.pretrained_tokenizer\n    if not tokenizer:\n        tokenizer = hf_settings.pretrained_model\n    if hf_settings.framework == \"tf\":\n        if hf_settings.inter_op_threads is not None:\n            tf.config.threading.set_inter_op_parallelism_threads(\n                hf_settings.inter_op_threads\n            )\n        if hf_settings.intra_op_threads is not None:\n            tf.config.threading.set_intra_op_parallelism_threads(\n                hf_settings.intra_op_threads\n            )\n    elif hf_settings.framework == \"pt\":\n        if hf_settings.inter_op_threads is not None:\n            torch.set_num_interop_threads(hf_settings.inter_op_threads)\n        if hf_settings.intra_op_threads is not None:\n            torch.set_num_threads(hf_settings.intra_op_threads)\n\n    hf_pipeline = pipeline(\n        hf_settings.task_name,\n        model=hf_settings.pretrained_model,\n        tokenizer=tokenizer,\n        device=hf_settings.device,\n        batch_size=batch_size,\n        framework=hf_settings.framework,\n    )\n\n    # If max_batch_size > 0 we need to ensure tokens are padded\n    if settings.max_batch_size:\n        model = hf_pipeline.model\n        eos_token_id = model.config.eos_token_id\n        hf_pipeline.tokenizer.pad_token_id = [str(eos_token_id)]  # type: ignore\n\n    return hf_pipeline\n\n\ndef _get_pipeline_class(hf_settings: HuggingFaceSettings) -> _PipelineConstructor:\n    if hf_settings.optimum_model:\n        return partial(opt_pipeline, accelerator=OPTIMUM_ACCELERATOR)\n\n    return trf_pipeline\n\n\nclass NumpyEncoder(json.JSONEncoder):\n    def default(self, obj):\n        if isinstance(obj, np.ndarray):\n            return obj.tolist()\n        return json.JSONEncoder.default(self, obj)\n", "path": "runtimes/huggingface/mlserver_huggingface/common.py"}, {"content": "import os\nimport orjson\n\nfrom typing import Optional, Dict, Union, NewType\nfrom pydantic import BaseSettings\nfrom distutils.util import strtobool\nfrom transformers.pipelines import SUPPORTED_TASKS\n\ntry:\n    # Optimum 1.7 changed the import name from `SUPPORTED_TASKS` to\n    # `ORT_SUPPORTED_TASKS`.\n    # We'll try to import the more recent one, falling back to the previous\n    # import name if not present.\n    # https://github.com/huggingface/optimum/blob/987b02e4f6e2a1c9325b364ff764da2e57e89902/optimum/pipelines/__init__.py#L18\n    from optimum.pipelines import ORT_SUPPORTED_TASKS as SUPPORTED_OPTIMUM_TASKS\nexcept ImportError:\n    from optimum.pipelines import SUPPORTED_TASKS as SUPPORTED_OPTIMUM_TASKS\n\nfrom mlserver.settings import ModelSettings\n\nfrom .errors import (\n    MissingHuggingFaceSettings,\n    InvalidTransformersTask,\n    InvalidOptimumTask,\n    InvalidModelParameter,\n    InvalidModelParameterType,\n)\n\nENV_PREFIX_HUGGINGFACE_SETTINGS = \"MLSERVER_MODEL_HUGGINGFACE_\"\nPARAMETERS_ENV_NAME = \"PREDICTIVE_UNIT_PARAMETERS\"\n\n\nclass HuggingFaceSettings(BaseSettings):\n    \"\"\"\n    Parameters that apply only to HuggingFace models\n    \"\"\"\n\n    class Config:\n        env_prefix = ENV_PREFIX_HUGGINGFACE_SETTINGS\n\n    # TODO: Document fields\n    task: str = \"\"\n    \"\"\"\n    Pipeline task to load.\n    You can see the available Optimum and Transformers tasks available in the\n    links below:\n\n    - `Optimum Tasks <https://huggingface.co/docs/optimum/onnxruntime/usage_guides/pipelines#inference-pipelines-with-the-onnx-runtime-accelerator>`_\n    - `Transformer Tasks <https://huggingface.co/docs/transformers/task_summary>`_\n    \"\"\"  # noqa: E501\n\n    task_suffix: str = \"\"\n    \"\"\"\n    Suffix to append to the base task name.\n    Useful for, e.g. translation tasks which require a suffix on the task name\n    to specify source and target.\n    \"\"\"\n\n    pretrained_model: Optional[str] = None\n    \"\"\"\n    Name of the model that should be loaded in the pipeline.\n    \"\"\"\n\n    pretrained_tokenizer: Optional[str] = None\n    \"\"\"\n    Name of the tokenizer that should be loaded in the pipeline.\n    \"\"\"\n\n    framework: Optional[str] = None\n    \"\"\"\n    The framework to use, either \"pt\" for PyTorch or \"tf\" for TensorFlow.\n    \"\"\"\n\n    optimum_model: bool = False\n    \"\"\"\n    Flag to decide whether the pipeline should use a Optimum-optimised model or\n    the standard Transformers model.\n    Under the hood, this will enable the model to use the optimised ONNX\n    runtime.\n    \"\"\"\n\n    device: int = -1\n    \"\"\"\n    Device in which this pipeline will be loaded (e.g., \"cpu\", \"cuda:1\", \"mps\",\n    or a GPU ordinal rank like 1).\n    \"\"\"\n\n    inter_op_threads: Optional[int] = None\n    \"\"\"\n    Threads used for parallelism between independent operations.\n    PyTorch:\n    https://pytorch.org/docs/stable/notes/cpu_threading_torchscript_inference.html\n    Tensorflow:\n    https://www.tensorflow.org/api_docs/python/tf/config/threading/set_inter_op_parallelism_threads\n    \"\"\"\n\n    intra_op_threads: Optional[int] = None\n    \"\"\"\n    Threads used within an individual op for parallelism.\n    PyTorch:\n    https://pytorch.org/docs/stable/notes/cpu_threading_torchscript_inference.html\n    Tensorflow:\n    https://www.tensorflow.org/api_docs/python/tf/config/threading/set_intra_op_parallelism_threads\n    \"\"\"\n\n    @property\n    def task_name(self):\n        if self.task == \"translation\":\n            return f\"{self.task}{self.task_suffix}\"\n        return self.task\n\n\nEXTRA_TYPE_DICT = {\n    \"INT\": int,\n    \"FLOAT\": float,\n    \"DOUBLE\": float,\n    \"STRING\": str,\n    \"BOOL\": bool,\n}\n\nExtraDict = NewType(\"ExtraDict\", Dict[str, Union[str, bool, float, int]])\n\n\ndef parse_parameters_from_env() -> ExtraDict:\n    \"\"\"\n    This method parses the environment variables injected via SCv1.\n\n    At least an empty dict is always returned.\n    \"\"\"\n    # TODO: Once support for SCv1 is deprecated, we should remove this method and rely\n    # purely on settings coming via the `model-settings.json` file.\n    parameters = orjson.loads(os.environ.get(PARAMETERS_ENV_NAME, \"[]\"))\n\n    parsed_parameters: ExtraDict = ExtraDict({})\n\n    # Guard: Exit early if there's no parameters\n    if len(parameters) == 0:\n        return parsed_parameters\n\n    for param in parameters:\n        name = param.get(\"name\")\n        value = param.get(\"value\")\n        type_ = param.get(\"type\")\n        if type_ == \"BOOL\":\n            parsed_parameters[name] = bool(strtobool(value))\n        else:\n            try:\n                parsed_parameters[name] = EXTRA_TYPE_DICT[type_](value)\n            except ValueError:\n                raise InvalidModelParameter(name, value, type_)\n            except KeyError:\n                raise InvalidModelParameterType(type_)\n\n    return parsed_parameters\n\n\ndef get_huggingface_settings(model_settings: ModelSettings) -> HuggingFaceSettings:\n    \"\"\"Get the HuggingFace settings provided to the runtime\"\"\"\n\n    env_params = parse_parameters_from_env()\n    extra = merge_huggingface_settings_extra(model_settings, env_params)\n    hf_settings = HuggingFaceSettings(**extra)  # type: ignore\n\n    if hf_settings.task not in SUPPORTED_TASKS:\n        raise InvalidTransformersTask(hf_settings.task, SUPPORTED_TASKS.keys())\n\n    if hf_settings.optimum_model:\n        if hf_settings.task not in SUPPORTED_OPTIMUM_TASKS:\n            raise InvalidOptimumTask(hf_settings.task, SUPPORTED_OPTIMUM_TASKS.keys())\n\n    return hf_settings\n\n\ndef merge_huggingface_settings_extra(\n    model_settings: ModelSettings, env_params: ExtraDict\n) -> ExtraDict:\n    \"\"\"\n    This function returns the Extra field of the Settings.\n\n    It merges them, iff they're both present, from the\n    environment AND model settings file. Precedence is\n    giving to the environment.\n    \"\"\"\n\n    # Both `parameters` and `extra` are Optional, so we\n    # need to get the value, or nothing.\n    settings_params = (\n        model_settings.parameters.extra\n        if model_settings.parameters is not None\n        else None\n    )\n\n    if settings_params is None and env_params == {}:\n        # There must be settings provided by at least the environment OR model settings\n        raise MissingHuggingFaceSettings()\n\n    # Set the default value\n    settings_params = settings_params or {}\n\n    # Overwrite any conflicting keys, giving precedence to the environment\n    settings_params.update(env_params)\n\n    return ExtraDict(settings_params)\n", "path": "runtimes/huggingface/mlserver_huggingface/settings.py"}], "after_files": [{"content": "import json\nimport numpy as np\n\nfrom typing import Callable\nfrom functools import partial\nfrom mlserver.settings import ModelSettings\n\nimport torch\nimport tensorflow as tf\n\nfrom optimum.pipelines import pipeline as opt_pipeline\nfrom transformers.pipelines import pipeline as trf_pipeline\nfrom transformers.pipelines.base import Pipeline\n\nfrom .settings import HuggingFaceSettings\n\n\nOPTIMUM_ACCELERATOR = \"ort\"\n\n_PipelineConstructor = Callable[..., Pipeline]\n\n\ndef load_pipeline_from_settings(\n    hf_settings: HuggingFaceSettings, settings: ModelSettings\n) -> Pipeline:\n    pipeline = _get_pipeline_class(hf_settings)\n\n    batch_size = 1\n    if settings.max_batch_size:\n        batch_size = settings.max_batch_size\n\n    model = hf_settings.pretrained_model\n    if not model:\n        model = settings.parameters.uri  # type: ignore\n    tokenizer = hf_settings.pretrained_tokenizer\n    if not tokenizer:\n        tokenizer = hf_settings.pretrained_model\n    if hf_settings.framework == \"tf\":\n        if hf_settings.inter_op_threads is not None:\n            tf.config.threading.set_inter_op_parallelism_threads(\n                hf_settings.inter_op_threads\n            )\n        if hf_settings.intra_op_threads is not None:\n            tf.config.threading.set_intra_op_parallelism_threads(\n                hf_settings.intra_op_threads\n            )\n    elif hf_settings.framework == \"pt\":\n        if hf_settings.inter_op_threads is not None:\n            torch.set_num_interop_threads(hf_settings.inter_op_threads)\n        if hf_settings.intra_op_threads is not None:\n            torch.set_num_threads(hf_settings.intra_op_threads)\n\n    hf_pipeline = pipeline(\n        hf_settings.task_name,\n        model=model,\n        tokenizer=tokenizer,\n        device=hf_settings.device,\n        batch_size=batch_size,\n        framework=hf_settings.framework,\n    )\n\n    # If max_batch_size > 0 we need to ensure tokens are padded\n    if settings.max_batch_size:\n        model = hf_pipeline.model\n        eos_token_id = model.config.eos_token_id  # type: ignore\n        hf_pipeline.tokenizer.pad_token_id = [str(eos_token_id)]  # type: ignore\n\n    return hf_pipeline\n\n\ndef _get_pipeline_class(hf_settings: HuggingFaceSettings) -> _PipelineConstructor:\n    if hf_settings.optimum_model:\n        return partial(opt_pipeline, accelerator=OPTIMUM_ACCELERATOR)\n\n    return trf_pipeline\n\n\nclass NumpyEncoder(json.JSONEncoder):\n    def default(self, obj):\n        if isinstance(obj, np.ndarray):\n            return obj.tolist()\n        return json.JSONEncoder.default(self, obj)\n", "path": "runtimes/huggingface/mlserver_huggingface/common.py"}, {"content": "import os\nimport orjson\n\nfrom typing import Optional, Dict, Union, NewType\nfrom pydantic import BaseSettings, Extra\nfrom distutils.util import strtobool\nfrom transformers.pipelines import SUPPORTED_TASKS\n\ntry:\n    # Optimum 1.7 changed the import name from `SUPPORTED_TASKS` to\n    # `ORT_SUPPORTED_TASKS`.\n    # We'll try to import the more recent one, falling back to the previous\n    # import name if not present.\n    # https://github.com/huggingface/optimum/blob/987b02e4f6e2a1c9325b364ff764da2e57e89902/optimum/pipelines/__init__.py#L18\n    from optimum.pipelines import ORT_SUPPORTED_TASKS as SUPPORTED_OPTIMUM_TASKS\nexcept ImportError:\n    from optimum.pipelines import SUPPORTED_TASKS as SUPPORTED_OPTIMUM_TASKS\n\nfrom mlserver.settings import ModelSettings\n\nfrom .errors import (\n    MissingHuggingFaceSettings,\n    InvalidTransformersTask,\n    InvalidOptimumTask,\n    InvalidModelParameter,\n    InvalidModelParameterType,\n)\n\nENV_PREFIX_HUGGINGFACE_SETTINGS = \"MLSERVER_MODEL_HUGGINGFACE_\"\nPARAMETERS_ENV_NAME = \"PREDICTIVE_UNIT_PARAMETERS\"\n\n\nclass HuggingFaceSettings(BaseSettings):\n    \"\"\"\n    Parameters that apply only to HuggingFace models\n    \"\"\"\n\n    class Config:\n        env_prefix = ENV_PREFIX_HUGGINGFACE_SETTINGS\n        extra = Extra.ignore\n\n    # TODO: Document fields\n    task: str = \"\"\n    \"\"\"\n    Pipeline task to load.\n    You can see the available Optimum and Transformers tasks available in the\n    links below:\n\n    - `Optimum Tasks <https://huggingface.co/docs/optimum/onnxruntime/usage_guides/pipelines#inference-pipelines-with-the-onnx-runtime-accelerator>`_\n    - `Transformer Tasks <https://huggingface.co/docs/transformers/task_summary>`_\n    \"\"\"  # noqa: E501\n\n    task_suffix: str = \"\"\n    \"\"\"\n    Suffix to append to the base task name.\n    Useful for, e.g. translation tasks which require a suffix on the task name\n    to specify source and target.\n    \"\"\"\n\n    pretrained_model: Optional[str] = None\n    \"\"\"\n    Name of the model that should be loaded in the pipeline.\n    \"\"\"\n\n    pretrained_tokenizer: Optional[str] = None\n    \"\"\"\n    Name of the tokenizer that should be loaded in the pipeline.\n    \"\"\"\n\n    framework: Optional[str] = None\n    \"\"\"\n    The framework to use, either \"pt\" for PyTorch or \"tf\" for TensorFlow.\n    \"\"\"\n\n    optimum_model: bool = False\n    \"\"\"\n    Flag to decide whether the pipeline should use a Optimum-optimised model or\n    the standard Transformers model.\n    Under the hood, this will enable the model to use the optimised ONNX\n    runtime.\n    \"\"\"\n\n    device: int = -1\n    \"\"\"\n    Device in which this pipeline will be loaded (e.g., \"cpu\", \"cuda:1\", \"mps\",\n    or a GPU ordinal rank like 1).\n    \"\"\"\n\n    inter_op_threads: Optional[int] = None\n    \"\"\"\n    Threads used for parallelism between independent operations.\n    PyTorch:\n    https://pytorch.org/docs/stable/notes/cpu_threading_torchscript_inference.html\n    Tensorflow:\n    https://www.tensorflow.org/api_docs/python/tf/config/threading/set_inter_op_parallelism_threads\n    \"\"\"\n\n    intra_op_threads: Optional[int] = None\n    \"\"\"\n    Threads used within an individual op for parallelism.\n    PyTorch:\n    https://pytorch.org/docs/stable/notes/cpu_threading_torchscript_inference.html\n    Tensorflow:\n    https://www.tensorflow.org/api_docs/python/tf/config/threading/set_intra_op_parallelism_threads\n    \"\"\"\n\n    @property\n    def task_name(self):\n        if self.task == \"translation\":\n            return f\"{self.task}{self.task_suffix}\"\n        return self.task\n\n\nEXTRA_TYPE_DICT = {\n    \"INT\": int,\n    \"FLOAT\": float,\n    \"DOUBLE\": float,\n    \"STRING\": str,\n    \"BOOL\": bool,\n}\n\nExtraDict = NewType(\"ExtraDict\", Dict[str, Union[str, bool, float, int]])\n\n\ndef parse_parameters_from_env() -> ExtraDict:\n    \"\"\"\n    This method parses the environment variables injected via SCv1.\n\n    At least an empty dict is always returned.\n    \"\"\"\n    # TODO: Once support for SCv1 is deprecated, we should remove this method and rely\n    # purely on settings coming via the `model-settings.json` file.\n    parameters = orjson.loads(os.environ.get(PARAMETERS_ENV_NAME, \"[]\"))\n\n    parsed_parameters: ExtraDict = ExtraDict({})\n\n    # Guard: Exit early if there's no parameters\n    if len(parameters) == 0:\n        return parsed_parameters\n\n    for param in parameters:\n        name = param.get(\"name\")\n        value = param.get(\"value\")\n        type_ = param.get(\"type\")\n        if type_ == \"BOOL\":\n            parsed_parameters[name] = bool(strtobool(value))\n        else:\n            try:\n                parsed_parameters[name] = EXTRA_TYPE_DICT[type_](value)\n            except ValueError:\n                raise InvalidModelParameter(name, value, type_)\n            except KeyError:\n                raise InvalidModelParameterType(type_)\n\n    return parsed_parameters\n\n\ndef get_huggingface_settings(model_settings: ModelSettings) -> HuggingFaceSettings:\n    \"\"\"Get the HuggingFace settings provided to the runtime\"\"\"\n\n    env_params = parse_parameters_from_env()\n    extra = merge_huggingface_settings_extra(model_settings, env_params)\n    hf_settings = HuggingFaceSettings(**extra)  # type: ignore\n\n    if hf_settings.task not in SUPPORTED_TASKS:\n        raise InvalidTransformersTask(hf_settings.task, SUPPORTED_TASKS.keys())\n\n    if hf_settings.optimum_model:\n        if hf_settings.task not in SUPPORTED_OPTIMUM_TASKS:\n            raise InvalidOptimumTask(hf_settings.task, SUPPORTED_OPTIMUM_TASKS.keys())\n\n    return hf_settings\n\n\ndef merge_huggingface_settings_extra(\n    model_settings: ModelSettings, env_params: ExtraDict\n) -> ExtraDict:\n    \"\"\"\n    This function returns the Extra field of the Settings.\n\n    It merges them, iff they're both present, from the\n    environment AND model settings file. Precedence is\n    giving to the environment.\n    \"\"\"\n\n    # Both `parameters` and `extra` are Optional, so we\n    # need to get the value, or nothing.\n    settings_params = (\n        model_settings.parameters.extra\n        if model_settings.parameters is not None\n        else None\n    )\n\n    if settings_params is None and env_params == {}:\n        # There must be settings provided by at least the environment OR model settings\n        raise MissingHuggingFaceSettings()\n\n    # Set the default value\n    settings_params = settings_params or {}\n\n    # Overwrite any conflicting keys, giving precedence to the environment\n    settings_params.update(env_params)\n\n    return ExtraDict(settings_params)\n", "path": "runtimes/huggingface/mlserver_huggingface/settings.py"}]}