from typing import Dict, Any, List import torch from transformers import AutoModelForCausalLM, AutoTokenizer class EndpointHandler: def __init__(self, path=""): """Initialize the model and tokenizer. Args: path (str): Path to the model directory. Defaults to empty string. """ self.device = "cuda" if torch.cuda.is_available() else "cpu" self.model = AutoModelForCausalLM.from_pretrained( path or "merged", torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32, device_map="auto" ) self.tokenizer = AutoTokenizer.from_pretrained(path or "merged") def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]: """Handle inference requests. Args: data (Dict[str, Any]): The input data. Can be in two formats: 1. Standard Hugging Face format: { "inputs": str, "parameters": Dict[str, Any] } 2. Custom format: { "instruction": str, "input": str (optional), "max_new_tokens": int (optional), "temperature": float (optional) } Returns: Dict[str, Any]: The model's response containing: - response (str): The generated text """ # Handle standard Hugging Face format if "inputs" in data: instruction = data["inputs"] parameters = data.get("parameters", {}) input_text = "" max_new_tokens = parameters.get("max_new_tokens", 512) temperature = parameters.get("temperature", 0.7) # Handle custom format else: instruction = data.get("instruction", "") input_text = data.get("input", "") max_new_tokens = data.get("max_new_tokens", 512) temperature = data.get("temperature", 0.7) # Create prompt prompt = f"""Below is an instruction that describes a task. Write a response that appropriately completes the request. ### Instruction: {instruction}""" if input_text: prompt += f""" ### Input: {input_text}""" prompt += """ ### Response:""" # Generate response inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device) outputs = self.model.generate( **inputs, max_new_tokens=max_new_tokens, temperature=temperature, do_sample=True, pad_token_id=self.tokenizer.eos_token_id ) # Decode and extract response full_response = self.tokenizer.decode(outputs[0], skip_special_tokens=True) response = full_response.split("### Response:")[-1].strip() return {"response": response}