사전 학습된 언어 모델(PML)

1. PML

PLM(Pre-trained Language Model)은 대량의 텍스트 데이터를 사전 학습하여 자연어 이해와 생성 능력을 갖춘 인공지능 모델입니다. 대표적으로 BERT, GPT, T5 등이 있으며, 이들은 대규모 데이터에서 단어의 의미와 문맥을 학습한 후, 특정 작업(예: 문장 분류, 번역, 질의응답 등)에 맞게 추가 학습(Fine-tuning)하여 활용됩니다. PLM은 문맥을 고려한 자연어 처리 능력이 뛰어나며, 다양한 언어 기반 AI 애플리케이션에서 핵심 기술로 사용됩니다.

 

1. BERT

BERT(Bidirectional Encoder Representations from Transformers)는 구글에서 개발한 사전 훈련된 자연어 처리(NLP) 모델로, Transformer 아키텍처를 기반으로 양방향 문맥 이해가 가능하도록 설계되었습니다. 기존 NLP 모델들은 단어의 앞 또는 뒤 방향으로만 문맥을 고려하는 경우가 많았지만, BERT는 문장의 양쪽 방향에서 동시에 문맥을 학습하여 더 정교한 언어 이해가 가능합니다. BERT는 두 가지 주요 사전 훈련 기법인 Masked Language Model(MLM)과 Next Sentence Prediction(NSP)을 사용하여 텍스트의 의미를 효과적으로 학습하며, 이를 통해 다양한 NLP 태스크(예: 문장 분류, 개체명 인식, 질의응답 시스템 등)에 적용될 수 있습니다. 또한, BERT는 기본적으로 사전 훈련된 후 특정 작업(Task-Specific Fine-tuning)을 통해 원하는 NLP 태스크에 맞게 미세 조정(fine-tuning)할 수 있어 강력한 성능을 발휘합니다.

 

2. GPT

GPT(Generative Pre-trained Transformer)는 OpenAI에서 개발한 사전 훈련된 자연어 처리(NLP) 모델로, Transformer 아키텍처를 기반으로 한 자동회귀(Auto-regressive) 모델입니다. GPT는 대량의 텍스트 데이터로 사전 훈련을 수행한 후, 특정 태스크(Task-Specific Fine-tuning) 없이도 문장 생성, 번역, 요약 등 다양한 언어 관련 작업을 수행할 수 있습니다. 특히, GPT는 단방향 학습(unidirectional learning) 방식을 사용하여 입력된 문맥을 기반으로 다음 단어를 예측하며 텍스트를 생성하는데, 이는 챗봇, 스토리 생성, 코드 자동 완성 등 창의적인 콘텐츠 생성에 강력한 성능을 발휘합니다. GPT 시리즈는 버전이 업그레이드될수록 모델 크기와 성능이 향상되었으며, 특히 GPT-3와 GPT-4는 대규모 데이터를 활용한 학습으로 더욱 정교한 언어 이해와 생성 능력을 갖추고 있습니다.

 

 

 

2. Hugging Face

Hugging Face는 자연어 처리(NLP)와 머신러닝(ML) 모델을 쉽게 활용할 수 있도록 지원하는 오픈소스 플랫폼이자 AI 회사입니다. 대표적으로 Transformers 라이브러리를 통해 BERT, GPT, T5 등 다양한 사전 학습된 모델을 제공하며, Datasets 라이브러리로 다양한 데이터셋을 효율적으로 활용할 수 있습니다. 또한, Hugging Face Hub에서는 연구자와 개발자들이 AI 모델과 데이터셋을 공유할 수 있으며, Gradio와 Spaces를 활용해 간단한 웹 애플리케이션으로 AI 모델을 배포할 수도 있습니다. PyTorch와 TensorFlow를 모두 지원하며, AI 연구 및 애플리케이션 개발을 보다 쉽게 만들어주는 강력한 도구입니다.

 

1. Hugging Face 모델 허브

