Qwen3-VL-8B-Instruct 파인튜닝 코드

Qwen3-VL-8B-Instruct 파인튜닝 코드를 정리해봤습니다.

 

커스텀 데이터를 사용하는지라, 데이터 로드(load_data 함수) 부분은 각자 데이터에 맞게 변경이 필요합니다.

참고만하시면 됩니다.

 

0. Library 준비

pip install transformers # >= 4.57.0
pip install datasets peft trl qwen_vl_utils
pip install torch # related by Personal CUDA Version(e.g., CUDA 12, 13, ...)
pip install PIL numpy tqdm json random # 필수는 아닐수도있음

 

transformers는 가장 최신의 버전을 추천합니다.

pip install --upgrade pip
pip install --upgrade transformers

 

1. 모델 테스트코드(Qwen3VL_Test.py)

from transformers import Qwen3VLForConditionalGeneration, AutoProcessor

# default: Load the model on the available device(s)
model = Qwen3VLForConditionalGeneration.from_pretrained(
    "Qwen/Qwen3-VL-8B-Instruct", dtype="auto", device_map="auto"
)

processor = AutoProcessor.from_pretrained("Qwen/Qwen3-VL-8B-Instruct")

messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "image",
                "image": "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen-VL/assets/demo.jpeg",
            },
            {"type": "text", "text": "Describe this image."},
        ],
    }
]

# Preparation for inference
inputs = processor.apply_chat_template(
    messages,
    tokenize=True,
    add_generation_prompt=True,
    return_dict=True,
    return_tensors="pt"
)
inputs = inputs.to(model.device)

# Inference: Generation of the output
generated_ids = model.generate(**inputs, max_new_tokens=128)
generated_ids_trimmed = [
    out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
]
output_text = processor.batch_decode(
    generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
)
print(output_text)

 

실행 결과:

>> Loading checkpoint shards: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████| 4/4 [00:05<00:00,  1.30s/it]
>> ['This is a heartwarming and serene photograph capturing a moment between a young woman and her dog on a beach at sunset or sunrise.\n\n**Scene Description:**\n\n*   **Main Subjects:** A young woman and a large, light-colored Labrador Retriever are the central focus. They are sitting on the sand, facing each other.\n*   **The Woman:** She has long, dark hair and is smiling warmly, looking down at her dog. She is dressed casually in a black-and-white plaid shirt and dark pants. She is sitting cross-legged in the sand, with one hand extended towards the dog, and the other hand holding']

 

2. FineTuning.py

저는 커스텀 데이터를 사용하여 json 포맷의 데이터시트를 로드하고, 데이터를 Train/Val Set으로 분리했습니다.

본인이 원하는 형태를 붙이면됩니다.

import torch
from transformers import Qwen3VLForConditionalGeneration, AutoTokenizer, AutoProcessor, BitsAndBytesConfig, Qwen3VLProcessor
from qwen_vl_utils import process_vision_info

from peft import LoraConfig, get_peft_model
from trl import SFTConfig, SFTTrainer
from datasets import concatenate_datasets, Dataset, load_dataset, load_from_disk
from tqdm import tqdm
import os
import base64
from PIL import Image
from io import BytesIO
import json
import pyarrow as pa
import random
import re
import numpy as np

# 시스템 메시지 (모델이 임상 문맥을 따르도록 안내)
SYSTEM_PROMPT = """
                [시스템 프롬프트]
                페르소나와 역할을 지정합니다.
                """            
MODEL_ID = "Qwen/Qwen3-VL-8B-Instruct"

USER_PROMPT = """
              [입력 프롬프트]
              QA형식에서 Q를 입력합니다.
              고정된 지시문이 아니여도 괜찮습니다.
              """

def format_data(sample):
    """
    단일 샘플을 모델 입력 메시지 구조로 포맷합니다.
    기존처럼 이미지 로딩 및 텍스트 프롬프트 생성도 포함합니다.
    """
    assistant_text = sample.get("ground_truth", "No gt")

    # 메시지 구조를 프롬프트와 함께 맞춤
    return [
        {
            "role": "system",
            "content": [{"type": "text", "text": SYSTEM_PROMPT}],
        },
        {
            "role": "user",
            "content": [
                {"type": "image", "image": sample["IMG"]},
                {"type": "text", "text": USER_PROMPT},
            ],
        },
        {
            "role": "assistant",
            "content": [{"type": "text", "text": assistant_text}],
        },
    ]

