51 lines
2.0 KiB
Python
51 lines
2.0 KiB
Python
import triton_python_backend_utils as pb_utils
|
|
from transformers import AutoModelForCausalLM, AutoTokenizer, GenerationConfig
|
|
import numpy as np
|
|
import json
|
|
|
|
class TritonPythonModel:
|
|
def initialize(self, args):
|
|
"""
|
|
모델이 로드될 때 딱 한 번만 호출됩니다.
|
|
`initialize` 함수를 구현하는 것은 선택 사항입니다. 이 함수를 통해 모델은
|
|
이 모델과 관련된 모든 상태를 초기화할 수 있습니다.
|
|
"""
|
|
self.logger = pb_utils.Logger
|
|
|
|
self.model_name = args["model_name"]
|
|
self.model_config = json.loads(args["model_config"])
|
|
|
|
self.logger.log_info(f"'{self.model_name}' 모델 초기화 완료")
|
|
|
|
|
|
def execute(self, requests):
|
|
"""
|
|
Triton이 각 추론 요청에 대해 호출하는 실행 함수입니다.
|
|
"""
|
|
responses = []
|
|
|
|
# 각 추론 요청을 순회하며 처리합니다.
|
|
for request in requests:
|
|
# Triton 입력 파싱
|
|
input_text = self._get_input_value(request, "INPUT")
|
|
self.logger.log_info(f"INPUT 출력:\n{input_text}")
|
|
|
|
|
|
output = "HELLO_WORLD"
|
|
self.logger.log_info(f"OUTPUT 출력:\n{output}")
|
|
|
|
# 생성된 텍스트를 Triton 출력 텐서로 변환합니다.
|
|
output_tensor = pb_utils.Tensor("text_output", np.array(output.encode('utf-8'), dtype=np.bytes_))
|
|
|
|
# 응답 객체를 생성하고 출력 텐서를 추가합니다.
|
|
responses.append(pb_utils.InferenceResponse(output_tensors=[output_tensor]))
|
|
|
|
return responses
|
|
|
|
def finalize(self):
|
|
"""
|
|
모델 실행이 완료된 후 Triton 서버가 종료될 때 한 번 호출되는 함수입니다.
|
|
`finalize` 함수를 구현하는 것은 선택 사항입니다. 이 함수를 통해 모델은
|
|
종료 전에 필요한 모든 정리 작업을 수행할 수 있습니다.
|
|
"""
|
|
pass |