멀티캠퍼스 데이터분석 6월 22일 수업 내용 — KoBART 인터뷰 요약 복습 실습, 텍스트 함의 인식(NLI) 데이터 구성 및 Seq2Seq 파인튜닝, 예측 함수 구현

---

📌 목차

KoBART 요약 복습
1. 인터뷰 데이터 KoBART 파인튜닝

텍스트 함의 인식 (NLI)

2. TL_text_entailment.json 파싱 & DataFrame 구성
3. 프롬프트 생성 & 데이터 전처리
4. Dataset 계층화 분할
5. Seq2Seq 토큰화 & 모델 학습
6. 평가 함수 — 생성 텍스트 → 0/1 변환
7. 예측 함수 구현 & 검증

---

KoBART 요약 복습

1. 인터뷰 데이터 KoBART 파인튜닝

데이터 로드 & DatasetDict

import pandas as pd
import numpy as np
from datasets import Dataset, DatasetDict
import evaluate
from transformers import (AutoTokenizer, AutoModelForSeq2SeqLM,
                          DataCollatorForSeq2Seq, Seq2SeqTrainer,
                          Seq2SeqTrainingArguments)
from konlpy.tag import Komoran

model_name = 'digit82/kobart-summarization'
komoran    = Komoran()

df       = pd.read_csv('인터뷰.csv')
answers  = df['answer'].tolist()
summarys = df['summary'].tolist()

# 상위 100개: train / 하위 20개: validation
raw_ds = DatasetDict({
    'train': Dataset.from_dict({
        'document': answers[:100], 'summary': summarys[:100]
    }),
    'validation': Dataset.from_dict({
        'document': answers[-20:], 'summary': summarys[-20:]
    })
})

💡 Dataset.from_pandas() 로도 동일하게 생성 가능

DatasetDict({
    'train'      : Dataset.from_pandas(df.iloc[:100]),
    'validation' : Dataset.from_pandas(df.iloc[-20:])
})

토큰화 — output 최대 길이 256

tokenizer = AutoTokenizer.from_pretrained(model_name, use_fast=True)
model     = AutoModelForSeq2SeqLM.from_pretrained(model_name)

max_input_len  = 512
max_target_len = 256   # 인터뷰 답변이 길어 뉴스 요약보다 크게 설정

def token_fn(batch):
    inputs = tokenizer(
        batch['document'], max_length=max_input_len,
        padding='max_length', truncation=True
    )
    labels = tokenizer(
        batch['summary'], max_length=max_target_len,
        padding='max_length', truncation=True
    )
    labels_ids = np.array(labels['input_ids'])
    labels_ids[labels_ids == tokenizer.pad_token_id] = -100
    inputs['labels'] = labels_ids.tolist()
    return inputs

tokenized_ds  = raw_ds.map(token_fn, batched=True, remove_columns=['document', 'summary'])
data_collator = DataCollatorForSeq2Seq(tokenizer=tokenizer, model=model)

ROUGE 평가 & 학습

rouge = evaluate.load('rouge')

def metrics(eval_pred):
    preds, labels = eval_pred
    labels    = np.where(labels != -100, labels, tokenizer.pad_token_id)
    pred_str  = [doc.strip() for doc in tokenizer.batch_decode(preds,  skip_special_tokens=True)]
    label_str = [doc.strip() for doc in tokenizer.batch_decode(labels, skip_special_tokens=True)]
    result    = rouge.compute(
        predictions=pred_str, references=label_str,
        tokenizer=lambda x: komoran.morphs(x)
    )
    return {k: round(v * 100, 2) for k, v in result.items()}

args = Seq2SeqTrainingArguments(
    output_dir='./kobart', eval_strategy='epoch', save_strategy='epoch',
    learning_rate=5e-5, num_train_epochs=5, logging_steps=1,
    predict_with_generate=True, generation_max_length=64, generation_num_beams=4,
    load_best_model_at_end=True, metric_for_best_model='rougeL',
    greater_is_better=True, dataloader_num_workers=3
)

trainer = Seq2SeqTrainer(
    model=model, args=args,
    train_dataset=tokenized_ds['train'], eval_dataset=tokenized_ds['validation'],
    processing_class=tokenizer, data_collator=data_collator, compute_metrics=metrics
)
trainer.train()

요약 텍스트 생성

test_text = answers[200]

inputs = tokenizer(
    test_text, return_tensors='pt', truncation=True,
    max_length=max_input_len, padding='max_length'
)

