Upload 2 files

.api/config.json

@@ -0,0 +1,21 @@
+{
+    "task": "text-generation",
+    "framework": "pytorch",
+    "runtime": "transformers",
+    "model_id": "johnuwaishe/Nigerian-health-llama-7b",
+    "revision": "main",
+    "handler_path": "handler.py",
+    "requirements": [
+        "torch>=2.0.0",
+        "transformers>=4.37.0",
+        "accelerate>=0.27.0"
+    ],
+    "parameters": {
+        "max_new_tokens": 512,
+        "temperature": 0.7,
+        "top_p": 0.95,
+        "top_k": 50,
+        "repetition_penalty": 1.1,
+        "do_sample": true
+    }
+} 

.api/handler.py

@@ -0,0 +1,119 @@
+from typing import Dict, List
+import torch
+from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
+from threading import Thread
+
+class EndpointHandler:
+    def __init__(self, path=""):
+        # Load model and tokenizer
+        self.tokenizer = AutoTokenizer.from_pretrained(path)
+        self.model = AutoModelForCausalLM.from_pretrained(
+            path,
+            torch_dtype=torch.bfloat16,
+            device_map="auto",
+            trust_remote_code=True
+        )
+        self.model.eval()
+
+    def __call__(self, data: Dict) -> Dict:
+        """Handle a request.
+        Args:
+            data (Dict): Input data for the request.
+                Expected format:
+                {
+                    "inputs": str,
+                    "parameters": {
+                        "max_new_tokens": int,
+                        "temperature": float,
+                        "top_p": float,
+                        "top_k": int,
+                        "repetition_penalty": float,
+                        "do_sample": bool
+                    }
+                }
+        Returns:
+            Dict: Response data.
+                Format:
+                {
+                    "generated_text": str
+                }
+        """
+        # Extract inputs and parameters
+        inputs = data.pop("inputs", data)
+        parameters = data.pop("parameters", {})
+
+        # Set default parameters if not provided
+        max_new_tokens = parameters.get("max_new_tokens", 100)
+        temperature = parameters.get("temperature", 0.7)
+        top_p = parameters.get("top_p", 0.95)
+        top_k = parameters.get("top_k", 50)
+        repetition_penalty = parameters.get("repetition_penalty", 1.1)
+        do_sample = parameters.get("do_sample", True)
+
+        # Tokenize inputs
+        input_ids = self.tokenizer(inputs, return_tensors="pt").input_ids
+        input_ids = input_ids.to(self.model.device)
+
+        # Generate
+        with torch.no_grad():
+            outputs = self.model.generate(
+                input_ids,
+                max_new_tokens=max_new_tokens,
+                temperature=temperature,
+                top_p=top_p,
+                top_k=top_k,
+                repetition_penalty=repetition_penalty,
+                do_sample=do_sample,
+                pad_token_id=self.tokenizer.eos_token_id
+            )
+
+        # Decode and return the generated text
+        generated_text = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
+
+        return {"generated_text": generated_text}
+
+    def stream(self, data: Dict) -> Dict:
+        """Handle a streaming request.
+        Args:
+            data (Dict): Same format as __call__
+        Returns:
+            Iterator[Dict]: Stream of responses.
+        """
+        # Extract inputs and parameters
+        inputs = data.pop("inputs", data)
+        parameters = data.pop("parameters", {})
+
+        # Set default parameters
+        max_new_tokens = parameters.get("max_new_tokens", 100)
+        temperature = parameters.get("temperature", 0.7)
+        top_p = parameters.get("top_p", 0.95)
+        top_k = parameters.get("top_k", 50)
+        repetition_penalty = parameters.get("repetition_penalty", 1.1)
+        do_sample = parameters.get("do_sample", True)
+
+        # Tokenize inputs
+        input_ids = self.tokenizer(inputs, return_tensors="pt").input_ids
+        input_ids = input_ids.to(self.model.device)
+
+        # Create streamer
+        streamer = TextIteratorStreamer(self.tokenizer, skip_special_tokens=True)
+
+        # Generate in a separate thread
+        generation_kwargs = dict(
+            input_ids=input_ids,
+            max_new_tokens=max_new_tokens,
+            temperature=temperature,
+            top_p=top_p,
+            top_k=top_k,
+            repetition_penalty=repetition_penalty,
+            do_sample=do_sample,
+            pad_token_id=self.tokenizer.eos_token_id,
+            streamer=streamer,
+        )
+
+        thread = Thread(target=self.model.generate, kwargs=generation_kwargs)
+        thread.start()
+
+        # Stream the output
+        for text in streamer:
+            yield {"token": {"text": text}}