We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Serialisation of the PromptModel is broken if the invocation_layer_class param is set, both from and to yaml
**Describe the bug**
A clear and concise description of what the bug is.

I am using a pipeline that uses a `PromptModel` with an `invocation_layer_class` specified, to deploy  Haystack API via docker.
On executing docker-compose up this will result in the error described below.

yml file:

name: PromptModel
params:
  api_key: EMPTY
  invocation_layer_class: !!python/name:haystack.nodes.prompt.invocation_layer.open_ai.OpenAIInvocationLayer ''
  max_length: 1024
  model_kwargs:
    api_base: http://localhost:8100/v1
    maximum_context_length: 2048
type: PromptModel

**Error message**
Error that was thrown (if available)

`yaml.constructor.ConstructorError: could not determine a constructor for the tag 'tag:yaml.org,2002:python/name:haystack.nodes.prompt.invocation_layer.open_ai.OpenAIInvocationLayer'`

**Expected behavior**
A clear and concise description of what you expected to happen.

I expected the pipleline to load correctly.

**Additional context**
Add any other context about the problem here, like document types / preprocessing steps / settings of reader etc.

The issue has been replicated by @anakin87

**To Reproduce**
Steps to reproduce the behavior
1. Serialize a pipeline with a `PromptModel `  with `invocation_layer_class` param, to yaml
2. Load the pipeline in the haystack api docker container