def load_data(PATH = 'your_data.json', split='train'):
    with open(PATH, 'r') as f:
        data = json.load(f)

    # split에 맞는 샘플만 필터링
    data = [sample for sample in data if sample["SPLIT"] == split]

    for sample in tqdm(data, desc="open image"):        
        sample["IMG"] = Image.open(sample["IMAGE_PATH"]).convert('RGB').resize((384,384))

    return data

def collate_fn(examples):
    """
    Data collator to prepare a batch of examples.

    This function applies the chat template to texts, processes the images,
    tokenizes the inputs, and creates labels with proper masking.
    """
    # Apply chat template to each example (no tokenization here)
    texts = [processor.apply_chat_template(example, tokenize=False) for example in examples]
    # Process visual inputs for each example
    image_inputs = [process_vision_info(example)[0] for example in examples]

    # Tokenize texts and images into tensors with padding
    batch = processor(
        text=texts,
        images=image_inputs,
        return_tensors="pt",
        padding=True,
    )

    # Create labels by cloning input_ids and mask the pad tokens
    labels = batch["input_ids"].clone()
    labels[labels == processor.tokenizer.pad_token_id] = -100

    # Determine image token IDs to mask in the labels (model specific)
    if isinstance(processor, Qwen3VLProcessor): #Qwen 버전에 맞춰 VLProcessor는 변할 수 있습니다.
        image_tokens = [151652, 151653, 151655]
    else:
        image_tokens = [processor.tokenizer.convert_tokens_to_ids(processor.image_token)]

    # Mask image token IDs in the labels
    for image_token_id in image_tokens:
        labels[labels == image_token_id] = -100

    batch["labels"] = labels
    return batch

def main():
    # BitsAndBytesConfig: 4비트 양자화 설정
    bnb_config = BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_quant_type="nf4",
        bnb_4bit_use_double_quant=True,
        bnb_4bit_compute_dtype=torch.bfloat16
    )

    # 모델과 processor 로드
    model = Qwen3VLForConditionalGeneration.from_pretrained(
        MODEL_ID, device_map="auto", dtype="auto", quantization_config=bnb_config
    )
    global processor
    processor = AutoProcessor.from_pretrained(MODEL_ID)

    print(processor.tokenizer)

    # LoRA 설정: 본인 데이터에 맞게 Grid Search가 필요합니다. (기본값은 r:8, alpha:16 추천합니다.)
    peft_config = LoraConfig(
        r=16,				# r 값은 본인의 GPU VRAM 상황과 Train Params 세팅에 따라 달라집니다.
        lora_alpha=192,		# alpha 값은 본인의 데이터 갯수에 따라 달라집니다.
        lora_dropout=0.05,
        bias="none",
        target_modules="all-linear",	# 모든 리니어 레이어 뒤에 어댑터를 붙입니다.
        # target_modules=[				# 원하는 타겟 모듈에 붙일 수 있습니다.
        #     "q_proj", "k_proj", "v_proj", "o_proj",
        #     "gate_proj", "up_proj", "down_proj"
        # ],
        task_type="CAUSAL_LM",
    )

    model = get_peft_model(model, peft_config)
    model.print_trainable_parameters()

    # 데이터 로드 및 포매팅
    train_dataset = load_data(split="train")
    train_dataset = [format_data(sample) for sample in train_dataset]
    eval_dataset = load_data(split="val")
    eval_dataset = [format_data(sample) for sample in eval_dataset]

    print("Train sample:", train_dataset[0])
    print("Eval sample:", eval_dataset[0])

    num_train_epochs = 3 # 데이터 갯수에 따라 조절할 수 있습니다. 1~3 epoch 수준에서 추천합니다.
    learning_rate = 2e-4 # 데이터 갯수에 따라 조절할 수 있습니다. Grid Search가 필요합니다.

    # SFT (Supervised Fine-Tuning) 설정
    training_args = SFTConfig(
        output_dir="./train/lora-adapter-r16_a192",
        num_train_epochs=num_train_epochs,
        per_device_train_batch_size=4,	# VRAM에 따라 세팅을 바꿀 수 있습니다.
        per_device_eval_batch_size=4,	# VRAM에 따라 세팅을 바꿀 수 있습니다.
        gradient_accumulation_steps=4,	# VRAM에 따라 세팅을 바꿀 수 있습니다.
        gradient_checkpointing=True,
        optim="adamw_torch_fused",
        logging_steps=10,
        save_strategy="epoch",
        eval_strategy="epoch",
        learning_rate=learning_rate,
        bf16=True,
        max_grad_norm=0.3,
        warmup_ratio=0.03,
        lr_scheduler_type="linear",
        report_to="wandb",                
        gradient_checkpointing_kwargs={"use_reentrant": False},
        dataset_kwargs={"skip_prepare_dataset": True},
        remove_unused_columns=False,
        label_names=["labels"],
    )
    training_args.remove_unused_columns = False

    trainer = SFTTrainer(
        model=model,
        args=training_args,
        train_dataset=train_dataset,
        eval_dataset=eval_dataset,  # 필요 시 eval 데이터도 함께 지정합니다
        data_collator=collate_fn,
        peft_config=peft_config,        
    )

    trainer.train()
    model.save_pretrained(training_args.output_dir, max_shard_size="4GB")
    processor.tokenizer.save_pretrained(training_args.output_dir)
    processor.save_pretrained(training_args.output_dir)

