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, BertConfig
from transformers import get_linear_schedule_with_warmup
from datasets import Dataset
from torch.utils.data import TensorDataset, DataLoader, RandomSampler, SequentialSampler
from torch.nn.utils.rnn import pad_sequence
from torch.optim import AdamW
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': 0, 'positive': 1, 'negative': 2}).astype('int64')
df.head()
train_cs, valid_cs = train_test_split(df, test_size=0.2, random_state=2025, stratify=df['labels'])
train_cs
train_cs, test_cs = train_test_split(train_cs, test_size=0.2, random_state=2025)
test_cs
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'].to_numpy(dtype='int64')
validation_labels = valid_cs['labels'].to_numpy(dtype='int64')
test_labels = test_cs['labels'].to_numpy(dtype='int64')
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('많은 중소기업들이 회사를 정리하고 있다')