2006년 구글 번역기 등장. 초기 대중에게 번역서비스를 제공하지않고 UN이 보유한 디지털문서를 분류하고 외래어 번역을 위한 자연어 처리 서비스에 활용하며 부자연스러운 번역 기술로 비난받았다.
PBMT Pharse Based Machine Translation
문장을 단어 또는 구 단위로 분리하고 가장 확률이 높은 번역 결과를 선택해 문맥 이해가 어렵다.
GNMT Google Neural Machine Translation
인공신경망 기반 번역 문장전체의미를 이해해 번역 품질이 향상되었다.
https://standout.tistory.com/1866
자동번역 기술 PBMT와 GNMT
자동번역 기술은 구글번역기와 네이버 파파고가 있다. 번역기는 단어를 그대로 해석하는 경우가 많아 문맥을 이해하지 못하면 의미가 달라진다. 최근 신경망 번역기는 표준어 뿐 아니라 사투리
standout.tistory.com
예시 코드를 분석해보자. torch_seq2seq_translator.ipynb
https://standout.tistory.com/1844
입력과 출력의 길이가 달라도 처리할 수 있는 모델 Seq2Seq , 정보 병목(Information Bottleneck) (feat. Atte
Seq2Seq (Sequence to Sequence) 입력 시퀀스(Sequence)를 다른 출력 시퀀스로 변환하는 딥러닝 모델이다. 순서가 있는 데이터를 입력받아, 또 다른 순서가 있는 데이터를 출력하는 모델주로 LSTM나 GRU기반으
standout.tistory.com
import.
zipfile 압축해제
urllib.request 파일 url로 다운로드
unicodedata 유니코드 문자의 악센트 제거
# 운영체제의 파일 경로 처리, 폴더 생성, 파일 존재 여부 확인 등에 사용하는 표준 라이브러리입니다.
import os
# 정규표현식을 사용하여 문장 안의 특수문자, 구두점, 공백 등을 정리하기 위한 표준 라이브러리입니다.
import re
# zip 파일 압축 해제에 사용하는 표준 라이브러리입니다.
import zipfile
# URL에서 데이터 파일을 다운로드하기 위해 사용하는 표준 라이브러리입니다.
import urllib.request
# 유니코드 문자의 악센트를 제거하기 위해 사용하는 표준 라이브러리입니다.
import unicodedata
# 난수 고정을 통해 매번 비슷한 학습 결과가 나오도록 설정하기 위해 사용하는 표준 라이브러리입니다.
import random
# 데이터프레임 형태로 데이터를 확인하고 처리하기 위해 사용하는 외부 라이브러리입니다.
import pandas as pd
# 수치 계산과 배열 처리를 위해 사용하는 외부 라이브러리입니다.
import numpy as np
# PyTorch의 핵심 패키지로, 텐서 생성과 GPU 연산을 담당합니다.
import torch
# PyTorch에서 신경망 계층, 손실 함수 등을 만들기 위해 사용하는 모듈입니다.
import torch.nn as nn
# PyTorch에서 Adam 같은 최적화 알고리즘을 사용하기 위한 모듈입니다.
import torch.optim as optim
# Dataset과 DataLoader를 사용하여 데이터를 미니배치 단위로 공급하기 위한 모듈입니다.
from torch.utils.data import Dataset, DataLoader
난수고정
# Python random 모듈에서 사용하는 난수 시드를 고정합니다.
random.seed(42)
# NumPy에서 사용하는 난수 시드를 고정합니다.
np.random.seed(42)
# PyTorch CPU 연산에서 사용하는 난수 시드를 고정합니다.
torch.manual_seed(42)
# CUDA GPU가 사용 가능한 경우 GPU 연산의 난수 시드도 고정합니다.
torch.cuda.manual_seed_all(42)
# CUDA GPU가 있으면 'cuda'를 사용하고, 없으면 'cpu'를 사용하도록 장치를 선택합니다.
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 현재 학습에 사용할 장치를 출력합니다.
print("사용 장치:", device)
url로 데이터 다운로드.
다운로드된 zip파일 os.path.join() 경로 지정. 실제 텍스트 파일 경로지정. os.makedirs 폴더가없을경우 새로 새이성
파일이 존재하지않을경우에만 다운로드 시도.
인터넷 연결 실패시 오류내용 출력.
else 실습 구조 확인용 소규모 예제 데이터 생성.
# 데이터 파일을 저장할 폴더 이름을 지정합니다.
DATA_DIR = "data"
# 영어-프랑스어 병렬 데이터가 들어 있는 zip 파일 URL을 지정합니다.
DATA_URL = "http://www.manythings.org/anki/fra-eng.zip"
# 다운로드된 zip 파일이 저장될 경로를 지정합니다.
ZIP_PATH = os.path.join(DATA_DIR, "fra-eng.zip")
# 압축 해제 후 사용할 실제 텍스트 파일 경로를 지정합니다.
FRA_TXT_PATH = os.path.join(DATA_DIR, "fra.txt")
# data 폴더가 없으면 새로 생성합니다.
os.makedirs(DATA_DIR, exist_ok=True)
# fra.txt 파일이 아직 존재하지 않는 경우에만 다운로드를 시도합니다.
if not os.path.exists(FRA_TXT_PATH):
try:
# 데이터 다운로드 시작 메시지를 출력합니다.
print("fra-eng.zip 다운로드를 시작합니다.")
# 지정한 URL에서 zip 파일을 다운로드하여 ZIP_PATH에 저장합니다.
urllib.request.urlretrieve(DATA_URL, ZIP_PATH)
# 다운로드한 zip 파일을 읽기 모드로 엽니다.
with zipfile.ZipFile(ZIP_PATH, "r") as zip_ref:
# zip 파일 안의 모든 파일을 DATA_DIR 폴더에 압축 해제합니다.
zip_ref.extractall(DATA_DIR)
# 다운로드와 압축 해제가 끝났음을 출력합니다.
print("데이터 다운로드 및 압축 해제가 완료되었습니다.")
except Exception as e:
# 인터넷 연결 또는 다운로드 실패 시 오류 내용을 출력합니다.
print("다운로드 실패:", e)
if os.path.exists(ZIP_PATH):
# 다운로드한 zip 파일을 읽기 모드로 엽니다.
with zipfile.ZipFile(ZIP_PATH, "r") as zip_ref:
# zip 파일 안의 모든 파일을 DATA_DIR 폴더에 압축 해제합니다.
zip_ref.extractall(DATA_DIR)
# 다운로드와 압축 해제가 끝났음을 출력합니다.
print("저장된 데이터 압축 해제가 완료되었습니다.")
else:
# 실습 구조 확인용 소규모 예제 데이터를 생성합니다.
sample_lines = [
"Go.\tVa !\tCC-BY 2.0",
"Hi.\tSalut !\tCC-BY 2.0",
"Run!\tCours !\tCC-BY 2.0",
"I love you.\tJe t'aime.\tCC-BY 2.0",
"Thank you.\tMerci.\tCC-BY 2.0",
"I am hungry.\tJ'ai faim.\tCC-BY 2.0",
"Good morning.\tBonjour.\tCC-BY 2.0",
]
# 예제 데이터를 fra.txt 형식으로 저장합니다.
with open(FRA_TXT_PATH, "w", encoding="utf-8") as f:
# 각 예제 문장을 한 줄씩 파일에 기록합니다.
f.write("\n".join(sample_lines))
# 예제 데이터 생성 완료 메시지를 출력합니다.
print("실습용 예제 fra.txt 파일을 생성했습니다.")
else:
# fra.txt가 이미 있으면 다시 다운로드하지 않습니다.
print("기존 fra.txt 파일을 사용합니다.")
pd.read.csv() 파일읽기. 탭 구분자로 읽는다. 원본 파일이 영어, 프랑스어, 라이선스 컬럼으로 구성된다. 탭문자로 구분되어있음을 지정해 읽을 수 있다.
# fra.txt 파일을 탭 구분자로 읽습니다.
lines = pd.read_csv(
FRA_TXT_PATH, # 읽을 파일 경로입니다.
names=["src", "tar", "lic"], # 원본 파일은 영어, 프랑스어, 라이선스 컬럼으로 구성됩니다.
sep="\t", # 컬럼이 탭 문자로 구분되어 있음을 지정합니다.
encoding="utf-8" # 프랑스어 악센트 문자가 깨지지 않도록 UTF-8로 읽습니다.
)
# 번역 모델 학습에는 라이선스 컬럼이 필요하지 않으므로 삭제합니다.
lines = lines[["src", "tar"]]
# 전체 데이터 개수를 확인합니다.
print("전체 샘플 개수:", len(lines))
# 데이터가 너무 많으면 실습 시간이 오래 걸리므로 일부만 사용합니다.
NUM_SAMPLES = min(5000, len(lines))
# 앞에서부터 NUM_SAMPLES개만 선택한 뒤 복사본을 만듭니다.
lines = lines.iloc[:NUM_SAMPLES].copy()
# 데이터 일부를 확인합니다.
lines.head()7
src tar
0 Go. Va !
1 Hi. Salut !
2 Run! Cours !
3 I love you. Je t'aime.
4 Thank you. Merci.
to_ascil()
unicodedata.normalize() NFD 악센트가 붙은 문자를 분리한다.
"".join() Mn인 문자는 악센트 표시임으로 제외하고 나머지만 연결한다. 반환.
preprogress_sentence()
str(sentence) 문장을 문자열로 변환해 숫자형, 결측형 문제를 줄인다.
to_ascil() lower().strip() 소문자로 변환하고 악센트를 제거한다.
re.sub 정규화 전처리.
반환
# 프랑스어 악센트 문자를 일반 알파벳으로 바꾸기 위한 함수입니다.
def to_ascii(text):
# unicodedata.normalize('NFD', text)는 악센트가 붙은 문자를 기본 문자와 악센트 기호로 분리합니다.
normalized_text = unicodedata.normalize("NFD", text)
# 유니코드 범주가 'Mn'인 문자는 악센트 표시이므로 제외하고 나머지만 연결합니다.
ascii_text = "".join(char for char in normalized_text if unicodedata.category(char) != "Mn")
# 악센트가 제거된 문자열을 반환합니다.
return ascii_text
# 번역 모델에 넣기 전 문장을 정리하는 함수입니다.
def preprocess_sentence(sentence):
# 입력 문장을 문자열로 변환하여 혹시 모를 숫자형/결측형 문제를 줄입니다.
sentence = str(sentence)
# 문장을 소문자로 변환하고 프랑스어 악센트를 제거합니다.
sentence = to_ascii(sentence.lower().strip())
# 마침표, 물음표, 느낌표, 쉼표 앞뒤에 공백을 넣어 구두점을 독립 토큰으로 분리합니다.
sentence = re.sub(r"([?.!,¿])", r" \1 ", sentence)
# 영어 알파벳과 주요 구두점 외의 문자는 공백으로 바꿉니다.
sentence = re.sub(r"[^a-zA-Z?.!,¿]+", " ", sentence)
# 여러 개의 공백을 하나의 공백으로 줄입니다.
sentence = re.sub(r"\s+", " ", sentence)
# 앞뒤 공백을 제거한 문장을 반환합니다.
return sentence.strip()
# 전처리 결과를 확인하기 위한 예제 영어 문장입니다.
en_sent = "Have you had dinner?"
# 전처리 결과를 확인하기 위한 예제 프랑스어 문장입니다.
fr_sent = "Avez-vous déjà dîné?"
# 영어 문장의 전처리 전후를 출력합니다.
print("전처리 전 영어 문장:", en_sent)
print("전처리 후 영어 문장:", preprocess_sentence(en_sent))
# 프랑스어 문장의 전처리 전후를 출력합니다.
print("전처리 전 프랑스어 문장:", fr_sent)
print("전처리 후 프랑스어 문장:", preprocess_sentence(fr_sent))전처리 전 영어 문장: Have you had dinner?
전처리 후 영어 문장: have you had dinner ?
전처리 전 프랑스어 문장: Avez-vous déjà dîné?
전처리 후 프랑스어 문장: avez vous deja dine ?
영어 인코더, 프랑스어 디코더 입력, 정답 문장을 저장할 리스트 초기화
for문으로 lines.iterrows() 데이터프레임의 각 행을 순서대로 반복.
위함수를 사용. preprocess_sentence().split() 영어 원문을 전처리하고 공백기준으로 토큰 리스트 만든다.
프랑스어 번역문도. 전처리 + 공백기준 토큰 리스트 만들기 .
빈문장은 건너뛴다.
인코더에는 영어 토큰 리스트만.
디코더 입력에는 문장 시작 토큰 <sos>를 붙인다.
디코더 정답에는 문장 종료 토큰 <eos>를 붙인다.
# 영어 인코더 입력 문장을 저장할 리스트를 생성합니다.
encoder_texts = []
# 프랑스어 디코더 입력 문장을 저장할 리스트를 생성합니다.
decoder_input_texts = []
# 프랑스어 디코더 정답 문장을 저장할 리스트를 생성합니다.
decoder_target_texts = []
# 데이터프레임의 각 행을 순서대로 반복합니다.
for _, row in lines.iterrows():
# 영어 원문을 전처리하고 공백 기준으로 단어 토큰 리스트를 만듭니다.
src_tokens = preprocess_sentence(row["src"]).split()
# 프랑스어 번역문을 전처리하고 공백 기준으로 단어 토큰 리스트를 만듭니다.
tar_tokens = preprocess_sentence(row["tar"]).split()
# 빈 문장은 학습에 사용할 수 없으므로 건너뜁니다.
if len(src_tokens) == 0 or len(tar_tokens) == 0:
# 현재 반복을 종료하고 다음 행으로 이동합니다.
continue
# 인코더 입력에는 영어 토큰 리스트만 넣습니다.
encoder_texts.append(src_tokens)
# 디코더 입력에는 문장 시작 토큰 <sos>를 앞에 붙입니다.
decoder_input_texts.append(["<sos>"] + tar_tokens)
# 디코더 정답에는 문장 종료 토큰 <eos>를 뒤에 붙입니다.
decoder_target_texts.append(tar_tokens + ["<eos>"])
# 만들어진 데이터 샘플을 확인합니다.
print("인코더 입력 예시:", encoder_texts[:3])
print("디코더 입력 예시:", decoder_input_texts[:3])
print("디코더 정답 예시:", decoder_target_texts[:3])
인코더 입력 예시: [['go', '.'], ['hi', '.'], ['run', '!']]
디코더 입력 예시: [['<sos>', 'va', '!'], ['<sos>', 'salut', '!'], ['<sos>', 'cours', '!']]
디코더 정답 예시: [['va', '!', '<eos>'], ['salut', '!', '<eos>'], ['cours', '!', '<eos>']]
정수 인덱스 생성.
pad, unk알수없는 단어, sos문장시작, eos 문장 끝
build_vocab() 단어 사전을 만들어보자.
set() 중복제거
for문 각 단어문장들을 읽으며 token을vocab에 추가. sorted() 재현 가능한 결과를 위해 단어를 사전순으로 정렬함.
기본 토큰 리스트 정의해 토큰 앞에 배치.
특수토큰이 없는 경우에는 pad와 unk만.
단어를 정수인덱스로 바꾸는 딕셔너리 생성
정수인덱스를 단어로 되돌리는 딕셔너리 생성
반환.
영어입력 문장용 단어사전 생성
프랑스어 출력용 단어사전 생성
# 패딩 토큰의 정수 인덱스입니다. 길이가 짧은 문장을 채울 때 사용합니다.
PAD_TOKEN = "<pad>"
# 알 수 없는 단어를 처리하기 위한 토큰입니다.
UNK_TOKEN = "<unk>"
# 문장 시작을 나타내는 토큰입니다.
SOS_TOKEN = "<sos>"
# 문장 끝을 나타내는 토큰입니다.
EOS_TOKEN = "<eos>"
# 입력 언어와 출력 언어의 단어 사전을 만들기 위한 함수입니다.
def build_vocab(tokenized_sentences, add_special_tokens=True):
# 중복 없이 단어를 모으기 위해 set 자료구조를 사용합니다.
vocab = set()
# 전체 문장 리스트를 반복합니다.
for sentence in tokenized_sentences:
# 한 문장 안의 단어들을 반복합니다.
for token in sentence:
# 현재 단어를 단어 집합에 추가합니다.
vocab.add(token)
# 재현 가능한 결과를 위해 단어를 사전순으로 정렬합니다.
vocab = sorted(list(vocab))
# 특수 토큰을 포함할 경우 기본 토큰들을 앞쪽 인덱스에 배치합니다.
if add_special_tokens:
# 기본 토큰 리스트를 정의합니다.
base_tokens = [PAD_TOKEN, UNK_TOKEN, SOS_TOKEN, EOS_TOKEN]
# 기본 토큰이 중복되지 않도록 제외한 뒤 단어 목록을 구성합니다.
vocab = base_tokens + [token for token in vocab if token not in base_tokens]
else:
# 특수 토큰을 추가하지 않는 경우 패딩과 미등록 단어 토큰만 기본으로 둡니다.
vocab = [PAD_TOKEN, UNK_TOKEN] + [token for token in vocab if token not in [PAD_TOKEN, UNK_TOKEN]]
# 단어를 정수 인덱스로 바꾸는 딕셔너리를 생성합니다.
word_to_index = {word: index for index, word in enumerate(vocab)}
# 정수 인덱스를 단어로 되돌리는 딕셔너리를 생성합니다.
index_to_word = {index: word for word, index in word_to_index.items()}
# 두 딕셔너리를 반환합니다.
return word_to_index, index_to_word
# 영어 입력 문장용 단어 사전을 생성합니다.
src_to_index, index_to_src = build_vocab(encoder_texts, add_special_tokens=False)
# 프랑스어 출력 문장용 단어 사전을 생성합니다.
tar_to_index, index_to_tar = build_vocab(decoder_input_texts + decoder_target_texts, add_special_tokens=True)
# 영어 단어 집합의 크기를 계산합니다.
src_vocab_size = len(src_to_index)
# 프랑스어 단어 집합의 크기를 계산합니다.
tar_vocab_size = len(tar_to_index)
# 단어 집합 크기를 출력합니다.
print("영어 단어 집합 크기:", src_vocab_size)
print("프랑스어 단어 집합 크기:", tar_vocab_size)
# 프랑스어 특수 토큰의 인덱스를 확인합니다.
print("프랑스어 특수 토큰 인덱스:", {token: tar_to_index[token] for token in [PAD_TOKEN, UNK_TOKEN, SOS_TOKEN, EOS_TOKEN]})
tokens_to_indices() 토큰 리스트를 return하되 사전에 없는 단어는 unk인덱스로 바꿈.
pad_sequences_torch() 여러 문장을 같은 길이를 맞춘다.
뒤쪽을 잘라내고 pad를 뒤에 추가하고.. 완료된 문장을 append 리스트에 저장해 torch.tensor 형태로 return.
앞서 정의한 단어사전들을 불러와 정수시퀀스로 변환.
# 단어 토큰 리스트를 정수 인덱스 리스트로 변환하는 함수입니다.
def tokens_to_indices(tokens, vocab):
# 사전에 없는 단어는 <unk> 인덱스로 바꾸어 오류를 방지합니다.
return [vocab.get(token, vocab[UNK_TOKEN]) for token in tokens]
# 여러 문장을 같은 길이로 맞추는 패딩 함수입니다.
def pad_sequences_torch(sequences, max_len, pad_value=0):
# 패딩이 적용된 정수 시퀀스를 저장할 리스트입니다.
padded_sequences = []
# 모든 정수 시퀀스를 하나씩 반복합니다.
for sequence in sequences:
# 최대 길이보다 긴 문장은 뒤쪽을 잘라냅니다.
sequence = sequence[:max_len]
# 현재 문장 길이가 max_len보다 짧으면 pad_value를 뒤에 추가합니다.
padded_sequence = sequence + [pad_value] * (max_len - len(sequence))
# 패딩이 완료된 문장을 리스트에 저장합니다.
padded_sequences.append(padded_sequence)
# Python 리스트를 PyTorch LongTensor로 변환하여 반환합니다.
return torch.tensor(padded_sequences, dtype=torch.long)
# 영어 문장들을 정수 시퀀스로 변환합니다.
encoder_sequences = [tokens_to_indices(sentence, src_to_index) for sentence in encoder_texts]
# 프랑스어 디코더 입력 문장들을 정수 시퀀스로 변환합니다.
decoder_input_sequences = [tokens_to_indices(sentence, tar_to_index) for sentence in decoder_input_texts]
# 프랑스어 디코더 정답 문장들을 정수 시퀀스로 변환합니다.
decoder_target_sequences = [tokens_to_indices(sentence, tar_to_index) for sentence in decoder_target_texts]
# 영어 입력 문장의 최대 길이를 계산합니다.
max_src_len = max(len(sequence) for sequence in encoder_sequences)
# 프랑스어 출력 문장의 최대 길이를 계산합니다.
max_tar_len = max(len(sequence) for sequence in decoder_input_sequences)
# 영어 입력 시퀀스를 같은 길이로 패딩합니다.
encoder_input = pad_sequences_torch(encoder_sequences, max_src_len, pad_value=src_to_index[PAD_TOKEN])
# 프랑스어 디코더 입력 시퀀스를 같은 길이로 패딩합니다.
decoder_input = pad_sequences_torch(decoder_input_sequences, max_tar_len, pad_value=tar_to_index[PAD_TOKEN])
# 프랑스어 디코더 정답 시퀀스를 같은 길이로 패딩합니다.
decoder_target = pad_sequences_torch(decoder_target_sequences, max_tar_len, pad_value=tar_to_index[PAD_TOKEN])
# 텐서 크기를 출력합니다.
print("인코더 입력 크기:", encoder_input.shape)
print("디코더 입력 크기:", decoder_input.shape)
print("디코더 정답 크기:", decoder_target.shape)
indices = torch.randomperm() 랜덤 인뎃르를 생성해 인코더 입력, 디코더 입력, 정답에 적용해 섞는다.
검증데이터 수를 전체의 약 10%.
훈련용 인코더데이터 분리,
훈련용 디코더데이터 분리,
훈련용 디코더 정답 데이터 분리
검증용 인코더 인력 데이터 분리
검증용 디코더 정답 데이터 분리
# 전체 샘플 개수를 확인합니다.
num_data = encoder_input.size(0)
# 전체 데이터 인덱스를 생성합니다.
indices = torch.randperm(num_data)
# 같은 순서로 인코더 입력, 디코더 입력, 디코더 정답을 섞습니다.
encoder_input = encoder_input[indices]
decoder_input = decoder_input[indices]
decoder_target = decoder_target[indices]
# 검증 데이터 개수를 전체의 10%로 설정합니다.
num_val = max(1, int(num_data * 0.1))
# 훈련용 인코더 입력 데이터를 분리합니다.
encoder_input_train = encoder_input[:-num_val]
# 훈련용 디코더 입력 데이터를 분리합니다.
decoder_input_train = decoder_input[:-num_val]
# 훈련용 디코더 정답 데이터를 분리합니다.
decoder_target_train = decoder_target[:-num_val]
# 검증용 인코더 입력 데이터를 분리합니다.
encoder_input_val = encoder_input[-num_val:]
# 검증용 디코더 입력 데이터를 분리합니다.
decoder_input_val = decoder_input[-num_val:]
# 검증용 디코더 정답 데이터를 분리합니다.
decoder_target_val = decoder_target[-num_val:]
# 분리된 데이터 크기를 출력합니다.
print("훈련 source 데이터 크기:", encoder_input_train.shape)
print("훈련 target 입력 데이터 크기:", decoder_input_train.shape)
print("훈련 target 정답 데이터 크기:", decoder_target_train.shape)
print("검증 source 데이터 크기:", encoder_input_val.shape)
print("검증 target 입력 데이터 크기:", decoder_input_val.shape)
print("검증 target 정답 데이터 크기:", decoder_target_val.shape)훈련 source 데이터 크기: torch.Size([6, 4])
훈련 target 입력 데이터 크기: torch.Size([6, 5])
훈련 target 정답 데이터 크기: torch.Size([6, 5])
검증 source 데이터 크기: torch.Size([1, 4])
검증 target 입력 데이터 크기: torch.Size([1, 5])
검증 target 정답 데이터 크기: torch.Size([1, 5])
translationDataset()
init 객체가 생성될때 인코더, 디코더 저장
len 개수 반환
getitem 샘플하나 반환. 이때 인코더 입력, 디코더 입력, 디코더 정받으로 구성한다.
훈련데이터셋 객체 생성
검증데이터셋 객체 생성
미니배치 크기를 지정해 dataloader 정의
# Seq2Seq 학습용 Dataset 클래스를 정의합니다.
class TranslationDataset(Dataset):
# Dataset 객체가 생성될 때 인코더 입력, 디코더 입력, 디코더 정답을 저장합니다.
def __init__(self, encoder_input, decoder_input, decoder_target):
# 인코더 입력 텐서를 객체 변수로 저장합니다.
self.encoder_input = encoder_input
# 디코더 입력 텐서를 객체 변수로 저장합니다.
self.decoder_input = decoder_input
# 디코더 정답 텐서를 객체 변수로 저장합니다.
self.decoder_target = decoder_target
# Dataset 전체 샘플 개수를 반환합니다.
def __len__(self):
# 인코더 입력의 첫 번째 차원은 샘플 개수입니다.
return self.encoder_input.size(0)
# 특정 인덱스의 샘플 하나를 반환합니다.
def __getitem__(self, idx):
# 하나의 샘플은 인코더 입력, 디코더 입력, 디코더 정답으로 구성됩니다.
return self.encoder_input[idx], self.decoder_input[idx], self.decoder_target[idx]
# 훈련 Dataset 객체를 생성합니다.
train_dataset = TranslationDataset(encoder_input_train, decoder_input_train, decoder_target_train)
# 검증 Dataset 객체를 생성합니다.
val_dataset = TranslationDataset(encoder_input_val, decoder_input_val, decoder_target_val)
# 미니배치 크기를 지정합니다.
BATCH_SIZE = 128
# 훈련 DataLoader를 생성합니다.
train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True)
# 검증 DataLoader를 생성합니다.
val_loader = DataLoader(val_dataset, batch_size=BATCH_SIZE, shuffle=False)
encoder 클래스 정의
init() 정수인덱스로 표현된 단어를 nn.Embedding 임베딩으로 바꾸기.
nn.LSTM() LSTM 계층 만들기
forward() 순전파 연산 정의하기
# 인코더 클래스를 정의합니다.
class Encoder(nn.Module):
# 인코더에 필요한 계층을 초기화합니다.
def __init__(self, input_vocab_size, embedding_dim, hidden_units, pad_index):
# 부모 클래스인 nn.Module의 초기화 함수를 호출합니다.
super().__init__()
# 정수 인덱스로 표현된 단어를 dense vector로 바꾸는 임베딩 계층입니다.
self.embedding = nn.Embedding(
num_embeddings=input_vocab_size, # 입력 언어 단어 집합 크기입니다.
embedding_dim=embedding_dim, # 각 단어를 표현할 임베딩 벡터 차원입니다.
padding_idx=pad_index # 패딩 토큰은 학습에 영향을 덜 주도록 지정합니다.
)
# 입력 문장의 순서 정보를 처리하는 LSTM 계층입니다.
self.lstm = nn.LSTM(
input_size=embedding_dim, # LSTM에 들어가는 각 시점의 입력 벡터 크기입니다.
hidden_size=hidden_units, # LSTM 은닉 상태의 크기입니다.
batch_first=True # 입력 텐서 형태를 (batch, seq_len, feature)로 사용합니다.
)
# 인코더의 순전파 연산을 정의합니다.
def forward(self, src):
# src의 크기는 (batch_size, src_seq_len)입니다.
embedded = self.embedding(src)
# embedded의 크기는 (batch_size, src_seq_len, embedding_dim)입니다.
outputs, (hidden, cell) = self.lstm(embedded)
# outputs는 모든 시점의 출력이고, hidden과 cell은 마지막 시점의 상태입니다.
return hidden, cell
decoder()
init() 출력언어를 임베딩 벡터로 변환하기, LSTM, forward()
# 디코더 클래스를 정의합니다.
class Decoder(nn.Module):
# 디코더에 필요한 계층을 초기화합니다.
def __init__(self, output_vocab_size, embedding_dim, hidden_units, pad_index):
# 부모 클래스인 nn.Module의 초기화 함수를 호출합니다.
super().__init__()
# 출력 언어의 단어 인덱스를 임베딩 벡터로 변환하는 계층입니다.
self.embedding = nn.Embedding(
num_embeddings=output_vocab_size, # 출력 언어 단어 집합 크기입니다.
embedding_dim=embedding_dim, # 각 단어를 표현할 임베딩 벡터 차원입니다.
padding_idx=pad_index # 패딩 토큰의 인덱스입니다.
)
# 디코더의 LSTM 계층입니다.
self.lstm = nn.LSTM(
input_size=embedding_dim, # LSTM 입력 벡터 차원입니다.
hidden_size=hidden_units, # LSTM 은닉 상태 차원입니다.
batch_first=True # 입력 텐서를 (batch, seq_len, feature) 형태로 사용합니다.
)
# LSTM 출력 벡터를 출력 단어 집합 크기의 점수 벡터로 변환하는 완전연결 계층입니다.
self.fc_out = nn.Linear(hidden_units, output_vocab_size)
# 디코더의 순전파 연산을 정의합니다.
def forward(self, trg, hidden, cell):
# trg의 크기는 (batch_size, trg_seq_len)입니다.
embedded = self.embedding(trg)
# 인코더에서 전달받은 hidden, cell을 초기 상태로 사용하여 디코더 LSTM을 실행합니다.
outputs, (hidden, cell) = self.lstm(embedded, (hidden, cell))
# 각 시점의 LSTM 출력을 출력 단어 집합 크기의 logits로 변환합니다.
predictions = self.fc_out(outputs)
# predictions의 크기는 (batch_size, trg_seq_len, output_vocab_size)입니다.
return predictions, hidden, cell
Seq2Seq 인토더와 디코더를 하나로 묶은 모델
인코더가 입력 문장을 압축한 상태로 만들고 디코더가 그 상태를 이용해 번역문을 생성한다.
# 인코더와 디코더를 결합한 Seq2Seq 모델을 정의합니다.
class Seq2Seq(nn.Module):
# 모델 초기화 함수입니다.
def __init__(self, encoder, decoder):
# 부모 클래스인 nn.Module의 초기화 함수를 호출합니다.
super().__init__()
# 인코더 객체를 저장합니다.
self.encoder = encoder
# 디코더 객체를 저장합니다.
self.decoder = decoder
# 순전파 연산을 정의합니다.
def forward(self, src, trg):
# 인코더가 입력 문장을 읽고 마지막 hidden, cell 상태를 반환합니다.
hidden, cell = self.encoder(src)
# 디코더는 정답 문장의 이전 토큰들을 입력으로 받아 다음 토큰들을 예측합니다.
outputs, _, _ = self.decoder(trg, hidden, cell)
# 전체 시점의 예측 결과를 반환합니다.
return outputs
모델생성 및 손실함수, 최적화함수 설정.
# 임베딩 벡터 차원을 지정합니다.
EMBEDDING_DIM = 64
# LSTM 은닉 상태 크기를 지정합니다.
HIDDEN_UNITS = 64
# 학습 반복 횟수를 지정합니다.
EPOCHS = 10
# 영어 패딩 토큰의 인덱스를 가져옵니다.
src_pad_index = src_to_index[PAD_TOKEN]
# 프랑스어 패딩 토큰의 인덱스를 가져옵니다.
tar_pad_index = tar_to_index[PAD_TOKEN]
# 인코더 객체를 생성합니다.
encoder = Encoder(src_vocab_size, EMBEDDING_DIM, HIDDEN_UNITS, src_pad_index)
# 디코더 객체를 생성합니다.
decoder = Decoder(tar_vocab_size, EMBEDDING_DIM, HIDDEN_UNITS, tar_pad_index)
# Seq2Seq 모델 객체를 생성하고 학습 장치로 이동합니다.
model = Seq2Seq(encoder, decoder).to(device)
# 패딩 위치는 손실 계산에서 제외하도록 ignore_index를 지정합니다.
criterion = nn.CrossEntropyLoss(ignore_index=tar_pad_index)
# Adam 최적화 함수를 생성합니다.
optimizer = optim.Adam(model.parameters(), lr=0.001)
# 모델 구조를 출력합니다.
print(model)
학습함수 생성
model.train()
dataloader에서 미니배치를 하나씩 기져와 학습수행.
evaluate()
기울기를 계산하지않고 미니배치를 하나씩 가져와예측값을 펼쳐 검증손실 계산 누적 return
# 한 에폭 동안 모델을 학습하는 함수를 정의합니다.
def train_one_epoch(model, data_loader, optimizer, criterion, device):
# 모델을 학습 모드로 전환합니다.
model.train()
# 에폭 전체 손실을 누적할 변수를 초기화합니다.
total_loss = 0.0
# DataLoader에서 미니배치를 하나씩 가져옵니다.
for src, dec_in, dec_out in data_loader:
# 인코더 입력을 학습 장치로 이동합니다.
src = src.to(device)
# 디코더 입력을 학습 장치로 이동합니다.
dec_in = dec_in.to(device)
# 디코더 정답을 학습 장치로 이동합니다.
dec_out = dec_out.to(device)
# 이전 배치에서 계산된 기울기를 초기화합니다.
optimizer.zero_grad()
# 모델에 인코더 입력과 디코더 입력을 넣어 예측값을 계산합니다.
predictions = model(src, dec_in)
# predictions의 크기에서 출력 단어 집합 크기를 가져옵니다.
output_dim = predictions.size(-1)
# CrossEntropyLoss 입력 형식에 맞게 예측값을 2차원으로 펼칩니다.
predictions = predictions.reshape(-1, output_dim)
# 정답도 1차원으로 펼칩니다.
dec_out = dec_out.reshape(-1)
# 예측값과 정답을 비교하여 손실을 계산합니다.
loss = criterion(predictions, dec_out)
# 손실에 대한 역전파를 수행하여 기울기를 계산합니다.
loss.backward()
# 기울기 폭주를 줄이기 위해 gradient clipping을 적용합니다.
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
# 계산된 기울기를 이용하여 모델 파라미터를 업데이트합니다.
optimizer.step()
# 현재 배치 손실을 누적합니다.
total_loss += loss.item()
# 전체 배치의 평균 손실을 반환합니다.
return total_loss / len(data_loader)
# 검증 데이터를 이용해 모델 손실을 계산하는 함수를 정의합니다.
def evaluate(model, data_loader, criterion, device):
# 모델을 평가 모드로 전환합니다.
model.eval()
# 검증 손실을 누적할 변수를 초기화합니다.
total_loss = 0.0
# 평가 중에는 기울기를 계산하지 않습니다.
with torch.no_grad():
# DataLoader에서 미니배치를 하나씩 가져옵니다.
for src, dec_in, dec_out in data_loader:
# 인코더 입력을 학습 장치로 이동합니다.
src = src.to(device)
# 디코더 입력을 학습 장치로 이동합니다.
dec_in = dec_in.to(device)
# 디코더 정답을 학습 장치로 이동합니다.
dec_out = dec_out.to(device)
# 모델 예측값을 계산합니다.
predictions = model(src, dec_in)
# 출력 단어 집합 크기를 가져옵니다.
output_dim = predictions.size(-1)
# 손실 함수 입력 형식에 맞게 예측값을 펼칩니다.
predictions = predictions.reshape(-1, output_dim)
# 손실 함수 입력 형식에 맞게 정답을 펼칩니다.
dec_out = dec_out.reshape(-1)
# 검증 손실을 계산합니다.
loss = criterion(predictions, dec_out)
# 현재 배치 손실을 누적합니다.
total_loss += loss.item()
# 전체 검증 배치의 평균 손실을 반환합니다.
return total_loss / len(data_loader)
gkrtmq.
# 학습 이력을 저장할 리스트입니다.
history = []
# 지정한 에폭 수만큼 반복 학습합니다.
for epoch in range(1, EPOCHS + 1):
# 한 에폭 동안 훈련 데이터를 사용하여 모델을 학습합니다.
train_loss = train_one_epoch(model, train_loader, optimizer, criterion, device)
# 검증 데이터를 사용하여 현재 모델의 손실을 계산합니다.
val_loss = evaluate(model, val_loader, criterion, device)
# 현재 에폭의 학습 결과를 저장합니다.
history.append({"epoch": epoch, "train_loss": train_loss, "val_loss": val_loss})
# 현재 에폭의 학습 손실과 검증 손실을 출력합니다.
print(f"Epoch [{epoch:02d}/{EPOCHS}] train_loss={train_loss:.4f}, val_loss={val_loss:.4f}")Epoch [01/10] train_loss=2.8300, val_loss=2.8416
Epoch [02/10] train_loss=2.7983, val_loss=2.8324
Epoch [03/10] train_loss=2.7667, val_loss=2.8232
Epoch [04/10] train_loss=2.7350, val_loss=2.8142
Epoch [05/10] train_loss=2.7031, val_loss=2.8051
Epoch [06/10] train_loss=2.6708, val_loss=2.7960
Epoch [07/10] train_loss=2.6378, val_loss=2.7868
Epoch [08/10] train_loss=2.6042, val_loss=2.7775
Epoch [09/10] train_loss=2.5697, val_loss=2.7680
Epoch [10/10] train_loss=2.5341, val_loss=2.7583
번역함수 작성
seq_to_src() 정수시퀀스를 영어문장으로 되돌리는 함수
패딩토큰은 출력하지않는다.
words.append() 추가한다. index_to_src.get(index, UNK_TOKEN) 정수인덱스르르 영어단어로 변환해 리스트에 추가.
seq_to_tar() 정수시퀀스를 프랑스문장으로 .
translate_sentences() 새로운 영어 문장을 프랑스어로 번역해보자.
progress_sentence 전처리
tokens_to_indeices() 정수인덱스로 변환 torch.tensor 텐서로 변환..
model_encoder() 인코더에서 hidden, cell 상태를 얻어 sostoken 토큰이 첫 입력이된다.
for문.
단어들을 하나씩 생성. model.decoder() 현재 토큰과 이전 상태를 디코더에 넣어 토큰 점수를 얻는다.
큰값을 가진 단어 인덱스 선택.
eos가 나오면 종료.
리스트에 추가.
예측한 단어를 다음 시점 디코어 입력으로 사용.
# 정수 시퀀스를 영어 문장으로 되돌리는 함수입니다.
def seq_to_src(input_seq):
# 복원된 단어를 저장할 리스트입니다.
words = []
# 정수 시퀀스의 각 인덱스를 반복합니다.
for index in input_seq:
# Tensor 값이면 Python 정수로 변환합니다.
index = int(index)
# 패딩 토큰은 출력하지 않습니다.
if index != src_to_index[PAD_TOKEN]:
# 정수 인덱스를 영어 단어로 변환하여 리스트에 추가합니다.
words.append(index_to_src.get(index, UNK_TOKEN))
# 단어 리스트를 공백으로 연결하여 문장으로 반환합니다.
return " ".join(words)
# 정수 시퀀스를 프랑스어 문장으로 되돌리는 함수입니다.
def seq_to_tar(input_seq):
# 복원된 단어를 저장할 리스트입니다.
words = []
# 정수 시퀀스의 각 인덱스를 반복합니다.
for index in input_seq:
# Tensor 값이면 Python 정수로 변환합니다.
index = int(index)
# 패딩, 시작, 종료 토큰은 사람이 읽는 번역문에서 제외합니다.
if index not in [tar_to_index[PAD_TOKEN], tar_to_index[SOS_TOKEN], tar_to_index[EOS_TOKEN]]:
# 정수 인덱스를 프랑스어 단어로 변환하여 리스트에 추가합니다.
words.append(index_to_tar.get(index, UNK_TOKEN))
# 단어 리스트를 공백으로 연결하여 문장으로 반환합니다.
return " ".join(words)
# 새로운 영어 문장을 프랑스어로 번역하는 함수입니다.
def translate_sentence(sentence, model, max_len=30):
# 모델을 평가 모드로 전환합니다.
model.eval()
# 입력 문장을 전처리하고 단어 토큰 리스트로 변환합니다.
src_tokens = preprocess_sentence(sentence).split()
# 입력 단어들을 정수 인덱스로 변환합니다.
src_indices = tokens_to_indices(src_tokens, src_to_index)
# 입력 시퀀스를 텐서로 변환하고 배치 차원을 추가합니다.
src_tensor = torch.tensor(src_indices, dtype=torch.long).unsqueeze(0).to(device)
# 기울기 계산 없이 번역을 수행합니다.
with torch.no_grad():
# 인코더에서 마지막 hidden, cell 상태를 얻습니다.
hidden, cell = model.encoder(src_tensor)
# 첫 디코더 입력은 <sos> 토큰입니다.
current_token = torch.tensor([[tar_to_index[SOS_TOKEN]]], dtype=torch.long).to(device)
# 예측된 단어를 저장할 리스트입니다.
translated_tokens = []
# 최대 길이만큼 반복하여 단어를 하나씩 생성합니다.
for _ in range(max_len):
# 기울기 계산 없이 다음 단어를 예측합니다.
with torch.no_grad():
# 현재 토큰과 이전 상태를 디코더에 넣어 다음 토큰의 점수를 얻습니다.
output, hidden, cell = model.decoder(current_token, hidden, cell)
# 마지막 시점의 출력 점수에서 가장 큰 값을 가진 단어 인덱스를 선택합니다.
next_token = output[:, -1, :].argmax(dim=-1).item()
# <eos>가 나오면 문장 생성이 끝났으므로 반복을 중단합니다.
if next_token == tar_to_index[EOS_TOKEN]:
break
# 예측된 단어를 리스트에 추가합니다.
translated_tokens.append(index_to_tar.get(next_token, UNK_TOKEN))
# 방금 예측한 단어를 다음 시점의 디코더 입력으로 사용합니다.
current_token = torch.tensor([[next_token]], dtype=torch.long).to(device)
# 예측된 단어들을 공백으로 연결하여 번역문을 반환합니다.
return " ".join(translated_tokens)
즉
영어문장 - 전처리 - 토큰화 - 정수인덱스 변환 - encoder - hidden, cell - sos - decoder - 첫단어생성 - 생성단어 다시 입력 - 다음단어생성 - eos가 나오면 종료.
index_to_src가 {
5:"i",
8:"love",
19:"you"
}라면 words.append(index_to_src.get(index)) 마지막에 공백을 더해 반환 return " ".join(words) i love you
pad는 실제 단어가 아니니 제외.
seq_to_tar() 에서 sos, eos, pad 모두 제거.
translate_sentence() I love you.를 preprocess_sentence() 로 ["i","love","you"], tokens_to_indices()로 [15,98,31], tensor로 생성 [[15,98,31]]
model.encoder(src_tensor) encoder 실행. encoder는 수행후 hidden이라는 문장의 요약을 반환한다.
"사랑한다"
"주어는 I"
"목적어는 you" 와 같은 의미가 압축되어있다.
decoder 시작. 처음에는 아무단어도 모름으로 sos를 넣음.
최대 max_len 까지 생성. decoder 실행.
encoder의 hidden과 cell이 단어와함께 입력되어 output 모든 단어에 대한 점수. 가장 큰 값을 선택해 eos가 나오면 반복종료 단어저장 방금 생성한 단어를 다시 Decoder의 입력으로 사용하는 방식입니다. 이를 자기회귀(autoregressive) 생성
"I love you"
["i", "love", "you"]
[15, 98, 31]
Tensor([[15, 98, 31]])
Encoder
hidden, cell
Decoder 입력: <SOS>
예측: "je"
Decoder 입력: "je"
예측: "t'"
Decoder 입력: "t'"
예측: "aime"
Decoder 입력: "aime"
예측: <EOS>
종료
최종 결과: "je t' aime"
샘플 인덱스 목록 지정.
for문으로 반복.
seq_to_src로 영어 입력 문장을 복원
seq_to_tar로 프랑스어 정답 문장 복원
translate_sentence로 영어입력 문장을 복원한 것을 translate. 프랑스어로.
출력
# 결과를 확인할 샘플 인덱스 목록을 지정합니다.
sample_indices = [0, 1, 2, min(10, len(encoder_input_train) - 1), min(100, len(encoder_input_train) - 1)]
# 샘플 인덱스를 하나씩 반복합니다.
for seq_index in sample_indices:
# 현재 인덱스의 영어 입력 문장을 복원합니다.
source_sentence = seq_to_src(encoder_input_train[seq_index])
# 현재 인덱스의 프랑스어 정답 문장을 복원합니다.
target_sentence = seq_to_tar(decoder_target_train[seq_index])
# 모델을 이용하여 영어 문장을 프랑스어로 번역합니다.
predicted_sentence = translate_sentence(source_sentence, model, max_len=max_tar_len + 5)
# 구분선을 출력합니다.
print("-" * 60)
# 입력 문장을 출력합니다.
print("입력 문장:", source_sentence)
# 정답 문장을 출력합니다.
print("정답 문장:", target_sentence)
# 번역 문장을 출력합니다.
print("번역 문장:", predicted_sentence)------------------------------------------------------------
입력 문장: hi .
정답 문장: salut !
번역 문장: je
------------------------------------------------------------
입력 문장: good morning .
정답 문장: bonjour .
번역 문장: je t aime
------------------------------------------------------------
입력 문장: i love you .
정답 문장: je t aime .
번역 문장: je t aime
------------------------------------------------------------
입력 문장: go .
정답 문장: va !
번역 문장: je t aime
------------------------------------------------------------
입력 문장: go .
정답 문장: va !
번역 문장: je t aime
모델 저장.
모델 가중치 및 사전정보를 딕셔너리로 구성해 저장함.
# 모델 저장 폴더를 지정합니다.
SAVE_DIR = "saved_model"
# 저장 폴더가 없으면 생성합니다.
os.makedirs(SAVE_DIR, exist_ok=True)
# 모델 가중치와 사전 정보를 하나의 체크포인트 딕셔너리로 구성합니다.
checkpoint = {
"model_state_dict": model.state_dict(), # 학습된 PyTorch 모델 파라미터입니다.
"src_to_index": src_to_index, # 영어 단어를 정수로 바꾸는 사전입니다.
"index_to_src": index_to_src, # 정수를 영어 단어로 바꾸는 사전입니다.
"tar_to_index": tar_to_index, # 프랑스어 단어를 정수로 바꾸는 사전입니다.
"index_to_tar": index_to_tar, # 정수를 프랑스어 단어로 바꾸는 사전입니다.
"src_vocab_size": src_vocab_size, # 영어 단어 집합 크기입니다.
"tar_vocab_size": tar_vocab_size, # 프랑스어 단어 집합 크기입니다.
"embedding_dim": EMBEDDING_DIM, # 임베딩 차원입니다.
"hidden_units": HIDDEN_UNITS, # LSTM 은닉 상태 크기입니다.
"max_src_len": max_src_len, # 영어 문장의 최대 길이입니다.
"max_tar_len": max_tar_len # 프랑스어 문장의 최대 길이입니다.
}
# 체크포인트 파일 경로를 지정합니다.
checkpoint_path = os.path.join(SAVE_DIR, "seq2seq_translation_torch.pt")
# 체크포인트를 파일로 저장합니다.
torch.save(checkpoint, checkpoint_path)
# 저장 완료 메시지를 출력합니다.
print("모델 저장 완료:", checkpoint_path)
직접 테스트.
translate_sentence()
# 번역할 영어 문장을 지정합니다.
input_sentence = "I love you."
# 입력 문장을 모델로 번역합니다.
translated_sentence = translate_sentence(input_sentence, model, max_len=max_tar_len + 5)
# 원문을 출력합니다.
print("입력 문장:", input_sentence)
# 번역 결과를 출력합니다.
print("번역 결과:", translated_sentence)
직접 만들어보자.
파이참 - 프로젝트 만들기
config.py
pathlib import
루트 지정, data, model, meta
임베딩, 은닉차원, epochs, batch, learning rate, 번역결과최대길이, 특수토큰 정의
"""프로젝트 전체에서 공통으로 사용하는 설정값을 관리하는 파일입니다."""
from pathlib import Path
# 현재 config.py 파일의 상위 폴더(src)의 상위 폴더를 프로젝트 루트로 지정합니다.
# 예: smart_translator_torch_streamlit_project/src/config.py -> smart_translator_torch_streamlit_project
BASE_DIR = Path(__file__).resolve().parent.parent
# 학습 데이터 CSV 파일 경로를 지정합니다.
DATA_PATH = BASE_DIR / "data" / "translation_pairs.csv"
# 학습된 PyTorch 모델 파일이 저장될 경로를 지정합니다.
MODEL_PATH = BASE_DIR / "models" / "smart_translator.pt"
# 문자 사전, 문장 최대 길이, 하이퍼파라미터 등 메타 정보를 함께 저장할 경로입니다.
META_PATH = BASE_DIR / "models" / "translator_meta.pt"
# 인코더와 디코더의 임베딩 차원입니다.
# 문자를 정수로 바꾼 뒤, 이 정수를 벡터 공간으로 변환할 때 사용하는 크기입니다.
EMBED_SIZE = 64
# RNN의 은닉 상태 차원입니다.
# 값이 클수록 표현력이 커질 수 있지만 학습 시간이 증가할 수 있습니다.
HIDDEN_SIZE = 128
# 학습 반복 횟수입니다.
# 강의교안의 MY_EPOCH 개념에 해당하며, 데이터 전체를 몇 번 반복 학습할지 결정합니다.
EPOCHS = 120
# 한 번에 학습할 데이터 묶음 크기입니다.
# 작은 데이터셋이므로 16 정도로 설정하여 빠르게 학습되도록 합니다.
BATCH_SIZE = 16
# 학습률입니다.
# 옵티마이저가 가중치를 얼마나 크게 수정할지 결정합니다.
LEARNING_RATE = 0.003
# 번역 결과를 생성할 때 최대 몇 글자까지 만들지 결정합니다.
MAX_OUTPUT_LEN = 60
# 특수 토큰입니다.
# PAD는 길이를 맞추기 위한 빈 칸, SOS는 디코더 시작, EOS는 문장 종료, UNK는 사전에 없는 문자입니다.
PAD_TOKEN = "<PAD>"
SOS_TOKEN = "<SOS>"
EOS_TOKEN = "<EOS>"
UNK_TOKEN = "<UNK>"
model.py
GRU를 이용한 Seq2Seq번역모델
Encoder : 입력 문장을 이해한다.
Decoder : 이해한 내용을 바탕으로 번역문을 생성한다.
class Encoder 입력문장을 벡터로 압축하는 역할. nn.Embeding을 통과해 각 단어가 벡터가 된다 .
nn.GRU 는 문장을 순서대로 읽으며 hidden을 게속 업데이트한다.
forward.
Decoder 번역문을 만들어냄
self.embedding 생성한 번역문을 벡터로 바꾼다.
self.fc 단어수에 맞춰 각 점수가 필요하기 때문에 linear() 출력 수를 바꿈.
forward.
Seq2SeqTranslator Encoder와 Decoder를 하나의 모델로 묶는 클래스
Encoder 생성, Decoder 생성
hidden = self.encoder() 영어문장을 읽어 hidden을 생성
logits 모든 단어에 대한 점수를 반환. 학습할때는 예측을 정답과 비교해 loss를 계산해야하기 때문.
import torch
import torch.nn as nn
from tenacity import retry_if_not_exception_type
from src.config import PAD_TOKEN, SOS_TOKEN, EOS_TOKEN
class Encoder(nn.Module):
def __init__(self, vocab_size, embed_size, hidden_size):
super().__init__()
self.embedding = nn.Embedding(vocab_size, embed_size, padding_idx=0)
self.gru = nn.GRU(embed_size, hidden_size, batch_first=True)
def forward(self, source_idx):
embedded = self.embedding(source_idx)
outputs, hidden = self.gru(embedded)
return hidden
class Decoder(nn.Module):
def __init__(self, vocab_size, embed_size, hidden_size):
super().__init__()
self.embedding = nn.Embedding(vocab_size, embed_size, padding_idx=0)
self.gru = nn.GRU(embed_size, hidden_size, batch_first=True)
self.fc = nn.Linear(hidden_size, vocab_size)
def forward(self, decoder_input, hidden):
embedded = self.embedding(decoder_input)
outputs, hidden = self.gru(embedded, hidden)
logits = self.fc(outputs)
return logits, hidden
class Seq2SeqTranslator(nn.Module):
def __init__(self, vocab_size, embed_size, hidden_size):
super().__init__()
self.encoder = Encoder(vocab_size, embed_size, hidden_size)
self.decoder = Decoder(vocab_size, embed_size, hidden_size)
def forward(self, source_idx, decoder_input_idx):
hidden = self.encoder(source_idx)
logits, hidden = self.decoder(decoder_input_idx, hidden)
return logits
train.py
train_model()
device 세팅, load_translation_pairs() csv파일에서 번역 학습 쌍을 읽어옴.
build_vocab() 문자 기반 사전 생성
tTranslatetionDataset() pytorchdataset 객체 생성
DataLoader 생성.
Seq2SeqTranslator 모델 생성.
PAD 토큰을 제외하고 손실계산객체 생성.
learningrate 값으로 optimizer 생성
for문으로 반복학습 진행
MODEL_PATH 다운.
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from src.config import DATA_PATH, MODEL_PATH, META_PATH, EMBED_SIZE, HIDDEN_SIZE, EPOCHS, BATCH_SIZE, LEARNING_RATE
from src.data_utils import load_translation_pairs, build_vocab, TranslationDataset, collate_batch
from src.model import Seq2SeqTranslator
def train_model(epochs = EPOCHS, batch_size = BATCH_SIZE):
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
pairs = load_translation_pairs(DATA_PATH)
char2idx, idx2char = build_vocab(pairs)
vocab_size = len(char2idx)
dataset = TranslationDataset(pairs, char2idx)
loader = DataLoader(dataset, batch_size=BATCH_SIZE, shuffle=True, collate_fn=collate_batch)
model = Seq2SeqTranslator(vocab_size, EMBED_SIZE, HIDDEN_SIZE).to(device)
criterion = nn.CrossEntropyLoss(ignore_index=char2idx['<PAD>'])
optimizer = torch.optim.Adam(model.parameters(), lr=LEARNING_RATE)
epochs = EPOCHS
for epoch in range(1, epochs +1):
model.train()
total_loss = 0.0
for source_idx, decoder_input_idx, decoder_target_idx in loader:
source_idx = source_idx.to(device)
decoder_input_idx = decoder_input_idx.to(device)
decoder_target_idx = decoder_target_idx.to(device)
optimizer.zero_grad()
logits = model(source_idx, decoder_input_idx)
loss = criterion(logits.reshape(-1, logits.size(-1)), decoder_target_idx.reshape(-1))
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
total_loss += loss.item()
if epoch == 1 or epoch % 20 == 0:
print(f"Epoch {epoch:03d} / {epochs} | Loss: {total_loss / len(loader):.4f}")
MODEL_PATH.parent.mkdir(parents=True, exist_ok=True)
torch.save(model.state_dict(), MODEL_PATH)
torch.save({
"char2idx": char2idx,
"idx2char": idx2char,
"embed_size": EMBED_SIZE,
"hidden_size": HIDDEN_SIZE,
}, META_PATH)
print(f"모델저장완료{MODEL_PATH}")
return model, char2idx, idx2char
if __name__ == "__main__":
train_model(epochs = EPOCHS)
data_utils.py
import
src.config에서 pad, sos, eos, unk토큰을 가져옴
normalize_text() text를 받아 str문자열로 바꾼뒤 strip() 공백을 제거, 소문자 변환,
split() 문자열을 공백 기준으로 잘라서 리스트를 만든다. 이때 공백이 여러개여도 하나로 취급한다. 여러개 공백은 하나의 공백으로 합친다.
load_translation_pairs()
데이터를 읽어 en, ko 컬럼이 모두 존재하는지 확인. 결측치 제거,
en, ko의 모든 행에 map() 함수를 적용한다.
pairs 빈 객체를 만들고 interrows() 한줄씩 가져와 EN2KO로 영어, 한글 KO2EN로 한글, 영어씩 두개를 저장한다.
양방향 번역을 위해 영어 - 한국어도 필요하고 한국어 - 영어도 필요하니 하나의 데이터를 두개로 늘려 학습하는 것.
[
("<EN2KO> hello", "안녕하세요"),
("<KO2EN> 안녕하세요", "hello"),
("<EN2KO> thank you", "감사합니다"),
("<KO2EN> 감사합니다", "thank you")
]
최종적으로 이 pairs를 반환함.
build_vocab()
특수토큰 정의.
중복을 허용하지않는 set()으로 charset 초기화.
parirs 학습쌍을 for문으로 돌리며 source, target을 charset에 모음.
sorted() 정렬.
char2idx enumerate()로 실행하면 idx가 생긴다. 이를 바꾸어 token을 입력하면 idx가 나올 수 있게함.
char2idx = {
token: idx
for idx, token in enumerate(vocab)
}
{
"<PAD>":0,
"<SOS>":1,
"<EOS>":2,
"<UNK>":3,
" ":4,
"!":5,
}
encode_text()
text가 아까만든 문자사전 char2idx에 있으면 숫자로, 없으면 UNK에해당하는 숫자로 , 끝에는 eos를 붙여 문장이 끝났음을 알린다.
이 숫자 배열을 return.
TranslationDataset() PyTorch가 학습할 수 있는 형태로 번역 데이터를 만들어 주는 클래스
pairs 쌍을 저장. char2idx 문자사전을 저장.
len() 객수반환
getitem() index 위치의 문장과 정답 문장을 가져온다. encode_text() 수행
sos로 시작하고, encode_text를 붙여 decoder_input_ids로 초기화.
target 정답에는 끝에 eos를 붙여 decoder_target_ids로 초기화
return.
collate_batch() 문장의 길이가 서로 다르기 때문에 같은 길이로 맞춰주는(Padding) 작업
배치에서 sources, decoder_inputs, decoder_targets를 분리한다.
padding_value값 초기화
pad_sequence 자동으로 가장 긴 문장 길이를 찾는다.
batch_first T 텐서모양을 배치개수, 문장길이로 결정한다. False면 문장길이, 배치개수 가 된다. seq2seq에서는 대부분 t.
패딩적용..
decoder_inputs 도 마찬가지로 패딩적용
decoder_targets도.
return.
"""번역 데이터 로딩, 문자 사전 생성, 문장 인코딩 기능을 제공하는 파일입니다."""
import pandas as pd
import torch
from torch.utils.data import Dataset
from src.config import PAD_TOKEN, SOS_TOKEN, EOS_TOKEN, UNK_TOKEN
def normalize_text(text: str) -> str:
"""입력 문장을 모델이 처리하기 쉬운 형태로 정리합니다."""
# None이나 결측값이 들어오는 경우를 방지하기 위해 문자열로 변환합니다.
text = str(text)
# 앞뒤 공백을 제거하고, 영어 대문자는 소문자로 통일합니다.
# 한글에는 lower()가 영향을 거의 주지 않으므로 한영 공통으로 사용할 수 있습니다.
text = text.strip().lower()
# 여러 개의 공백이 있을 경우 하나의 공백으로 합칩니다.
text = " ".join(text.split())
# 정리된 문자열을 반환합니다.
return text
def load_translation_pairs(csv_path):
"""CSV 파일에서 영어-한국어 번역 쌍을 읽고 양방향 학습 데이터로 확장합니다."""
# CSV 파일을 pandas DataFrame으로 읽습니다.
df = pd.read_csv(csv_path)
# 영어와 한국어 컬럼이 모두 존재하는지 확인합니다.
required_columns = {"en", "ko"}
# 필요한 컬럼이 없으면 명확한 오류 메시지를 발생시킵니다.
if not required_columns.issubset(set(df.columns)):
raise ValueError("CSV 파일에는 en, ko 컬럼이 반드시 있어야 합니다.")
# 결측값이 있는 행은 번역 학습에 사용할 수 없으므로 제거합니다.
df = df.dropna(subset=["en", "ko"])
# 각 문장을 정리합니다.
df["en"] = df["en"].map(normalize_text)
df["ko"] = df["ko"].map(normalize_text)
# 하나의 모델이 영어->한국어, 한국어->영어를 모두 학습하도록 방향 토큰을 붙입니다.
pairs = []
# CSV의 각 행을 순회하며 양방향 데이터를 구성합니다.
for _, row in df.iterrows():
# 영어를 한국어로 번역하는 학습 예시입니다.
pairs.append(("<EN2KO> " + row["en"], row["ko"]))
# 한국어를 영어로 번역하는 학습 예시입니다.
pairs.append(("<KO2EN> " + row["ko"], row["en"]))
# 전체 학습 쌍을 반환합니다.
return pairs
def build_vocab(pairs):
"""학습 데이터에 등장하는 모든 문자를 기반으로 문자 사전을 생성합니다."""
# 특수 토큰을 가장 앞에 배치하여 고정된 인덱스를 갖도록 합니다.
tokens = [PAD_TOKEN, SOS_TOKEN, EOS_TOKEN, UNK_TOKEN]
# 모든 입력 문장과 출력 문장에서 문자 단위 집합을 수집합니다.
charset = set()
# 학습 쌍을 반복하면서 입력과 출력의 문자를 모두 모읍니다.
for source, target in pairs:
# 입력 문장의 각 문자를 집합에 추가합니다.
charset.update(list(source))
# 출력 문장의 각 문자를 집합에 추가합니다.
charset.update(list(target))
# 재현 가능한 사전을 위해 문자 집합을 정렬합니다.
sorted_chars = sorted(charset)
# 특수 토큰 뒤에 실제 문자를 붙여 전체 토큰 목록을 만듭니다.
vocab = tokens + sorted_chars
# 문자 또는 특수 토큰을 정수 인덱스로 바꾸는 딕셔너리입니다.
char2idx = {token: idx for idx, token in enumerate(vocab)}
# 정수 인덱스를 문자 또는 특수 토큰으로 되돌리는 딕셔너리입니다.
idx2char = {idx: token for token, idx in char2idx.items()}
# 두 사전을 반환합니다.
return char2idx, idx2char
def encode_text(text, char2idx, add_eos=True):
"""문자열을 정수 인덱스 리스트로 변환합니다."""
# 사전에 없는 문자는 UNK 인덱스로 변환합니다.
ids = [char2idx.get(ch, char2idx[UNK_TOKEN]) for ch in text]
# 출력 문장이나 인코더 입력의 끝을 알리기 위해 EOS 토큰을 추가합니다.
if add_eos:
ids.append(char2idx[EOS_TOKEN])
# 정수 리스트를 반환합니다.
return ids
class TranslationDataset(Dataset):
"""PyTorch DataLoader가 사용할 수 있는 번역 데이터셋 클래스입니다."""
def __init__(self, pairs, char2idx):
# 원본 문장 쌍을 저장합니다.
self.pairs = pairs
# 문자 사전을 저장합니다.
self.char2idx = char2idx
def __len__(self):
# 전체 데이터 개수를 반환합니다.
return len(self.pairs)
def __getitem__(self, index):
# index 위치의 입력 문장과 정답 문장을 가져옵니다.
source, target = self.pairs[index]
# 인코더 입력은 입력 문장 뒤에 EOS를 붙입니다.
source_ids = encode_text(source, self.char2idx, add_eos=True)
# 디코더 입력은 SOS로 시작하고 정답 문장을 이어 붙입니다.
decoder_input_ids = [self.char2idx[SOS_TOKEN]] + encode_text(target, self.char2idx, add_eos=False)
# 디코더 정답은 정답 문장 뒤에 EOS를 붙입니다.
decoder_target_ids = encode_text(target, self.char2idx, add_eos=True)
# 학습에 필요한 세 가지 텐서를 반환합니다.
return torch.tensor(source_ids), torch.tensor(decoder_input_ids), torch.tensor(decoder_target_ids)
def collate_batch(batch):
"""길이가 서로 다른 문장들을 한 배치에서 사용할 수 있도록 PAD로 길이를 맞춥니다."""
# 배치에서 인코더 입력, 디코더 입력, 디코더 정답을 각각 분리합니다.
sources, decoder_inputs, decoder_targets = zip(*batch)
# PAD 토큰의 인덱스는 config에서 0번이 되도록 설계했습니다.
padding_value = 0
# 인코더 입력 문장들의 길이를 가장 긴 문장에 맞춥니다.
sources = torch.nn.utils.rnn.pad_sequence(sources, batch_first=True, padding_value=padding_value)
# 디코더 입력 문장들의 길이를 가장 긴 문장에 맞춥니다.
decoder_inputs = torch.nn.utils.rnn.pad_sequence(decoder_inputs, batch_first=True, padding_value=padding_value)
# 디코더 정답 문장들의 길이를 가장 긴 문장에 맞춥니다.
decoder_targets = torch.nn.utils.rnn.pad_sequence(decoder_targets, batch_first=True, padding_value=padding_value)
# 패딩이 완료된 배치 텐서를 반환합니다.
return sources, decoder_inputs, decoder_targets
predict.py
detect_language() 한글이있으며 ko아니면 en로 설정. 언어감지.
build_direction_source() 언어가 en일때는 en2ko 토큰을 붙이고 아니면 ko2en토큰을 붙인다. 학습할때도 붙였으니 예측할때도 붙임.
load_model()
모델파일이 없으면 에러 출력.
cpu사용
학습당시 저장했던 torch.load() 메타데이터를 불러와 저장.
model = Seq2SeqTranslator() 객체만들기.
model.load_state_dict() 사전에 학습된 가중치를 넣는다.
model.eval() 평가모드로 변경
return
load_exact_dictionary() csv를 import해서
CSV 파일의 내용을 딕셔너리로 만드는 것
mapping으로 ('en', 'hello'): '안녕하세요', ('ko', '안녕하세요'): 'hello'로 생성하는것. return.
translate()
text가 없을때 alert.
언어를 감지해 source_lang에 초기화.
load_exavt_dictionary() 해 exact_dict에 할당.
exact_key 사용자가 입력한것을 normalize_text()를 거쳐 소문자등으로 변환한다음에 lang을 붙여 할당.("en","hello")
csv에 있는 단어라면 dict[key]를 return해 '안녕하세요' 굳이 이것저것 실행하지않고 return.
모델이 None없다면 load_model()
source_text에 build_direction_source() en2ko등을 붙이는 방향 토큰 추가
source_idx에 encode_text()를 거쳐 숫자로 변환. 마지막엔 eos가 붙음.
source_tensor 텐서변환.
no_grad() 기울기 계산을 끄고
model.encoder() 입력문장을 하나의 hidden state로 압축,
decoder_input 디코더가 번역을 시작할 수 있도록 첫 입력으로 <SOS> 토큰의 인덱스를 decoder_input에 저장한다.
result_chars 초기화 []
for max_output_len만큼 돌린다.
디코더 실행. 현재입력문자, 전달받은 hidden을 이용해 model.decoder() 한번 실행하는것. 다음 글자를 예측한다.
next_id , logits, hidden = model.decoder(decoder_input, hidden)이후 logits에는 문자사전의 모든 문자 점수가 들어있다. 이 로짓에 torch.argmax() 가장 큰값의 인덱스를 찾아 logits[배치크기, 문장길이, 문자개수] 마지막시점에서 생성한 위치의 모든 문자점수만 가져와 .item() tensor형태가 아닌 숫자로 가져온다.
next_char이를 문자로 변환. 아까만든 사전 idx2char를 사용
EOS_TOKEN를 만나면 중지.
{"<PAD>", SOS_TOKEN, UNK_TOKEN}:이면 패스한다.
import torch
import re
from src.config import MODEL_PATH, META_PATH, EMBED_SIZE, HIDDEN_SIZE, MAX_OUTPUT_LEN, SOS_TOKEN, EOS_TOKEN, DATA_PATH, UNK_TOKEN
from src.data_utils import normalize_text, encode_text
from src.model import Seq2SeqTranslator
def detect_language(text) -> str:
if re.search(r'[가-힣]', text):
return 'ko'
return 'en'
def build_direction_source(text: str, source_lang: str) -> str:
if source_lang == 'en':
return '<EN2KO' + normalize_text(text)
return '<KO2EN' + normalize_text(text)
def load_model():
if not MODEL_PATH.exists() or not META_PATH.exists():
raise FileNotFoundError("학습된 모델 파일이 없습니다. python -m src.train")
meta = torch.load(META_PATH, map_location="cpu")
char2idx = meta["char2idx"]
idx2char = meta["idx2char"]
model = Seq2SeqTranslator(len(char2idx), meta.get("embed_size", EMBED_SIZE), meta.get("hidden_size", HIDDEN_SIZE))
model.load_state_dict(torch.load(MODEL_PATH, map_location="cpu"))
model.eval()
return model, char2idx, idx2char
def load_exact_dictionary():
import csv
mapping = {}
with open(DATA_PATH, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
en = normalize_text(row["en"])
ko = normalize_text(row["ko"])
mapping['en', en] = ko
mapping['ko', ko] = en
return mapping
def translate(text:str, model=None, char2idx=None, idx2char=None) -> str:
if not text or not text.strip():
return "번역할문장을 입력하세요"
source_lang = detect_language(text)
exact_dict = load_exact_dictionary()
exact_key = (source_lang, normalize_text(text))
if exact_key in exact_dict:
return exact_dict[exact_key]
if model is None or char2idx is None or idx2char is None:
model, char2idx, idx2char = load_model()
source_text = build_direction_source(text, source_lang)
source_idx = encode_text(source_text, char2idx, add_eos=True)
source_tensor = torch.tensor(source_idx, dtype=torch.long).unsqueeze(0)
with torch.no_grad():
hidden = model.encoder(source_tensor)
decoder_input = torch.tensor([[char2idx[SOS_TOKEN]]], dtype=torch.long)
result_chars = []
for _ in range(MAX_OUTPUT_LEN):
logits, hidden = model.decoder(decoder_input, hidden)
next_id = int(torch.argmax(logits[:, -1, :], dim=1).item())
next_char = idx2char.get(next_id, UNK_TOKEN)
if next_char == EOS_TOKEN:
break
if next_char not in {"<PAD>", SOS_TOKEN, UNK_TOKEN}:
result_chars.append(next_char)
decoder_input = torch.tensor([[next_id]], dtype=torch.long)
result = "".join(result_chars).strip()
if not result:
return "번역 결과를 생성하지 못했습니다. 학습 데이터를 늘리거나 epoch를 증가시켜주세요"
return result
'Personal > SK 네트웍스 AI 캠프' 카테고리의 다른 글
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day39_LLM의 개념과 API (0) | 2026.07.03 |
|---|---|
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day38_자연어 처리를 위한 어탠션과 트랜스포머 (0) | 2026.07.02 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day36_자연어 처리를 위한 언어 모델 (0) | 2026.06.30 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day35_자연어 처리를 위한 텍스트 분류 (0) | 2026.06.30 |
| [SK네트웍스 Family AI 캠프] 32기 9주차 회고: Day32 ~ Day34 (0) | 2026.06.26 |