**FAQ Check**
- [x] Have you had a look at [our new FAQ page](https://docs.haystack.deepset.ai/docs/faq)?

**System:**
 - OS: Windows11 + WSL2
 - GPU/CPU: GPU
 - Haystack version (commit or version number): 1.22.1
 - DocumentStore: WeaviateDocumentStore
 - Reader: -
 - Retriever: EmbeddingRetriever


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `haystack/nodes/prompt/prompt_model.py`
Content:
```
1 import inspect
2 import logging
3 from typing import Any, Dict, List, Optional, Tuple, Type, Union, overload
4 
5 from haystack.nodes.base import BaseComponent
6 from haystack.nodes.prompt.invocation_layer import PromptModelInvocationLayer
7 from haystack.schema import Document, MultiLabel
8 from haystack.lazy_imports import LazyImport
9 
10 with LazyImport(message="Run 'pip install farm-haystack[inference]'") as torch_import:
11     import torch
12 
13 
14 logger = logging.getLogger(__name__)
15 
16 
17 class PromptModel(BaseComponent):
18     """
19     The PromptModel class is a component that uses a pre-trained model to perform tasks defined in a prompt. Out of
20     the box, it supports model invocation layers for:
21     - Hugging Face transformers (all text2text-generation and text-generation models)
22     - OpenAI InstructGPT models
23     - Azure OpenAI InstructGPT models
24 
25     Although it's possible to use PromptModel to make prompt invocations on the underlying model, use
26     PromptNode to interact with the model. PromptModel instances are a way for multiple
27     PromptNode instances to use a single PromptNode, and thus save computational resources.
28 
29     For more details, refer to [PromptModels](https://docs.haystack.deepset.ai/docs/prompt_node#models).
30     """
31 
32     outgoing_edges = 1
33 
34     def __init__(
35         self,
36         model_name_or_path: str = "google/flan-t5-base",
37         max_length: Optional[int] = 100,
38         api_key: Optional[str] = None,
39         timeout: Optional[float] = None,
40         use_auth_token: Optional[Union[str, bool]] = None,
41         use_gpu: Optional[bool] = None,
42         devices: Optional[List[Union[str, "torch.device"]]] = None,
43         invocation_layer_class: Optional[Type[PromptModelInvocationLayer]] = None,
44         model_kwargs: Optional[Dict] = None,
45     ):
46         """
47         Creates an instance of PromptModel.
48 
49         :param model_name_or_path: The name or path of the underlying model.
50         :param max_length: The maximum number of tokens the output text generated by the model can have.
51         :param api_key: The API key to use for the model.
52         :param use_auth_token: The Hugging Face token to use.
53         :param use_gpu: Whether to use GPU or not.
54         :param devices: The devices to use where the model is loaded.
55         :param invocation_layer_class: The custom invocation layer class to use. If None, known invocation layers are used.
56         :param model_kwargs: Additional keyword arguments passed to the underlying model.
57 
58         Note that Azure OpenAI InstructGPT models require two additional parameters: azure_base_url (The URL for the
59         Azure OpenAI API endpoint, usually in the form `https://<your-endpoint>.openai.azure.com') and
60         azure_deployment_name (the name of the Azure OpenAI API deployment). You should add these parameters
61         in the `model_kwargs` dictionary.
62         """
63         super().__init__()
64         self.model_name_or_path = model_name_or_path
65         self.max_length = max_length
66         self.api_key = api_key
67         self.timeout = timeout
68         self.use_auth_token = use_auth_token
69         self.use_gpu = use_gpu
70         self.devices = devices
71 
72         self.model_kwargs = model_kwargs if model_kwargs else {}
73         self.model_invocation_layer = self.create_invocation_layer(invocation_layer_class=invocation_layer_class)
74 
75     def create_invocation_layer(
76         self, invocation_layer_class: Optional[Type[PromptModelInvocationLayer]]
77     ) -> PromptModelInvocationLayer:
78         kwargs = {
79             "api_key": self.api_key,
80             "timeout": self.timeout,
81             "use_auth_token": self.use_auth_token,
82             "use_gpu": self.use_gpu,
83             "devices": self.devices,
84         }
85         all_kwargs = {**self.model_kwargs, **kwargs}
86 
87         if invocation_layer_class:
88             return invocation_layer_class(
89                 model_name_or_path=self.model_name_or_path, max_length=self.max_length, **all_kwargs
90             )
91 
92         for invocation_layer in PromptModelInvocationLayer.invocation_layer_providers:
93             if inspect.isabstract(invocation_layer):
94                 continue
95             if invocation_layer.supports(self.model_name_or_path, **all_kwargs):
96                 return invocation_layer(
97                     model_name_or_path=self.model_name_or_path, max_length=self.max_length, **all_kwargs
98                 )
99         raise ValueError(
100             f"Model {self.model_name_or_path} is not supported - no matching invocation layer found."
101             f" Currently supported invocation layers are: {PromptModelInvocationLayer.invocation_layer_providers}"
102             f" You can implement and provide custom invocation layer for {self.model_name_or_path} by subclassing "
103             "PromptModelInvocationLayer."
104             f" Also please ensure you are authorised to load the model {self.model_name_or_path} and you are "
105             "logged-in into the huggingface cli."
106         )
107 
108     def invoke(self, prompt: Union[str, List[str], List[Dict[str, str]]], **kwargs) -> List[str]:
109         """
110         Takes in a prompt and returns a list of responses using the underlying invocation layer.
111 
112         :param prompt: The prompt to use for the invocation. It can be a single prompt or a list of prompts.
113         :param kwargs: Additional keyword arguments to pass to the invocation layer.
114         :return: A list of model-generated responses for the prompt or prompts.
115         """
116         output = self.model_invocation_layer.invoke(prompt=prompt, **kwargs)
117         return output
118 
119     async def ainvoke(self, prompt: Union[str, List[str], List[Dict[str, str]]], **kwargs) -> List[str]:
120         """
121         Drop-in replacement asyncio version of the `invoke` method, see there for documentation.
122         """
123         if hasattr(self.model_invocation_layer, "ainvoke"):
124             return await self.model_invocation_layer.ainvoke(prompt=prompt, **kwargs)
125 
126         # The underlying invocation layer doesn't support asyncio
127         return self.model_invocation_layer.invoke(prompt=prompt, **kwargs)
128 
129     @overload
130     def _ensure_token_limit(self, prompt: str) -> str:
131         ...
132 
133     @overload
134     def _ensure_token_limit(self, prompt: List[Dict[str, str]]) -> List[Dict[str, str]]:
135         ...
136 
137     def _ensure_token_limit(self, prompt: Union[str, List[Dict[str, str]]]) -> Union[str, List[Dict[str, str]]]:
138         """Ensure that length of the prompt and answer is within the maximum token length of the PromptModel.
139 
140         :param prompt: Prompt text to be sent to the generative model.
141         """
142         return self.model_invocation_layer._ensure_token_limit(prompt=prompt)
143 
144     def run(
145         self,
146         query: Optional[str] = None,
147         file_paths: Optional[List[str]] = None,
148         labels: Optional[MultiLabel] = None,
149         documents: Optional[List[Document]] = None,
150         meta: Optional[dict] = None,
151     ) -> Tuple[Dict, str]:
152         raise NotImplementedError("This method should never be implemented in the derived class")
153 
154     def run_batch(
155         self,
156         queries: Optional[Union[str, List[str]]] = None,
157         file_paths: Optional[List[str]] = None,
158         labels: Optional[Union[MultiLabel, List[MultiLabel]]] = None,
159         documents: Optional[Union[List[Document], List[List[Document]]]] = None,
160         meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,
161         params: Optional[dict] = None,
162         debug: Optional[bool] = None,
163     ):
164         raise NotImplementedError("This method should never be implemented in the derived class")
165 
166     def __repr__(self):
167         return "{}({!r})".format(self.__class__.__name__, self.__dict__)
168
```
--- 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/haystack/nodes/prompt/prompt_model.py b/haystack/nodes/prompt/prompt_model.py
--- a/haystack/nodes/prompt/prompt_model.py
+++ b/haystack/nodes/prompt/prompt_model.py
@@ -1,5 +1,6 @@
 import inspect
 import logging
+import importlib
 from typing import Any, Dict, List, Optional, Tuple, Type, Union, overload
 
 from haystack.nodes.base import BaseComponent
@@ -40,7 +41,7 @@
         use_auth_token: Optional[Union[str, bool]] = None,
         use_gpu: Optional[bool] = None,
         devices: Optional[List[Union[str, "torch.device"]]] = None,
-        invocation_layer_class: Optional[Type[PromptModelInvocationLayer]] = None,
+        invocation_layer_class: Optional[Union[Type[PromptModelInvocationLayer], str]] = None,
         model_kwargs: Optional[Dict] = None,
     ):
         """
@@ -73,7 +74,7 @@
         self.model_invocation_layer = self.create_invocation_layer(invocation_layer_class=invocation_layer_class)
 
     def create_invocation_layer(
-        self, invocation_layer_class: Optional[Type[PromptModelInvocationLayer]]
+        self, invocation_layer_class: Optional[Union[Type[PromptModelInvocationLayer], str]]
     ) -> PromptModelInvocationLayer:
         kwargs = {
             "api_key": self.api_key,
@@ -84,6 +85,18 @@
         }
         all_kwargs = {**self.model_kwargs, **kwargs}
 
+        if isinstance(invocation_layer_class, str):
+            module_name, class_name = invocation_layer_class.rsplit(".", maxsplit=1)
+            try:
+                module = importlib.import_module(module_name)
+            except ImportError as e:
+                msg = f"Can't find module {module_name}"
+                raise ValueError(msg) from e
+            invocation_layer_class = getattr(module, class_name)
+            if invocation_layer_class is None:
+                msg = f"Can'f find class {class_name} in module {module_name}"
+                ValueError(msg)
+
         if invocation_layer_class:
             return invocation_layer_class(
                 model_name_or_path=self.model_name_or_path, max_length=self.max_length, **all_kwargs

{"golden_diff": "diff --git a/haystack/nodes/prompt/prompt_model.py b/haystack/nodes/prompt/prompt_model.py\n--- a/haystack/nodes/prompt/prompt_model.py\n+++ b/haystack/nodes/prompt/prompt_model.py\n@@ -1,5 +1,6 @@\n import inspect\n import logging\n+import importlib\n from typing import Any, Dict, List, Optional, Tuple, Type, Union, overload\n \n from haystack.nodes.base import BaseComponent\n@@ -40,7 +41,7 @@\n         use_auth_token: Optional[Union[str, bool]] = None,\n         use_gpu: Optional[bool] = None,\n         devices: Optional[List[Union[str, \"torch.device\"]]] = None,\n-        invocation_layer_class: Optional[Type[PromptModelInvocationLayer]] = None,\n+        invocation_layer_class: Optional[Union[Type[PromptModelInvocationLayer], str]] = None,\n         model_kwargs: Optional[Dict] = None,\n     ):\n         \"\"\"\n@@ -73,7 +74,7 @@\n         self.model_invocation_layer = self.create_invocation_layer(invocation_layer_class=invocation_layer_class)\n \n     def create_invocation_layer(\n-        self, invocation_layer_class: Optional[Type[PromptModelInvocationLayer]]\n+        self, invocation_layer_class: Optional[Union[Type[PromptModelInvocationLayer], str]]\n     ) -> PromptModelInvocationLayer:\n         kwargs = {\n             \"api_key\": self.api_key,\n@@ -84,6 +85,18 @@\n         }\n         all_kwargs = {**self.model_kwargs, **kwargs}\n \n+        if isinstance(invocation_layer_class, str):\n+            module_name, class_name = invocation_layer_class.rsplit(\".\", maxsplit=1)\n+            try:\n+                module = importlib.import_module(module_name)\n+            except ImportError as e:\n+                msg = f\"Can't find module {module_name}\"\n+                raise ValueError(msg) from e\n+            invocation_layer_class = getattr(module, class_name)\n+            if invocation_layer_class is None:\n+                msg = f\"Can'f find class {class_name} in module {module_name}\"\n+                ValueError(msg)\n+\n         if invocation_layer_class:\n             return invocation_layer_class(\n                 model_name_or_path=self.model_name_or_path, max_length=self.max_length, **all_kwargs\n", "issue": "Serialisation of the PromptModel is broken if the invocation_layer_class param is set, both from and to yaml\n**Describe the bug**\r\nA clear and concise description of what the bug is.\r\n\r\nI am using a pipeline that uses a `PromptModel` with an `invocation_layer_class` specified, to deploy  Haystack API via docker.\r\nOn executing docker-compose up this will result in the error described below.\r\n\r\nyml file:\r\n\r\nname: PromptModel\r\nparams:\r\n  api_key: EMPTY\r\n  invocation_layer_class: !!python/name:haystack.nodes.prompt.invocation_layer.open_ai.OpenAIInvocationLayer ''\r\n  max_length: 1024\r\n  model_kwargs:\r\n    api_base: http://localhost:8100/v1\r\n    maximum_context_length: 2048\r\ntype: PromptModel\r\n\r\n**Error message**\r\nError that was thrown (if available)\r\n\r\n`yaml.constructor.ConstructorError: could not determine a constructor for the tag 'tag:yaml.org,2002:python/name:haystack.nodes.prompt.invocation_layer.open_ai.OpenAIInvocationLayer'`\r\n\r\n**Expected behavior**\r\nA clear and concise description of what you expected to happen.\r\n\r\nI expected the pipleline to load correctly.\r\n\r\n**Additional context**\r\nAdd any other context about the problem here, like document types / preprocessing steps / settings of reader etc.\r\n\r\nThe issue has been replicated by @anakin87\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior\r\n1. Serialize a pipeline with a `PromptModel `  with `invocation_layer_class` param, to yaml\r\n2. Load the pipeline in the haystack api docker container\r\n\r\n\r\n**FAQ Check**\r\n- [x] Have you had a look at [our new FAQ page](https://docs.haystack.deepset.ai/docs/faq)?\r\n\r\n**System:**\r\n - OS: Windows11 + WSL2\r\n - GPU/CPU: GPU\r\n - Haystack version (commit or version number): 1.22.1\r\n - DocumentStore: WeaviateDocumentStore\r\n - Reader: -\r\n - Retriever: EmbeddingRetriever\r\n\n", "before_files": [{"content": "import inspect\nimport logging\nfrom typing import Any, Dict, List, Optional, Tuple, Type, Union, overload\n\nfrom haystack.nodes.base import BaseComponent\nfrom haystack.nodes.prompt.invocation_layer import PromptModelInvocationLayer\nfrom haystack.schema import Document, MultiLabel\nfrom haystack.lazy_imports import LazyImport\n\nwith LazyImport(message=\"Run 'pip install farm-haystack[inference]'\") as torch_import:\n    import torch\n\n\nlogger = logging.getLogger(__name__)\n\n\nclass PromptModel(BaseComponent):\n    \"\"\"\n    The PromptModel class is a component that uses a pre-trained model to perform tasks defined in a prompt. Out of\n    the box, it supports model invocation layers for:\n    - Hugging Face transformers (all text2text-generation and text-generation models)\n    - OpenAI InstructGPT models\n    - Azure OpenAI InstructGPT models\n\n    Although it's possible to use PromptModel to make prompt invocations on the underlying model, use\n    PromptNode to interact with the model. PromptModel instances are a way for multiple\n    PromptNode instances to use a single PromptNode, and thus save computational resources.\n\n    For more details, refer to [PromptModels](https://docs.haystack.deepset.ai/docs/prompt_node#models).\n    \"\"\"\n\n    outgoing_edges = 1\n\n    def __init__(\n        self,\n        model_name_or_path: str = \"google/flan-t5-base\",\n        max_length: Optional[int] = 100,\n        api_key: Optional[str] = None,\n        timeout: Optional[float] = None,\n        use_auth_token: Optional[Union[str, bool]] = None,\n        use_gpu: Optional[bool] = None,\n        devices: Optional[List[Union[str, \"torch.device\"]]] = None,\n        invocation_layer_class: Optional[Type[PromptModelInvocationLayer]] = None,\n        model_kwargs: Optional[Dict] = None,\n    ):\n        \"\"\"\n        Creates an instance of PromptModel.\n\n        :param model_name_or_path: The name or path of the underlying model.\n        :param max_length: The maximum number of tokens the output text generated by the model can have.\n        :param api_key: The API key to use for the model.\n        :param use_auth_token: The Hugging Face token to use.\n        :param use_gpu: Whether to use GPU or not.\n        :param devices: The devices to use where the model is loaded.\n        :param invocation_layer_class: The custom invocation layer class to use. If None, known invocation layers are used.\n        :param model_kwargs: Additional keyword arguments passed to the underlying model.\n\n        Note that Azure OpenAI InstructGPT models require two additional parameters: azure_base_url (The URL for the\n        Azure OpenAI API endpoint, usually in the form `https://<your-endpoint>.openai.azure.com') and\n        azure_deployment_name (the name of the Azure OpenAI API deployment). You should add these parameters\n        in the `model_kwargs` dictionary.\n        \"\"\"\n        super().__init__()\n        self.model_name_or_path = model_name_or_path\n        self.max_length = max_length\n        self.api_key = api_key\n        self.timeout = timeout\n        self.use_auth_token = use_auth_token\n        self.use_gpu = use_gpu\n        self.devices = devices\n\n        self.model_kwargs = model_kwargs if model_kwargs else {}\n        self.model_invocation_layer = self.create_invocation_layer(invocation_layer_class=invocation_layer_class)\n\n    def create_invocation_layer(\n        self, invocation_layer_class: Optional[Type[PromptModelInvocationLayer]]\n    ) -> PromptModelInvocationLayer:\n        kwargs = {\n            \"api_key\": self.api_key,\n            \"timeout\": self.timeout,\n            \"use_auth_token\": self.use_auth_token,\n            \"use_gpu\": self.use_gpu,\n            \"devices\": self.devices,\n        }\n        all_kwargs = {**self.model_kwargs, **kwargs}\n\n        if invocation_layer_class:\n            return invocation_layer_class(\n                model_name_or_path=self.model_name_or_path, max_length=self.max_length, **all_kwargs\n            )\n\n        for invocation_layer in PromptModelInvocationLayer.invocation_layer_providers:\n            if inspect.isabstract(invocation_layer):\n                continue\n            if invocation_layer.supports(self.model_name_or_path, **all_kwargs):\n                return invocation_layer(\n                    model_name_or_path=self.model_name_or_path, max_length=self.max_length, **all_kwargs\n                )\n        raise ValueError(\n            f\"Model {self.model_name_or_path} is not supported - no matching invocation layer found.\"\n            f\" Currently supported invocation layers are: {PromptModelInvocationLayer.invocation_layer_providers}\"\n            f\" You can implement and provide custom invocation layer for {self.model_name_or_path} by subclassing \"\n            \"PromptModelInvocationLayer.\"\n            f\" Also please ensure you are authorised to load the model {self.model_name_or_path} and you are \"\n            \"logged-in into the huggingface cli.\"\n        )\n\n    def invoke(self, prompt: Union[str, List[str], List[Dict[str, str]]], **kwargs) -> List[str]:\n        \"\"\"\n        Takes in a prompt and returns a list of responses using the underlying invocation layer.\n\n        :param prompt: The prompt to use for the invocation. It can be a single prompt or a list of prompts.\n        :param kwargs: Additional keyword arguments to pass to the invocation layer.\n        :return: A list of model-generated responses for the prompt or prompts.\n        \"\"\"\n        output = self.model_invocation_layer.invoke(prompt=prompt, **kwargs)\n        return output\n\n    async def ainvoke(self, prompt: Union[str, List[str], List[Dict[str, str]]], **kwargs) -> List[str]:\n        \"\"\"\n        Drop-in replacement asyncio version of the `invoke` method, see there for documentation.\n        \"\"\"\n        if hasattr(self.model_invocation_layer, \"ainvoke\"):\n            return await self.model_invocation_layer.ainvoke(prompt=prompt, **kwargs)\n\n        # The underlying invocation layer doesn't support asyncio\n        return self.model_invocation_layer.invoke(prompt=prompt, **kwargs)\n\n    @overload\n    def _ensure_token_limit(self, prompt: str) -> str:\n        ...\n\n    @overload\n    def _ensure_token_limit(self, prompt: List[Dict[str, str]]) -> List[Dict[str, str]]:\n        ...\n\n    def _ensure_token_limit(self, prompt: Union[str, List[Dict[str, str]]]) -> Union[str, List[Dict[str, str]]]:\n        \"\"\"Ensure that length of the prompt and answer is within the maximum token length of the PromptModel.\n\n        :param prompt: Prompt text to be sent to the generative model.\n        \"\"\"\n        return self.model_invocation_layer._ensure_token_limit(prompt=prompt)\n\n    def run(\n        self,\n        query: Optional[str] = None,\n        file_paths: Optional[List[str]] = None,\n        labels: Optional[MultiLabel] = None,\n        documents: Optional[List[Document]] = None,\n        meta: Optional[dict] = None,\n    ) -> Tuple[Dict, str]:\n        raise NotImplementedError(\"This method should never be implemented in the derived class\")\n\n    def run_batch(\n        self,\n        queries: Optional[Union[str, List[str]]] = None,\n        file_paths: Optional[List[str]] = None,\n        labels: Optional[Union[MultiLabel, List[MultiLabel]]] = None,\n        documents: Optional[Union[List[Document], List[List[Document]]]] = None,\n        meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,\n        params: Optional[dict] = None,\n        debug: Optional[bool] = None,\n    ):\n        raise NotImplementedError(\"This method should never be implemented in the derived class\")\n\n    def __repr__(self):\n        return \"{}({!r})\".format(self.__class__.__name__, self.__dict__)\n", "path": "haystack/nodes/prompt/prompt_model.py"}], "after_files": [{"content": "import inspect\nimport logging\nimport importlib\nfrom typing import Any, Dict, List, Optional, Tuple, Type, Union, overload\n\nfrom haystack.nodes.base import BaseComponent\nfrom haystack.nodes.prompt.invocation_layer import PromptModelInvocationLayer\nfrom haystack.schema import Document, MultiLabel\nfrom haystack.lazy_imports import LazyImport\n\nwith LazyImport(message=\"Run 'pip install farm-haystack[inference]'\") as torch_import:\n    import torch\n\n\nlogger = logging.getLogger(__name__)\n\n\nclass PromptModel(BaseComponent):\n    \"\"\"\n    The PromptModel class is a component that uses a pre-trained model to perform tasks defined in a prompt. Out of\n    the box, it supports model invocation layers for:\n    - Hugging Face transformers (all text2text-generation and text-generation models)\n    - OpenAI InstructGPT models\n    - Azure OpenAI InstructGPT models\n\n    Although it's possible to use PromptModel to make prompt invocations on the underlying model, use\n    PromptNode to interact with the model. PromptModel instances are a way for multiple\n    PromptNode instances to use a single PromptNode, and thus save computational resources.\n\n    For more details, refer to [PromptModels](https://docs.haystack.deepset.ai/docs/prompt_node#models).\n    \"\"\"\n\n    outgoing_edges = 1\n\n    def __init__(\n        self,\n        model_name_or_path: str = \"google/flan-t5-base\",\n        max_length: Optional[int] = 100,\n        api_key: Optional[str] = None,\n        timeout: Optional[float] = None,\n        use_auth_token: Optional[Union[str, bool]] = None,\n        use_gpu: Optional[bool] = None,\n        devices: Optional[List[Union[str, \"torch.device\"]]] = None,\n        invocation_layer_class: Optional[Union[Type[PromptModelInvocationLayer], str]] = None,\n        model_kwargs: Optional[Dict] = None,\n    ):\n        \"\"\"\n        Creates an instance of PromptModel.\n\n        :param model_name_or_path: The name or path of the underlying model.\n        :param max_length: The maximum number of tokens the output text generated by the model can have.\n        :param api_key: The API key to use for the model.\n        :param use_auth_token: The Hugging Face token to use.\n        :param use_gpu: Whether to use GPU or not.\n        :param devices: The devices to use where the model is loaded.\n        :param invocation_layer_class: The custom invocation layer class to use. If None, known invocation layers are used.\n        :param model_kwargs: Additional keyword arguments passed to the underlying model.\n\n        Note that Azure OpenAI InstructGPT models require two additional parameters: azure_base_url (The URL for the\n        Azure OpenAI API endpoint, usually in the form `https://<your-endpoint>.openai.azure.com') and\n        azure_deployment_name (the name of the Azure OpenAI API deployment). You should add these parameters\n        in the `model_kwargs` dictionary.\n        \"\"\"\n        super().__init__()\n        self.model_name_or_path = model_name_or_path\n        self.max_length = max_length\n        self.api_key = api_key\n        self.timeout = timeout\n        self.use_auth_token = use_auth_token\n        self.use_gpu = use_gpu\n        self.devices = devices\n\n        self.model_kwargs = model_kwargs if model_kwargs else {}\n        self.model_invocation_layer = self.create_invocation_layer(invocation_layer_class=invocation_layer_class)\n\n    def create_invocation_layer(\n        self, invocation_layer_class: Optional[Union[Type[PromptModelInvocationLayer], str]]\n    ) -> PromptModelInvocationLayer:\n        kwargs = {\n            \"api_key\": self.api_key,\n            \"timeout\": self.timeout,\n            \"use_auth_token\": self.use_auth_token,\n            \"use_gpu\": self.use_gpu,\n            \"devices\": self.devices,\n        }\n        all_kwargs = {**self.model_kwargs, **kwargs}\n\n        if isinstance(invocation_layer_class, str):\n            module_name, class_name = invocation_layer_class.rsplit(\".\", maxsplit=1)\n            try:\n                module = importlib.import_module(module_name)\n            except ImportError as e:\n                msg = f\"Can't find module {module_name}\"\n                raise ValueError(msg) from e\n            invocation_layer_class = getattr(module, class_name)\n            if invocation_layer_class is None:\n                msg = f\"Can'f find class {class_name} in module {module_name}\"\n                ValueError(msg)\n\n        if invocation_layer_class:\n            return invocation_layer_class(\n                model_name_or_path=self.model_name_or_path, max_length=self.max_length, **all_kwargs\n            )\n\n        for invocation_layer in PromptModelInvocationLayer.invocation_layer_providers:\n            if inspect.isabstract(invocation_layer):\n                continue\n            if invocation_layer.supports(self.model_name_or_path, **all_kwargs):\n                return invocation_layer(\n                    model_name_or_path=self.model_name_or_path, max_length=self.max_length, **all_kwargs\n                )\n        raise ValueError(\n            f\"Model {self.model_name_or_path} is not supported - no matching invocation layer found.\"\n            f\" Currently supported invocation layers are: {PromptModelInvocationLayer.invocation_layer_providers}\"\n            f\" You can implement and provide custom invocation layer for {self.model_name_or_path} by subclassing \"\n            \"PromptModelInvocationLayer.\"\n            f\" Also please ensure you are authorised to load the model {self.model_name_or_path} and you are \"\n            \"logged-in into the huggingface cli.\"\n        )\n\n    def invoke(self, prompt: Union[str, List[str], List[Dict[str, str]]], **kwargs) -> List[str]:\n        \"\"\"\n        Takes in a prompt and returns a list of responses using the underlying invocation layer.\n\n        :param prompt: The prompt to use for the invocation. It can be a single prompt or a list of prompts.\n        :param kwargs: Additional keyword arguments to pass to the invocation layer.\n        :return: A list of model-generated responses for the prompt or prompts.\n        \"\"\"\n        output = self.model_invocation_layer.invoke(prompt=prompt, **kwargs)\n        return output\n\n    async def ainvoke(self, prompt: Union[str, List[str], List[Dict[str, str]]], **kwargs) -> List[str]:\n        \"\"\"\n        Drop-in replacement asyncio version of the `invoke` method, see there for documentation.\n        \"\"\"\n        if hasattr(self.model_invocation_layer, \"ainvoke\"):\n            return await self.model_invocation_layer.ainvoke(prompt=prompt, **kwargs)\n\n        # The underlying invocation layer doesn't support asyncio\n        return self.model_invocation_layer.invoke(prompt=prompt, **kwargs)\n\n    @overload\n    def _ensure_token_limit(self, prompt: str) -> str:\n        ...\n\n    @overload\n    def _ensure_token_limit(self, prompt: List[Dict[str, str]]) -> List[Dict[str, str]]:\n        ...\n\n    def _ensure_token_limit(self, prompt: Union[str, List[Dict[str, str]]]) -> Union[str, List[Dict[str, str]]]:\n        \"\"\"Ensure that length of the prompt and answer is within the maximum token length of the PromptModel.\n\n        :param prompt: Prompt text to be sent to the generative model.\n        \"\"\"\n        return self.model_invocation_layer._ensure_token_limit(prompt=prompt)\n\n    def run(\n        self,\n        query: Optional[str] = None,\n        file_paths: Optional[List[str]] = None,\n        labels: Optional[MultiLabel] = None,\n        documents: Optional[List[Document]] = None,\n        meta: Optional[dict] = None,\n    ) -> Tuple[Dict, str]:\n        raise NotImplementedError(\"This method should never be implemented in the derived class\")\n\n    def run_batch(\n        self,\n        queries: Optional[Union[str, List[str]]] = None,\n        file_paths: Optional[List[str]] = None,\n        labels: Optional[Union[MultiLabel, List[MultiLabel]]] = None,\n        documents: Optional[Union[List[Document], List[List[Document]]]] = None,\n        meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,\n        params: Optional[dict] = None,\n        debug: Optional[bool] = None,\n    ):\n        raise NotImplementedError(\"This method should never be implemented in the derived class\")\n\n    def __repr__(self):\n        return \"{}({!r})\".format(self.__class__.__name__, self.__dict__)\n", "path": "haystack/nodes/prompt/prompt_model.py"}]}