Hugging Face 모델 허브는 BERT, GPT, T5, Stable Diffusion 등 다양한 사전 학습된 AI 모델을 제공하는 공개 플랫폼입니다. 사용자는 자연어 처리(NLP), 컴퓨터 비전, 음성 인식 등 다양한 분야의 모델을 검색하고, PyTorch 또는 TensorFlow와 함께 손쉽게 활용할 수 있습니다. 모델마다 설명, 예제 코드, 라이선스 정보가 제공되며, 직접 학습한 모델을 업로드하거나 다른 연구자들이 공유한 모델을 다운로드하여 활용할 수도 있습니다. 이를 통해 연구자와 개발자는 AI 모델을 빠르게 실험하고 실제 애플리케이션에 적용할 수 있습니다. 검색창에 'ko' 또는 'kor'를 입력하여 한국어 모델을 서치할 수 있습니다.

 

2. Hugging Face를 이용한 토큰화

Hugging Face의 각 모델은 각 모델과 맵핑되는 토크나이저가 존재합니다. A모델을 사용한다면 A모델의 토크나이저를 사용해야 합니다. 각 토크나이저는 Vocabulary 정보를 담고 있으므로 A모델에 B토크나이저를 사용할 경우, 입력을 이해하지 못하는 상황이 발생합니다. 예를 들어 A모델은 단어 '사과'가 3번이고, B모델은 단어 '사과'가 51번인데, 입력으로 '사과가 먹고싶다'라는 문장이 들어올 경우 입력의 혼선이 발생하게됩니다.

 

3. 한국어 금융 뉴스 긍정, 부정, 중립 분류하기

금융 뉴스 문장 감성 분석 데이터셋(Finance Sentiment Corpus)은 금융 관련 뉴스 문장을 긍정(Positive), 부정(Negative), 중립(Neutral)으로 분류한 감성 분석 데이터셋입니다. 이 데이터셋은 주식시장 예측, 금융 리포트 분석, 투자 전략 수립 등을 위해 자연어 처리(NLP) 모델을 훈련하는 데 활용됩니다. 일반적으로 주식 시장의 변동성과 연관된 뉴스 기사의 영향을 분석하는 데 유용하며, 금융 전문가의 의견, 기업 실적 발표, 경제 지표 등의 내용을 감성 레이블과 함께 제공합니다.

 

!pip install transformers
!pip install datasets

 

!wget https://raw.githubusercontent.com/ukairia777/finance_sentiment_corpus/main/finance_data.csv

 

import pandas as pd
import numpy as np
import random
import time
import datetime
import csv
import os
import torch
import torch.nn.functional as F

from tqdm import tqdm
from datasets import load_dataset
from torch.nn.utils.rnn import pad_sequence

from transformers import pipeline
from transformers import BertTokenizer
from transformers import BertForSequenceClassification, AdamW, BertConfig
from transformers import get_linear_schedule_with_warmup
from torch.utils.data import TensorDataset, DataLoader, RandomSampler, SequentialSampler

from torch.nn.utils.rnn import pad_sequence

from sklearn.model_selection import train_test_split
from sklearn.metrics import f1_score, roc_auc_score, accuracy_score, hamming_loss

 

df = pd.read_csv('finance_data.csv')
df.head()

 

df['labels'] = df['labels'].replace(['neutral', 'positive', 'negative'],[0, 1, 2])
df.head()

 

all_data = load_dataset(
        "csv",
        data_files={
            "train": "finance_data.csv",
        },
    )
    
all_data

 

cs = all_data['train'].train_test_split(0.2)
train_cs = cs["train"]
test_cs = cs["test"]

train_cs

 

test_cs

 

cs = train_cs.train_test_split(0.2)
train_cs = cs["train"]
valid_cs = cs["test"]

 

train_cs

 

valid_cs

 

test_cs

 

print('두번째 샘플 출력 :', train_cs['kor_sentence'][1])
print('두번째 샘플의 레이블 출력 :', train_cs['labels'][1])

 

train_sentences = list(map(lambda x: '[CLS] ' + str(x) + ' [SEP]', train_cs['kor_sentence']))
validation_sentences = list(map(lambda x: '[CLS] ' + str(x) + ' [SEP]', valid_cs['kor_sentence']))
test_sentences = list(map(lambda x: '[CLS] ' + str(x) + ' [SEP]', test_cs['kor_sentence']))

 

train_labels = train_cs['labels']
validation_labels = valid_cs['labels']
test_labels = test_cs['labels']

 

test_sentences[:5]

 

test_labels[:5]

 