if __name__ == "__main__":
    main()

 

728x90

3. Inference.py

학습한 LoRA 어댑터를 로드하여, Qwen 모델에 붙인 뒤 인퍼런스를 수행할 수 있는 샘플 코드입니다.

import torch
import json
import os
import numpy as np
from PIL import Image
from transformers import AutoProcessor, AutoModelForImageTextToText, BitsAndBytesConfig
from peft import PeftModel
from typing import List, Dict, Any
import pandas as pd
from tqdm import tqdm
import argparse
import re
from collections import Counter
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.metrics import confusion_matrix
from FineTuning import load_data

# ==================== Qwen 인퍼런스 클래스 ====================
class QwenInference:
    def __init__(self, base_model_id: str, adapter_path: str):
        self.base_model_id = base_model_id
        self.adapter_path = adapter_path
        self.model = None
        self.processor = None
        self.dtype = None
        
        self.SYSTEM_PROMPT = """
                시스템 프롬프트 / 학습과 동일하게 설정합니다.
            """
        self.USER_PROMPT = """
              	유저 프롬프트 / 학습과 동일하게 설정합니다.
              """

    def load_model(self):
        print("Qwen base 모델과 adapter(LoRA) 불러오는 중...")
        # GPU bfloat16 지원 체크
        if torch.cuda.get_device_capability()[0] < 8:
            print("Warning: GPU does not support bfloat16, using float16 instead.")
            self.dtype = torch.float16
        else:
            self.dtype = torch.bfloat16

        model_kwargs = dict(
            dtype=self.dtype,
            device_map="cuda:0",
        )
        # base model 로드
        base_model = AutoModelForImageTextToText.from_pretrained(self.base_model_id, **model_kwargs)
        self.processor = AutoProcessor.from_pretrained(self.base_model_id)
        # adapter(LoRA) weight 로드 및 merge
        print("Adapter(LoRA) weight를 base model에 merge합니다...")
        model_with_adapter = PeftModel.from_pretrained(base_model, self.adapter_path)
        self.model = model_with_adapter.merge_and_unload()
        # self.model = base_model

        print("모델 로딩 및 병합 완료!")
        # 오른쪽 패딩 사용
        self.processor.tokenizer.padding_side = "right"

    def generate_report(self, image_path: str, max_new_tokens: int = 512, temperature: float = 0.1):        
    	# 포매팅을 진행한 뒤, 이미지를 입력해 텍스트를 생성합니다.
        try:
            image = image_path
            messages = [
                {
                    "role": "system",
                    "content": [
                        {"type": "text", "text": self.SYSTEM_PROMPT}
                    ]
                },
                {
                    "role": "user",
                    "content": [
                        {"type": "image"},
                        {"type": "text", "text": self.USER_PROMPT}
                    ]
                }
            ]
            prompt_text = self.processor.apply_chat_template(
                messages,
                add_generation_prompt=True,
                tokenize=False
            ).strip()
            inputs = self.processor(
                text=[prompt_text],
                images=[[image]],
                return_tensors="pt",
                padding=True,
            )
            for k, v in inputs.items():
                if isinstance(v, torch.Tensor):
                    inputs[k] = v.to(self.model.device)
            with torch.inference_mode():
                out = self.model.generate(
                    **inputs,
                    disable_compile=True,
                    max_new_tokens=max_new_tokens,
                    do_sample=True,
                    temperature=temperature,
                    pad_token_id=self.processor.tokenizer.eos_token_id,
                    repetition_penalty=1.2,
                    no_repeat_ngram_size=4
                )
                generated = out[0][inputs["input_ids"].shape[-1]:]
            return self.processor.decode(generated, skip_special_tokens=True).strip()
        except Exception as e:
            print(f"Error processing {image_path}: {str(e)}")
            return f"Error: {str(e)}"

    def run_inference(self, dataset: List[Dict], output_dir: str = "./results", max_samples: int = None):
        os.makedirs(output_dir, exist_ok=True)
        print('output folder: ', output_dir)
        
        results = []
        print("Starting inference on dataset...")
        for i, sample in enumerate(tqdm(dataset, desc="Generating")):
            gt = sample.get("grount_truth", "No gt")
            image = sample["IMG"]
            generated = self.generate_report(image)
            result = {
                'Image ID': sample.get("Image ID"),
                'ground_truth': gt,
                'generated_report': generated
            }
            results.append(result)
        print("\nInference completed!")
        self.save_and_analyze_results(results, output_dir)
        return results

	# 생성한 결과를 csv / json으로 저장합니다.
    def save_and_analyze_results(self, results: List[Dict], output_dir: str):
        results_df = pd.DataFrame(results)
        output_csv = os.path.join(output_dir, "inference_results.csv")
        results_df.to_csv(output_csv, index=False, encoding='utf-8')
        print(f"Results saved to {output_csv}")
        output_json = os.path.join(output_dir, "inference_results.json")
        with open(output_json, 'w', encoding='utf-8') as f:
            json.dump(results, f, ensure_ascii=False, indent=2)
        print(f"Results also saved to {output_json}")

    def cleanup(self):
        if self.model:
            del self.model
        torch.cuda.empty_cache()
        print("Memory cleaned up!")