gen_ids = model.generate(
    **inputs.to(model.device),
    max_new_tokens=500, min_new_tokens=10,
    num_beams=4, do_sample=True,
    length_penalty=0.6, no_repeat_ngram_size=2,
    repetition_penalty=3.0,
    eos_token_id=tokenizer.eos_token_id,
    pad_token_id=tokenizer.pad_token_id
)

print(tokenizer.decode(gen_ids[0], skip_special_tokens=True))
print('실제 요약:', summarys[200])

저장된 모델 로드 & 추론

model2  = AutoModelForSeq2SeqLM.from_pretrained('./kobart_exam')
gen_ids = model2.generate(**inputs.to(model.device), max_new_tokens=500, ...)
print(tokenizer.decode(gen_ids[0], skip_special_tokens=True))

---

텍스트 함의 인식 (NLI)

2. TL_text_entailment.json 파싱 & DataFrame 구성

NLI(Natural Language Inference) — 원문(context)과 가설(question)이 주어졌을 때, 가설이 참(Yes) 인지 거짓(No) 인지 판별하는 태스크입니다.

JSON 파싱

import json
import pandas as pd
from pprint import pprint

with open('./data/TL_text_entailment.json', 'r', encoding='utf-8') as f:
    raw = json.load(f)

data = []
for doc in raw['data']:
    for para in doc['paragraphs']:
        context  = para['context']
        for qa in para['qas']:
            question = qa['question']
            answer   = qa['answers']['text']
            data.append({
                'context' : context,
                'question': question,
                'answer'  : answer
            })

df = pd.DataFrame(data)
df['answer'].value_counts()   # Yes / No 분포 확인

---

3. 프롬프트 생성 & 데이터 전처리

입력 프롬프트 형식 정의

ans2id   = {'No': 0, 'Yes': 1}
id2target = {0: '아니오', 1: '예'}

def build_prompt(context, question):
    return (
        f"다음 글을 읽고 문장이 참이면 '예', 거짓이면 '아니오'로 답하시오\n"
        f"원문 : {context}\n"
        f"문장 : {question}"
    )

# context + question → 프롬프트 결합
inputs  = df[['context', 'question']].apply(
    lambda x: build_prompt(*(x.tolist())), axis=1
).tolist()

# Yes/No → 예/아니오
targets = df['answer'].map(ans2id).map(id2target).tolist()

ds = Dataset.from_dict({
    'input_text' : inputs,
    'target_text': targets
})

💡 프롬프트 엔지니어링이 중요한 이유
Seq2Seq 모델은 입력 텍스트의 형식에 따라 출력이 크게 달라집니다.
학습 시와 추론 시에 동일한 프롬프트 형식을 사용해야 일관된 결과를 얻을 수 있습니다.

---

4. Dataset 계층화 분할

Hugging Face Dataset의 train_test_split() 은 계층화에 ClassLabel 타입이 필요합니다.

# strat 컬럼 추가 (계층화용)
ds = ds.map(lambda x: {'strat': 0 if x['target_text'] == '아니오' else 1})

# ClassLabel 타입으로 변환 (계층화 분할에 필수)
ds = ds.class_encode_column('strat')

# 1차 분할: train 80% / test 20%
split1    = ds.train_test_split(test_size=0.2, seed=42, stratify_by_column='strat')
train_ds  = split1['train'].remove_columns('strat')

# 2차 분할: test를 validation 50% / test 50%
split2    = split1['test'].train_test_split(test_size=0.5, seed=42, stratify_by_column='strat')
valid_ds  = split2['train'].remove_columns('strat')
test_ds   = split2['test'].remove_columns('strat')

💡 class_encode_column()이 필요한 이유
Hugging Face Dataset의 train_test_split(stratify_by_column=...) 은
해당 컬럼이 ClassLabel 타입이어야만 계층화 분할이 가능합니다.
일반 정수 컬럼은 class_encode_column() 으로 변환해야 합니다.

---

5. Seq2Seq 토큰화 & 모델 학습

토큰화 — 짧은 출력(예/아니오)에 맞게 max_target_len=8

model_name     = 'digit82/kobart-summarization'
max_input_len  = 512
max_target_len = 8   # '예' / '아니오' — 매우 짧으므로 8로 설정

tokenizer = AutoTokenizer.from_pretrained(model_name)

def token_fn(batch):
    inputs = tokenizer(
        batch['input_text'], max_length=max_input_len,
        truncation=True, return_token_type_ids=False
    )
    labels = tokenizer(
        batch['target_text'], max_length=max_target_len,
        truncation=True, return_token_type_ids=False
    )
    # NLI는 출력이 짧아 -100 마스킹 없이 그대로 사용
    inputs['labels'] = labels['input_ids']
    return inputs