tokenizer = BertTokenizer.from_pretrained('klue/bert-base')

 

MAX_LEN = 128

def data_to_tensor(sentences, labels, tokenizer):
    # 정수 인코딩: 각 텍스트를 토큰화한 후 Vocabulary에 맵핑되는 정수 시퀀스로 변환
    tokenized_texts = [tokenizer.tokenize(sent) for sent in sentences]
    input_ids = [torch.tensor(tokenizer.convert_tokens_to_ids(x)) for x in tokenized_texts]

    # PyTorch의 pad_sequence 사용 (batch_first=True로 설정)
    padded_inputs = pad_sequence(input_ids, batch_first=True, padding_value=0)

    # 고정된 MAX_LEN 길이로 패딩 적용 (부족하면 0 추가, 길면 자름)
    if padded_inputs.shape[1] < MAX_LEN:
        padded_inputs = F.pad(padded_inputs, (0, MAX_LEN - padded_inputs.shape[1]), value=0)
    else:
        padded_inputs = padded_inputs[:, :MAX_LEN]

    # Attention Mask 생성 (0이 아닌 값은 1, 0은 0)
    attention_masks = (padded_inputs != 0).float()

    # 텐서 변환
    tensor_inputs = padded_inputs
    tensor_labels = torch.tensor(labels)
    tensor_masks = attention_masks

    return tensor_inputs, tensor_labels, tensor_masks

 

train_inputs, train_labels, train_masks = data_to_tensor(train_sentences, train_labels, tokenizer)
validation_inputs, validation_labels, validation_masks = data_to_tensor(validation_sentences, validation_labels, tokenizer)
test_inputs, test_labels, test_masks = data_to_tensor(test_sentences, test_labels, tokenizer)

print(train_inputs[0])
print(train_masks[0])

 

print(tokenizer.decode([2]))
print(tokenizer.decode([3]))

 

batch_size = 32

train_data = TensorDataset(train_inputs, train_masks, train_labels)
train_sampler = RandomSampler(train_data)
train_dataloader = DataLoader(train_data, sampler=train_sampler, batch_size=batch_size)

validation_data = TensorDataset(validation_inputs, validation_masks, validation_labels)
validation_sampler = SequentialSampler(validation_data)
validation_dataloader = DataLoader(validation_data, sampler=validation_sampler, batch_size=batch_size)

test_data = TensorDataset(test_inputs, test_masks, test_labels)
test_sampler = RandomSampler(test_data)
test_dataloader = DataLoader(test_data, sampler=test_sampler, batch_size=batch_size)

print('훈련 데이터의 크기:', len(train_labels))
print('검증 데이터의 크기:', len(validation_labels))
print('테스트 데이터의 크기:', len(test_labels))

 

num_labels = 3
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

model = BertForSequenceClassification.from_pretrained("klue/bert-base", num_labels=num_labels)
model.cuda()

 

optimizer = AdamW(model.parameters(), lr = 2e-5, eps = 1e-8)
epochs = 5
total_steps = len(train_dataloader) * epochs
scheduler = get_linear_schedule_with_warmup(optimizer, num_warmup_steps = 0, num_training_steps = total_steps)

 

def format_time(elapsed):
    elapsed_rounded = int(round((elapsed)))
    return str(datetime.timedelta(seconds=elapsed_rounded))  # hh:mm:ss
    
def metrics(predictions, labels):
    y_pred = predictions
    y_true = labels

    # 사용 가능한 메트릭들을 사용한다.
    accuracy = accuracy_score(y_true, y_pred)
    f1_macro_average = f1_score(y_true=y_true, y_pred=y_pred, average='macro', zero_division=0)
    f1_micro_average = f1_score(y_true=y_true, y_pred=y_pred, average='micro', zero_division=0)
    f1_weighted_average = f1_score(y_true=y_true, y_pred=y_pred, average='weighted', zero_division=0)

    # 메트릭 결과에 대해서 리턴
    metrics = {'accuracy': accuracy,
               'f1_macro': f1_macro_average,
               'f1_micro': f1_micro_average,
               'f1_weighted': f1_weighted_average}

    return metrics

 

# 랜덤 시드값.
seed_val = 777
random.seed(seed_val)
np.random.seed(seed_val)
torch.manual_seed(seed_val)
torch.cuda.manual_seed_all(seed_val)