def main():
    parser = argparse.ArgumentParser(description="Qwen LoRA Adapter Inference")
    parser.add_argument("--base_model", type=str, default="Qwen/Qwen3-VL-8B-Instruct",
                        help="Base model ID")
    parser.add_argument("--adapter_path", type=str, default="./train/lora-adapter-r16_a192/checkpoint-201",
                        help="Path to fine-tuned adapter (LoRA), checkpoint 경로를 변경할 수 있습니다.")
    parser.add_argument("--split", type=str, default="test", choices=["val", "test"],
                        help="Which split to use (val/test)")
    parser.add_argument("--output_dir", type=str, default='./inference_results/lora-adapter-r16_a192/checkpoint-201',
                        help="Output directory for results")
    parser.add_argument("--max_samples", type=int, default=None,
                        help="Maximum number of samples to run inference on")
    parser.add_argument("--max_tokens", type=int, default=512,
                        help="Maximum number of tokens to generate")
    parser.add_argument("--temperature", type=float, default=0.01,
                        help="Generation temperature")
    args = parser.parse_args()

    # adapter_path에서 폴더명만 추출하여 결과 디렉토리명에 반영
    import re
    adapter_name = re.sub(r'[\\/]', '_', args.adapter_path.strip('./'))
    if args.output_dir is None:
        output_dir = f"./inference_results_{adapter_name}"
    else:
        output_dir = args.output_dir

    # 데이터셋 로드
    print(f"Loading dataset split: {args.split}")
    dataset = load_data(split=args.split)
    print(f"Total samples available in {args.split}: {len(dataset)}")

    # QwenInference 클래스로 객체를 생성한 뒤, 인퍼런스를 수행합니다.
    inference = QwenInference(args.base_model, args.adapter_path)
    try:
        inference.load_model()
        print("\n" + "="*50)
        print(f"BATCH INFERENCE ({args.split.upper()} SET)")
        print("="*50)
        results = inference.run_inference(dataset, output_dir, args.max_samples)
        # results를 DataFrame으로 변환하고 display
        results_df = pd.DataFrame(results)
        print("\n===== Inference Results (DataFrame) =====")
        # 결과 DataFrame을 엑셀 파일로 저장합니다.
        output_excel = os.path.join(output_dir, "inference_results.xlsx")
        results_df.to_excel(output_excel, index=False)
        print(f"Results also saved to {output_excel}")
    finally:
        inference.cleanup()

if __name__ == "__main__":
    main()

 

마치며

Qwen3-VL-8B 모델의 경우 오픈소스 VLM에서 상당히 좋은 성능을 보여주는 모델입니다.

VLM 파인튜닝이 필요한 경우, Unsloth와 같은 라이브러리를 이용할 수 있지만, Python 코드로도 이렇게 진행할 수 있습니다.

참고하시면 좋겠습니다 :)