LLaVA에 대해 알아보자 - Predict

Vision과 LLM이 결합된 VLM, 그 중 하나인 LLaVA에 대해 코드 수준으로 알아보려 한다.

핵심 기능 하나씩 어떤 역할을 하는지 알아보자

 

오늘 알아볼 내용은 LLaVA의 Predict 과정입니다.

알아보기 앞서, LLavA는 Vision을 담당하는 ViT모델 - CLIP 모델, Language를 담당하는 LLM으로 이루어져있습니다.

 

추론이 작동하는 방식은 다음 단계와 같습니다.

 

1. 이미지를 Vision Model로 디지털 데이터화 (Feature)
2. 데이터를 이후에 같이 LLM에 넣을 언어 데이터 형식과 동일하게 변경 (Tensor)
3. 언어(프롬프트)를 LLM에 넣을 형식으로 변경 (Tensor)
4. 두 데이터를 하나로 묶음 (Linear)
5. LLM에 입력 후 추론 & 결과 출력

 

이제 하나씩 보도록 하겠습니다.

코드는 LLaVA 공식 깃허브에서 가져왔습니다.

URL : https://github.com/haotian-liu/LLaVA

GitHub - haotian-liu/LLaVA: [NeurIPS'23 Oral] Visual Instruction Tuning (LLaVA) built towards GPT-4V level capabilities and beyo

 

1. load_image (Function)

이미지의 경로(로컬 혹은 온라인)을 받아 Image 형식으로 변환해주는 함수입니다.

코드

def load_image(image_file):
    if image_file.startswith('http') or image_file.startswith('https'):
        response = requests.get(image_file)
        image = Image.open(BytesIO(response.content)).convert('RGB')
    else:
        image = Image.open(image_file).convert('RGB')
    return image

 

2. Predictor (Class)

Predict를 수행하는 클래스입니다.

클래스 내 구조는 아래와 같습니다.

class Predictor(BasePredictor):
    def setup(self) -> None:
        """Load the model into memory to make running multiple predictions efficient"""
        
    def predict(
        self,
        image: Path = Input(description="Input image"),
        prompt: str = Input(description="Prompt to use for text generation"),
        top_p: float = Input(description="When decoding text, samples from the top p percentage of most likely tokens; lower to ignore less likely tokens", ge=0.0, le=1.0, default=1.0),
        temperature: float = Input(description="Adjusts randomness of outputs, greater than 1 is random and 0 is deterministic", default=0.2, ge=0.0),
        max_tokens: int = Input(description="Maximum number of tokens to generate. A word is generally 2-3 tokens", default=1024, ge=0),
    ) -> ConcatenateIterator[str]:
        """Run a single prediction on the model"""

1-1. setup

코드

    def setup(self) -> None:
        """Load the model into memory to make running multiple predictions efficient"""
        for weight in weights:
            download_weights(weight["src"], weight["dest"], weight["files"])
        disable_torch_init()
    
        self.tokenizer, self.model, self.image_processor, self.context_len = load_pretrained_model("liuhaotian/llava-v1.5-13b", model_name="llava-v1.5-13b", model_base=None, load_8bit=False, load_4bit=False)

Predictor 클래스를 준비하는 메소드 입니다.

• Tokenizer (텍스트를 받아서 토큰으로 변환해주는 함수 등이 있는 객체)
• Model (LLM 모델)
• ImageProcessor (Vision Model에서부터 나온 Vision Model에 이미지를 넣을 수 있도록 전처리 해주는 함수)
• ContextLengh (Model에 최대로 들어갈 수 있는 토큰의 수)

를 준비하여 클래스 안에 준비해둡니다.

1-2. predict

이미지와 언어(프롬프트)를 받아 LLM모델에 들어가서 추론 및 결과를 리턴해주는 메소드입니다.

 

더 자세히 설명하도록 하겠습니다.

 

1-2-1. 입력 변수

코드

image: Path = Input(description="Input image"),
prompt: str = Input(description="Prompt to use for text generation"),
top_p: float = Input(description="When decoding text, samples from the top p percentage of most likely tokens; lower to ignore less likely tokens", ge=0.0, le=1.0, default=1.0),
temperature: float = Input(description="Adjusts randomness of outputs, greater than 1 is random and 0 is deterministic", default=0.2, ge=0.0),
max_tokens: int = Input(description="Maximum number of tokens to generate. A word is generally 2-3 tokens", default=1024, ge=0),

• image : 이미지 경로를 입력 받습니다.
• prompt : 언어(프롬프트) 데이터를 받습니다.
• top_p : LLM이 결과를 생성할 때, 다음 결과로 나올 수 있는 경우의 수를 정합니다.
• 0~1사이 값이며, 1에 가까울 수록 경우의 수가 많으며, 0에 가까울 수 록 경우의 수가 한정적입니다. (답변의 다양성)
• temperature : LLM이 결과를 생성할 때, 다음 결과로 나올 수 있는 단어의 무작위성을 정합니다.
• 숫자가 클 수록 창의적이고 무작위로 다음 답변이 생성되며, 0에 가까울 수록 예측 가능한 답변이 생성됩니다.
• max_tokens : 생성할 토큰의 최대치를 설정합니다. (영어 한 단어당 보통 2~3 토큰)

 

1-2-2. LLM에 넣을 데이터를 담을 그릇 만들기

코드

conv_mode = "llava_v1"
conv = conv_templates[conv_mode].copy()

 

• conv_mode : 모델 종류 선택
• conv_templates : 모델 별 대화 탬플릿 모음집
• 즉, conv = conv_templates[... 가 의미하는 바는, 모델 별 대화 탬플릿 모음집으로부터 우리가 쓸 모델의 탬플릿을 복사한다는 것입니다.

 

1-2-3. 이미지 불러오기 & 인코딩

코드

image_data = load_image(str(image))
image_tensor = self.image_processor.preprocess(image_data, return_tensors='pt')['pixel_values'].half().cuda()

• load_image 함수를 통해 이미지를 불러오고,
• image_processor의 preprocess 메소드를 이용해서 이미지 인코딩

 

1-2-4. 언어 데이터를 프롬프트 양식에 담기

코드

inp = DEFAULT_IMAGE_TOKEN + '\n' + prompt
conv.append_message(conv.roles[0], inp)

conv.append_message(conv.roles[1], None)
prompt = conv.get_prompt()

• inp = DEFAULT... → 언어 데이터 앞에 이미지 토큰이 들어올 자리를 명시
• conv.append_message(conv.roles[0]... → 프롬프트 양식 중에 "role"(보통 User)에 언어 데이터 입력
• conv.append_message(conv.roles[1]... → 프롬프트 양식 중에 "role"(보통 Assistant) 파트. AI가 답한 내용이 없으니 None.
• prompt = ... → 양식에 맞춰 넣은 데이터를 다시 프롬프트로 저장

 

1-2-5. LLM에 데이터 입력할 준비

코드

input_ids = tokenizer_image_token(prompt, self.tokenizer, IMAGE_TOKEN_INDEX, return_tensors='pt').unsqueeze(0).cuda()
stop_str = conv.sep if conv.sep_style != SeparatorStyle.TWO else conv.sep2
keywords = [stop_str]
streamer = TextIteratorStreamer(self.tokenizer, skip_prompt=True, timeout=20.0)

• input_ids = ... → 프롬프트, tokenizer, IMAGE_TOKEN_INDEX(이미지가 들어갈 장소를 알려주는 정보)를 넣고 정보 토큰화 (이미지 들어갈 공간은 빼둔 상태)
• stop_str = ... → LLM이 답변을 생성하다가 멈추는 지점을 알려주는 문자열 선택. (LLM이 답변 생성하다가 해당 문자열이 나오면 다음 답변 생성 중단)
• keywords ... → stop_str담은 배열
• streamer = Text... → 답변을 하나씩 출력하기 위한 스트리머 객체(yield처럼)

 

1-2-6. 멀티쓰레드로 LLM 추론 실행

코드

with torch.inference_mode():
      thread = Thread(target=self.model.generate, kwargs=dict(
          inputs=input_ids,
          images=image_tensor,
          do_sample=True,
          temperature=temperature,
          top_p=top_p,
          max_new_tokens=max_tokens,
          streamer=streamer,
          use_cache=True))
      thread.start()

• 모델과 입력 데이터, 파라미터를 넣어 LLM 추론 시작

 

1-2-7. LLM 추론 결과 하나씩 출력

코드

prepend_space = False
for new_text in streamer:
    if new_text == " ":
        prepend_space = True
        continue
    if new_text.endswith(stop_str):
        new_text = new_text[:-len(stop_str)].strip()
        prepend_space = False
    elif prepend_space:
        new_text = " " + new_text
        prepend_space = False
    if len(new_text):
        yield new_text
if prepend_space:
    yield " "
thread.join()

 

streamer를 통해서 나오는 결과를 하나씩 출력

• if new_text == " ": → 공백이 나올 경우 일단 저장 후 다음 출력때 공백 붙여서 출력.(자연스러운 출력을 위한 장치)
• if new_text.endswitch(stop_str): → 모델이 답변을 종료했다는 단어를 출력할 경우, 해당 단어 빼고 출력
• elif prepend_space: → 위에서 저장한 공백을 붙여서 출력
• if len(new_text): → 준비된 텍스트가 있을 경우 일단 출력. Chat GPT 등이 단어 하나씩 출력하는 방법도 이런 종류
• thread.join() → 추론 쓰레드 끝날때 까지 대기

 

 

여기까지가 Predict 코드입니다.

앞으로 역순으로 하나씩 깊게 파고 들어 포스팅 해보도록 하겠습니다.

 

질문 혹은 틀린점 지적은 언제든 환영입니다.