1) Unsloth
Unsloth는 오픈소스 경량화 모델 최적화 라이브러리로,
Meta의 Llama 3, Google의 Gemma, Microsoft의 Phi-4, Mistral 등의 최신 모델을
더 빠르고 적은 VRAM으로 파인튜닝할 수 있도록 설계된 툴킷입니다.
특히 Hugging Face의 transformers, trl 기반과 호환되며,
4-bit 및 8-bit QLoRA를 활용해 최소한의 메모리로도 고성능 학습이 가능한 것이 장점입니다.
Unsloth의 주요 특징
- 기존 방식 대비 2배 빠른 파인튜닝 속도
- VRAM 사용량 70~80% 절감 가능 (4-bit QLoRA)
- 4-bit 및 8-bit Quantization 지원
- Llama, Qwen, Gemma, Mistral 등 다양한 모델 지원
- GGUF, Ollama, vLLM 등 다양한 포맷 변환 가능
- Hugging Face 및 LangChain과 호환
- Colab에서도 무료로 실행 가능 (로컬 및 저사양 환경도 가능)
2) 기본 챗봇 구축
1) 모델 로드 및 기본 설정
Hugging Face에서 제공하는 MLP-KTLim/llama-3-Korean-Bllossom-8B 모델
- 한국어 특화 모델
- LLaMA-3 기반으로 한국어에 특화된 모델이며, 다양한 한국어 질의에 자연스럽게 응답이 가능합니다.
- Llama3대비 대략 25% 더 긴 길이의 한국어 Context 처리가능
- 한국어-영어 Pararell Corpus를 활용한 한국어-영어 지식연결 (사전학습)
- 한국어 문화, 언어를 고려해 언어학자가 제작한 데이터를 활용한 미세조정
2) 기본 챗봇 응답 생성 과정
파인튜닝 전에 사전 학습된 모델의 기본 성능을 확인합니다.
사용자 질문에 대한 응답을 generate()로 직접 생성해보겠습니다.
model.generate : 모델이 응답을 생성할 때 품질을 조절하는 다양한 하이퍼파라미터를 설정합니다.
주요 파라미터
- temperature : 예측값에 무작위성 부여, 낮을수록 일관적인 응답, 높을수록 창의적 응답. (기본: 0.5, 무작위성 없는 상태))
- top_p : 누적 확률 기준으로 상위 토큰만 샘플링. 낮을수록 안정적, 높을수록 다양함.
- repetition_penalty : 같은 문장이나 단어 반복을 억제. 자연스러운 응답 유도.
- do_sample : 샘플링 방식을 적용하여 다양한 답변 생성 가능 (True일 경우 랜덤성 부여).
- max_new_tokens : 생성되는 응답의 최대 토큰 수를 제한 (응답 길이 제어).
user_input = "선생님은 무엇인가요?"
# ✅ 입력을 모델이 처리할 수 있는 텐서로 변환
inputs = tokenizer(user_input, return_tensors="pt").to("cuda")
print(f"📝 질문: {user_input}")
# ✅ 모델 응답 생성 (추론 옵션 적용)
outputs = model.generate(
**inputs,
max_new_tokens=100, # 응답 최대 길이 제한
temperature=0.3, # 온도 조절을 통해 일관된 응답 유도
top_p=0.1, # 상위 확률 토큰만 샘플링
repetition_penalty=1.2, # 반복 방지
do_sample=True # 샘플링 방식 적용
)
# ✅ 응답 디코딩
decoded_output = tokenizer.decode(outputs[0], skip_special_tokens=True)
print("🤖 응답:", decoded_output)
3) LoRA(QLoRA)를 활용한 모델 경량 미세튜닝
(1) 한국어 데이터셋 준비
이번 실습에서는 한국어 법률 정보 챗봇을 구축하기 위해, Hugging Face에서 제공하는 jihye-moon/LawQA-Ko 데이터셋을 사용합니다.
from datasets import load_dataset
# jihye-moon/LawQA-Ko 데이터셋을 train split 기준으로 불러옵니다.
dataset = load_dataset("jihye-moon/LawQA-Ko", split="train")
dataset
Dataset({
features: ['question', 'precedent', 'answer'],
num_rows: 14819
})
파인튜닝을 위해서는 아래와 같은 Alpaca 포맷으로 변경해야 합니다:
{
"instruction": "질문 내용",
"input": "", // 선택사항 (없으면 빈 문자열)
"output": "답변 내용",
"text": "Alpaca-style 데이터 포맷으로 구성된 최종 학습 문장"
}
# ✅ Step 1: instruction / output 필드 생성
def to_alpaca_format(example):
return {
"instruction": example["question"],
"output": example["answer"]
}
dataset = dataset.map(to_alpaca_format) #instruction 과 output을 붙인다.
'question', 'precedent', 'answer'
+ 'instruction', 'output'
길이 필터링
# ✅ Step 2: output 길이 필터링 (토큰 길이 516 이하)
def is_output_short(example):
tokenized = tokenizer(example["output"], truncation=False, add_special_tokens=False)
return len(tokenized["input_ids"]) <= 516
filtered_dataset = dataset.filter(is_output_short)
print(len(dataset), len(filtered_dataset))
14819 10535
프롬프트에 해당하는 text 필드 생성
alpaca_prompt = """Below is an instruction that describes a task. Write a response that appropriately completes the request.
### Instruction:
{}
### Response:
{}"""
text = alpaca_prompt.format(instruction, output) + EOS_TOKEN
.format은 다음과 같습니다. f "....{instruction} .....{output}...."
# ✅ Step 3: text 필드 생성
def formatting_prompts_func(examples):
instructions = examples["instruction"]
outputs = examples["output"]
texts = []
for instruction, output in zip(instructions, outputs):
text = alpaca_prompt.format(instruction, output) + EOS_TOKEN
texts.append(text)
return { "text": texts }
formatted_dataset = filtered_dataset.map(formatting_prompts_func, batched=True)
# ✅ 결과 확인
print(f"📊 필터링 후 샘플 수: {len(formatted_dataset)}")
print(formatted_dataset[0])
(2) LoRA 설정 및 모델 준비(QLoRA)
QLoRA는 기존 LoRA에 양자화를 추가한 것이다. 양자화란 파라미터(weight)의 자료형 크기를 바꾸는 것이다. 무거운 모델을 양자화해서, 로드하게 해준다.
Unsloth 활용
- FastLanguageModel : Unsloth의 핵심 클래스. 빠른 로딩 및 LoRA 적용 지원
- load_in_4bit=True : GPU 메모리 절약을 위한 4bit 양자화 적용 -> QLoRA
- LoRA Adapter : 파인튜닝 시 일부 레이어만 학습하여 효율 극대화
LoRA
- 원본 모델 전체를 FP16(16-bit 부동소수점)으로 유지
QLoRA - Quantized LoRA
- 모델 가중치를 4-bit로 양자화하여 로드
- 특정 연산(예: 커스텀 CUDA 커널 지원)이 필요하므로, bitsandbytes 등 추가 라이브러리가 있어야 함
from unsloth import FastLanguageModel
from transformers import TextStreamer
import torch
model, tokenizer = FastLanguageModel.from_pretrained(
model_name = "MLP-KTLim/llama-3-Korean-Bllossom-8B", # ✅ 사용 모델
max_seq_length = 2048, # ✅ 권장 설정
dtype = None, # ✅ float16 or bfloat16 자동 선택
load_in_4bit = True # ✅ 4bit QLoRA 양자화 로딩 활성화
)
# 이미 로드한 모델과 토크나이저에 LoRA 설정만 적용
model = FastLanguageModel.get_peft_model(
model,
r = 16, # LoRA 랭크 설정. 8, 16, 32, 64, 128 권장. r 값이 클수록 모델이 더 많은 정보를 학습할 수 있지만, 너무 크면 메모리를 많이 사용
target_modules = ["q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj"], # PEFT 적용할 모듈 목록. 모델의 특정 부분(모듈)에만 학습
lora_alpha = 16, # LoRA 알파 설정 LoRA라는 기술이 얼마나 강하게 작용할지 조절
lora_dropout = 0, # LoRA 드롭아웃 설정. 0으로 최적화
bias = "none", # 바이어스 설정. "none"으로 최적화
# "unsloth" 사용 시 VRAM 절약 및 배치 사이즈 2배 증가
# 학습할 때 메모리를 절약하는 방법을 사용하는 설정
use_gradient_checkpointing = "unsloth", # 매우 긴 컨텍스트를 위해 "unsloth" 설정
random_state = 42, # 랜덤 시드 설정
use_rslora = False, # 랭크 안정화 LoRA 사용 여부
loftq_config = None, # LoftQ 설정 (사용하지 않음)
)
FastLanguageModel.get_peft_model
- r : LoRA의 랭크 값. 클수록 성능 좋지만 메모리 사용 증가 (보통 8~64)
- lora_alpha : 학습 속도와 안정성 제어하는 스케일링 계수
- lora_dropout : LoRA 레이어에만 적용되는 드롭아웃 확률 (0.05 권장)
- bias : "none"으로 설정 시 LoRA matrix만 학습 (보통 이 값 사용)
- target_modules : LoRA를 적용할 Linear 계층 명시 (q_proj, v_proj 등)
- use_gradient_checkpointing : 메모리 절약 위한 역전파 시 체크포인팅 사용
- use_rslora : Rescaled LoRA 사용 여부 (False로 설정)
- loftq_config : optional, QLoRA에 LoFTQ 기법 적용할 경우 지정
(3) LoRA(QLoRA)를 이용한 모델 파인튜닝
SFTTrainer를 활용하여 실제로 LoRA(QLoRA) 기반 파인튜닝을 수행합니다.
주요 학습 인자 설명
- per_device_train_batch_size: 각 디바이스(GPU) 당 학습 배치 크기
- gradient_accumulation_steps: 메모리 제한 시 그레이디언트 누적 단계 수
- max_steps: 전체 학습 반복 횟수 (데모는 100, 실전은 수천 이상 설정)
- learning_rate: 파인튜닝 시 모델 가중치 업데이트 속도
- fp16, bf16: GPU 환경에 맞춰 혼합 정밀도 학습 설정
- optim: 양자화된 모델에 최적화된 옵티마이저 사용 (adamw_8bit)
- output_dir: 학습 결과(모델) 저장 경로
- packing: 여러 샘플을 하나로 패킹할지 여부 (False 권장)
from trl import SFTTrainer
from transformers import TrainingArguments
from unsloth import is_bfloat16_supported # BFloat16 지원 여부 확인 함수 임포트
max_seq_length = 1024
# SFTTrainer를 사용하여 모델 학습 설정
# SFTTrainer 인스턴스 생성
trainer = SFTTrainer(
model = model, # 학습할 모델
tokenizer = tokenizer, # 사용할 토크나이저
train_dataset = formatted_dataset, # 학습할 데이터셋 ★★★★★★★★
dataset_text_field = "text", # 데이터셋의 텍스트 필드 이름 ★★★★★★★★
max_seq_length = max_seq_length, # 최대 시퀀스 길이
dataset_num_proc = 2, # 데이터셋 전처리에 사용할 프로세스 수 cpu
packing = False, # 짧은 시퀀스의 경우 packing을 비활성화 (학습 속도 5배 향상 가능)
args = TrainingArguments(
per_device_train_batch_size = 2, # 디바이스 당 배치 사이즈
gradient_accumulation_steps = 4, # 그래디언트 누적 단계 수
warmup_steps = 5, # 워밍업 스텝 수
# num_train_epochs = 1, # 전체 학습 에폭 수 설정 가능
max_steps = 60, # 최대 학습 스텝 수
learning_rate = 2e-4, # 학습률
fp16 = not is_bfloat16_supported(), # BFloat16 지원 여부에 따라 FP16 사용
bf16 = is_bfloat16_supported(), # BFloat16 사용 여부
logging_steps = 1, # 로깅 빈도
optim = "adamw_8bit", # 옵티마이저 설정 (8비트 AdamW)
weight_decay = 0.01, # 가중치 감쇠
lr_scheduler_type = "linear", # 학습률 스케줄러 타입
seed = 3407, # 랜덤 시드 설정
output_dir = "outputs", # 출력 디렉토리
report_to = "none",
),
)
trainer_stats = trainer.train()
학습된 모델 저장
# 모델 저장 로컬폴더에다가 저장하는 방식
model.save_pretrained("Qlora_model") # Local saving
tokenizer.save_pretrained("Qlora_model")
# 이 이외에 허깅페이스나 다른 hub에 push해서 저장하는 방법이 있음
# 다만, 업로드 속도와 다운로드 속도를 고려해야함.
(4) 파인튜닝된 모델 평가 (Inference)
학습된모델 불러오기
from unsloth import FastLanguageModel
import torch
# 저장된 경로 지정
save_directory = "Qlora_model"
# 모델과 토크나이저 불러오기
model, tokenizer = FastLanguageModel.from_pretrained(
model_name = save_directory,
max_seq_length = 2048,
dtype = None,
load_in_4bit = True, # 양자화 옵션을 동일하게 설정
)
추론 해보기
# 추론해보기
FastLanguageModel.for_inference(model) # 네이티브 2배 빠른 추론 활성화
# 추론을 위한 입력 준비
inputs = tokenizer(
[
alpaca_prompt.format(
"최저임금제도란 무엇이며, 최저임금액은 어떻게 결정되는지요?", # 인스트럭션 (명령어)
"", # 출력 - 생성할 답변을 비워둠
)
], return_tensors = "pt").to("cuda") # 텐서를 PyTorch 형식으로 변환하고 GPU로 이동
text_streamer = TextStreamer(tokenizer) # 토크나이저를 사용하여 스트리머 초기화
# 모델을 사용하여 텍스트 생성 및 스트리밍 출력
_ = model.generate(**inputs, streamer = text_streamer, max_new_tokens = 128) # 최대 128개의 새로운 토큰 생성
각 모델의 성능을 정량적으로 비교하기 위해 챗봇 평가 매트릭스(BLEU, ROUGE, Distinct-N, GPT Score 등)를 통해 평가를 진행하겠습니다.
1️⃣ BLEU (Bilingual Evaluation Understudy)
응답이 정답과 얼마나 일치하는가? Precision과 유사
번역 품질 평가 지표지만, 챗봇 응답이 정답과 얼마나 일치하는지 측정하는 데 활용 가능.
단어 n-gram의 일치율을 기반으로 평가 (BLEU-1, BLEU-2 등).
2️⃣ ROUGE (Recall-Oriented Understudy for Gisting Evaluation)
실제 정답과 비교했을 때, 정답 토큰이 얼마나 있는가? (Recall)
문서 요약 평가에 자주 쓰이지만, 챗봇 응답의 요약 품질을 평가하는 데도 활용 가능.
ROUGE-1 (단어 기반, '나는', '오늘'), ROUGE-2 (2-gram 기반, '나는-오늘'), ROUGE-L (Longest Common Subsequence 기반) 사용 가능.
3️⃣ Distinct-N Score
생성된 응답의 다양성을 평가하는 지표.
Distinct-1 (고유한 단어 비율), Distinct-2 (고유한 2-gram 비율)를 측정하여 반복적인 응답을 방지하는 능력을 확인.
4️⃣ GPT-Score / BERTScore
GPT-4 또는 BERT 기반 평가 모델을 활용하여 챗봇 응답의 자연스러움과 의미적 유사도를 비교.
!pip install rouge
from nltk.translate.bleu_score import sentence_bleu
from rouge import Rouge
import nltk
nltk.download('punkt')
eval_size = 10
bleu1_list, bleu2_list = [], []
rouge1_list, rouge2_list, rougel_list = [], [], []
distinct1_list, distinct2_list = [], []
rouge = Rouge()
for i in range(eval_size):
reference = formatted_dataset[i]["output"].strip()
instruction = formatted_dataset[i]["instruction"].strip()
prompt = f"""Below is an instruction that describes a task. Write a response that appropriately completes the request.
### Instruction:
{instruction}
### Response:
"""
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
outputs = model.generate(
**inputs,
max_new_tokens=512,
do_sample=True,
temperature=0.7,
top_p=0.95,
eos_token_id=tokenizer.eos_token_id,
)
decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
if "### Response:" in decoded:
generated = decoded.split("### Response:")[-1].strip()
else:
generated = decoded.strip()
try:
bleu1 = sentence_bleu([reference.split()], generated.split(), weights=(1, 0, 0, 0))
bleu2 = sentence_bleu([reference.split()], generated.split(), weights=(0.5, 0.5, 0, 0))
rouge_scores = rouge.get_scores(generated, reference)[0]
tokens = generated.split()
bigrams = list(zip(tokens, tokens[1:]))
distinct1 = len(set(tokens)) / max(len(tokens), 1)
distinct2 = len(set(bigrams)) / max(len(bigrams), 1)
bleu1_list.append(bleu1)
bleu2_list.append(bleu2)
rouge1_list.append(rouge_scores["rouge-1"]["f"])
rouge2_list.append(rouge_scores["rouge-2"]["f"])
rougel_list.append(rouge_scores["rouge-l"]["f"])
distinct1_list.append(distinct1)
distinct2_list.append(distinct2)
except:
print(f"⚠️ {i}번째 샘플 평가 실패")
print("📊 [1000개 샘플 평균 평가 결과]")
print(f"BLEU-1 평균: {sum(bleu1_list)/len(bleu1_list):.4f}")
print(f"BLEU-2 평균: {sum(bleu2_list)/len(bleu2_list):.4f}")
print('-------------------------')
print(f"ROUGE-1 평균: {sum(rouge1_list)/len(rouge1_list):.4f}")
print(f"ROUGE-2 평균: {sum(rouge2_list)/len(rouge2_list):.4f}")
print(f"ROUGE-L 평균: {sum(rougel_list)/len(rougel_list):.4f}")
print('-------------------------')
print(f"Distinct-1 평균: {sum(distinct1_list)/len(distinct1_list):.4f}")
print(f"Distinct-2 평균: {sum(distinct2_list)/len(distinct2_list):.4f}")
📊 [1000개 샘플 평균 평가 결과]
BLEU-1 평균: 0.0520
BLEU-2 평균: 0.0255
-------------------------
ROUGE-1 평균: 0.0939
ROUGE-2 평균: 0.0235
ROUGE-L 평균: 0.0871
-------------------------
Distinct-1 평균: 0.7418
Distinct-2 평균: 0.8750
(5) DoRA: Weight-Decomposed LoRA
DoRA(Weight-Decomposed LoRA)는 기존 LoRA의 성능 한계를 극복하기 위해 제안된 새로운 PEFT(파라미터 효율적 미세조정) 방식입니다.
LoRA는 간단하고 효율적이지만, 일부 경우 정확도가 감소하거나 훈련이 불안정해질 수 있습니다.
DoRA는 이러한 한계를 해결하면서 더 적은 파라미터로도 더 높은 성능을 보일 수 있습니다.
런타임 - 세션 다시 시작 후 실
from unsloth import FastLanguageModel
import torch
model, tokenizer = FastLanguageModel.from_pretrained(
model_name = "MLP-KTLim/llama-3-Korean-Bllossom-8B", # ✅ 사용 모델
max_seq_length = 2048, # ✅ 권장 설정
dtype = None, # ✅ float16 or bfloat16 자동 선택
load_in_4bit = True # ✅ 4bit QLoRA 양자화 로딩 비활성화
)
model = FastLanguageModel.get_peft_model(
model,
r = 16,
lora_alpha = 16,
lora_dropout = 0.0,
bias = "none",
target_modules = ["q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj"],
use_gradient_checkpointing = "unsloth",
use_dora = True, # ✅ DoRA 활성화
)
데이터셋 로드
from datasets import load_dataset
# ✅ 데이터셋 로드
dataset = load_dataset("jihye-moon/LawQA-Ko", split="train")
# ✅ EOS 토큰
EOS_TOKEN = tokenizer.eos_token
# ✅ Alpaca-style 프롬프트 템플릿
alpaca_prompt = """Below is an instruction that describes a task. Write a response that appropriately completes the request.
### Instruction:
{}
### Response:
{}"""
# ✅ Step 1: instruction / output 필드 생성
def to_alpaca_format(example):
return {
"instruction": example["question"],
"output": example["answer"]
}
dataset = dataset.map(to_alpaca_format)
# ✅ Step 2: output 길이 필터링 (토큰 길이 516 이하)
def is_output_short(example):
tokenized = tokenizer(example["output"], truncation=False, add_special_tokens=False)
return len(tokenized["input_ids"]) <= 516
filtered_dataset = dataset.filter(is_output_short)
# ✅ Step 3: text 필드 생성
def formatting_prompts_func(examples):
instructions = examples["instruction"]
outputs = examples["output"]
texts = []
for instruction, output in zip(instructions, outputs):
text = alpaca_prompt.format(instruction, output) + EOS_TOKEN
texts.append(text)
return { "text": texts }
formatted_dataset = filtered_dataset.map(formatting_prompts_func, batched=True)
# ✅ 결과 확인
print(f"📊 필터링 후 샘플 수: {len(formatted_dataset)}")
print(formatted_dataset[0])
from trl import SFTTrainer
from transformers import TrainingArguments
from unsloth import is_bfloat16_supported # BFloat16 지원 여부 확인 함수 임포트
# SFTTrainer를 사용하여 모델 학습 설정
# SFTTrainer 인스턴스 생성
trainer = SFTTrainer(
model = model, # 학습할 모델
tokenizer = tokenizer, # 사용할 토크나이저
train_dataset = formatted_dataset, # 학습할 데이터셋 ★★★★★★★★
dataset_text_field = "text", # 데이터셋의 텍스트 필드 이름 ★★★★★★★★
max_seq_length = 1024, # 최대 시퀀스 길이
dataset_num_proc = 2, # 데이터셋 전처리에 사용할 프로세스 수 cpu
packing = False, # 짧은 시퀀스의 경우 packing을 비활성화 (학습 속도 5배 향상 가능)
args = TrainingArguments(
per_device_train_batch_size = 2, # 디바이스 당 배치 사이즈
gradient_accumulation_steps = 4, # 그래디언트 누적 단계 수
warmup_steps = 5, # 워밍업 스텝 수
# num_train_epochs = 1, # 전체 학습 에폭 수 설정 가능
max_steps = 60, # 최대 학습 스텝 수
learning_rate = 2e-4, # 학습률
fp16 = not is_bfloat16_supported(), # BFloat16 지원 여부에 따라 FP16 사용
bf16 = is_bfloat16_supported(), # BFloat16 사용 여부
logging_steps = 1, # 로깅 빈도
optim = "adamw_8bit", # 옵티마이저 설정 (8비트 AdamW)
weight_decay = 0.01, # 가중치 감쇠
lr_scheduler_type = "linear", # 학습률 스케줄러 타입
seed = 3407, # 랜덤 시드 설정
output_dir = "outputs", # 출력 디렉토리
report_to = "none",
),
)
학습 실행
trainer_stats = trainer.train()
# 모델 저장 로컬폴더에다가 저장하는 방식
model.save_pretrained("Dora_model") # Local saving
tokenizer.save_pretrained("Dora_model")
# 이 이외에 허깅페이스나 다른 hub에 push해서 저장하는 방법이 있음
# 다만, 업로드 속도와 다운로드 속도를 고려해야함.
(6) DoRA로 파인튜닝된 DoRA 모델 평가 (Inference)
'런타임 - 세션 다시 시작' 후 실행
from unsloth import FastLanguageModel
import torch
# 저장된 경로 지정
save_directory = "Dora_model"
# 모델과 토크나이저 불러오기
model, tokenizer = FastLanguageModel.from_pretrained(
model_name = save_directory,
max_seq_length = 2048,
dtype = None,
load_in_4bit = True, # 양자화 옵션을 동일하게 설정
)
from transformers import TextStreamer
# ✅ Alpaca-style 프롬프트 템플릿
alpaca_prompt = """Below is an instruction that describes a task. Write a response that appropriately completes the request.
### Instruction:
{}
### Response:
{}"""
# 추론해보기
FastLanguageModel.for_inference(model) # 네이티브 2배 빠른 추론 활성화
# 추론을 위한 입력 준비
inputs = tokenizer(
[
alpaca_prompt.format(
"최저임금제도란 무엇이며, 최저임금액은 어떻게 결정되는지요?", # 인스트럭션 (명령어)
"", # 출력 - 생성할 답변을 비워둠
)
], return_tensors = "pt").to("cuda") # 텐서를 PyTorch 형식으로 변환하고 GPU로 이동
text_streamer = TextStreamer(tokenizer) # 토크나이저를 사용하여 스트리머 초기화
# 모델을 사용하여 텍스트 생성 및 스트리밍 출력
_ = model.generate(**inputs, streamer = text_streamer, max_new_tokens = 256)
4) 외부 지식 기반 답변 생성 (RAG 적용)
RAG (Retrieval-Augmented Generation)는 LLM이 응답을 생성할 때,
외부 지식(문서, DB 등)을 검색하여 함께 활용하는 방식입니다.
1단계, Retrieval (검색)
- 사용자의 질문에 대해 외부 문서에서 관련 정보를 검색합니다.
- 의미 기반 임베딩 검색(Dense Retrieval) 또는 키워드 기반 검색(BM25 등)을 활용할 수 있습니다.
2단계, Augmented Generation (응답 생성)
- 검색된 정보를 LLM의 입력 프롬프트에 함께 넣고 응답을 생성합니다.
- 문서 내용을 반영한 더 정확하고 풍부한 응답을 유도할 수 있습니다.
구성 요소 설명
Document Encoder 문서들을 사전에 임베딩해 벡터 DB로 저장
Query Encoder 사용자의 질문을 임베딩하여 검색에 사용
Retriever FAISS 등으로 벡터 유사도 기반 문서 검색
LLM 검색된 정보를 기반으로 답변 생성 (ex. QLoRA 튜닝된 LLM)
(1) Dense Retrieval (임베딩 기반 검색)
Dense Retrieval은 사용자의 질문(Query)을 임베딩 벡터로 변환한 후, 문서(또는 대화 기록)들을 동일한 벡터 공간에 투영하여 가장 유사한 문서를 검색하는 방식입니다.
기존의 키워드 기반 검색(BM25 등)과 달리, 의미 기반 유사도(MEANING-BASED SIMILARITY)를 파악할 수 있어 자연어 질문에 더 강력한 검색 성능을 보입니다.
구현 흐름
- 문서 구성: jihye-moon/LawQA-Ko 데이터셋의 질문 + 답변을 하나의 문서로 구성
- 문서 임베딩: sentence-transformers 기반 KoSBERT 또는 E5 모델로 임베딩 수행
- 인덱스 생성: FAISS를 이용한 벡터 DB 구축
- 검색: 사용자 질문 → 임베딩 → 유사도 기반 가장 유사한 문서 검색
- 출력: 검색된 문서를 기반으로 LLM 프롬프트에 삽입하여 응답 생성
SBERT (센텐스버트, Sentence-BERT)
SBERT는 기본적으로 BERT의 문장 임베딩의 성능을 우수하게 개선시킨 모델입니다. SBERT는 위에서 언급한 BERT의 문장 임베딩을 응용하여 BERT를 파인 튜닝합니다.
KoSBERT
- SBERT(Sentence-BERT)의 구조를 기반으로 하며,
- 한국어 자연어 추론(NLI) 데이터로 파인튜닝 되어 있어 문장 간 의미 유사도 계산에 탁월합니다.
- 본 실습에서는 sentence-transformers 라이브러리를 통해 jhgan/ko-sbert-nli 모델을 사용합니다.
- 이를 통해 "질문과 의미적으로 가장 비슷한 문서"를 빠르게 찾을 수 있습니다.
기존의 문장수준의 임베딩과 차이는 무엇인가?
기존에는 문장으로 묶어서 그 문장안의 토큰에 대해서 빈도수와 가중치 값으로 표현하는 것이다. (BOW, TF-IDF, LLM기반 BERT 등) 센텐스버트는 문장을 하나의 벡터로 압축하는 것이다. Pooling과 같이 생각하면 된다.
Step 1. 문서 준비 : 질문 + 답변 하나의 문장으로 구성
검색 문서는 단일 문장 형태로 구성합니다.
"질문\n답변" 형태로 결합하여 임베딩 벡터로 변환합니다
from datasets import load_dataset
# 데이터셋 로드
dataset = load_dataset("jihye-moon/LawQA-Ko", split="train")
# 문서 리스트 생성
documents = []
for example in dataset.select(range(1000)): # 예시로 1000개만 사용
question = example["question"].strip()
answer = example["answer"].strip()
full_text = f"{question}\n{answer}" # 질문과 답변 결합
documents.append(full_text)
print(f"✅ 문서 수: {len(documents)}")
print("📄 첫 번째 문서:\n", documents[0])
Step 2. 문서 임베딩: KoSBERT 기반
sentence-transformers의 jhgan/ko-sbert-nli 모델을 사용합니다.
모든 문서를 벡터화하여 numpy 배열로 저장합니다.
from sentence_transformers import SentenceTransformer
# KoSBERT 모델 로드
embedder = SentenceTransformer("jhgan/ko-sbert-nli")
# 문서 임베딩
doc_embeddings = embedder.encode(documents, convert_to_numpy=True, show_progress_bar=True)
Step 3. FAISS를 이용한 벡터 인덱스 생성
FAISS는 고속 벡터 검색 라이브러리로, 문서 간 유사도 기반 검색에 활용됩니다.
import faiss
import numpy as np
# 임베딩 차원 확인
dimension = doc_embeddings.shape[1]
# 벡터 인덱스 생성
index = faiss.IndexFlatL2(dimension)
index.add(doc_embeddings)
print("✅ FAISS 인덱스 생성 완료!")
Step 4. 유사 문서 검색 함수 정의
def search_documents(query, top_k=3):
# 쿼리 임베딩
query_vec = embedder.encode([query])
# 유사도 기반 검색
D, I = index.search(np.array(query_vec), k=top_k)
# 상위 문서 반환
return [documents[i] for i in I[0]]
# 예시: 질문을 입력해 관련 문서 검색
query = "부동산 계약 해지 조건은 어떻게 되나요?"
results = search_documents(query, top_k=3)
print("🧠 사용자 질문:", query)
print("\n📄 검색된 유사 문서:")
for i, doc in enumerate(results):
print(f"\n🔹 [Top {i+1}]\n{doc[:300]}...")
Step 5. 검색 결과를 기반으로 LLM에 질문 프롬프트 구성
Dense Retrieval로 가져온 문서들을 LLM 입력 프롬프트에 삽입합니다. 우리는 Alpaca-style 모델을 사용하고 있으므로, 프롬프트는 아래와 같은 형태로 구성합니다:
from unsloth import FastLanguageModel
from transformers import TextStreamer
import torch
# ✅ 모델 로드
model, tokenizer = FastLanguageModel.from_pretrained(
model_name = "MLP-KTLim/llama-3-Korean-Bllossom-8B", # finetuning 이전 원본 모델로 실습
max_seq_length = 2048,
dtype = None,
load_in_4bit = True
)
# ✅ 추론 최적화
FastLanguageModel.for_inference(model)
tokenizer.pad_token = tokenizer.eos_token
user_input = "부동산 계약 해지 조건은 어떻게 되나요?"
# ✅ 입력을 모델이 처리할 수 있는 텐서로 변환
inputs = tokenizer(user_input, return_tensors="pt").to("cuda")
print(f"📝 질문: {user_input}")
# ✅ 모델 응답 생성 (추론 옵션 적용)
outputs = model.generate(
**inputs,
max_new_tokens=256, # 응답 최대 길이 제한
temperature=0.5, # 온도 조절을 통해 일관된 응답 유도
top_p=0.85, # 상위 확률 토큰만 샘플링
repetition_penalty=1.2, # 반복 방지
do_sample=True # 샘플링 방식 적용
)
# ✅ 응답 디코딩
decoded_output = tokenizer.decode(outputs[0], skip_special_tokens=True)
print("🤖 응답:", decoded_output)
Step 6. LLM 프롬프트 구성 및 응답 생성
def generate_answer_with_retrieval(query, top_k=3, max_context_len=1500):
# 🔍 문서 검색
retrieved_docs = search_documents(query, top_k=top_k)
# 📄 문서 병합 (너무 길면 자름)
context = "\n\n---\n\n".join(retrieved_docs)
if len(context) > max_context_len:
context = context[:max_context_len]
# 🧾 프롬프트 구성 (Alpaca-style)
prompt = (
f"다음은 참고할 수 있는 문서들입니다:\n{context}\n\n"
f"### Instruction:\n{query}\n\n"
f"### Response:\n"
)
# 🔢 토크나이즈
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
# 🤖 응답 생성
outputs = model.generate(
**inputs,
max_new_tokens=256,
temperature=0.7,
top_p=0.9,
repetition_penalty=1.1,
)
# 🧾 디코딩
decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
if "### Response:" in decoded:
response = decoded.split("### Response:")[-1].strip()
else:
response = decoded.strip()
return response
# 최종 실행 예시
query = "부동산 계약 해지 조건은 어떻게 되나요?"
response = generate_answer_with_retrieval(query)
print("🧠 사용자 질문:", query)
print("\n🤖 LLM 응답:\n", response)
(2) Hybrid Search (BM25 + 벡터 검색)
Hybrid Search는 키워드 기반의 BM25 검색과 의미 기반의 임베딩 검색(Dense Retrieval)을 결합하여 정확도와 다양성을 모두 확보할 수 있는 검색 기법입니다.
- BM25는 텍스트 기반 키워드 매칭에 강하고,
- Dense Retrieval은 질문 의미와 유사한 문서를 찾아내는 데 효과적입니다.
- 두 검색 결과를 통합하여 평균 점수를 기반으로 최종 순위를 산출합니다.
BM25는 일반적으로 Whoosh, Elasticsearch, Lucene, scikit-learn TfidfVectorizer 등을 사용합니다. Colab에서는 간편하게 TfidfVectorizer를 활용한 BM25 유사도 기반 접근을 사용할 수 있습니다.
구현 흐름
- 문서 준비 (앞 단계 동일)
- 임베딩 벡터 & TF-IDF 벡터 준비
- Hybrid 검색 함수 정의
- LLM 프롬프트 구성 및 응답 생성
- Hybri 검색 기반 응답 함수 정의
Step 1. 문서 준비 (앞 단계 동일)
from datasets import load_dataset
dataset = load_dataset("jihye-moon/LawQA-Ko", split="train")
documents = []
for example in dataset.select(range(1000)):
question = example["question"].strip()
answer = example["answer"].strip()
full_text = f"{question}\n{answer}"
documents.append(full_text)
Step 2. 임베딩 벡터 & TF-IDF 벡터 준비
from sentence_transformers import SentenceTransformer
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
import faiss
# KoSBERT 임베딩
embedder = SentenceTransformer("jhgan/ko-sbert-nli")
doc_embeddings = embedder.encode(documents, convert_to_numpy=True, show_progress_bar=True)
# TF-IDF 벡터화
vectorizer = TfidfVectorizer()
tfidf_matrix = vectorizer.fit_transform(documents)
# FAISS 인덱스 생성
dimension = doc_embeddings.shape[1]
index = faiss.IndexFlatL2(dimension)
index.add(doc_embeddings)
Step 3. Hybrid 검색 함수 정의
def hybrid_search(query, top_k=3):
# 1. TF-IDF (BM25 유사도)
query_tfidf = vectorizer.transform([query])
bm25_scores = cosine_similarity(query_tfidf, tfidf_matrix).flatten()
# 2. Dense 임베딩 유사도
query_dense = embedder.encode([query])
D, I_dense = index.search(np.array(query_dense), k=top_k * 2) # 후보 넉넉히
# 3. Hybrid 점수 계산
hybrid_scores = {}
dense_indices = I_dense[0]
bm25_top = np.argsort(bm25_scores)[-top_k*2:]
for i in set(dense_indices).union(set(bm25_top)):
bm25_score = bm25_scores[i]
dense_score = 1 - np.linalg.norm(query_dense - doc_embeddings[i])
hybrid_scores[i] = (bm25_score + dense_score) / 2
# 4. 상위 문서 반환
sorted_docs = sorted(hybrid_scores.items(), key=lambda x: x[1], reverse=True)[:top_k]
return [documents[i] for i, _ in sorted_docs]
Step 4. LLM 프롬프트 구성 및 응답 생성
user_input = "부동산 계약 해지 조건은 어떻게 되나요?"
# ✅ 입력을 모델이 처리할 수 있는 텐서로 변환
inputs = tokenizer(user_input, return_tensors="pt").to("cuda")
print(f"📝 질문: {user_input}")
# ✅ 모델 응답 생성 (추론 옵션 적용)
outputs = model.generate(
**inputs,
max_new_tokens=256, # 응답 최대 길이 제한
temperature=0.5, # 온도 조절을 통해 일관된 응답 유도
top_p=0.85, # 상위 확률 토큰만 샘플링
repetition_penalty=1.2, # 반복 방지
do_sample=True # 샘플링 방식 적용
)
# ✅ 응답 디코딩
decoded_output = tokenizer.decode(outputs[0], skip_special_tokens=True)
print("🤖 응답:", decoded_output)
Step 5. Hybrid 검색 기반 응답 함수 정의
def generate_hybrid_answer(query, top_k=3, max_context_len=1500):
retrieved_docs = hybrid_search(query, top_k=top_k)
context = "\n\n---\n\n".join(retrieved_docs)
if len(context) > max_context_len:
context = context[:max_context_len]
prompt = (
f"다음은 참고할 수 있는 문서들입니다:\n{context}\n\n"
f"### Instruction:\n{query}\n\n"
f"### Response:\n"
)
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
outputs = model.generate(
**inputs,
max_new_tokens=200,
temperature=0.7,
top_p=0.9,
repetition_penalty=1.1,
)
decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
if "### Response:" in decoded:
response = decoded.split("### Response:")[-1].strip()
else:
response = decoded.strip()
return response
query = "부동산 계약 해지 조건은 어떻게 되나요?"
response = generate_hybrid_answer(query)
print("🧠 사용자 질문:", query)
print("\n🤖 LLM 응답:\n", response)
(3) Reranking을 통한 검색 결과 개선
Reranking은 Dense Search 또는 Hybrid Search로 먼저 후보 문서를 추출한 후, 질문과 문서 간의 의미적 일치도를 더 정밀하게 재측정하여 최종 순위를 다시 정렬하는 기법입니다.
- 초기 검색 단계에서는 빠르게 top-k 후보 문서를 추출
- 그 다음, CrossEncoder 같은 문서-질문 쌍 비교 모델을 사용해 정확한 순위 재조정
- LLM에는 최종 상위 문서만 전달하여 더 정확하고 정보성 높은 응답 생성 가능
주요 용어 정리
- Cross-Encoder : 문서-질문 쌍을 하나의 입력으로 받아 의미 유사도를 정밀하게 예측하는 모델 (ex. "질문 [SEP] 문서" 형태로 입력 → relevance score 출력)
- MS-MARCO : Microsoft가 만든 대규모 질의응답 데이터셋으로, 검색 및 rerank 학습에 많이 사용됨.
- KoBERT 기반 CrossEncoder : 한국어 검색 정제에 적합한 형태로 사전학습된 문장 쌍 비교 모델 (MS-MARCO 기반 KoSimCSE 등)
구현 흐름
- Hybrid Search 기반 초기 후보 문서 10개 추출
- Reranker 모델로 문서 순위 재조정
- LLM 프롬프트 구성 및 응답 생성
Step 1. Hybrid Search 기반 초기 후보 문서 10개 추출
# 사용자 질문
query = "부동산 계약 해지 조건은 어떻게 되나요?"
# 1. TF-IDF 유사도 계산
query_tfidf = vectorizer.transform([query])
bm25_scores = cosine_similarity(query_tfidf, tfidf_matrix).flatten()
bm25_top = np.argsort(bm25_scores)[::-1][:10]
# 2. Dense 임베딩 유사도 계산
query_dense = embedder.encode([query])
D, I_dense = index.search(np.array(query_dense), k=10)
dense_top = I_dense[0]
# 3. Hybrid Score 계산 (BM25 + Dense 평균)
hybrid_scores = {}
for i in set(bm25_top).union(set(dense_top)):
bm25_score = bm25_scores[i]
dense_score = 1 - np.linalg.norm(query_dense - doc_embeddings[i])
hybrid_scores[i] = (bm25_score + dense_score) / 2
# 후보 문서 추출 (10개)
top_candidates = sorted(hybrid_scores.items(), key=lambda x: x[1], reverse=True)[:10]
candidate_docs = [(query, documents[idx]) for idx, _ in top_candidates]
Step 2. Reranker 모델로 문서 순위 재조정
from sentence_transformers import CrossEncoder
# CrossEncoder 로드 (MS-MARCO 학습 기반)
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
# 문서 쌍 → 유사도 점수 계산
rerank_scores = reranker.predict(candidate_docs)
# 유사도 기반 재정렬
reranked = sorted(zip(candidate_docs, rerank_scores), key=lambda x: x[1], reverse=True)
# 최종 top-k 문서 선택 (예: 3개)
top_k = 3
reranked_docs = [doc for (query, doc), score in reranked[:top_k]]
Step 3. LLM 프롬프트 구성 및 응답 생성
from unsloth import FastLanguageModel
from transformers import TextStreamer
# 프롬프트용 문서 병합
context = "\n\n---\n\n".join(reranked_docs)
# LLM 프롬프트 구성 (Alpaca-style)
prompt = (
f"다음은 참고할 수 있는 문서들입니다:\n{context}\n\n"
f"### Instruction:\n{query}\n\n"
f"### Response:\n"
)
# 토크나이즈
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
# 응답 생성
outputs = model.generate(
**inputs,
max_new_tokens=200,
temperature=0.7,
top_p=0.9,
repetition_penalty=1.1,
)
# 응답 디코딩
decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
response = decoded.split("### Response:")[-1].strip()
# 출력
print("🧠 사용자 질문:", query)
print("\n🤖 LLM 응답:\n", response)
(4) Auto-merging Retrieval (자동 청킹 + 병합 전략)
Auto-merging Retrieval은 검색된 여러 문서가 너무 짧거나 내용이 흩어져 있을 경우, 이들을 하나로 병합하여 LLM이 더 잘 이해할 수 있도록 돕는 RAG 전략입니다.
이 전략은 다음과 같은 상황에서 유용합니다:
- 개별 문서의 길이가 짧아 정보가 부족할 때
- 여러 문서가 서로 연관된 정보를 가지고 있을 때
- LLM의 context window를 고려하여 효율적인 정보 입력이 필요할 때
구현 흐름
- Hybrid Retrieval 또는 Dense Retrieval을 통해 Top-K 문서를 검색
- 각 문서를 적절히 청크 또는 연결하여 하나의 context로 병합
- 병합된 문서를 프롬프트에 삽입하여 LLM 응답 생성
Step 1. 사용자 질문 입력 및 문서 검색
query = "부동산 계약 해지 조건은 어떻게 되나요?"
# 기존 hybrid_search 또는 search_documents 사용 가능
documents_to_merge = hybrid_search(query, top_k=5) # 또는 search_documents(query, top_k=5)
Step 2. 문서 병합 (자동 청킹)
# 병합된 context 구성
merged_context = "\n\n---\n\n".join(documents_to_merge)
# LLM 입력 길이 고려하여 context 길이 제한 (예: 3500자)
max_char_len = 3500
if len(merged_context) > max_char_len:
merged_context = merged_context[:max_char_len]
Step 3. 프롬프트 구성 및 응답 생성
# Alpaca-style 프롬프트 구성
prompt = (
f"다음은 참고할 수 있는 문서들입니다:\n{merged_context}\n\n"
f"### Instruction:\n{query}\n\n"
f"### Response:\n"
)
# 토크나이즈 및 LLM 추론
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
outputs = model.generate(
**inputs,
max_new_tokens=200,
temperature=0.7,
top_p=0.9,
repetition_penalty=1.1,
)
# 응답 디코딩
decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
response = decoded.split("### Response:")[-1].strip()
# 출력
print("🧠 사용자 질문:", query)
print("\n📄 병합된 문서 (context 일부):\n", merged_context[:500], "...")
print("\n🤖 LLM 응답:\n", response)
참고: 청킹 전략이 중요한 이유
- LLM은 제한된 context window 내에서만 정보를 처리할 수 있으므로, 너무 많은 문서를 그대로 넣으면 잘리는 정보가 발생할 수 있습니다.
- 이럴 때 관련된 문서끼리 병합하여 하나의 흐름 있는 context를 제공하면 응답 품질이 향상됩니다.
(5) Self-Query를 활용한 메타데이터 필터링
Self-Query 기반 메타데이터 필터링은 사용자의 질문에서 의미 있는 키워드나 주제(메타데이터)를 추출하고, 이 키워드를 기반으로 검색 대상 문서 자체를 선별하는 방법입니다.
즉, "검색 전에 관련 있는 문서만 골라서 검색"하는 전략입니다.
장점
- 검색 정확도 향상
- LLM 프롬프트 축소
- 응답 품질 향상
구현 흐름
- 사용자 질문에서 키워드 추출
- 메타데이터 조건에 맞는 문서 필터링
- Dense 임베딩 검색 수행
- LLM 프롬프트 구성 및 응답 생성
Step 1. 사용자 질문에서 키워드 추출
간단한 키워드 리스트 기반 추출을 사용하지만, 실제 서비스에서는 NER, Keyphrase Extraction, BERT 기반 분류기로 고도화할 수 있습니다.
# ✅ 사용자 질문
query = "부동산 계약 해지 조건은 어떻게 되나요?"
# ✅ 주요 키워드 리스트 (주제별 법률 용어 정의)
keyword_list = ["부동산", "계약 해지"]
# ✅ 질문에 포함된 키워드 추출
keywords = [kw for kw in keyword_list if kw in query]
print("🧠 추출된 키워드:", keywords)
Step 2. 메타데이터 조건에 맞는 문서 필터링
이전 단계에서 구성한 documents 리스트 중에서, 질문 키워드를 모두 포함한 문서만 추출합니다.
filtered_documents = []
filtered_embeddings = []
for i, doc in enumerate(documents):
if all(keyword in doc for keyword in keywords): # 모든 키워드 포함 시 선택
filtered_documents.append(doc)
filtered_embeddings.append(doc_embeddings[i])
print(f"✅ 필터링된 문서 수: {len(filtered_documents)}")
Step 3. Dense 임베딩 검색 수행
import numpy as np
# ✅ 사용자 질문 임베딩
query_vec = embedder.encode([query])
filtered_embeddings = np.array(filtered_embeddings)
# ✅ 유사도 기반 검색 (Cosine 유사도)
similarities = np.dot(filtered_embeddings, query_vec.T).squeeze()
top_k = 3
top_indices = np.argsort(similarities)[::-1][:top_k]
# ✅ 최종 검색 문서 선택
top_docs = [filtered_documents[i] for i in top_indices]
top_docs
Step 4. LLM 프롬프트 구성 및 응답 생성
# ✅ 문서 병합 및 context 구성
merged_context = "\n\n---\n\n".join(top_docs)
max_char_len = 1800
if len(merged_context) > max_char_len:
merged_context = merged_context[:max_char_len]
# ✅ Alpaca 스타일 프롬프트 생성
prompt = (
f"다음은 참고할 수 있는 문서들입니다:\n{merged_context}\n\n"
f"### Instruction:\n{query}\n\n"
f"### Response:\n"
)
# ✅ LLM 추론
inputs = tokenizer(prompt, return_tensors="pt").to("cuda")
outputs = model.generate(
**inputs,
max_new_tokens=200,
temperature=0.7,
top_p=0.9,
repetition_penalty=1.1,
)
# ✅ 응답 디코딩
decoded = tokenizer.decode(outputs[0], skip_special_tokens=True)
if "### Response:" in decoded:
response = decoded.split("### Response:")[-1].strip()
else:
response = decoded.strip()
print("🧠 사용자 질문:", query)
print("\n🤖 LLM 응답:\n", response)
Tags:
AI개발_교육