cols     = ['input_text', 'target_text']
train_ds = train_ds.map(token_fn, batched=True, remove_columns=cols)
valid_ds = valid_ds.map(token_fn, batched=True, remove_columns=cols)
test_ds  = test_ds.map(token_fn,  batched=True, remove_columns=cols)

💡 요약 태스크와 NLI 토큰화의 차이

	요약 태스크	NLI 태스크
출력 길이	길다 (128~256)	매우 짧다 (8 이하)
패딩 -100 처리	필요 (패딩이 손실에 영향)	불필요 (출력이 너무 짧아 패딩 거의 없음)
return_token_type_ids	기본값 사용	False (KoBART 미사용)

모델 생성 & 학습

import torch
from transformers import AutoModelForSeq2SeqLM, DataCollatorForSeq2Seq

model = AutoModelForSeq2SeqLM.from_pretrained(
    model_name,
    num_labels = 2,
    id2label   = {0: 'LABEL_0', 1: 'LABEL_1'},
    label2id   = {'LABEL_0': 0, 'LABEL_1': 1}
)

data_collator = DataCollatorForSeq2Seq(
    tokenizer=tokenizer, model=model, label_pad_token_id=-100
)

args = Seq2SeqTrainingArguments(
    output_dir   = './kobart',
    num_train_epochs = 3,
    learning_rate    = 3e-5,
    weight_decay     = 0.01,
    warmup_ratio     = 0.1,
    eval_strategy    = 'epoch',
    save_strategy    = 'epoch',
    load_best_model_at_end   = True,
    metric_for_best_model    = 'f1',
    greater_is_better        = True,
    generation_max_length    = max_target_len,
    logging_steps            = 1,
    per_device_train_batch_size = 16,
    per_device_eval_batch_size  = 32
)

trainer = Seq2SeqTrainer(
    model        = model,
    args         = args,
    train_dataset = train_ds.select(range(500)),   # 메모리 절약을 위해 상위 500개만
    eval_dataset  = valid_ds.select(range(100)),
    processing_class = tokenizer,
    data_collator    = data_collator,
    compute_metrics  = metrics
)
trainer.train()

---

6. 평가 함수 — 생성 텍스트 → 0/1 변환

생성된 텍스트가 '예'로 시작하면 1, '아니오'로 시작하면 0으로 변환합니다.

import numpy as np
from sklearn.metrics import accuracy_score, f1_score

def text_to_id(text):
    """생성된 텍스트 → 정수 레이블 변환"""
    t = text.strip()
    if t.startswith('예'):
        return 1
    elif t.startswith('아니'):
        return 0
    else:
        return -1   # 파싱 실패

def metrics(eval_pred):
    preds, labels = eval_pred

    # 튜플 형태로 들어오는 경우 처리
    if isinstance(preds, tuple):
        preds = preds[0]
    # 3차원 → 2차원 (argmax)
    if len(preds.shape) == 3:
        preds = np.argmax(preds, axis=-1)

    labels    = np.where(labels != -100, labels, tokenizer.pad_token_id)
    pred_str  = tokenizer.batch_decode(preds,  skip_special_tokens=True)
    label_str = tokenizer.batch_decode(labels, skip_special_tokens=True)

    pred_ids  = [text_to_id(t) for t in pred_str]
    label_ids = [text_to_id(t) for t in label_str]

    acc       = accuracy_score(pred_ids, label_ids)
    f1        = f1_score(pred_ids, label_ids, average='macro')
    fail_rate = np.mean([p == -1 for p in pred_ids])   # 파싱 실패율

    return {
        'accuracy'       : round(acc * 100, 2),
        'f1'             : f1,
        'parse_fail_rate': fail_rate
    }

💡 parse_fail_rate를 지표로 추가하는 이유
Seq2Seq 모델이 간혹 '예'/'아니오' 외의 텍스트를 생성할 수 있습니다.
이 경우 -1로 처리하며, parse_fail_rate로 비정상 생성 비율을 모니터링합니다.

---

7. 예측 함수 구현 & 검증

단일 예측 함수 — context + question 입력

model2 = AutoModelForSeq2SeqLM.from_pretrained('./best')

def predict(context, question):
    """원문과 가설 질문을 입력받아 참/거짓 판별"""
    prompt = build_prompt(context, question)
    inputs = tokenizer(
        prompt, max_length=max_input_len,
        truncation=True, return_tensors='pt',
        return_token_type_ids=False
    ).to(model.device)

    with torch.no_grad():
        out_ids = model.generate(**inputs, max_new_tokens=max_target_len, num_beams=1)

    text    = tokenizer.decode(out_ids[0], skip_special_tokens=True).strip()
    pred_id = text_to_id(text)
    label   = {0: '아니오(거짓)', 1: '예(참)', -1: '측정불가'}[pred_id]
    return label, text

