From 536131678e94c4ef029f49d824f3bb4e8be33fb2 Mon Sep 17 00:00:00 2001 From: groupuser Date: Fri, 21 Nov 2025 00:42:00 +0000 Subject: [PATCH] =?UTF-8?q?Automatically=20created=20from=20=EB=AA=A8?= =?UTF-8?q?=EB=8D=B8=20=EB=B0=B0=ED=8F=AC(529:Llama-3.2-1B-Instruct)=20by?= =?UTF-8?q?=20=EA=B7=B8=EB=A3=B9=EC=82=AC=EC=9A=A9=EC=9E=90(groupuser)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- 1/model.py | 287 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 287 insertions(+) create mode 100644 1/model.py diff --git a/1/model.py b/1/model.py new file mode 100644 index 0000000..b83692b --- /dev/null +++ b/1/model.py @@ -0,0 +1,287 @@ +import torch +import numpy as np +import json +import triton_python_backend_utils as pb_utils +from transformers import AutoModelForCausalLM, AutoTokenizer, GenerationConfig +from peft import PeftModel, PeftConfig + +class TritonPythonModel: + def initialize(self, args): + """ + 모델이 로드될 때 딱 한 번만 호출됩니다. + `initialize` 함수를 구현하는 것은 선택 사항입니다. 이 함수를 통해 모델은 + 이 모델과 관련된 모든 상태를 초기화할 수 있습니다. + + Parameters + ---------- + args : dict + Both keys and values are strings. The dictionary keys and values are: + * model_config: A JSON string containing the model configuration + * model_instance_kind: A string containing model instance kind + * model_instance_device_id: A string containing model instance device + ID + * model_repository: Model repository path + * model_version: Model version + * model_name: Model name + """ + self.logger = pb_utils.Logger + + self.model_config = json.loads(args["model_config"]) + + self.model_name = args["model_name"] + self.base_model_path = self._get_config_parameter("base_model_path") + self.is_adapter_model = self._get_config_parameter("is_adapter_model").strip().lower() == "true" + self.adapter_model_path = self._get_config_parameter("adapter_model_path") + self.quantization = self._get_config_parameter("quantization") + + self.logger.log_info(f"base_model_path: {self.base_model_path}") + self.logger.log_info(f"is_adapter_model: {self.is_adapter_model}") + self.logger.log_info(f"adapter_model_path: {self.adapter_model_path}") + self.logger.log_info(f"quantization: {self.quantization}") + + self.load_model() + + def load_model(self): + """ + Load model + """ + self.bnb_config = None + torch_dtype = torch.float16 # 기본 dtype (필요시 bfloat16 등으로 조절) + + # 양자화 옵션 체크 + if self.quantization == "int4": + self.bnb_config = BitsAndBytesConfig( + load_in_4bit=True, + bnb_4bit_use_double_quant=True, + bnb_4bit_quant_type="nf4", + bnb_4bit_compute_dtype=torch_dtype + ) + elif self.quantization == "int8": + self.bnb_config = BitsAndBytesConfig( + load_in_8bit=True, + llm_int8_threshold=6.0, + llm_int8_has_fp16_weight=True + ) + else: + self.bnb_config = None + + if self.is_adapter_model: + # 어댑터 설정 정보 로드 + peft_config = PeftConfig.from_pretrained(self.adapter_model_path) + self.base_model_path = peft_config.base_model_name_or_path + + # base 모델 로드 + base_model = AutoModelForCausalLM.from_pretrained( + peft_config.base_model_name_or_path, + torch_dtype=torch.float16, + quantization_config=self.bnb_config if self.bnb_config else None, + device_map="auto" + ) + + # adapter 모델 로드 (base 모델 위에 덧씌움) + self.model = PeftModel.from_pretrained(base_model, self.adapter_model_path) + else: + # 일반 모델인 경우 로드 + self.model = AutoModelForCausalLM.from_pretrained( + pretrained_model_name_or_path=self.base_model_path, + local_files_only=True, + quantization_config=self.bnb_config if self.bnb_config else None, + device_map="auto" + ) + + # Tokenizer는 base model의 tokenizer 사용 + self.tokenizer = AutoTokenizer.from_pretrained(self.base_model_path) + self.tokenizer.pad_token_id = self.tokenizer.eos_token_id + self.supports_chat_template = self._check_chat_template_support() + + 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, "text_input") + + text = "" + conversation = "" + input_token_length = 0 # 입력 토큰 길이를 저장할 변수 + + # 입력 텍스트가 JSON 형식의 대화 기록인지 확인합니다. + try: + conversation = json.loads(input_text) + is_chat = True + self.logger.log_info(f"입력 conversation 출력:\n{conversation}") + except: + # JSON 파싱에 실패하면 일반 텍스트로 처리합니다. + text = input_text + is_chat = False + self.logger.log_info(f"입력 text 출력:\n{text}") + + # 입력 텍스트를 토큰화합니다. + if self.supports_chat_template and is_chat: + self.logger.log_info(f"Chat 템플릿을 적용하여 토큰화합니다.") + inputs = self.tokenizer.apply_chat_template( + conversation, + tokenize=True, + add_generation_prompt=True, + return_tensors="pt", + return_dict=True + ).to(device=self.model.device) + else: + self.logger.log_info(f"입력 텍스트를 토큰화합니다.") + inputs = self.tokenizer( + text, + return_tensors="pt").to(device=self.model.device) + + input_ids = inputs["input_ids"] + attention_mask = inputs["attention_mask"] + input_token_length = inputs["input_ids"].shape[-1] + + + # 언어 모델을 사용하여 텍스트를 생성합니다. + gened = self.model.generate( + input_ids=input_ids, + attention_mask=attention_mask, + generation_config=self._process_generation_config(request), + pad_token_id=self.tokenizer.pad_token_id, + ) + + # 생성된 토큰 시퀀스를 텍스트로 디코딩하고 입력 텍스트는 제외합니다. + generated_tokens = gened[0][input_token_length:] # 입력 토큰 이후부터 슬라이싱 + gened_text = self.tokenizer.decode(generated_tokens, skip_special_tokens=True) + self.logger.log_info(f"모델이 생성한 토큰 시퀀스 (입력 텍스트 제외):\n{gened_text}") + + output = gened_text.strip() + + # 생성된 텍스트를 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 _process_generation_config(self, request): + """ + 추론 요청에서 생성 설정 관련 파라미터들을 추출하여 GenerationConfig 객체를 생성합니다. + + Args: + request (pb_utils.InferenceRequest): Triton 추론 요청 객체. + + Returns: + transformers.GenerationConfig: GenerationConfig 객체. + """ + max_length = self._get_input_value(request, "max_length", default=20) + max_new_tokens = self._get_input_value(request, "max_new_tokens") + temperature = self._get_input_value(request, "temperature") + do_sample = self._get_input_value(request, "do_sample") + top_k = self._get_input_value(request, "top_k") + top_p = self._get_input_value(request, "top_p") + repetition_penalty = self._get_input_value(request, "repetition_penalty") + stream = self._get_input_value(request, "stream") + + generation_config = GenerationConfig( + max_length=max_length, + max_new_tokens=max_new_tokens, + temperature=temperature, + do_sample=do_sample, + top_k=top_k, + top_p=top_p, + repetition_penalty=repetition_penalty, + stream=stream, + ) + + self.logger.log_info(f"추론 요청 GenerationConfig:\n{generation_config}") + + return generation_config + + def _get_config_parameter(self, parameter_name): + """ + 모델 설정(config.pbtxt)에서 특정 파라미터의 문자열 값을 가져옵니다. + + Args: + parameter_name (str): 가져올 파라미터의 이름. + + Returns: + str or None: 파라미터의 'string_value' 또는 해당 파라미터가 없거나 'string_value' 키가 없는 경우 None. + """ + self.parameters = self.model_config.get('parameters', {}) + parameter_dict = self.parameters.get(parameter_name) + + if isinstance(parameter_dict, dict) and 'string_value' in parameter_dict: + return parameter_dict['string_value'] + + return None + + def _check_chat_template_support(self): + """ + 주어진 허깅페이스 Transformer 모델이 Chat 템플릿을 지원하는지 확인하고 결과를 출력합니다. + + Returns: + bool: Chat 템플릿 지원 여부 (True 또는 False). + """ + try: + if hasattr(self.tokenizer, "chat_template") and self.tokenizer.chat_template is not None: + self.logger.log_info(f"'{self.model_name}' 모델의 토크나이저는 Chat 템플릿을 지원합니다.") + self.logger.log_info("Chat 템플릿 내용:") + self.logger.log_info(self.tokenizer.chat_template) + return True + else: + self.logger.log_info(f"'{self.model_name}' 모델의 토크나이저는 Chat 템플릿을 직접적으로 지원하지 않거나, Chat 템플릿 정보가 없습니다.") + return False + except Exception as e: + self.logger.log_info(f"'{self.model_name}' 모델의 토크나이저를 로드하는 동안 오류가 발생했습니다: {e}") + return False + + + def _get_input_value(self, request, input_name: str, default=None): + """ + Triton 추론 요청에서 특정 이름의 입력 텐서 값을 가져옵니다. + + Args: + request (pb_utils.InferenceRequest): Triton 추론 요청 객체. + input_name (str): 가져올 입력 텐서의 이름. + default (any, optional): 입력 텐서가 없을 경우 반환할 기본값. Defaults to None. + + Returns: + any: 디코딩된 입력 텐서의 값. 텐서가 없으면 기본값을 반환합니다. + """ + tensor_value = pb_utils.get_input_tensor_by_name(request, input_name) + + if tensor_value is None: + return default + + return self._np_decoder(tensor_value.as_numpy()[0]) + + def _np_decoder(self, obj): + """ + NumPy 객체의 데이터 타입을 확인하고 Python 기본 타입으로 변환합니다. + + Args: + obj (numpy.ndarray element): 변환할 NumPy 배열의 요소. + + Returns: + any: 해당 NumPy 요소에 대응하는 Python 기본 타입 (str, int, float, bool). + bytes 타입인 경우 UTF-8로 디코딩합니다. + """ + if isinstance(obj, bytes): + return obj.decode('utf-8') + if np.issubdtype(obj, np.integer): + return int(obj) + if np.issubdtype(obj, np.floating): + return round(float(obj), 3) + if isinstance(obj, np.bool_): + return bool(obj) + + def finalize(self): + """ + 모델 실행이 완료된 후 Triton 서버가 종료될 때 한 번 호출되는 함수입니다. + `finalize` 함수를 구현하는 것은 선택 사항입니다. 이 함수를 통해 모델은 + 종료 전에 필요한 모든 정리 작업을 수행할 수 있습니다. + """ + pass