model.zero_grad()
for epoch_i in range(0, epochs):
    print('======== Epoch {:} / {:} ========'.format(epoch_i + 1, epochs))
    t0 = time.time()
    total_loss = 0

    model.train()

    for step, batch in tqdm(enumerate(train_dataloader)):
        if step % 500 == 0 and not step == 0:
            elapsed = format_time(time.time() - t0)
            print('  Batch {:>5,}  of  {:>5,}.    Elapsed: {:}.'.format(step, len(train_dataloader), elapsed))

        batch = tuple(t.to(device) for t in batch)
        b_input_ids, b_input_mask, b_labels = batch

        outputs = model(b_input_ids,
                        token_type_ids=None,
                        attention_mask=b_input_mask,
                        labels=b_labels)

        loss = outputs[0]
        total_loss += loss.item()
        loss.backward()

        torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)  # gradient clipping if it is over a threshold
        optimizer.step()
        scheduler.step()

        model.zero_grad()

    avg_train_loss = total_loss / len(train_dataloader)

    print("")
    print("  Average training loss: {0:.4f}".format(avg_train_loss))
    print("  Training epcoh took: {:}".format(format_time(time.time() - t0)))

 

t0 = time.time()
model.eval()
accum_logits, accum_label_ids = [], []

for batch in validation_dataloader:
    batch = tuple(t.to(device) for t in batch)
    b_input_ids, b_input_mask, b_labels = batch

    with torch.no_grad():
        outputs = model(b_input_ids,
                        token_type_ids=None,
                        attention_mask=b_input_mask)

    logits = outputs[0]
    logits = logits.detach().cpu().numpy()
    label_ids = b_labels.to('cpu').numpy()

    for b in logits:
        # 3개의 값 중 가장 큰 값을 예측한 인덱스로 결정
        # ex) [ 3.5134246  -0.30875662 -2.111316  ] ==> 0
        accum_logits.append(np.argmax(b))

    for b in label_ids:
        accum_label_ids.append(b)

accum_logits = np.array(accum_logits)
accum_label_ids = np.array(accum_label_ids)
results = metrics(accum_logits, accum_label_ids)

print("Accuracy: {0:.4f}".format(results['accuracy']))
print("F1 (Macro) Score: {0:.4f}".format(results['f1_macro']))
print("F1 (Micro) Score: {0:.4f}".format(results['f1_micro']))
print("F1 (Weighted) Score: {0:.4f}".format(results['f1_weighted']))

 

%pwd
%mkdir model

 

path = '/content/model/'

torch.save(model.state_dict(), path+"BERT_news_positive_negative_model.pt")

 

model.load_state_dict(torch.load(path+"BERT_news_positive_negative_model.pt"))

 

t0 = time.time()
model.eval()
accum_logits, accum_label_ids = [], []

for step, batch in tqdm(enumerate(test_dataloader)):
    if step % 100 == 0 and not step == 0:
        elapsed = format_time(time.time() - t0)
        print('  Batch {:>5,}  of  {:>5,}.    Elapsed: {:}.'.format(step, len(test_dataloader), elapsed))

    batch = tuple(t.to(device) for t in batch)
    b_input_ids, b_input_mask, b_labels = batch

    with torch.no_grad():
        outputs = model(b_input_ids,
                        token_type_ids=None,
                        attention_mask=b_input_mask)

    logits = outputs[0]
    logits = logits.detach().cpu().numpy()
    label_ids = b_labels.to('cpu').numpy()

    for b in logits:
        # 3개의 값 중 가장 큰 값을 예측한 인덱스로 결정
        # ex) [ 3.5134246  -0.30875662 -2.111316  ] ==> 0
        accum_logits.append(np.argmax(b))

    for b in label_ids:
        accum_label_ids.append(b)

accum_logits = np.array(accum_logits)
accum_label_ids = np.array(accum_label_ids)
results = metrics(accum_logits, accum_label_ids)

print("Accuracy: {0:.4f}".format(results['accuracy']))
print("F1 (Macro) Score: {0:.4f}".format(results['f1_macro']))
print("F1 (Micro) Score: {0:.4f}".format(results['f1_micro']))
print("F1 (Weighted) Score: {0:.4f}".format(results['f1_weighted']))

 