# 예시
context_test  = "최고위원 선거에서는 처럼회 멤버로 강성 친문 김용민 의원이 가장 많은 표를 얻었다."
question_test = "처럼회는 처음으로 국회의원 직무를 수행하는 민주당 의원들의 모임이야"

label, raw = predict(context_test, question_test)
print('생성된 텍스트:', raw, '/ 판정:', label)

test_ds 상위 5개 예측 & 정확도 확인

def predict2(input_ids, attention_mask):
    """토큰화된 Dataset 데이터로 직접 예측"""
    with torch.no_grad():
        out_ids = model2.generate(
            input_ids=input_ids, attention_mask=attention_mask,
            max_new_tokens=max_target_len, num_beams=1
        )
    text    = tokenizer.decode(out_ids[0], skip_special_tokens=True).strip()
    pred_id = text_to_id(text)
    label   = {0: '아니오', 1: '예', -1: '판정불가'}[pred_id]
    return label, pred_id

cnt       = 0
total_cnt = 5

for i in range(total_cnt):
    data = test_ds[i]

    # [seq_len] → [1, seq_len] (배치 차원 추가)
    input_ids      = torch.tensor(data['input_ids'],      device=model2.device).unsqueeze(0)
    attention_mask = torch.tensor(data['attention_mask'], device=model2.device).unsqueeze(0)

    pred_str, pred_id = predict2(input_ids, attention_mask)

    # 실젯값 디코딩
    label_str = tokenizer.decode(data['labels'], skip_special_tokens=True).replace('\n', ' ').strip()
    label_id  = text_to_id(label_str)

    ok   = (pred_id == label_id)
    cnt += ok
    mark = '예측 성공' if ok else '예측 실패'

    # 입력 텍스트 확인 (앞 100자)
    show_text = tokenizer.decode(data['input_ids'], skip_special_tokens=True).strip()
    print(show_text[:100])
    print(mark, '/ 실젯값:', label_str, '/ 예측값:', pred_str)

print(f'정확도: {(cnt / total_cnt) * 100}%')

JSON 파일에서 직접 예측 — predict_json_sample

def predict_json_sample(path, samples):
    """
    path    : JSON 파일 경로
    samples : [(context 인덱스, 질문 문자열), ...]
    """
    with open(path, 'r', encoding='utf-8') as f:
        raw = json.load(f)

    contexts = [p['context'] for doc in raw['data'] for p in doc['paragraphs']]

    model2.eval()
    for idx, question in samples:
        context      = contexts[idx]
        label, gen   = predict(context, question)
        print(f'\n [본문 #{idx}] {context[:100]}...')
        print(f'    질문 : {question}')
        print(f'    예측 : {label}  (모델 생성: {gen})')

# 사용 예시
file_path    = './data/VS_text_entailment.json'
demo_samples = [
    (0, '오로고보맙은 난소암 치료를 위한 면역 항암 신약이다.'),    # 기대: 예
    (0, '오로고보맙의 미국 FDA 임상3상은 중단되었다.'),           # 기대: 아니오
    (1, '법무부는 이태원 클럽을 방문한 외국인에게 코로나19 자진 검사를 독려했다')  # 기대: 예
]

predict_json_sample(file_path, demo_samples)

---

📎 핵심 개념 요약

개념	설명
NLI	원문(context)과 가설(question)의 참/거짓 판별
프롬프트 엔지니어링	학습/추론 시 동일한 입력 형식 유지 필수
build_prompt()	context + question → 구조화된 입력 텍스트 생성
class_encode_column()	계층화 분할을 위해 정수→ClassLabel 타입 변환
stratify_by_column	Hugging Face Dataset 계층화 분할 매개변수
remove_columns('strat')	학습에 불필요한 보조 컬럼 제거
text_to_id()	생성 텍스트 → 0/1/-1 정수 변환
parse_fail_rate	'예'/'아니오' 외의 텍스트 생성 비율 모니터링
isinstance(preds, tuple)	예측 결과 형태 동적 처리
max_target_len=8	NLI는 출력이 짧아 작은 값 사용
return_token_type_ids=False	KoBART는 token_type_ids 미사용
dataset.select(range(N))	상위 N개만 선택 (메모리/시간 절약)
unsqueeze(0)	[seq_len] → [1, seq_len] 배치 차원 추가
predict_json_sample()	JSON 파일에서 context 추출 후 즉시 예측
dataloader_num_workers	데이터 로더 병렬 처리 수 (CPU 코어 수 기준)