| from typing import Dict, Any |
| from deepsparse import Pipeline |
| from time import perf_counter |
|
|
| class EndpointHandler: |
|
|
| def __init__(self, path=""): |
|
|
| self.pipeline = Pipeline.create(task="text-classification", model_path=path, scheduler="sync") |
|
|
| def __call__(self, data: Dict[str, Any]) -> Dict[str, str]: |
| """ |
| Args: |
| data (:obj:): prediction input text |
| """ |
| inputs = data.pop("inputs", data) |
|
|
| start = perf_counter() |
| prediction = self.pipeline(inputs) |
| end = perf_counter() |
| latency = end - start |
|
|
| return { |
| "labels": prediction.labels, |
| "scores": prediction.scores, |
| "latency (secs.)": latency |
| } |
