Updated handler for CPU usage

handler.py

@@ -0,0 +1,89 @@
+from typing import Any, Dict, List
+from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
+import torch
+import os
+
+MAX_INPUT_LENGTH = 256   
+MAX_OUTPUT_LENGTH = 128 
+
+class EndpointHandler:
+    def __init__(self, model_dir: str = "", num_threads: int | None = None, generation_config: Dict[str, Any] | None = None, **kwargs: Any) -> None:
+        # Set environment hints for CPU efficiency
+        os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
+
+        # Configure torch threading for CPU
+        if num_threads:
+            try:
+                torch.set_num_threads(num_threads)
+                torch.set_num_interop_threads(max(1, num_threads // 2))
+            except Exception:
+                pass
+            os.environ.setdefault("OMP_NUM_THREADS", str(num_threads))
+            os.environ.setdefault("MKL_NUM_THREADS", str(num_threads))
+
+        self.device = "cpu"  # Force CPU usage
+
+        # Load tokenizer & model with CPU-friendly settings
+        self.tokenizer = AutoTokenizer.from_pretrained(model_dir)
+        self.model = AutoModelForSeq2SeqLM.from_pretrained(model_dir, low_cpu_mem_usage=True)
+        self.model.eval()
+        self.model.to(self.device)
+
+        # Optional bfloat16 cast on CPU (beneficial on Sapphire Rapids/oneDNN)
+        self._use_bf16 = False
+        if os.getenv("ENABLE_BF16", "1") == "1":
+            try:
+                self.model = self.model.to(dtype=torch.bfloat16)
+                self._use_bf16 = True
+            except Exception:
+                self._use_bf16 = False
+
+        # Determine a safe pad token id
+        pad_id = self.tokenizer.pad_token_id if self.tokenizer.pad_token_id is not None else self.tokenizer.eos_token_id
+
+        # Default fast generation config (greedy) overridable by caller
+        default_gen = {
+            "max_length": MAX_OUTPUT_LENGTH,
+            "num_beams": 1,              # Greedy for CPU speed
+            "do_sample": False,
+            "no_repeat_ngram_size": 3,
+            "early_stopping": True,
+            "use_cache": True,
+            "pad_token_id": pad_id,
+        }
+        if generation_config:
+            default_gen.update(generation_config)
+        self.generation_args = default_gen
+
+    def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
+        inputs = data.get("inputs")
+        if not inputs:
+            raise ValueError("No 'inputs' found in the request data.")
+
+        if isinstance(inputs, str):
+            inputs = [inputs]
+
+        # Allow per-request overrides under 'parameters'
+        per_request_params = data.get("parameters") or {}
+        gen_args = {**self.generation_args, **per_request_params}
+
+        tokenized_inputs = self.tokenizer(
+            inputs,
+            max_length=MAX_INPUT_LENGTH,
+            padding=True,
+            truncation=True,
+            return_tensors="pt"
+        ).to(self.device)
+
+        try:
+            with torch.inference_mode():
+                outputs = self.model.generate(
+                    tokenized_inputs["input_ids"],
+                    attention_mask=tokenized_inputs["attention_mask"],
+                    **gen_args
+                )
+            decoded_outputs = self.tokenizer.batch_decode(outputs, skip_special_tokens=True)
+            results = [{"generated_text": text} for text in decoded_outputs]
+            return results
+        except Exception as e:
+            return [{"generated_text": f"Error: {str(e)}"}]