自定義服務¶
本文內容¶
自定義處理器¶
透過編寫 Python 腳本來自定義 TorchServe 的行為，當您使用模型封存器時，該腳本會與模型一起封裝。 TorchServe 在執行時會執行此程式碼。
提供自定義腳本以
初始化模型實例
在將輸入資料發送到模型以進行推論或 Captum 解釋之前，對其進行預處理
自定義模型如何被呼叫以進行推論或解釋
在發送回回應之前，對模型的輸出進行後處理
以下適用於所有類型的自定義處理器
data - 來自傳入請求的輸入資料
context - 是 TorchServe context。 您可以使用以下資訊進行自定義：model_name、model_dir、manifest、batch_size、gpu 等。
從 BaseHandler 開始！¶
BaseHandler 實現了您需要的大部分功能。 您可以從它派生一個新類，如範例和預設處理器中所示。 大多數時候，您只需要覆蓋 preprocess 或 postprocess。
具有 module 層級進入點的自定義處理器¶
自定義處理器檔案必須定義一個模組層級函數，作為執行的進入點。 該函數可以有任何名稱，但它必須接受以下參數並返回預測結果。
進入點函數的簽名是
# Create model object
model = None
def entry_point_function_name(data, context):
"""
Works on data and context to create model object or process inference request.
Following sample demonstrates how model object can be initialized for jit mode.
Similarly you can do it for eager mode models.
:param data: Input data for prediction
:param context: context contains model server system properties
:return: prediction output
"""
global model
if not data:
manifest = context.manifest
properties = context.system_properties
model_dir = properties.get("model_dir")
device = torch.device("cuda:" + str(properties.get("gpu_id")) if torch.cuda.is_available() else "cpu")
# Read model serialize/pt file
serialized_file = manifest['model']['serializedFile']
model_pt_path = os.path.join(model_dir, serialized_file)
if not os.path.isfile(model_pt_path):
raise RuntimeError("Missing the model.pt file")
model = torch.jit.load(model_pt_path)
else:
#infer and return result
return model(data)
此進入點在兩種情況下被啟用
要求 TorchServe 擴展模型以增加後端工作人員的數量（這可以透過
PUT /models/{model_name}請求或具有initial-workers選項的POST /models請求，或者在使用--models選項時在 TorchServe 啟動期間（torchserve --start --models {model_name=model.mar}），即您提供要載入的模型）TorchServe 收到
POST /predictions/{model_name}請求。
(1) 用於擴大或縮小模型的 Worker 數量。(2) 則是用於針對模型執行推論的標準方法。(1) 也稱為模型載入時間。通常，您會希望模型初始化程式碼在模型載入時執行。您可以在TorchServe 管理 API 和 TorchServe 推論 API 中找到更多關於這些 API 和其他 TorchServe API 的資訊。
具有 class 層級進入點的自訂 Handler¶
您可以建立具有任何名稱的 Class 來建立自訂 Handler，但它必須具有 initialize 和 handle 方法。
注意 - 如果您計劃在同一個 Python 模組/檔案中有多個 Class，請確保 Handler Class 是列表中的第一個。
進入點 Class 和函式的簽章是
class ModelHandler(object):
"""
A custom model handler implementation.
"""
def __init__(self):
self._context = None
self.initialized = False
self.model = None
self.device = None
def initialize(self, context):
"""
Invoke by torchserve for loading a model
:param context: context contains model server system properties
:return:
"""
# load the model
self.manifest = context.manifest
properties = context.system_properties
model_dir = properties.get("model_dir")
self.device = torch.device("cuda:" + str(properties.get("gpu_id")) if torch.cuda.is_available() else "cpu")
# Read model serialize/pt file
serialized_file = self.manifest['model']['serializedFile']
model_pt_path = os.path.join(model_dir, serialized_file)
if not os.path.isfile(model_pt_path):
raise RuntimeError("Missing the model.pt file")
self.model = torch.jit.load(model_pt_path)
self.initialized = True
def handle(self, data, context):
"""
Invoke by TorchServe for prediction request.
Do pre-processing of data, prediction using model and postprocessing of prediciton output
:param data: Input data for prediction
:param context: Initial context contains model server system properties.
:return: prediction output
"""
pred_out = self.model.forward(data)
return pred_out
進階自訂 Handler¶
返回自訂錯誤代碼¶
透過具有 module 層級進入點的自訂 Handler，將自訂錯誤代碼返回給使用者。
from ts.utils.util import PredictionException
def handle(data, context):
# Some unexpected error - returning error code 513
raise PredictionException("Some Prediction Error", 513)
透過具有 class 層級進入點的自訂 Handler，將自訂錯誤代碼返回給使用者。
from ts.torch_handler.base_handler import BaseHandler
from ts.utils.util import PredictionException
class ModelHandler(BaseHandler):
"""
A custom model handler implementation.
"""
def handle(self, data, context):
# Some unexpected error - returning error code 513
raise PredictionException("Some Prediction Error", 513)
從頭開始編寫用於預測和解釋請求的自訂 Handler¶
您通常應該從 BaseHandler 派生，並且僅覆寫需要更改行為的方法！ 正如您在範例中所見，大多數時候您只需要覆寫 preprocess 或 postprocess
儘管如此，您可以從頭開始編寫一個 Class。下面是一個例子。基本上，它遵循典型的 Init-Pre-Infer-Post 模式來建立可維護的自訂 Handler。
# custom handler file
# model_handler.py
"""
ModelHandler defines a custom model handler.
"""
from ts.torch_handler.base_handler import BaseHandler
class ModelHandler(BaseHandler):
"""
A custom model handler implementation.
"""
def __init__(self):
self._context = None
self.initialized = False
self.explain = False
self.target = 0
def initialize(self, context):
"""
Initialize model. This will be called during model loading time
:param context: Initial context contains model server system properties.
:return:
"""
self._context = context
self.initialized = True
# load the model, refer 'custom handler class' above for details
def preprocess(self, data):
"""
Transform raw input into model input data.
:param batch: list of raw requests, should match batch size
:return: list of preprocessed model input data
"""
# Take the input data and make it inference ready
preprocessed_data = data[0].get("data")
if preprocessed_data is None:
preprocessed_data = data[0].get("body")
return preprocessed_data
def inference(self, model_input):
"""
Internal inference methods
:param model_input: transformed model input data
:return: list of inference output in NDArray
"""
# Do some inference call to engine here and return output
model_output = self.model.forward(model_input)
return model_output
def postprocess(self, inference_output):
"""
Return inference result.
:param inference_output: list of inference output
:return: list of predict results
"""
# Take output from network and post-process to desired format
postprocess_output = inference_output
return postprocess_output
def handle(self, data, context):
"""
Invoke by TorchServe for prediction request.
Do pre-processing of data, prediction using model and postprocessing of prediciton output
:param data: Input data for prediction
:param context: Initial context contains model server system properties.
:return: prediction output
"""
model_input = self.preprocess(data)
model_output = self.inference(model_input)
return self.postprocess(model_output)
請參考 waveglow_handler 獲取更多詳細資訊。
用於自訂 Handler 的 Captum 解釋¶
TorchServe 返回用於圖像分類、文本分類和 BERT 模型的 Captum 解釋。 它是通過放置以下請求實現的： POST /explanations/{model_name}
這些解釋是作為 Base Handler 的 explain_handle 方法的一部分編寫的。 Base Handler 調用這個 explain_handle_method。 傳遞給 Explain Handle 方法的參數是預處理的數據和原始數據。 它會調用自定義 Handler 的 get insights 函式，該函式會返回 Captum 歸因。 用戶應該編寫自己的 get_insights 功能來獲得解釋。
為了提供自定義 Handler，應該在 Handler 的 Initialize 函式中初始化 Captum 演算法。
使用者可以覆寫自訂 Handler 中的 explain_handle 函式。 用戶應該定義他們的 get_insights 方法，以便自訂 Handler 獲取 Captum 歸因。
上述 ModelHandler Class 應具有以下具有 Captum 功能的方法。
def initialize(self, context):
"""
Load the model and its artifacts
"""
.....
self.lig = LayerIntegratedGradients(
captum_sequence_forward, self.model.bert.embeddings
)
def handle(self, data, context):
"""
Invoke by TorchServe for prediction/explanation request.
Do pre-processing of data, prediction using model and postprocessing of prediction/explanations output
:param data: Input data for prediction/explanation
:param context: Initial context contains model server system properties.
:return: prediction/ explanations output
"""
model_input = self.preprocess(data)
if not self._is_explain():
model_output = self.inference(model_input)
model_output = self.postprocess(model_output)
else :
model_output = self.explain_handle(model_input, data)
return model_output
# Present in the base_handler, so override only when neccessary
def explain_handle(self, data_preprocess, raw_data):
"""Captum explanations handler
Args:
data_preprocess (Torch Tensor): Preprocessed data to be used for captum
raw_data (list): The unprocessed data to get target from the request
Returns:
dict : A dictionary response with the explanations response.
"""
output_explain = None
inputs = None
target = 0
logger.info("Calculating Explanations")
row = raw_data[0]
if isinstance(row, dict):
logger.info("Getting data and target")
inputs = row.get("data") or row.get("body")
target = row.get("target")
if not target:
target = 0
output_explain = self.get_insights(data_preprocess, inputs, target)
return output_explain
def get_insights(self,**kwargs):
"""
Functionality to get the explanations.
Called from the explain_handle method
"""
pass
擴展預設 Handler¶
TorchServe 具有以下預設 Handler。
如果需要，可以擴展以上 Handler 來建立自訂 Handler。此外，您可以擴展抽象 base_handler。
要在 Python 腳本中導入預設 Handler，請使用以下 import 語句。
from ts.torch_handler.<default_handler_name> import <DefaultHandlerClass>
以下是擴展預設 image_classifier Handler 的自訂 Handler 範例。
from ts.torch_handler.image_classifier import ImageClassifier
class CustomImageClassifier(ImageClassifier):
def preprocess(self, data):
"""
Overriding this method for custom preprocessing.
:param data: raw data to be transformed
:return: preprocessed data for model input
"""
# custom pre-procsess code goes here
return data
更多詳細資訊請參考以下範例
使用進入點建立模型封存檔¶
TorchServe 從 Manifest 檔案中識別自訂服務的進入點。當您建立模型封存檔時，請使用 --handler 選項指定進入點的位置。
model-archiver 工具使您可以建立 TorchServe 可以提供的模型封存檔。
torch-model-archiver --model-name <model-name> --version <model_version_number> --handler model_handler[:<entry_point_function_name>] [--model-file <path_to_model_architecture_file>] --serialized-file <path_to_state_dict_file> [--extra-files <comma_seperarted_additional_files>] [--export-path <output-dir> --model-path <model_dir>] [--runtime python3]
注意 -
[] 中的選項是可選的。
如果
entry_point_function_name在您的 Handler 模組中命名為handle或者 Handler 是 Python Class，則可以跳過它。
這將在目錄 <output-dir> 中為 python3 運行時建立檔案 <model-name>.mar。--runtime 參數允許在運行時使用特定的 Python 版本。預設情況下，它使用系統的預設 Python 發行版本。
範例
torch-model-archiver --model-name waveglow_synthesizer --version 1.0 --model-file waveglow_model.py --serialized-file nvidia_waveglowpyt_fp32_20190306.pth --handler waveglow_handler.py --extra-files tacotron.zip,nvidia_tacotron2pyt_fp32_20190306.pth
處理多個 GPU 上的模型執行¶
TorchServe 在 vCPU 或 GPU 上擴展後端 Worker。 在多個 GPU 的情況下，TorchServe 以循環方式選擇 GPU 設備，並將此設備 ID 傳遞給上下文物件中的模型 Handler。 用戶應使用此 GPU ID 建立 PyTorch 設備物件，以確保所有 Worker 都沒有在同一個 GPU 中建立。 以下程式碼片段可用於模型 Handler 中來建立 PyTorch 設備物件。
import torch
class ModelHandler(object):
"""
A base Model handler implementation.
"""
def __init__(self):
self.device = None
def initialize(self, context):
properties = context.system_properties
self.device = torch.device("cuda:" + str(properties.get("gpu_id")) if torch.cuda.is_available() else "cpu")
安裝模型特定的 Python 相依性¶
自訂模型/ Handler 可能依賴於不同的 Python 包，這些包預設情況下未作為 TorchServe 安裝的一部分安裝。
以下步驟允許使用者提供要由 TorchServe 安裝的自訂 Python 包列表，以實現無縫模型服務。
