SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day34_자연어 처리를 위한 딥러닝 모델

자연어 처리 NLP Natural Language Processing

사람이 사용하는 언어를 컴퓨터가 이해하고 처리하도록 학습하는 인공지능 모델

https://standout.tistory.com/1822

자연어와 자연어처리란? , Natural Language, NLP Natural Language Processing (feat. 정형데이터와 비정형데이

 

 

자연어 처리에서 사용하는 대표적인 딥러닝 모델

현재 많이 사용하는 모델은 DNN, CNN, RNN, LSTM, GRU, Seq2Seq, Attention, Transformer, BERT, GPT가 있다. 

https://standout.tistory.com/1848

자연어 처리에서 사용하는 대표적인 딥러닝 모델: DNN, CNN , RNN , LSTM , GRU , Seq2Seq , Attention, Transforme

 

 

전체 모델들의 발전과정

DNN
 │
 ├── CNN
 │
 └── RNN
      │
      ├── LSTM
      │      │
      │      └── GRU
      │
      └── Seq2Seq
             │
             └── Attention
                    │
                    └── Transformer
                           ├── BERT (Encoder 기반)
                           └── GPT (Decoder 기반)

 

 

 

 

DNN

가장 기본적인 딥러닝 모델, 구조가 간단하고 현이 쉽지만 문장의 순서를 몰라 나는 밥을 먹었다.와 먹었다 나는 밥을.를 거의 동일하게 본다. 

https://standout.tistory.com/1537

딥러닝: Deep Neural Network (DNN), Convolutional Neural Network (CNN), Recurrent Neural Network (RNN)

 

 

 

CNN

이미지 분석용이나 자연어처리에도 많이 사용되는데 연속된 단어 패턴을 잘 찾는다. 

https://standout.tistory.com/1744

CNN 이란?: CNN Concolutional Neural Network 합성곱 신경망, 이미지의 특징을 자동으로 찾자!

 

 

 

RNN

문장의 순서를 기억해 앞의 정보를 기억한다. 하지만 긴 문장을 잘 기억하지 못한다. 

https://standout.tistory.com/1842

RNN 순환신경망의 한 종류, LSTM이란?: 장기 의존성 문제를 해결하기 위해 개발된 딥러닝 모델

 

 

 

LSTM

RNN의 단점을 필요한 정보만 기억하고 불필요한 정보를 삭제하는 등의 방법으로 단점을 해결했다. 

https://standout.tistory.com/1842

RNN 순환신경망의 한 종류, LSTM이란?: 장기 의존성 문제를 해결하기 위해 개발된 딥러닝 모델

 

 

 

 

GRU

LSTM을 단순화해 기존 게이트 3개를 2개로 줄여 메모리를 적게 사용하고 성능을 거의 비슷하게 했다. 

https://standout.tistory.com/1842

RNN 순환신경망의 한 종류, LSTM이란?: 장기 의존성 문제를 해결하기 위해 개발된 딥러닝 모델

 

 

 

Seq2Seq

문장을 다른 문장으로 바꾸는 번역모델으로 번역, 요약, 챗봇에 많이 사용한다.  안녕하세요.-> Hello.

https://standout.tistory.com/1844

입력과 출력의 길이가 달라도 처리할 수 있는 모델 Seq2Seq , 정보 병목(Information Bottleneck) (feat. Atte

 

 

 

Attension

Seq2Seq의 단점을 해결해 문장이 길어지면 중간 벡터 하나만으로 부족함으로 필요한 단어를 다시 보며 긴 문장에서도 성능이 향상되도록 했다. 

https://standout.tistory.com/1844

입력과 출력의 길이가 달라도 처리할 수 있는 모델 Seq2Seq , 정보 병목(Information Bottleneck) (feat. Atte

 

 

 

 

Transformer

현재 자연어 처리의 핵심 모델으로 RNN을 사용하지않고 Attension만 사용한다.

https://standout.tistory.com/1846

Attention 메커니즘만을 이용하여 문장을 처리하는 딥러닝 모델 Transformer (feat.Self-Attention, Multi-Head A

 

 

 

BERT

Transformer의 Encoder 구조만 사용하는 모델로양쪽 방향을 동시에 읽는다. 

https://standout.tistory.com/1847

앞 뒤를 모두 참고하는 Bidirectional 모델 BERT ,특수 토큰 CLS와 SEP , 두가지 학습방법 MLM, NSP (feat. Tra

 

 

GPT

Transformer의 Decoder 구조를 사용하는 생성형 언어 모델

 

 

 

 

 

오늘 자연어 처리를 위한 CNN과 RNN을 해보자. 

 

CNN

대상에 대한 특징을 인식하는 방식을 모델링해 이미지 학습을 위한 효율적인 알고리즘, 

Convolution과 Pooling, Fully Connected Layer로 여러가지 사물중 해당 항목에 대한 확률을 출력한다. 

https://standout.tistory.com/1744

CNN 이란?: CNN Concolutional Neural Network 합성곱 신경망, 이미지의 특징을 자동으로 찾자!

 

 

 

CNN을 기반으로 이미지가 가지고 있는 특징을 추출하고 앞에서 추출한 특징을 사용하여 새로운 이미지로 식별이 가능하다.

입력 데이터 주변을 특정 값으로 채워출력크기를 조정하기 위한 Padding과 이미지 이동 간격 설정하는 Stride를 함께 수행해 다른 이미지와의 일치 여부를 확인하는 Convolution 과정을 거친다. 

Max Pooling 최댓값을 추출해 ( 혹은 평균 풀링으로 평균값을 선택할 수 있다. Average Pooling)

Flattening 1차원 구조로 변환해 완전연결층으로 연결해 다시 학습한다. 

Softmax 어떤 분류에 속하는지 판단한다.

이 Convolution은 깊어질수록 더 복잡한 특징을 추출가능하고 고차원의 잠재적인 특징을 학습가능하다. 

 

 

 

RNN

문장의 순서를 기억해 앞의 정보를 기억한다. 하지만 긴 문장을 잘 기억하지 못한다. 

https://standout.tistory.com/1842

RNN 순환신경망의 한 종류, LSTM이란?: 장기 의존성 문제를 해결하기 위해 개발된 딥러닝 모델

 

 

RNN

torchtext를 이용해 field 객체를 지정한다.

torchtext를 이용해 IMDB 데이터 세트를 로드해 text_field과 label_field에 할당하고 .splits함수를 이용하여 train, test 데이터세트로 분리한다. 

build_vocab 함수를 이용해 Vocabulary를 생성하고 Bucketlterator클래스를 이용해 데이터 로더를 구현한다 .

 

 

외

pyTorch의 High Level 인터페이스를 제공하는 오픈소스 Python라이브러리 pyTorch Lightning 로 Traing, testind를 보다 간결한 코드로 사용할 수 있도록 구현된 프레임워크에 LightningModule 클래스를 상속받아 사용할 수 있다 .

 

 

 

 

 

 

직접 실습해보자. 

 

nlp_model 프로젝트 실습

python -m pip install --upgrade pip setuptools wheel
pip install -r requirements.txt
python src/cnn_spam_classifier.py

 

 

requirements.txt

numpy>=1.26
pandas>=2.1
matplotlib>=3.8
scikit-learn>=1.3

torch>=2.2
torchvision>=0.17
torchaudio>=2.2

pytorch-lightning>=2.2
torchmetrics>=1.3

torchtext>=0.17

 

 

시드고정

def set_seed(seed: int = 42) -> None:
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed(seed)

 

토큰화하기

소문자로 변경한뒤 text를 split해 토큰화한다. 

def simple_tokenize(text: str) -> list[str]:
    text = str(text).lower()
    tokens = text.split()
    return tokens

 

 

build_vocab

text를 가져오되 min_freq는 1로 최소 설정했다. 

dict형의 counter와 함께 tuple로 생성한다.  return이 여러개를 return하면 자동 tuple이 된다. 

앞서 정의한 함수 simple_tokenize() 를 통해 문장을 토큰화하고 0에는 패딩토큰, 1에는 미등록 단어 토큰이 들어가도록 세팅,

for문을 돌려 min_freq이상 등록된 단어를 사전에 포함하고 아닐경우에는 len() 이후 숫자를 ++해 추가하도록 한다. 

def build_vocab(texts: pd.Series, min_freq: int = 1) -> tuple[dict[str, int], Counter]:
    counter = Counter()
    for text in texts:
        counter.update(simple_tokenize(text)) # 문장토큰화하고 counter에 빈도누적
    word_to_index = {"<PAD>": 0} # 0 은 패팅코큰
    word_to_index["<UNK>"] = 1 # 1 은 미등록 단어 토큰

    for word, freq in counter.most_common(): # min_freq이상 등장한 단어를 사전에 포함한다.
        if freq >= min_freq:
            if word not in word_to_index:
                word_to_index[word] = len(word_to_index) # 사전에 없다면 새로운 번호를 부여해 추가.
    return word_to_index, counter

 

 

 

 각 단어를 받아 사전번호로 세팅하고, max_len보다 길면 자르고 짧으면 PAD를 추가해 형태를 맞춘다 .

def encode_and_pad(text:str, word_to_index: dict[str, int], max_len: int) -> list[int]:
    tokens = simple_tokenize(text)
    encoded = [word_to_index.get(token, word_to_index["<UNK>"]) for token in tokens]
    encodeed = encoded[:max_len]
    padded = encoded + [0] * [word_to_index["<PAD>"]] * (max_len - len(encoded))
    return padded

 

 

 

SMSDataset 생성자.

생성자 init 생성. 문장 text 를 list로 변환해 저장, label을 astype해서 정수 list로 변환해서 저장

word_to_index 사전 초기화

max_len 초기화

len 함수, textx의 len을 반환

__getitem__ 게터 설정 x에 정수시퀀스로 변환하고 패딩처리.

y에는 labels.

x_tensor는 long형으로 변환, y_tensor는 float32의 tensor형으로 변환

# SMS 문장과 레이블(ham | spam, 정답)을  Torch DataLoader 가 읽을 수 있도록 구성하는 Dataset 클래스 정의
class SMSDataset(Dataset):
    # 생성자가 해당 클래스 객체 생성시, 문장, 레이블, 단어 사전, 최대 길이를 전달받도록 함
    def __init__(self, texts: pd.Series, labels: pd.Series, word_to_index: dict[str, int], max_len: int):
        # 입력 문장을 리스트로 변환하고 저장
        self.texts = list(texts)

        # 정답 레이블도 정수 리스트로 변환하고 저장
        self.labels = list(labels.astype(int))

        # 단어를 정수 번호로 바꾸기 위한 사전 저장
        self.word_to_index = word_to_index

        # 모든 문장을 맞출 최대 길이도 저장
        self.max_len = max_len

    # Dataset 의 전체 샘플 개수 반환하는 메소드(멤버함수 : 클래스에 소속된 함수) 정의
    def __len__(self) -> int:
        # 저장된 문장의 갯수를 반환
        return len(self.texts)

    # 특정 인덱스 하나에 해당하는 입력 텐서와 정답 텐서를 반환하는 메소드 정의
    def __getitem__(self, index: int) -> tuple[torch.Tensor, torch.Tensor]:
        # index 위치의 문장을 정수 시퀀스로 변환하고 패딩 처리함
        x = encode_and_pad(self.texts[index], self.word_to_index, self.max_len)

        # index 위치의 정답 레이블도 꺼냄
        y = self.labels[index]

        # 입력 문장은 정수 시퀀스이므로, LongTensor 로 변환함
        # Embedding 계층은 정수 인덱스를 입력으로 받기 때문임
        x_tensor = torch.tensor(x, dtype=torch.long)

        # 이진 분류 손실함수 BCEWithLogitsLoss 에 맞추기 위해 정답 레이블 float 텐서로 변환함
        # logits (점수) : 0.0123 (0에 가까움), 0.987 (1에 가까움)
        y_tensor = torch.tensor(y, dtype=torch.float32)

        # 입력 텐서와 정답 텐서를 반환함
        return x_tensor, y_tensor
# ------------------------------------------------

 

 

 

모델설계

초기화. 임베딩 32, 필터 32, 커널사이즈 5, 드롭아웃 0.3 설정

임베딩 계층 선언 padding_idx는 0 으로 설정해 0으로 값 채우도록. 

과적합을 줄이기 위해 nn.Dropout.

conv1d 설정 nn.Conv1d 1차원 합성곱 계층. relu, dropout, nn.leanear로 스팸일 가능성을 나타내는 로짓 1개를 출력, 

forward 입력 데이터가 모델 내부에서 어떤 순서로 계산됬는지를 정의한다.  embedding, dropout, permute로 차원순서변경, conv1d, relu, torch.max, dropout2, fc로 로짓계산해 logits.squeeze 형태를 받기위해 마지막차원 제거

# 모델 설계
# Embedding - Conv1D - GlobalMaxPooling - Linear 구조의 CNN 문장 분류 모델
class TextCNN(nn.Module):
    # 모델 계층 초기화
    def __init__(self,
                 vocab_size: int,
                 embedding_dim: int = 32,
                 num_filters: int = 32,
                 kernel_size: int = 5,
                 dropout_ratio: float = 0.3):
        # 부모 클래스 생성자 호출  (후손 생성자 안에서만 호출할 수 있음, 생성자 앉 첫줄에 기입할 것)
        super().__init__()

        # 신경망 계층에 대한 설정 선언
        # 1. Embedding 계층
        self.embedding = nn.Embedding(
            num_embeddings=vocab_size,
            embedding_dim=embedding_dim,
            padding_idx=0
        )

        # 2. 과적합 줄이기 위해 Dropout 계층
        self.dropout1 = nn.Dropout(p=dropout_ratio)

        # 3. 1차원 합성곱 계층 : 문장 안의 연속된 단어 패턴을 추출
        self.conv1d = nn.Conv1d(
            in_channels=embedding_dim,
            out_channels=num_filters,
            kernel_size=kernel_size,
        )

        # ReLU 활성화함수 : 음수를 0으로 바꿈 => 양수 특징만 통과시킴
        self.relu = nn.ReLU()

        # 두 번째 Dropout 계층
        self.dropout2 = nn.Dropout(p=dropout_ratio)

        # 출력층 : 스팸일 가능성을 나타내는 로짓 1개 출력하는 완전 연결층
        self.fc = nn.Linear(num_filters, 1)

    # 입력 데이터가 모델 내부에서 어떤 순서로 계산되는지 정의합니다.
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # 입력 x의 형태는 [배치 크기, 문장 길이]입니다.
        # Embedding 후 형태는 [배치 크기, 문장 길이, 임베딩 차원]입니다.
        x = self.embedding(x)
        # 임베딩 결과에 Dropout을 적용합니다.
        x = self.dropout1(x)
        # Conv1D는 [배치 크기, 채널 수, 길이] 형태를 입력으로 받으므로 차원 순서를 바꿉니다.
        x = x.permute(0, 2, 1)
        # 1차원 합성곱을 적용하여 문장의 지역적 단어 패턴을 추출합니다.
        x = self.conv1d(x)
        # ReLU 활성화 함수를 적용하여 중요한 양수 특징을 남깁니다.
        x = self.relu(x)
        # 시간축, 즉 문장 길이 방향 전체에서 가장 큰 값을 선택하여 Global Max Pooling을 수행합니다.
        x = torch.max(x, dim=2).values
        # 풀링된 특징에 Dropout을 적용합니다.
        x = self.dropout2(x)
        # 완전 연결층을 통과시켜 각 샘플마다 로짓 1개를 계산합니다.
        logits = self.fc(x)
        # BCEWithLogitsLoss가 [배치 크기] 형태를 받기 쉽도록 마지막 차원을 제거합니다.
        logits = logits.squeeze(1)
        # sigmoid 적용 전의 로짓 값을 반환합니다.
        return logits

    # 한 epoch 동안 모델을 학습하는 함수입니다.

 

 

 

 

학습 .model.train() 학습모드로 전환. 손실합계 저장할 변수들 초기화.

DataLoader에서 미니배치 가져와 zero_grad 계사니된 기울기 초기화, model() 로짓계산. 

criterion 손실계산, loss.backward() 기울기계산. optimizer.step() 바니영

평균 손실계산, 정확도 계산, 반환

def train_one_epoch(model: nn.Module, data_loader: DataLoader, criterion: nn.Module,
                    optimizer: torch.optim.Optimizer, device: torch.device) -> tuple[float, float]:
    # 모델을 학습 모드로 전환합니다. Dropout이 활성화됩니다.
    model.train()
    # 전체 손실 합계를 저장할 변수를 초기화합니다.
    total_loss = 0.0
    # 맞게 예측한 샘플 수를 저장할 변수를 초기화합니다.
    total_correct = 0
    # 전체 샘플 수를 저장할 변수를 초기화합니다.
    total_count = 0
    # DataLoader에서 미니배치를 하나씩 가져옵니다.
    for x_batch, y_batch in data_loader:
        # 입력 배치를 CPU 또는 GPU 장치로 이동합니다.
        x_batch = x_batch.to(device)
        # 정답 배치를 CPU 또는 GPU 장치로 이동합니다.
        y_batch = y_batch.to(device)
        # 이전 미니배치에서 계산된 기울기를 초기화합니다.
        optimizer.zero_grad()
        # 모델에 입력 배치를 넣어 로짓을 계산합니다.
        logits = model(x_batch)
        # 로짓과 정답 레이블을 이용해 이진 분류 손실을 계산합니다.
        loss = criterion(logits, y_batch)
        # 손실을 기준으로 역전파를 수행하여 각 파라미터의 기울기를 계산합니다.
        loss.backward()
        # 계산된 기울기를 이용해 모델 파라미터를 업데이트합니다.
        optimizer.step()
        # 현재 미니배치의 손실에 샘플 수를 곱해 누적합니다.
        total_loss += loss.item() * x_batch.size(0)
        # 로짓에 sigmoid를 적용해 스팸일 확률로 변환합니다.
        probs = torch.sigmoid(logits)
        # 확률이 0.5 이상이면 스팸(1), 아니면 정상(0)으로 예측합니다.
        preds = (probs >= 0.5).float()
        # 예측값과 정답이 같은 샘플 수를 누적합니다.
        total_correct += (preds == y_batch).sum().item()
        # 현재 미니배치 샘플 수를 전체 샘플 수에 누적합니다.
        total_count += x_batch.size(0)
    # 전체 평균 손실을 계산합니다.
    avg_loss = total_loss / total_count
    # 전체 정확도를 계산합니다.
    accuracy = total_correct / total_count
    # 평균 손실과 정확도를 반환합니다.
    return avg_loss, accuracy

 

 

 

검증함수.

model.eval() 평가모드 전환

손실합계 저장할 변수 초기화

torch.no_grad() 메모리 사용량 줄이기dataloader에서 미니배치 하나씩 가져오기

torch.sigmoid()  로짓 확률로 변환.

확률이 0.5이상이면 스팸으로 예측. 

누적해 평균손실, 정확도 반환.

# 검증 또는 테스트 데이터를 이용해 모델을 평가하는 함수입니다.
def evaluate(model: nn.Module, data_loader: DataLoader, criterion: nn.Module, device: torch.device) -> tuple[
    float, float]:
    # 모델을 평가 모드로 전환합니다. Dropout이 비활성화됩니다.
    model.eval()
    # 전체 손실 합계를 저장할 변수를 초기화합니다.
    total_loss = 0.0
    # 맞게 예측한 샘플 수를 저장할 변수를 초기화합니다.
    total_correct = 0
    # 전체 샘플 수를 저장할 변수를 초기화합니다.
    total_count = 0
    # 평가 단계에서는 기울기 계산이 필요 없으므로 no_grad를 사용해 메모리 사용량을 줄입니다.
    with torch.no_grad():
        # DataLoader에서 미니배치를 하나씩 가져옵니다.
        for x_batch, y_batch in data_loader:
            # 입력 배치를 CPU 또는 GPU 장치로 이동합니다.
            x_batch = x_batch.to(device)
            # 정답 배치를 CPU 또는 GPU 장치로 이동합니다.
            y_batch = y_batch.to(device)
            # 모델에 입력 배치를 넣어 로짓을 계산합니다.
            logits = model(x_batch)
            # 로짓과 정답 레이블을 이용해 손실을 계산합니다.
            loss = criterion(logits, y_batch)
            # 현재 미니배치 손실을 누적합니다.
            total_loss += loss.item() * x_batch.size(0)
            # 로짓을 확률로 변환합니다.
            probs = torch.sigmoid(logits)
            # 확률이 0.5 이상이면 스팸으로 예측합니다.
            preds = (probs >= 0.5).float()
            # 정답과 일치한 예측 개수를 누적합니다.
            total_correct += (preds == y_batch).sum().item()
            # 전체 샘플 수를 누적합니다.
            total_count += x_batch.size(0)
    # 평균 손실을 계산합니다.
    avg_loss = total_loss / total_count
    # 정확도를 계산합니다.
    accuracy = total_correct / total_count
    # 평균 손실과 정확도를 반환합니다.
    return avg_loss, accuracy

 

 

 

 

 

main()함수

난수 고정. 

gpu cpu 사용

데이터셋 url지정해 csv 파일에 저장, 파일이 없으면 인터넷에서 내려받는다. latin1인코딩 방식으로 읽으며 분석에 필요없는 열들을 삭제한다. ham은 0 spam은 1로 변환한다. 

본문이 중복되어있으면 하나만 남기고 제거한다. 

입력데이터와 정답 레이블 데이터를 분리한다. 

검증데이터도.훈련데이터문장으로만 단어 사전을 만든다. 

전체 단어수를 계싼한다. 희귀단어기준은 2로 설정. 등장빈도가 thresholde보다 작은 단어수를 계산해 합계 계산. 

전체 문장 길이 분포를 히스토그램으로 시각화한다. 

max_len은 189, 학습할미니배치 크기는 64, 훈련, 검증data객체 생성SMSDataset(), CNN TextCNN() 텍스트 분류 모델 생성

BCEWithLogitsLoss() , 옵티마이저 생성

가장 좋은 모델을 저장할 경로지정, 

for문으로 rpoch수만큼 학습 train_one_epoch(), 그리고 평가 evaluate()

best_val_acc 성능이 더 좋으면 저장

# 전체 데이터 준비, 모델 생성, 학습, 평가 흐름을 실행하는 메인 함수입니다.
def main() -> None:
    # 난수 시드를 고정하여 실행할 때마다 최대한 비슷한 결과가 나오게 합니다.
    set_seed(42)

    # GPU가 사용 가능하면 cuda를 사용하고, 아니면 cpu를 사용합니다.
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

    # 현재 사용하는 장치를 출력합니다.sm
    print(f"사용 장치: {device}")

    # SMS 스팸 분류 데이터셋 URL을 지정합니다.
    dataset_url = "https://raw.githubusercontent.com/mohitgupta-omg/Kaggle-SMS-Spam-Collection-Dataset-/master/spam.csv"

    # 로컬에 저장할 CSV 파일명을 지정합니다.
    os.makedirs("../data", exist_ok=True)
    dataset_path = "./data/spam.csv"

    # spam.csv 파일이 없으면 인터넷에서 내려받습니다.
    if not os.path.exists(dataset_path):
        # 지정한 URL에서 CSV 파일을 내려받아 spam.csv로 저장합니다.
        urllib.request.urlretrieve(dataset_url, filename=dataset_path)

    # CSV 파일을 latin1 인코딩으로 읽습니다.
    data = pd.read_csv(dataset_path, encoding="latin1")

    # 전체 샘플 수를 출력합니다.
    print("총 샘플의 수 :", len(data))

    # 분석에 필요하지 않은 열들을 삭제합니다.
    data = data.drop(columns=["Unnamed: 2", "Unnamed: 3", "Unnamed: 4"])

    # ham은 0, spam은 1로 변환합니다.
    data["v1"] = data["v1"].replace(["ham", "spam"], [0, 1])

    # 결측값 여부를 출력합니다.
    print("결측값 여부 :", data.isnull().values.any())

    # 메일 본문 v2 열의 고유 문장 수를 출력합니다.
    print("v2열의 유니크한 값 :", data["v2"].nunique())

    # 같은 메일 본문이 중복되어 있으면 하나만 남기고 제거합니다.
    data = data.drop_duplicates(subset=["v2"]).reset_index(drop=True)

    # 중복 제거 후 전체 샘플 수를 출력합니다.
    print("중복 제거 후 총 샘플의 수 :", len(data))

    # 레이블별 개수를 출력합니다.
    print(data.groupby("v1").size().reset_index(name="count"))

    # 정상 메일 비율을 출력합니다.
    print(f'정상 메일의 비율 = {round(data["v1"].value_counts()[0] / len(data) * 100, 3)}%')

    # 스팸 메일 비율을 출력합니다.
    print(f'스팸 메일의 비율 = {round(data["v1"].value_counts()[1] / len(data) * 100, 3)}%')

    # 입력 문장 데이터와 정답 레이블 데이터를 분리합니다.
    X_data = data["v2"]
    y_data = data["v1"]

    # 훈련+검증 데이터 80%, 테스트 데이터 20%로 분리합니다.
    X_train_val, X_test, y_train_val, y_test = train_test_split(
        X_data,
        y_data,
        test_size=0.2,
        random_state=0,
        stratify=y_data,
    )

    # 훈련+검증 데이터 중 20%를 검증 데이터로 분리합니다.
    X_train, X_val, y_train, y_val = train_test_split(
        X_train_val,
        y_train_val,
        test_size=0.2,
        random_state=0,
        stratify=y_train_val,
    )

    # 훈련 데이터 문장으로만 단어 사전을 만듭니다.
    word_to_index, word_counter = build_vocab(X_train, min_freq=1)

    # 전체 단어 수를 계산합니다.
    total_cnt = len(word_counter)

    # 희귀 단어 기준을 2로 설정합니다.
    threshold = 2

    # 등장 빈도가 threshold보다 작은 단어 수를 계산합니다.
    rare_cnt = sum(1 for _, freq in word_counter.items() if freq < threshold)

    # 전체 단어 등장 횟수 합계를 계산합니다.
    total_freq = sum(word_counter.values())

    # 희귀 단어 등장 횟수 합계를 계산합니다.
    rare_freq = sum(freq for _, freq in word_counter.items() if freq < threshold)

    # 희귀 단어 통계를 출력합니다.
    print("등장 빈도가 %s번 이하인 희귀 단어의 수: %s" % (threshold - 1, rare_cnt))
    print("단어 집합(vocabulary)에서 희귀 단어의 비율:", (rare_cnt / total_cnt) * 100)
    print("전체 등장 빈도에서 희귀 단어 등장 빈도 비율:", (rare_freq / total_freq) * 100)

    # <PAD>와 <UNK>까지 포함한 최종 단어 집합 크기를 계산합니다.
    vocab_size = len(word_to_index)

    # 단어 집합 크기를 출력합니다.
    print("단어 집합의 크기: {}".format(vocab_size))

    # 훈련 문장의 토큰 길이 목록을 계산합니다.
    train_lengths = [len(simple_tokenize(text)) for text in X_train]

    # 훈련 문장 중 가장 긴 문장의 길이를 출력합니다.
    print("메일의 최대 길이 : %d" % max(train_lengths))

    # 훈련 문장의 평균 길이를 출력합니다.
    print("메일의 평균 길이 : %f" % (sum(train_lengths) / len(train_lengths)))

    # 전체 문장 길이 분포를 히스토그램으로 표시합니다.
    plt.hist([len(simple_tokenize(text)) for text in X_data], bins=50)
    plt.xlabel("length of samples")
    plt.ylabel("number of samples")
    plt.show()

    # 원본 노트북과 동일하게 문장의 최대 길이를 189로 지정합니다.
    max_len = 189

    # 한 번에 학습할 미니배치 크기를 지정합니다.
    batch_size = 64

    # 훈련 Dataset 객체를 생성합니다.
    train_dataset = SMSDataset(X_train, y_train, word_to_index, max_len)

    # 검증 Dataset 객체를 생성합니다.
    val_dataset = SMSDataset(X_val, y_val, word_to_index, max_len)

    # 테스트 Dataset 객체를 생성합니다.
    test_dataset = SMSDataset(X_test, y_test, word_to_index, max_len)

    # 훈련 DataLoader를 생성합니다. shuffle=True는 매 epoch마다 데이터 순서를 섞습니다.
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

    # 검증 DataLoader를 생성합니다.
    val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False)

    # 테스트 DataLoader를 생성합니다.
    test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

    # PyTorch CNN 텍스트 분류 모델을 생성합니다.
    model = TextCNN(
        vocab_size=vocab_size,
        embedding_dim=32,
        num_filters=32,
        kernel_size=5,
        dropout_ratio=0.3,
    )

    # 모델을 CPU 또는 GPU 장치로 이동합니다.
    model = model.to(device)
    # 모델 구조를 출력합니다.
    print(model)
    # 이진 분류에서 sigmoid와 binary cross entropy를 함께 안정적으로 계산하는 손실 함수를 사용합니다.
    criterion = nn.BCEWithLogitsLoss()
    # Adam 옵티마이저를 생성합니다.
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    # 최대 학습 epoch 수를 지정합니다.
    num_epochs = 10
    # 검증 정확도가 가장 좋을 때의 값을 저장합니다.
    best_val_acc = 0.0
    # 가장 좋은 모델을 저장할 경로를 지정합니다.
    best_model_path = "./model/best_torch_textcnn.pt"

    # 지정한 epoch 수만큼 학습을 반복합니다.
    for epoch in range(1, num_epochs + 1):
        # 한 epoch 동안 훈련 데이터를 이용해 모델을 학습합니다.
        train_loss, train_acc = train_one_epoch(model, train_loader, criterion, optimizer, device)
        # 검증 데이터를 이용해 현재 모델 성능을 평가합니다.
        val_loss, val_acc = evaluate(model, val_loader, criterion, device)
        # 현재 epoch의 학습 결과를 출력합니다.
        print(
            f"Epoch {epoch:02d}/{num_epochs} | "
            f"train_loss={train_loss:.4f}, train_acc={train_acc:.4f} | "
            f"val_loss={val_loss:.4f}, val_acc={val_acc:.4f}"
        )

        # 검증 정확도가 이전 최고 성능보다 좋으면 모델을 저장합니다.
        if val_acc > best_val_acc:
            # 최고 검증 정확도를 갱신합니다.
            best_val_acc = val_acc
            # 현재 모델의 파라미터를 파일로 저장합니다.
            torch.save(model.state_dict(), best_model_path)
            # 모델이 저장되었음을 출력합니다.
            print(f"최고 검증 정확도 갱신: {best_val_acc:.4f}, 모델 저장: {best_model_path}")

    # 저장된 최고 성능 모델 파일이 있으면 불러옵니다.
    if os.path.exists(best_model_path):
        # 저장된 모델 파라미터를 현재 모델에 적용합니다.
        model.load_state_dict(torch.load(best_model_path, map_location=device))
    # 테스트 데이터로 최종 모델 성능을 평가합니다.
    test_loss, test_acc = evaluate(model, test_loader, criterion, device)
    # 최종 테스트 손실과 정확도를 출력합니다.
    print(f"\n테스트 손실: {test_loss:.4f}")
    print(f"테스트 정확도: {test_acc:.4f}")

# 이 파일을 직접 실행할 때만 main() 함수를 실행합니다.
if __name__ == "__main__":
    main()

 

 

 

 

 

 

수령받은 lstm_imdb_lightning.py 코드도 분석해보자. 

다만 실행중 에러가났고 아쉽게도 2022년.. 한 4년도정도 된 오래된 코드임을 알게됬다. 

python 3.9를 다운하고 돌려봤지만 여러 환경이 충돌될수있다.. 그냥 코드분석에만 집중해보자.

https://www.python.org/downloads/release/python-390/
3.9다운 - Windows x86-64 executable installer
설치.

차례대로 터미널에 명령.
deactivate
py -0p 
버전 3.9 잘 설치됫는지 확인

내문서 - 파일경로 열어서 .venv 삭제

py -3.9 -m venv .venv
.\.venv\Scripts\Activate.ps1

pip install torch==1.9.1
pip install torchtext==0.10.1
pip install pytorch-lightning==1.5.10

 

 

 

 

import

torch.nn 신경망 계층을 만듬,

torch.nn.functional 활성화함수나 손실함수 등을 함수형태로 사용할 수 있게 한다. 

pytorch_lighting Pytorch 학습 코드를 더 구조적으로 작성하게해주는 라이브러리로 반복 학습문, 검증과 gpu 설정등을 간결하게 관리할 수 있다. 

torchtext.legacy.data 텍스트 데이터와 라벨 데이터를 전처리 하기 위한 객체. 텍스트를 토큰화, 길이를 맞추고 , vocab를 만드는데 사용한다 .Field

torchtext.legacy,data BucketIterator 문장 길이가 비슷한 데이터끼리 배치로 묶어준다 . 문장 차이가 크면 RNN LSTM 계열에서 패딩이 많이 발생하기에 비슷한 길이끼리 묶으면 학습효율을 높일 수 있다 .

torch.text.legacy.datasets IMDB 영화리뷰 분석 데이터셋. 리뷰 문장과 긍정/부정 라벨로 구성되어있다. 

# torch는 PyTorch의 핵심 라이브러리입니다.
# 텐서 연산, GPU 사용, 딥러닝 모델 학습 등에 사용됩니다.
import torch

# torch.nn은 신경망 계층을 만들 때 사용하는 모듈입니다.
# LSTM, Linear, CrossEntropyLoss 같은 딥러닝 구성요소를 제공합니다.
import torch.nn as nn

# torch.nn.functional은 활성화 함수나 손실 함수 등을 함수 형태로 사용할 수 있게 해 줍니다.
# 여기서는 ELU 활성화 함수 F.elu()를 사용합니다.
import torch.nn.functional as F

# Adam은 딥러닝 모델의 가중치를 업데이트하는 최적화 알고리즘입니다.
# 학습 과정에서 손실값을 줄이는 방향으로 파라미터를 조정합니다.
from torch.optim import Adam

# PyTorch Lightning은 PyTorch 학습 코드를 더 구조적으로 작성하게 해 주는 라이브러리입니다.
# 반복 학습문, 검증 과정, GPU 설정 등을 간결하게 관리할 수 있습니다.
import pytorch_lightning as pl

# Field는 torchtext.legacy에서 텍스트 데이터와 라벨 데이터를 전처리하기 위한 객체입니다.
# 텍스트를 토큰화하고, 길이를 맞추고, vocabulary를 만드는 데 사용됩니다.
from torchtext.legacy.data import Field

# BucketIterator는 문장 길이가 비슷한 데이터끼리 배치로 묶어 주는 데이터 로더입니다.
# RNN/LSTM 계열 모델에서는 문장 길이 차이가 크면 패딩이 많이 생기므로,
# 비슷한 길이끼리 묶으면 학습 효율을 높일 수 있습니다.
from torchtext.legacy.data import BucketIterator

# IMDB는 torchtext.legacy.datasets에서 제공하는 영화 리뷰 감성 분석 데이터셋입니다.
# 리뷰 문장과 긍정/부정 라벨로 구성되어 있습니다.
from torchtext.legacy.datasets import IMDB

 

 

 

데이터 전처리 Field 객체 생성. 

sequential T 단어들이 순서대로 나열된 문장 데이터. 

include_length T 각 문장의 실제 길이 정보를 함께 저장한다. 

fix_length = 200 모든 문장을 최대 200개 토큰길이로 맞춘다. 길이가 200보다 짧은 문장은 패딩, 200보다 긴 문장은 잘린다. 

lebel 데이터들은 pos 또는 neg 하나의 값만 사용하여 순서가 있는 데이터가 아니라 단일값이다 .

# fix_length=200은 모든 문장을 최대 200개 토큰 길이로 맞추겠다는 뜻입니다.
# 길이가 200보다 짧은 문장은 패딩되고, 200보다 긴 문장은 잘릴 수 있습니다.
text_field = Field(
    sequential=True,
    include_lengths=True,
    fix_length=200
)

label_field = Field(
    sequential=False
)

 

 

 

 

IMDB.splits() 데이터셋을 나누어 불러온다. 

train, test = IMDB.splits(
    text_field,
    label_field
)

 

 

vocab 설정

build_vocab() 는 훈련 데이터에 등장한 단어들을 모아 단어 사전을 생성한다. 

vectors='fasttext.simple.300d' 사전학습된 FastText 300 차원 임베딩 벡터를 사용하겠다는 의미. 

text_field.build_vocab(
    train,
    vectors="fasttext.simple.300d"
)

label_field.build_vocab(train)

 

 

 

 

cuda gpu cpu 설정

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

 

 

batch 사이즈 설정 및 BucketIterator.splits() 를 사용해 train과 test 를 배치 단위로 공급하는 반복자를 만든다.

batch_size = 32

# BucketIterator.splits()는 train과 test 데이터셋을 배치 단위로 공급하는 반복자를 만듭니다.
# train_iter는 훈련용 데이터 로더이고, test_iter는 검증 또는 평가용 데이터 로더입니다.
# device=device는 배치 데이터를 CPU 또는 GPU에 올려서 사용하겠다는 의미입니다.
train_iter, test_iter = BucketIterator.splits(
    (train, test),
    batch_size=batch_size,
    device=device
)

 

 

 

 

LSTM 모델 클래스 정의

RNNModel 클래스만들기

init() 생성자 함수. 

output size는 IMDB는 긍정/부정으로 2개 클래스 임으로 실제 환견에서는 2가 더 자연스럽다. 

 nn.LSTM 객체 만들기. nn.linear 각 클래스에 대한 점수 출력, nn.CrossEntropuLoss() 손실함수, 

pl.metrics.Accuracy()  Pytorch Lightning에서 정확도 계산 객체 생성.  에러발생시 Accracy 사용. 

forward 모델의 순전파를 정의. 임베딩, LSTM 활성화 함수, 선형층을 거쳐 예측값을 반환한다. 

training_step() 배치 하나에 수행할 연산을 정의한다. 모델 입력 형태에 맞춰 self() 즉 forward()

정확도 계산. 

log 기록. prog_bar T 진행률 표시줄에 train_acc 값을 보여준다. 

validation_step() 검증 배치 하나에 대해 수행할 연산 정의. 모델 학습 중 검증 데이터에 대한 성능을 확인할때 사용

입력 문장과 정답 라벨 분리, self(), 손실계산, 정확도계산 log기록

데이터로더 반환, 옵티마이저 반환

# RNNModel 클래스는 PyTorch Lightning의 LightningModule을 상속합니다.
# LightningModule을 사용하면 모델 구조, 학습 단계, 검증 단계, 최적화 설정을 한 클래스 안에 정리할 수 있습니다.
class RNNModel(pl.LightningModule):

    # __init__()은 모델 객체가 생성될 때 한 번 실행되는 생성자 함수입니다.
    # embedding은 text_field.vocab.vectors로부터 전달되는 사전 학습 단어 임베딩 행렬입니다.
    # lstm_input_size=300은 FastText 임베딩 차원이 300차원이기 때문에 입력 크기를 300으로 설정한 것입니다.
    # lstm_hidden_size=100은 LSTM이 내부적으로 저장하는 은닉 상태 벡터 크기입니다.
    # output_size=3은 출력 클래스 개수입니다.
    # 단, IMDB는 보통 긍정/부정 2개 클래스이므로 실제 환경에서는 output_size=2가 더 자연스럽습니다.
    def __init__(
        self,
        embedding,
        lstm_input_size=300,
        lstm_hidden_size=100,
        output_size=3
    ):
        # 부모 클래스인 pl.LightningModule의 초기화 기능을 실행합니다.
        # Lightning 내부 기능을 정상적으로 사용하기 위해 반드시 호출해야 합니다.
        super().__init__()

        # 사전 학습된 임베딩 벡터 행렬을 모델의 멤버 변수로 저장합니다.
        # 이 코드는 nn.Embedding 레이어가 아니라 이미 만들어진 벡터 행렬을 직접 인덱싱하는 방식입니다.
        self.embedding = embedding

        # nn.LSTM은 순서가 있는 데이터, 즉 문장이나 시계열 데이터를 처리하는 계층입니다.
        # 입력 크기는 300차원 임베딩 벡터이고, 출력 은닉 상태 크기는 100입니다.
        self.lstm = nn.LSTM(
            lstm_input_size,
            lstm_hidden_size
        )

        # nn.Linear는 LSTM에서 나온 100차원 특징을 클래스 개수만큼의 출력값으로 변환합니다.
        # 최종적으로 각 클래스에 대한 점수(logit)를 출력합니다.
        self.lin = nn.Linear(
            lstm_hidden_size,
            output_size
        )

        # CrossEntropyLoss는 다중 클래스 분류에서 많이 사용하는 손실 함수입니다.
        # 모델이 예측한 클래스 점수와 실제 정답 라벨의 차이를 계산합니다.
        self.loss_function = nn.CrossEntropyLoss()

        # PyTorch Lightning 버전에 따라 pl.metrics.Accuracy()가 동작하지 않을 수 있습니다.
        # 최신 환경에서는 torchmetrics.Accuracy 사용이 권장됩니다.
        try:
            # 구버전 PyTorch Lightning에서 정확도 계산 객체를 생성합니다.
            self.train_accuracy = pl.metrics.Accuracy()

            # 검증 정확도 계산 객체를 생성합니다.
            self.val_accuracy = pl.metrics.Accuracy()
        except AttributeError:
            # 최신 PyTorch Lightning에서는 pl.metrics가 제거되었을 수 있으므로 torchmetrics를 사용합니다.
            from torchmetrics import Accuracy

            # 다중 클래스 정확도 계산 객체를 생성합니다.
            self.train_accuracy = Accuracy(task="multiclass", num_classes=output_size)

            # 검증 정확도 계산 객체를 생성합니다.
            self.val_accuracy = Accuracy(task="multiclass", num_classes=output_size)

    # forward()는 모델의 순전파를 정의하는 함수입니다.
    # 입력 데이터 X가 들어오면 임베딩, LSTM, 활성화 함수, 선형층을 거쳐 예측값을 반환합니다.
    def forward(self, X: torch.Tensor):

        x = self.embedding[X].to(self.device).permute(1, 0, 2)

        x, _ = self.lstm(x)

        x = F.elu(x.permute(1, 0, 2))

        x = self.lin(x)

        x = x.sum(dim=1)

        return x

    # training_step()은 훈련 배치 하나에 대해 수행할 연산을 정의합니다.
    # PyTorch Lightning은 학습 중 이 함수를 자동으로 반복 호출합니다.
    def training_step(self, batch, batch_idx):

        # batch.text[0]에는 정수 인코딩된 문장 텐서가 들어 있습니다.
        # .T는 텐서를 전치하여 모델 입력 형태에 맞춥니다.
        # batch.label에는 각 문장의 실제 정답 라벨이 들어 있습니다.
        x, y = batch.text[0].T, batch.label

        # self(x)는 forward(x)를 호출하는 것과 같습니다.
        # 모델이 각 문장에 대한 클래스 점수를 예측합니다.
        y_hat = self(x)

        # 손실 함수를 사용하여 예측값 y_hat과 실제 정답 y의 차이를 계산합니다.
        loss = self.loss_function(y_hat, y)

        # 훈련 정확도를 계산합니다.
        train_acc = self.train_accuracy(y_hat, y)

        # self.log()는 학습 과정에서 지표를 기록합니다.
        # prog_bar=True는 진행률 표시줄에 train_acc 값을 보여 주겠다는 의미입니다.
        self.log(
            "train_acc",
            train_acc,
            prog_bar=True
        )

        # Lightning은 반환된 loss를 이용하여 역전파와 파라미터 업데이트를 수행합니다.
        return {
            "loss": loss
        }

    # validation_step()은 검증 배치 하나에 대해 수행할 연산을 정의합니다.
    # 모델 학습 중 검증 데이터에 대한 성능을 확인할 때 사용됩니다.
    def validation_step(self, batch, batch_idx):

        # 검증 배치에서 입력 문장과 정답 라벨을 분리합니다.
        x, y = batch.text[0].T, batch.label

        # 검증 데이터에 대한 예측값을 계산합니다.
        y_hat = self(x)

        # 검증 손실을 계산합니다.
        loss = self.loss_function(y_hat, y)

        # 검증 정확도를 계산합니다.
        val_acc = self.val_accuracy(y_hat, y)

        # 검증 정확도를 진행률 표시줄에 기록합니다.
        self.log(
            "val_acc",
            val_acc,
            prog_bar=True
        )

        # 검증 손실을 반환합니다.
        return {
            "validation_loss": loss
        }

    # train_dataloader()는 훈련에 사용할 데이터 로더를 반환합니다.
    # PyTorch Lightning의 trainer.fit()은 이 함수를 호출하여 훈련 데이터를 가져옵니다.
    def train_dataloader(self):

        # 앞에서 만든 train_iter를 훈련 데이터 로더로 반환합니다.
        return train_iter

    # val_dataloader()는 검증에 사용할 데이터 로더를 반환합니다.
    def val_dataloader(self):

        # 앞에서 만든 test_iter를 검증 데이터 로더로 반환합니다.
        return test_iter

    # configure_optimizers()는 모델 학습에 사용할 최적화 알고리즘을 정의합니다.
    # PyTorch Lightning은 이 함수를 호출하여 optimizer를 자동으로 설정합니다.
    def configure_optimizers(self):

        # Adam optimizer는 학습률을 적응적으로 조절하는 최적화 알고리즘입니다.
        # self.parameters()는 학습 가능한 모델 파라미터 전체를 의미합니다.
        # lr=0.01은 학습률을 0.01로 설정한다는 의미입니다.
        return Adam(
            self.parameters(),
            lr=0.01
        )

 

 

 

모델 객체 생성

생성된 사전 학습 입베딩 벡터 행렬. 

# text_field.vocab.vectors는 build_vocab()에서 생성된 사전 학습 임베딩 벡터 행렬입니다.
# 이 벡터 행렬을 RNNModel에 전달하여 모델이 단어 임베딩을 사용할 수 있게 합니다.
model = RNNModel(
    text_field.vocab.vectors
)

 

 

 

pytorch lightning 

pl.Trainer()

gpus =1은 gpu1개를 사용하겠다는 의미. max_epochs = 3은 전체 훈련 데이터를 3번 반복해서 학습하겠다는 의미

trainer.fit(model) 모델학습시작. 

# PyTorch Lightning의 버전에 따라 GPU 설정 인자가 달라질 수 있습니다.
# 구버전에서는 gpus=1을 사용하고, 최신 버전에서는 accelerator/devices를 사용하는 경우가 많습니다.
try:
    # 구버전 PyTorch Lightning 방식입니다.
    # gpus=1은 GPU 1개를 사용하겠다는 의미입니다.
    # max_epochs=3은 전체 훈련 데이터를 3번 반복해서 학습하겠다는 의미입니다.
    trainer = pl.Trainer(
        gpus=1 if torch.cuda.is_available() else 0,
        max_epochs=3
    )
except TypeError:
    # 최신 PyTorch Lightning 방식입니다.
    # accelerator는 GPU가 있으면 "gpu", 없으면 "cpu"로 설정합니다.
    # devices=1은 사용할 장치 개수를 1개로 지정합니다.
    trainer = pl.Trainer(
        accelerator="gpu" if torch.cuda.is_available() else "cpu",
        devices=1,
        max_epochs=3
    )

# trainer.fit()은 모델 학습을 시작합니다.
# Lightning은 내부적으로 train_dataloader(), validation_step(), configure_optimizers() 등을 호출합니다.
trainer.fit(model)