Generon_QC_merge / handler.py
DinoLiu's picture
Upload handler.py with huggingface_hub
3b01827 verified
Raw
History Blame Contribute Delete
4.49 kB
from typing import Dict, List, Any
from transformers import pipeline
import torch
from datetime import datetime
import pytz
from pathlib import Path
class EndpointHandler:
def __init__(self, model_path: str = "", tokenizer_path: str = None):
"""
Initialize the endpoint handler with model and system instruction
Args:
model_path: Path to the model
tokenizer_path: Path to the tokenizer (if different from model_path)
"""
# Set default tokenizer path if not provided
if tokenizer_path is None:
tokenizer_path = model_path
# Initialize the pipeline
self.pipeline = pipeline(
"text-generation",
model=model_path,
tokenizer=tokenizer_path,
max_length=2000,
device_map="cuda" if torch.cuda.is_available() else "cpu"
)
# System instruction
self.system_instruction =
"""You are an AI that generates a user-friendly user interface based on a user query using the µUI script language. You must output the µUI script covering the sufficient interface that is needed without any other content.
You must include the information mentioned by the user usable to construct your interface.
Only when the request is unclear or require mandatory context or information to generate UI for it, you generate the right UI to collect such missing information or clarify the context. You also must use µUI (MicroUI) Language for the information collection UI generation as well.
Don't output any other content except µUI script. """
# Default location
self.default_location = "San Francisco, USA"
def get_formatted_time(self, timezone: str = "America/Los_Angeles") -> str:
"""Get formatted time for given timezone"""
try:
tz = pytz.timezone(timezone)
current_time = datetime.now(tz)
return current_time.strftime('%Y-%m-%d %H:%M:%S %Z')
except:
# Default to PT if timezone is invalid
tz = pytz.timezone("America/Los_Angeles")
current_time = datetime.now(tz)
return current_time.strftime('%Y-%m-%d %H:%M:%S %Z')
def format_message(self, query: str, location: str = None, time: str = None) -> str:
"""Format the user message with location and time"""
location = location if location else self.default_location
time = time if time else self.get_formatted_time()
return f"Current User query: {query}\nCurrent Location: {location}\nCurrent Time: {time}"
def __call__(self, data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""
Handle the inference request
Args:
data: Dictionary containing:
- inputs (str): User query
- location (str, optional): User location
- time (str, optional): Custom time
Returns:
List[Dict]: Model response with metadata
"""
try:
# Extract inputs
if isinstance(data, dict):
inputs = data.pop("inputs", None)
if inputs is None:
# If no inputs field, treat entire data as input
inputs = data
location = data.pop("location", None)
time = data.pop("time", None)
else:
# If data is not a dict, treat it as the input
inputs = data
location = None
time = None
# Format messages
messages = [
{"role": "system", "content": self.system_instruction},
{"role": "user", "content": self.format_message(inputs, location, time)}
]
# Run inference
output = self.pipeline(messages)
# Extract and format response
response = {
"generated_text": output[0]["generated_text"][2]["content"],
"metadata": {
"location": location if location else self.default_location,
"time": time if time else self.get_formatted_time()
}
}
return [response]
except Exception as e:
return [{"error": f"Inference failed: {str(e)}"}]