code: def completion(
      self,
      model: str,
      messages: list,
      api_base: Optional[str],
      headers: Optional[dict],
      model_response: ModelResponse,
      print_verbose: Callable,
      timeout: float,
      encoding,
      api_key,
      logging_obj,
      custom_prompt_dict={},
      acompletion: bool = False,
      optional_params=None,
      litellm_params=None,
      logger_fn=None,
  ):
      super().completion()
      exception_mapping_worked = False
      try:
          headers = self.validate_environment(api_key, headers)
          task = get_hf_task_for_model(model)
          print_verbose(f"{model}, {task}")
          completion_url = ""
          input_text = ""
          if "https" in model:
              completion_url = model
          elif api_base:
              completion_url = api_base
          elif "HF_API_BASE" in os.environ:
              completion_url = os.getenv("HF_API_BASE", "")
          elif "HUGGINGFACE_API_BASE" in os.environ:
              completion_url = os.getenv("HUGGINGFACE_API_BASE", "")
          else:
              completion_url = (
                  f"https://api-inference.huggingface.co/models/{model}"
              )
  
          ## Load Config
          config = litellm.HuggingfaceConfig.get_config()
          for k, v in config.items():
              if (
                  k not in optional_params
              ):  # completion(top_k=3) > huggingfaceConfig(top_k=3) <- allows for dynamic variables to be passed in
                  optional_params[k] = v
  
          ### MAP INPUT PARAMS
          if task == "conversational":
              inference_params = copy.deepcopy(optional_params)
              inference_params.pop("details")
              inference_params.pop("return_full_text")
              past_user_inputs = []
              generated_responses = []
              text = ""
              for message in messages:
                  if message["role"] == "user":
                      if text != "":
                          past_user_inputs.append(text)
                      text = message["content"]
                  elif (
                      message["role"] == "assistant"
                      or message["role"] == "system"
                  ):
                      generated_responses.append(message["content"])
              data = {
                  "inputs": {
                      "text": text,
                      "past_user_inputs": past_user_inputs,
                      "generated_responses": generated_responses,
                  },
                  "parameters": inference_params,
              }
              input_text = "".join(
                  message["content"] for message in messages
              )
          elif task == "text-generation-inference":
              # always send "details" and "return_full_text" as params
              if model in custom_prompt_dict:
                  # check if the model has a registered custom prompt
                  model_prompt_details = custom_prompt_dict[model]
                  prompt = custom_prompt(
                      role_dict=model_prompt_details.get("roles", None),
                      initial_prompt_value=model_prompt_details.get(
                          "initial_prompt_value", ""
                      ),
                      final_prompt_value=model_prompt_details.get(
                          "final_prompt_value", ""
                      ),
                      messages=messages,
                  )
              else:
                  prompt = prompt_factory(
                      model=model, messages=messages
                  )
              data = {
                  "inputs": prompt,
                  "parameters": optional_params,
                  "stream": True
                  if "stream" in optional_params
                  and optional_params["stream"] == True
                  else False,
              }
              input_text = prompt
          else:
              # Non TGI and Conversational llms
              # We need this branch, it removes 'details' and 'return_full_text' from params
              if model in custom_prompt_dict:
                  # check if the model has a registered custom prompt
                  model_prompt_details = custom_prompt_dict[model]
                  prompt = custom_prompt(
                      role_dict=model_prompt_details.get("roles", {}),
                      initial_prompt_value=model_prompt_details.get(
                          "initial_prompt_value", ""
                      ),
                      final_prompt_value=model_prompt_details.get(
                          "final_prompt_value", ""
                      ),
                      bos_token=model_prompt_details.get(
                          "bos_token", ""
                      ),
                      eos_token=model_prompt_details.get(
                          "eos_token", ""
                      ),
                      messages=messages,
                  )
              else:
                  prompt = prompt_factory(
                      model=model, messages=messages
                  )
              inference_params = copy.deepcopy(optional_params)
              inference_params.pop("details")
              inference_params.pop("return_full_text")
              data = {
                  "inputs": prompt,
                  "parameters": inference_params,
                  "stream": True
                  if "stream" in optional_params
                  and optional_params["stream"] == True
                  else False,
              }
              input_text = prompt
          ## LOGGING
          logging_obj.pre_call(
              input=input_text,
              api_key=api_key,
              additional_args={
                  "complete_input_dict": data,
                  "task": task,
                  "headers": headers,
                  "api_base": completion_url,
                  "acompletion": acompletion,
              },
          )
          ## COMPLETION CALL
          if acompletion is True:
              ### ASYNC STREAMING
              if optional_params.get("stream", False):
                  return self.async_streaming(logging_obj=logging_obj, api_base=completion_url, data=data, headers=headers, model_response=model_response, model=model, timeout=timeout)  # type: ignore
              else:
                  ### ASYNC COMPLETION
                  return self.acompletion(api_base=completion_url, data=data, headers=headers, model_response=model_response, task=task, encoding=encoding, input_text=input_text, model=model, optional_params=optional_params, timeout=timeout)  # type: ignore
          ### SYNC STREAMING
          if (
              "stream" in optional_params
              and optional_params["stream"] == True
          ):
              response = requests.post(
                  completion_url,
                  headers=headers,
                  data=json.dumps(data),
                  stream=optional_params["stream"],
              )
              return response.iter_lines()
          ### SYNC COMPLETION
          else:
              response = requests.post(
                  completion_url, headers=headers, data=json.dumps(data)
              )
  
              ## Some servers might return streaming responses even though stream was not set to true. (e.g. Baseten)
              is_streamed = False
              if (
                  response.__dict__["headers"].get("Content-Type", "")
                  == "text/event-stream"
              ):
                  is_streamed = True
  
              # iterate over the complete streamed response, and return the final answer
              if is_streamed:
                  streamed_response = CustomStreamWrapper(
                      completion_stream=response.iter_lines(),
                      model=model,
                      custom_llm_provider="huggingface",
                      logging_obj=logging_obj,
                  )
                  content = ""
                  for chunk in streamed_response:
                      content += chunk["choices"][0]["delta"]["content"]
                  completion_response: List[Dict[str, Any]] = [
                      {"generated_text": content}
                  ]
                  ## LOGGING
                  logging_obj.post_call(
                      input=input_text,
                      api_key=api_key,
                      original_response=completion_response,
                      additional_args={
                          "complete_input_dict": data,
                          "task": task,
                      },
                  )
              else:
                  ## LOGGING
                  logging_obj.post_call(
                      input=input_text,
                      api_key=api_key,
                      original_response=response.text,
                      additional_args={
                          "complete_input_dict": data,
                          "task": task,
                      },
                  )
                  ## RESPONSE OBJECT
                  try:
                      completion_response = response.json()
                      if isinstance(completion_response, dict):
                          completion_response = [completion_response]
                  except:
                      import traceback
  
                      raise HuggingfaceError(
                          message=f"Original Response received: {response.text}; Stacktrace: {traceback.format_exc()}",
                          status_code=response.status_code,
                      )
              print_verbose(f"response: {completion_response}")
              if (
                  isinstance(completion_response, dict)
                  and "error" in completion_response
              ):
                  print_verbose(
                      f"completion error: {completion_response['error']}"
                  )
                  print_verbose(
                      f"response.status_code: {response.status_code}"
                  )
                  raise HuggingfaceError(
                      message=completion_response["error"],
                      status_code=response.status_code,
                  )
              return self.convert_to_model_response_object(
                  completion_response=completion_response,
                  model_response=model_response,
                  task=task,
                  optional_params=optional_params,
                  encoding=encoding,
                  input_text=input_text,
                  model=model,
              )
      except HuggingfaceError as e:
          exception_mapping_worked = True
          raise e
      except Exception as e:
          if exception_mapping_worked:
              raise e
          else:
              import traceback
  
              raise HuggingfaceError(
                  status_code=500, message=traceback.format_exc()
              )
  
imports: import os, copy, types
  import json
  from enum import Enum
  import httpx, requests
  from .base import BaseLLM
  import time
  import litellm
  from typing import Callable, Dict, List, Any
  from litellm.utils import ModelResponse, Choices, Message, CustomStreamWrapper, Usage
  from typing import Optional
  from .prompt_templates.factory import prompt_factory, custom_prompt
label_desc: Use image node_id 4 for calling the Huggingface obj's underlying member method code with example usage: obj.completion(model, messages, api_base, headers, model_response, print_verbose, timeout, encoding, api_key, logging_obj, custom_prompt_dict, acompletion, optional_params, litellm_params, logger_fn) and returns: response, self, self, self