| # import torch | |
| # from typing import Dict, List, Any | |
| # from transformers import AutoTokenizer, AutoModelForCausalLM | |
| # class EndpointHandler: | |
| # def __init__(self, path: str = ""): | |
| # self.tokenizer = AutoTokenizer.from_pretrained(path, padding_side = "left") | |
| # self.model = AutoModelForCausalLM.from_pretrained(path, device_map = "auto", torch_dtype=torch.float16) | |
| # def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]: | |
| # """ | |
| # Args: | |
| # data (:obj:): | |
| # includes the input data and the parameters for the inference. | |
| # Return: | |
| # A :obj:`list`:. The list contains the answer and scores of the inference inputs | |
| # """ | |
| # # process input | |
| # inputs_dict = data.pop("inputs", data) | |
| # parameters = data.pop("parameters", {}) | |
| # prompts = [f"<human>: {prompt}\n<bot>:" for prompt in inputs_dict] | |
| # self.tokenizer.pad_token = self.tokenizer.eos_token | |
| # inputs = self.tokenizer(prompts, truncation=True, max_length=2048-512, | |
| # return_tensors='pt', padding=True).to(self.model.device) | |
| # input_length = inputs.input_ids.shape[1] | |
| # if parameters.get("deterministic", False): | |
| # torch.manual_seed(42) | |
| # outputs = self.model.generate( | |
| # **inputs, max_new_tokens=512, do_sample=True, temperature=0.7, top_p=0.7, top_k=50 | |
| # ) | |
| # output_strs = self.tokenizer.batch_decode(outputs[:, input_length:], skip_special_tokens=True) | |
| # return {"generated_text": output_strs} | |
| # import torch | |
| # from typing import Dict, List, Any | |
| # from transformers import AutoTokenizer, AutoModelForCausalLM | |
| # class EndpointHandler(): | |
| # def __init__(self, path: str = ""): | |
| # self.tokenizer = AutoTokenizer.from_pretrained(path, padding_side = "left") | |
| # self.model = AutoModelForCausalLM.from_pretrained(path, device_map = "auto", torch_dtype=torch.float16) | |
| # def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]: | |
| # """ | |
| # Args: | |
| # data (:obj:): | |
| # includes the input data and the parameters for the inference. | |
| # Return: | |
| # A :obj:`list`:. The list contains the answer and scores of the inference inputs | |
| # """ | |
| # # process input | |
| # inputs_list = data.pop("inputs", data) | |
| # parameters = data.pop("parameters", {}) | |
| # prompts = [f"<human>: {prompt}\n<bot>:" for prompt in inputs_list] | |
| # self.tokenizer.pad_token = self.tokenizer.eos_token | |
| # inputs = self.tokenizer(prompts, truncation=True, max_length=2048-512, | |
| # return_tensors='pt', padding=True).to(self.model.device) | |
| # input_length = inputs.input_ids.shape[1] | |
| # if parameters.get("deterministic", False): | |
| # torch.manual_seed(42) | |
| # outputs = self.model.generate( | |
| # **inputs, max_new_tokens=512, do_sample=True, temperature=0.7, top_p=0.7, top_k=50 | |
| # ) | |
| # output_strs = self.tokenizer.batch_decode(outputs[:, input_length:], skip_special_tokens=True) | |
| # return {"generated_text": output_strs} | |
| import torch | |
| from transformers import AutoTokenizer, AutoModelForCausalLM, StoppingCriteria, StoppingCriteriaList | |
| from typing import Dict, List, Any | |
| class StopWordsCriteria(StoppingCriteria): | |
| def __init__(self, stop_words, tokenizer): | |
| self.tokenizer = tokenizer | |
| self.stop_words = stop_words | |
| self._cache_str = '' | |
| def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor, **kwargs) -> bool: | |
| self._cache_str += self.tokenizer.decode(input_ids[0, -1]) | |
| for stop_words in self.stop_words: | |
| if stop_words in self._cache_str: | |
| return True | |
| return False | |
| class EndpointHandler(): | |
| def __init__(self, path: str = ""): | |
| self.tokenizer = AutoTokenizer.from_pretrained(path, padding_side = "left") | |
| self.tokenizer.pad_token = self.tokenizer.eos_token | |
| self.model = AutoModelForCausalLM.from_pretrained(path, device_map = "auto", torch_dtype=torch.float16) | |
| def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Args: | |
| data (:obj:): | |
| includes the input data and the parameters for the inference. | |
| Return: | |
| A :obj:`list`:. The list contains the answer and scores of the inference inputs | |
| """ | |
| # process input | |
| inputs_list = data.pop("inputs", data) | |
| parameters = data.pop("parameters", {}) | |
| prompts = [f"<human>: {prompt}\n<bot>:" for prompt in inputs_list] | |
| if parameters.get("preset_truncation_token"): | |
| preset_truncation_token_value = parameters["preset_truncation_token"] | |
| DELIMETER = " " | |
| prompts = [DELIMETER.join(prompt.split(DELIMETER)[:preset_truncation_token_value]) for prompt in prompts] | |
| print("45", prompts) | |
| del parameters["preset_truncation_token"] | |
| with torch.no_grad(): | |
| inputs = self.tokenizer(prompts, truncation=True, max_length=2048-512, | |
| return_tensors='pt', padding=True).to(self.model.device) | |
| input_length = inputs.input_ids.shape[1] | |
| if parameters.get("deterministic_seed", False): | |
| torch.manual_seed(parameters["deterministic_seed"]) | |
| del parameters["deterministic_seed"] | |
| outputs = self.model.generate( | |
| **inputs, **parameters, | |
| stopping_criteria=StoppingCriteriaList( | |
| [StopWordsCriteria(['\n<human>:'], self.tokenizer)] | |
| ) | |
| ) | |
| output_strs = self.tokenizer.batch_decode(outputs[:, input_length:], skip_special_tokens=True) | |
| output_strs = [output_str.replace("\n<human>:", "") for output_str in output_strs] | |
| return {"generated_text": output_strs} |