Upload handler.py with huggingface_hub

handler.py

@@ -0,0 +1,89 @@
+from typing import Dict, List, Any
+from transformers import AutoModelForCausalLM, AutoTokenizer
+import torch
+
+
+class EndpointHandler:
+    """
+    Custom handler for DoloresAI model on HuggingFace Inference Endpoints.
+    """
+
+    def __init__(self, path=""):
+        """
+        Initialize the handler with the model and tokenizer.
+
+        Args:
+            path (str): Path to the model directory
+        """
+        # Load tokenizer and model
+        self.tokenizer = AutoTokenizer.from_pretrained(path)
+        self.model = AutoModelForCausalLM.from_pretrained(
+            path,
+            torch_dtype=torch.float16,
+            device_map="auto",
+            low_cpu_mem_usage=True
+        )
+
+        # Verify vocab sizes match
+        assert self.model.config.vocab_size == len(self.tokenizer), \
+            f"Vocab size mismatch: model={self.model.config.vocab_size}, tokenizer={len(self.tokenizer)}"
+
+    def __call__(self, data: Dict[str, Any]) -> List[Dict[str, str]]:
+        """
+        Process inference requests.
+
+        Args:
+            data (Dict): Input data with format:
+                {
+                    "inputs": str,  # The prompt text
+                    "parameters": {  # Optional generation parameters
+                        "max_new_tokens": int,
+                        "temperature": float,
+                        "top_p": float,
+                        "do_sample": bool,
+                        "repetition_penalty": float
+                    }
+                }
+
+        Returns:
+            List[Dict]: Generated text response
+        """
+        # Extract inputs
+        inputs = data.pop("inputs", data)
+        parameters = data.pop("parameters", {})
+
+        # Default generation parameters
+        max_new_tokens = parameters.get("max_new_tokens", 512)
+        temperature = parameters.get("temperature", 0.7)
+        top_p = parameters.get("top_p", 0.9)
+        do_sample = parameters.get("do_sample", True)
+        repetition_penalty = parameters.get("repetition_penalty", 1.1)
+
+        # Tokenize input
+        input_ids = self.tokenizer(
+            inputs,
+            return_tensors="pt",
+            truncation=True,
+            max_length=self.model.config.max_position_embeddings - max_new_tokens
+        ).input_ids.to(self.model.device)
+
+        # Generate response
+        with torch.no_grad():
+            outputs = self.model.generate(
+                input_ids,
+                max_new_tokens=max_new_tokens,
+                temperature=temperature,
+                top_p=top_p,
+                do_sample=do_sample,
+                repetition_penalty=repetition_penalty,
+                pad_token_id=self.tokenizer.eos_token_id,
+                eos_token_id=self.tokenizer.eos_token_id,
+            )
+
+        # Decode output
+        generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
+
+        # Remove the input prompt from the response
+        response_text = generated_text[len(inputs):].strip()
+
+        return [{"generated_text": response_text}]