pipe = pipeline("text-classification", model=model.cuda(), tokenizer=tokenizer, device=0, max_length=512,
                return_all_scores=True, function_to_apply='softmax')

 

result = pipe('SK하이닉스가 매출이 급성장하였다')
print(result)

 

pipe = pipeline("text-classification", model=model.cuda(), tokenizer=tokenizer, device=0, max_length=512, function_to_apply='softmax')

 

result = pipe('SK하이닉스가 매출이 급성장하였다')
print(result)

 

label_dict = {'LABEL_0' : '중립', 'LABEL_1' : '긍정', 'LABEL_2' : '부정'}

 

def prediction(text):
  result = pipe(text)
  return [label_dict[result[0]['label']]]

 

prediction('인공지능 기술의 발전으로 누군가는 기회를 얻을 것이고, 누군가는 얻지 못할 것이다')

 

prediction('많은 중소기업들이 회사를 정리하고 있다')

 

 

4. 학습 테크닉의 발전

1. Transfer Learning

전이 학습(Transfer Learning)은 한 작업(Task)에서 학습한 지식을 다른 관련 작업에 적용하는 머신러닝 기법입니다. 일반적으로 대량의 데이터와 연산 자원이 필요한 모델을 처음부터 새롭게 훈련하는 대신, 사전 훈련된 모델(Pre-trained Model)의 가중치를 가져와 새로운 데이터에 맞게 일부 층을 조정(Fine-tuning)하거나 특정 기능을 고정(Feature Extraction)하여 활용합니다. 예를 들어, 이미지 분류 모델로 사전 훈련된 CNN(합성곱 신경망) 모델의 초기 계층을 고정하고, 마지막 분류 계층만 새로운 데이터에 맞게 학습시키는 방식이 대표적입니다. 이는 데이터가 적거나 학습 시간이 제한된 상황에서 높은 성능을 얻을 수 있도록 도와주며, 자연어 처리(NLP), 컴퓨터 비전(CV) 등 다양한 분야에서 널리 사용됩니다.

 

2. Multi-task Learning

멀티 태스크 러닝(Multi-task Learning, MTL)은 하나의 모델이 여러 관련된 작업(Task)을 동시에 학습하도록 하는 머신러닝 기법입니다. 이는 여러 개의 작업이 공통된 정보를 공유할 수 있다는 가정하에, 하나의 모델이 다양한 태스크에서 학습한 지식을 활용하여 성능을 향상시키도록 합니다. 일반적으로 MTL 모델은 공유된 표현 학습(Shared Representation Learning)을 통해 서로 다른 태스크 간에 특징(feature)을 공유하며, 이를 통해 데이터 효율성을 높이고 과적합(overfitting)을 방지할 수 있습니다. 예를 들어, 자연어 처리(NLP)에서는 문장 분류, 개체명 인식(NER), 감정 분석 등의 작업을 하나의 모델에서 함께 학습할 수 있으며, 컴퓨터 비전에서는 물체 감지(Object Detection)와 이미지 분할(Segmentation)을 동시에 수행할 수 있습니다. MTL은 모델이 일반화 성능을 높이고 특정 태스크에 대한 편향을 줄이는 데 유용하게 활용됩니다.

 

3. Zero-shot & Few-shot Learning

Zero-shot Learning(ZSL)과 Few-shot Learning(FSL)은 적은 양의 데이터 또는 전혀 학습되지 않은 클래스에 대해 모델이 일반화하여 예측할 수 있도록 하는 기법입니다. Zero-shot Learning(ZSL)은 특정 태스크나 클래스에 대한 사전 학습 없이, 모델이 기존에 학습한 개념과 관계를 활용하여 새로운 개념을 이해하고 예측하는 방법입니다. 예를 들어, 이미지 분류에서 "코끼리"라는 클래스를 학습한 적이 없더라도, "큰 몸집, 긴 코"와 같은 속성 정보를 기반으로 새로운 클래스를 예측할 수 있습니다. 반면, Few-shot Learning(FSL)은 매우 적은 데이터 샘플(예: 1~5개)만 가지고도 새로운 태스크를 학습할 수 있도록 하는 방법으로, 대표적인 접근법으로는 메타 학습(Meta-learning)과 사전 훈련된 대규모 모델을 활용한 방법이 있습니다. 이러한 기법들은 특히 자연어 처리(NLP), 이미지 분류, 의료 진단 등 데이터가 제한적인 환경에서 강력한 성능을 발휘합니다.

 

