57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284 | class HuggingFaceInferenceAPI(CustomLLM):
"""
Wrapper on the Hugging Face's Inference API.
Overview of the design:
- Synchronous uses InferenceClient, asynchronous uses AsyncInferenceClient
- chat uses the conversational task: https://huggingface.co/tasks/conversational
- complete uses the text generation task: https://huggingface.co/tasks/text-generation
Note: some models that support the text generation task can leverage Hugging
Face's optimized deployment toolkit called text-generation-inference (TGI).
Use InferenceClient.get_model_status to check if TGI is being used.
Relevant links:
- General Docs: https://huggingface.co/docs/api-inference/index
- API Docs: https://huggingface.co/docs/huggingface_hub/main/en/package_reference/inference_client
- Source: https://github.com/huggingface/huggingface_hub/tree/main/src/huggingface_hub/inference
"""
@classmethod
def class_name(cls) -> str:
return "HuggingFaceInferenceAPI"
# Corresponds with huggingface_hub.InferenceClient
model_name: Optional[str] = Field(
default=None,
description=(
"The model to run inference with. Can be a model id hosted on the Hugging"
" Face Hub, e.g. bigcode/starcoder or a URL to a deployed Inference"
" Endpoint. Defaults to None, in which case a recommended model is"
" automatically selected for the task (see Field below)."
),
)
token: Union[str, bool, None] = Field(
default=None,
description=(
"Hugging Face token. Will default to the locally saved token. Pass "
"token=False if you don’t want to send your token to the server."
),
)
timeout: Optional[float] = Field(
default=None,
description=(
"The maximum number of seconds to wait for a response from the server."
" Loading a new model in Inference API can take up to several minutes."
" Defaults to None, meaning it will loop until the server is available."
),
)
headers: Dict[str, str] = Field(
default=None,
description=(
"Additional headers to send to the server. By default only the"
" authorization and user-agent headers are sent. Values in this dictionary"
" will override the default values."
),
)
cookies: Dict[str, str] = Field(
default=None, description="Additional cookies to send to the server."
)
task: Optional[str] = Field(
default=None,
description=(
"Optional task to pick Hugging Face's recommended model, used when"
" model_name is left as default of None."
),
)
_sync_client: "InferenceClient" = PrivateAttr()
_async_client: "AsyncInferenceClient" = PrivateAttr()
_get_model_info: "Callable[..., ModelInfo]" = PrivateAttr()
context_window: int = Field(
default=DEFAULT_CONTEXT_WINDOW,
description=(
LLMMetadata.model_fields["context_window"].description
+ " This may be looked up in a model's `config.json`."
),
)
num_output: int = Field(
default=DEFAULT_NUM_OUTPUTS,
description=LLMMetadata.model_fields["num_output"].description,
)
is_chat_model: bool = Field(
default=False,
description=(
LLMMetadata.model_fields["is_chat_model"].description
+ " Unless chat templating is intentionally applied, Hugging Face models"
" are not chat models."
),
)
is_function_calling_model: bool = Field(
default=False,
description=(
LLMMetadata.model_fields["is_function_calling_model"].description
+ " As of 10/17/2023, Hugging Face doesn't support function calling"
" messages."
),
)
def _get_inference_client_kwargs(self) -> Dict[str, Any]:
"""Extract the Hugging Face InferenceClient construction parameters."""
return {
"model": self.model_name,
"token": self.token,
"timeout": self.timeout,
"headers": self.headers,
"cookies": self.cookies,
}
def __init__(self, **kwargs: Any) -> None:
"""Initialize.
Args:
kwargs: See the class-level Fields.
"""
if kwargs.get("model_name") is None:
task = kwargs.get("task", "")
# NOTE: task being None or empty string leads to ValueError,
# which ensures model is present
kwargs["model_name"] = InferenceClient.get_recommended_model(task=task)
logger.debug(
f"Using Hugging Face's recommended model {kwargs['model_name']}"
f" given task {task}."
)
if kwargs.get("task") is None:
task = "conversational"
else:
task = kwargs["task"].lower()
super().__init__(**kwargs) # Populate pydantic Fields
self._sync_client = InferenceClient(**self._get_inference_client_kwargs())
self._async_client = AsyncInferenceClient(**self._get_inference_client_kwargs())
self._get_model_info = model_info
def validate_supported(self, task: str) -> None:
"""
Confirm the contained model_name is deployed on the Inference API service.
Args:
task: Hugging Face task to check within. A list of all tasks can be
found here: https://huggingface.co/tasks
"""
all_models = self._sync_client.list_deployed_models(frameworks="all")
try:
if self.model_name not in all_models[task]:
raise ValueError(
"The Inference API service doesn't have the model"
f" {self.model_name!r} deployed."
)
except KeyError as exc:
raise KeyError(
f"Input task {task!r} not in possible tasks {list(all_models.keys())}."
) from exc
def get_model_info(self, **kwargs: Any) -> "ModelInfo":
"""Get metadata on the current model from Hugging Face."""
return self._get_model_info(self.model_name, **kwargs)
@property
def metadata(self) -> LLMMetadata:
return LLMMetadata(
context_window=self.context_window,
num_output=self.num_output,
is_chat_model=self.is_chat_model,
is_function_calling_model=self.is_function_calling_model,
model_name=self.model_name,
)
def chat(self, messages: Sequence[ChatMessage], **kwargs: Any) -> ChatResponse:
# default to conversational task as that was the previous functionality
if self.task == "conversational" or self.task is None:
output: "ConversationalOutput" = self._sync_client.conversational(
**{**chat_messages_to_conversational_kwargs(messages), **kwargs}
)
return ChatResponse(
message=ChatMessage(
role=MessageRole.ASSISTANT, content=output["generated_text"]
)
)
else:
# try and use text generation
prompt = self.messages_to_prompt(messages)
completion = self.complete(prompt)
return ChatResponse(
message=ChatMessage(role=MessageRole.ASSISTANT, content=completion.text)
)
def complete(
self, prompt: str, formatted: bool = False, **kwargs: Any
) -> CompletionResponse:
return CompletionResponse(
text=self._sync_client.text_generation(
prompt, **{**{"max_new_tokens": self.num_output}, **kwargs}
)
)
def stream_chat(
self, messages: Sequence[ChatMessage], **kwargs: Any
) -> ChatResponseGen:
raise NotImplementedError
def stream_complete(
self, prompt: str, formatted: bool = False, **kwargs: Any
) -> CompletionResponseGen:
raise NotImplementedError
async def achat(
self, messages: Sequence[ChatMessage], **kwargs: Any
) -> ChatResponse:
raise NotImplementedError
async def acomplete(
self, prompt: str, formatted: bool = False, **kwargs: Any
) -> CompletionResponse:
response = await self._async_client.text_generation(
prompt, **{**{"max_new_tokens": self.num_output}, **kwargs}
)
return CompletionResponse(text=response)
async def astream_chat(
self, messages: Sequence[ChatMessage], **kwargs: Any
) -> ChatResponseAsyncGen:
raise NotImplementedError
async def astream_complete(
self, prompt: str, formatted: bool = False, **kwargs: Any
) -> CompletionResponseAsyncGen:
raise NotImplementedError
|