4. Hierachical Attention

Hierarchical Attention(계층적 어텐션)은 자연어 처리(NLP)에서 문서 또는 긴 문장을 보다 효과적으로 이해하기 위해 어텐션 메커니즘을 계층적으로 적용하는 기법입니다. 일반적인 어텐션 메커니즘은 모든 단어에 동일한 방식으로 가중치를 부여하지만, 계층적 어텐션은 문서 구조를 반영하여 단어 수준(word-level) 어텐션과 문장 수준(sentence-level) 어텐션을 각각 학습합니다. 먼저 단어 수준 어텐션을 사용하여 각 문장에서 중요한 단어를 강조한 후, 문장 수준 어텐션을 통해 전체 문서에서 중요한 문장을 추출하는 방식으로 동작합니다. 이를 통해 모델은 긴 문서에서도 핵심 정보를 효과적으로 추출할 수 있으며, 감성 분석, 문서 요약, 기계 번역 등 다양한 NLP 태스크에서 성능을 향상시키는 데 활용됩니다. 대표적인 예로는 Hierarchical Attention Networks(HAN) 모델이 있으며, 이는 문맥을 보다 깊이 있게 반영하여 문서 분류 작업에서 강력한 성능을 보입니다.

 

5. Knowledge Distillation

Knowledge Distillation(지식 증류)은 크고 복잡한 모델(Teacher Model)의 지식을 작은 모델(Student Model)에 전달하여 성능을 유지하면서도 경량화하는 모델 압축 기법입니다. 일반적으로 대규모 신경망은 높은 성능을 가지지만 연산 비용이 크고 실시간 응용에 적합하지 않은 경우가 많습니다. 이를 해결하기 위해 Knowledge Distillation에서는 Teacher Model이 예측한 소프트 확률 분포(Soft Targets)를 Student Model이 학습하도록 하여 일반적인 정답 레이블(Hard Labels)보다 더 풍부한 정보(예: 클래스 간 관계)를 제공받을 수 있도록 합니다. 이 과정에서 온도 매개변수(Temperature Scaling)를 활용하여 Soft Targets의 정보를 조절하며, Student Model은 이를 통해 일반화 성능을 개선하고 모델의 크기와 추론 속도를 최적화할 수 있습니다. Knowledge Distillation은 모바일 기기, 임베디드 시스템 등 제한된 자원 환경에서 강력한 모델을 구현하는 데 유용하며, 자연어 처리(NLP), 컴퓨터 비전(CV) 등 다양한 분야에서 널리 활용됩니다.

 

6. Human Feedback 기반의 Reinforcement Learning

Human Feedback 기반 강화 학습(RLHF, Reinforcement Learning from Human Feedback)은 전통적인 강화 학습(RL)에서 보상 함수(Reward Function)를 사람이 직접 제공하는 피드백을 활용하여 학습하는 방법입니다. 일반적인 RL에서는 환경에서 자동으로 계산된 보상을 기반으로 에이전트가 최적의 행동을 학습하지만, RLHF에서는 사람이 직접 행동에 대한 선호도(Preference)나 보상을 제공하여 보다 직관적이고 정교한 학습이 가능합니다. 이 방법은 특히 명확한 보상 설계가 어려운 자연어 처리(NLP)나 윤리적 AI 개발에서 효과적이며, 예를 들어 OpenAI의 GPT 모델은 RLHF를 활용하여 사용자에게 더욱 자연스럽고 유용한 응답을 생성하도록 학습되었습니다. RLHF는 먼저 사람이 데이터에 태그를 달아 보상 모델(Reward Model)을 학습하고, 이후 강화 학습을 통해 모델이 인간의 선호도를 반영하도록 최적화하는 과정을 거칩니다. 이를 통해 AI 모델이 단순히 성능 최적화뿐만 아니라, 보다 인간 친화적이고 윤리적인 결과를 생성할 수 있도록 돕습니다.